Event-driven Architecture ও Event Sourcing — state যখন log

"""eventstore.py — mini event-store (~110 line)"""
import json

class VersionConflict(Exception):
    pass

FOLDERS = {
    'Opened':    lambda st, e: {**st, 'bal': e['data'].get('initial', 0)},
    'Deposited': lambda st, e: {**st, 'bal': st.get('bal', 0) + e['data']['amount']},
    'Withdrawn': lambda st, e: {**st, 'bal': st.get('bal', 0) - e['data']['amount']},
}

def fold_events(stream, state=None):
    st = state if state is not None else {}
    for e in stream:
        st = FOLDERS[e['type']](st, e)
    return st

class EventStore:
    def __init__(self, snapshot_every=2):
        self.streams = {}          # agg-id -> list[event]
        self.snapshots = {}        # agg-id -> (version, state)
        self.every = snapshot_every

    def append(self, agg, etype, data, expected_version):
        stream = self.streams.setdefault(agg, [])
        if len(stream) != expected_version:                 # CAS-style guard
            raise VersionConflict(
                f'{agg}: want v{expected_version}, store has v{len(stream)}')
        e = {'agg': agg, 'type': etype, 'version': len(stream) + 1,
             'data': data}
        stream.append(e)
        if len(stream) % self.every == 0:                   # checkpoint!
            self.snapshots[agg] = (len(stream), fold_events(stream))
        return e['version']

    def load(self, agg):
        """snapshot থেকে শুরু — শুধু tail fold"""
        stream = self.streams.get(agg, [])
        start, base = self.snapshots.get(agg, (0, None))
        return fold_events(stream[start:], state=base)

    def raw_fold(self, agg):
        return fold_events(self.streams.get(agg, []))       # no-snapshot path


class Projection:
    """live read-model — নতুন event আসতেই apply"""
    def __init__(self):
        self.balances = {}

    def apply(self, e):
        agg, amt = e['agg'], e['data'].get('amount')
        cur = self.balances.get(agg, 0)
        if e['type'] == 'Opened':
            self.balances[agg] = e['data'].get('initial', 0)
        elif e['type'] == 'Deposited':
            self.balances[agg] = cur + amt
        elif e['type'] == 'Withdrawn':
            self.balances[agg] = cur - amt

    def rebuild_from(self, store):
        self.balances = {}                                  # scratch!
        for agg in sorted(store.streams):
            for e in store.streams[agg]:
                self.apply(e)
        return self


if __name__ == '__main__':
    store = EventStore(snapshot_every=2)

    v1 = store.append('A1', 'Opened', {'initial': 0}, expected_version=0)
    v2 = store.append('A1', 'Deposited', {'amount': 5000}, expected_version=v1)
    v3 = store.append('A1', 'Withdrawn', {'amount': 1200}, expected_version=v2)

    try:                                                    # stale writer!
        store.append('A1', 'Deposited', {'amount': 9},
                     expected_version=1)
    except VersionConflict as err:
        print(f'conflict blocked: {err}')

    store.append('A1', 'Deposited', {'amount': 300}, expected_version=v3)
    store.append('B2', 'Opened', {'initial': 100}, expected_version=0)

    print(f"load(A1) w/ snapshot : {store.load('A1')}")
    print(f"raw fold(A1)         : {store.raw_fold('A1')}")
    assert store.load('A1') == store.raw_fold('A1')

    proj = Projection()
    for agg in ('A1', 'B2'):                                # live-apply all
        for e in store.streams[agg]:
            proj.apply(e)
    fresh = Projection().rebuild_from(store)                # scratch rebuild
    print(f"live projection : {proj.balances}")
    print(f"rebuilt         : {fresh.balances}")
    assert proj.balances == fresh.balances
    print("PASS — snapshot-load==raw-fold, live==rebuilt")

নমুনা output:

conflict blocked: A1: want v1, store has v3
load(A1) w/ snapshot : {'bal': 4100}
raw fold(A1)         : {'bal': 4100}
live projection : {'A1': 4100, 'B2': 100}
rebuilt         : {'A1': 4100, 'B2': 100}
PASS — snapshot-load==raw-fold, live==rebuilt

তিনটা পাঠ:

• Stale-writer rejection-টাই aggregate-concurrency — distributed lock ছাড়াই safety, conflict-এ retry।
• Snapshot ভুল হলেও raw_fold শেষ কথা — dual-path assertion production event-store-এর nightly-integrity-check।
• Live-projection আর rebuild-এর equality প্রমাণ করে read-model disposable — এটাই CQRS-এর সাহস: query-side ভয়-হীনভাবে change করা যায়।

নিজে বাড়ান:

1. Upcaster-chain যোগ করুন — Withdrawn v1(amount) → v2({amount, currency:'BDT'}), পুরনো fixture দিয়ে পরীক্ষা
2. Crypto-shredding stub — personal-field user-key-এ encrypt, key-delete-এ fold-এ field অপাঠ্য হওয়া simulate
3. Multi-aggregate projection (‘total deposits per day’) — event timestamp-এ group করে আলাদা read-dict