"""muse code impact — transitive blast-radius analysis.

Answers the question every engineer asks before touching a function:
*"If I change this, what else could break?"*

``muse code impact`` builds the reverse call graph for the committed snapshot,
then performs a BFS from the target symbol's bare name through every caller,
then every caller's callers, until the full transitive closure is reached.

The result is a depth-ordered blast-radius map: depth 1 = direct callers,
depth 2 = callers of callers, and so on.  This tells you exactly how far a
change propagates through the codebase.

With ``--compare REF``, the command diffs the blast radius between two commits,
showing exactly which callers were *added* (new risk) and which were *removed*
(coupling reduced) — the ideal pre-merge signal for any proposal that touches a
widely-called symbol.

With ``--forward``, the direction reverses: instead of "who calls this?", it
answers "what does this call?" — the full transitive dependency fan-out of the
target symbol.

This is structurally impossible in Git.  Git stores files as blobs — it has
no concept of call relationships between functions.  You would need an
external static-analysis tool and a separate dependency graph.  In Muse,
the symbol graph is a first-class citizen of every committed snapshot.

Usage::

    muse code impact "src/billing.py::compute_invoice_total"
    muse code impact "src/billing.py::compute_invoice_total" --depth 2
    muse code impact "src/auth.py::validate_token" --commit HEAD~5
    muse code impact "src/core.py::content_hash" --json
    muse code impact "src/billing.py::process_order" --compare HEAD~10
    muse code impact "src/billing.py::process_order" --forward --depth 3
    muse code impact "src/api.py::handle_request" --file src/billing.py
    muse code impact "src/core.py::content_hash" --count

Output::

    Impact analysis: src/billing.py::compute_invoice_total
    ──────────────────────────────────────────────────────────────

    Depth 1 — direct callers (2):
      src/api.py::create_invoice
      src/billing.py::process_order

    Depth 2 — callers of callers (1):
      src/api.py::handle_request

    ──────────────────────────────────────────────────────────────
    Total blast radius: 3 symbols across 2 files
    🔴 High impact — add tests before changing this symbol.

Flags:

``--depth, -d N``
    Stop BFS after N levels (default: 0 = unlimited).

``--commit, -c REF``
    Analyse a historical snapshot instead of HEAD.

``--compare REF``
    Diff blast radius between HEAD (or ``--commit``) and REF.
    Shows added callers (new risk) and removed callers (reduced coupling).

``--forward``
    Show callees (dependencies) instead of callers.  Answers: "what does
    this symbol depend on?" rather than "what depends on this symbol?"

``--file PATH``
    Restrict the blast-radius output to callers from this file only.

``--count``
    Print only the total count of callers (scriptable).

``--json``
    Emit the full blast-radius map as JSON.
"""

from __future__ import annotations

import argparse
import json
import logging
import pathlib
import sys

from muse.core.errors import ExitCode
from muse.core.repo import read_repo_id, require_repo
from typing import TypedDict

from muse.core.store import (
    CommitRecord,
    get_commit_snapshot_manifest,
    read_current_branch,
    resolve_commit_ref,
)
from muse.core.symbol_cache import load_symbol_cache
from muse.plugins.registry import read_domain
from muse.plugins.code._callgraph import (
    ForwardGraph,
    ReverseGraph,
    build_forward_graph,
    build_reverse_graph,
    transitive_callees,
    transitive_callers,
)
from muse.plugins.code._framework import (
    ImplicitEntryEdge,
    ImplicitEdgeGraph,
    build_implicit_edge_graph,
)
from muse.plugins.code._query import language_of
from muse.core.validation import clamp_int, sanitize_display


type _BlastRadius = dict[str, list[str]]
type _StrMeta = dict[str, str]
logger = logging.getLogger(__name__)


class _EntryPointJson(TypedDict):
    """One framework entry-point edge in JSON output."""

    framework_id: str
    kind: str
    metadata: _StrMeta


class _ImpactJson(TypedDict, total=False):
    """JSON payload for ``muse code impact`` output."""

    mode: str
    address: str
    target_name: str
    commit_id: str
    depth_limit: int
    file_filter: str | None
    blast_radius: _BlastRadius
    total: int
    entry_points: list[_EntryPointJson]
    compare_commit_id: str
    added_callers: list[str]
    removed_callers: list[str]
    net_change: int


def _flat_addrs(blast: dict[int, list[str]]) -> set[str]:
    """Return the flat set of all addresses across all depths."""
    return {addr for addrs in blast.values() for addr in addrs}


# ---------------------------------------------------------------------------
# Knowtation domain path
# ---------------------------------------------------------------------------


def _run_knowtation_impact(
    *,
    root: pathlib.Path,
    address: str,
    forward: bool,
    depth: int,
    count_only: bool,
    as_json: bool,
    include_attachments: bool = True,
) -> None:
    """Knowtation-domain path for ``muse code impact``.

    Treats *address* as a vault-relative note path and walks the link graph
    transitively.  Default direction is reverse BFS (notes that depend on the
    target); pass ``forward=True`` for forward BFS (notes the target links
    to).

    Args:
        root:                Repository root.
        address:             Vault-relative note path.
        forward:             Forward BFS instead of reverse.
        depth:               BFS depth cap (``<= 0`` interpreted as unlimited).
        count_only:          Emit only the total count of notes.
        as_json:             Emit JSON instead of text.
        include_attachments: Honour the include_attachments flag.
    """
    from muse.plugins.knowtation.code_intel import note_impact
    from muse.plugins.knowtation.link_index import build_link_index
    from muse.plugins.knowtation._query import is_note

    if not is_note(address):
        print(
            f"❌  '{address}' is not a Markdown note.  "
            "Knowtation impact requires a .md / .markdown / .mdx path.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    candidate = root / address
    if not candidate.is_file():
        print(f"❌  Note '{address}' not found in vault.", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    try:
        index = build_link_index(root)
    except ValueError as exc:
        print(f"❌  Could not index vault: {exc}", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    bfs_depth = depth if depth > 0 else 0
    result = note_impact(
        index,
        address,
        forward=forward,
        depth=bfs_depth,
        include_attachments=include_attachments,
    )

    if count_only:
        print(result["total"])
        return

    if as_json:
        result["domain"] = "knowtation"
        print(json.dumps(result, indent=2))
        return

    direction = "Forward dependencies of" if forward else "Blast radius (callers) of"
    print(f"\n{direction}  {sanitize_display(address)}")
    print("─" * 62)
    if not result["by_depth"]:
        msg = "no outgoing links" if forward else "no incoming links"
        print(f"  ({msg})")
    else:
        for d_str in sorted(result["by_depth"], key=int):
            notes = result["by_depth"][d_str]
            print(f"\n  depth {d_str}:")
            for note in notes:
                print(f"    {sanitize_display(note)}")
    print(f"\n{result['total']} note(s) across {len(result['by_depth'])} depth(s)")


def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
    """Register the impact subcommand."""
    parser = subparsers.add_parser(
        "impact",
        help="Show the transitive blast-radius of changing a symbol.",
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "address", metavar="ADDRESS",
        help='Symbol address, e.g. "src/billing.py::compute_invoice_total".',
    )
    parser.add_argument(
        "--depth", "-d", type=int, default=None, metavar="N",
        help="Maximum BFS depth (default: 5, max: 50). Use --transitive for full traversal.",
    )
    parser.add_argument(
        "--commit", "-c", default=None, metavar="REF", dest="ref",
        help="Analyse a historical snapshot instead of HEAD.",
    )
    parser.add_argument(
        "--compare", default=None, metavar="REF", dest="compare_ref",
        help="Diff blast radius against this commit reference.",
    )
    parser.add_argument(
        "--forward", action="store_true", dest="forward",
        help="Show callees (dependencies) instead of callers.",
    )
    parser.add_argument(
        "--file", "-f", default=None, metavar="PATH", dest="file_filter",
        help="Restrict blast-radius output to callers from this file.",
    )
    parser.add_argument(
        "--count", action="store_true", dest="count_only",
        help="Print only the total count of matching symbols.",
    )
    parser.add_argument(
        "--json", action="store_true", dest="as_json",
        help="Emit results as JSON.",
    )
    parser.add_argument(
        "--include-attachments",
        action=argparse.BooleanOptionalAction,
        default=True,
        dest="include_attachments",
        help=(
            "Knowtation only: follow mist-attachment co-references during impact "
            "(default on).  Use --no-include-attachments to disable."
        ),
    )
    parser.set_defaults(func=run)


def run(args: argparse.Namespace) -> None:
    """Show the transitive blast-radius of changing a symbol.

    Builds the reverse call graph for the committed snapshot, then BFS-walks
    it from the target symbol outwards.  Depth 1 = direct callers; depth 2 =
    callers of callers; and so on until no new callers are found.

    With ``--compare`` the blast radius is diffed against another commit,
    revealing exactly which callers were added or removed.

    With ``--forward`` the direction inverts: depth 1 = direct callees.

    Python only (call-graph analysis uses stdlib ``ast``).
    """
    address: str = args.address
    _depth_raw: int | None = args.depth
    depth: int = clamp_int(_depth_raw, 1, 50, "depth") if _depth_raw is not None else 5
    ref: str | None = args.ref
    compare_ref: str | None = args.compare_ref
    forward: bool = args.forward
    file_filter: str | None = args.file_filter
    count_only: bool = args.count_only
    as_json: bool = args.as_json

    if forward and compare_ref:
        print("❌ --forward and --compare are mutually exclusive.", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    root = require_repo()
    repo_id = read_repo_id(root)
    branch = read_current_branch(root)

    try:
        active_domain = read_domain(root)
    except Exception:
        active_domain = "code"

    if active_domain == "knowtation":
        _run_knowtation_impact(
            root=root,
            address=address,
            forward=forward,
            depth=depth,
            count_only=count_only,
            as_json=as_json,
            include_attachments=getattr(args, "include_attachments", True),
        )
        return

    lang = language_of(address.split("::")[0]) if "::" in address else ""
    if lang and lang != "Python":
        print(
            f"⚠️  Impact analysis is currently Python-only.  '{address}' is {lang}.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    commit = resolve_commit_ref(root, repo_id, branch, ref)
    if commit is None:
        print(f"❌ Commit '{ref or 'HEAD'}' not found.", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {}

    cache = load_symbol_cache(root)

    target_name = address.split("::")[-1].split(".")[-1] if "::" in address else address

    if forward:
        fwd: ForwardGraph = build_forward_graph(root, manifest, cache=cache)
        callee_map = transitive_callees(address, fwd, max_depth=depth)
        cache.save()
        _render_forward(
            address=address,
            target_name=target_name,
            commit=commit,
            callee_map=callee_map,
            depth_limit=depth,
            count_only=count_only,
            as_json=as_json,
        )
        return

    rev: ReverseGraph = build_reverse_graph(root, manifest, cache=cache)
    implicit: ImplicitEdgeGraph = build_implicit_edge_graph(root, manifest, cache=cache)
    blast = transitive_callers(target_name, rev, max_depth=depth)

    if file_filter:
        blast = {
            d: [a for a in addrs if "::" in a and a.split("::")[0] == file_filter]
            for d, addrs in blast.items()
        }
        blast = {d: addrs for d, addrs in blast.items() if addrs}

    compare_commit: CommitRecord | None = None
    added: set[str] = set()
    removed: set[str] = set()

    if compare_ref:
        compare_commit = resolve_commit_ref(root, repo_id, branch, compare_ref)
        if compare_commit is None:
            print(f"❌ --compare commit '{compare_ref}' not found.", file=sys.stderr)
            raise SystemExit(ExitCode.USER_ERROR)
        compare_manifest = get_commit_snapshot_manifest(root, compare_commit.commit_id) or {}
        compare_rev: ReverseGraph = build_reverse_graph(root, compare_manifest, cache=cache)
        compare_blast = transitive_callers(target_name, compare_rev, max_depth=depth)

        if file_filter:
            compare_blast = {
                d: [a for a in addrs if "::" in a and a.split("::")[0] == file_filter]
                for d, addrs in compare_blast.items()
            }
            compare_blast = {d: addrs for d, addrs in compare_blast.items() if addrs}

        current_flat = _flat_addrs(blast)
        compare_flat = _flat_addrs(compare_blast)
        added = current_flat - compare_flat
        removed = compare_flat - current_flat

    cache.save()

    entry_points = implicit.get(address, [])

    _render_reverse(
        address=address,
        target_name=target_name,
        commit=commit,
        blast=blast,
        entry_points=entry_points,
        depth_limit=depth,
        file_filter=file_filter,
        compare_commit=compare_commit,
        added=added,
        removed=removed,
        count_only=count_only,
        as_json=as_json,
    )


# ---------------------------------------------------------------------------
# Renderers
# ---------------------------------------------------------------------------


def _render_forward(
    *,
    address: str,
    target_name: str,
    commit: CommitRecord,
    callee_map: dict[int, list[str]],
    depth_limit: int,
    count_only: bool,
    as_json: bool,
) -> None:
    """Render forward (callees) analysis."""
    total = sum(len(v) for v in callee_map.values())

    if count_only and not as_json:
        print(total)
        return

    if as_json:
        print(json.dumps(
            {
                "mode": "forward",
                "address": address,
                "target_name": target_name,
                "commit_id": commit.commit_id,
                "depth_limit": depth_limit,
                "callees": {str(d): names for d, names in sorted(callee_map.items())},
                "total": total,
            },
            indent=2,
        ))
        return

    print(f"\nDependency fan-out: {sanitize_display(address)}")
    print("─" * 62)

    if not callee_map:
        print(f"\n  ('{target_name}' calls nothing detectable — leaf function or dynamic dispatch)")
        return

    for d in sorted(callee_map.keys()):
        label = "direct callees" if d == 1 else f"depth-{d} callees"
        print(f"\nDepth {d} — {label} ({len(callee_map[d])}):")
        for name in sorted(callee_map[d]):
            print(f"  {sanitize_display(name)}")

    print("\n" + "─" * 62)
    print(f"Total dependency fan-out: {total} callee(s)")
    print("\nNote: analysis covers Python call-sites only.")


def _render_reverse(
    *,
    address: str,
    target_name: str,
    commit: CommitRecord,
    blast: dict[int, list[str]],
    entry_points: list[ImplicitEntryEdge],
    depth_limit: int,
    file_filter: str | None,
    compare_commit: CommitRecord | None,
    added: set[str],
    removed: set[str],
    count_only: bool,
    as_json: bool,
) -> None:
    """Render reverse (callers / blast-radius) analysis."""
    total = sum(len(v) for v in blast.values())

    if count_only and not as_json:
        print(total)
        return

    if as_json:
        payload = _ImpactJson(
            mode="reverse",
            address=address,
            target_name=target_name,
            commit_id=commit.commit_id,
            depth_limit=depth_limit,
            file_filter=file_filter,
            blast_radius={str(d): list(addrs) for d, addrs in sorted(blast.items())},
            total=total,
        )
        if entry_points:
            payload["entry_points"] = [
                _EntryPointJson(
                    framework_id=ep.framework_id,
                    kind=ep.kind,
                    metadata=ep.metadata,
                )
                for ep in entry_points
            ]
        if compare_commit is not None:
            payload["compare_commit_id"] = compare_commit.commit_id
            payload["added_callers"] = sorted(added)
            payload["removed_callers"] = sorted(removed)
            payload["net_change"] = len(added) - len(removed)
        print(json.dumps(payload, indent=2))
        return

    print(f"\nImpact analysis: {sanitize_display(address)}")
    if file_filter:
        print(f"(filtered to callers in: {file_filter})")
    print("─" * 62)

    if entry_points:
        print("\n  Framework entry point — externally reachable via:")
        for ep in entry_points:
            meta_str = "  ".join(f"{k}={v}" for k, v in ep.metadata.items() if v)
            print(f"    [{ep.framework_id}]  kind={ep.kind}  {meta_str}")

    if not blast:
        if entry_points:
            print("\n  (no explicit callers in user code — wired by the framework above)")
        else:
            print(
                f"\n  (no callers detected — '{target_name}' may be an entry point or dead code)"
            )
        print("\n  Note: analysis covers Python only; external callers are not detected.")
    else:
        all_files: set[str] = set()
        for d in sorted(blast.keys()):
            callers = blast[d]
            if d == 1:
                label = "direct callers"
            elif d == 2:
                label = "callers of callers"
            else:
                label = f"depth-{d} callers"
            print(f"\nDepth {d} — {label} ({len(callers)}):")
            for addr in sorted(callers):
                marker = " ← NEW" if addr in added else ""
                print(f"  {sanitize_display(addr)}{marker}")
                if "::" in addr:
                    all_files.add(addr.split("::")[0])

        print("\n" + "─" * 62)
        file_label = "file" if len(all_files) == 1 else "files"
        print(f"Total blast radius: {total} symbol(s) across {len(all_files)} {file_label}")
        if total >= 10:
            print("🔴 High impact — add tests before changing this symbol.")
        elif total >= 3:
            print("🟡 Medium impact — review callers before changing this symbol.")
        else:
            print("🟢 Low impact — change is well-contained.")
        print(
            "\nNote: analysis covers Python call-sites only."
            " Dynamic dispatch (getattr, decorators) is not detected."
        )

    if compare_commit is not None:
        print(f"\nBlast-radius diff vs {compare_commit.commit_id[:8]}:")
        print("─" * 62)
        if not added and not removed:
            print("  No change in blast radius.")
        else:
            net = len(added) - len(removed)
            sign = "+" if net >= 0 else ""
            print(f"  Net change: {sign}{net} caller(s)")
            if added:
                print(f"\n  Added ({len(added)}):")
                for a in sorted(added):
                    print(f"    + {a}")
            if removed:
                print(f"\n  Removed ({len(removed)}):")
                for r in sorted(removed):
                    print(f"    - {r}")