leocrm/app/plugins/miniapp_registry.py

"""Universal MiniApp registry — platform-level (Phase M1).

MiniApps are the platform's universal UI building blocks, hostable in Chat,
Dashboard, Windows and AI agent output. Every plugin registers its MiniApps
here (manifest-driven via BasePlugin lifecycle or programmatically at
runtime). The registry is deliberately host-agnostic: hosts declare where
an app may appear, the registry itself never renders anything.

History: grew out of kommunikation/miniapp_registry.py (chat-only, 92
lines). That module is now a compatibility bridge re-exporting this one —
all existing importers (kommunikation contracts, automation routes) keep
working unchanged.

Security model (fail-closed):
- ``MiniAppDef.permission`` — apps requiring a permission the user lacks are
  filtered out server-side by the /api/v1/miniapps listing and answered
  with 403 on single-app lookup. Empty permission = visible to everyone
  (backward compatible with the old chat miniapps).
- Hosts additionally restrict where an app may appear (?host= filter).
"""

from __future__ import annotations

import logging
from typing import Any

from pydantic import BaseModel, Field

logger = logging.getLogger(__name__)

DEFAULT_HOSTS: list[str] = ["chat", "dashboard", "window"]


class MiniAppDef(BaseModel):
    """Definition of a MiniApp that plugins can register."""

    app_id: str = Field(..., description="Unique app identifier")
    name: str = Field(..., description="Display name")
    icon: str = Field(default="AppWindow", description="Icon name (lucide or emoji)")
    description: str = Field(default="", description="App description")
    plugin_name: str = Field(..., description="Plugin that registered this app")
    render_schema: dict[str, Any] = Field(
        default_factory=dict, description="JSON schema for frontend rendering"
    )
    # ── Phase M1 extensions ──
    permission: str = Field(
        default="",
        description="Required permission (fail-closed). Empty = everyone.",
    )
    settings_schema: dict[str, Any] = Field(
        default_factory=dict,
        description="JSON schema for the per-instance settings form",
    )
    col_span: int = Field(default=1, ge=1, le=12, description="Grid column span")
    row_span: int = Field(default=1, ge=1, le=12, description="Grid row span")
    hosts: list[str] = Field(
        default_factory=lambda: list(DEFAULT_HOSTS),
        description="Hosts this app may appear in (chat, dashboard, window)",
    )
    component: str = Field(
        default="", description="Frontend component path (for manifest-contributed apps)"
    )
    order: int = Field(default=100, description="Sort order in palettes")
    builtin: bool = Field(
        default=True, description="Registered in-process (vs. future remote apps)"
    )


class MiniAppRegistry:
    """Registry for MiniApps provided by system and plugins."""

    def __init__(self) -> None:
        self._apps: dict[str, MiniAppDef] = {}

    def register(
        self,
        app_id: str,
        name: str,
        icon: str = "AppWindow",
        description: str = "",
        plugin_name: str = "system",
        render_schema: dict[str, Any] | None = None,
        permission: str = "",
        settings_schema: dict[str, Any] | None = None,
        col_span: int = 1,
        row_span: int = 1,
        hosts: list[str] | None = None,
        component: str = "",
        order: int = 100,
    ) -> None:
        """Register or replace a MiniApp."""
        app = MiniAppDef(
            app_id=app_id,
            name=name,
            icon=icon,
            description=description,
            plugin_name=plugin_name,
            render_schema=render_schema or {},
            permission=permission,
            settings_schema=settings_schema or {},
            col_span=col_span,
            row_span=row_span,
            hosts=hosts if hosts is not None else list(DEFAULT_HOSTS),
            component=component,
            order=order,
        )
        self._apps[app_id] = app
        logger.info("MiniApp registered: %s by %s", app_id, plugin_name)

    def unregister(self, app_id: str) -> None:
        """Unregister a MiniApp."""
        app = self._apps.pop(app_id, None)
        if app:
            logger.info("MiniApp unregistered: %s", app_id)

    def unregister_plugin(self, plugin_name: str) -> None:
        """Unregister all MiniApps from a specific plugin."""
        to_remove = [
            app_id for app_id, app in self._apps.items() if app.plugin_name == plugin_name
        ]
        for app_id in to_remove:
            self._apps.pop(app_id, None)
        if to_remove:
            logger.info("Unregistered %d MiniApps from plugin %s", len(to_remove), plugin_name)

    def list_apps(self, host: str | None = None) -> list[dict[str, Any]]:
        """List MiniApp definitions, optionally filtered by host."""
        apps = self._apps.values()
        if host:
            apps = [a for a in apps if host in a.hosts]
        return [app.model_dump() for app in apps]

    def get_app(self, app_id: str) -> MiniAppDef | None:
        """Get a specific MiniApp definition."""
        return self._apps.get(app_id)


# ─── Singleton helpers ──

_registry: MiniAppRegistry | None = None


def get_miniapp_registry() -> MiniAppRegistry:
    """Return the shared singleton MiniAppRegistry instance."""
    global _registry
    if _registry is None:
        _registry = MiniAppRegistry()
    return _registry


def reset_miniapp_registry() -> None:
    """Reset the singleton instance (useful for tests)."""
    global _registry
    _registry = None


def user_permits(current_user: dict[str, Any], app: dict[str, Any]) -> bool:
    """Check whether *current_user* may see/use the MiniApp *app*.

    Empty permission = visible to everyone; otherwise fail-closed check
    (system admins always pass). Shared by /api/v1/miniapps and the
    personal dashboard seed (Phase M2) so both apply identical rules.
    """
    from app.core.permissions import check_permission

    required = app.get("permission") or ""
    if not required:
        return True
    if current_user.get("is_system_admin"):
        return True
    return check_permission(current_user, required)