#!/usr/bin/env python3
"""Fail-closed sweep of orphaned `.git/objects/pack/tmp_pack_*` debris.

Sole periodic owner is system-monitor-disk-maintain (via disk-check.timer),
same as worktree_gc.py. A `tmp_pack_*` file is git's own scratch name for a
pack still being written by `index-pack`/`unpack-objects`/a fetch/push; a
live writer keeps updating its mtime, so an age gate (default 90 min) is the
whole safety story here — no process check is needed or attempted. Debris
regenerates after any ENOSPC/OOM kill of a git operation; this sweep only
removes stray temp files by exact name + age, never a real pack and never
`git gc`/`git prune`.

Linked worktrees (`.worktrees/<slug>`, `.claude/worktrees/<slug>`) share the
main checkout's object store via `.git/worktrees/<name>/commondir` — they
have no `objects/pack` dir of their own (confirmed: `git worktree add`
never creates one). `discover_repos` finds repo roots; each root's real
objects/pack dir is resolved through `.git` (file or dir) and `commondir`
so a discovered path that is itself a linked worktree still finds the one
shared pack dir instead of silently finding nothing.
"""

from __future__ import annotations

import argparse
import json
import os
import stat as stat_mod
import sys
import time
from pathlib import Path
from typing import Any

LIB_DIR = Path(__file__).resolve().parent
if str(LIB_DIR) not in sys.path:
    sys.path.insert(0, str(LIB_DIR))

from worktree_gc import discover_repos  # noqa: E402

TMP_PACK_PREFIX = "tmp_pack_"
DEFAULT_MIN_AGE_S = 90 * 60  # 90 minutes — matches the manual sweep that made tonight's incident safe.
DEFAULT_PROJECT_ROOTS = [os.path.join(os.path.expanduser("~"), "Projects")]


def resolve_common_git_dir(repo_root: str) -> str | None:
    """Return the real `.git` dir that owns `objects/`, following worktree commondir.

    None if `repo_root/.git` is missing or unreadable — caller must skip, never error.
    """
    dot_git = os.path.join(repo_root, ".git")
    try:
        st = os.lstat(dot_git)
    except OSError:
        return None
    if os.path.islink(dot_git):
        # Never follow a symlinked `.git` — treat as unreadable rather than chase it.
        return None
    if os.path.isdir(dot_git):
        git_dir = dot_git
    elif os.path.isfile(dot_git):
        try:
            content = Path(dot_git).read_text(encoding="utf-8", errors="strict")
        except OSError:
            return None
        prefix = "gitdir:"
        line = content.strip()
        if not line.startswith(prefix):
            return None
        git_dir = os.path.realpath(os.path.join(repo_root, line[len(prefix):].strip()))
    else:
        return None
    if not os.path.isdir(git_dir):
        return None
    commondir_file = os.path.join(git_dir, "commondir")
    if os.path.isfile(commondir_file):
        try:
            common = Path(commondir_file).read_text(encoding="utf-8").strip()
        except OSError:
            return None
        git_dir = os.path.realpath(os.path.join(git_dir, common))
    return git_dir if os.path.isdir(git_dir) else None


def pack_dir_for_repo(repo_root: str) -> str | None:
    git_dir = resolve_common_git_dir(repo_root)
    if git_dir is None:
        return None
    pack_dir = os.path.join(git_dir, "objects", "pack")
    return pack_dir if os.path.isdir(pack_dir) else None


def is_eligible_tmp_pack(entry_path: str, *, now_s: float, min_age_s: float) -> bool:
    """HARD SAFETY gate: name, type, and age — nothing else authorizes removal."""
    name = os.path.basename(entry_path)
    if not name.startswith(TMP_PACK_PREFIX):
        return False
    try:
        st = os.lstat(entry_path)
    except OSError:
        return False
    if not stat_mod.S_ISREG(st.st_mode):
        # Covers symlinks (never followed), dirs, and anything else non-regular.
        return False
    if (now_s - st.st_mtime) < min_age_s:
        return False
    return True


def find_repo_tmp_packs(repo_root: str, *, now_s: float, min_age_s: float) -> list[str]:
    pack_dir = pack_dir_for_repo(repo_root)
    if pack_dir is None:
        return []
    eligible: list[str] = []
    try:
        entries = list(os.scandir(pack_dir))
    except OSError:
        return []
    # os.scandir only yields direct children of the already-resolved pack_dir
    # (entry.path == pack_dir/name); nothing here can name a path outside it.
    for entry in entries:
        if is_eligible_tmp_pack(entry.path, now_s=now_s, min_age_s=min_age_s):
            eligible.append(entry.path)
    return eligible


def _statvfs_avail_bytes(path: str) -> int | None:
    try:
        vfs = os.statvfs(path)
    except OSError:
        return None
    return vfs.f_bavail * vfs.f_frsize


def sweep_tmp_packs(options: dict[str, Any] | None = None) -> dict[str, Any]:
    opts = options or {}
    apply = opts.get("apply") is True
    min_age_s = opts["min_age_s"] if isinstance(opts.get("min_age_s"), (int, float)) else DEFAULT_MIN_AGE_S
    project_roots = (
        opts["project_roots"]
        if isinstance(opts.get("project_roots"), list) and opts["project_roots"]
        else DEFAULT_PROJECT_ROOTS
    )
    now_s = opts["now_s"] if isinstance(opts.get("now_s"), (int, float)) else time.time()

    summary: dict[str, Any] = {
        "scanned": 0,
        "removed": [],
        "kept": [],
        "errors": [],
        "freedBytes": 0,
        "dryRun": not apply,
    }

    repos = discover_repos(project_roots)
    candidates: dict[str, list[str]] = {}
    for repo_root in repos:
        found = find_repo_tmp_packs(repo_root, now_s=now_s, min_age_s=min_age_s)
        if found:
            candidates[repo_root] = found
        summary["scanned"] += 1

    if not candidates:
        return summary

    # statvfs delta, grouped by filesystem device, so a hardlinked/shared mount
    # is never double counted the way summed `du` over multiple targets would be.
    devices: dict[int, str] = {}
    for repo_root in candidates:
        try:
            st_dev = os.stat(repo_root).st_dev
        except OSError:
            continue
        devices.setdefault(st_dev, repo_root)

    before_avail = {dev: _statvfs_avail_bytes(path) for dev, path in devices.items()}

    for repo_root, files in candidates.items():
        for target in files:
            summary["removed_candidate_count"] = summary.get("removed_candidate_count", 0) + 1
            if not apply:
                summary["kept"].append({"path": target, "reason": "dry-run"})
                continue
            try:
                os.unlink(target)
            except OSError as exc:
                summary["errors"].append({"path": target, "error": str(exc)})
                continue
            summary["removed"].append(target)

    if apply:
        after_avail = {dev: _statvfs_avail_bytes(path) for dev, path in devices.items()}
        freed = 0
        for dev, before in before_avail.items():
            after = after_avail.get(dev)
            if before is None or after is None:
                continue
            delta = after - before
            if delta > 0:
                freed += delta
        summary["freedBytes"] = freed

    summary.pop("removed_candidate_count", None)
    return summary


def _parse_project_roots(raw: str | None) -> list[str]:
    if not raw:
        return list(DEFAULT_PROJECT_ROOTS)
    return [part for part in raw.split(os.pathsep) if part]


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Fail-closed sweep of orphaned .git/objects/pack/tmp_pack_* debris"
    )
    parser.add_argument("--apply", action="store_true", help="Actually remove eligible temp packs (default: dry-run)")
    parser.add_argument("--project-roots", default=os.environ.get("SM_TMP_PACK_PROJECT_ROOTS"))
    parser.add_argument(
        "--min-age-s",
        type=float,
        default=float(os.environ.get("SM_TMP_PACK_MIN_AGE_S", DEFAULT_MIN_AGE_S)),
    )
    args = parser.parse_args(argv)

    summary = sweep_tmp_packs({
        "apply": args.apply,
        "project_roots": _parse_project_roots(args.project_roots),
        "min_age_s": args.min_age_s,
    })
    print(json.dumps({
        "scanned": summary["scanned"],
        "removed": len(summary["removed"]),
        "kept": len(summary["kept"]),
        "errors": summary["errors"],
        "freedBytes": summary["freedBytes"],
        "dryRun": summary["dryRun"],
    }, sort_keys=True))
    return 0 if not summary["errors"] else 1


if __name__ == "__main__":
    sys.exit(main())