#!/usr/bin/env python3
"""Dual writes diverge two stores forever; one ordered CDC log converges them, DDIA 2e Ch. 12.

Two clients concurrently SET the same key X: client 1 wants X=A, client 2 wants
X=B. Each client writes to a "database" AND to a "search index" as two SEPARATE
writes (a dual write). There is no shared order between the two write paths, so
the four writes can interleave. Interleave them (DDIA Figure 12-4) so the DATABASE
applies A-then-B (ends on B) while the INDEX applies B-then-A (ends on A). The two
stores now disagree -- db[X]=B, index[X]=A -- permanently, and nothing threw.

The fix (Figure 12-5): route every change through ONE ordered change-data-capture
log. A leader serializes the writes into a single sequence [(X,A),(X,B)]. The db
applies the log in order (ends on B); the index is a FOLLOWER that replays the SAME
log in the SAME order (also ends on B). Both derived stores are now a deterministic
function of one order, so they cannot diverge -- they converge to B.

Pure standard library. The interleaving is modeled explicitly, so it is
deterministic and reproduces exactly every run.
"""

KEY = "X"


def banner(title):
    print(title)
    print("-" * len(title))


# --- Part 1: dual writes -- two independent, unordered write paths ---------------
#
# Two stores, each a plain dict. Each client issues FOUR writes total across the
# two clients, as separate ops with no shared order. We model one specific
# interleaving (Figure 12-4) as an explicit list of (store, client, value) ops.
# Each client preserves its own db-before-index order; the clients interleave.

def dual_write_divergence():
    db = {}
    index = {}

    # The four writes, in the interleaved order they actually commit. Client 1
    # writes A, client 2 writes B; each writes db first, then index. The clients
    # race, so the commit order is:
    ops = [
        ("db",    "client 1", "A"),   # 1. client 1 sets db[X]   = A
        ("db",    "client 2", "B"),   # 2. client 2 sets db[X]   = B   -> db ends on B
        ("index", "client 2", "B"),   # 3. client 2 sets index[X]= B
        ("index", "client 1", "A"),   # 4. client 1 sets index[X]= A   -> index ends on A
    ]

    banner("Part 1 -- dual writes: two clients, two stores, four unordered writes")
    print(f'    client 1 wants {KEY}=A ; client 2 wants {KEY}=B')
    print(f'    each client writes the DB and the INDEX as two separate writes\n')
    print("    the four writes interleave and commit in this order (Figure 12-4):")
    for i, (store, client, value) in enumerate(ops, 1):
        if store == "db":
            db[KEY] = value
        else:
            index[KEY] = value
        print(f'      {i}. {client:8s} -> {store:5s}[{KEY}] = {value}')

    print(f'\n    final state:  db[{KEY}] = {db[KEY]}   index[{KEY}] = {index[KEY]}')
    diverged = db[KEY] != index[KEY]
    assert diverged, "expected the stores to diverge"
    print(f'    db != index  ->  DIVERGED, and nothing raised an error.')
    print(f'    the database says {db[KEY]}, the search index says {index[KEY]} -- forever.\n')
    return db[KEY], index[KEY]


# --- Part 2: one ordered CDC log -- the index is a follower that replays it -------
#
# Now the leader serializes both clients' writes into a SINGLE ordered log. The db
# is the leader (it applies the log to produce its state); the index is a follower
# that replays the exact same log in the exact same order. Only one write path now.

def cdc_convergence():
    # One leader decides one order for the two SETs. This is the whole fix.
    log = [(KEY, "A"), (KEY, "B")]

    banner("Part 2 -- one ordered CDC log: the leader serializes, the index follows")
    print("    both clients' writes go through ONE ordered change log (Figure 12-5):")
    for i, (k, v) in enumerate(log, 1):
        print(f'      offset {i}: SET {k} = {v}')
    print()

    # The database applies the log in order.
    db = {}
    for k, v in log:
        db[k] = v

    # The search index is a FOLLOWER: it replays the SAME log in the SAME order.
    index = {}
    for k, v in log:
        index[k] = v

    print(f'    db     replays the log in order -> db[{KEY}]    = {db[KEY]}')
    print(f'    index  replays the SAME log     -> index[{KEY}] = {index[KEY]}')
    converged = db[KEY] == index[KEY]
    assert converged and db[KEY] == "B", "expected both stores to converge to B"
    print(f'    db == index == {db[KEY]}  ->  CONVERGED.')
    print(f'    the index no longer races the db; it just replays the db\'s order.\n')
    return db[KEY], index[KEY]


def main():
    db_v, index_v = dual_write_divergence()
    cdc_db, cdc_index = cdc_convergence()

    banner("Summary")
    print(f'    dual writes:  db={db_v}  index={index_v}   -> DIVERGED (no error)')
    print(f'    one CDC log:  db={cdc_db}  index={cdc_index}   -> CONVERGED')


if __name__ == "__main__":
    main()