#!/usr/bin/env python3
"""Synthetic restartable-backfill lab. Python 3.11+, standard library.

Actual SQLite commits, parent/child interleaving and abrupt owned worker exit.
Four fictional rows; batches of two. No production database paths, network,
load test, replication, power failure or continuous writer synchronization.
"""
import argparse
import json
import multiprocessing as mp
import os
from pathlib import Path
import platform
import sqlite3
import tempfile

CASES = ("cursor-first", "data-first", "atomic")
EXIT_AFTER_BOUNDARY = 23
BOUND = 4
BATCH = 2


def connect(path):
    db = sqlite3.connect(path, isolation_level=None, timeout=2)
    db.row_factory = sqlite3.Row
    db.execute("PRAGMA foreign_keys=ON")
    db.execute("PRAGMA synchronous=FULL")
    return db


def initialize(path):
    db = connect(path)
    try:
        db.executescript("""
          CREATE TABLE items (id INTEGER PRIMARY KEY, owner TEXT NOT NULL,
            revision INTEGER NOT NULL, new_owner TEXT, new_revision INTEGER);
          CREATE TABLE progress (singleton INTEGER PRIMARY KEY CHECK(singleton=1), cursor INTEGER NOT NULL);
          INSERT INTO progress VALUES (1,0);
          CREATE TABLE pending (item_id INTEGER PRIMARY KEY REFERENCES items(id), reason TEXT NOT NULL);
          CREATE TABLE attempts (id INTEGER PRIMARY KEY, item_id INTEGER REFERENCES items(id), captured_revision INTEGER, result TEXT NOT NULL);
        """)
        db.executemany("INSERT INTO items VALUES (?, 'Queue A',1,NULL,NULL)", [(i,) for i in range(1, BOUND + 1)])
    finally:
        db.close()


def capture(path, pending=False):
    db = connect(path)
    try:
        if pending:
            return [dict(r) for r in db.execute("SELECT i.id,i.owner,i.revision FROM items i JOIN pending p ON p.item_id=i.id ORDER BY i.id LIMIT ?", (BATCH,))]
        cursor = db.execute("SELECT cursor FROM progress").fetchone()[0]
        return [dict(r) for r in db.execute("SELECT id,owner,revision FROM items WHERE id>? AND id<=? ORDER BY id LIMIT ?", (cursor, BOUND, BATCH))]
    finally:
        db.close()


def write_batch(db, captured):
    for row in captured:
        count = db.execute("UPDATE items SET new_owner=?,new_revision=? WHERE id=? AND revision=? AND new_owner IS NULL", (row["owner"], row["revision"], row["id"], row["revision"])).rowcount
        if count:
            result = "applied"
            db.execute("DELETE FROM pending WHERE item_id=?", (row["id"],))
        else:
            current = db.execute("SELECT * FROM items WHERE id=?", (row["id"],)).fetchone()
            if current["new_owner"] == current["owner"] and current["new_revision"] == current["revision"]:
                result = "already_current"
                db.execute("DELETE FROM pending WHERE item_id=?", (row["id"],))
            else:
                result = "needs_reconciliation"
                db.execute("INSERT OR IGNORE INTO pending VALUES (?,'captured revision or target no longer fits')", (row["id"],))
        db.execute("INSERT INTO attempts(item_id,captured_revision,result) VALUES (?,?,?)", (row["id"], row["revision"], result))


def interrupted_worker(path, mode, channel):
    captured = capture(path)
    channel.send(captured)
    if not channel.poll(10) or channel.recv() != "writer committed":
        os._exit(97)
    db = connect(path)
    if mode == "cursor-first":
        db.execute("UPDATE progress SET cursor=?", (captured[-1]["id"],))
        db.execute("BEGIN IMMEDIATE")
        write_batch(db, captured)
        # Intentional abrupt exit with data and pending records uncommitted.
        os._exit(EXIT_AFTER_BOUNDARY)
    db.execute("BEGIN IMMEDIATE")
    write_batch(db, captured)
    if mode == "atomic":
        db.execute("UPDATE progress SET cursor=?", (captured[-1]["id"],))
    db.execute("COMMIT")
    # data-first exits before writing the cursor; atomic committed it with data.
    os._exit(EXIT_AFTER_BOUNDARY)


def change_source(path, owner):
    db = connect(path)
    try:
        db.execute("BEGIN IMMEDIATE")
        db.execute("UPDATE items SET owner=?,revision=revision+1 WHERE id=2", (owner,))
        db.execute("COMMIT")
    finally:
        db.close()


def interrupt(path, mode):
    context = mp.get_context("spawn")
    parent, child = context.Pipe()
    process = context.Process(target=interrupted_worker, args=(str(path), mode, child))
    process.start()
    child.close()
    try:
        if not parent.poll(10):
            raise RuntimeError("worker_capture_timeout")
        captured = parent.recv()
        change_source(path, "Queue B")
        parent.send("writer committed")
        process.join(10)
        if process.is_alive() or process.exitcode != EXIT_AFTER_BOUNDARY:
            raise RuntimeError("worker_boundary_not_observed")
        return {"pid": process.pid, "exit_code": process.exitcode, "captured_batch": captured}
    finally:
        if process.is_alive():
            process.terminate()
            process.join(5)
        parent.close()
        process.close()


def resume(path, pending=False):
    # At most four fixture records; never an unbounded retry loop.
    for _ in range(BOUND):
        rows = capture(path, pending)
        if not rows:
            return
        db = connect(path)
        try:
            db.execute("BEGIN IMMEDIATE")
            write_batch(db, rows)
            if not pending:
                db.execute("UPDATE progress SET cursor=?", (rows[-1]["id"],))
            db.execute("COMMIT")
        except BaseException:
            if db.in_transaction:
                db.execute("ROLLBACK")
            raise
        finally:
            db.close()
        if pending:
            return  # One bounded reconciliation pass; unresolved targets stay visible.
    raise RuntimeError("fixture_batch_budget_exceeded")


def inspect(path):
    db = connect(path)
    try:
        if db.execute("PRAGMA integrity_check").fetchone()[0] != "ok":
            raise RuntimeError("integrity_failed")
        rows = [dict(r) for r in db.execute("SELECT * FROM items ORDER BY id")]
        for row in rows:
            row["matches_current_source"] = row["new_owner"] == row["owner"] and row["new_revision"] == row["revision"]
        return {"cursor": db.execute("SELECT cursor FROM progress").fetchone()[0],
                "fixed_upper_id": BOUND, "rows": rows,
                "unmatched_ids": [r["id"] for r in rows if not r["matches_current_source"]],
                "pending": [dict(r) for r in db.execute("SELECT * FROM pending ORDER BY item_id")],
                "attempts": [dict(r) for r in db.execute("SELECT * FROM attempts ORDER BY id")]}
    finally:
        db.close()


def run_case(path, mode, late_change):
    initialize(path)
    worker = interrupt(path, mode)
    stopped = inspect(path)
    resume(path)
    scanned = inspect(path)
    if late_change:
        change_source(path, "Queue C")
    before_reconcile = inspect(path)
    resume(path, pending=True)
    final = inspect(path)
    return {"case": mode, "worker": worker, "late_change": late_change,
            "after_worker_exit": stopped, "after_scan_resume": scanned,
            "before_reconciliation": before_reconcile, "final": final}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    choice = parser.add_mutually_exclusive_group(required=True)
    choice.add_argument("--all", action="store_true")
    choice.add_argument("--case", choices=CASES)
    parser.add_argument("--late-change", action="store_true", help="Atomic case only: change pending row 2 again before reconciliation")
    args = parser.parse_args()
    if args.late_change and args.case != "atomic":
        parser.error("--late-change requires --case atomic")
    with tempfile.TemporaryDirectory(prefix="lockedin-backfill-") as folder:
        reports = [run_case(Path(folder) / (mode + ".sqlite"), mode, args.late_change) for mode in CASES if args.all or args.case == mode]
        print(json.dumps({"python_version": platform.python_version(), "sqlite_version": sqlite3.sqlite_version, "batch_size": BATCH, "cases": reports}, indent=2))


if __name__ == "__main__":
    main()