Event sourcing আর CQRS — event থেকে state আর আলাদা read model

১ · খাতার শেষ লাইন, না পুরো খাতা — state রাখা বনাম ঘটনা রাখা

পাড়ার মুদি দোকানের বাকির খাতাটি ভাবুন। দোকানি কখনো লেখেন না “রহিম সাহেবের বাকি এখন ৭৫০ টাকা” আর আগের লাইন কেটে দেন না। তিনি লেখেন: “৩ তারিখ — চাল, ৫০০ টাকা”, “৮ তারিখ — তেল, ৪০০ টাকা”, “১২ তারিখ — নগদ জমা ১৫০ টাকা”। মোট বাকি জানতে চাইলে তিনি লাইনগুলো যোগ-বিয়োগ করেন। কোনো লাইন ভুল হলে সেটি ঘষে মোছেন না — নিচে একটি সংশোধনী লাইন লেখেন। মাস শেষে ঝগড়া হলে খাতা খুলে দেখানো যায় কোন দিন কী হয়েছিল।

সফটওয়্যারের বেশিরভাগ ডেটাবেস কিন্তু উল্টোটা করে। একটি mobile wallet সেবার (কাল্পনিক) accounts টেবিলে একটি সারি থাকে: A, balance = 1450. টাকা তুললে সারিটি UPDATE হয়ে যায় — আগের মান হারিয়ে যায়। একে বলে state-oriented নকশা: ডেটাবেস জানে এখন কী আছে, কিন্তু কীভাবে এখানে এলো, তা জানে না।

এবার একটি বাসের সিট। ঈদের আগে একটি দূরপাল্লার বাসের (কাল্পনিক অপারেটর) সিট A3-এর state-oriented সারিতে আছে A3 · BOOKED · যাত্রী: নাসরিন. গ্রাহকসেবায় আরেকজন যাত্রী ফোন করে বললেন, “আমি তো সকালে এই সিটই কিনেছিলাম!” সারিটি কিছুই বলতে পারে না। Event রাখা নকশায় সিটের stream-এ থাকে: SeatHeld(রবিন, ০৯:০২), HoldExpired(০৯:১৭), SeatHeld(নাসরিন, ০৯:১৮), SeatBooked(নাসরিন, ০৯:২০). রবিনের hold-এর মেয়াদ ১৫ মিনিটে ফুরিয়েছিল, টাকা কাটেনি — এক নজরে উত্তর মেলে।

এই ধারণা নতুন নয়। ব্যাংকের ledger ঠিক এভাবে কাজ করে — হিসাবরক্ষকেরা কয়েকশো বছর ধরে লেনদেনের লাইন জমিয়ে balance বের করেন (অংশ ৯-এ ফিরব)। ডেটাবেসের ভেতরেও একই কৌশল: পর্ব ১৯-এর write-ahead log প্রতিটি বদল আগে log-এ লেখে, টেবিলটি পরে সেই log থেকেই মেলানো যায়। Event sourcing এই log-কে ডেটাবেসের ভেতর থেকে বের করে এনে ব্যবসার ভাষায় লেখে — “page 812-এর byte বদলেছে” নয়, “রহিম ৮০০ টাকা পাঠিয়েছেন”।

২ · Event sourcing-এর যন্ত্র — command, event, aggregate আর fold

চারটি শব্দ বুঝলেই পুরো যন্ত্রটি পরিষ্কার।

গাণিতিকভাবে লিখলে: s₀ = প্রাথমিক অবস্থা, আর s_n = apply(s_n−1, e_n). তাই s_n = apply(…apply(apply(s₀, e₁), e₂)…, e_n). এর থেকে দুটি কঠোর শর্ত আসে:

• apply হবে বিশুদ্ধ (pure): শুধু পুরোনো অবস্থা আর event দেখে নতুন অবস্থা ফেরত দেবে — ঘড়ি দেখবে না, random সংখ্যা নেবে না, অন্য সেবাকে ডাকবে না। তাহলেই আজ আর দশ বছর পরে একই event থেকে একই অবস্থা আসবে।
• সিদ্ধান্ত আর side effect থাকবে command-এর ধাপে: “balance যথেষ্ট কি না” যাচাই হয় command নেওয়ার সময়; SMS পাঠানো হয় event লেখার পরে, আলাদা কোনো subscriber থেকে। apply-এর ভেতরে SMS থাকলে প্রতিটি rebuild-এ গ্রাহক পুরোনো সব SMS আবার পেতেন।

নিচের stepper-এ একটি wallet-এর চারটি command আর শেষে একটি পুরো rebuild দেখুন। খেয়াল করুন, প্রত্যাখ্যাত command কোনো event রেখে যায় না।

৩ · Event store আর expected version — একই stream-এ দুজন লেখক

Event store হলো event রাখার ডেটাবেস। তার কাছে মূলত দুটি কাজ চাওয়া হয়: (১) একটি stream-এর সব event ক্রমানুসারে পড়া, আর (২) একটি stream-এর শেষে নতুন event যোগ করা — একটি শর্তসহ। কোনো UPDATE নেই, সাধারণত DELETE-ও নেই। আলাদা পণ্য আছে (যেমন open-source EventStoreDB), আবার PostgreSQL-এর মতো সাধারণ relational ডেটাবেসেও একটি টেবিল দিয়ে কাজটি ভালোভাবে চলে।

শর্তটি কেন? ধরুন wallet A-তে এখন version ৪, balance ৩,৩০০. দুটি app সার্ভার প্রায় একসাথে দুটি command পেল — একটিতে ২,০০০ তোলা, অন্যটিতে ২,৫০০. দুজনেই stream পড়ে দেখল ৩,৩০০, দুজনেরই নিয়ম মিলল। দুজনেই event যোগ করলে balance হবে −১,২০০ — নিয়ম ভাঙল, যদিও প্রত্যেকে আলাদাভাবে ঠিক কাজ করেছে। এটি পর্ব ২৪-এর lost update-এরই আরেক রূপ।

সমাধান optimistic concurrency, stream-এর ওপর। প্রতিটি append-এর সাথে client বলে: “আমি version ৪ দেখে এই সিদ্ধান্ত নিয়েছি — stream এখনো ৪-এ থাকলে তবেই লিখুন।” একে বলে expected version. Store মিলিয়ে দেখে; না মিললে লেখা প্রত্যাখ্যান করে, আর client stream আবার পড়ে, নিয়ম আবার যাচাই করে। ব্যাংকের কাউন্টারে টোকেনের মতো: আপনার টোকেন নম্বর ডাকার আগেই অন্য কেউ সেই কাউন্টারে বসে গেলে আপনাকে আবার নতুন করে ডাকা হবে।

Relational ডেটাবেসে এই শর্তের সবচেয়ে সহজ রূপ একটি UNIQUE constraint: UNIQUE (stream_id, version). একই stream-এ একই version দুবার থাকতে পারবে না, তাই দুজন একই expected version থেকে version = 5 লিখতে গেলে দ্বিতীয়জন ব্যর্থ হবেই — app-এর কোনো ভুল এটি এড়াতে পারে না। নিচের live SQL-এ চালিয়ে দেখুন; শেষ লাইনটি ইচ্ছা করে ব্যর্থ হয়:

-- Two servers read wallet A at version 4 (balance 3300). Each tries to append version 5.
-- Server 1 appends first and wins.
INSERT INTO events (stream_id, version, type, amount, at) VALUES ('A', 5, 'Withdrew', 2000, '2026-09-15');
-- Server 2 uses the same expected version; ON CONFLICT turns the clash into "0 rows".
INSERT INTO events (stream_id, version, type, amount, at) VALUES ('A', 5, 'Withdrew', 2500, '2026-09-15')
ON CONFLICT (stream_id, version) DO NOTHING;
-- Server 2 reloads: version and balance have moved, so it must re-check its rule.
SELECT MAX(version) AS version,
       SUM(CASE WHEN type = 'Deposited' OR dir = 'in' THEN amount ELSE -amount END) AS balance
FROM events WHERE stream_id = 'A';
-- Without the reload, a plain INSERT at the stale version is refused by the database itself:
INSERT INTO events (stream_id, version, type, amount, at) VALUES ('A', 5, 'Withdrew', 2500, '2026-09-15');

৪ · Rebuild আর snapshot — replay-র দাম হিসাব

প্রতিটি command-এর আগে বর্তমান অবস্থা লাগে, আর অবস্থা আসে fold থেকে। একটি wallet-এ দিনে গড়ে ১০টি event হলে তিন বছরে প্রায় ১০,৯৫০টি। প্রতিবার সবগুলো পড়া কি সম্ভব? হিসাব করে দেখি। ধরা যাক একটি event store থেকে পড়ে প্রয়োগ করতে r = ২০ µs লাগে (অনুমান)।

এক command-এর আগে ২১৯ ms শুধু অতীত পড়তে — অগ্রহণযোগ্য। সমাধান snapshot: প্রতি k-টি event পরপর ওই মুহূর্তের অবস্থা (version আর balance) আলাদা করে লিখে রাখা। লোড করার সময় সর্বশেষ snapshot পড়ে শুধু তার পরের event-গুলো fold করা হয়। Snapshot একটি cache মাত্র — হারালে বা ভুল হলে মুছে event থেকে আবার বানানো যায়; আসল সত্য event-ই।

হিসাবটি সংখ্যায়

Stream-এ L-টি event, snapshot প্রতি k-টিতে, snapshot পড়তে c, লিখতে S সময়। তাহলে:

• Replay-র event সংখ্যা = L mod k. সবচেয়ে খারাপ অবস্থায় k − ১, গড়ে মোটামুটি (k − ১)/২ (শেষ snapshot-এর পরে কোথায় আছেন, তা ০ থেকে k − ১-এর মধ্যে সমান সম্ভাবনায় ধরে)।
• উদাহরণ: L = ১০,৯৫০, k = ১০০, c = ১ ms (অনুমান)। L mod k = ৫০, তাই এখন ১ + ৫০ × ০.০২ = ২ ms. সবচেয়ে খারাপে ১ + ৯৯ × ০.০২ = ২.৯৮ ms, গড়ে ১ + ৪৯.৫ × ০.০২ ≈ ১.৯৯ ms. ২১৯ ms থেকে প্রায় ১০০ গুণ কম।
• দাম: প্রতি k-টি append-এ একটি বাড়তি snapshot লেখা — তিন বছরে ⌊১০,৯৫০ / ১০০⌋ = ১০৯টি, যদিও সাধারণত শুধু সর্বশেষটিই রাখা হয়।

সর্বোত্তম k — একটি ছোট প্রমাণ

ধরুন প্রতিটি command আগে stream লোড করে (memory-তে কিছু রাখা হয় না)। তাহলে প্রতি command-এ গড় খরচ = replay (r × (k − ১)/২) + snapshot লেখার ভাগ (S/k), সঙ্গে snapshot পড়ার স্থির খরচ c:

দ্বিতীয় অন্তরজ ২S/k³ > ০, তাই এটি সত্যিই সর্বনিম্ন। S = ২ ms, r = ০.০২ ms হলে k* = √২০০ ≈ ১৪; তখন বাড়তি খরচ প্রায় ০.২৭ ms, যেখানে k = ১০০-তে প্রায় ১.০১ ms. ছোট k-তে বারবার snapshot লেখার খরচ, বড় k-তে বেশি replay-র খরচ — মাঝখানে বর্গমূলের নিয়ম। বাস্তবে aggregate প্রায়ই memory-তে cache থাকে, তখন replay শুধু ঠান্ডা শুরুতে (restart, নতুন সার্ভার) হয় — আর snapshot লেখা প্রায়ই command-এর পথ থেকে সরিয়ে পেছনের একটি কাজে দেওয়া হয়। তাই সূত্রটি শুরু-বিন্দু, চূড়ান্ত উত্তর নয়; মাপুন।

এবার পুরো wallet-টি একটি simulator-এ: তিনটি event type, fold, প্রতিটি append-এ expected version-এর যাচাই, আর প্রতি ৪টি event-এ snapshot:

// Event-sourced wallet: balance = fold(events). Expected-version check + a snapshot every N events.
const N = 4, store = [], snaps = {};     // store: append-only list of events; snaps: latest per stream
function apply(bal, e) {                 // pure: old state + one event -> new state
  if (e.type === "Deposited") return bal + e.amt;
  if (e.type === "Withdrew") return bal - e.amt;
  return e.dir === "out" ? bal - e.amt : bal + e.amt;            // Transferred
}
function load(id, useSnap = true) {      // rebuild current state; count how many events were replayed
  const s = (useSnap && snaps[id]) || { version: 0, bal: 0 };
  const evs = store.filter((e) => e.stream === id && e.version > s.version);
  return { version: s.version + evs.length, bal: evs.reduce(apply, s.bal), replayed: evs.length };
}
function append(id, expected, e) {      // the store refuses if someone appended after we read
  const cur = load(id).version;
  if (cur !== expected) throw new Error(`conflict on ${id}: expected v${expected}, found v${cur}`);
  store.push({ ...e, stream: id, version: cur + 1 });
  if ((cur + 1) % N === 0) snaps[id] = { version: cur + 1, bal: load(id).bal };
}
function command(id, type, amt, extra = {}) {   // read -> check the rule -> append with expected version
  const s = load(id);
  if ((type === "Withdrew" || extra.dir === "out") && s.bal < amt) return console.log(`REJECTED ${type} ${amt} on ${id}: balance ${s.bal}`);
  append(id, s.version, { type, amt, ...extra });
}
command("A", "Deposited", 5000); command("A", "Withdrew", 1200);
command("A", "Transferred", 800, { dir: "out" }); command("B", "Transferred", 800, { dir: "in" });
command("A", "Deposited", 300); command("A", "Withdrew", 9000); command("A", "Deposited", 150);
const c1 = load("A"), c2 = load("A");          // two app servers read A at the same version
append("A", c1.version, { type: "Withdrew", amt: 2000 });
try { append("A", c2.version, { type: "Withdrew", amt: 2500 }); } catch (err) { console.log(err.message); }
command("A", "Withdrew", 2500);                // c2 reloads, re-checks the rule, then retries
for (const id of ["A", "B"]) {
  const a = load(id), b = load(id, false);
  console.log(`${id}: v${a.version} balance=${a.bal} | snapshot v${snaps[id] ? snaps[id].version : 0} -> replayed ${a.replayed} | no snapshot -> replayed ${b.replayed}, balance=${b.bal}`);
}
console.log("events in store: " + store.length + ", without the version check A would be " + (c2.bal - 2000 - 2500));

REJECTED Withdrew 9000 on A: balance 3300
conflict on A: expected v5, found v6
REJECTED Withdrew 2500 on A: balance 1450
A: v6 balance=1450 | snapshot v4 -> replayed 2 | no snapshot -> replayed 6, balance=1450
B: v1 balance=800 | snapshot v0 -> replayed 1 | no snapshot -> replayed 1, balance=800
events in store: 7, without the version check A would be -1050

৫ · Event-এর গঠন বদলায় — versioning আর upcasting

State-oriented ডেটাবেসে কলাম বদলালে একটি migration চালিয়ে সব সারি নতুন রূপে আনা যায়। Event store-এ নিয়ম উল্টো: event পরিবর্তনহীন, তিন বছর আগের Deposited {amt: 5000} চিরকাল সেই রূপেই থাকবে। অথচ এর মধ্যে ব্যবসা বদলেছে — টাকা এখন পয়সায় রাখা হয় (দশমিকের গোলমাল এড়াতে), আর কোন চ্যানেলে জমা (এজেন্ট, ব্যাংক, কার্ড) সেটিও লাগে। এটি পর্ব ১২-এর schema evolution-এর সবচেয়ে কঠিন রূপ, কারণ পুরোনো ডেটা কখনো মরে না।

প্রচলিত কৌশলগুলো:

• প্রতিটি event-এ version নম্বর: Deposited v1, v2, v3 — পড়ার কোড জানে কোনটি কোন রূপ।
• Upcasting: পড়ার সময় একটি ছোট ফাংশনের শিকল পুরোনো event-কে ধাপে ধাপে সর্বশেষ রূপে তোলে (v1 → v2 → v3). Store-এর event বদলায় না; fold আর projection শুধু সর্বশেষ রূপ চেনে।
• যোগ করুন, মানে বদলাবেন না: নতুন ঐচ্ছিক field যোগ নিরাপদ; পুরোনো field-এর অর্থ বদলানো (টাকা থেকে পয়সা, একই নামে) সবচেয়ে বিপজ্জনক — নতুন নাম বা নতুন version নিন।
• শেষ উপায় — stream কপি করে রূপান্তর: পুরো stream পড়ে নতুন রূপে নতুন stream-এ লেখা। Audit-এর কারণে পুরোনোটিও রেখে দিতে হয়, আর কাজটি ঝুঁকিপূর্ণ।

// Upcasting: old events stay as written; on read, a chain of small functions lifts them to the latest shape.
const stored = [                                   // three eras of the same wallet, never rewritten
  { type: "Deposited", v: 1, data: { amt: 5000 } },                                 // 2024: whole taka
  { type: "Deposited", v: 2, data: { poisha: 30000 } },                             // 2025: poisha, no channel
  { type: "Deposited", v: 3, data: { poisha: 15050, channel: "agent" } },           // 2026: current shape
  { type: "Withdrew",  v: 1, data: { amt: 1200 } },
];
const upcasters = {                                // "type@version" -> function returning the next version
  "Deposited@1": (d) => ({ poisha: d.amt * 100 }),
  "Deposited@2": (d) => ({ ...d, channel: "unknown" }),
  "Withdrew@1":  (d) => ({ poisha: d.amt * 100 }),
};
const LATEST = { Deposited: 3, Withdrew: 2 };
function upcast(e) {
  let { v, data } = e, steps = 0;
  while (v < LATEST[e.type]) { data = upcasters[e.type + "@" + v](data); v++; steps++; }
  return { type: e.type, v, data, steps };
}
let poisha = 0;
for (const e of stored.map(upcast)) {             // the fold only ever sees the latest shape
  poisha += (e.type === "Deposited" ? 1 : -1) * e.data.poisha;
  console.log(`${e.type} v${e.v} (${e.steps} upcast steps) ${JSON.stringify(e.data)}`);
}
console.log("balance: " + (poisha / 100).toFixed(2) + " taka; stored events unchanged: " + JSON.stringify(stored[0].data));

Deposited v3 (2 upcast steps) {"poisha":500000,"channel":"unknown"}
Deposited v3 (1 upcast steps) {"poisha":30000,"channel":"unknown"}
Deposited v3 (0 upcast steps) {"poisha":15050,"channel":"agent"}
Withdrew v2 (1 upcast steps) {"poisha":120000}
balance: 4250.50 taka; stored events unchanged: {"amt":5000}

৬ · CQRS — লেখার মডেল এক, পড়ার মডেল অনেক

রেস্তোরাঁর রান্নাঘর আর ওয়েটারের কথা ভাবুন। রান্নাঘর (লেখার দিক) কঠোর নিয়মে চলে — কোন অর্ডার আগে, কোন উপকরণ শেষ, কোন পদ বাতিল। ওয়েটার (পড়ার দিক) রান্নাঘরের খাতা খুলে গ্রাহককে উত্তর দেন না; তাঁর হাতে থাকে একটি সাজানো মেনু-বোর্ড আর টেবিলভিত্তিক অর্ডারের তালিকা — রান্নাঘরের খবর থেকে বানানো, গ্রাহকের প্রশ্নের মাপে সাজানো।

CQRS ঠিক এই ভাগ। Write model command নেয়, নিয়ম যাচাই করে, event (বা state) লেখে — এখানে সংগতি আর সঠিকতা প্রধান। Read model শুধু প্রশ্নের উত্তর দেয়, আর প্রতিটি পাতা বা প্রশ্নের জন্য আলাদা আকারে সাজানো থাকে — balance পাতার জন্য একটি ছোট key-value, মাসিক statement-এর জন্য একটি টেবিল, খোঁজার জন্য একটি search index (পর্ব ২৭)। Read model আসলে একটি materialized view বা denormalized কপি (পর্ব ২১), যা event থেকে বানানো হয়।

Event থেকে read model বানানোর কাজটিকে বলে projection, আর যে প্রক্রিয়া এটি চালায় তাকে projector. প্রতিটি projector event log পড়ে, প্রতিটি event-এ নিজের টেবিল আপডেট করে, আর মনে রাখে কোন event পর্যন্ত পড়া হয়েছে — তার checkpoint. ব্যাপারটি পর্ব ৩৩-এর consumer-এর offset রাখার মতোই।

একই event log থেকে দুটি projection, একটি পিছিয়ে থাকা projector, তারপর অক্টোবরে আসা নতুন একটি চাহিদা — “২,০০০ বা তার বেশি টাকার সব লেনদেনের তালিকা” — যার read model শুরু থেকে replay করে পেছনের তারিখসহ বানানো হলো:

// One event log, several read models (projections). Each projector keeps its own checkpoint.
const log = [   // [position, date, stream, type, amount, direction]
  [1, "2026-08-03", "A", "Deposited", 5000], [2, "2026-08-10", "A", "Withdrew", 1200],
  [3, "2026-08-21", "A", "Transferred", 800, "out"], [4, "2026-08-21", "B", "Transferred", 800, "in"],
  [5, "2026-09-02", "A", "Deposited", 300], [6, "2026-09-05", "B", "Withdrew", 500],
  [7, "2026-09-15", "A", "Withdrew", 2000], [8, "2026-10-01", "B", "Deposited", 6000],
  [9, "2026-10-04", "A", "Deposited", 150],
].map(([pos, date, stream, type, amt, dir]) => ({ pos, date, stream, type, amt, dir }));
const sign = (e) => (e.type === "Deposited" || e.dir === "in" ? +1 : -1);
function projector(name, handler, view) {                  // a projector = handler + view + checkpoint
  const p = { name, view, checkpoint: 0, applied: 0 };
  p.catchUp = (upto = log.length) => {
    p.applied = 0;
    for (const e of log) if (e.pos > p.checkpoint && e.pos <= upto) { handler(p.view, e); p.checkpoint = e.pos; p.applied++; }
    return p;
  };
  return p;
}
const balances = projector("balances", (v, e) => { v[e.stream] = (v[e.stream] || 0) + sign(e) * e.amt; }, {});
const monthly = projector("monthly", (v, e) => {
  const k = e.stream + " " + e.date.slice(0, 7), row = (v[k] = v[k] || { in: 0, out: 0 });
  row[sign(e) > 0 ? "in" : "out"] += e.amt;
}, {});
balances.catchUp(6); monthly.catchUp();                     // the balance projector is 3 events behind
console.log("balances @" + balances.checkpoint + ": " + JSON.stringify(balances.view) + "  (stale)");
balances.catchUp();
console.log("balances @" + balances.checkpoint + ": " + JSON.stringify(balances.view) + "  (caught up: " + balances.applied + " new events)");
for (const [k, r] of Object.entries(monthly.view).sort()) console.log("statement " + k + ": in " + r.in + ", out " + r.out);
// New requirement in October: list every movement of 2000 or more -- built from position 1.
const large = projector("large", (v, e) => { if (e.amt >= 2000) v.push(e.date + " " + e.stream + " " + e.type + " " + e.amt); }, []);
console.log("large (new, rebuilt from start): " + large.catchUp().view.join(" | "));
balances.view = {}; balances.checkpoint = 0;                // drop and rebuild a read model
console.log("balances rebuilt: " + JSON.stringify(balances.catchUp().view) + ", replayed " + balances.applied + " events");

balances @6: {"A":3300,"B":300}  (stale)
balances @9: {"A":1450,"B":6300}  (caught up: 3 new events)
statement A 2026-08: in 5000, out 2000
statement A 2026-09: in 300, out 2000
statement A 2026-10: in 150, out 0
statement B 2026-08: in 800, out 0
statement B 2026-09: in 0, out 500
statement B 2026-10: in 6000, out 0
large (new, rebuilt from start): 2026-08-03 A Deposited 5000 | 2026-09-15 A Withdrew 2000 | 2026-10-01 B Deposited 6000
balances rebuilt: {"A":1450,"B":6300}, replayed 9 events

Event store যদি SQL টেবিল হয়, projection একটি query মাত্র — আর সেটি টেবিল হিসেবে জমা রাখলেই read model. নিচে balance, মাসিক statement, “৩১ আগস্টে balance কত ছিল” (time travel) আর একটি জমা-রাখা read model:

-- Read model 1: current balance per wallet (a projection = a fold over events).
SELECT stream_id, MAX(version) AS version,
       SUM(CASE WHEN type = 'Deposited' OR dir = 'in' THEN amount ELSE -amount END) AS balance
FROM events GROUP BY stream_id ORDER BY stream_id;
-- Read model 2: monthly statement of wallet A.
SELECT substr(at, 1, 7) AS month,
       SUM(CASE WHEN type = 'Deposited' OR dir = 'in' THEN amount ELSE 0 END) AS money_in,
       SUM(CASE WHEN type = 'Deposited' OR dir = 'in' THEN 0 ELSE amount END) AS money_out
FROM events WHERE stream_id = 'A' GROUP BY month ORDER BY month;
-- Time travel: what was A's balance at the end of August?
SELECT SUM(CASE WHEN type = 'Deposited' OR dir = 'in' THEN amount ELSE -amount END) AS a_on_aug_31
FROM events WHERE stream_id = 'A' AND at <= '2026-08-31';
-- Materialise read model 1 as a table that queries can hit directly.
CREATE TABLE balance_view AS
SELECT stream_id, MAX(pos) AS checkpoint,
       SUM(CASE WHEN type = 'Deposited' OR dir = 'in' THEN amount ELSE -amount END) AS balance
FROM events GROUP BY stream_id;
SELECT * FROM balance_view ORDER BY stream_id;

৭ · Projection পিছিয়ে থাকে — eventual consistency আর read-your-writes

Write model-এ event লেখা আর read model-এ তা দেখা যাওয়ার মাঝে একটি ফাঁক থাকে — projector-কে event পড়তে, প্রয়োগ করতে আর checkpoint সরাতে হয়। সাধারণত এটি মিলিসেকেন্ড থেকে কয়েক সেকেন্ড, কিন্তু projector বিকল হলে বা ভারী rebuild চললে মিনিট-ঘণ্টাও হতে পারে। তাই read model eventually consistent: নতুন লেখা থামলে শেষ পর্যন্ত মিলবে, মাঝের সময়ে পুরোনো মান দেখাতে পারে। সবচেয়ে বিরক্তিকর রূপটি হলো নিজের লেখা নিজে না দেখা — পর্ব ২২-এর read-your-writes ভাঙা।

Read-your-writes রক্ষার প্রচলিত কৌশল:

• Position বা version দিয়ে অপেক্ষা: উপরের ধাপ ৫ — reply-তে position, পড়ার সময় checkpoint ≥ position পর্যন্ত অপেক্ষা (একটি সময়সীমাসহ)।
• নিজের ডেটা write model থেকে: লেখার পরের কয়েক সেকেন্ড শুধু সেই ব্যবহারকারীর নিজের aggregate fold করে দেখানো।
• UI-তে আশাবাদী আপডেট: সফল reply পেলে পর্দায় নিজেই নতুন মান বসানো — কিন্তু অন্য ডিভাইসে খুললে কাজ করে না।
• একই transaction-এ read model: এক ডেটাবেসে থাকলে event আর ছোট একটি read model একসাথে commit — তখন সেই read model আর eventual নয়, তবে লেখা ধীর হয় আর projector-এর স্বাধীনতা কমে।

৮ · লাভ, দাম আর কখন নয় — audit, time travel, মুছে ফেলা আর over-engineering

মুছে ফেলার অধিকার বনাম পরিবর্তনহীন log

অনেক দেশের ব্যক্তিগত তথ্য সুরক্ষা আইন ব্যবহারকারীকে নিজের তথ্য মুছে দিতে বলার অধিকার দেয় — সবচেয়ে পরিচিত উদাহরণ ইউরোপীয় ইউনিয়নের GDPR (২০১৮ সাল থেকে প্রযোজ্য), যার “right to erasure”. অথচ event store-এর মূলনীতি “কখনো মুছবেন না”। প্রচলিত কৌশল:

• ব্যক্তিগত তথ্য event-এর বাইরে: event-এ শুধু একটি customer ID; নাম, ফোন, ঠিকানা আলাদা একটি সাধারণ টেবিলে, যেখান থেকে মোছা যায়।
• Crypto-shredding: প্রতিটি ব্যক্তির তথ্য তাঁর নিজস্ব একটি key দিয়ে encrypt করে event-এ রাখা; মুছতে বললে শুধু key-টি ধ্বংস — event থাকে, কিন্তু পড়া যায় না। Key-এর ব্যবস্থাপনা নিরাপদ হতে হয় (পর্ব ৫২)। এটি আইনের চোখে “মুছে ফেলা” হিসেবে যথেষ্ট কি না, তা এখতিয়ার আর আইনি ব্যাখ্যাভেদে আলাদা — আইনজ্ঞের পরামর্শ ছাড়া ধরে নেবেন না।
• Backup, snapshot আর read model-ও: সব কপি থেকে মোছার পরিকল্পনা লাগে — শুধু event store নয়।

CQRS কখন সাহায্য করে, কখন অকারণ

৯ · Log, ledger আর double-entry — বড় ছবিতে event sourcing

Log-এর সাথে সম্পর্ক। পর্ব ৩৩-এর log-ভিত্তিক stream (partition, offset, consumer group) দেখতে event store-এর মতো, আর projector-কে চালাতে এটি চমৎকার। কিন্তু দুটি এক নয়। Event store-এর কাছে চাওয়া হয় একটি aggregate-এর stream আলাদা করে দ্রুত পড়া আর stream-প্রতি expected version — অথচ একটি partition-এ হাজারো aggregate মিশে থাকে, আর সাধারণ log-ভিত্তিক broker-এ stream-প্রতি expected version-এর যাচাই সাধারণত সরাসরি থাকে না (পণ্য ও সংস্করণভেদে আলাদা)। Retention-এর মেয়াদে পুরোনো event মুছে যাওয়াও আরেক ঝুঁকি — event store-এ event-ই সত্য, তাই retention “চিরকাল” হতে হয়। প্রচলিত নকশা: event store (প্রায়ই একটি SQL টেবিল) সত্য রাখে, আর সেখান থেকে outbox বা CDC দিয়ে event log-ভিত্তিক broker-এ যায়, যাতে অন্য সেবা আর projector শুনতে পারে। Event-এর রকমভেদ — শুধু খবর দেওয়া event বনাম পুরো state বহনকারী event বনাম event sourcing-এর event — পর্ব ৩২-এ, আর queue-এর ভিত্তি পর্ব ৩১-এ।

Double-entry-র সাথে সম্পর্ক। হিসাববিজ্ঞানের double-entry bookkeeping-এ প্রতিটি লেনদেন দুটি হিসাবে সমান ও বিপরীত entry হিসেবে লেখা হয় — A থেকে ৮০০ বের হলে B-তে ৮০০ ঢোকে, আর সব entry-র যোগফল সবসময় শূন্য। এটি event sourcing-এর একটি বিশেষ, কঠোর রূপ: entry কখনো মোছা হয় না, ভুল শোধরাতে বিপরীত entry (compensating event) লেখা হয়, আর balance হলো entry-গুলোর fold. পার্থক্য হলো double-entry একটি অতিরিক্ত অপরিবর্তনীয় শর্ত যোগ করে — যোগফল শূন্য — যা যেকোনো মুহূর্তে যাচাই করা যায়। টাকার সিস্টেমে event sourcing আর double-entry ledger প্রায়ই একসাথে চলে; পর্ব ৬০-এ ticket booking আর payment-এর case study-তে পুরো নকশা দেখবেন।

পরের পর্বগুলোর সাথে। Projector আসলে একটি stream processor — window, watermark আর বড় আকারের replay নিয়ে পর্ব ৩৬। আর দীর্ঘ কাজ (যেমন একটি transfer saga) নির্ভরযোগ্যভাবে চালানোর workflow engine-ও প্রায়ই ভেতরে event-এর ইতিহাস রেখে replay করে নিজের অবস্থা ফিরে পায় — পর্ব ৩৭।

১০ · AI-এর যুগে event sourcing আর CQRS-এর নকশা যাচাই

“একটি mobile wallet-এর architecture বানিয়ে দিন” লিখলে AI প্রায়ই চোখ ধাঁধানো একটি তালিকা দেয়: “Event sourcing, CQRS, Kafka, microservices, Redis”. শব্দগুলো ঠিক, কিন্তু নকশা নয়। কোন aggregate, কোন event, কোথায় নিয়ম, read model কতটা পিছিয়ে থাকতে পারে — এগুলো প্রায়ই অনুপস্থিত থাকে।

AI-এর নকশা হাতে পেলে — যাচাইয়ের চেকলিস্ট

• প্রতিটি event-এর নাম অতীত কালে আর ব্যবসার ভাষায়, XUpdated-ধাঁচের নয়
• নিয়ম যাচাই write model-এ, expected version-সহ; read model শুধু পড়ার জন্য
• Snapshot interval আর replay-র সময় নিজে হিসাব করে মিলিয়েছি
• Projector বিকল, rebuild আর duplicate event — তিনটিরই উত্তর আছে
• Event-এর schema বদলানোর আর ব্যক্তিগত তথ্য মোছার পরিকল্পনা আছে
• যে অংশে ইতিহাসের দরকার নেই, সেখানে সরল CRUD রাখা হয়েছে