#!/usr/bin/env python3 """The migration regression oracle — v1 snapshot vs v2 Postgres. docker exec garden2-app python scripts/diff_harness.py \ --snapshot /data/v1-snapshot/snapshot_v1.json \ [--media-manifest /data/v1-snapshot/media_manifest_v1.json] \ [--data-root /data] Nine named invariants (per the approved plan). ZERO failures = cutover gate. I1_ROW_COUNTS per-table row counts match exactly I2_ROW_EQUALITY per-row, per-column equality (canonicalized), by id I3_FK_INTEGRITY polymorphic + real FK references resolve in PG I4_MEDIA_FILES every referenced media file exists w/ matching sha256 I5_ARTIFACT_FILES every artifacts.filename exists w/ matching sha256 I6_TARGET_TRIPLES field_note_targets triple set + is_primary identical I7_DERIVED_COUNTS per-area plant counts + per-planting counts match I8_NOTES_DERIVATION notes row per bodied v1 note; body byte-equal; note_targets mirror field_note_targets I9_SEQUENCES nextval > max(id) for every id table """ from __future__ import annotations import argparse import hashlib import json import os import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from canonical import V1_COLUMN_KINDS, V2_ONLY_COLUMNS, canon_row # noqa: E402 import psycopg # noqa: E402 from psycopg.rows import dict_row # noqa: E402 FAILURES: list[str] = [] PASSES: list[str] = [] def check(invariant: str, ok: bool, detail: str = "") -> None: if ok: PASSES.append(invariant) else: FAILURES.append(f"{invariant}: {detail}") print(f"FAIL {invariant}: {detail}") def fetch_table(cur, table: str) -> list[dict]: order = "area_id, station_id" if table == "area_stations" else "id" cur.execute(f"SELECT * FROM {table} ORDER BY {order}") return cur.fetchall() def main() -> None: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--snapshot", required=True) ap.add_argument("--media-manifest", default="") ap.add_argument("--data-root", default="/data") args = ap.parse_args() with open(args.snapshot, encoding="utf-8") as f: snap = json.load(f) v1_tables: dict = snap["tables"] # v1 ran SQLite with FK enforcement OFF; orphaned child rows (targets of # deleted notes, messages of deleted sessions) are NOT migrated. Apply # the same rule to the v1 side here so the comparison is apples-to-apples. # This is a documented deviation, printed loudly, never silent. v1_note_ids = {r["id"] for r in v1_tables["field_notes"]} v1_session_ids = {r["id"] for r in v1_tables["chat_sessions"]} fnt_all = v1_tables["field_note_targets"] fnt_kept = [r for r in fnt_all if r["field_note_id"] in v1_note_ids] if len(fnt_kept) != len(fnt_all): print(f"NOTE: excluding {len(fnt_all) - len(fnt_kept)} orphaned " f"field_note_targets from v1 side (FK-off wart)") v1_tables["field_note_targets"] = fnt_kept cm_all = v1_tables["chat_messages"] cm_kept = [r for r in cm_all if r["session_id"] in v1_session_ids] if len(cm_kept) != len(cm_all): print(f"NOTE: excluding {len(cm_all) - len(cm_kept)} orphaned " f"chat_messages from v1 side (FK-off wart)") v1_tables["chat_messages"] = cm_kept dsn = os.environ.get("DATABASE_URL", "").replace( "postgresql+psycopg://", "postgresql://" ) pg = psycopg.connect(dsn, row_factory=dict_row) cur = pg.cursor() # ---------- I1 + I2: counts + row equality ---------- for table, v1_rows in sorted(v1_tables.items()): v2_rows = fetch_table(cur, table) check( "I1_ROW_COUNTS", len(v1_rows) == len(v2_rows), f"{table}: v1={len(v1_rows)} v2={len(v2_rows)}", ) if len(v1_rows) != len(v2_rows): continue # column-set check: v2 may only add the explicit allowlist if v1_rows: v1_cols = set(v1_rows[0].keys()) v2_cols = set(v2_rows[0].keys()) allowed_extra = V2_ONLY_COLUMNS.get(table, set()) unexpected_extra = v2_cols - v1_cols - allowed_extra missing = v1_cols - v2_cols check( "I2_ROW_EQUALITY", not unexpected_extra and not missing, f"{table}: unexpected v2 cols={sorted(unexpected_extra)} " f"missing={sorted(missing)}", ) mismatches = 0 first_detail = "" for r1, r2 in zip(v1_rows, v2_rows): c1 = canon_row(table, r1) c2 = canon_row(table, r2, drop_v2_only=True) if c1 != c2: mismatches += 1 if not first_detail: diffs = { k: (c1.get(k), c2.get(k)) for k in set(c1) | set(c2) if c1.get(k) != c2.get(k) } first_detail = f"{table} id={r1.get('id')}: {diffs}" check("I2_ROW_EQUALITY", mismatches == 0, f"{table}: {mismatches} row mismatches; first: {first_detail}") # ---------- I3: FK integrity ---------- fk_checks = [ ("plants", "species_id", "species"), ("plants", "variety_id", "varieties"), ("plants", "area_id", "areas"), ("plants", "planting_id", "plantings"), ("varieties", "species_id", "species"), ("plantings", "year_id", "years"), ("areas", "parent_id", "areas"), ("field_notes", "parent_id", "field_notes"), ("field_note_targets", "field_note_id", "field_notes"), ("chat_messages", "session_id", "chat_sessions"), ] for table, col, target in fk_checks: cur.execute( f"""SELECT COUNT(*) AS n FROM {table} t WHERE t.{col} IS NOT NULL AND NOT EXISTS (SELECT 1 FROM {target} x WHERE x.id = t.{col})""" ) n = cur.fetchone()["n"] check("I3_FK_INTEGRITY", n == 0, f"{table}.{col} -> {target}: {n} dangling") # areas.parent_id acyclic cur.execute( """WITH RECURSIVE walk(id, path, cycle) AS ( SELECT id, ARRAY[id], false FROM areas UNION ALL SELECT a.parent_id, path || a.parent_id, a.parent_id = ANY(path) FROM walk w JOIN areas a ON a.id = w.id WHERE a.parent_id IS NOT NULL AND NOT cycle ) SELECT COUNT(*) AS n FROM walk WHERE cycle""" ) check("I3_FK_INTEGRITY", cur.fetchone()["n"] == 0, "areas.parent_id cycle") # ---------- I4 + I5: media + artifact files ---------- def file_sha256(path: str) -> str: h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() manifest = {} if args.media_manifest: with open(args.media_manifest, encoding="utf-8") as f: manifest = json.load(f)["files"] media_refs: set = set() cur.execute("SELECT audio_path, photo_paths FROM field_notes") for row in cur.fetchall(): if row["audio_path"]: media_refs.add(row["audio_path"]) for p in (row["photo_paths"] or "").split("|"): if p: media_refs.add(p) cur.execute("SELECT photo_paths FROM chat_messages") for row in cur.fetchall(): for p in (row["photo_paths"] or "").split("|"): if p: media_refs.add(p) for table in ("areas", "plantings", "plants", "species", "varieties", "watering_stations", "supplies", "years"): cur.execute(f"SELECT featured_image_path FROM {table}") for row in cur.fetchall(): if row["featured_image_path"]: media_refs.add(row["featured_image_path"]) missing_files = 0 hash_mismatch = 0 first = "" for rel in sorted(media_refs): abs_path = os.path.join(args.data_root, "notes", rel) if not os.path.isfile(abs_path): missing_files += 1 first = first or f"missing {rel}" continue key = f"notes/{rel}" if manifest and key in manifest: if file_sha256(abs_path) != manifest[key]["sha256"]: hash_mismatch += 1 first = first or f"sha256 mismatch {rel}" check("I4_MEDIA_FILES", missing_files == 0 and hash_mismatch == 0, f"{missing_files} missing, {hash_mismatch} hash mismatches " f"of {len(media_refs)} refs; first: {first}") cur.execute("SELECT filename FROM artifacts") art_missing = 0 for row in cur.fetchall(): p = os.path.join(args.data_root, "artifacts", row["filename"]) if not os.path.isfile(p): art_missing += 1 elif manifest: key = f"artifacts/{row['filename']}" if key in manifest and file_sha256(p) != manifest[key]["sha256"]: art_missing += 1 check("I5_ARTIFACT_FILES", art_missing == 0, f"{art_missing} bad artifacts") # ---------- I6: target triples ---------- v1_triples = { (r["field_note_id"], r["target_type"], r["target_id"], bool(r["is_primary"])) for r in v1_tables["field_note_targets"] } cur.execute( "SELECT field_note_id, target_type, target_id, is_primary FROM field_note_targets" ) v2_triples = { (r["field_note_id"], r["target_type"], r["target_id"], bool(r["is_primary"])) for r in cur.fetchall() } check("I6_TARGET_TRIPLES", v1_triples == v2_triples, f"v1-only={len(v1_triples - v2_triples)} v2-only={len(v2_triples - v1_triples)}") # ---------- I7: derived counts ---------- from collections import Counter v1_by_area = Counter( r["area_id"] for r in v1_tables["plants"] if r["area_id"] is not None ) cur.execute( "SELECT area_id, COUNT(*) AS n FROM plants WHERE area_id IS NOT NULL GROUP BY area_id" ) v2_by_area = {r["area_id"]: r["n"] for r in cur.fetchall()} check("I7_DERIVED_COUNTS", dict(v1_by_area) == v2_by_area, "plants-per-area drift") v1_by_planting = Counter( r["planting_id"] for r in v1_tables["plants"] if r["planting_id"] is not None ) cur.execute( "SELECT planting_id, COUNT(*) AS n FROM plants " "WHERE planting_id IS NOT NULL GROUP BY planting_id" ) v2_by_planting = {r["planting_id"]: r["n"] for r in cur.fetchall()} check("I7_DERIVED_COUNTS", dict(v1_by_planting) == v2_by_planting, "plants-per-planting drift") # ---------- I8: notes derivation ---------- v1_bodied = { r["id"]: r["body"] for r in v1_tables["field_notes"] if r["body"] not in (None, "") } cur.execute("SELECT id, source_capture_id, body_md, author FROM notes") v2_notes = {r["id"]: r for r in cur.fetchall()} body_bad = sum( 1 for nid, body in v1_bodied.items() if nid not in v2_notes or v2_notes[nid]["body_md"] != body or v2_notes[nid]["source_capture_id"] != nid or v2_notes[nid]["author"] != "user" ) check("I8_NOTES_DERIVATION", body_bad == 0 and len(v2_notes) == len(v1_bodied), f"v1 bodied={len(v1_bodied)} v2 notes={len(v2_notes)} bad={body_bad}") # note_targets mirror v1_nt = { (r["field_note_id"], r["target_type"], r["target_id"]) for r in v1_tables["field_note_targets"] if r["field_note_id"] in v1_bodied } cur.execute("SELECT note_id, target_type, target_id FROM note_targets") v2_nt = {(r["note_id"], r["target_type"], r["target_id"]) for r in cur.fetchall()} check("I8_NOTES_DERIVATION", v1_nt == v2_nt, f"note_targets mirror: v1-only={len(v1_nt - v2_nt)} v2-only={len(v2_nt - v1_nt)}") # ---------- I9: sequences ---------- for table in list(v1_tables.keys()) + ["notes", "gardens", "users"]: if table == "area_stations": continue cur.execute( f"""SELECT COALESCE(MAX(id), 0) AS mx, pg_sequence_last_value(pg_get_serial_sequence('{table}','id')) AS last FROM {table}""" ) r = cur.fetchone() ok = r["mx"] == 0 or (r["last"] is not None and r["last"] >= r["mx"]) check("I9_SEQUENCES", ok, f"{table}: max(id)={r['mx']} seq={r['last']}") # ---------- summary ---------- print() counts: dict = {} for p in PASSES: counts[p] = counts.get(p, 0) + 1 for inv in sorted(counts): print(f"PASS {inv} ({counts[inv]} checks)") if FAILURES: print(f"\n{len(FAILURES)} FAILURES — cutover gate CLOSED") sys.exit(1) print("\nALL INVARIANTS GREEN — cutover gate open") if __name__ == "__main__": main()