#!/usr/bin/env python3
"""Synthetic transactional-outbox replay lab; Python 3.11+, standard library.

Two temporary SQLite files represent a caller and a separate local receiver.
The receiver's effect is a database row, not a real dispatch. An owned child
exits after receiver commit and before returning an acknowledgment. No network,
credentials, authentication, production integration, or timing sleeps in the
original cases. Optional --http uses an owned 127.0.0.1 listener for real HTTP
disconnects, an in-flight timeout and receipt lookup; no external host is used.

Run --all, --case guarded, or --self-test. No database paths are accepted.
"""
import argparse
import hashlib
import json
import multiprocessing as mp
import os
from pathlib import Path
import sqlite3
import sys
import tempfile

CASES = ("naive", "guarded", "changed-payload", "scoped-key")
PAYLOAD = {"work_order": "WO-1", "note": "Inspect the isolation valve."}
EXIT_AFTER_COMMIT = 23
CHILD_TIMEOUT = 10


def require(condition, message):
    if not condition:
        raise RuntimeError(message)


def canonical(payload):
    if not isinstance(payload, dict) or set(payload) != {"work_order", "note"}:
        raise ValueError("invalid_payload")
    if not all(isinstance(v, str) and 0 < len(v) <= 512 for v in payload.values()):
        raise ValueError("invalid_payload")
    return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False)


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


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


def initialize(caller_path, receiver_path, guarded):
    db = connect(caller_path)
    try:
        db.executescript("""
            CREATE TABLE intents (
                id INTEGER PRIMARY KEY, tenant TEXT NOT NULL,
                operation_key TEXT NOT NULL, body TEXT NOT NULL);
            CREATE TABLE outbox (
                intent_id INTEGER PRIMARY KEY REFERENCES intents(id),
                state TEXT NOT NULL CHECK(state IN ('pending','acknowledged')),
                receipt TEXT);
        """)
        # One real caller transaction: intent and delivery obligation commit together.
        db.execute("BEGIN IMMEDIATE")
        try:
            db.execute("INSERT INTO intents VALUES (1,?,?,?)", ("north", "OP-1", canonical(PAYLOAD)))
            db.execute("INSERT INTO outbox VALUES (1,'pending',NULL)")
            db.execute("COMMIT")
        except BaseException:
            db.execute("ROLLBACK")
            raise
    finally:
        db.close()
    db = connect(receiver_path)
    try:
        db.executescript("""
            CREATE TABLE effects (
                id INTEGER PRIMARY KEY, tenant TEXT NOT NULL, body TEXT NOT NULL);
            CREATE TABLE receipts (
                effect_id INTEGER PRIMARY KEY REFERENCES effects(id),
                tenant TEXT NOT NULL, operation_key TEXT NOT NULL,
                payload_digest TEXT NOT NULL);
        """)
        if guarded:
            db.execute("CREATE UNIQUE INDEX scoped_operation ON receipts(tenant,operation_key)")
    finally:
        db.close()


def caller_delivery(path):
    db = connect(path)
    try:
        tenant, key, body = db.execute("SELECT tenant,operation_key,body FROM intents WHERE id=1").fetchone()
        return {"tenant": tenant, "operation_key": key, "payload": json.loads(body)}
    finally:
        db.close()


def caller_view(path):
    db = connect(path)
    try:
        state, receipt = db.execute("SELECT state,receipt FROM outbox WHERE intent_id=1").fetchone()
        return {"outbox": state, "outcome": "confirmed" if receipt else "unknown",
                "receipt": json.loads(receipt) if receipt else None}
    finally:
        db.close()


def destination_observer(path):
    """Exercise observer only: delivery/acknowledgment code never uses this read."""
    db = connect(path)
    try:
        require(db.execute("PRAGMA integrity_check").fetchone()[0] == "ok", "receiver_integrity_failed")
        return {"knowledge": "independent observer; unavailable to caller",
                "effects": db.execute("SELECT COUNT(*) FROM effects").fetchone()[0],
                "receipts": db.execute("SELECT COUNT(*) FROM receipts").fetchone()[0]}
    finally:
        db.close()


def receive(path, delivery, guarded, allowed_keys=("OP-1",)):
    # Tenant is a synthetic partition label, not an authenticated identity.
    tenant, key = delivery["tenant"], delivery["operation_key"]
    if tenant not in ("north", "south") or key not in allowed_keys:
        raise ValueError("invalid_scope_or_key")
    body = canonical(delivery["payload"])
    fingerprint = digest(body)
    db = connect(path)
    try:
        db.execute("BEGIN IMMEDIATE")
        try:
            existing = db.execute(
                "SELECT effect_id,payload_digest FROM receipts WHERE tenant=? AND operation_key=?",
                (tenant, key)).fetchone() if guarded else None
            if existing:
                if existing[1] != fingerprint:
                    raise ValueError("key_payload_conflict")
                effect_id, status = existing[0], "replayed"
            else:
                effect_id = db.execute("INSERT INTO effects(tenant,body) VALUES (?,?)", (tenant, body)).lastrowid
                db.execute("INSERT INTO receipts VALUES (?,?,?,?)", (effect_id, tenant, key, fingerprint))
                status = "applied"
            # Guarded effect + receipt belong to this SAME receiver transaction.
            db.execute("COMMIT")
        except BaseException:
            if db.in_transaction:
                db.execute("ROLLBACK")
            raise
        return {"status": status, "receipt": {"tenant": tenant, "operation_key": key,
                "payload_digest": fingerprint, "effect_id": effect_id}}
    finally:
        db.close()


def receiver_child(path, delivery, guarded, lose_ack, channel):
    try:
        result = receive(path, delivery, guarded)
        if lose_ack:
            os._exit(EXIT_AFTER_COMMIT)  # Commit finished; no acknowledgment sent.
        channel.send(result)
    except ValueError as error:
        channel.send({"status": "conflict", "error": str(error)})
    finally:
        channel.close()


def deliver(path, delivery, guarded, lose_ack=False):
    """Local process boundary, not HTTP. Kill/reap only this owned child."""
    context = mp.get_context("spawn")
    incoming, outgoing = context.Pipe(duplex=False)
    process = context.Process(target=receiver_child, args=(path, delivery, guarded, lose_ack, outgoing))
    try:
        process.start()
        outgoing.close()
        process.join(CHILD_TIMEOUT)
        if process.is_alive():
            raise RuntimeError("owned_receiver_child_timed_out")
        try:
            acknowledgment = incoming.recv() if incoming.poll() else None
        except EOFError:
            acknowledgment = None
        expected_exit = EXIT_AFTER_COMMIT if lose_ack else 0
        require(process.exitcode == expected_exit, "unexpected_receiver_child_exit")
        require((acknowledgment is None) == lose_ack, "unexpected_acknowledgment_boundary")
        return {"child_exit": process.exitcode, "acknowledgment": acknowledgment}
    finally:
        outgoing.close()
        incoming.close()
        if process.pid is not None:
            if process.is_alive():
                process.kill()
                process.join(2)
            require(not process.is_alive(), "owned_receiver_child_not_reaped")
            process.close()


def acknowledge(caller_path, result):
    acknowledgment = result["acknowledgment"]
    if acknowledgment is None or acknowledgment["status"] == "conflict":
        return
    original = caller_delivery(caller_path)
    receipt = acknowledgment["receipt"]
    # A receipt for an altered delivery must never acknowledge the original intent.
    require(all(receipt[field] == original[field] for field in ("tenant", "operation_key"))
            and receipt["payload_digest"] == digest(canonical(original["payload"])), "receipt_intent_mismatch")
    db = connect(caller_path)
    try:
        db.execute("UPDATE outbox SET state='acknowledged',receipt=? WHERE intent_id=1",
                   (json.dumps(receipt, sort_keys=True),))
    finally:
        db.close()


def run_case(name):
    guarded = name != "naive"
    with tempfile.TemporaryDirectory(prefix="fde-outbox-") as directory:
        caller_path, receiver_path = (str(Path(directory) / leaf) for leaf in ("caller.sqlite", "receiver.sqlite"))
        initialize(caller_path, receiver_path, guarded)
        original = caller_delivery(caller_path)
        crash = deliver(receiver_path, original, guarded, lose_ack=True)
        acknowledge(caller_path, crash)
        caller_before = caller_view(caller_path)
        observer_before = destination_observer(receiver_path)
        require(caller_before == {"outbox": "pending", "outcome": "unknown", "receipt": None}, "premature_caller_certainty")
        require((observer_before["effects"], observer_before["receipts"]) == (1, 1), "missing_committed_receiver_effect")

        redelivery = caller_delivery(caller_path)
        if name == "changed-payload":
            redelivery["payload"]["note"] = "A different action."
        elif name == "scoped-key":
            redelivery["tenant"] = "south"
        elif name == "guarded":
            redelivery["payload"] = dict(reversed(list(redelivery["payload"].items())))
        replay = deliver(receiver_path, redelivery, guarded)
        if name == "scoped-key":
            try:
                acknowledge(caller_path, replay)
            except RuntimeError as error:
                require(str(error) == "receipt_intent_mismatch", "unexpected_acknowledgment_failure")
                replay["caller_acknowledgment_error"] = str(error)
            else:
                raise RuntimeError("other_partition_receipt_acknowledged_original_intent")
        else:
            acknowledge(caller_path, replay)
        require(caller_delivery(caller_path) == original, "durable_caller_intent_changed")
        if name == "guarded":
            again = deliver(receiver_path, original, guarded)
            require(again["acknowledgment"]["receipt"] == replay["acknowledgment"]["receipt"], "receipt_changed_on_replay")
            replay["same_receipt_on_further_replay"] = True
        caller_after = caller_view(caller_path)
        observer_after = destination_observer(receiver_path)
        expected_effects = 2 if name in ("naive", "scoped-key") else 1
        require(observer_after["effects"] == observer_after["receipts"] == expected_effects, "unexpected_receiver_effect_count")
        if name == "changed-payload":
            require(replay["acknowledgment"] == {"status": "conflict", "error": "key_payload_conflict"}, "changed_payload_not_rejected")
            require(caller_after == caller_before, "conflict_acknowledged_original_intent")
        elif name == "scoped-key":
            require(replay["acknowledgment"]["status"] == "applied"
                    and replay["acknowledgment"]["receipt"]["tenant"] == "south", "other_partition_not_separate")
            require(caller_after == caller_before, "other_partition_changed_original_caller_state")
        else:
            require(caller_after["outbox"] == "acknowledged", "redelivery_not_acknowledged")
            require(replay["acknowledgment"]["status"] == ("applied" if name == "naive" else "replayed"), "unexpected_replay_status")
        return {"case": name, "interrupted_delivery": crash,
                "caller_after_crash": caller_before, "destination_observer_after_crash": observer_before,
                "replay": replay, "caller_after_replay": caller_after,
                "destination_observer_after_replay": observer_after}


HTTP_CASES = ("drop-before-effect", "drop-after-commit", "timeout-in-flight")


def lookup_receipt(path, tenant, key):
    if tenant not in ("north", "south") or key not in ("OP-1", "OP-2"):
        raise ValueError("invalid_scope_or_key")
    db = connect(path)
    try:
        row = db.execute("SELECT effect_id,payload_digest FROM receipts WHERE tenant=? AND operation_key=?", (tenant, key)).fetchone()
        return None if row is None else {"tenant": tenant, "operation_key": key, "effect_id": row[0], "payload_digest": row[1]}
    finally:
        db.close()


class HttpFixture:
    """An owned loopback receiver; gates are fixture controls, never request options."""
    def __init__(self, path, case):
        import http.server
        import socket
        import threading
        import urllib.parse
        self.path, self.case = path, case
        self.entered, self.release, self.finished = (threading.Event() for _ in range(3))
        self.lock = threading.Lock()
        self.first = True
        self.trace, self.errors = [], []
        fixture = self

        class Server(http.server.ThreadingHTTPServer):
            daemon_threads = False
            def get_request(self):
                connection, address = super().get_request()
                connection.settimeout(2)
                return connection, address

        class Handler(http.server.BaseHTTPRequestHandler):
            def log_message(self, *args):
                pass

            def reply(self, status, body):
                data = json.dumps(body, sort_keys=True).encode()
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(data)))
                self.send_header("Cache-Control", "no-store")
                self.end_headers()
                self.wfile.write(data)

            def drop(self):
                self.close_connection = True
                try:
                    self.connection.shutdown(socket.SHUT_RDWR)
                except OSError:
                    pass
                self.connection.close()

            def do_GET(self):
                parsed = urllib.parse.urlsplit(self.path)
                query = urllib.parse.parse_qs(parsed.query)
                if parsed.path != "/receipt" or set(query) != {"tenant", "operation_key"} or any(len(v) != 1 for v in query.values()):
                    self.reply(400, {"error": "invalid_fixture_lookup"}); return
                try:
                    receipt = lookup_receipt(fixture.path, query["tenant"][0], query["operation_key"][0])
                    fixture.record("receipt_lookup", found=receipt is not None, operation_key=query["operation_key"][0])
                    self.reply(200 if receipt else 404, {"status": "found", "receipt": receipt} if receipt else {"status": "not_found"})
                except ValueError:
                    self.reply(400, {"error": "invalid_scope_or_key"})

            def do_POST(self):
                first = False
                try:
                    if self.path != "/work" or self.headers.get_content_type() != "application/json":
                        self.reply(400, {"error": "invalid_fixture_request"}); return
                    length = int(self.headers.get("Content-Length", "0"))
                    if not 1 <= length <= 4096:
                        self.reply(400, {"error": "invalid_body_size"}); return
                    delivery = json.loads(self.rfile.read(length))
                    if not isinstance(delivery, dict) or set(delivery) != {"tenant", "operation_key", "payload"} or delivery["payload"] != PAYLOAD:
                        self.reply(400, {"error": "use_the_authored_fixture"}); return
                    with fixture.lock:
                        first, fixture.first = fixture.first, False
                    fixture.record("request_received", operation_key=delivery["operation_key"], first=first)
                    fixture.entered.set()
                    if first and fixture.case == "timeout-in-flight":
                        fixture.record("waiting_before_effect")
                        require(fixture.release.wait(5), "fixture_release_not_received")
                    if first and fixture.case == "drop-before-effect":
                        fixture.record("connection_closed_before_effect"); self.drop(); return
                    result = receive(fixture.path, delivery, True, allowed_keys=("OP-1", "OP-2"))
                    fixture.record("receiver_result", status=result["status"], effect_id=result["receipt"]["effect_id"], operation_key=delivery["operation_key"])
                    if first:
                        fixture.record("connection_closed_after_commit"); self.drop(); return
                    self.reply(201 if result["status"] == "applied" else 200, result)
                except (ValueError, KeyError, TypeError):
                    self.reply(409, {"status": "conflict", "error": "invalid_or_conflicting_fixture"})
                except Exception as error:
                    fixture.errors.append(type(error).__name__)
                    self.drop()
                finally:
                    if first:
                        fixture.finished.set()

        self.server = Server(("127.0.0.1", 0), Handler)
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(target=self.server.serve_forever, kwargs={"poll_interval": 0.02})
        self.thread.start()

    def record(self, event, **fields):
        with self.lock:
            self.trace.append({"event": event, **fields})

    def close(self):
        import socket
        self.release.set()
        self.server.shutdown()
        self.server.server_close()  # Own handlers have bounded reads and gate waits.
        self.thread.join(2)
        require(not self.thread.is_alive(), "owned_http_server_not_stopped")
        try:
            with socket.create_connection(("127.0.0.1", self.port), timeout=0.2):
                raise RuntimeError("owned_http_listener_still_open")
        except ConnectionRefusedError:
            return True


def http_call(port, method, path, delivery=None, timeout=2):
    import http.client
    connection = http.client.HTTPConnection("127.0.0.1", port, timeout=timeout)
    try:
        body = json.dumps(delivery, sort_keys=True) if delivery is not None else None
        connection.request(method, path, body=body, headers={"Content-Type": "application/json"} if body else {})
        response = connection.getresponse()
        return {"method": method, "path": path, "http_status": response.status,
                "body": json.loads(response.read()), "exception": None}
    except (http.client.RemoteDisconnected, TimeoutError, ConnectionResetError) as error:
        return {"method": method, "path": path, "http_status": None, "body": None,
                "exception": type(error).__name__}
    finally:
        connection.close()


def accept_http_result(caller_path, result):
    if result["http_status"] in (200, 201):
        acknowledge(caller_path, {"acknowledgment": result["body"]})


def run_http_case(name, retry_key):
    import threading
    with tempfile.TemporaryDirectory(prefix="fde-http-outcome-") as directory:
        caller_path, receiver_path = (str(Path(directory) / leaf) for leaf in ("caller.sqlite", "receiver.sqlite"))
        initialize(caller_path, receiver_path, True)
        original = caller_delivery(caller_path)
        fixture = HttpFixture(receiver_path, name)
        first_result, request_thread, report = [], None, None
        try:
            early_lookup = None
            if name == "timeout-in-flight":
                request_thread = threading.Thread(target=lambda: first_result.append(http_call(fixture.port, "POST", "/work", original, timeout=0.4)))
                request_thread.start()
                require(fixture.entered.wait(3), "request_was_not_observed")
                request_thread.join(2)
                require(not request_thread.is_alive() and first_result and first_result[0]["exception"] == "TimeoutError", "caller_timeout_was_not_observed")
                early_lookup = http_call(fixture.port, "GET", "/receipt?tenant=north&operation_key=OP-1")
                require(early_lookup["http_status"] == 404 and not fixture.finished.is_set(), "early_lookup_did_not_precede_effect")
                fixture.release.set()
            else:
                first_result.append(http_call(fixture.port, "POST", "/work", original))
            require(fixture.finished.wait(3) and not fixture.errors, "first_request_did_not_finish_cleanly")
            require(first_result[0]["http_status"] is None, "first_fault_returned_a_response")
            caller_before = caller_view(caller_path)
            require(caller_before["outcome"] == "unknown", "transport_error_became_rollback")
            observer_before = destination_observer(receiver_path)
            expected_before = 0 if name == "drop-before-effect" else 1
            require(observer_before["effects"] == expected_before, "unexpected_initial_effect_count")
            replay_delivery = caller_delivery(caller_path)
            replay_delivery["operation_key"] = retry_key
            replay = http_call(fixture.port, "POST", "/work", replay_delivery)
            receipt_error = None
            try:
                accept_http_result(caller_path, replay)
            except RuntimeError as error:
                require(str(error) == "receipt_intent_mismatch", "unexpected_receipt_failure")
                receipt_error = str(error)
            caller_after_retry = caller_view(caller_path)
            lookup = http_call(fixture.port, "GET", "/receipt?tenant=north&operation_key=OP-1")
            accept_http_result(caller_path, lookup)
            final = destination_observer(receiver_path)
            expected_final = expected_before + (1 if retry_key == "OP-2" or expected_before == 0 else 0)
            require(final["effects"] == final["receipts"] == expected_final and not fixture.errors, "unexpected_final_effect_count")
            require(caller_delivery(caller_path) == original, "original_caller_intent_changed")
            report = {"case": name, "listener": {"host": "127.0.0.1", "port": fixture.port},
                      "original_delivery": original, "first_http_result": first_result[0],
                      "early_lookup_while_in_flight": early_lookup, "caller_after_fault": caller_before,
                      "independent_observer_after_fault": observer_before,
                      "retry_key": retry_key, "replay": replay, "retry_receipt_error": receipt_error,
                      "caller_after_retry": caller_after_retry, "original_receipt_lookup": lookup,
                      "caller_after_reconciliation": caller_view(caller_path), "independent_observer_final": final,
                      "server_trace": fixture.trace, "server_errors": fixture.errors}
        finally:
            fixture.release.set()
            if request_thread is not None:
                request_thread.join(3)
            removed = fixture.close()
            if report is not None:
                report["owned_listener_closed"] = removed
        return report


def run_http(retry_key):
    return {"mode": "http", "basis": "Actual loopback HTTP requests and two temporary SQLite databases per case; effects are fictional receiver rows.",
            "scope": "No cloud, external host, credentials, TLS, real authentication, driver pool or production outcome. Scope labels are supplied fixtures.",
            "retry_key": retry_key, "python_version": sys.version.split()[0], "sqlite_version": sqlite3.sqlite_version,
            "cases": [run_http_case(case, retry_key) for case in HTTP_CASES]}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    action = parser.add_mutually_exclusive_group(required=True)
    action.add_argument("--case", choices=CASES, help="run one named case")
    action.add_argument("--all", action="store_true", help="run all four cases and verify their outcomes")
    action.add_argument("--self-test", action="store_true", help="verify all four cases; print a compact result")
    action.add_argument("--http", action="store_true", help="Run actual loopback HTTP disconnect, timeout and reconciliation cases")
    parser.add_argument("--retry-key", choices=("OP-1", "OP-2"), help="HTTP-mode replay identity (default OP-1); OP-2 deliberately changes the operation")
    args = parser.parse_args()
    if args.retry_key is not None and not args.http:
        parser.error("--retry-key requires --http")
    if args.http:
        print(json.dumps(run_http(args.retry_key or "OP-1"), indent=2))
        return 0
    cases = CASES if args.all or args.self_test else (args.case,)
    reports = [run_case(name) for name in cases]
    if args.self_test:
        print(json.dumps({"self_test": "passed", "cases": list(cases),
                          "verified": "actual child exits, unknown caller state, duplicate naive effect, stable guarded receipt, payload conflict, partition separation, mismatched receipt rejected"}, indent=2))
    else:
        print(json.dumps({"basis": "two temporary local SQLite databases; effect is a receiver row",
                          "scope": "caller outbox durability does not establish a single receiver effect",
                          "cases": reports}, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())