#!/usr/bin/env python3
"""At-least-once delivery double-counts a sum; stamping the offset makes it exact, DDIA 2e Ch. 12.

A consumer reads an append-only log of 100 "+1" events (offsets 0..99) and applies
each one to an EXTERNAL total. It records its consumer offset only PERIODICALLY --
a checkpoint every 10 events -- so the offset it has committed usually lags the
events it has already applied.

Now crash it in that gap: it has APPLIED events through offset 49, but its last
committed checkpoint is at offset 39. On restart the consumer resumes from the
checkpoint, offset 40, and REPROCESSES events 40..49 -- at-least-once delivery, so
those ten events are delivered a second time.

The surprise is what the redelivery does to the total. The naive consumer just
increments, so the ten replayed events are applied twice: the counter ends at 110,
not 100 -- redelivered messages INFLATE it. The fix is idempotence via the offset:
the external store records, alongside the value, the HIGHEST offset it has already
applied, and rejects any event whose offset is not strictly newer. The same replay
of 40..49 is now a no-op, and the total is exactly 100.

Pure standard library, deterministic -> reproducible.
"""

# --- The log: 100 "+1" events at offsets 0..99. Applying event i adds DELTA. ---

N_EVENTS = 100
DELTA = 1
TRUE_TOTAL = N_EVENTS * DELTA          # every event applied exactly once -> 100

CHECKPOINT_EVERY = 10                  # the consumer commits its offset this often
APPLIED_THROUGH = 49                   # last offset applied before the crash
LAST_CHECKPOINT = 39                   # last offset committed before the crash
RESUME_FROM = LAST_CHECKPOINT + 1      # restart replays from here -> 40


# --- Two external stores. Both hold a running total; one also stamps the offset. ---

class NaiveStore:
    """The side effect is a bare increment. It cannot tell a replay from a new event."""

    def __init__(self):
        self.total = 0

    def apply(self, offset, delta):
        # No memory of which offsets have been seen: every delivery increments.
        self.total += delta
        return True                    # "applied"


class IdempotentStore:
    """Stores the highest applied offset next to the value: a set-if-newer write."""

    def __init__(self):
        self.total = 0
        self.highest_applied = -1      # no offset applied yet

    def apply(self, offset, delta):
        # Idempotence via the offset: only apply an event strictly newer than the
        # highest we have already folded in. A replayed offset is a no-op.
        if offset <= self.highest_applied:
            return False               # duplicate -> skipped
        self.total += delta
        self.highest_applied = offset
        return True                    # applied


def run(store, label):
    """Drive `store` through the crash-and-replay delivery schedule, tracing skips."""
    print(f"{label}:")

    # --- First run: apply 0..49, checkpoint every 10, then crash. ---
    committed = None
    for offset in range(0, APPLIED_THROUGH + 1):
        store.apply(offset, DELTA)
        # Periodic offset checkpoint -- except the one that WOULD fire right after
        # the final applied offset: the crash lands in the gap between applying the
        # side effect and committing that checkpoint, so it never gets recorded.
        if (offset + 1) % CHECKPOINT_EVERY == 0 and offset < APPLIED_THROUGH:
            committed = offset
    print(f"    first run: applied offsets 0..{APPLIED_THROUGH}, "
          f"total = {store.total}, last checkpoint committed = offset {committed}")
    print(f"    *** CRASH *** (applied through {APPLIED_THROUGH}, "
          f"but offset only committed up to {committed})")

    # --- Restart: resume from the committed offset, so 40..49 are redelivered. ---
    replayed = list(range(RESUME_FROM, APPLIED_THROUGH + 1))
    skipped = []
    for offset in range(RESUME_FROM, N_EVENTS):
        applied = store.apply(offset, DELTA)
        if not applied:
            skipped.append(offset)
    print(f"    restart: resume from committed offset {RESUME_FROM} -> "
          f"redelivers offsets {replayed[0]}..{replayed[-1]} "
          f"({len(replayed)} events), then 50..99")
    if skipped:
        print(f"    skipped {len(skipped)} already-applied offsets: "
              f"{skipped[0]}..{skipped[-1]}")
    else:
        print(f"    skipped 0 offsets -- every redelivered event was applied again")

    inflation = store.total - TRUE_TOTAL
    verdict = (f"INFLATED by {inflation}" if inflation
               else "exactly the number of events")
    print(f"    final total = {store.total}  (true total = {TRUE_TOTAL}, {verdict})\n")
    return store.total


def main():
    print(f"An append-only log of {N_EVENTS} \"+{DELTA}\" events, offsets 0..{N_EVENTS - 1}.")
    print(f"A consumer applies each event to an external total, checkpointing its")
    print(f"offset every {CHECKPOINT_EVERY} events. It crashes having applied through "
          f"offset {APPLIED_THROUGH}")
    print(f"with its offset last committed at {LAST_CHECKPOINT} -- so restart replays "
          f"40..{APPLIED_THROUGH}.")
    print(f"    every event is \"+{DELTA}\", so the true total is {TRUE_TOTAL}.\n")

    naive_total = run(NaiveStore(), "NAIVE increment (at-least-once, no dedup)")
    idem_total = run(IdempotentStore(), "IDEMPOTENT set-if-newer (dedup by offset)")

    print("summary:")
    print(f"    naive        -> {naive_total}   "
          f"(+{naive_total - TRUE_TOTAL} from the 10 replayed events)")
    print(f"    idempotent   -> {idem_total}   "
          f"(replayed events rejected as duplicates -> effectively-once)")


if __name__ == "__main__":
    main()