"""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()) # ── 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." ) # ── 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, ) 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 # 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 # 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