What are event sourcing and CQRS?

// Event sourcing + CQRS in miniature. The event store is an append-only array with
// per-stream versions and optimistic concurrency; the read side is a projection that
// catches up from a checkpoint. In-process only: no real event store or database.
const log = []; // global, ordered: {pos, stream, version, type, data}
function readStream(stream) { return log.filter((e) => e.stream === stream); }
function append(stream, expectedVersion, events) {
  const current = readStream(stream).length;
  if (current !== expectedVersion) throw new Error(`conflict: expected v${expectedVersion}, stream is at v${current}`);
  events.forEach((ev, i) => log.push({ pos: log.length + 1, stream, version: current + i + 1, ...ev }));
}

// Write side: state is a fold over the events.
function evolve(s, e) {
  if (e.type === "Opened") return { balance: 0, open: true };
  if (e.type === "Deposited") return { ...s, balance: s.balance + e.data.amount };
  if (e.type === "Withdrawn") return { ...s, balance: s.balance - e.data.amount };
  return s;
}
function load(stream, upTo = Infinity) {
  const evs = readStream(stream).filter((e) => e.version <= upTo);
  return { state: evs.reduce(evolve, null), version: evs.length, applied: evs.length };
}
function withdraw(stream, amount, loaded) {
  const { state, version } = loaded ?? load(stream);
  if (state.balance < amount) return `rejected: balance ${state.balance} < ${amount}`;
  append(stream, version, [{ type: "Withdrawn", data: { amount } }]);
  return `ok, balance now ${state.balance - amount}`;
}

// Read side: a projection with its own table and checkpoint.
const balances = new Map();
let checkpoint = 0;
function project() {
  for (const e of log.filter((x) => x.pos > checkpoint)) {
    if (e.type === "Opened") balances.set(e.stream, 0);
    if (e.type === "Deposited") balances.set(e.stream, balances.get(e.stream) + e.data.amount);
    if (e.type === "Withdrawn") balances.set(e.stream, balances.get(e.stream) - e.data.amount);
    checkpoint = e.pos;
  }
}

console.log("== 1. Commands append events; state is rebuilt by replaying them");
append("acct-1", 0, [{ type: "Opened", data: {} }, { type: "Deposited", data: { amount: 100 } }]);
console.log("  stream acct-1: " + readStream("acct-1").map((e) => `v${e.version} ${e.type}${e.data.amount ? " " + e.data.amount : ""}`).join(", "));
console.log(`  load -> balance ${load("acct-1").state.balance} at v${load("acct-1").version}`);

console.log("\n== 2. Two withdrawals of 70 race, both loaded at v2");
const a = load("acct-1"), b = load("acct-1");
console.log("  A: " + withdraw("acct-1", 70, a));
try { withdraw("acct-1", 70, b); } catch (e) { console.log("  B: " + e.message + " -> reload and retry"); }
console.log("  B retry: " + withdraw("acct-1", 70));
console.log(`  without the version check both would succeed: balance ${100 - 70 - 70}`);

console.log("\n== 3. CQRS read model is eventually consistent");
project();
append("acct-1", load("acct-1").version, [{ type: "Deposited", data: { amount: 500 } }]);
console.log(`  write side balance ${load("acct-1").state.balance}; read model says ${balances.get("acct-1")} (checkpoint ${checkpoint} of ${log.length})`);
project();
console.log(`  after the projector catches up: read model ${balances.get("acct-1")} (checkpoint ${checkpoint} of ${log.length})`);

console.log("\n== 4. Time travel and a projection added later");
console.log(`  balance as of v3: ${load("acct-1", 3).state.balance}`);
const bigDeposits = log.filter((e) => e.type === "Deposited" && e.data.amount >= 100).map((e) => `${e.stream} ${e.data.amount} @pos${e.pos}`);
console.log(`  new "large deposits" view rebuilt from history: [${bigDeposits.join(", ")}]`);

console.log("\n== 5. Snapshots bound replay cost");
append("acct-2", 0, [{ type: "Opened", data: {} }]);
for (let i = 1; i < 1000; i++) append("acct-2", i, [{ type: "Deposited", data: { amount: 1 } }]);
const full = load("acct-2");
const snap = { state: load("acct-2", 950).state, version: 950 }; // stored every 50 events
const tail = readStream("acct-2").filter((e) => e.version > snap.version);
const fromSnap = tail.reduce(evolve, snap.state);
console.log(`  full replay: ${full.applied} events -> balance ${full.state.balance}`);
console.log(`  snapshot at v${snap.version} + ${tail.length} events -> balance ${fromSnap.balance}`);
console.log(`  events in the store: ${log.length}; nothing was ever updated in place`);