"""Repository detection utilities for the Muse CLI.

Walking up the directory tree to locate a ``.muse/`` directory is the
single most-called internal primitive. Every subcommand uses it. Keeping
the semantics clear (``None`` on miss, never raises) makes callers simpler
and test isolation easier (``MUSE_REPO_ROOT`` env-var override).

:func:`read_repo_id` is the canonical way to read the repository ID from
``.muse/repo.json``.  It replaces 73 copy-pasted ``_read_repo_id`` functions
that had diverged into four different error-handling behavioural variants.
The canonical implementation uses ``REPO_NOT_FOUND`` on a missing file and
``INTERNAL_ERROR`` on a malformed file — errors go through the logger, not
stderr, so commands that want to emit a user-facing message can catch the
``SystemExit`` and print their own context.

:func:`require_repo` performs a **startup GC sweep** on every invocation,
removing stale temp files left by a prior SIGKILL.  All three temp-file
families used by Muse are covered:

* ``.obj-tmp-*`` / ``.restore-tmp-*`` — object-store shard directories
  (delegated to :func:`muse.core.object_store.cleanup_stale_object_temps`)
* ``.muse-tmp-*`` — created by :func:`~muse.core.store.write_text_atomic`
  and :func:`~muse.core.store._write_msgpack_atomic` in ``.muse/`` and its
  subdirectories (branches, commits, snapshots, tags, releases, …)
* ``.stat_cache_*.tmp`` — created by :class:`~muse.core.stat_cache.StatCache`
  in ``.muse/``
"""

from __future__ import annotations

import datetime
import json
import logging
import os
import pathlib
import sys

from muse.core.errors import ExitCode, UntrustedRepositoryError
from muse.core.validation import assert_not_symlink

logger = logging.getLogger(__name__)

# Subdirectories of .muse/ (excluding objects/) that can hold stale temps.
# objects/ is handled separately by cleanup_stale_object_temps.
_MUSE_SWEEP_DIRS: tuple[str, ...] = (
    "",           # .muse/ root itself (HEAD, stat_cache, config, merge-state, etc.)
    "branches",   # write_branch_ref
    "commits",    # write_commit via _write_msgpack_atomic
    "snapshots",  # write_snapshot via _write_msgpack_atomic
    "tags",       # write_tag via _write_msgpack_atomic
    "releases",   # write_release via _write_msgpack_atomic
    "refs/heads", # write_branch_ref alternative path
    "code",       # code-domain index writes
    "indices",    # index writes
    "coordination", # create_intent / create_reservation
    "worktrees",  # _save_meta
    "stash",      # stash writes
    "rr-cache",   # reuse-recorded resolution cache
    "logs",       # log writes
    "remotes",    # remote config writes
)

# File-name prefixes that identify stale temps within the above directories.
# .muse-tmp-: write_text_atomic + _write_msgpack_atomic
# .stat_cache_: StatCache.save (suffix .tmp, but prefix is unique enough)
_MUSE_TEMP_PREFIXES: tuple[str, ...] = (".muse-tmp-", ".stat_cache_")


def _cleanup_muse_dir_temps(muse_dir: pathlib.Path) -> int:
    """Remove stale ``.muse-tmp-*`` and ``.stat_cache_*.tmp`` files.

    Iterates only the known subdirectory set that Muse writes to, so the
    object store (handled separately) and user files in the workdir are
    never touched.

    Returns:
        Number of stale temp files removed.
    """
    if not muse_dir.is_dir():
        return 0
    removed = 0
    for subdir in _MUSE_SWEEP_DIRS:
        target = muse_dir / subdir if subdir else muse_dir
        # Skip symlinked subdirectories — never delete files inside an
        # attacker-controlled location that was swapped in via symlink.
        if not target.is_dir() or target.is_symlink():
            continue
        for entry in target.iterdir():
            if entry.is_file() and any(
                entry.name.startswith(pfx) for pfx in _MUSE_TEMP_PREFIXES
            ):
                try:
                    entry.unlink()
                    removed += 1
                    logger.warning(
                        "⚠️  Removed stale muse temp %s (left by prior crash)", entry
                    )
                except OSError as exc:
                    logger.warning(
                        "⚠️  Could not remove stale temp %s: %s", entry, exc
                    )
    return removed


_CRITICAL_MUSE_DIRS: tuple[str, ...] = (
    "objects",
    "commits",
    "snapshots",
    "refs",
    "refs/heads",
    "tags",
)


def _verify_muse_dir_integrity(muse_dir: pathlib.Path) -> None:
    """Assert that critical ``.muse/`` subdirectories are real directories.

    Any of these being a symlink would redirect writes to an attacker-
    controlled location.  Called by :func:`require_repo` on every invocation
    so the check runs at the trust boundary, before any store operation.

    Args:
        muse_dir: Absolute path to the ``.muse/`` directory.

    Raises:
        SystemExit(1): If any critical subdirectory is a symbolic link.
    """
    for subname in _CRITICAL_MUSE_DIRS:
        candidate = muse_dir / subname
        if not candidate.exists():
            continue  # not yet created — first-use, not an attack
        try:
            assert_not_symlink(candidate, label=f".muse/{subname}")
        except ValueError as exc:
            logger.error("❌ %s", exc)
            raise SystemExit(1) from exc


def _startup_gc(repo_root: pathlib.Path) -> None:
    """Sweep all stale temp files left by a prior SIGKILL crash.

    Called by :func:`require_repo` on every command invocation so that any
    orphaned temp file from the previous crash is cleaned before the current
    command reads or writes the store.  The sweep is fast (< 5 ms on a
    typical repo) because it only touches small, bounded directories.

    Three temp-file families are covered:

    1. Object-store temps (``.obj-tmp-*``, ``.restore-tmp-*``) via
       :func:`~muse.core.object_store.cleanup_stale_object_temps`.
    2. Store/config temps (``.muse-tmp-*``) via
       :func:`_cleanup_muse_dir_temps`.
    3. Stat-cache temps (``.stat_cache_*.tmp``) via the same sweep (the
       ``.stat_cache_`` prefix is included in :data:`_MUSE_TEMP_PREFIXES`).
    """
    from muse.core.object_store import cleanup_stale_object_temps

    cleanup_stale_object_temps(repo_root)
    _cleanup_muse_dir_temps(repo_root / ".muse")


def _resolve_worktree_pointer(pointer_path: pathlib.Path) -> pathlib.Path | None:
    """Read a ``.muse`` worktree pointer file and return the main repo root.

    The file must contain a line of the form::

        musestore: /absolute/path/to/main/.muse

    Returns the parent of the ``.muse/`` store (i.e. the main repo root), or
    ``None`` on any parse or validation failure.  Never raises.
    """
    try:
        text = pointer_path.read_text(encoding="utf-8", errors="strict").strip()
    except Exception as exc:
        logger.debug("Could not read worktree pointer %s: %s", pointer_path, exc)
        return None

    prefix = "musestore: "
    if not text.startswith(prefix):
        logger.debug("Worktree pointer %s has unexpected format: %r", pointer_path, text[:80])
        return None

    raw_store = text[len(prefix):].strip()
    if not raw_store:
        return None
    if any(ord(c) < 0x20 or ord(c) == 0x7F for c in raw_store):
        logger.warning("⚠️ Worktree pointer %s contains control characters — ignoring", pointer_path)
        return None
    if len(raw_store) > 4096:
        logger.warning("⚠️ Worktree pointer %s path too long — ignoring", pointer_path)
        return None

    store_path = pathlib.Path(raw_store).resolve()
    if not store_path.is_dir() or store_path.is_symlink():
        logger.debug("Worktree pointer %s → %s is not a valid store dir", pointer_path, store_path)
        return None

    repo_root = store_path.parent
    # Loop guard: resolved root must not be the worktree directory itself.
    worktree_dir = pointer_path.parent.resolve()
    if repo_root == worktree_dir:
        logger.warning("⚠️ Worktree pointer %s loops back to its own directory", pointer_path)
        return None

    return repo_root


def _is_repo_trusted(repo_root: pathlib.Path) -> bool:
    """Return ``True`` if *repo_root* is in the caller's trust list.

    Checks two sources (in order):
    1. ``MUSE_SAFE_DIRS`` environment variable — colon-separated absolute paths.
    2. ``~/.muse/config.toml`` ``[security] safe_dirs`` list.

    Root (uid == 0) is always trusted.
    """
    if os.getuid() == 0:
        return True

    canonical = str(repo_root.resolve())

    # 1. MUSE_SAFE_DIRS env var (colon-separated, for Docker/CI).
    env_raw = os.environ.get("MUSE_SAFE_DIRS", "")
    if env_raw.strip():
        for raw_dir in env_raw.split(":"):
            if raw_dir.strip() and pathlib.Path(raw_dir.strip()).resolve() == pathlib.Path(canonical):
                return True

    # 2. ~/.muse/config.toml [security] safe_dirs.
    try:
        from muse.cli.config import get_global_safe_dirs
        for safe_path in get_global_safe_dirs():
            if pathlib.Path(safe_path).resolve() == pathlib.Path(canonical):
                return True
    except Exception:  # noqa: BLE001
        pass

    return False


def _check_repo_ownership(repo_root: pathlib.Path) -> None:
    """Raise :class:`~muse.core.errors.UntrustedRepositoryError` if ownership mismatch.

    Implements a CVE-2022-24765–equivalent check: the ``.muse/`` directory must
    be owned by the current user, or the repository must be explicitly trusted.

    Skipped when:
    - Current uid is 0 (root has unrestricted access anyway).
    - The repo is in ``MUSE_SAFE_DIRS`` or ``~/.muse/config.toml`` safe_dirs.

    Args:
        repo_root: The repository root directory (parent of ``.muse/``).

    Raises:
        UntrustedRepositoryError: When owner UID does not match current UID and
            the path is not in the trust list.
    """
    current_uid = os.getuid()
    if current_uid == 0:
        return  # root bypass

    muse_dir = repo_root / ".muse"
    try:
        st = muse_dir.stat()
    except OSError:
        # Can't stat — not a concern for ownership check; other code handles
        # missing .muse/
        return

    owner_uid = st.st_uid
    if owner_uid == current_uid:
        return  # owned by us — safe

    # Different owner — check trust list before raising.
    if _is_repo_trusted(repo_root):
        return

    raise UntrustedRepositoryError(
        path=str(repo_root),
        owner_uid=owner_uid,
        current_uid=current_uid,
    )


def find_repo_root(start: pathlib.Path | None = None) -> pathlib.Path | None:
    """Walk up from *start* (default ``Path.cwd()``) looking for ``.muse/``.

    Returns the first directory that contains ``.muse/``, or ``None`` if no
    such ancestor exists. Never raises — callers decide what to do on miss.

    The ``MUSE_REPO_ROOT`` environment variable overrides discovery entirely;
    set it in tests to avoid ``os.chdir`` calls.

    Security hardening for ``MUSE_REPO_ROOT``:
    - Empty or whitespace-only values are silently ignored (falls through to
      directory walk) rather than being resolved to the current working
      directory, which would bypass the explicit intent to override.
    - Values longer than the OS ``PATH_MAX`` (4096 on Linux/macOS) are
      rejected — overly long paths indicate an injection attempt.
    - Control characters in the value are rejected — they indicate a crafted
      payload rather than a genuine file-system path.
    - Symlinked ``.muse/`` directories are rejected even when the path comes
      from the env override, consistent with the directory-walk path.

    Ownership check (CVE-2022-24765 equivalent):
    - After locating ``.muse/``, ``_check_repo_ownership`` verifies that the
      directory is owned by the current user.
    - Raises :class:`~muse.core.errors.UntrustedRepositoryError` on mismatch
      unless the path is in ``MUSE_SAFE_DIRS`` or ``~/.muse/config.toml``.
    """
    raw_env = os.environ.get("MUSE_REPO_ROOT")
    if raw_env is not None:
        # Silently ignore empty or whitespace-only values — fall through to walk.
        stripped = raw_env.strip()
        if not stripped:
            logger.debug("MUSE_REPO_ROOT is empty or whitespace — ignoring, using cwd walk")
        else:
            # Reject values containing control characters.
            if any(ord(c) < 0x20 or ord(c) == 0x7F for c in stripped):
                logger.warning(
                    "⚠️ MUSE_REPO_ROOT contains control characters — ignoring for safety"
                )
                return None
            # Reject unreasonably long paths (OS PATH_MAX is 4096 on Linux/macOS).
            if len(stripped) > 4096:
                logger.warning(
                    "⚠️ MUSE_REPO_ROOT is too long (%d chars) — ignoring for safety",
                    len(stripped),
                )
                return None
            p = pathlib.Path(stripped).resolve()
            logger.debug("⚠️ MUSE_REPO_ROOT override active: %s", p)
            muse_candidate = p / ".muse"
            # Reject symlinked .muse/ even when the path comes from the env override.
            if muse_candidate.is_dir() and not muse_candidate.is_symlink():
                _check_repo_ownership(p)
                return p
            return None

    current = (start or pathlib.Path.cwd()).resolve()
    while True:
        muse_dir = current / ".muse"
        # Reject symlinked .muse/ — a symlink here redirects all subsequent
        # writes to an attacker-controlled location outside the repo root.
        if muse_dir.is_dir() and not muse_dir.is_symlink():
            _check_repo_ownership(current)
            return current
        # Linked worktree: .muse is a file containing "musestore: /path/to/.muse"
        if muse_dir.is_file() and not muse_dir.is_symlink():
            resolved = _resolve_worktree_pointer(muse_dir)
            if resolved is not None:
                _check_repo_ownership(resolved)
                return resolved
        parent = current.parent
        if parent == current:
            return None
        current = parent


_NOT_A_REPO_MSG = (
    'fatal: not a muse repository (or any parent up to mount point /)\n'
    'Run "muse init" to initialize a new repository.'
)


def require_repo(start: pathlib.Path | None = None) -> pathlib.Path:
    """Return the repo root or exit 2 with a clear error message.

    Wraps ``find_repo_root()`` for command callbacks that must be inside a
    Muse repository. The error text is written to stderr so the shell always
    surfaces it; our ``CliRunner`` merges stderr into ``result.output``.

    **Startup GC sweep:** after locating the repo root, performs a lightweight
    sweep of all ``.muse/`` subdirectories to remove stale temp files left by
    a prior ``SIGKILL``.  The sweep covers all three temp-file families
    produced by the store layer (``.muse-tmp-*``, ``.stat_cache_*.tmp``,
    ``.obj-tmp-*``, ``.restore-tmp-*``).  The cost is < 5 ms on a typical
    repo because only small, bounded directories are listed.
    """
    root = find_repo_root(start)
    if root is None:
        print(_NOT_A_REPO_MSG, file=sys.stderr)
        raise SystemExit(ExitCode.REPO_NOT_FOUND)
    _verify_muse_dir_integrity(root / ".muse")
    _startup_gc(root)
    return root


#: Public alias.
require_repo_root = require_repo


def parse_date_arg(value: str, flag: str) -> datetime.datetime:
    """Parse an ISO-8601 date or datetime string from a CLI flag.

    Accepts ``YYYY-MM-DD`` and ``YYYY-MM-DDTHH:MM:SS``.  Always returns a
    UTC-aware :class:`datetime.datetime`.  Exits with code 1 and a clear
    error message on parse failure, naming the offending *flag*.

    This is the canonical implementation replacing per-command inline date
    parsing that used different formats and error messages.

    Args:
        value: The raw string value from the CLI argument.
        flag:  The flag name (e.g. ``--since``) used in the error message.

    Returns:
        A UTC-aware :class:`datetime.datetime`.

    Raises:
        SystemExit(1): when *value* cannot be parsed as a recognised format.
    """
    for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"):
        try:
            return datetime.datetime.strptime(value, fmt).replace(
                tzinfo=datetime.timezone.utc
            )
        except ValueError:
            continue
    print(
        f"❌ Invalid date for {flag}: {value!r}"
        " — expected YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS",
        file=sys.stderr,
    )
    raise SystemExit(1)


def read_repo_id(repo_root: pathlib.Path) -> str:
    """Read the ``repo_id`` from ``.muse/repo.json``.

    This is the canonical implementation.  It replaces 73 copy-pasted
    ``_read_repo_id`` functions that had diverged into four different
    error-handling variants across the codebase.

    Raises:
        SystemExit(REPO_NOT_FOUND): when ``.muse/repo.json`` does not exist.
        SystemExit(INTERNAL_ERROR): when the file exists but is not valid JSON
            or does not contain the expected ``repo_id`` key.
    """
    repo_json = repo_root / ".muse" / "repo.json"
    try:
        return str(json.loads(repo_json.read_text(encoding="utf-8"))["repo_id"])
    except FileNotFoundError as exc:
        logger.debug(".muse/repo.json not found: %s", exc)
        raise SystemExit(ExitCode.REPO_NOT_FOUND) from exc
    except (json.JSONDecodeError, KeyError) as exc:
        logger.debug(".muse/repo.json malformed: %s", exc)
        raise SystemExit(ExitCode.INTERNAL_ERROR) from exc