"""Framework entry-point detection for the code-domain symbol graph.

Problem
-------
Static call-graph analysis only records *explicit* call edges — ``A()`` inside
``B`` produces edge ``B→A``.  Framework entry points (FastAPI route handlers,
Flask views, Celery tasks, …) are wired at *runtime* through decorators and DI
containers.  No explicit call site exists in user code, so they appear dead to
:func:`~muse.cli.commands.dead.run` and have empty blast-radius in
:func:`~muse.cli.commands.impact.run`.

Solution
--------
This module adds a complementary edge type — :class:`ImplicitEntryEdge` — that
represents "this symbol is externally reachable via the framework named
``framework_id``."  Edges are synthesised by :class:`FrameworkPlugin`
implementations, one per framework, during a single linear pass over the
manifest.  The resulting :data:`ImplicitEdgeGraph` is consumed by the impact
and dead-code commands to:

* Surface entry-point annotations instead of "no callers detected".
* Exclude entry-point symbols from dead-code results.

Adding a new framework
----------------------
1. Create a class that satisfies the :class:`FrameworkPlugin` protocol.
2. Append an instance to :data:`BUILTIN_PLUGINS`.
3. Write tests in ``tests/test_framework_plugins.py``.

No changes to the core graph, impact, or dead-code modules are required.

Configuration
-------------
``.muse/code_config.toml`` is the single configuration file for all code-domain
intelligence.  It has two independent sections:

``[code]`` — controls which tree-sitter language grammars are active::

    [code]
    # Restrict to Python + TypeScript + shell; all other grammars become
    # file-level fallbacks even if the grammar packages are installed.
    languages = ["python", "typescript", "tsx", "css", "bash"]

``[framework_detection]`` — controls entry-point detection::

    [framework_detection]
    auto_detect = true
    disabled_plugins = []          # e.g. ["celery"] to silence false positives

    [[framework_detection.custom_entry_points]]
    language = "Python"
    kind = "custom-rpc"
    decorator_names = ["rpc_handler", "grpc_handler"]

Both sections are optional; omitting either reproduces the zero-config defaults.
"""

from __future__ import annotations

import ast
import logging
import pathlib
import tomllib
from dataclasses import dataclass, field
from typing import Protocol, runtime_checkable

from muse.core._types import Manifest
from muse.core.object_store import read_object
from muse.core.symbol_cache import SymbolCache
from muse.core.validation import MAX_AST_BYTES
from muse.plugins.code.ast_parser import SymbolTree, parse_symbols

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Core data types
# ---------------------------------------------------------------------------

#: String-keyed metadata dict for per-framework entry-point annotations.
#: Keys vary by framework (e.g. "method", "path" for HTTP; "hook" for Flask).
type _Metadata = dict[str, str]


@dataclass(frozen=True, slots=True)
class ImplicitEntryEdge:
    """One framework-implied reachability edge for a symbol.

    Attributes:
        framework_id:    Stable lowercase identifier, e.g. ``"fastapi"``.
        symbol_address:  Full symbol address, e.g.
                         ``"app/routers/runs.py::create_run"``.
        kind:            Semantic category of the entry point:
                         ``"http-route"``, ``"task"``, ``"cli-command"``, …
        metadata:        Arbitrary per-framework annotations.  Common keys:
                         ``method`` (HTTP verb), ``path`` (URL pattern),
                         ``queue`` (Celery queue name).
    """

    framework_id: str
    symbol_address: str
    kind: str
    metadata: _Metadata = field(default_factory=dict)


#: Maps ``symbol_address → [ImplicitEntryEdge, …]`` for every framework-wired
#: symbol in a snapshot.  Analogous to :data:`~._callgraph.ReverseGraph` but
#: for implicit framework callers rather than explicit call sites.
ImplicitEdgeGraph = dict[str, list[ImplicitEntryEdge]]


# ---------------------------------------------------------------------------
# FrameworkPlugin protocol
# ---------------------------------------------------------------------------


@runtime_checkable
class FrameworkPlugin(Protocol):
    """Protocol every framework entry-point plugin must satisfy.

    Implementations should be stateless singletons.  The contract is:

    * ``language`` — the primary language this plugin targets.  Used to skip
      files whose suffix does not match, avoiding unnecessary parses.
    * ``framework_id`` — a stable lowercase slug, e.g. ``"fastapi"``.
    * ``detect_entry_points`` — pure, thread-safe, returns an empty list when
      the file contains no framework-wired symbols.
    """

    @property
    def language(self) -> str:
        """Primary language, e.g. ``"Python"`` or ``"TypeScript"``."""
        ...

    @property
    def framework_id(self) -> str:
        """Stable lowercase slug, e.g. ``"fastapi"``."""
        ...

    def detect_entry_points(
        self,
        file_path: str,
        sym_tree: SymbolTree,
        source: bytes,
    ) -> list[ImplicitEntryEdge]:
        """Return implicit entry edges for framework-wired symbols in *file_path*.

        Args:
            file_path:  Workspace-relative POSIX path, e.g.
                        ``"app/routers/runs.py"``.
            sym_tree:   Pre-parsed symbol tree for the file.  Plugins may
                        use this to correlate decorator findings with known
                        symbol addresses without re-parsing.
            source:     Raw file bytes.  Plugins that need full AST detail
                        (e.g. decorator arguments) must call
                        ``ast.parse(source)`` themselves.

        Returns:
            A (possibly empty) list of :class:`ImplicitEntryEdge` objects.
            One edge per framework-wired symbol per entry-point type.
        """
        ...


# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------


@dataclass
class _CustomEntryPointRule:
    """One user-defined entry-point pattern from ``.muse/code_config.toml``."""

    language: str
    kind: str
    decorator_names: list[str] = field(default_factory=list)


@dataclass
class FrameworkConfig:
    """Parsed ``[framework_detection]`` section from ``.muse/code_config.toml``.

    Defaults reflect the "zero-config" happy path: all built-in plugins active,
    no custom rules.
    """

    auto_detect: bool = True
    disabled_plugins: frozenset[str] = field(default_factory=frozenset)
    custom_entry_points: list[_CustomEntryPointRule] = field(default_factory=list)


def load_framework_config(root: pathlib.Path) -> FrameworkConfig:
    """Read ``[framework_detection]`` from ``.muse/code_config.toml``.

    Returns default :class:`FrameworkConfig` if the file is absent or the
    section is missing — callers should not need to handle ``None``.
    """
    config_path = root / ".muse" / "code_config.toml"
    if not config_path.exists():
        return FrameworkConfig()

    try:
        with open(config_path, "rb") as fh:
            raw = tomllib.load(fh)
    except Exception as exc:  # noqa: BLE001
        logger.warning("⚠️  Could not parse .muse/code_config.toml: %s", exc)
        return FrameworkConfig()

    section = raw.get("framework_detection", {})
    if not isinstance(section, dict):
        return FrameworkConfig()

    auto_detect: bool = bool(section.get("auto_detect", True))
    disabled: list[str] = section.get("disabled_plugins", [])
    custom: list[_CustomEntryPointRule] = []
    for rule in section.get("custom_entry_points", []):
        if not isinstance(rule, dict):
            continue
        lang = rule.get("language", "Python")
        kind = rule.get("kind", "custom")
        names = rule.get("decorator_names", [])
        if isinstance(names, list):
            custom.append(_CustomEntryPointRule(language=lang, kind=kind, decorator_names=names))

    return FrameworkConfig(
        auto_detect=auto_detect,
        disabled_plugins=frozenset(disabled),
        custom_entry_points=custom,
    )


# ---------------------------------------------------------------------------
# Custom pattern plugin (reads user-defined rules from FrameworkConfig)
# ---------------------------------------------------------------------------


class _CustomPatternPlugin:
    """Synthesises entry edges from ``[[framework_detection.custom_entry_points]]`` rules."""

    language = "Python"
    framework_id = "custom"

    def __init__(self, rules: list[_CustomEntryPointRule]) -> None:
        self._rules = [r for r in rules if r.language == "Python"]

    def detect_entry_points(
        self,
        file_path: str,
        sym_tree: SymbolTree,
        source: bytes,
    ) -> list[ImplicitEntryEdge]:
        if not self._rules or len(source) > MAX_AST_BYTES:
            return []
        try:
            tree = ast.parse(source)
        except SyntaxError:
            return []

        decorator_to_kind = {}  # bare_name → kind string
        for rule in self._rules:
            for name in rule.decorator_names:
                decorator_to_kind[name] = rule.kind

        if not decorator_to_kind:
            return []

        edges: list[ImplicitEntryEdge] = []
        for node in ast.walk(tree):
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            for dec in node.decorator_list:
                dec_name = _decorator_bare_name(dec)
                if dec_name and dec_name in decorator_to_kind:
                    addr = _resolve_address(file_path, node.name, sym_tree)
                    if addr:
                        edges.append(ImplicitEntryEdge(
                            framework_id=self.framework_id,
                            symbol_address=addr,
                            kind=decorator_to_kind[dec_name],
                            metadata={"decorator": dec_name},
                        ))
        return edges


# ---------------------------------------------------------------------------
# Built-in plugins
# ---------------------------------------------------------------------------


class _FastAPIPlugin:
    """Detects FastAPI / APIRouter route handlers.

    Recognises the decorator forms::

        @app.get("/path")
        @router.post("/path", ...)
        @api_router.put("/path")

    Any variable name is accepted as the router/app object; only the HTTP
    *method attribute* is checked (``get``, ``post``, ``put``, ``delete``,
    ``patch``, ``head``, ``options``, ``trace``).
    """

    language = "Python"
    framework_id = "fastapi"

    _HTTP_METHODS: frozenset[str] = frozenset(
        {"get", "post", "put", "delete", "patch", "head", "options", "trace"}
    )

    def detect_entry_points(
        self,
        file_path: str,
        sym_tree: SymbolTree,
        source: bytes,
    ) -> list[ImplicitEntryEdge]:
        if len(source) > MAX_AST_BYTES:
            return []
        try:
            tree = ast.parse(source)
        except SyntaxError:
            return []

        edges: list[ImplicitEntryEdge] = []
        for node in ast.walk(tree):
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            for dec in node.decorator_list:
                edge = self._edge_from_decorator(dec, node.name, file_path, sym_tree)
                if edge is not None:
                    edges.append(edge)
        return edges

    def _edge_from_decorator(
        self,
        dec: ast.expr,
        func_name: str,
        file_path: str,
        sym_tree: SymbolTree,
    ) -> ImplicitEntryEdge | None:
        # Must be a Call: @router.get("/path") or @router.get("/path", ...)
        if not isinstance(dec, ast.Call):
            return None
        func = dec.func
        # Must be an attribute: <anything>.get / .post / etc.
        if not isinstance(func, ast.Attribute):
            return None
        method = func.attr.lower()
        if method not in self._HTTP_METHODS:
            return None

        # Extract the path from the first positional argument (optional).
        path = ""
        if dec.args and isinstance(dec.args[0], ast.Constant):
            path = str(dec.args[0].value)

        addr = _resolve_address(file_path, func_name, sym_tree)
        if addr is None:
            return None

        return ImplicitEntryEdge(
            framework_id=self.framework_id,
            symbol_address=addr,
            kind="http-route",
            metadata={"method": method.upper(), "path": path},
        )


class _FlaskPlugin:
    """Detects Flask and Blueprint route handlers and lifecycle hooks.

    Recognises::

        @app.route("/path", methods=["GET", "POST"])
        @bp.route("/path")
        @app.before_request / @app.after_request / @app.teardown_appcontext
        @app.errorhandler(404)
        @app.cli.command()
    """

    language = "Python"
    framework_id = "flask"

    _ROUTE_ATTRS: frozenset[str] = frozenset({"route"})
    _LIFECYCLE_ATTRS: frozenset[str] = frozenset(
        {"before_request", "after_request", "teardown_appcontext",
         "before_app_request", "after_app_request"}
    )
    _ERROR_ATTRS: frozenset[str] = frozenset({"errorhandler"})

    def detect_entry_points(
        self,
        file_path: str,
        sym_tree: SymbolTree,
        source: bytes,
    ) -> list[ImplicitEntryEdge]:
        if len(source) > MAX_AST_BYTES:
            return []
        try:
            tree = ast.parse(source)
        except SyntaxError:
            return []

        edges: list[ImplicitEntryEdge] = []
        for node in ast.walk(tree):
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            for dec in node.decorator_list:
                edge = self._edge_from_decorator(dec, node.name, file_path, sym_tree)
                if edge is not None:
                    edges.append(edge)
        return edges

    def _edge_from_decorator(
        self,
        dec: ast.expr,
        func_name: str,
        file_path: str,
        sym_tree: SymbolTree,
    ) -> ImplicitEntryEdge | None:
        addr = _resolve_address(file_path, func_name, sym_tree)
        if addr is None:
            return None

        # @app.route("/path", methods=[...])
        if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute):
            attr = dec.func.attr
            if attr in self._ROUTE_ATTRS:
                path = ""
                if dec.args and isinstance(dec.args[0], ast.Constant):
                    path = str(dec.args[0].value)
                methods = _extract_methods_kwarg(dec) or ["GET"]
                return ImplicitEntryEdge(
                    framework_id=self.framework_id,
                    symbol_address=addr,
                    kind="http-route",
                    metadata={"method": ",".join(methods), "path": path},
                )
            if attr in self._LIFECYCLE_ATTRS:
                return ImplicitEntryEdge(
                    framework_id=self.framework_id,
                    symbol_address=addr,
                    kind="lifecycle-hook",
                    metadata={"hook": attr},
                )
            if attr in self._ERROR_ATTRS:
                return ImplicitEntryEdge(
                    framework_id=self.framework_id,
                    symbol_address=addr,
                    kind="error-handler",
                    metadata={},
                )

        # Bare @app.before_request (no call parens)
        if isinstance(dec, ast.Attribute) and dec.attr in self._LIFECYCLE_ATTRS:
            return ImplicitEntryEdge(
                framework_id=self.framework_id,
                symbol_address=addr,
                kind="lifecycle-hook",
                metadata={"hook": dec.attr},
            )

        return None


class _CeleryPlugin:
    """Detects Celery task functions.

    Recognises::

        @app.task
        @celery.task
        @shared_task
        @app.task(bind=True)
        @app.periodic_task(run_every=…)
    """

    language = "Python"
    framework_id = "celery"

    _TASK_BARE: frozenset[str] = frozenset({"shared_task"})
    _TASK_ATTRS: frozenset[str] = frozenset({"task", "periodic_task"})

    def detect_entry_points(
        self,
        file_path: str,
        sym_tree: SymbolTree,
        source: bytes,
    ) -> list[ImplicitEntryEdge]:
        if len(source) > MAX_AST_BYTES:
            return []
        try:
            tree = ast.parse(source)
        except SyntaxError:
            return []

        edges: list[ImplicitEntryEdge] = []
        for node in ast.walk(tree):
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            for dec in node.decorator_list:
                edge = self._edge_from_decorator(dec, node.name, file_path, sym_tree)
                if edge is not None:
                    edges.append(edge)
        return edges

    def _edge_from_decorator(
        self,
        dec: ast.expr,
        func_name: str,
        file_path: str,
        sym_tree: SymbolTree,
    ) -> ImplicitEntryEdge | None:
        addr = _resolve_address(file_path, func_name, sym_tree)
        if addr is None:
            return None

        # @shared_task (bare Name)
        if isinstance(dec, ast.Name) and dec.id in self._TASK_BARE:
            return ImplicitEntryEdge(
                framework_id=self.framework_id,
                symbol_address=addr,
                kind="task",
                metadata={},
            )

        # @shared_task(...) (called Name)
        if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Name):
            if dec.func.id in self._TASK_BARE:
                return ImplicitEntryEdge(
                    framework_id=self.framework_id,
                    symbol_address=addr,
                    kind="task",
                    metadata={},
                )

        # @app.task / @celery.task (bare Attribute)
        if isinstance(dec, ast.Attribute) and dec.attr in self._TASK_ATTRS:
            kind = "periodic-task" if dec.attr == "periodic_task" else "task"
            return ImplicitEntryEdge(
                framework_id=self.framework_id,
                symbol_address=addr,
                kind=kind,
                metadata={},
            )

        # @app.task(...) / @app.periodic_task(...) (called Attribute)
        if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute):
            if dec.func.attr in self._TASK_ATTRS:
                kind = "periodic-task" if dec.func.attr == "periodic_task" else "task"
                return ImplicitEntryEdge(
                    framework_id=self.framework_id,
                    symbol_address=addr,
                    kind=kind,
                    metadata={},
                )

        return None


#: Default plugin instances — extend this list to add new frameworks.
BUILTIN_PLUGINS: list[FrameworkPlugin] = [
    _FastAPIPlugin(),
    _FlaskPlugin(),
    _CeleryPlugin(),
]


# ---------------------------------------------------------------------------
# Graph construction
# ---------------------------------------------------------------------------


_PY_SUFFIXES: frozenset[str] = frozenset({".py", ".pyi"})


def build_implicit_edge_graph(
    root: pathlib.Path,
    manifest: Manifest,
    cache: SymbolCache | None = None,
    config: FrameworkConfig | None = None,
) -> ImplicitEdgeGraph:
    """Build the implicit entry-edge graph from *manifest*.

    Runs all active :data:`BUILTIN_PLUGINS` (plus any custom rules from
    *config*) over every file in the manifest and aggregates the results.

    Args:
        root:     Repository root — used to read blobs from the object store.
        manifest: Snapshot manifest mapping file path → SHA-256 object ID.
        cache:    Optional :class:`~muse.core.symbol_cache.SymbolCache`.
                  When provided, symbol trees are served from cache,
                  avoiding redundant parses.
        config:   Framework configuration.  When ``None``,
                  :func:`load_framework_config` is called with *root*.

    Returns:
        :data:`ImplicitEdgeGraph` — ``{symbol_address: [ImplicitEntryEdge, …]}``.
        Symbols with no implicit callers are absent from the mapping.
    """
    if config is None:
        config = load_framework_config(root)

    if not config.auto_detect:
        return {}

    active_plugins: list[FrameworkPlugin] = [
        p for p in BUILTIN_PLUGINS
        if p.framework_id not in config.disabled_plugins
    ]
    if config.custom_entry_points:
        active_plugins.append(_CustomPatternPlugin(config.custom_entry_points))

    if not active_plugins:
        return {}

    # Partition plugins by language so we only run Python plugins on .py files.
    py_plugins = [p for p in active_plugins if p.language == "Python"]

    graph: ImplicitEdgeGraph = {}

    for file_path, obj_id in manifest.items():
        suffix = pathlib.PurePosixPath(file_path).suffix.lower()

        plugins_for_file = py_plugins if suffix in _PY_SUFFIXES else []
        if not plugins_for_file:
            continue

        raw = read_object(root, obj_id)
        if raw is None or len(raw) > MAX_AST_BYTES:
            continue

        sym_tree: SymbolTree = (
            cache.get(obj_id) or parse_symbols(raw, file_path)
            if cache is not None
            else parse_symbols(raw, file_path)
        )

        for plugin in plugins_for_file:
            try:
                edges = plugin.detect_entry_points(file_path, sym_tree, raw)
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "⚠️  %s plugin raised on %s: %s",
                    plugin.framework_id, file_path, exc,
                )
                continue
            for edge in edges:
                graph.setdefault(edge.symbol_address, []).append(edge)

    return graph


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------


def _resolve_address(
    file_path: str,
    func_name: str,
    sym_tree: SymbolTree,
) -> str | None:
    """Return the full symbol address for *func_name* within *file_path*.

    Tries exact match first (``file::name``), then a scan for method forms
    (``file::ClassName.name``).  Returns ``None`` if the name is not in the
    symbol tree — this prevents synthesising edges for lambdas or inline
    functions that the AST parser does not index.
    """
    direct = f"{file_path}::{func_name}"
    if direct in sym_tree:
        return direct
    # Method match: "file::SomeClass.func_name"
    for addr in sym_tree:
        if addr.endswith(f".{func_name}") and addr.startswith(file_path + "::"):
            return addr
    return None


def _decorator_bare_name(dec: ast.expr) -> str | None:
    """Return the bare decorator name for simple ``@name`` or ``@name(...)`` forms."""
    if isinstance(dec, ast.Name):
        return dec.id
    if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Name):
        return dec.func.id
    if isinstance(dec, ast.Attribute):
        return dec.attr
    if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute):
        return dec.func.attr
    return None


def _extract_methods_kwarg(call_node: ast.Call) -> list[str] | None:
    """Extract HTTP methods from a ``methods=[...]`` keyword in a decorator call."""
    for kw in call_node.keywords:
        if kw.arg == "methods" and isinstance(kw.value, ast.List):
            methods: list[str] = []
            for elt in kw.value.elts:
                if isinstance(elt, ast.Constant):
                    methods.append(str(elt.value).upper())
            return methods or None
    return None