ai-agent/backend/githist.py

"""
Bucketed git history, cached.

The services/projects catalogs used to fork two ``git log -- <prefix>/<name>``
subprocesses per directory on every request — 34 services × ~160 ms of
history-walk each made ``/api/services`` an 11-second endpoint. Instead we walk
the superproject's history **once** (``git log --name-only -- <prefix>``),
bucket the commits by ``<prefix>/<name>``, and cache the result keyed by the
repo's HEAD: a request pays one cheap ``rev-parse`` (TTL-gated) and a dict
lookup, and the walk re-runs only after a new commit.

Standalone project repos (their own ``.git``) can't join the bucketed walk;
:func:`repo_log` gives them a per-repo TTL cache instead.
"""

import datetime
import subprocess
import threading
import time

# Re-check HEAD at most this often; between checks the cache is trusted as-is.
HEAD_TTL_SECS = 15.0
# Standalone repos: re-run their (cheap, small-repo) log at most this often.
REPO_TTL_SECS = 60.0
_LOG_FORMAT = "%x01%h%x1f%ct%x1f%cI%x1f%s"


def _git(args: list[str], cwd, timeout: int = 30) -> str | None:
    cmd = ["git", "-c", "safe.directory=*", "-C", str(cwd), *args]
    try:
        out = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    except (OSError, subprocess.SubprocessError):
        return None
    return out.stdout if out.returncode == 0 else None


def _parse_log(out: str | None) -> list[dict]:
    """``--format=_LOG_FORMAT --name-only`` output → [{hash, ts, date, subject,
    files}] newest-first."""
    commits: list[dict] = []
    cur: dict | None = None
    for line in (out or "").splitlines():
        if line.startswith("\x01"):
            parts = line[1:].split("\x1f")
            if len(parts) == 4:
                try:
                    ts = int(parts[1])
                except ValueError:
                    ts = None
                cur = {"hash": parts[0], "ts": ts, "date": parts[2],
                       "subject": parts[3], "files": []}
                commits.append(cur)
            else:
                cur = None
        elif line.strip() and cur is not None:
            cur["files"].append(line.strip())
    return commits


class BucketedHistory:
    """Per-directory commit history under ``<repo>/<prefix>/``, one walk."""

    def __init__(self, repo_dir, prefix: str, per_bucket: int = 8):
        self.repo_dir = repo_dir
        self.prefix = prefix.strip("/")
        self.per_bucket = per_bucket
        self._lock = threading.Lock()
        self._head: str | None = None
        self._checked = 0.0
        self._buckets: dict[str, dict] = {}

    def _walk(self) -> dict[str, dict]:
        out = _git(["log", f"--format={_LOG_FORMAT}", "--name-only",
                    "--", self.prefix], self.repo_dir)
        buckets: dict[str, dict] = {}
        pre = self.prefix + "/"
        for c in _parse_log(out):
            names = set()
            for f in c["files"]:
                if f.startswith(pre):
                    tail = f[len(pre):]
                    name = tail.split("/", 1)[0]
                    if name:
                        names.add(name)
            for name in names:
                b = buckets.setdefault(name, {"ts": c["ts"], "iso": c["date"],
                                              "commits": []})
                if len(b["commits"]) < self.per_bucket:
                    b["commits"].append({"hash": c["hash"], "date": c["date"],
                                         "subject": c["subject"]})
        return buckets

    def get(self, name: str) -> dict | None:
        """{ts, iso, commits} for one directory, or None (no commits yet)."""
        now = time.time()
        with self._lock:
            if self._head is not None and now - self._checked < HEAD_TTL_SECS:
                return self._buckets.get(name)
            head = (_git(["rev-parse", "HEAD"], self.repo_dir) or "?").strip()
            self._checked = now
            if head != self._head:
                self._buckets = self._walk()
                self._head = head
            return self._buckets.get(name)


class RepoLogCache:
    """TTL-cached ``git log`` for standalone repos (a project's own .git)."""

    def __init__(self, per_repo: int = 8):
        self.per_repo = per_repo
        self._lock = threading.Lock()
        self._cache: dict[str, tuple[float, dict | None]] = {}

    def get(self, repo_dir) -> dict | None:
        """{ts, iso, commits} for the repo's recent history, or None."""
        key = str(repo_dir)
        now = time.time()
        with self._lock:
            hit = self._cache.get(key)
            if hit and now - hit[0] < REPO_TTL_SECS:
                return hit[1]
        out = _git(["log", f"-{self.per_repo}", f"--format={_LOG_FORMAT}"],
                   repo_dir)
        rec: dict | None = None
        commits = _parse_log(out)
        if commits:
            newest = commits[0]
            rec = {"ts": newest["ts"], "iso": newest["date"],
                   "commits": [{"hash": c["hash"], "date": c["date"],
                                "subject": c["subject"]} for c in commits]}
        with self._lock:
            self._cache[key] = (now, rec)
        return rec


def parse_iso_ts(iso: str | None) -> int | None:
    if not iso:
        return None
    try:
        return int(datetime.datetime.fromisoformat(iso).timestamp())
    except ValueError:
        return None