Event Sourcing in Financial Systems: Patterns and Practices

1// Traditional approach: Store current state
2interface Account {
3  id: string;
4  balance: number;
5  updatedAt: Date;
6}
7
8// Event sourcing: Store events
9type AccountEvent = 
10  | { type: 'AccountOpened'; accountId: string; initialDeposit: number }
11  | { type: 'MoneyDeposited'; accountId: string; amount: number; timestamp: Date }
12  | { type: 'MoneyWithdrawn'; accountId: string; amount: number; timestamp: Date };
13

1function computeBalance(events: AccountEvent[]): number {
2  return events.reduce((balance, event) => {
3    switch (event.type) {
4      case 'AccountOpened':
5        return event.initialDeposit;
6      case 'MoneyDeposited':
7        return balance + event.amount;
8      case 'MoneyWithdrawn':
9        return balance - event.amount;
10    }
11  }, 0);
12}
13

1SELECT * FROM events 
2WHERE aggregate_id = 'account-123'
3ORDER BY timestamp;
4
5-- Results:
6-- 2024-01-01 10:00:00 | AccountOpened    | { initialDeposit: 10000 }
7-- 2024-01-15 14:30:00 | MoneyDeposited   | { amount: 5000 }
8-- 2024-02-01 09:15:00 | MoneyWithdrawn   | { amount: 2000 }
9-- 2024-02-10 16:45:00 | MoneyDeposited   | { amount: 3000 }
10

1function balanceAt(events: AccountEvent[], date: Date): number {
2  const relevantEvents = events.filter(e => e.timestamp <= date);
3  return computeBalance(relevantEvents);
4}
5
6// What was the balance on January 31st?
7const balance = balanceAt(events, new Date('2024-01-31'));
8// Answer: 15000 (10000 + 5000)
9

1// Replay production events in test environment
2const productionEvents = fetchEventsFromProduction('account-123');
3const testAccount = new Account();
4productionEvents.forEach(event => testAccount.apply(event));
5
6// Now debug with exact production state
7

1class EventStore {
2  async append(streamId: string, event: Event): Promise<void> {
3    await db.query(
4      `INSERT INTO events (stream_id, event_type, data, timestamp)
5       VALUES ($1, $2, $3, NOW())`,
6      [streamId, event.type, JSON.stringify(event)]
7    );
8  }
9  
10  async getStream(streamId: string): Promise<Event[]> {
11    const rows = await db.query(
12      `SELECT * FROM events 
13       WHERE stream_id = $1 
14       ORDER BY timestamp`,
15      [streamId]
16    );
17    return rows.map(r => JSON.parse(r.data));
18  }
19}
20

1class SnapshotStore {
2  async saveSnapshot(
3    streamId: string, 
4    state: any, 
5    version: number
6  ): Promise<void> {
7    await db.query(
8      `INSERT INTO snapshots (stream_id, state, version)
9       VALUES ($1, $2, $3)
10       ON CONFLICT (stream_id) DO UPDATE SET state = $2, version = $3`,
11      [streamId, JSON.stringify(state), version]
12    );
13  }
14  
15  async loadState(streamId: string): Promise<any> {
16    // Load latest snapshot
17    const snapshot = await db.query(
18      `SELECT * FROM snapshots WHERE stream_id = $1`,
19      [streamId]
20    );
21    
22    // Replay events since snapshot
23    const events = await eventStore.getStream(
24      streamId, 
25      snapshot.version
26    );
27    
28    return applyEvents(snapshot.state, events);
29  }
30}
31

1// Write side: Append events
2class AccountCommandHandler {
3  async deposit(accountId: string, amount: number): Promise<void> {
4    const event = {
5      type: 'MoneyDeposited',
6      accountId,
7      amount,
8      timestamp: new Date()
9    };
10    await eventStore.append(accountId, event);
11  }
12}
13
14// Read side: Materialized view
15class AccountProjection {
16  async updateFromEvent(event: AccountEvent): Promise<void> {
17    switch (event.type) {
18      case 'MoneyDeposited':
19        await db.query(
20          `UPDATE accounts 
21           SET balance = balance + $1 
22           WHERE id = $2`,
23          [event.amount, event.accountId]
24        );
25        break;
26    }
27  }
28}
29

1// Domain events
2type TransactionEvent =
3  | { type: 'TransactionInitiated'; id: string; from: string; to: string; amount: number }
4  | { type: 'FundsReserved'; id: string; from: string; amount: number }
5  | { type: 'FundsTransferred'; id: string; to: string; amount: number }
6  | { type: 'TransactionCompleted'; id: string; completedAt: Date }
7  | { type: 'TransactionFailed'; id: string; reason: string };
8
9// Aggregate root
10class Transaction {
11  private events: TransactionEvent[] = [];
12  
13  initiate(from: string, to: string, amount: number): void {
14    this.apply({
15      type: 'TransactionInitiated',
16      id: uuid(),
17      from,
18      to,
19      amount
20    });
21  }
22  
23  private apply(event: TransactionEvent): void {
24    this.events.push(event);
25    // Update internal state based on event
26  }
27  
28  getUncommittedEvents(): TransactionEvent[] {
29    return this.events;
30  }
31}
32

1class EventUpcaster {
2  upcast(event: any): Event {
3    // V1: { type: 'MoneyDeposited', amount: number }
4    // V2: { type: 'MoneyDeposited', amount: number, currency: string }
5    
6    if (event.version === 1) {
7      return {
8        ...event,
9        currency: 'USD', // Add default
10        version: 2
11      };
12    }
13    return event;
14  }
15}
16

1const SNAPSHOT_FREQUENCY = 100;
2
3if (events.length % SNAPSHOT_FREQUENCY === 0) {
4  await snapshotStore.save(streamId, currentState, events.length);
5}
6

1class ReadModel {
2  private lastProcessedVersion = 0;
3  
4  async update(event: Event): Promise<void> {
5    if (event.version <= this.lastProcessedVersion) {
6      return; // Already processed
7    }
8    
9    // Apply update
10    await this.applyEvent(event);
11    this.lastProcessedVersion = event.version;
12  }
13}
14