CQRS & Event Sourcing for High-Performance Financial Applications

export interface Command {
  id: string;
  type: string;
  timestamp: Date;
}

export interface DepositFundsCommand extends Command {
  type: 'DepositFunds';
  accountId: string;
  amount: number;
  currency: string;
  transactionId: string;
}

export interface WithdrawFundsCommand extends Command {
  type: 'WithdrawFunds';
  accountId: string;
  amount: number;
  currency: string;
  transactionId: string;
}

export interface TransferFundsCommand extends Command {
  type: 'TransferFunds';
  fromAccountId: string;
  toAccountId: string;
  amount: number;
  currency: string;
  transactionId: string;
}

export interface Event {
  id: string;
  type: string;
  aggregateId: string;
  aggregateType: string;
  version: number;
  timestamp: Date;
  payload: any;
  metadata?: any;
}

export interface FundsDepositedEvent extends Event {
  type: 'FundsDeposited';
  payload: {
    accountId: string;
    amount: number;
    currency: string;
    newBalance: number;
    transactionId: string;
  };
}

export interface FundsWithdrawnEvent extends Event {
  type: 'FundsWithdrawn';
  payload: {
    accountId: string;
    amount: number;
    currency: string;
    newBalance: number;
    transactionId: string;
  };
}

export interface FundsTransferredEvent extends Event {
  type: 'FundsTransferred';
  payload: {
    fromAccountId: string;
    toAccountId: string;
    amount: number;
    currency: string;
    fromAccountNewBalance: number;
    toAccountNewBalance: number;
    transactionId: string;
  };
}

import { FundsDepositedEvent, FundsWithdrawnEvent, FundsTransferredEvent, Event } from '../events/account-events';
import { DepositFundsCommand, WithdrawFundsCommand, TransferFundsCommand } from '../commands/account-commands';

export class Account {
  public id: string;
  public balance: number;
  public currency: string;
  public version: number;
  private uncommittedEvents: Event[] = [];

  constructor(id: string, currency: string) {
    this.id = id;
    this.balance = 0;
    this.currency = currency;
    this.version = 0;
  }

  public static fromEvents(id: string, currency: string, history: Event[]): Account {
    const account = new Account(id, currency);
    history.forEach(event => account.apply(event, false));
    account.uncommittedEvents = []; // Clear uncommitted after loading history
    return account;
  }

  private apply(event: Event, isNew: boolean = true): void {
    switch (event.type) {
      case 'FundsDeposited':
        this.applyFundsDeposited(event as FundsDepositedEvent);
        break;
      case 'FundsWithdrawn':
        this.applyFundsWithdrawn(event as FundsWithdrawnEvent);
        break;
      case 'FundsTransferred':
        // Special handling for transfer; could be two separate events
        // For simplicity, we'll assume a single aggregate handles internal transfer
        // Or, more typically, two aggregates (from, to) with distributed transaction/saga
        break;
      default:
        throw new Error(`Unknown event type: ${event.type}`);
    }
    this.version = event.version; // Update aggregate version
    if (isNew) {
      this.uncommittedEvents.push(event);
    }
  }

  // Command Handlers
  public depositFunds(command: DepositFundsCommand): void {
    if (command.amount <= 0) {
      throw new Error('Deposit amount must be positive');
    }
    if (command.currency !== this.currency) {
      throw new Error(`Cannot deposit ${command.currency} into a ${this.currency} account.`);
    }

    const newBalance = this.balance + command.amount;
    const event: FundsDepositedEvent = {
      id: `evt-${Date.now()}`,
      type: 'FundsDeposited',
      aggregateId: this.id,
      aggregateType: 'Account',
      version: this.version + 1, // Increment version for new event
      timestamp: new Date(),
      payload: {
        accountId: this.id,
        amount: command.amount,
        currency: this.currency,
        newBalance: newBalance,
        transactionId: command.transactionId,
      },
    };
    this.apply(event);
  }

  public withdrawFunds(command: WithdrawFundsCommand): void {
    if (command.amount <= 0) {
      throw new Error('Withdrawal amount must be positive');
    }
    if (this.balance < command.amount) {
      throw new Error('Insufficient funds');
    }
    if (command.currency !== this.currency) {
        throw new Error(`Cannot withdraw ${command.currency} from a ${this.currency} account.`);
    }

    const newBalance = this.balance - command.amount;
    const event: FundsWithdrawnEvent = {
      id: `evt-${Date.now()}`,
      type: 'FundsWithdrawn',
      aggregateId: this.id,
      aggregateType: 'Account',
      version: this.version + 1,
      timestamp: new Date(),
      payload: {
        accountId: this.id,
        amount: command.amount,
        currency: this.currency,
        newBalance: newBalance,
        transactionId: command.transactionId,
      },
    };
    this.apply(event);
  }

  // Event Appliers
  private applyFundsDeposited(event: FundsDepositedEvent): void {
    this.balance = event.payload.newBalance;
  }

  private applyFundsWithdrawn(event: FundsWithdrawnEvent): void {
    this.balance = event.payload.newBalance;
  }

  public getUncommittedEvents(): Event[] {
    return this.uncommittedEvents;
  }
}

import { Event } from '../events/account-events';

export interface EventStore {
  save(aggregateId: string, expectedVersion: number, events: Event[]): Promise<void>;
  load(aggregateId: string): Promise<Event[]>;
}

// Example of a simple in-memory event store (for demonstration)
export class InMemoryEventStore implements EventStore {
  private events: { [aggregateId: string]: Event[] } = {};

  public async save(aggregateId: string, expectedVersion: number, newEvents: Event[]): Promise<void> {
    const currentEvents = this.events[aggregateId] || [];
    if (currentEvents.length !== expectedVersion) {
      throw new Error('Concurrency conflict: Expected version mismatch.');
    }
    this.events[aggregateId] = [...currentEvents, ...newEvents];
    console.log(`Saved ${newEvents.length} events for aggregate ${aggregateId}. New version: ${currentEvents.length + newEvents.length}`);
  }

  public async load(aggregateId: string): Promise<Event[]> {
    return this.events[aggregateId] || [];
  }
}

import { DepositFundsCommand } from '../commands/account-commands';
import { Account } from '../domain/account-aggregate';
import { EventStore } from '../infrastructure/event-store';

export class DepositFundsCommandHandler {
  private eventStore: EventStore;

  constructor(eventStore: EventStore) {
    this.eventStore = eventStore;
  }

  public async handle(command: DepositFundsCommand): Promise<void> {
    // 1. Load aggregate history
    const history = await this.eventStore.load(command.accountId);

    // 2. Reconstruct aggregate state
    // For a new account, we might need an initial event or default constructor
    // Assuming account already exists for deposit
    if (history.length === 0) {
        // In a real system, you might first create an account, then deposit
        throw new Error(`Account ${command.accountId} not found.`);
    }
    const initialAccount = new Account(command.accountId, history[0].payload.currency || 'USD'); // Assuming currency from first event
    const account = Account.fromEvents(command.accountId, initialAccount.currency, history);

    // 3. Execute command on aggregate
    account.depositFunds(command);

    // 4. Save uncommitted events to event store
    const newEvents = account.getUncommittedEvents();
    await this.eventStore.save(account.id, account.version - newEvents.length, newEvents); // Pass expected version

    console.log(`Deposit successful for account ${command.accountId}. New events:`, newEvents.map(e => e.type));
  }
}

import { FundsDepositedEvent, FundsWithdrawnEvent, FundsTransferredEvent, Event } from '../events/account-events';

export interface AccountBalanceReadModel {
  accountId: string;
  currentBalance: number;
  currency: string;
  lastUpdated: Date;
}

// A projector subscribes to events and updates a read model database
export class AccountBalanceProjector {
  private readModelStore: Map<string, AccountBalanceReadModel>; // In-memory for simplicity, usually a DB

  constructor(readModelStore: Map<string, AccountBalanceReadModel>) {
    this.readModelStore = readModelStore;
  }

  public async handleEvent(event: Event): Promise<void> {
    switch (event.type) {
      case 'FundsDeposited':
        await this.handleFundsDeposited(event as FundsDepositedEvent);
        break;
      case 'FundsWithdrawn':
        await this.handleFundsWithdrawn(event as FundsWithdrawnEvent);
        break;
      case 'FundsTransferred':
          // For transfers, you'd update two accounts
          await this.handleFundsTransferred(event as FundsTransferredEvent);
          break;
      // Add handlers for other events that affect account balance
    }
  }

  private async handleFundsDeposited(event: FundsDepositedEvent): Promise<void> {
    const { accountId, newBalance, currency } = event.payload;
    this.readModelStore.set(accountId, {
      accountId,
      currentBalance: newBalance,
      currency,
      lastUpdated: event.timestamp,
    });
    console.log(`Read model updated for account ${accountId}: New balance ${newBalance}`);
  }

  private async handleFundsWithdrawn(event: FundsWithdrawnEvent): Promise<void> {
    const { accountId, newBalance, currency } = event.payload;
    this.readModelStore.set(accountId, {
      accountId,
      currentBalance: newBalance,
      currency,
      lastUpdated: event.timestamp,
    });
    console.log(`Read model updated for account ${accountId}: New balance ${newBalance}`);
  }

  private async handleFundsTransferred(event: FundsTransferredEvent): Promise<void> {
    const { fromAccountId, toAccountId, fromAccountNewBalance, toAccountNewBalance, currency } = event.payload;
    this.readModelStore.set(fromAccountId, {
      accountId: fromAccountId,
      currentBalance: fromAccountNewBalance,
      currency,
      lastUpdated: event.timestamp,
    });
    this.readModelStore.set(toAccountId, {
      accountId: toAccountId,
      currentBalance: toAccountNewBalance,
      currency,
      lastUpdated: event.timestamp,
    });
    console.log(`Read model updated for transfer: ${fromAccountId} -> ${toAccountId}`);
  }

  public getAccountBalance(accountId: string): AccountBalanceReadModel | undefined {
    return this.readModelStore.get(accountId);
  }
}