plugins/manager.py

"""Plugin manager — loads, registers, and manages plugins."""

import json
import logging
from typing import Optional
from pathlib import Path

from config import config
from database.repositories import PluginRepo

logger = logging.getLogger("synapse.plugins")


class PluginAPI:
    """API surface exposed to plugins."""

    def __init__(self):
        self.tools: dict = {}
        self.hooks: dict = {}

    def register_tool(self, name: str, description: str, handler, parameters: dict = None):
        self.tools[name] = {
            "name": name,
            "description": description,
            "handler": handler,
            "parameters": parameters or {},
        }
        logger.info(f"Plugin registered tool: {name}")

    def register_hook(self, event: str, handler):
        if event not in self.hooks:
            self.hooks[event] = []
        self.hooks[event].append(handler)
        logger.info(f"Plugin registered hook: {event}")

    async def trigger_hook(self, event: str, **kwargs):
        results = []
        for handler in self.hooks.get(event, []):
            try:
                if hasattr(handler, '__call__'):
                    result = handler(**kwargs)
                    results.append(result)
            except Exception as e:
                logger.error(f"Hook {event} error: {e}")
        return results


class PluginManager:
    """Manages plugin lifecycle."""

    def __init__(self):
        self.api = PluginAPI()
        self._loaded_plugins: dict[str, object] = {}

    async def load_enabled_plugins(self):
        plugins = await PluginRepo.get_enabled()
        for plugin in plugins:
            await self._load_plugin(plugin)

    async def _load_plugin(self, plugin_data: dict):
        plugin_id = plugin_data["id"]
        name = plugin_data["name"]
        try:
            plugin_dir = Path(f"plugins/builtin/{name}")
            if plugin_dir.exists():
                import importlib.util
                spec = importlib.util.spec_from_file_location(
                    f"plugins.builtin.{name}", plugin_dir / "main.py"
                )
                if spec and spec.loader:
                    module = importlib.util.module_from_spec(spec)
                    spec.loader.exec_module(module)
                    if hasattr(module, "register"):
                        module.register(self.api)
                    self._loaded_plugins[plugin_id] = module
                    logger.info(f"Loaded plugin: {name}")
        except Exception as e:
            logger.error(f"Failed to load plugin {name}: {e}")

    async def install_plugin(self, name: str, description: str = "",
                             author: str = "") -> dict:
        from database.repositories import PluginRepo
        plugin_id = PluginRepo.new_id()
        from database import engine as db_engine
        await db_engine.execute(
            "INSERT INTO plugins (id, name, description, author) VALUES (?, ?, ?, ?)",
            (plugin_id, name, description, author),
        )
        return {"id": plugin_id, "name": name}

    async def toggle_plugin(self, plugin_id: str) -> dict:
        await PluginRepo.toggle(plugin_id)
        plugin = await PluginRepo.get_by_id(plugin_id)
        return plugin

    async def get_all_plugins(self) -> list[dict]:
        from database import engine as db_engine
        return await db_engine.fetch_all("SELECT * FROM plugins ORDER BY name")

    def get_registered_tools(self) -> list[dict]:
        return [
            {"name": t["name"], "description": t["description"], "parameters": t["parameters"]}
            for t in self.api.tools.values()
        ]


plugin_manager = PluginManager()