+-----------------------------------------------------------------------------------------+
|                  Enterprise Event Sourcing & CQRS Topology                              |
+-----------------------------------------------------------------------------------------+

[COMMAND / WRITE SIDE (Strong Consistency)]
[User / API Client] ---> (Command: 'WithdrawMoney { amount: $50 }')
                                 |
                                 v
+-----------------------------------------------------------------+
| Command Handler & Domain Aggregate (Account Aggregate):         |
| 1. Replays events to verify invariant: balance >= $50.          |
| 2. Appends new event: 'MoneyWithdrawn { amount: $50, v: 4 }'    |
| 3. Optimistic Concurrency Check (Expected Version: 3).          |
+--------------------------------+--------------------------------+
                                 |
                                 v (Appends to Immutable Stream: /account-984)
[WRITE STORE: EventStoreDB / PostgreSQL Outbox]
                                 |
                                 v (Real-Time Change Data Capture / Event Stream)
[EVENT BROKER: Apache Kafka / Redpanda]
                                 |
        +------------------------+------------------------+
        | (Subscription 1)                                | (Subscription 2)
        v                                                 v
+---------------------------------+             +---------------------------------+
| PROJECTION WORKER (PostgreSQL)  |             | PROJECTION WORKER (Elasticsearch|
| - Updates SQL table:            |             | - Indexes transaction search:   |
|   'account_balances_view'       |             |   'transactions_audit_index'    |
+----------------+----------------+             +----------------+----------------+
                 |                                               |
                 v                                               v
[READ SIDE: Fast REST / GraphQL Queries]        [READ SIDE: Full-Text Audit Logs]

+-----------------------------------------------------------------------------------------+
|                  EventStoreDB vs Apache Kafka Responsibility Matrix                     |
+-----------------------------------------------------------------------------------------+

// domain/AccountAggregate.ts
export interface DomainEvent {
  type: string;
  data: any;
  metadata?: any;
}

export class BankAccountAggregate {
  public id: string;
  public balance: number = 0;
  public isClosed: boolean = false;
  public version: number = -1;

  constructor(id: string) {
    this.id = id;
  }

  // 1. Rehydrate state by replaying historical events
  public loadFromHistory(events: DomainEvent[]) {
    for (const event of events) {
      this.applyEvent(event);
      this.version++;
    }
  }

  // 2. Command: Withdraw Money (Enforces Business Invariant!)
  public withdrawMoney(amount: number): DomainEvent {
    if (this.isClosed) {
      throw new Error("Cannot withdraw from a closed account!");
    }
    if (this.balance < amount) {
      throw new Error(`Insufficient funds: Balance is `{this.balance}, requested `{amount}`);
    }

    const event: DomainEvent = {
      type: "MoneyWithdrawn",
      data: { accountId: this.id, amount, timestamp: Date.now() },
    };

    this.applyEvent(event);
    return event;
  }

  // 3. State Mutation Mutator
  private applyEvent(event: DomainEvent) {
    switch (event.type) {
      case "AccountOpened":
        this.balance = event.data.initialDeposit;
        break;
      case "MoneyDeposited":
        this.balance += event.data.amount;
        break;
      case "MoneyWithdrawn":
        this.balance -= event.data.amount;
        break;
      case "AccountClosed":
        this.isClosed = true;
        break;
    }
  }
}

import { EventStoreDBClient, jsonEvent, NO_STREAM } from "@eventstore/db-client";

const client = EventStoreDBClient.fromConnectionString("esdb://localhost:2113?tls=false");

export async function handleWithdrawCommand(accountId: string, amount: number) {
  const streamName = `account-${accountId}`;

  // 1. Read existing event stream
  const events = client.readStream(streamName);
  const aggregate = new BankAccountAggregate(accountId);
  const history = [];

  for await (const resolvedEvent of events) {
    history.push({ type: resolvedEvent.event!.type, data: resolvedEvent.event!.data });
  }

  aggregate.loadFromHistory(history);

  // 2. Execute Business Logic & Produce New Event
  const newEvent = aggregate.withdrawMoney(amount);

  // 3. Append to EventStoreDB with Optimistic Locking!
  // If another thread modified the account in the meantime, it throws WrongExpectedVersion!
  await client.appendToStream(streamName, [jsonEvent({ type: newEvent.type, data: newEvent.data })], {
    expectedRevision: aggregate.version === -1 ? NO_STREAM : BigInt(aggregate.version),
  });

  return { success: true, newBalance: aggregate.balance };
}

+-----------------------------------------------------------------------------------------+
|                  Aggregate Snapshotting Acceleration Model                              |
+-----------------------------------------------------------------------------------------+

[STREAM: 10,240 Events Total]
- Events 0 to 10,200: Pre-calculated into Snapshot: { balance: $94,200, isClosed: false }
- Events 10,201 to 10,240: (Only 40 events to replay!)
                                       |
                                       v
[Rehydration Time Drops from 450ms down to 1.8ms!]

+-----------------------------------------------------------------------------------------+
|                  Runtime Event Upcaster Architecture                                    |
+-----------------------------------------------------------------------------------------+

[Raw Event on Disk: UserRegistered.v1 { firstName: "Sachin", lastName: "Sharma" }]
                                         |
                                         v (Read by Upcaster Middleware)
[Upcaster Transforms: Appends 'fullName: "Sachin Sharma"', Removes deprecated fields]
                                         |
                                         v
[Aggregate receives modern UserRegistered.v2 Schema seamlessly in memory!]

// upcasters/userRegisteredUpcaster.ts
export function upcastUserRegistered(event: DomainEvent): DomainEvent {
  if (event.type === "UserRegistered.v1") {
    return {
      type: "UserRegistered.v2",
      data: {
        userId: event.data.userId,
        fullName: ``{event.data.firstName} `{event.data.lastName}`, // Transformed on the fly!
        email: event.data.email,
      },
    };
  }
  return event;
}