"""Never twice: run a side effect at most once across racing workers and crash retries.
Python 3.8+, standard library only. Set CASCADE_API_KEY (and CASCADE_API for a local server).

The claim on once/<key> stops CONCURRENT duplicates. already_done() plus passing `key`
downstream as the idempotency key stop LATE duplicates (a retry after a crash). If
do_it() finished but the worker died before record(), the next worker calls do_it()
again; only the downstream idempotency key makes that harmless.
"""
import json, os, random, threading, time, urllib.error, urllib.request


def cascade(path, body):
    api = os.environ.get("CASCADE_API", "https://api.cascadedynamics.ai/v1")
    for attempt in range(6):
        req = urllib.request.Request(api + path, data=json.dumps(body).encode(), method="POST", headers={
            "Authorization": "Bearer " + os.environ["CASCADE_API_KEY"],
            "Content-Type": "application/json", "User-Agent": "cascade-recipes/1.0"})
        try:
            with urllib.request.urlopen(req, timeout=30) as r:
                return json.loads(r.read())
        except urllib.error.HTTPError as e:
            if e.code in (429, 503) and attempt < 5:
                time.sleep(0.25 * 2 ** attempt)
                continue
            raise


def keep_held(cell, agent, ttl_ms, lost):
    """Re-claim every ttl/3 (one metered decision each). Sets `lost` if the cell is taken.
    Returns stop(); it waits for a renewal already on the wire, so that renewal cannot
    land after the release and re-take the cell."""
    halt = threading.Event()

    def loop():
        while not halt.wait(ttl_ms / 3000):
            try:
                if not cascade("/claim", {"cell": cell, "agent": agent, "ttl_ms": ttl_ms}).get("ok"):
                    lost.set()
            except Exception:
                pass  # a blip; if it never recovers the TTL lapses
    t = threading.Thread(target=loop, daemon=True)
    t.start()

    def stop():
        halt.set()
        t.join()
    return stop


def once_only(key, agent, do_it, already_done, record=None, ttl_ms=30000, poll_s=0.25):
    """do_it(key, lost_event) -> result. Returns {"did": bool, "result", "receipt"}."""
    if already_done(key):
        return {"did": False}
    cell = "once/" + key
    while True:
        c = cascade("/claim", {"cell": cell, "agent": agent, "ttl_ms": ttl_ms})
        if c.get("ok"):
            break
        time.sleep(poll_s + random.random() * poll_s)  # repeat polls are free
    lost = threading.Event()
    stop = keep_held(cell, agent, ttl_ms, lost)
    try:
        if already_done(key):
            return {"did": False}
        result = do_it(key, lost)
        if record:
            record(key, {"result": result, "agent": agent, "receipt": c["entry_hash"]})
        return {"did": True, "result": result, "receipt": c["entry_hash"]}
    finally:
        stop()
        try:
            cascade("/release", {"cell": cell, "agent": agent})
        except Exception:
            pass  # the TTL frees it