leocrm/app/core/hooks.py

"""WordPress-style hooks: actions (fire-and-forget) and filters (modify data).

Actions are fire-and-forget event callbacks with no return value.
Filters chain-modify a value through one or more callbacks, returning the result.

Usage in services::

    from app.core.hooks import do_action, apply_filters

    # Action — no return value, side effects only
    await do_action("contact.before_create", contact_data, db=db)

    # Filter — returns modified value
    display_name = await apply_filters("contact.format_display_name", contact.name)

Usage in plugins (on_activate)::

    from app.core.hooks import get_hook_registry

    async def on_activate(self, db, service_container, event_bus):
        await super().on_activate(db, service_container, event_bus)
        reg = get_hook_registry()
        reg.register_action("contact.before_create", self._on_contact_create, priority=10)
        reg.register_filter("contact.format_display_name", self._format_name, priority=10)

Priority: lower numbers run first (default=10).
"""

from __future__ import annotations

import logging
from collections import defaultdict
from collections.abc import Callable
from typing import Any

logger = logging.getLogger(__name__)


class HookRegistry:
    """Central registry for actions and filters.

    Actions: ``do_action('contact.before_create', data)`` — no return value.
    Filters: ``result = apply_filters('contact.format_name', name)`` — returns modified value.

    Priority: lower numbers run first (default=10).
    """

    _instance: HookRegistry | None = None

    def __new__(cls) -> HookRegistry:
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._actions: dict[str, list[tuple[int, Callable]]] = defaultdict(list)
            cls._instance._filters: dict[str, list[tuple[int, Callable]]] = defaultdict(list)
        return cls._instance

    # ─── Registration ───

    def register_action(self, hook_name: str, callback: Callable, priority: int = 10, owner_tag: str | None = None) -> None:
        """Register an action callback for *hook_name*.

        Args:
            owner_tag: Optional tag identifying the owning plugin. Used by
                unregister_actions_by_owner() to remove only this plugin's hooks.
        """
        self._actions[hook_name].append((priority, callback, owner_tag))
        self._actions[hook_name].sort(key=lambda x: x[0])
        logger.debug("Action registered: %s (priority=%d, owner=%s)", hook_name, priority, owner_tag)

    def register_filter(self, hook_name: str, callback: Callable, priority: int = 10, owner_tag: str | None = None) -> None:
        """Register a filter callback for *hook_name*."""
        self._filters[hook_name].append((priority, callback, owner_tag))
        self._filters[hook_name].sort(key=lambda x: x[0])
        logger.debug("Filter registered: %s (priority=%d, owner=%s)", hook_name, priority, owner_tag)

    # ─── Unregistration ───

    def unregister(self, hook_name: str, callback: Callable) -> None:
        """Remove a specific callback from both actions and filters."""
        self._actions[hook_name] = [
            (p, c, o) for p, c, o in self._actions.get(hook_name, []) if c != callback
        ]
        self._filters[hook_name] = [
            (p, c, o) for p, c, o in self._filters.get(hook_name, []) if c != callback
        ]
        if not self._actions[hook_name]:
            self._actions.pop(hook_name, None)
        if not self._filters[hook_name]:
            self._filters.pop(hook_name, None)

    def unregister_all_for_plugin(self, plugin_name: str) -> None:
        """Remove all hooks whose callback belongs to a plugin.

        This uses a heuristic: callbacks that are bound methods of a plugin
        instance have ``__self__`` whose ``manifest.name`` matches.
        Free functions are skipped (not plugin-owned).
        """
        for hook_dict in (self._actions, self._filters):
            for hook_name in list(hook_dict.keys()):
                kept: list[tuple[int, Callable]] = []
                for priority, callback, _owner in hook_dict[hook_name]:
                    owner = getattr(callback, "__self__", None)
                    plugin_manifest_name = getattr(getattr(owner, "manifest", None), "name", None)
                    if plugin_manifest_name == plugin_name:
                        logger.debug("Unregistered hook %s for plugin %s", hook_name, plugin_name)
                        continue
                    kept.append((priority, callback, _owner))
                if kept:
                    hook_dict[hook_name] = kept
                else:
                    hook_dict.pop(hook_name, None)

    def clear_actions(self, hook_name: str) -> None:
        """Remove all action callbacks for a given hook name.

        Used by plugins to unregister hooks that were registered via
        register_history_hooks() (which creates free functions, not bound methods).
        """
        self._actions.pop(hook_name, None)
        logger.debug("Cleared all actions for hook: %s", hook_name)

    def unregister_actions_by_owner(self, hook_name: str, owner_tag: str) -> None:
        """Remove only the action callbacks for *hook_name* that were registered
        with the given *owner_tag*.

        This prevents a plugin from accidentally removing another plugin's
        handlers for the same event.
        """
        callbacks = self._actions.get(hook_name, [])
        kept = [(p, c, o) for p, c, o in callbacks if o != owner_tag]
        if kept:
            self._actions[hook_name] = kept
        else:
            self._actions.pop(hook_name, None)
        logger.debug("Unregistered %d actions for hook %s owner=%s", len(callbacks) - len(kept), hook_name, owner_tag)

    # ─── Execution ───

    async def do_action(self, hook_name: str, *args: Any, **kwargs: Any) -> None:
        """Execute all action callbacks for *hook_name* in priority order."""
        for _, callback, _owner in self._actions.get(hook_name, []):
            try:
                result = callback(*args, **kwargs)
                if hasattr(result, "__await__"):
                    await result
            except Exception:
                logger.exception("Error in action %s", hook_name)

    async def apply_filters(self, hook_name: str, value: Any, *args: Any, **kwargs: Any) -> Any:
        """Pass *value* through all filter callbacks for *hook_name* in priority order."""
        for _, callback, _owner in self._filters.get(hook_name, []):
            try:
                result = callback(value, *args, **kwargs)
                if hasattr(result, "__await__"):
                    result = await result
                value = result
            except Exception:
                logger.exception("Error in filter %s", hook_name)
        return value

    # ─── Introspection ───

    def list_actions(self) -> list[str]:
        """Return all registered action hook names."""
        return sorted(self._actions.keys())

    def list_filters(self) -> list[str]:
        """Return all registered filter hook names."""
        return sorted(self._filters.keys())

    def has_action(self, hook_name: str) -> bool:
        return bool(self._actions.get(hook_name))

    def has_filter(self, hook_name: str) -> bool:
        return bool(self._filters.get(hook_name))

    # ─── Testing ───

    def _reset_for_testing(self) -> None:
        """Clear all state — for unit tests only."""
        self._actions.clear()
        self._filters.clear()


# ─── Module-level helpers ───


def get_hook_registry() -> HookRegistry:
    """Return the global :class:`HookRegistry` singleton."""
    return HookRegistry()


async def do_action(hook_name: str, *args: Any, **kwargs: Any) -> None:
    """Execute all action callbacks for *hook_name*."""
    await get_hook_registry().do_action(hook_name, *args, **kwargs)


async def apply_filters(hook_name: str, value: Any, *args: Any, **kwargs: Any) -> Any:
    """Pass *value* through all filter callbacks for *hook_name*."""
    return await get_hook_registry().apply_filters(hook_name, value, *args, **kwargs)


def reset_hook_registry_for_testing() -> HookRegistry:
    """Return a fresh singleton — for unit tests only."""
    reg = get_hook_registry()
    reg._reset_for_testing()
    return reg