#!/usr/bin/env python3
"""Original synthetic queue recovery workshop. Python 3.11+, standard library.

Actual temporary SQLite transactions; single-worker delivery/settlement model.
No broker, network, clock, process crash, authorization or remote business action.
Run --all, --case reader-fixed, or --case settlement-fixed --replay-key new.
No persistent paths or external destinations are accepted.
"""
import argparse
from contextlib import closing
import hashlib
import json
from pathlib import Path
import platform
import sqlite3
import tempfile

CASES = ("unchanged", "reader-fixed", "settlement-fixed")


def connect(path):
    db = sqlite3.connect(path)
    db.row_factory = sqlite3.Row
    db.execute("PRAGMA foreign_keys=ON")
    return db


def fingerprint(body):
    return hashlib.sha256(body.encode("utf-8")).hexdigest()


def initialize(path, case):
    with closing(connect(path)) as db, db:
        db.executescript("""
          CREATE TABLE deliveries (
            id INTEGER PRIMARY KEY, source_id INTEGER REFERENCES deliveries(id),
            operation_key TEXT NOT NULL, body TEXT NOT NULL,
            state TEXT NOT NULL CHECK(state IN ('ready','settled','quarantined','moved')),
            attempts INTEGER NOT NULL DEFAULT 0);
          CREATE TABLE effects (id INTEGER PRIMARY KEY, operation_key TEXT NOT NULL, body TEXT NOT NULL);
          CREATE TABLE receipts (
            operation_key TEXT PRIMARY KEY, body_digest TEXT NOT NULL,
            effect_id INTEGER NOT NULL UNIQUE REFERENCES effects(id));
          CREATE TABLE attempts (
            id INTEGER PRIMARY KEY, delivery_id INTEGER REFERENCES deliveries(id),
            attempt INTEGER, reader INTEGER, result TEXT, effect_id INTEGER,
            error TEXT, settled INTEGER NOT NULL);
          CREATE TABLE quarantine (
            id INTEGER PRIMARY KEY, delivery_id INTEGER NOT NULL UNIQUE REFERENCES deliveries(id),
            reason TEXT NOT NULL, attempts INTEGER NOT NULL);
        """)
        for key, schema in [("JOB-1", 1 if case == "settlement-fixed" else 2), ("JOB-2", 1)]:
            body = json.dumps({"schema": schema, "case": key, "draft": "Inspect synthetic work."}, sort_keys=True, separators=(",", ":"))
            db.execute("INSERT INTO deliveries(operation_key,body,state) VALUES (?,?,'ready')", (key, body))


def apply(path, key, body, reader):
    # The receipt and local effect share one transaction. Settlement is later.
    payload = json.loads(body)
    if payload["schema"] > reader:
        raise ValueError("unsupported_schema")
    with closing(connect(path)) as db, db:
        db.execute("BEGIN IMMEDIATE")
        prior = db.execute("SELECT * FROM receipts WHERE operation_key=?", (key,)).fetchone()
        if prior:
            if prior["body_digest"] != fingerprint(body):
                raise ValueError("operation_payload_conflict")
            return "reused", prior["effect_id"]
        effect = db.execute("INSERT INTO effects(operation_key,body) VALUES (?,?)", (key, body)).lastrowid
        db.execute("INSERT INTO receipts VALUES (?,?,?)", (key, fingerprint(body), effect))
        return "applied", effect


def step(path, reader, limit, omit_settlement=False):
    with closing(connect(path)) as db, db:
        db.execute("BEGIN IMMEDIATE")
        row = db.execute("SELECT * FROM deliveries WHERE state='ready' ORDER BY id LIMIT 1").fetchone()
        if row is None:
            return False
        attempt = row["attempts"] + 1
        db.execute("UPDATE deliveries SET attempts=? WHERE id=?", (attempt, row["id"]))
    result, effect, error = "not_applied", None, None
    try:
        result, effect = apply(path, row["operation_key"], row["body"], reader)
        # Explicitly omitted settlement models a separate broker boundary.
        if omit_settlement and row["id"] == 1:
            error = "settlement_unavailable"
    except ValueError as failure:
        error = str(failure)
    with closing(connect(path)) as db, db:
        db.execute("BEGIN IMMEDIATE")
        if error is None:
            db.execute("UPDATE deliveries SET state='settled' WHERE id=?", (row["id"],))
        elif attempt >= limit:
            db.execute("UPDATE deliveries SET state='quarantined' WHERE id=?", (row["id"],))
            db.execute("INSERT INTO quarantine(delivery_id,reason,attempts) VALUES (?,?,?)", (row["id"], error, attempt))
        db.execute("INSERT INTO attempts(delivery_id,attempt,reader,result,effect_id,error,settled) VALUES (?,?,?,?,?,?,?)", (row["id"], attempt, reader, result, effect, error, int(error is None)))
    return True


def drain(path, reader, limit, omit_settlement=False):
    # Explicit finite budget: two initial deliveries and at most one replay.
    for _ in range(3 * limit + 1):
        if not step(path, reader, limit, omit_settlement):
            return
    raise RuntimeError("fixture_attempt_budget_exceeded")


def replay(path, key_mode):
    with closing(connect(path)) as db, db:
        db.execute("BEGIN IMMEDIATE")
        row = db.execute("SELECT * FROM deliveries WHERE id=1 AND state='quarantined'").fetchone()
        if row is None:
            raise ValueError("original_not_quarantined")
        key = row["operation_key"] if key_mode == "original" else "JOB-1-NEW"
        replay_id = db.execute("INSERT INTO deliveries(source_id,operation_key,body,state) VALUES (?,?,?,'ready')", (row["id"], key, row["body"])).lastrowid
        db.execute("UPDATE deliveries SET state='moved' WHERE id=?", (row["id"],))
        return replay_id


def inspect(path):
    # Fresh reads of committed tables, including the retained quarantine history.
    with closing(connect(path)) as db:
        if db.execute("PRAGMA integrity_check").fetchone()[0] != "ok":
            raise RuntimeError("integrity_failed")
        deliveries = [dict(r) for r in db.execute("SELECT * FROM deliveries ORDER BY id")]
        return {"deliveries": deliveries,
                "active_quarantine": sum(d["state"] == "quarantined" for d in deliveries),
                "ready": sum(d["state"] == "ready" for d in deliveries),
                "effects": [dict(r) for r in db.execute("SELECT * FROM effects ORDER BY id")],
                "receipts": [dict(r) for r in db.execute("SELECT * FROM receipts ORDER BY operation_key")],
                "attempts": [dict(r) for r in db.execute("SELECT * FROM attempts ORDER BY id")],
                "quarantine_history": [dict(r) for r in db.execute("SELECT * FROM quarantine ORDER BY id")]}


def run_case(path, case, limit, key_mode):
    initialize(path, case)
    initial = inspect(path)
    drain(path, 1, limit, omit_settlement=case == "settlement-fixed")
    before = inspect(path)
    replay_id = replay(path, key_mode)
    moved = inspect(path)
    drain(path, 2 if case == "reader-fixed" else 1, limit)
    final = inspect(path)
    return {"case": case, "attempt_limit_per_delivery": limit,
            "replay_key_mode": key_mode, "replay_delivery_id": replay_id,
            "initial": initial, "before_replay": before, "after_move_before_work": moved, "final": final}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    selection = parser.add_mutually_exclusive_group(required=True)
    selection.add_argument("--all", action="store_true")
    selection.add_argument("--case", choices=CASES)
    parser.add_argument("--attempt-limit", type=int, choices=range(2, 6), default=3)
    parser.add_argument("--replay-key", choices=("original", "new"), default="original")
    args = parser.parse_args()
    with tempfile.TemporaryDirectory(prefix="lockedin-queue-recovery-") as folder:
        reports = [run_case(Path(folder) / (case + ".sqlite"), case, args.attempt_limit, args.replay_key) for case in CASES if args.all or args.case == case]
        print(json.dumps({"mode": "single-worker SQLite delivery model", "python_version": platform.python_version(), "sqlite_version": sqlite3.sqlite_version, "cases": reports}, indent=2))


if __name__ == "__main__":
    main()