"""Plugin registry — manages discovered, installed, and active plugins.""" from __future__ import annotations import importlib import logging from typing import Any from fastapi import FastAPI from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession from app.core.event_bus import EventBus, get_event_bus from app.core.service_container import ServiceContainer, get_container from app.models.plugin import Plugin as PluginModel from app.plugins.base import BasePlugin from app.plugins.migration_runner import MigrationRunner logger = logging.getLogger(__name__) class PluginRegistry: """Registry for managing plugin lifecycle. Maintains in-memory state for discovered plugins and their runtime instances, backed by database records for persistent status tracking. """ def __init__(self) -> None: # name -> BasePlugin instance (discovered/loaded) self._plugins: dict[str, BasePlugin] = {} # name -> PluginModel DB record (installed status) self._db_status: dict[str, PluginModel] = {} # name -> list of actual route objects added to the FastAPI app # (tracked by object identity to avoid cross-plugin route removal) self._mounted_routes: dict[str, list[Any]] = {} self._engine: AsyncEngine | None = None self._event_bus: EventBus = get_event_bus() self._container: ServiceContainer = get_container() self._migration_runner: MigrationRunner | None = None self._app: FastAPI | None = None self._initialized = False def initialize(self, engine: AsyncEngine, app: FastAPI | None = None) -> None: """Initialize the registry with the DB engine and optional FastAPI app.""" self._engine = engine self._app = app self._migration_runner = MigrationRunner(engine) self._initialized = True @property def migration_runner(self) -> MigrationRunner: if self._migration_runner is None: raise RuntimeError("Registry not initialized — call initialize() first") return self._migration_runner @property def engine(self) -> AsyncEngine: if self._engine is None: raise RuntimeError("Registry not initialized — call initialize() first") return self._engine # ── Discovery ── def discover_builtins(self) -> list[str]: """Discover and load built-in plugins from app.plugins.builtins. Scans the builtins package for modules that export a BasePlugin subclass. Returns list of discovered plugin names. """ discovered: list[str] = [] try: builtins_pkg = importlib.import_module("app.plugins.builtins") except ImportError: return discovered pkg_path = getattr(builtins_pkg, "__path__", None) if pkg_path is None: return discovered import pkgutil for _importer, modname, _ispkg in pkgutil.iter_modules(pkg_path): if modname.startswith("_"): continue try: full_name = f"app.plugins.builtins.{modname}" module = importlib.import_module(full_name) # Look for BasePlugin subclass in module for attr_name in dir(module): attr = getattr(module, attr_name) if ( isinstance(attr, type) and issubclass(attr, BasePlugin) and attr is not BasePlugin ): instance = attr() if instance.name not in self._plugins: self._plugins[instance.name] = instance discovered.append(instance.name) logger.info(f"Discovered plugin: {instance.name} v{instance.version}") except Exception as exc: logger.warning(f"Failed to load builtin plugin module {modname}: {exc}") return discovered def register_plugin(self, plugin: BasePlugin) -> None: """Manually register a plugin instance.""" self._plugins[plugin.name] = plugin logger.info(f"Registered plugin: {plugin.name} v{plugin.version}") def get_plugin(self, name: str) -> BasePlugin | None: """Get a registered plugin instance by name.""" return self._plugins.get(name) def list_discovered(self) -> list[str]: """List all discovered plugin names.""" return list(self._plugins.keys()) # ── Notification Type Sync ── async def sync_notification_types(self, db: AsyncSession) -> None: """Sync notification types from all active plugins to DB. For each active plugin, calls get_notification_types() and upserts the returned types into the notification_types table. Removes types belonging to plugins that are no longer active. """ from app.models.notification import NotificationType # Collect all notification types from active plugins active_types: dict[str, dict[str, Any]] = {} # type_key -> type_data for name, plugin in self._plugins.items(): record = await self._get_plugin_record(db, name) if record is None or not record.active: continue for nt in plugin.get_notification_types(): key = nt.get("type_key", "") if not key: continue active_types[key] = { "type_key": key, "plugin_name": name, "category": nt.get("category", "general"), "label": nt.get("label", key), "description": nt.get("description"), "is_enabled_by_default": nt.get("is_enabled_by_default", True), } # Fetch existing types from DB result = await db.execute(select(NotificationType)) existing = {row.type_key: row for row in result.scalars().all()} # Upsert active types for key, data in active_types.items(): if key in existing: row = existing[key] row.plugin_name = data["plugin_name"] row.category = data["category"] row.label = data["label"] row.description = data["description"] row.is_enabled_by_default = data["is_enabled_by_default"] else: new_row = NotificationType( type_key=key, plugin_name=data["plugin_name"], category=data["category"], label=data["label"], description=data["description"], is_enabled_by_default=data["is_enabled_by_default"], ) db.add(new_row) # Remove types from plugins that are no longer active active_keys = set(active_types.keys()) for key, row in existing.items(): if key not in active_keys: # Check if the plugin is still active plugin_name = row.plugin_name plugin_record = await self._get_plugin_record(db, plugin_name) if plugin_record is None or not plugin_record.active: await db.delete(row) await db.flush() logger.info("Synced %d notification types from active plugins", len(active_types)) # ── DB Status Sync ── async def sync_db_status(self, db: AsyncSession) -> None: """Load plugin status records from the database.""" result = await db.execute(select(PluginModel)) self._db_status = {row.name: row for row in result.scalars().all()} def get_db_status(self, name: str) -> PluginModel | None: """Get the DB status record for a plugin.""" return self._db_status.get(name) # ── Dependency Resolution ── async def _check_dependencies_installed(self, db: AsyncSession, name: str) -> None: """Verify that all declared dependencies of a plugin are installed. Raises ValueError listing missing dependencies. """ plugin = self.get_plugin(name) if plugin is None: raise ValueError(f"Plugin '{name}' not found in registry") dependencies = plugin.manifest.dependencies if not dependencies: return missing: list[str] = [] for dep_name in dependencies: dep_record = await self._get_plugin_record(db, dep_name) if dep_record is None or not dep_record.installed: missing.append(dep_name) if missing: raise ValueError( f"Plugin '{name}' has uninstalled dependencies: {', '.join(missing)}. " f"Install them first." ) async def _check_dependencies_active(self, db: AsyncSession, name: str) -> None: """Verify that all declared dependencies of a plugin are active. Raises ValueError listing inactive dependencies. """ plugin = self.get_plugin(name) if plugin is None: raise ValueError(f"Plugin '{name}' not found in registry") dependencies = plugin.manifest.dependencies if not dependencies: return inactive: list[str] = [] for dep_name in dependencies: dep_record = await self._get_plugin_record(db, dep_name) if dep_record is None or not dep_record.active: inactive.append(dep_name) if inactive: raise ValueError( f"Plugin '{name}' has inactive dependencies: {', '.join(inactive)}. " f"Activate them first." ) # ── Topological Sort & Dependents ── def resolve_load_order(self) -> list[str]: """Return plugin names in dependency-aware load order using Kahn's algorithm. Core plugins (is_core=True) are emitted first, then non-core plugins in topological order based on declared dependencies. Raises RuntimeError on circular dependencies or missing dependencies. """ names = set(self._plugins.keys()) # Validate that all declared dependencies exist in the registry for name in names: plugin = self._plugins[name] for dep in plugin.manifest.dependencies: if dep not in names: raise RuntimeError( f"Plugin '{name}' depends on '{dep}' which is not discovered. " f"Available plugins: {', '.join(sorted(names))}" ) # Build adjacency: dep -> list of plugins that depend on it # and in-degree: plugin -> count of unsatisfied dependencies in_degree: dict[str, int] = {name: 0 for name in names} dependents_map: dict[str, list[str]] = {name: [] for name in names} for name in names: plugin = self._plugins[name] for dep in plugin.manifest.dependencies: dependents_map[dep].append(name) in_degree[name] += 1 # Single Kahn pass with priority: core plugins before non-core at each level. # Queue entries are tuples: (priority, name) where priority 0 = core, 1 = non-core. import heapq heap: list[tuple[int, str]] = [] for name in names: if in_degree[name] == 0: priority = 0 if self._plugins[name].manifest.is_core else 1 heapq.heappush(heap, (priority, name)) result: list[str] = [] while heap: _, current = heapq.heappop(heap) result.append(current) for dependent in dependents_map[current]: in_degree[dependent] -= 1 if in_degree[dependent] == 0: priority = 0 if self._plugins[dependent].manifest.is_core else 1 heapq.heappush(heap, (priority, dependent)) if len(result) != len(names): unresolved = names - set(result) raise RuntimeError( f"Circular dependency detected among plugins: {', '.join(sorted(unresolved))}" ) return result def get_dependents(self, name: str) -> list[str]: """Return list of discovered plugin names that depend on the given plugin. This examines manifest.dependencies of all discovered plugins. """ dependents: list[str] = [] for plugin_name, plugin in self._plugins.items(): if name in plugin.manifest.dependencies: dependents.append(plugin_name) return sorted(dependents) # ── Permission Validation (soft check) ── def _check_permissions(self, name: str) -> list[str]: """Soft-check that declared permissions exist in the system. Returns a list of warning messages for permissions that are not recognised in the available system permissions set. Does not raise — this is a warning-only check. """ plugin = self.get_plugin(name) if plugin is None: return [] declared = plugin.manifest.permissions if not declared: return [] # Build a set of all known permission strings from every discovered plugin available: set[str] = set() for p in self._plugins.values(): available.update(p.manifest.permissions) warnings: list[str] = [] for perm in declared: if perm not in available: warnings.append( f"Plugin '{name}' declares permission '{perm}' " f"which is not found in any discovered plugin's permissions." ) return warnings # ── Version Comparison & Update Path ── async def _check_and_run_version_migrations( self, db: AsyncSession, name: str, record: PluginModel ) -> bool: """Compare manifest version with DB version and run migrations if different. If the plugin's manifest version differs from the stored DB version, re-run all migrations to bring the schema up to date, then update the DB record's version field. Returns True if migrations were run, False otherwise. """ plugin = self.get_plugin(name) if plugin is None: return False manifest_version = plugin.manifest.version db_version = record.version if manifest_version == db_version: return False logger.info( f"Plugin '{name}' version mismatch: DB={db_version}, manifest={manifest_version}. " f"Running migrations to update." ) # Re-run migrations to apply any new migration files if plugin.manifest.migrations: await self.migration_runner.run_all_migrations(db, name, plugin.manifest.migrations) # Update DB version to match manifest record.version = manifest_version await db.flush() self._db_status[name] = record logger.info(f"Plugin '{name}' updated to version {manifest_version}.") return True # ── Install / Activate / Deactivate / Uninstall ── async def install(self, db: AsyncSession, name: str) -> PluginModel: """Install a plugin: run migrations and create DB record. Idempotent: if already installed, checks for version updates and returns existing record. Checks that all declared dependencies are installed first. """ plugin = self.get_plugin(name) if plugin is None: raise ValueError(f"Plugin '{name}' not found in registry") # Check if already installed in DB existing = await self._get_plugin_record(db, name) if existing is not None: # Check for version update — run migrations if version changed await self._check_and_run_version_migrations(db, name, existing) return existing # Check dependencies are installed await self._check_dependencies_installed(db, name) # Run migrations if plugin.manifest.migrations: await self.migration_runner.run_all_migrations(db, name, plugin.manifest.migrations) # Call on_install hook await plugin.on_install(db, self._container) # Create DB record record = PluginModel( name=name, display_name=plugin.manifest.display_name, version=plugin.manifest.version, status="installed", installed=True, active=False, is_core=plugin.manifest.is_core, ) db.add(record) await db.flush() self._db_status[name] = record return record async def activate(self, db: AsyncSession, name: str) -> PluginModel: """Activate a plugin: register routes, event listeners, set status=active. Idempotent: if already active, checks for version updates and returns existing record without error. Checks that all declared dependencies are active first. Performs a soft permission check and logs warnings for unknown permissions. """ plugin = self.get_plugin(name) if plugin is None: raise ValueError(f"Plugin '{name}' not found in registry") record = await self._get_plugin_record(db, name) if record is None: raise ValueError(f"Plugin '{name}' is not installed — install first") # Check for version update — run migrations if version changed await self._check_and_run_version_migrations(db, name, record) # Idempotent: already active if record.active and record.status == "active": return record # Check dependencies are active await self._check_dependencies_active(db, name) # Soft permission check — log warnings for unknown permissions perm_warnings = self._check_permissions(name) for warning in perm_warnings: logger.warning(warning) # Call on_activate hook (registers event listeners) await plugin.on_activate(db, self._container, self._event_bus) # Register routes on FastAPI app if available # Track actual route objects by identity to avoid cross-plugin removal if self._app is not None: routers = plugin.get_routes() mounted_routes: list[Any] = [] for router in routers: # Snapshot existing route object IDs before inclusion existing_ids = {id(r) for r in self._app.router.routes} self._app.include_router(router) # Collect newly added route objects for r in self._app.router.routes: if id(r) not in existing_ids: mounted_routes.append(r) self._mounted_routes[name] = mounted_routes # Sync notification types from this plugin await self.sync_notification_types(db) # Update DB record record.status = "active" record.active = True await db.flush() self._db_status[name] = record return record async def deactivate(self, db: AsyncSession, name: str) -> PluginModel: """Deactivate a plugin: unregister event listeners, set status=inactive. Idempotent: if already inactive, returns existing record without error. """ plugin = self.get_plugin(name) if plugin is None: raise ValueError(f"Plugin '{name}' not found in registry") record = await self._get_plugin_record(db, name) if record is None: raise ValueError(f"Plugin '{name}' is not installed") # Idempotent: already inactive if not record.active and record.status == "inactive": return record # Reject deactivation of core plugins if plugin.manifest.is_core or record.is_core: raise ValueError( f"Plugin '{name}' is a core plugin and cannot be deactivated" ) # Reject deactivation if other active plugins depend on it dependents = self.get_dependents(name) active_dependents: list[str] = [] for dep_name in dependents: dep_record = await self._get_plugin_record(db, dep_name) if dep_record is not None and dep_record.active: active_dependents.append(dep_name) if active_dependents: raise ValueError( f"Plugin '{name}' cannot be deactivated because active plugins " f"depend on it: {', '.join(active_dependents)}" ) # Call on_deactivate hook (unregisters event listeners) await plugin.on_deactivate(db, self._container, self._event_bus) # Unregister only the specific route objects that belong to this plugin # (by object identity, not by path — prevents cross-plugin route removal) if self._app is not None and name in self._mounted_routes: mounted_routes = self._mounted_routes.pop(name, []) mounted_ids = {id(r) for r in mounted_routes} self._app.router.routes = [ r for r in self._app.router.routes if id(r) not in mounted_ids ] # Update DB record record.status = "inactive" record.active = False await db.flush() self._db_status[name] = record return record async def uninstall( self, db: AsyncSession, name: str, remove_data: bool = False ) -> PluginModel: """Uninstall a plugin: deactivate, optionally drop tables, remove DB record. Args: db: Async database session. name: Plugin name to uninstall. remove_data: If True, drop all plugin-created tables. Returns: The plugin record before deletion (for response). Raises: ValueError if plugin not installed. """ plugin = self.get_plugin(name) if plugin is None: raise ValueError(f"Plugin '{name}' not found in registry") record = await self._get_plugin_record(db, name) if record is None: raise ValueError(f"Plugin '{name}' is not installed") # Deactivate first if active if record.active: await self.deactivate(db, name) # Refetch record after deactivate record = await self._get_plugin_record(db, name) if record is None: raise ValueError(f"Plugin '{name}' disappeared during uninstall") # Call on_uninstall hook await plugin.on_uninstall(db, self._container) # Optionally drop plugin tables dropped_tables: list[str] = [] if remove_data: dropped_tables = await self.migration_runner.drop_plugin_tables(db, name) # Remove DB record await db.delete(record) await db.flush() self._db_status.pop(name, None) # Return a detached copy for response record_dropped_tables = dropped_tables record.status = "uninstalled" record.dropped_tables = record_dropped_tables return record async def list_plugins(self, db: AsyncSession) -> list[dict[str, Any]]: """List all plugins with their current status. Merges discovered (in-memory) plugins with installed (DB) records. """ result = await db.execute(select(PluginModel)) db_records = {row.name: row for row in result.scalars().all()} plugins_list: list[dict[str, Any]] = [] for name, plugin in self._plugins.items(): record = db_records.get(name) if record is not None: plugins_list.append( { "name": name, "display_name": record.display_name, "version": record.version, "status": record.status, "installed": record.installed, "active": record.active, "description": plugin.manifest.description, "dependencies": plugin.manifest.dependencies, "events": plugin.manifest.events, "migrations": plugin.manifest.migrations, "permissions": plugin.manifest.permissions, } ) else: plugins_list.append( { "name": name, "display_name": plugin.manifest.display_name, "version": plugin.version, "status": "discovered", "installed": False, "active": False, "description": plugin.manifest.description, "dependencies": plugin.manifest.dependencies, "events": plugin.manifest.events, "migrations": plugin.manifest.migrations, "permissions": plugin.manifest.permissions, } ) # Also include DB-only records (plugins that were installed but no longer discovered) for name, record in db_records.items(): if name not in self._plugins: plugins_list.append( { "name": name, "display_name": record.display_name, "version": record.version, "status": record.status, "installed": record.installed, "active": record.active, "description": "", "dependencies": [], "events": [], "migrations": [], "permissions": [], } ) return plugins_list # ── Internal Helpers ── async def _get_plugin_record(self, db: AsyncSession, name: str) -> PluginModel | None: """Fetch a plugin record from DB by name.""" result = await db.execute(select(PluginModel).where(PluginModel.name == name)) return result.scalar_one_or_none() # Global registry instance _registry: PluginRegistry | None = None def get_registry() -> PluginRegistry: """Get the global plugin registry.""" global _registry if _registry is None: _registry = PluginRegistry() return _registry def reset_registry_for_testing() -> PluginRegistry: """Create a fresh registry for testing.""" global _registry _registry = PluginRegistry() return _registry