#!/usr/bin/env python3
"""Multishard exactly-once transfer through a crash -- WITHOUT 2PC.

DDIA 2e, Ch. 13, Figure 13-2: a payment REQUEST is written to a request log as
ONE atomic action. A DETERMINISTIC stream processor consumes it and emits two
events -- a credit to the payee shard and a debit to the payer shard -- each
carrying the original request id. The ONLY atomic step is the append to the
single request log; everything after is a deterministic function of it. So a
crash mid-processing (after applying the first emitted event, before the second,
before recording completion) forces at-least-once REDELIVERY: on recovery the
request is reprocessed from the log and re-emits byte-identical events.

Because the shards DEDUPE by (request-derived) event id, the re-emitted event
that was already applied is skipped -- at-least-once delivery becomes
effectively-once, and money conservation holds across both shards with no
distributed transaction. Drop the dedup and the same crash double-credits the
payee: money is created and the invariant breaks.

Everything is modelled in one process. The crash and redelivery are explicit,
so the transcript is fixed and reproducible.
"""

SEP = "=" * 68


class Shard:
    """A single account shard: a balance plus (optionally) a dedup set of the
    event ids it has already applied. dedup=False models the naive shard that
    keeps no memory of what it has seen -- so a redelivered event is applied
    again."""

    def __init__(self, name, balance, dedup=True):
        self.name = name
        self.balance = balance
        self.dedup = dedup
        self.applied = set()   # event ids already applied
        self.touches = 0       # how many times an event actually mutated us

    def apply(self, event):
        eid = event["event_id"]
        if self.dedup and eid in self.applied:
            print(f"      {self.name}: event {eid} already applied -> SKIP "
                  f"(balance stays {self.balance})")
            return False
        self.balance += event["delta"]
        self.applied.add(eid)
        self.touches += 1
        sign = "+" if event["delta"] >= 0 else ""
        print(f"      {self.name}: apply {eid} ({sign}{event['delta']}) "
              f"-> balance {self.balance}")
        return True


class Crash(Exception):
    """Raised to model a process crash at a chosen point."""


def process_request(req, payer, payee, status, crash_after_first=False):
    """The DETERMINISTIC stream processor. Consumes ONE request from the log and
    derives its effects: a credit to the payee shard and a debit to the payer
    shard, each tagged with an event id DERIVED from the request id (not freshly
    generated) so a replay re-emits byte-identical events.

    Effects are applied in order [credit, debit]. crash_after_first models a
    crash after the credit is applied but before the debit is emitted and before
    the request is marked done -- so the request is not yet complete and must be
    redelivered.
    """
    rid = req["rid"]
    credit = {"event_id": f"{rid}:credit", "delta": +req["amount"], "shard": payee}
    debit = {"event_id": f"{rid}:debit",  "delta": -req["amount"], "shard": payer}
    print(f"    processor: derive events from request {rid} "
          f"(transfer {req['amount']} {payer.name}->{payee.name})")

    print(f"    processor: emit credit {credit['event_id']} to {payee.name}")
    payee.apply(credit)
    if crash_after_first:
        raise Crash("crashed after crediting payee, "
                    "before debiting payer or recording completion")

    print(f"    processor: emit debit  {debit['event_id']} to {payer.name}")
    payer.apply(debit)

    status[rid] = "done"
    print(f"    processor: mark request {rid} done")


def invariant(payer, payee, total):
    s = payer.balance + payee.balance
    ok = (s == total)
    print(f"    invariant: {payer.name}={payer.balance} + {payee.name}="
          f"{payee.balance} = {s}  (started at {total})  "
          f"-> {'HELD' if ok else 'VIOLATED (money ' + ('created' if s > total else 'lost') + ')'}")
    touched_once = (payer.touches == 1 and payee.touches == 1)
    print(f"    touches: {payer.name} touched {payer.touches}x, "
          f"{payee.name} touched {payee.touches}x  "
          f"-> {'each account touched exactly once' if touched_once else 'an account was touched more than once'}")
    return ok


def scenario_happy():
    print(SEP)
    print("SCENARIO 1 -- happy path (no crash, dedup on)")
    print(SEP)
    payer = Shard("A", 100, dedup=True)
    payee = Shard("B", 100, dedup=True)
    total = payer.balance + payee.balance
    print(f"    initial: A=100, B=100, total={total}")
    print("    request-log append (the ONE atomic action): "
          "req-1 = transfer 11 A->B")
    status = {}
    process_request({"rid": "req-1", "amount": 11}, payer, payee, status)
    invariant(payer, payee, total)
    print()


def scenario_crash_dedup():
    print(SEP)
    print("SCENARIO 2 -- crash mid-transfer, dedup ON (redelivery is safe)")
    print(SEP)
    payer = Shard("A", 100, dedup=True)
    payee = Shard("B", 100, dedup=True)
    total = payer.balance + payee.balance
    print(f"    initial: A=100, B=100, total={total}")
    print("    request-log append (the ONE atomic action): "
          "req-1 = transfer 11 A->B")
    req = {"rid": "req-1", "amount": 11}
    status = {}
    print("    --- first delivery: crash after crediting payee ---")
    try:
        process_request(req, payer, payee, status, crash_after_first=True)
    except Crash as e:
        print(f"    !! CRASH: {e}")
        print(f"    !! request req-1 status = "
              f"{status.get('req-1', 'NOT done')} -> must be redelivered")
    print("    --- recovery: request still un-done, reprocess from the log "
          "(at-least-once) ---")
    process_request(req, payer, payee, status)
    invariant(payer, payee, total)
    print()


def scenario_crash_nodedup():
    print(SEP)
    print("SCENARIO 3 -- SAME crash, dedup OFF (redelivery double-credits)")
    print(SEP)
    payer = Shard("A", 100, dedup=False)
    payee = Shard("B", 100, dedup=False)
    total = payer.balance + payee.balance
    print(f"    initial: A=100, B=100, total={total}")
    print("    request-log append (the ONE atomic action): "
          "req-1 = transfer 11 A->B")
    req = {"rid": "req-1", "amount": 11}
    status = {}
    print("    --- first delivery: crash after crediting payee ---")
    try:
        process_request(req, payer, payee, status, crash_after_first=True)
    except Crash as e:
        print(f"    !! CRASH: {e}")
        print(f"    !! request req-1 status = "
              f"{status.get('req-1', 'NOT done')} -> must be redelivered")
    print("    --- recovery: request still un-done, reprocess from the log "
          "(at-least-once) ---")
    process_request(req, payer, payee, status)
    invariant(payer, payee, total)
    print()


if __name__ == "__main__":
    scenario_happy()
    scenario_crash_dedup()
    scenario_crash_nodedup()
    print(SEP)
    print("SUMMARY")
    print(SEP)
    print("  1 happy path:       A=89,  B=111, sum=200  invariant HELD, each touched once")
    print("  2 crash + dedup:    A=89,  B=111, sum=200  invariant HELD, each touched once")
    print("  3 crash - no dedup: A=89,  B=122, sum=211  invariant VIOLATED, payee touched 2x (money created)")