"""Nightly rehearsal: snapshot -> scrub -> legacy -> transform -> extract both sides -> compare -> status line (the FULL rehearsal per 05-migration-plan; "cutover day is the hundredth run, not the first"). Scheduled daily at ``settings.snapshot_time`` by an asyncio task in app.main (which also fires it at boot when the sidecar has no ``legacy`` schema yet, or when RUN_SNAPSHOT_ON_START=1); run on demand with ``docker exec with-app python -m scripts.nightly_snapshot``. 1. pg_dump the Supabase source — schema ``public`` plus table ``auth.users``, pre-data + data sections only (no owners, privileges, indexes, FKs, triggers or policies: the landed copy is read-only forensics/extractor input). MUST run inside the app container — the image ships postgresql-client-17 to match the source server's major version (host pg_dump is v16). 2. Rewrite schema qualifiers (public./auth. -> legacy.) outside COPY data blocks. 3. Load atomically: drop+recreate schema ``legacy``, load, NULL every credential column in legacy.users, verify the scrub — all in ONE transaction, so unscrubbed credentials are never visible and a failed run leaves the previous snapshot intact. 4. Run the diff-harness old-side extractor (harness/extract_legacy.py) against the sidecar, in-process (the DSN never appears on argv). 5. Run the migration transforms (app/services/migrate — legacy.* -> public.*, idempotent, one transaction, run report persisted). 6. Run the new-side emitter (harness/extract_new.py) and compare both canonical documents (harness/compare.py + allowlist). 7. Append a status line to harness/out/nightly-status.jsonl either way. Raw dumps (which contain credential hashes until the scrub) exist only in a mode-700 temp dir that is deleted before the run returns. """ import asyncio import fcntl import gzip import json import logging import os import re import shutil import subprocess import sys import tempfile import time from collections.abc import Iterable, Iterator from datetime import UTC, datetime, timedelta from pathlib import Path from typing import IO, Any from sqlalchemy import text from app.config import settings from app.db import get_engine log = logging.getLogger("with.snapshot") ROOT = Path(__file__).resolve().parents[1] #: Credential-bearing columns in auth.users (05-migration-plan step 1). SCRUB_COLUMNS = ( "encrypted_password", "confirmation_token", "recovery_token", "email_change_token_new", "email_change_token_current", "phone_change_token", "reauthentication_token", ) _QUALIFIED = re.compile(r"\b(public|auth)\.") _SCHEMA_STMT = re.compile(r"^(CREATE|DROP|COMMENT ON) SCHEMA ", re.IGNORECASE) _HEADER_SQL = """\ -- Generated by scripts/nightly_snapshot.py — do not edit. DROP SCHEMA IF EXISTS legacy CASCADE; CREATE SCHEMA legacy; -- Shims for column defaults that reference Supabase's extensions schema. CREATE SCHEMA IF NOT EXISTS extensions; CREATE EXTENSION IF NOT EXISTS "uuid-ossp" WITH SCHEMA extensions; CREATE EXTENSION IF NOT EXISTS pgcrypto WITH SCHEMA extensions; """ # --- pure helpers ------------------------------------------------------- def rewrite_sql(lines: Iterable[str]) -> Iterator[str]: """Rewrite schema qualifiers ``public.``/``auth.`` -> ``legacy.`` on command lines, leaving COPY data blocks byte-identical. Schema-level statements (CREATE/DROP/COMMENT ON SCHEMA) are commented out — the header owns schema lifecycle.""" in_copy = False for line in lines: if in_copy: yield line if line.rstrip("\n") == "\\.": in_copy = False continue if _SCHEMA_STMT.match(line): yield f"-- [snapshot] skipped: {line}" continue rewritten = _QUALIFIED.sub("legacy.", line) if rewritten.startswith("COPY ") and rewritten.rstrip().endswith("FROM stdin;"): in_copy = True yield rewritten def _scrub_sql() -> str: sets = ",\n ".join(f"{c} = NULL" for c in SCRUB_COLUMNS) checks = " OR ".join(f"{c} IS NOT NULL" for c in SCRUB_COLUMNS) return f""" -- Credential scrub — same transaction as the load, so unscrubbed -- credentials are never visible outside it. UPDATE legacy.users SET {sets}; DO $scrub$ DECLARE remaining integer; BEGIN SELECT count(*) INTO remaining FROM legacy.users WHERE {checks}; IF remaining > 0 THEN RAISE EXCEPTION 'credential scrub failed: % rows still populated', remaining; END IF; END $scrub$; """ def _redact(message: str) -> str: for secret in (settings.supabase_db_password, settings.with_db_password): if secret: message = message.replace(secret, "***") return message def _sidecar_dsn() -> str: """libpq/psycopg DSN for the sidecar (strip the SQLAlchemy driver tag).""" if not settings.with_db_url: raise RuntimeError("WITH_DB_URL is not set") return settings.with_db_url.replace("postgresql+psycopg://", "postgresql://", 1) def _seconds_until(hhmm: str) -> float: hour, minute = (int(part) for part in hhmm.split(":")) now = datetime.now() target = now.replace(hour=hour, minute=minute, second=0, microsecond=0) if target <= now: target += timedelta(days=1) return (target - now).total_seconds() # --- pipeline steps ----------------------------------------------------- def _pg_dump(extra: list[str], out_path: Path) -> None: if shutil.which("pg_dump") is None: raise RuntimeError("pg_dump not found — run inside the app container " "(it ships postgresql-client-17)") cmd = [ "pg_dump", "--no-owner", "--no-privileges", "--section=pre-data", "--section=data", "--host", settings.supabase_db_host, "--port", str(settings.supabase_db_port), "--username", settings.supabase_db_user, *extra, "--file", str(out_path), settings.supabase_db_name, ] env = dict(os.environ, PGPASSWORD=settings.supabase_db_password) result = subprocess.run(cmd, env=env, capture_output=True, text=True, timeout=600) if result.returncode != 0: raise RuntimeError(f"pg_dump failed: {_redact(result.stderr.strip()[-1500:])}") def _dump_source(tmpdir: Path) -> dict: if not (settings.supabase_db_host and settings.supabase_db_user and settings.supabase_db_password): raise RuntimeError("SUPABASE_DB_* not configured (.supabase-export.env)") _pg_dump(["--table=auth.users"], tmpdir / "auth_users.sql") _pg_dump(["--schema=public"], tmpdir / "public.sql") return { "auth_users_bytes": (tmpdir / "auth_users.sql").stat().st_size, "public_bytes": (tmpdir / "public.sql").stat().st_size, } def _rewrite_into(src: Path, dst: IO[str]) -> None: with src.open("r", encoding="utf-8") as fh: for line in rewrite_sql(fh): dst.write(line) def _load_into_legacy(tmpdir: Path) -> dict: load_sql = tmpdir / "load.sql" with load_sql.open("w", encoding="utf-8") as fh: fh.write(_HEADER_SQL) _rewrite_into(tmpdir / "auth_users.sql", fh) _rewrite_into(tmpdir / "public.sql", fh) fh.write(_scrub_sql()) result = subprocess.run( ["psql", _sidecar_dsn(), "--single-transaction", "-v", "ON_ERROR_STOP=1", "--quiet", "--no-psqlrc", "--file", str(load_sql)], capture_output=True, text=True, timeout=600, ) if result.returncode != 0: raise RuntimeError(f"psql load failed: {_redact(result.stderr.strip()[-1500:])}") return {"load_sql_bytes": load_sql.stat().st_size} def _record_row_counts(status: dict) -> dict: with get_engine().connect() as conn: tables = [ row[0] for row in conn.execute(text( "SELECT table_name FROM information_schema.tables " "WHERE table_schema = 'legacy' AND table_type = 'BASE TABLE' " "ORDER BY table_name" )) ] counts = { table: conn.execute(text(f'SELECT count(*) FROM legacy."{table}"')).scalar_one() for table in tables } status["row_counts"] = counts return {"tables": len(counts), "total_rows": sum(counts.values())} def _out_files(out_dir: Path) -> tuple[Path, Path]: day = datetime.now().date().isoformat() return out_dir / f"legacy-{day}.json", out_dir / f"new-{day}.json" def _run_extractor(out_dir: Path) -> dict: if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) try: from harness.extract_legacy import main as extract_main except ImportError as exc: # harness/ is mounted, owned by workstream C raise RuntimeError(f"harness/extract_legacy.py not importable: {exc}") from exc out_file = _out_files(out_dir)[0] rc = extract_main(["--dsn", _sidecar_dsn(), "--schema", "legacy", "--out", str(out_file)]) if rc != 0: raise RuntimeError(f"extractor exited {rc}") return {"out_file": out_file.name, "bytes": out_file.stat().st_size} def _run_transforms() -> dict: """Legacy -> R1 transforms (05 step 2) against the sidecar.""" from app.services.migrate import run_all # books_source="auto" folds the PlusROI sheet-books import into the # nightly rebuild (R2, ADR #016) while that tenant is SHEET-owned. It # shares the transform transaction: a sheet reconciliation failure or # missing/shrunken export FAILS the run (rolls back — ADR #019). An # APP-owned tenant (books_owner="app", ADR #021) is skipped and its # rows preserved by the tenant-scoped reset. report = run_all(engine=get_engine(), books_source="auto") detail = { "ok": report.ok, "straggler_count": report.straggler_count, "warnings": len(report.warnings), "errors": report.errors[:5], "steps": { s.name: {"source": s.source_count, "written": s.written} for s in report.steps }, } if not report.ok: raise RuntimeError(f"transforms failed: {report.errors[:3]}") return detail def _run_extract_new(out_dir: Path) -> dict: if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from harness.extract_new import main as extract_new_main out_file = _out_files(out_dir)[1] rc = extract_new_main(["--dsn", _sidecar_dsn(), "--out", str(out_file)]) if rc != 0: raise RuntimeError(f"new-side extractor exited {rc}") return {"out_file": out_file.name, "bytes": out_file.stat().st_size} def _run_compare(out_dir: Path) -> dict: if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from harness import compare as compare_mod old_file, new_file = _out_files(out_dir) old_doc = json.loads(old_file.read_text(encoding="utf-8")) new_doc = json.loads(new_file.read_text(encoding="utf-8")) allowlist = compare_mod.load_allowlist(compare_mod.DEFAULT_ALLOWLIST) report = compare_mod.run_compare(old_doc, new_doc, allowlist) rendered = compare_mod.render_report( report, old_doc.get("meta", {}), new_doc.get("meta", {}) ) log.info("harness compare:\n%s", rendered) detail = { "result": "GREEN" if report.ok else "RED", "failures": len(report.failures), "failure_paths": [d.path for d in report.failures[:10]], "allowed": sum(1 for d in report.diffs if d.allowed), } if not report.ok: raise RuntimeError( f"harness compare RED — {len(report.failures)} unexplained diff(s);" f" first: {detail['failure_paths'][:3]}" ) return detail def legacy_schema_present() -> bool: """Does the sidecar already hold a landed ``legacy`` snapshot?""" with get_engine().connect() as conn: return conn.execute(text( "SELECT EXISTS (SELECT 1 FROM information_schema.tables" " WHERE table_schema = 'legacy' AND table_name = 'time_entries')" )).scalar_one() # --- orchestration ------------------------------------------------------ def _step(status: dict, name: str, fn, *args) -> bool: t0 = time.monotonic() try: detail = fn(*args) or {} status["steps"][name] = { "ok": True, "duration_s": round(time.monotonic() - t0, 2), **detail } return True except Exception as exc: status["steps"][name] = { "ok": False, "duration_s": round(time.monotonic() - t0, 2), "error": _redact(str(exc))[:2000], } status["error"] = f"{name}: {_redact(str(exc))}"[:2000] return False def _append_status(out_dir: Path, status: dict) -> None: out_dir.mkdir(parents=True, exist_ok=True) with (out_dir / "nightly-status.jsonl").open("a", encoding="utf-8") as fh: fh.write(json.dumps(status, separators=(",", ":"), default=str) + "\n") def _backup_sidecar(out_dir: Path) -> dict: """Dump the sidecar's public schema (the books) to the mounted backups volume with a 14-file rotation. This is the app-side half of the backup regime (05-migration-plan); the host/NAS rotation picks the files up from ./backups on the project directory.""" backups = ROOT / "backups" backups.mkdir(mode=0o750, exist_ok=True) stamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S") out_path = backups / f"with-public-{stamp}.sql.gz" dsn = _sidecar_dsn() dump = subprocess.run( ["pg_dump", "--dbname", dsn, "--schema", "public", "--no-owner", "--no-privileges"], capture_output=True, timeout=600, ) if dump.returncode != 0: raise RuntimeError(f"sidecar pg_dump failed: {_redact(dump.stderr.decode()[-400:])}") with gzip.open(out_path, "wb") as fh: fh.write(dump.stdout) kept = sorted(backups.glob("with-public-*.sql.gz")) for old in kept[:-14]: old.unlink() return {"file": out_path.name, "bytes": out_path.stat().st_size, "retained": min(len(kept), 14)} def run_snapshot(out_dir: Path | str | None = None) -> dict[str, Any]: """Run the full pipeline once. Idempotent; always appends a status line.""" t0 = time.monotonic() out = Path(out_dir) if out_dir else ROOT / "harness" / "out" status: dict[str, Any] = { "ts": datetime.now(UTC).isoformat(timespec="seconds"), "ok": False, "duration_s": None, "row_counts": None, "steps": {}, "error": None, } lock = (Path(tempfile.gettempdir()) / "with-snapshot.lock").open("w") try: try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: status["error"] = "skipped: another snapshot run is in progress" return status with tempfile.TemporaryDirectory(prefix="with-snapshot-") as tmp: tmpdir = Path(tmp) ok = _step(status, "dump", _dump_source, tmpdir) ok = ok and _step(status, "load_scrub", _load_into_legacy, tmpdir) ok = ok and _step(status, "row_counts", _record_row_counts, status) ok = ok and _step(status, "extract", _run_extractor, out) # the rehearsal half (05: transform -> extract both -> compare) ok = ok and _step(status, "transform", _run_transforms) ok = ok and _step(status, "extract_new", _run_extract_new, out) ok = ok and _step(status, "compare", _run_compare, out) ok = ok and _step(status, "backup_sidecar", _backup_sidecar, out) status["ok"] = ok finally: status["duration_s"] = round(time.monotonic() - t0, 2) _append_status(out, status) lock.close() return status async def snapshot_on_boot() -> None: """Boot-time rehearsal (M2): if the sidecar has no ``legacy`` schema yet — the nightly hasn't fired since the volume was created — or RUN_SNAPSHOT_ON_START=1, run the full pipeline immediately so the transforms have data. Fired as a background task: startup never blocks on it.""" try: need = settings.run_snapshot_on_start or not await asyncio.to_thread( legacy_schema_present ) except Exception: log.exception("boot snapshot check failed — skipping boot run") return if not need: return log.info("boot: legacy snapshot missing or RUN_SNAPSHOT_ON_START set — " "running the snapshot+rehearsal pipeline now") try: status = await asyncio.to_thread(run_snapshot) log.info("boot snapshot finished ok=%s error=%s", status["ok"], status["error"]) except Exception: log.exception("boot snapshot crashed") async def snapshot_scheduler() -> None: """id-auth-style sleep loop — created as a background task by app.main.""" log.info("nightly snapshot scheduled daily at %s", settings.snapshot_time) while True: await asyncio.sleep(_seconds_until(settings.snapshot_time)) log.info("nightly snapshot starting") try: status = await asyncio.to_thread(run_snapshot) log.info("nightly snapshot finished ok=%s error=%s", status["ok"], status["error"]) except Exception: # keep the loop alive; the status file has detail log.exception("nightly snapshot crashed") if __name__ == "__main__": result = run_snapshot() print(json.dumps(result, indent=2, default=str)) raise SystemExit(0 if result["ok"] else 1)