#!/usr/bin/env python3
"""Offline simulation of the side_effect_ledger guard (sqlite3 stands in for Postgres).

Runs the same decision logic as guard.n8n.json:
  claim -> (claimed) send -> mark done / failed / leave 'started'
        -> (not claimed) done: skip | failed: reclaim and send | started: needs review

Default is --mock: the remote system is an in-process fake driven by sample/scenarios.json.
Only http_post is swapped. A non-mock run needs --url and --apply (it would POST for real).
"""
import argparse
import json
import os
import sqlite3
import sys
import urllib.error
import urllib.request

HERE = os.path.dirname(os.path.abspath(__file__))

DDL = """
CREATE TABLE side_effect_ledger (
  idem_key     TEXT PRIMARY KEY,
  action       TEXT NOT NULL,
  status       TEXT NOT NULL CHECK (status IN ('started','done','failed')),
  execution_id TEXT,
  started_at   TEXT NOT NULL DEFAULT (strftime('%Y-%m-%d %H:%M:%S','now')),
  finished_at  TEXT,
  response     TEXT
)"""

# Same statements as schema.sql, with ? placeholders and sqlite time functions.
CLAIM = ("INSERT INTO side_effect_ledger (idem_key, action, status, execution_id) "
         "VALUES (?, ?, 'started', ?) ON CONFLICT (idem_key) DO NOTHING RETURNING idem_key")
MARK_DONE = ("UPDATE side_effect_ledger SET status='done', finished_at=strftime('%Y-%m-%d %H:%M:%S','now'), "
             "response=? WHERE idem_key=? AND status='started' RETURNING idem_key")
MARK_FAILED = ("UPDATE side_effect_ledger SET status='failed', finished_at=strftime('%Y-%m-%d %H:%M:%S','now'), "
               "response=? WHERE idem_key=? AND status='started' RETURNING idem_key")
LOOKUP = ("SELECT status, CAST(strftime('%s','now') AS INTEGER) - CAST(strftime('%s', started_at) AS INTEGER) "
          "FROM side_effect_ledger WHERE idem_key=?")
RECLAIM = ("UPDATE side_effect_ledger SET status='started', started_at=strftime('%Y-%m-%d %H:%M:%S','now'), "
           "finished_at=NULL, execution_id=? WHERE idem_key=? AND status='failed' RETURNING idem_key")


# ---------------------------------------------------------------- HTTP layer
def http_post(url, headers, body, timeout=30):
    """Live HTTP layer. Returns (status_code, parsed_body). Timeouts/connection errors raise OSError."""
    data = json.dumps(body).encode()
    req = urllib.request.Request(url, data=data, method="POST",
                                 headers={"Content-Type": "application/json", **headers})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            raw = r.read().decode() or "{}"
            return r.status, _maybe_json(raw)
    except urllib.error.HTTPError as e:  # a real HTTP response with an error status
        return e.code, _maybe_json(e.read().decode() or "{}")


def _maybe_json(raw):
    try:
        return json.loads(raw)
    except ValueError:
        return {"raw": raw}


class FakeRemote:
    """Fake external system. Does NOT dedupe on Idempotency-Key (worst case, like many chat/email APIs)."""

    def __init__(self, behaviors):
        self.behaviors = behaviors
        self.calls = {}
        self.effects = {}

    def http_post(self, url, headers, body, timeout=30):
        rid = body["id"]
        n = self.calls[rid] = self.calls.get(rid, 0) + 1
        seq = self.behaviors.get(rid, [])
        mode = seq[n - 1] if n <= len(seq) else "ok"
        if mode == "reject_422":
            return 422, {"error": "invalid"}
        self.effects[rid] = self.effects.get(rid, 0) + 1  # the write is committed
        if mode == "timeout_after_accept":
            raise TimeoutError("simulated: remote committed, response lost")
        if mode == "server_error_after_accept":
            return 500, {"error": "simulated: committed then errored"}
        return 201, {"created": rid}


HTTP_POST = http_post  # main() swaps this in mock mode


# ---------------------------------------------------------------- ledger + guard
def classify(status):
    """done = success, failed = definite rejection (safe to resend), unknown = remote may have acted."""
    if 200 <= status < 300:
        return "done"
    if 400 <= status < 500 and status not in (408, 409):
        return "failed"
    return "unknown"


def process_item(db, action, rec, execution_id, url):
    key = f"{action}:{rec['id']}"
    row = db.execute(CLAIM, (key, action, execution_id)).fetchone()
    db.commit()
    result_prefix = ""
    if row is None:
        got = db.execute(LOOKUP, (key,)).fetchone()
        status = got[0] if got else None
        if status == "done":
            return key, "skipped_done"
        if status == "failed":
            won = db.execute(RECLAIM, (execution_id, key)).fetchone()
            db.commit()
            if won is None:
                return key, "review"
            result_prefix = "reclaimed_"
        else:
            return key, "review"  # started: outcome unknown, a human decides
    try:
        code, body = HTTP_POST(url, {"Idempotency-Key": key}, rec)
    except OSError:  # includes TimeoutError and URLError; leave the row 'started'
        return key, "unknown"
    outcome = classify(code)
    if outcome == "done":
        db.execute(MARK_DONE, (json.dumps(body), key)).fetchone()
        db.commit()
        return key, result_prefix + "sent"
    if outcome == "failed":
        db.execute(MARK_FAILED, (json.dumps(body), key)).fetchone()
        db.commit()
        return key, result_prefix + "failed"
    return key, "unknown"


def run_batch(db, action, records, execution_id, url):
    return {r["id"]: process_item(db, action, r, execution_id, url) for r in records}


def naive_batch(action, records, url):
    """No ledger: what a plain retry does."""
    for r in records:
        try:
            HTTP_POST(url, {}, r)
        except OSError:
            pass


# ---------------------------------------------------------------- scenarios
def run_scenario(sc, data, url):
    global HTTP_POST
    action, records = data["action"], data["records"]
    ids = [r["id"] for r in records]
    print(f"\n=== Scenario {sc['name']} ===")

    # Baseline: no guard, execute then retry the whole batch.
    remote = FakeRemote(sc["behaviors"])
    HTTP_POST = remote.http_post
    naive_batch(action, records, url)
    naive_batch(action, records, url)
    print("Baseline WITHOUT guard (run + retry): effects per item = " +
          ", ".join(f"{i}:{remote.effects.get(i, 0)}" for i in ids))
    dupes = [i for i in ids if remote.effects.get(i, 0) > 1]
    print(f"  duplicated items without guard: {dupes}")

    # With guard.
    remote = FakeRemote(sc["behaviors"])
    HTTP_POST = remote.http_post
    db = sqlite3.connect(":memory:")
    db.execute(DDL)
    r1 = run_batch(db, action, records, "exec-1", url)
    r2 = run_batch(db, action, records, "exec-2", url)  # retry of the whole batch, NEW execution id
    for label, res in (("run 1", r1), ("retry (whole batch)", r2)):
        print(f"{label}:")
        for rid in ids:
            print(f"  {res[rid][0]:<28} {res[rid][1]}")
    print("External calls / effects per item WITH guard:")
    for rid in ids:
        print(f"  {rid}: calls={remote.calls.get(rid, 0)} effects={remote.effects.get(rid, 0)}")
    review = sorted(k for k, s in r2.values() if s == "review")
    print(f"Sent to human review: {review}")

    exp = sc["expect"]
    assert {i: s for i, (_, s) in r1.items()} == exp["run1"], f"run1 mismatch {r1}"
    assert {i: s for i, (_, s) in r2.items()} == exp["run2"], f"run2 mismatch {r2}"
    assert {i: remote.effects.get(i, 0) for i in ids} == exp["effects"], "effects mismatch"
    assert {i: remote.calls.get(i, 0) for i in ids} == exp["calls"], "calls mismatch"
    assert review == exp["review"], f"review mismatch {review}"
    assert all(v <= 1 for v in remote.effects.values()), "an external effect happened more than once"
    print("PASS: every external effect happened at most once; unknown items went to review, not resent.")
    return dupes


def claim_race_check():
    """Two claimers for the same key: exactly one wins."""
    db = sqlite3.connect(":memory:")
    db.execute(DDL)
    a = db.execute(CLAIM, ("create_crm_contact:ORD-9", "create_crm_contact", "exec-A")).fetchone()
    b = db.execute(CLAIM, ("create_crm_contact:ORD-9", "create_crm_contact", "exec-B")).fetchone()
    assert a is not None and b is None, "second claim must return no row"
    n = db.execute("SELECT count(*) FROM side_effect_ledger").fetchone()[0]
    assert n == 1
    print("\n=== Claim check ===\nSecond claim for the same key returned no row; ledger has 1 row. PASS")


def main():
    global HTTP_POST
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--mock", action="store_true", help="offline run against sample/ fixtures (default if no --url)")
    ap.add_argument("--url", help="live endpoint to POST to (needs --apply)")
    ap.add_argument("--apply", action="store_true", help="allow real HTTP writes (live mode only)")
    ap.add_argument("--records", default=os.path.join(HERE, "sample", "records.json"))
    ap.add_argument("--scenarios", default=os.path.join(HERE, "sample", "scenarios.json"))
    args = ap.parse_args()

    if args.url and not args.mock:
        if not args.apply:
            print("Refusing to send real requests without --apply. Use --mock for the offline test.")
            return 2
        data = json.load(open(args.records))
        db = sqlite3.connect(os.path.join(HERE, "live_ledger.sqlite"))
        db.execute(DDL.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS"))
        for rid, (key, st) in run_batch(db, data["action"], data["records"], "live-run", args.url).items():
            print(f"{key:<28} {st}")
        return 0

    data = json.load(open(args.records))
    scenarios = json.load(open(args.scenarios))["scenarios"]
    url = "https://example.com/mock/write"
    any_dupes = False
    for sc in scenarios:
        any_dupes |= bool(run_scenario(sc, data, url))
    claim_race_check()
    assert any_dupes, "baseline should show duplicates, otherwise the test proves nothing"
    print("\nALL SCENARIOS PASSED")
    return 0


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