320 lines
11 KiB
Python
320 lines
11 KiB
Python
"""
|
||
State store for doc-hygiene.
|
||
|
||
Provides:
|
||
- resolve_project_root(start_dir, fs) — pure function, no side effects
|
||
- StateStore — confines all writes to <project_root>/.dochygiene/
|
||
|
||
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 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>/.dochygiene/ (invariant #3).
|
||
No global index is maintained; each project has its own independent store.
|
||
The store never opens or edits .gitignore (invariant #3).
|
||
|
||
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 / ".dochygiene"
|
||
|
||
# ------------------------------------------------------------------
|
||
# Directory bootstrap
|
||
# ------------------------------------------------------------------
|
||
|
||
def _ensure_state_dir(self) -> Path:
|
||
"""Create .dochygiene/ if it does not exist. Never touches .gitignore."""
|
||
self._state_dir.mkdir(parents=True, exist_ok=True)
|
||
return self._state_dir
|
||
|
||
# ------------------------------------------------------------------
|
||
# 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 .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)."""
|
||
path = self._state_path()
|
||
try:
|
||
return json.loads(path.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.
|
||
"""
|
||
json_path = self._state_dir / _REPORT_JSON
|
||
md_path = self._state_dir / _REPORT_MD
|
||
try:
|
||
return json_path.read_text(), md_path.read_text()
|
||
except FileNotFoundError:
|
||
return None
|