Event-Driven Architecture Patterns

Deep Dive: Events in Financial Systems

This detailed version expands on the overview with implementation specifics, code examples, and production considerations.

Event Structure and Schema

Every event should contain:

interface DomainEvent {
  // Unique identifier for this event
  eventId: string;
  
  // Type discriminator for event routing
  eventType: string;
  
  // Aggregate/entity this event belongs to
  aggregateId: string;
  aggregateType: string;
  
  // Ordering within the aggregate
  sequenceNumber: number;
  
  // When it happened (business time)
  occurredAt: Date;
  
  // When it was recorded (system time)
  recordedAt: Date;
  
  // Who/what caused it
  causationId?: string;
  correlationId: string;
  
  // The actual event data
  payload: Record<string, unknown>;
  
  // Schema version for evolution
  version: number;
}

Event Naming Conventions

Events should be named as past-tense facts:

❌ Bad	✅ Good
CreateLoan	LoanCreated
ProcessPayment	PaymentProcessed
UpdateAddress	AddressUpdated
SendNotification	NotificationSent

The name should be domain-specific, not technical:

❌ Technical	✅ Domain
RecordInserted	LoanOriginated
StatusChanged	LoanDefaulted
MessagePublished	StatementGenerated

Event Sourcing Implementation

Here’s a minimal event-sourced aggregate:

abstract class EventSourcedAggregate {
  private uncommittedEvents: DomainEvent[] = [];
  private version: number = 0;

  protected apply(event: DomainEvent): void {
    // Apply the event to update state
    this.when(event);
    this.version++;
    this.uncommittedEvents.push(event);
  }

  // Override in subclass to handle specific events
  protected abstract when(event: DomainEvent): void;

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

  public loadFromHistory(events: DomainEvent[]): void {
    for (const event of events) {
      this.when(event);
      this.version++;
    }
  }
}

Example Loan aggregate:

class Loan extends EventSourcedAggregate {
  private status: LoanStatus = 'draft';
  private balance: number = 0;
  private borrowerId: string = '';

  static originate(
    loanId: string, 
    borrowerId: string, 
    amount: number
  ): Loan {
    const loan = new Loan();
    loan.apply({
      eventType: 'LoanOriginated',
      aggregateId: loanId,
      aggregateType: 'Loan',
      payload: { borrowerId, amount },
      // ... other fields
    });
    return loan;
  }

  makePayment(amount: number): void {
    if (this.status !== 'active') {
      throw new Error('Cannot pay inactive loan');
    }
    
    this.apply({
      eventType: 'PaymentMade',
      aggregateId: this.id,
      aggregateType: 'Loan',
      payload: { amount, previousBalance: this.balance },
      // ...
    });

    if (this.balance <= 0) {
      this.apply({
        eventType: 'LoanPaidOff',
        aggregateId: this.id,
        aggregateType: 'Loan',
        payload: {},
        // ...
      });
    }
  }

  protected when(event: DomainEvent): void {
    switch (event.eventType) {
      case 'LoanOriginated':
        this.borrowerId = event.payload.borrowerId;
        this.balance = event.payload.amount;
        this.status = 'active';
        break;
      case 'PaymentMade':
        this.balance -= event.payload.amount;
        break;
      case 'LoanPaidOff':
        this.status = 'paid_off';
        break;
      case 'LoanDefaulted':
        this.status = 'defaulted';
        break;
    }
  }
}

CQRS Projection Patterns

Projections transform events into queryable read models:

interface ProjectionHandler<TEvent, TReadModel> {
  eventTypes: string[];
  handle(event: TEvent, current: TReadModel | null): TReadModel;
}

// Example: Loan Dashboard Projection
const loanDashboardProjection: ProjectionHandler<DomainEvent, LoanDashboard> = {
  eventTypes: ['LoanOriginated', 'PaymentMade', 'LoanDefaulted', 'LoanPaidOff'],
  
  handle(event: DomainEvent, current: LoanDashboard | null): LoanDashboard {
    const dashboard = current || { 
      loanId: event.aggregateId,
      status: 'unknown',
      balance: 0,
      paymentCount: 0,
      lastPaymentDate: null,
    };

    switch (event.eventType) {
      case 'LoanOriginated':
        return {
          ...dashboard,
          status: 'active',
          balance: event.payload.amount,
          originatedAt: event.occurredAt,
        };
      
      case 'PaymentMade':
        return {
          ...dashboard,
          balance: dashboard.balance - event.payload.amount,
          paymentCount: dashboard.paymentCount + 1,
          lastPaymentDate: event.occurredAt,
        };
      
      case 'LoanPaidOff':
        return { ...dashboard, status: 'paid_off', balance: 0 };
      
      case 'LoanDefaulted':
        return { ...dashboard, status: 'defaulted' };
      
      default:
        return dashboard;
    }
  }
};

Eventual Consistency Strategies

1. Optimistic UI with Confirmation Polling

async function processPaymentWithOptimisticUI(
  loanId: string, 
  amount: number
): Promise<void> {
  // 1. Submit the command
  const { correlationId } = await submitPaymentCommand(loanId, amount);
  
  // 2. Immediately show optimistic UI
  updateUIOptimistically(loanId, -amount);
  
  // 3. Poll for confirmation (with backoff)
  const confirmed = await pollForProjectionUpdate(
    loanId,
    correlationId,
    { maxAttempts: 10, backoffMs: 500 }
  );
  
  // 4. If not confirmed, show warning or refresh
  if (!confirmed) {
    showStaleDataWarning();
    await refreshLoanData(loanId);
  }
}

2. Read-Your-Writes Pattern

class ReadYourWritesClient {
  private lastKnownVersion: Map<string, number> = new Map();

  async submitCommand(command: Command): Promise<void> {
    const result = await this.commandHandler.handle(command);
    // Store the version after our write
    this.lastKnownVersion.set(result.aggregateId, result.newVersion);
  }

  async query(aggregateId: string): Promise<ReadModel> {
    const minVersion = this.lastKnownVersion.get(aggregateId) || 0;
    
    // Wait for projection to catch up to our write
    return await this.queryService.queryWithMinVersion(
      aggregateId, 
      minVersion,
      { timeoutMs: 5000 }
    );
  }
}

Event Store Considerations

Key requirements for a financial event store:

1. Immutability: Events are append-only, never modified
2. Ordering: Strict ordering within an aggregate
3. Idempotency: Duplicate writes should be safe
4. Optimistic Concurrency: Check expected version on write

interface EventStore {
  // Append events with optimistic concurrency
  append(
    aggregateId: string,
    events: DomainEvent[],
    expectedVersion: number
  ): Promise<{ newVersion: number }>;

  // Read events for replay
  getEvents(
    aggregateId: string,
    fromVersion?: number
  ): AsyncIterable<DomainEvent>;

  // Subscribe to new events (for projections)
  subscribe(
    fromPosition: string,
    handler: (event: DomainEvent) => Promise<void>
  ): Subscription;
}

Production Checklist

Before going to production with event-driven architecture:

• Schema Registry: Track event schemas and versions
• Dead Letter Queue: Handle events that can’t be processed
• Idempotent Consumers: Handle duplicate event delivery
• Monitoring: Track projection lag, event throughput
• Replay Tools: Ability to rebuild projections from events
• Event Versioning Strategy: How to evolve schemas over time
• Backup & Recovery: Event store backup strategy
• GDPR Compliance: How to handle data deletion requests

Common Pitfalls

1. Too Many Events: Not every change needs an event. Start with domain events.

2. Anemic Events: Events with just IDs requiring additional lookups. Include relevant data.

3. Coupling Through Events: Services depending on internal event structure. Use published language.

4. Missing Correlation: Can’t trace a request through multiple events. Always include correlation IDs.

5. Projection Sprawl: Too many specialized projections. Start with general-purpose, optimize when needed.

Further Reading

• Martin Fowler: Event Sourcing
• Greg Young: CQRS Documents
• Udi Dahan: Clarified CQRS