"""Compact/sidebar-summary menu override API. Routes (prefix ``/api/compact-menu-overrides``): GET "" read effective/default/personal compact-menu overrides. ``menu_type`` omitted / empty → type A (original legacy data). PUT "" replace that type in the caller's editable scope. ``menu_type`` omitted / empty → type A, does not wipe type B. DELETE "" reset that type in the caller's editable scope. """ from __future__ import annotations from datetime import datetime from typing import Literal from fastapi import APIRouter, HTTPException, Query, Request from pydantic import BaseModel, Field from app.gateway.auth.disabled_mode import is_auth_disabled from app.gateway.catalog_cache import cached_catalog, catalog_key, invalidate_catalog from app.gateway.deps import get_optional_user_from_request from deerflow.persistence.compact_menu_overrides import ( DEFAULT_COMPACT_MENU_TYPE, CompactMenuOverrideStore, normalize_compact_menu_type, ) router = APIRouter(prefix="/api/compact-menu-overrides", tags=["compact-menu-overrides"]) class CompactMenuOverride(BaseModel): node_id: str = Field(..., min_length=1, max_length=128) placement: Literal["sidebar", "settings", "toolbox"] = "sidebar" sort_order: int = 0 label: str | None = Field(default=None, max_length=255) disabled: bool = False class CompactMenuOverrideResponse(CompactMenuOverride): menu_type: str = DEFAULT_COMPACT_MENU_TYPE updated_by: str | None = None updated_at: datetime | str | None = None class CompactMenuOverrideListResponse(BaseModel): menu_type: str = DEFAULT_COMPACT_MENU_TYPE overrides: list[CompactMenuOverrideResponse] default_overrides: list[CompactMenuOverrideResponse] = Field(default_factory=list) user_overrides: list[CompactMenuOverrideResponse] = Field(default_factory=list) has_user_override: bool = False editable_scope: Literal["default", "user"] = "default" class CompactMenuOverrideReplaceRequest(BaseModel): menu_type: str | None = Field( default=None, max_length=32, description="A or B. Omitted/empty = A (original compact-menu set).", ) overrides: list[CompactMenuOverride] = Field(default_factory=list) def _get_store(request: Request) -> CompactMenuOverrideStore: store = getattr(request.app.state, "compact_menu_override_store", None) if store is None: raise HTTPException(status_code=503, detail="Compact menu override store not available") return store async def _actor(request: Request) -> tuple[str | None, bool]: """Return ``(user_id, is_admin)`` while preserving auth-disabled mode.""" user = await get_optional_user_from_request(request) if user is None: if is_auth_disabled(): return None, True raise HTTPException(status_code=401, detail="Not authenticated") return str(user.id), getattr(user, "system_role", None) == "admin" def _merge_overrides(default_rows: list[dict], user_rows: list[dict] | None) -> list[dict]: """Overlay personal rows and retain administrator rows for new nodes.""" if user_rows is None: return [dict(row) for row in default_rows] merged: list[dict] = [] seen: set[str] = set() for row in user_rows: node_id = str(row.get("node_id") or "") if not node_id or node_id in seen: continue merged.append(dict(row)) seen.add(node_id) for row in default_rows: node_id = str(row.get("node_id") or "") if node_id and node_id not in seen: merged.append(dict(row)) seen.add(node_id) return merged async def _read_response( request: Request, store: CompactMenuOverrideStore, menu_type: str, *, actor: tuple[str | None, bool] | None = None, ) -> CompactMenuOverrideListResponse: user_id, is_admin = actor if actor is not None else await _actor(request) default_rows = await store.list_overrides(menu_type) if is_admin or user_id is None: return CompactMenuOverrideListResponse( menu_type=menu_type, overrides=[CompactMenuOverrideResponse(**row) for row in default_rows], default_overrides=[CompactMenuOverrideResponse(**row) for row in default_rows], user_overrides=[], has_user_override=False, editable_scope="default", ) user_rows = await store.list_user_overrides(user_id, menu_type) effective = _merge_overrides(default_rows, user_rows) return CompactMenuOverrideListResponse( menu_type=menu_type, overrides=[CompactMenuOverrideResponse(**row) for row in effective], default_overrides=[CompactMenuOverrideResponse(**row) for row in default_rows], user_overrides=[CompactMenuOverrideResponse(**row) for row in (user_rows or [])], has_user_override=user_rows is not None, editable_scope="user", ) @router.get("", response_model=CompactMenuOverrideListResponse) async def list_compact_menu_overrides( request: Request, menu_type: str | None = Query( default=None, max_length=32, description="A or B. Omitted/empty = A, returning the original compact-menu rows.", ), ) -> CompactMenuOverrideListResponse: t = normalize_compact_menu_type(menu_type) actor = await _actor(request) store = _get_store(request) return await cached_catalog( request, "compact_menu_overrides", catalog_key(*actor, t), lambda: _read_response(request, store, t, actor=actor), CompactMenuOverrideListResponse, ) @router.put("", response_model=CompactMenuOverrideListResponse) async def replace_compact_menu_overrides( request: Request, body: CompactMenuOverrideReplaceRequest, ) -> CompactMenuOverrideListResponse: t = normalize_compact_menu_type(body.menu_type) store = _get_store(request) user_id, is_admin = await _actor(request) payload = [o.model_dump() for o in body.overrides] if is_admin or user_id is None: await store.replace_all(payload, menu_type=t, updated_by=user_id) else: await store.replace_user_overrides(user_id, payload, menu_type=t, updated_by=user_id) await invalidate_catalog(request, "compact_menu_overrides") return await _read_response(request, store, t) @router.delete("", response_model=CompactMenuOverrideListResponse) async def reset_compact_menu_overrides( request: Request, menu_type: str | None = Query(default=None, max_length=32), ) -> CompactMenuOverrideListResponse: """Clear the caller's personal compact layout for one variant.""" t = normalize_compact_menu_type(menu_type) store = _get_store(request) user_id, is_admin = await _actor(request) if is_admin or user_id is None: await store.replace_all([], menu_type=t, updated_by=user_id) else: await store.replace_user_overrides(user_id, [], menu_type=t, updated_by=user_id) await invalidate_catalog(request, "compact_menu_overrides") return await _read_response(request, store, t)