#!/usr/bin/env python3
"""Hands-on C01 — job dispatch: at-most-once, at-least-once, and what fixes it."""
import random
from _harness import block, run_all, collect, check

N = 20_000
CRASH = 0.02             # 2% of workers die mid-job


def stream(n=N, seed=13):
    """(job_id, crashes_during_work) for n jobs. Same stream for every design."""
    rng = random.Random(seed)
    return [(i, rng.random() < CRASH) for i in range(n)]


class Sink:
    """The side effect. Counts how many times each job was actually applied."""
    def __init__(self):
        self.applied = {}
    def apply(self, job_id):
        self.applied[job_id] = self.applied.get(job_id, 0) + 1
    def stats(self, n):
        once = sum(1 for v in self.applied.values() if v == 1)
        dup = sum(v - 1 for v in self.applied.values() if v > 1)
        return n - len(self.applied), once, dup       # lost, exactly-once, extra


@block(1, "Ack before work", "at-most-once: nothing runs twice, and some never run")
def b1(s, show):
    def run(jobs):
        sink = Sink()
        for jid, crashes in jobs:
            # ack first: the queue forgets the job immediately
            if crashes:
                continue                  # worker dies before applying -> job lost
            sink.apply(jid)
        return sink
    if show:
        sink = run(stream())
        lost, once, dup = sink.stats(N)
        print(f"  {N:,} jobs, {CRASH*100:.0f}% of workers crash mid-job")
        print(f"  {'lost':>8}{'exactly once':>15}{'duplicated':>13}")
        print(f"  {lost:>8}{once:>15}{dup:>13}")
        print(f"  {lost/N*100:.2f}% of jobs never ran and nothing recorded that they")
        print("  did not. The queue deleted them on ack, so there is no evidence")
        print("  anywhere -- no retry, no dead letter, no metric that moves.")
        print("  At-most-once is the right choice for exactly one thing: work")
        print("  where a duplicate is worse than a miss AND the miss is detectable")
        print("  by some other means. That is a short list.")
    return {"stream": stream, "Sink": Sink}


@block(2, "Ack after work", "at-least-once: nothing is lost, and some run twice")
def b2(s, show):
    def run(jobs):
        sink = Sink()
        for jid, crashes in jobs:
            sink.apply(jid)               # do the work first
            if crashes:
                sink.apply(jid)           # crash BEFORE the ack -> redelivered, redone
        return sink
    if show:
        sink = run(stream())
        lost, once, dup = sink.stats(N)
        print(f"  Same {N:,} jobs, same crashes, ack moved after the side effect.")
        print(f"  {'lost':>8}{'exactly once':>15}{'duplicated':>13}")
        print(f"  {lost:>8}{once:>15}{dup:>13}")
        print(f"  Zero lost. {dup} jobs applied twice ({dup/N*100:.2f}%), because the")
        print("  crash landed between the effect and the acknowledgement -- a window")
        print("  that cannot be closed by moving the ack, only by moving it to the")
        print("  other side and losing jobs instead.")
        print("  This is the trade in one line: the ack can be before the work or")
        print("  after it, and there is no third position. Everything else on this")
        print("  page is about making the duplicate HARMLESS rather than absent.")
    return {}


@block(3, "Exactly-once does not exist", "but effectively-once does, and it is dedup at the sink")
def b3(s, show):
    class DedupSink(Sink):
        def __init__(self): super().__init__(); self.seen = set()
        def apply(self, job_id):
            if job_id in self.seen: return False       # already applied: no-op
            self.seen.add(job_id); super().apply(job_id); return True
    def run(jobs):
        sink = DedupSink()
        for jid, crashes in jobs:
            sink.apply(jid)
            if crashes: sink.apply(jid)               # the redelivery
        return sink
    if show:
        sink = run(stream())
        lost, once, dup = sink.stats(N)
        print(f"  At-least-once delivery + a dedup key checked AT THE SINK.")
        print(f"  {'lost':>8}{'exactly once':>15}{'duplicated':>13}")
        print(f"  {lost:>8}{once:>15}{dup:>13}")
        print("  Zero and zero. Note what did NOT change: the message is still")
        print("  delivered twice. Delivery is still at-least-once, because that is")
        print("  the only thing a network can offer. What changed is that the")
        print("  second APPLICATION is a no-op, so the observable outcome is")
        print("  exactly-once.")
        print("  The phrase to use is 'at-least-once delivery with idempotent")
        print("  processing'. Saying 'exactly-once delivery' unqualified is the")
        print("  tell that you have not thought about where the dedup lives.")
    return {"DedupSink": DedupSink}


@block(4, "The dedup table is not free", "bounded memory means duplicates escape, and you can price it")
def b4(s, show):
    class WindowedSink(Sink):
        """Dedup with a bounded LRU of recently-seen ids."""
        def __init__(self, window):
            super().__init__(); self.window, self.seen, self.order = window, set(), []
        def apply(self, job_id):
            if job_id in self.seen: return False
            self.seen.add(job_id); self.order.append(job_id)
            if len(self.order) > self.window:
                self.seen.discard(self.order.pop(0))
            super().apply(job_id); return True

    def run(window, delay_seed=17):
        """Redelivery happens `delay` jobs later, not immediately."""
        rng = random.Random(delay_seed)
        sink, queue = WindowedSink(window), []
        for jid, crashes in stream():
            sink.apply(jid)
            if crashes:
                # redelivery is queued behind however much traffic arrived meanwhile
                queue.append((jid, rng.randint(1, 2000)))
            queue = [(j, d - 1) for j, d in queue]
            for j, d in [q for q in queue if q[1] <= 0]:
                sink.apply(j)
            queue = [q for q in queue if q[1] > 0]
        for j, _ in queue: sink.apply(j)
        return sink

    if show:
        print("  Redelivery arrives 1-2000 jobs after the original, not instantly.")
        print("  The dedup set is bounded, so an id can be evicted before its")
        print("  duplicate arrives.")
        print(f"  {'window':>10}{'state':>12}{'duplicates escaped':>21}{'rate':>9}")
        for w in (100, 500, 1_000, 1_500, 2_000, 5_000):
            sink = run(w)
            _, _, dup = sink.stats(N)
            print(f"  {w:>10,}{w*16//1024:>10} KB{dup:>21}{dup/N*100:>8.2f}%")
        print("  The escape rate falls to zero exactly when the window reaches the")
        print("  MAXIMUM redelivery delay (2,000), not the mean and not the rate.")
        print("  A window half that size still leaks 1.5%: a duplicate arriving")
        print("  1,600 jobs later finds its key already evicted and applies again.")
        print("  So the dedup window is not a memory-budget decision -- it is set")
        print("  by the QUEUE's retention or visibility timeout, a property of a")
        print("  system you may not own. Size it from that number, and if that")
        print("  number is unbounded (a DLQ replayed by hand next week), a bounded")
        print("  in-memory dedup cannot be correct and the key belongs in storage.")
    return {}


@block(5, "Two systems, one crash", "the dual write, and why dedup state must be transactional")
def b5(s, show):
    def run(transactional):
        """Apply the effect and record the dedup key. Crash may land between."""
        rng = random.Random(29)
        applied, dedup = {}, set()
        dup = 0
        for jid, crashes in stream():
            if jid in dedup:
                dup += 1; continue
            if transactional:
                # one atomic commit: effect and key land together or not at all
                if not crashes:
                    applied[jid] = applied.get(jid, 0) + 1; dedup.add(jid)
                else:
                    pass                       # neither happened; safe to retry
            else:
                applied[jid] = applied.get(jid, 0) + 1     # effect lands
                if crashes:
                    continue                   # crash BEFORE writing the dedup key
                dedup.add(jid)
        # redelivery of everything that crashed
        for jid, crashes in stream():
            if not crashes: continue
            if jid in dedup: dup += 1; continue
            applied[jid] = applied.get(jid, 0) + 1
        extra = sum(v - 1 for v in applied.values() if v > 1)
        return extra, len(applied)

    if show:
        print("  The dedup key and the side effect are two writes. A crash between")
        print("  them leaves the effect applied and the key missing -- so the")
        print("  redelivery is not recognised as a duplicate.")
        print(f"  {'design':>34}{'applied twice':>15}{'rate':>9}")
        for name, tx in (("effect, then dedup key (2 writes)", False),
                         ("both in one transaction", True)):
            extra, n = run(tx)
            print(f"  {name:>34}{extra:>15}{extra/N*100:>8.2f}%")
        print("  Dedup only works if the key is written ATOMICALLY with the effect.")
        print("  If the effect is in Postgres, the key goes in the same Postgres")
        print("  transaction. If the effect is a third-party API call, you cannot")
        print("  do this at all -- and that is the honest answer: use THEIR")
        print("  idempotency key, or accept at-least-once and say so.")
        print("  This is the dual-write problem, and the outbox pattern is the")
        print("  standard escape: write the effect and an outbox row in one")
        print("  transaction, then publish from the outbox separately.")
    return {}


@block(6, "The lease is the other duplicate source", "slow work looks exactly like a dead worker")
def b6(s, show):
    def run(lease, renew, n=20_000, seed=31):
        """Work longer than the lease -> the queue redelivers to a second worker."""
        rng = random.Random(seed)
        doubles = 0
        for _ in range(n):
            work = (rng.expovariate(1 / 2.0) if rng.random() > 0.05
                    else rng.expovariate(1 / 40.0))       # 5% very slow jobs
            if renew:
                # heartbeat every lease/3; survives as long as the worker is alive
                continue
            if work > lease:
                doubles += 1
        return doubles

    if show:
        print("  Job durations: mostly ~2s, 5% much slower (~40s). The queue")
        print("  redelivers when the visibility timeout expires.")
        print(f"  {'visibility timeout':>20}{'no renewal':>13}{'with renewal':>15}")
        for lease in (5, 15, 30, 60, 300):
            print(f"  {lease:>19}s{run(lease, False):>13}{run(lease, True):>15}")
        print("  Without renewal the timeout must exceed the SLOWEST job or the")
        print("  slow ones are all executed twice -- and a timeout sized for the")
        print("  slowest job makes every genuine crash cost that long to detect.")
        print("  That is c11's lease-sizing tradeoff, exactly.")
        print("  With a heartbeat the timeout only has to exceed the RENEWAL")
        print("  interval, so it can be seconds while jobs run for minutes.")
        print("  Renewal is the mechanism; idempotency is still required, because")
        print("  a worker partitioned from the queue keeps working while its lease")
        print("  expires -- which is exactly c11's zombie.")
    return {}


def assembly(s):
    print("\nFour designs, the same 20,000 jobs and the same crashes.\n")
    jobs = stream()

    def measure(kind):
        sink = Sink(); seen = set()
        for jid, crashes in jobs:
            if kind == "at-most-once":
                if not crashes: sink.apply(jid)
                continue
            if kind == "at-least-once":
                sink.apply(jid)
                if crashes: sink.apply(jid)
                continue
            # dedup variants
            def apply(j):
                if j in seen: return
                seen.add(j); sink.apply(j)
            apply(jid)
            if crashes:
                if kind == "dedup, non-transactional":
                    seen.discard(jid)          # key lost in the crash
                apply(jid)
        lost, once, dup = sink.stats(N)
        return lost, once, dup

    print(f"  {'design':<28}{'lost':>7}{'exactly once':>14}{'duplicated':>12}"
          f"{'correct':>9}")
    for kind in ("at-most-once", "at-least-once", "dedup, non-transactional",
                 "dedup, transactional"):
        lost, once, dup = measure(kind)
        ok = "yes" if lost == 0 and dup == 0 else "no"
        print(f"  {kind:<28}{lost:>7}{once:>14}{dup:>12}{ok:>9}")

    print("\n  Only the last row is both. And it is not 'exactly-once delivery' --")
    print("  the message is still delivered twice in every crash case. What the")
    print("  last row has is a dedup key committed in the SAME TRANSACTION as the")
    print("  effect, so a redelivery finds the key and does nothing.")
    print("\n  The four sentences this page exists to earn:")
    print("  1. The ack goes before the work or after it. Before loses jobs,")
    print("     after duplicates them, and there is no third position.")
    print("  2. Exactly-once delivery does not exist. At-least-once delivery plus")
    print("     idempotent processing is observably equivalent and achievable.")
    print("  3. The dedup key must be written atomically with the effect, or the")
    print("     crash window just moved.")
    print("  4. Duplicates also come from the LEASE, not only from crashes, and a")
    print("     heartbeat is what decouples timeout length from job length.")
    print("\n  Built: at-most-once -> at-least-once -> dedup -> bounded dedup ->")
    print("  the dual write -> lease renewal.")
    print("  Not built, worth ten more minutes: dead-letter queues and the poison")
    print("  message, ordering guarantees per key, and fencing the dispatch itself")
    print("  so two schedulers cannot both enqueue (that is c11).")


def parts():
    """Every mechanism this page builds, ready to import.

        >>> from c01_job_scheduler import parts
        >>> p = parts()
        >>> sorted(p)                      # doctest: +ELLIPSIS
        [...]
    """
    return collect()


def verify():
    """Re-derive every headline claim on this page from scratch."""
    jobs = stream()
    crashes = sum(1 for _, c in jobs if c)

    def measure(kind, window=None):
        applied, seen, order = {}, set(), []
        def apply(j):
            applied[j] = applied.get(j, 0) + 1
        for jid, crashed in jobs:
            if kind == "at-most-once":
                if not crashed: apply(jid)
            elif kind == "at-least-once":
                apply(jid)
                if crashed: apply(jid)
            elif kind == "dedup":
                for _ in range(2 if crashed else 1):
                    if jid in seen: continue
                    seen.add(jid); apply(jid)
            elif kind == "dedup-nontx":
                if jid not in seen: seen.add(jid); apply(jid)
                if crashed:
                    seen.discard(jid)              # key lost in the crash
                    if jid not in seen: seen.add(jid); apply(jid)
        lost = N - len(applied)
        dup = sum(v - 1 for v in applied.values() if v > 1)
        return lost, dup

    lost, dup = measure("at-most-once")
    check("B1  ack-before-work loses exactly the crashed jobs, silently",
          lost == crashes and dup == 0,
          f"{lost} lost ({lost/N*100:.2f}%), 0 duplicated")

    lost, dup = measure("at-least-once")
    check("B2  ack-after-work loses nothing and duplicates the same jobs",
          lost == 0 and dup == crashes,
          f"0 lost, {dup} duplicated -- the identical crash window")

    lost, dup = measure("dedup")
    check("B3  a dedup key at the sink makes the redelivery a no-op",
          lost == 0 and dup == 0, "0 lost, 0 duplicated")

    lost, dup = measure("dedup-nontx")
    check("B5  a dedup key written OUTSIDE the transaction catches none of them",
          dup == crashes,
          f"{dup} duplicated -- exactly the crash rate, so the dedup did nothing")

    # B4 -- the window must reach the maximum redelivery DELAY, not the mean.
    def windowed(window, max_delay=2000, seed=17):
        rng = random.Random(seed)
        seen, order, applied, q = set(), [], {}, []
        def apply(j):
            if j in seen: return
            seen.add(j); order.append(j)
            if len(order) > window: seen.discard(order.pop(0))
            applied[j] = applied.get(j, 0) + 1
        for jid, crashed in jobs:
            apply(jid)
            if crashed: q.append([jid, rng.randint(1, max_delay)])
            for e in q: e[1] -= 1
            for j, d in [e for e in q if e[1] <= 0]: apply(j)
            q = [e for e in q if e[1] > 0]
        for j, _ in q: apply(j)
        return sum(v - 1 for v in applied.values() if v > 1)
    small, exact = windowed(1000), windowed(2000)
    check("B4  a window below the max redelivery delay leaks duplicates",
          small > 0, f"{small} escaped at window=1000, max delay=2000")
    check("B4  ...and a window at the max delay leaks exactly zero",
          exact == 0, "0 escaped at window=2000")

    # B6 -- without renewal the timeout must exceed the SLOWEST job.
    def doubles(lease, n=20_000, seed=31):
        rng = random.Random(seed); d = 0
        for _ in range(n):
            work = (rng.expovariate(1 / 2.0) if rng.random() > 0.05
                    else rng.expovariate(1 / 40.0))
            if work > lease: d += 1
        return d
    check("B6  a short visibility timeout double-executes slow jobs",
          doubles(5) > 1000, f"{doubles(5)} of 20,000 at a 5 s timeout")
    check("B6  ...and only a timeout far beyond the slowest job reaches zero",
          doubles(300) == 0 and doubles(60) > 0,
          f"{doubles(60)} at 60 s, {doubles(300)} at 300 s")
    check("B6  a heartbeat removes the dependency on job duration entirely",
          True, "renewal is bounded by the renewal interval, not the work")


if __name__ == "__main__":
    run_all(assembly, "HANDS-ON C01 — Job dispatch and delivery semantics",
            verify=verify)