cc-os/plugins/os-doc-hygiene/scripts/state_store.py

"""
State store for doc-hygiene.

Provides:
  - resolve_project_root(start_dir, fs) — pure function, no side effects
  - StateStore — confines all writes to <project_root>/.cc-os/dochygiene/

Per ADR-027, the canonical state directory is <project_root>/.cc-os/dochygiene/.
The legacy path <project_root>/.dochygiene/ is still supported for reads: if the
canonical directory has no state, reads fall back to the legacy directory. On the
first write, any existing legacy files are migrated into the canonical directory
and the legacy directory is removed.

Design invariants honoured:
  #3  State lives in-project; no global index; never edit .gitignore.
  #4  Report rollover keeps exactly one .md + .json pair.
  #6  Deterministic-first; no model invoked here.
  #9  (scanner's concern, not ours)

Atomic-write mechanism: write to a temp file in the same directory,
fsync the file descriptor, then os.replace (POSIX-atomic) onto the target.
A concurrent reader therefore observes either the prior complete file or the
new complete file, never a partial write.
"""

from __future__ import annotations

import json
import os
import shutil
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, Protocol


# ---------------------------------------------------------------------------
# Lightweight filesystem abstraction used only by resolve_project_root
# so the pure function can be tested without touching real disk.
# ---------------------------------------------------------------------------

class _RootFS(Protocol):
    """Minimal filesystem surface for root resolution."""
    def is_dir(self, path: Path) -> bool: ...
    def parent(self, path: Path) -> Path: ...


class RealRootFS:
    """Production implementation — delegates to pathlib/os."""

    def is_dir(self, path: Path) -> bool:
        return path.is_dir()

    def parent(self, path: Path) -> Path:
        return path.parent


# ---------------------------------------------------------------------------
# Pure root-resolution function (task 3.1)
# ---------------------------------------------------------------------------

def resolve_project_root(start_dir: Path, fs: Optional[_RootFS] = None) -> Path:
    """
    Walk upward from *start_dir* (inclusive) looking for a .git directory.

    Returns the first ancestor (or start_dir itself) that contains a .git
    directory.  If no git root is found, returns start_dir unchanged.

    This function is PURE: it never calls os.getcwd() or any other
    stateful function; all filesystem access goes through *fs*.  Production
    callers pass os.getcwd() as start_dir; tests pass a fake fs and a
    synthetic path.
    """
    if fs is None:
        fs = RealRootFS()

    current = Path(start_dir)
    while True:
        if fs.is_dir(current / ".git"):
            return current
        parent = fs.parent(current)
        if parent == current:
            # Reached the filesystem root without finding .git.
            break
        current = parent

    return Path(start_dir)


# ---------------------------------------------------------------------------
# Clock abstraction (task 3.3 / design D3)
# ---------------------------------------------------------------------------

class _Clock(Protocol):
    """Returns the current UTC datetime."""
    def now(self) -> datetime: ...


class RealClock:
    """Production clock — returns the real UTC time."""

    def now(self) -> datetime:
        return datetime.now(timezone.utc)


# ---------------------------------------------------------------------------
# StateStore (tasks 3.2–3.5)
# ---------------------------------------------------------------------------

_STATE_FILE = "state.json"
_REPORT_JSON = "report.json"
_REPORT_MD = "report.md"

# The set of filenames that are *not* report files and must never be deleted
# during rollover.  Explicit allowlist is safer than trying to infer.
_NON_REPORT_FILES = {_STATE_FILE}


class StateStore:
    """
    Manages all persistent state for doc-hygiene within a single project.

    All writes are confined to <project_root>/.cc-os/dochygiene/ (invariant #3,
    ADR-027).  No global index is maintained; each project has its own
    independent store.  The store never opens or edits .gitignore (invariant #3).

    Backward compatibility (ADR-027): the legacy location
    <project_root>/.dochygiene/ is read as a fallback when the canonical
    directory has no state.  The first write migrates any legacy files into the
    canonical directory and removes the legacy directory.

    Parameters
    ----------
    project_root:
        The resolved project root directory (output of resolve_project_root).
        Injected so the store is testable with a tmp_path.
    clock:
        Provides "now".  Injected for testability (design D3).
    """

    TIMESTAMPS = ("last_check", "last_clean", "last_reminded")

    def __init__(self, project_root: Path, clock: Optional[_Clock] = None) -> None:
        self._root = Path(project_root)
        self._clock = clock or RealClock()
        self._state_dir = self._root / ".cc-os" / "dochygiene"
        self._legacy_dir = self._root / ".dochygiene"

    # ------------------------------------------------------------------
    # Directory bootstrap
    # ------------------------------------------------------------------

    def _ensure_state_dir(self) -> Path:
        """Create .cc-os/dochygiene/ if it does not exist, migrating any legacy
        .dochygiene/ contents in first.  Never touches .gitignore."""
        self._migrate_legacy()
        self._state_dir.mkdir(parents=True, exist_ok=True)
        return self._state_dir

    def _migrate_legacy(self) -> None:
        """
        One-time migration of <root>/.dochygiene/ into <root>/.cc-os/dochygiene/.

        No-op if the legacy directory does not exist. Copies every file found
        in the legacy directory into the canonical directory (canonical files,
        if any already exist, take precedence and are not overwritten), then
        removes the legacy directory entirely.
        """
        if not self._legacy_dir.is_dir():
            return

        self._state_dir.mkdir(parents=True, exist_ok=True)
        for entry in self._legacy_dir.iterdir():
            if not entry.is_file():
                continue
            target = self._state_dir / entry.name
            if not target.exists():
                target.write_bytes(entry.read_bytes())

        shutil.rmtree(self._legacy_dir, ignore_errors=True)

    # ------------------------------------------------------------------
    # Atomic write (task 3.4 / design D9)
    # ------------------------------------------------------------------

    def _atomic_write(self, target: Path, data: bytes) -> None:
        """
        Write *data* to *target* atomically.

        Strategy: write to a NamedTemporaryFile in the same directory,
        fsync, then os.replace onto the target.  os.replace is POSIX-atomic
        within one filesystem, so a concurrent reader sees either the prior
        complete file or the new complete file, never a partial write.
        """
        # Confirm the target is under our managed directory (confinement check).
        self._assert_confined(target)

        target_dir = target.parent
        target_dir.mkdir(parents=True, exist_ok=True)

        fd, tmp_path_str = tempfile.mkstemp(
            dir=str(target_dir),
            prefix="." + target.name + ".tmp_",
        )
        tmp_path = Path(tmp_path_str)
        try:
            with os.fdopen(fd, "wb") as fh:
                fh.write(data)
                fh.flush()
                os.fsync(fh.fileno())
            os.replace(tmp_path_str, str(target))
        except Exception:
            # Best-effort cleanup on failure; do not mask the original error.
            try:
                tmp_path.unlink(missing_ok=True)
            except OSError:
                pass
            raise

    def _assert_confined(self, path: Path) -> None:
        """Raise ValueError if *path* is not under self._state_dir."""
        try:
            path.resolve().relative_to(self._state_dir.resolve())
        except ValueError:
            raise ValueError(
                f"StateStore attempted to write outside .cc-os/dochygiene/: {path}"
            )

    # ------------------------------------------------------------------
    # State JSON helpers
    # ------------------------------------------------------------------

    def _state_path(self) -> Path:
        return self._state_dir / _STATE_FILE

    def _read_state(self) -> dict:
        """Read state.json; return {} if missing or unreadable (never raises).

        Falls back to the legacy .dochygiene/state.json (ADR-027) if the
        canonical file is absent. The legacy file is only *read* here — it is
        migrated into the canonical location on the next write.
        """
        try:
            return json.loads(self._state_path().read_bytes())
        except (FileNotFoundError, json.JSONDecodeError, OSError):
            pass
        try:
            return json.loads((self._legacy_dir / _STATE_FILE).read_bytes())
        except (FileNotFoundError, json.JSONDecodeError, OSError):
            return {}

    def _write_state(self, state: dict) -> None:
        self._ensure_state_dir()
        data = json.dumps(state, indent=2, sort_keys=True).encode()
        self._atomic_write(self._state_path(), data)

    # ------------------------------------------------------------------
    # Lifecycle timestamps (task 3.3)
    # ------------------------------------------------------------------

    def get_timestamp(self, key: str) -> Optional[datetime]:
        """
        Return the stored datetime for *key*, or None if absent/unset.

        The value is stored as an ISO-8601 string in state.json.
        """
        if key not in self.TIMESTAMPS:
            raise ValueError(f"Unknown timestamp key: {key!r}")
        raw = self._read_state().get(key)
        if raw is None:
            return None
        return datetime.fromisoformat(raw)

    def set_timestamp(self, key: str, value: Optional[datetime] = None) -> None:
        """
        Write *key* to state.json.

        If *value* is None, uses the injected clock's now().
        The value is serialised as an ISO-8601 string (UTC).
        """
        if key not in self.TIMESTAMPS:
            raise ValueError(f"Unknown timestamp key: {key!r}")
        ts = value if value is not None else self._clock.now()
        # Normalise to UTC ISO-8601 string.
        if ts.tzinfo is None:
            ts = ts.replace(tzinfo=timezone.utc)
        state = self._read_state()
        state[key] = ts.isoformat()
        self._write_state(state)

    # Convenience shorthands
    def set_last_check(self, value: Optional[datetime] = None) -> None:
        self.set_timestamp("last_check", value)

    def get_last_check(self) -> Optional[datetime]:
        return self.get_timestamp("last_check")

    def set_last_clean(self, value: Optional[datetime] = None) -> None:
        self.set_timestamp("last_clean", value)

    def get_last_clean(self) -> Optional[datetime]:
        return self.get_timestamp("last_clean")

    def set_last_reminded(self, value: Optional[datetime] = None) -> None:
        self.set_timestamp("last_reminded", value)

    def get_last_reminded(self) -> Optional[datetime]:
        return self.get_timestamp("last_reminded")

    # ------------------------------------------------------------------
    # Report rollover (task 3.5 / design D10)
    # ------------------------------------------------------------------

    def _delete_existing_reports(self) -> None:
        """
        Delete any existing report.json and report.md in .dochygiene/.

        Only report files (report.json, report.md) are deleted.
        state.json and any other files are never touched.
        """
        for name in (_REPORT_JSON, _REPORT_MD):
            p = self._state_dir / name
            try:
                p.unlink()
            except FileNotFoundError:
                pass

    def write_report(self, json_blob: str, md_blob: str) -> None:
        """
        Write a new report pair, atomically, after deleting any prior pair.

        After this call exactly one .json and one .md report file exist in
        .dochygiene/ (invariant #4).

        Parameters
        ----------
        json_blob:
            The machine-readable report JSON (as a string).
        md_blob:
            The human-readable report Markdown.
        """
        self._ensure_state_dir()
        # Delete prior pair first (rollover).
        self._delete_existing_reports()
        # Atomically write the new pair.
        self._atomic_write(
            self._state_dir / _REPORT_JSON,
            json_blob.encode(),
        )
        self._atomic_write(
            self._state_dir / _REPORT_MD,
            md_blob.encode(),
        )

    def read_report(self) -> Optional[tuple[str, str]]:
        """
        Return (json_blob, md_blob) if a report exists, else None.

        Falls back to the legacy .dochygiene/ directory (ADR-027) if the
        canonical directory has no report pair.
        """
        try:
            return (
                (self._state_dir / _REPORT_JSON).read_text(),
                (self._state_dir / _REPORT_MD).read_text(),
            )
        except FileNotFoundError:
            pass
        try:
            return (
                (self._legacy_dir / _REPORT_JSON).read_text(),
                (self._legacy_dir / _REPORT_MD).read_text(),
            )
        except FileNotFoundError:
            return None