"""muse plumbing for-each-ref — iterate all refs with rich commit metadata.

Enumerates every branch ref and emits the full commit metadata it points to.
Supports sorting by any commit field, glob-pattern filtering, and an optional
``--no-commits`` fast-path so agent pipelines can slice the ref list without
loading every commit record.

Hierarchical branch names (e.g. ``feat/my-thing``, ``bugfix/PROJ-42``) are
fully supported — the command recursively walks ``.muse/refs/heads/``.

Output (JSON, default)::

    {
      "refs": [
        {
          "ref":          "refs/heads/dev",
          "branch":       "dev",
          "commit_id":    "<sha256>",
          "author":       "gabriel",
          "message":      "Add verse melody",
          "committed_at": "2026-01-01T00:00:00+00:00",
          "snapshot_id":  "<sha256>"
        }
      ],
      "count": 1
    }

With ``--no-commits`` the ``author``, ``message``, ``committed_at``, and
``snapshot_id`` fields are omitted::

    {
      "refs": [
        {"ref": "refs/heads/dev", "branch": "dev", "commit_id": "<sha256>"}
      ],
      "count": 1
    }

Text output (``--format text``)::

    <sha256>  refs/heads/dev  2026-01-01T00:00:00+00:00  gabriel

Text output with ``--no-commits``::

    <sha256>  refs/heads/dev

Plumbing contract
-----------------

- Exit 0: refs emitted (list may be empty).
- Exit 1: unknown ``--sort`` field; bad ``--format``; negative ``--count``.
- Exit 3: I/O error reading refs or commit records.

Agent use
---------

Cheapest full ref list (skip commit I/O)::

    muse plumbing for-each-ref --no-commits --json

Latest commit on every feat/* branch (sorted newest first)::

    muse plumbing for-each-ref --pattern 'refs/heads/feat/*' \\
        --sort committed_at --desc --json

Count branches matching a pattern::

    muse plumbing for-each-ref --pattern 'refs/heads/bugfix/*' --json \\
        | python3 -c "import sys,json; print(json.load(sys.stdin)['count'])"

Get the tip commit of exactly N most-recently-committed branches::

    muse plumbing for-each-ref --sort committed_at --desc --count 5 --json
"""

from __future__ import annotations

import argparse
import fnmatch
import json
import logging
import pathlib
import sys
from typing import TypedDict

from muse.core.errors import ExitCode
from muse.core.repo import require_repo
from muse.core.store import read_commit
from muse.core.validation import sanitize_display, validate_object_id

logger = logging.getLogger(__name__)

_FORMAT_CHOICES = ("json", "text")
_SORT_FIELDS = (
    "ref",
    "branch",
    "commit_id",
    "author",
    "committed_at",
    "message",
    "snapshot_id",
)


class _RefDetail(TypedDict, total=False):
    """One ref entry.

    The ``author``, ``message``, ``committed_at``, and ``snapshot_id``
    fields are omitted when ``--no-commits`` is used.
    """

    ref: str
    branch: str
    commit_id: str
    author: str
    message: str
    committed_at: str
    snapshot_id: str


class _ForEachRefResult(TypedDict):
    refs: list[_RefDetail]
    count: int


def _list_all_refs(root: pathlib.Path) -> list[tuple[str, str]]:
    """Return sorted (branch_name, commit_id) pairs from ``.muse/refs/heads/``.

    Uses ``rglob("*")`` so hierarchical branch names (``feat/my-thing``,
    ``bugfix/PROJ-42``) are discovered correctly.  Symlinks are skipped to
    prevent path-traversal attacks via crafted symlinks in the ref store.
    Ref files whose contents are not a valid 64-char hex SHA-256 are also
    skipped with a debug log.
    """
    heads_dir = root / ".muse" / "refs" / "heads"
    if not heads_dir.exists():
        return []
    pairs: list[tuple[str, str]] = []
    for child in sorted(heads_dir.rglob("*")):
        if child.is_symlink():
            logger.debug("for-each-ref: skipping symlink ref %s", child)
            continue
        if not child.is_file():
            continue
        branch = child.relative_to(heads_dir).as_posix()
        commit_id = child.read_text(encoding="utf-8").strip()
        try:
            validate_object_id(commit_id)
        except ValueError:
            logger.debug(
                "for-each-ref: skipping ref %s — invalid commit ID %r",
                branch,
                commit_id[:20],
            )
            continue
        pairs.append((branch, commit_id))
    return pairs


def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
    """Register the for-each-ref subcommand."""
    parser = subparsers.add_parser(
        "for-each-ref",
        help="Iterate all refs with rich commit metadata.",
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "--pattern", "-p",
        default=None,
        dest="pattern",
        metavar="GLOB",
        help=(
            "fnmatch glob filter applied to the full ref name "
            "(e.g. 'refs/heads/feat/*').  Omit to include all refs."
        ),
    )
    parser.add_argument(
        "--sort", "-s",
        default="ref",
        dest="sort_by",
        metavar="FIELD",
        help=f"Field to sort by. One of: {', '.join(_SORT_FIELDS)}. (default: ref)",
    )
    parser.add_argument(
        "--desc", "-d",
        action="store_true",
        dest="descending",
        help="Reverse the sort order (descending).",
    )
    parser.add_argument(
        "--count", "-n",
        type=int,
        default=0,
        dest="count_limit",
        metavar="N",
        help="Limit output to the first N refs after sorting (0 = unlimited).",
    )
    parser.add_argument(
        "--no-commits",
        action="store_true",
        dest="no_commits",
        help=(
            "Skip loading commit records.  Emits only ``ref``, ``branch``, "
            "and ``commit_id`` fields.  Significantly faster on large repos "
            "when full commit metadata is not needed."
        ),
    )
    parser.add_argument(
        "--format", "-f",
        dest="fmt",
        default="json",
        metavar="FORMAT",
        help="Output format: json or text. (default: json)",
    )
    parser.add_argument(
        "--json", action="store_const", const="json", dest="fmt",
        help="Shorthand for --format json.",
    )
    parser.set_defaults(func=run)


def run(args: argparse.Namespace) -> None:
    """Iterate all branch refs with full commit metadata.

    Emits each branch ref together with the commit it points to, including
    the author, message, timestamp, and snapshot ID.  Pass ``--no-commits``
    to skip commit record loading for a fast bulk ref enumeration.
    """
    fmt: str = args.fmt
    pattern: str | None = args.pattern
    sort_by: str = args.sort_by
    descending: bool = args.descending
    count_limit: int = args.count_limit
    no_commits: bool = args.no_commits

    if fmt not in _FORMAT_CHOICES:
        print(
            json.dumps(
                {"error": f"Unknown format {fmt!r}. Valid: {', '.join(_FORMAT_CHOICES)}"}
            ),
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    if sort_by not in _SORT_FIELDS:
        print(
            json.dumps({
                "error": (
                    f"Unknown sort field {sort_by!r}. "
                    f"Valid: {', '.join(_SORT_FIELDS)}"
                )
            }),
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    if count_limit < 0:
        print(
            json.dumps({"error": f"--count must be >= 0, got {count_limit}"}),
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    # --no-commits + sorting by commit-only fields is contradictory.
    _commit_only_fields = {"author", "message", "committed_at", "snapshot_id"}
    if no_commits and sort_by in _commit_only_fields:
        print(
            json.dumps({
                "error": (
                    f"Cannot sort by {sort_by!r} with --no-commits "
                    "(field is not loaded). Use a ref-level field: "
                    "ref, branch, commit_id."
                )
            }),
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    root = require_repo()

    try:
        pairs = _list_all_refs(root)
    except OSError as exc:
        logger.debug("for-each-ref I/O error listing refs: %s", exc)
        print(json.dumps({"error": str(exc)}), file=sys.stderr)
        raise SystemExit(ExitCode.INTERNAL_ERROR)

    # Apply glob filter.
    if pattern is not None:
        pairs = [
            (b, c) for b, c in pairs if fnmatch.fnmatch(f"refs/heads/{b}", pattern)
        ]

    # Build detailed ref list.
    details: list[_RefDetail] = []
    for branch, commit_id in pairs:
        if no_commits:
            details.append(
                _RefDetail(
                    ref=f"refs/heads/{branch}",
                    branch=branch,
                    commit_id=commit_id,
                )
            )
            continue

        record = None
        try:
            record = read_commit(root, commit_id)
        except (OSError, ValueError, KeyError) as exc:
            logger.debug(
                "for-each-ref: cannot read commit %s: %s", commit_id[:12], exc
            )

        if record is None:
            details.append(
                _RefDetail(
                    ref=f"refs/heads/{branch}",
                    branch=branch,
                    commit_id=commit_id,
                    author="",
                    message="(commit record missing)",
                    committed_at="",
                    snapshot_id="",
                )
            )
        else:
            details.append(
                _RefDetail(
                    ref=f"refs/heads/{branch}",
                    branch=branch,
                    commit_id=commit_id,
                    author=record.author,
                    message=record.message,
                    committed_at=record.committed_at.isoformat(),
                    snapshot_id=record.snapshot_id,
                )
            )

    # Sort — explicit dispatcher avoids TypedDict key constraint on subscript.
    def _sort_key(d: _RefDetail) -> str:
        if sort_by == "branch":
            return d.get("branch", "")
        if sort_by == "commit_id":
            return d.get("commit_id", "")
        if sort_by == "author":
            return d.get("author", "")
        if sort_by == "committed_at":
            return d.get("committed_at", "")
        if sort_by == "message":
            return d.get("message", "")
        if sort_by == "snapshot_id":
            return d.get("snapshot_id", "")
        return d.get("ref", "")

    details.sort(key=_sort_key, reverse=descending)

    # Limit.
    if count_limit > 0:
        details = details[:count_limit]

    if fmt == "text":
        for d in details:
            if no_commits:
                print(
                    f"{sanitize_display(d.get('commit_id', ''))}  "
                    f"{sanitize_display(d.get('ref', ''))}"
                )
            else:
                print(
                    f"{sanitize_display(d.get('commit_id', ''))}  "
                    f"{sanitize_display(d.get('ref', ''))}  "
                    f"{sanitize_display(d.get('committed_at', ''))}  "
                    f"{sanitize_display(d.get('author', ''))}"
                )
        return

    result: _ForEachRefResult = {"refs": details, "count": len(details)}
    print(json.dumps(result))