tooling: verified backup, state snapshot, retargetable apply validator

Promotion prep for the contract-apply cutover, which unlike the quick-sell promotion moves BOTH binaries and applies a schema migration. fifa17-promotion-backup.py uses SQLite's online backup API, not cp. Production runs WAL with a routinely uncheckpointed WAL (515 KB at capture time); copying the main file alone is not atomic against a live writer and carries no guarantee the WAL holds no newer committed state. Emits a checksummed backup, a metadata record and a RESTORE-*.sh that removes the stale -wal/-shm BEFORE restoring -- omit that and SQLite replays the old journal over the file you just put back, resurrecting the state you were abandoning. fifa17-promotion-snapshot.py is read-only (mode=ro) and counts EVERY table rather than a hand-picked list, so a delta cannot hide in a table nobody thought to name. It also fingerprints the ownership rows, catching a row silently rewritten when counts alone would match. fifa17-contract-apply-validate.py gains --host/--db so one tool serves staging, the migration rehearsal and the production acceptance run. Defaults stay staging: there is deliberately no production default, so a bare invocation cannot touch production.
2026-08-22 18:40:12 +00:00
parent 6c97bc4e2b
commit f0c6dcf238
3 changed files with 264 additions and 1 deletions
@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""READ-ONLY state capture of an OpenFUT Core SQLite DB, for promotion gating.
+
+Opened `mode=ro` and never written to, so it is safe against live production.
+Emits a deterministic JSON document: run it before a promotion and again after,
+then `diff` the two. Every table is counted, so a delta cannot hide in a table
+nobody thought to list.
+
+Usage:
+    fifa17-promotion-snapshot.py <db-path> [out.json]
+"""
+import hashlib
+import json
+import sqlite3
+import sys
+
+
+def snapshot(db_path):
+    con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
+    con.row_factory = sqlite3.Row
+    one = lambda s: con.execute(s).fetchone()[0]
+
+    snap = {
+        "db_path": db_path,
+        "journal_mode": one("PRAGMA journal_mode"),
+        "page_size": one("PRAGMA page_size"),
+        "page_count": one("PRAGMA page_count"),
+        "sqlite_version": sqlite3.sqlite_version,
+    }
+
+    # Schema version + the full applied-migration ledger. A promotion that
+    # claims "0028 applied" must show it here, with success=1.
+    snap["migration_max"] = one("SELECT MAX(version) FROM _sqlx_migrations")
+    snap["migration_count"] = one("SELECT COUNT(*) FROM _sqlx_migrations")
+    snap["migrations_failed"] = one(
+        "SELECT COUNT(*) FROM _sqlx_migrations WHERE success <> 1")
+
+    # Every table, counted. Deliberately not a hand-picked list.
+    tables = [r[0] for r in con.execute(
+        "SELECT name FROM sqlite_master WHERE type='table' "
+        "AND name NOT LIKE 'sqlite_%' ORDER BY name")]
+    snap["table_counts"] = {t: one(f'SELECT COUNT(*) FROM "{t}"') for t in tables}
+
+    # The economically load-bearing figures, called out so a diff is readable.
+    snap["coins"] = dict(con.execute("SELECT id, coins FROM clubs ORDER BY id").fetchall())
+    snap["owned_total"] = one("SELECT COUNT(*) FROM owned_cards")
+    snap["owned_by_kind"] = dict(
+        con.execute("SELECT content_kind, COUNT(*) FROM owned_cards "
+                    "GROUP BY 1 ORDER BY 1").fetchall())
+    snap["squad_players"] = one("SELECT COUNT(*) FROM squad_players")
+    snap["market_listings"] = one("SELECT COUNT(*) FROM market_listings")
+    snap["market_listings_sold"] = one("SELECT COUNT(*) FROM market_listings WHERE sold = 1")
+    snap["market_listings_npc"] = one("SELECT COUNT(*) FROM market_listings WHERE is_npc = 1")
+    snap["game_entity_ext"] = one("SELECT COUNT(*) FROM game_entity_ext")
+    snap["consumable_applications"] = one("SELECT COUNT(*) FROM consumable_applications")
+    snap["packs_total"] = one("SELECT COUNT(*) FROM packs")
+    snap["packs_unopened"] = one("SELECT COUNT(*) FROM packs WHERE opened = 0")
+    snap["match_completions"] = one("SELECT COUNT(*) FROM match_completions")
+
+    # Present only after 0028. Absent => pre-0028 DB, which is itself the signal.
+    cols = [r[1] for r in con.execute("PRAGMA table_info(owned_cards)")]
+    snap["owned_cards_columns"] = cols
+    snap["has_0028"] = "contract_matches" in cols
+    if snap["has_0028"]:
+        snap["contract_matches_set"] = one(
+            "SELECT COUNT(*) FROM owned_cards WHERE contract_matches IS NOT NULL")
+        snap["contract_matches_sum"] = one(
+            "SELECT COALESCE(SUM(contract_matches), 0) FROM owned_cards")
+
+    # A content fingerprint over ownership: catches a row silently rewritten
+    # even when every count stays identical.
+    h = hashlib.sha256()
+    for r in con.execute(
+            "SELECT id, club_id, card_id, is_loan, loan_matches_remaining, "
+            "chemistry_style, position_override, training_bonus, content_kind, "
+            "quantity FROM owned_cards ORDER BY id"):
+        h.update(("|".join("" if v is None else str(v) for v in r)).encode())
+    snap["owned_cards_fingerprint_pre0028_columns"] = h.hexdigest()
+
+    snap["integrity_check"] = one("PRAGMA integrity_check")
+    snap["foreign_key_check"] = len(con.execute("PRAGMA foreign_key_check").fetchall())
+    con.close()
+    return snap
+
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print(__doc__)
+        raise SystemExit(2)
+    snap = snapshot(sys.argv[1])
+    text = json.dumps(snap, indent=2, sort_keys=True)
+    print(text)
+    if len(sys.argv) > 2:
+        open(sys.argv[2], "w").write(text + "\n")