webmail-client/backend/app/plugins/loader.py

"""Plugin discovery + core wiring (SPEC §11).

Discovers plugins via the ``webmail.plugins`` entry-point group, then:
- mounts each plugin router at ``/api/plugins/{name}`` (API process),
- forwards §8.3 events to plugin hooks (both API and worker processes),
- aggregates UI manifests for ``GET /api/plugins/ui-manifest``.

Plugin background jobs (§11.2) and migrations (§11.1) are wired in later M3 phases.
"""

from __future__ import annotations

import logging
from importlib.metadata import entry_points
from typing import TYPE_CHECKING, Any

from app.events import EventBus
from app.plugins.types import Page, Plugin, Slot, UiManifest

if TYPE_CHECKING:
    from fastapi import FastAPI

logger = logging.getLogger(__name__)

ENTRYPOINT_GROUP = "webmail.plugins"

_plugins: list[Plugin] | None = None
# Plugin names whose hooks are already subscribed on a given bus, so repeated
# create_app()/worker starts in one process don't double-register (tests, reload).
_hooks_registered: set[tuple[int, str]] = set()


def _coerce_plugin(obj: Any) -> Plugin:
    """An entry point may resolve to a ``Plugin`` or a module exposing ``plugin``."""
    if isinstance(obj, Plugin):
        return obj
    candidate = getattr(obj, "plugin", None)
    if isinstance(candidate, Plugin):
        return candidate
    raise TypeError(f"entry point {obj!r} does not expose a Plugin (SPEC §11)")


def discover_plugins() -> list[Plugin]:
    """Load every plugin registered under the ``webmail.plugins`` group."""
    found: list[Plugin] = []
    seen: set[str] = set()
    for ep in entry_points(group=ENTRYPOINT_GROUP):
        try:
            plugin = _coerce_plugin(ep.load())
        except Exception:  # noqa: BLE001 - one bad plugin must not break the app
            logger.exception("failed to load plugin entry point %s", ep.name)
            continue
        if plugin.name in seen:
            logger.error("duplicate plugin name %r; skipping", plugin.name)
            continue
        seen.add(plugin.name)
        found.append(plugin)
        logger.info("loaded plugin %s v%s", plugin.name, plugin.version)
    return found


def get_plugins() -> list[Plugin]:
    """Return the process-wide discovered plugin list (cached)."""
    global _plugins
    if _plugins is None:
        _plugins = discover_plugins()
    return _plugins


def set_plugins_for_test(plugins: list[Plugin] | None) -> None:
    """Override (or reset with ``None``) the plugin registry — tests only."""
    global _plugins
    _plugins = plugins
    _hooks_registered.clear()


def register_hooks(bus: EventBus) -> None:
    """Subscribe each plugin's hooks to the bus (idempotent per process/bus).

    Called in both the API and the worker so hooks fire wherever the event is
    published (e.g. ``message.new`` originates in the worker)."""
    for plugin in get_plugins():
        marker = (id(bus), plugin.name)
        if marker in _hooks_registered:
            continue
        for event, handler in plugin.hooks.items():
            bus.subscribe(event, handler)
        _hooks_registered.add(marker)


def mount_routers(app: FastAPI) -> None:
    """Mount each plugin's router at ``/api/plugins/{name}`` (SPEC §11)."""
    for plugin in get_plugins():
        if plugin.router is not None:
            app.include_router(plugin.router, prefix=f"/api/plugins/{plugin.name}")


def _slot_dict(slot: Slot) -> dict[str, str]:
    d = {"name": slot.name, "bundle": slot.bundle}
    if slot.title:
        d["title"] = slot.title
    return d


def _page_dict(page: Page) -> dict[str, Any]:
    return {
        "route": page.route,
        "bundle": page.bundle,
        "nav": {"icon": page.nav.icon, "label": page.nav.label, "order": page.nav.order},
    }


def _manifest_dict(plugin: Plugin, manifest: UiManifest) -> dict[str, Any]:
    return {
        "name": plugin.name,
        "version": plugin.version,
        "builtin": plugin.builtin,  # core plugins the UI must not offer to remove
        "slots": [_slot_dict(s) for s in manifest.slots],
        "pages": [_page_dict(p) for p in manifest.pages],
        "capabilities": dict(manifest.capabilities),
    }


def build_ui_manifest() -> dict[str, Any]:
    """Aggregate every plugin's UI manifest for the frontend loader (SPEC §11.3)."""
    plugins = [_manifest_dict(p, p.ui_manifest) for p in get_plugins() if p.ui_manifest is not None]
    return {"plugins": plugins}


def get_plugin(name: str) -> Plugin | None:
    """Return the loaded plugin named ``name``, or None."""
    return next((p for p in get_plugins() if p.name == name), None)


class PluginRemovalError(Exception):
    """Raised when trying to uninstall a builtin (core) plugin (SPEC §11 / §12)."""


def assert_removable(name: str) -> Plugin:
    """Return the plugin, or raise if it's unknown or builtin (non-removable).

    Core plugins (calendar, contacts) ship ``builtin=True`` and may never be
    uninstalled by a user; the uninstall path calls this before touching data."""
    plugin = get_plugin(name)
    if plugin is None:
        raise PluginRemovalError(f"unknown plugin {name!r}")
    if plugin.builtin:
        raise PluginRemovalError(f"plugin {name!r} is builtin and cannot be removed")
    return plugin