from __future__ import annotations from dataclasses import dataclass from typing import Callable, Iterable, Mapping from sqlalchemy.orm import Session from govoplan_core.core.concurrency import ( RevisionConflictError, assert_revision_precondition, strong_resource_etag, ) from govoplan_core.core.registry import PlatformRegistry from govoplan_quick_access.backend.db.models import QuickAccessProfile from govoplan_quick_access.backend.schemas import ( CatalogueCategoryResponse, CatalogueResponse, CatalogueToolResponse, EffectiveCategoryResponse, EffectiveQuickAccessResponse, EffectiveStalePreferenceResponse, EffectiveToolResponse, PreferenceEntry, ProfileResponse, ProfileUpdateRequest, ) BASE_CATEGORIES = ( CatalogueCategoryResponse( id="work", label="i18n:govoplan-quick-access.category.work", description="i18n:govoplan-quick-access.category.work_description", icon="list-checks", order=10, ), CatalogueCategoryResponse( id="calendar", label="i18n:govoplan-quick-access.category.calendar", description="i18n:govoplan-quick-access.category.calendar_description", icon="calendar", order=20, ), CatalogueCategoryResponse( id="messages", label="i18n:govoplan-quick-access.category.messages", description="i18n:govoplan-quick-access.category.messages_description", icon="messages-square", order=30, ), CatalogueCategoryResponse( id="files", label="i18n:govoplan-quick-access.category.files", description="i18n:govoplan-quick-access.category.files_description", icon="files", order=40, ), ) def scope_key(scope_type: str, tenant_id: str | None, scope_id: str | None) -> str: if scope_type == "system": return "system:*" if scope_type == "tenant" and tenant_id: return f"tenant:{tenant_id}" if scope_type == "user" and tenant_id and scope_id: return f"user:{tenant_id}:{scope_id}" raise ValueError("Invalid Quick Access profile scope") def build_catalogue( registry: PlatformRegistry, *, permission_checker: Callable[[str], bool] | None = None, allowed_module_ids: Iterable[str] | None = None, ) -> CatalogueResponse: allowed_modules = ( None if allowed_module_ids is None else frozenset(allowed_module_ids) ) tools: list[CatalogueToolResponse] = [] for manifest in registry.manifests(): if allowed_modules is not None and manifest.id not in allowed_modules: continue frontend = manifest.frontend if frontend is None: continue for tool in frontend.quick_access_tools: if permission_checker is not None: if tool.required_all and not all(permission_checker(scope) for scope in tool.required_all): continue if tool.required_any and not any(permission_checker(scope) for scope in tool.required_any): continue tools.append( CatalogueToolResponse( contract_version=tool.contract_version, id=tool.id, module_id=tool.module_id, category_id=tool.category_id, label=tool.label, description=tool.description, icon=tool.icon, surface_id=tool.surface_id, full_page_path=tool.full_page_path, required_all=list(tool.required_all), required_any=list(tool.required_any), order=tool.order, default_enabled=tool.default_enabled, modes=list(tool.modes), availability=tool.availability, accepted_reference_kinds=list(tool.accepted_reference_kinds), returned_reference_kinds=list(tool.returned_reference_kinds), help_context_id=tool.help_context_id, ) ) tools.sort(key=lambda item: (item.category_id, item.order, item.id)) return CatalogueResponse(categories=list(BASE_CATEGORIES), tools=tools) def get_profile( session: Session, *, scope_type: str, tenant_id: str | None, scope_id: str | None, ) -> QuickAccessProfile | None: key = scope_key(scope_type, tenant_id, scope_id) return session.query(QuickAccessProfile).filter(QuickAccessProfile.scope_key == key).one_or_none() def profile_response( row: QuickAccessProfile | None, *, scope_type: str, tenant_id: str | None, scope_id: str | None, catalogue: CatalogueResponse, ) -> ProfileResponse: revision = row.revision if row is not None else 1 resource_id = scope_key(scope_type, tenant_id, scope_id) category_preferences = _preference_map(row.category_preferences if row else {}) tool_preferences = _preference_map(row.tool_preferences if row else {}) category_ids = {item.id for item in catalogue.categories} tool_ids = {item.id for item in catalogue.tools} return ProfileResponse( scope_type=scope_type, tenant_id=tenant_id, scope_id=scope_id, revision=revision, etag=strong_resource_etag("quick_access_profile", resource_id, revision), category_preferences=category_preferences, tool_preferences=tool_preferences, stale_category_ids=sorted(set(category_preferences) - category_ids), stale_tool_ids=sorted(set(tool_preferences) - tool_ids), ) def update_profile( session: Session, *, scope_type: str, tenant_id: str | None, scope_id: str | None, actor_id: str, payload: ProfileUpdateRequest, if_match: str | None, ) -> QuickAccessProfile: key = scope_key(scope_type, tenant_id, scope_id) assert_revision_precondition( if_match, resource_type="quick_access_profile", resource_id=key, submitted_base_revision=payload.base_revision, ) row = get_profile( session, scope_type=scope_type, tenant_id=tenant_id, scope_id=scope_id, ) current_revision = row.revision if row is not None else 1 if current_revision != payload.base_revision: raise RevisionConflictError( resource_type="quick_access_profile", resource_id=key, current_revision=current_revision, submitted_base_revision=payload.base_revision, refresh_path=_profile_path(scope_type), current_etag=strong_resource_etag( "quick_access_profile", key, current_revision ), ) if scope_type == "user" and any( entry.forced for entry in ( *payload.category_preferences.values(), *payload.tool_preferences.values(), ) ): raise ValueError("Personal Quick Access preferences cannot force items") category_preferences = _serialized_preferences(payload.category_preferences) tool_preferences = _serialized_preferences(payload.tool_preferences) if row is None: row = QuickAccessProfile( scope_type=scope_type, tenant_id=tenant_id, scope_id=scope_id, scope_key=key, revision=2, created_by=actor_id, updated_by=actor_id, category_preferences=category_preferences, tool_preferences=tool_preferences, ) session.add(row) else: row.revision += 1 row.updated_by = actor_id row.category_preferences = category_preferences row.tool_preferences = tool_preferences session.flush() return row def resolve_effective( session: Session, *, registry: PlatformRegistry, tenant_id: str, account_id: str, permission_checker: Callable[[str], bool] | None = None, allowed_module_ids: Iterable[str] | None = None, catalogue: CatalogueResponse | None = None, ) -> EffectiveQuickAccessResponse: if catalogue is None: catalogue = build_catalogue( registry, permission_checker=permission_checker, allowed_module_ids=allowed_module_ids, ) profiles = ( ("system", get_profile(session, scope_type="system", tenant_id=None, scope_id=None)), ("tenant", get_profile(session, scope_type="tenant", tenant_id=tenant_id, scope_id=tenant_id)), ("user", get_profile(session, scope_type="user", tenant_id=tenant_id, scope_id=account_id)), ) diagnostics: list[str] = [] stale_preferences = _stale_preferences(profiles, catalogue) category_states: dict[str, _EffectiveState] = {} for category in catalogue.categories: state = _EffectiveState(enabled=True, order=category.order) for source, profile in profiles: entry = _profile_entry(profile, "category_preferences", category.id, diagnostics) state.apply(entry, source=source) category_states[category.id] = state tools_by_category: dict[str, list[EffectiveToolResponse]] = { category.id: [] for category in catalogue.categories } for tool in catalogue.tools: state = _EffectiveState(enabled=tool.default_enabled, order=tool.order) for source, profile in profiles: entry = _profile_entry(profile, "tool_preferences", tool.id, diagnostics) state.apply(entry, source=source) category_state = category_states.get(tool.category_id) enabled = state.enabled and bool(category_state and category_state.enabled) availability_source = ( category_state.availability_source if state.enabled and category_state and not category_state.enabled else state.availability_source ) tool_payload = tool.model_dump() tool_payload["order"] = state.order tools_by_category.setdefault(tool.category_id, []).append( EffectiveToolResponse( **tool_payload, enabled=enabled, forced=state.forced, locked_by=state.locked_by, availability_state=_availability_state(enabled, state.forced), availability_source=availability_source, order_source=state.order_source, ) ) categories: list[EffectiveCategoryResponse] = [] for category in catalogue.categories: state = category_states[category.id] tools = sorted( tools_by_category.get(category.id, []), key=lambda item: (item.order, item.id), ) enabled = state.enabled and any(tool.enabled for tool in tools) category_payload = category.model_dump() category_payload["order"] = state.order categories.append( EffectiveCategoryResponse( **category_payload, enabled=enabled, forced=state.forced, locked_by=state.locked_by, availability_state=_availability_state(enabled, state.forced), availability_source=state.availability_source, order_source=state.order_source, tools=tools, ) ) categories.sort(key=lambda item: (item.order, item.id)) return EffectiveQuickAccessResponse( categories=categories, stale_preferences=stale_preferences, diagnostics=list(dict.fromkeys(diagnostics)), ) @dataclass(slots=True) class _EffectiveState: enabled: bool order: int forced: bool = False locked_by: str | None = None availability_source: str = "module" order_source: str = "module" def apply(self, entry: PreferenceEntry | None, *, source: str) -> None: if entry is None: return if entry.order is not None: self.order = entry.order self.order_source = source if self.locked_by is not None: return if entry.enabled is not None: self.enabled = entry.enabled self.availability_source = source if source != "user" and entry.enabled is False: self.forced = False self.locked_by = source elif source != "user" and entry.forced: self.enabled = True self.forced = True self.locked_by = source self.availability_source = source def _availability_state(enabled: bool, forced: bool) -> str: if not enabled: return "blocked" return "forced" if forced else "available" def _stale_preferences( profiles: tuple[tuple[str, QuickAccessProfile | None], ...], catalogue: CatalogueResponse, ) -> list[EffectiveStalePreferenceResponse]: category_ids = {item.id for item in catalogue.categories} tool_ids = {item.id for item in catalogue.tools} stale: list[EffectiveStalePreferenceResponse] = [] for source, profile in profiles: if profile is None: continue for kind, field, known_ids in ( ("category", "category_preferences", category_ids), ("tool", "tool_preferences", tool_ids), ): values = getattr(profile, field, {}) if not isinstance(values, Mapping): continue stale.extend( EffectiveStalePreferenceResponse(id=str(item_id), kind=kind, source=source) for item_id in values if str(item_id) not in known_ids ) return sorted(stale, key=lambda item: (item.source, item.kind, item.id)) def _profile_entry( profile: QuickAccessProfile | None, field: str, item_id: str, diagnostics: list[str], ) -> PreferenceEntry | None: if profile is None: return None raw_map = getattr(profile, field, {}) if not isinstance(raw_map, Mapping): diagnostics.append(f"Ignored malformed {field} in {profile.scope_key}.") return None raw = raw_map.get(item_id) if raw is None: return None try: return PreferenceEntry.model_validate(raw) except Exception: diagnostics.append(f"Ignored malformed preference for {item_id} in {profile.scope_key}.") return None def _preference_map(raw: object) -> dict[str, PreferenceEntry]: if not isinstance(raw, Mapping): return {} result: dict[str, PreferenceEntry] = {} for item_id, value in raw.items(): try: result[str(item_id)] = PreferenceEntry.model_validate(value) except Exception: continue return result def _serialized_preferences( preferences: Mapping[str, PreferenceEntry], ) -> dict[str, dict[str, object]]: if len(preferences) > 500: raise ValueError("Quick Access profiles may contain at most 500 preferences") return { str(item_id): entry.model_dump(mode="json", exclude_none=True) for item_id, entry in preferences.items() if str(item_id).strip() } def _profile_path(scope_type: str) -> str: return { "system": "/api/v1/quick-access/profiles/system", "tenant": "/api/v1/quick-access/profiles/tenant", "user": "/api/v1/quick-access/profiles/me", }[scope_type] __all__ = [ "BASE_CATEGORIES", "build_catalogue", "get_profile", "profile_response", "resolve_effective", "scope_key", "update_profile", ]