1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

/**
 * The left-fold pattern for event application
 */
 
// Generic event application signature
type EventApplier<TState, TEvent> = (state: TState, event: TEvent) => TState;
 
// Generic fold implementation
function foldEvents<TState, TEvent>(
  events: TEvent[],
  initialState: TState,
  apply: EventApplier<TState, TEvent>
): TState {
  return events.reduce(apply, initialState);
}
 
// Concrete example: Shopping Cart
interface CartState {
  cartId: string;
  customerId: string | null;
  items: Map<string, CartItem>;
  status: 'active' | 'checked_out' | 'abandoned';
  createdAt: Date | null;
  lastModified: Date | null;
}
 
interface CartItem {
  productId: string;
  productName: string;
  quantity: number;
  unitPrice: number;
}
 
// Initial state factory
const createEmptyCartState = (cartId: string): CartState => ({
  cartId,
  customerId: null,
  items: new Map(),
  status: 'active',
  createdAt: null,
  lastModified: null,
});
 
// Event application - pure function, no side effects
function applyCartEvent(state: CartState, event: CartEvent): CartState {
  switch (event.type) {
    case 'CartCreated':
      return {
        ...state,
        customerId: event.data.customerId,
        createdAt: event.timestamp,
        lastModified: event.timestamp,
      };
 
    case 'ItemAdded': {
      const newItems = new Map(state.items);
      const existing = newItems.get(event.data.productId);
      
      newItems.set(event.data.productId, {
        productId: event.data.productId,
        productName: event.data.productName,
        quantity: (existing?.quantity ?? 0) + event.data.quantity,
        unitPrice: event.data.unitPrice,
      });
 
      return {
        ...state,
        items: newItems,
        lastModified: event.timestamp,
      };
    }
 
    case 'ItemRemoved': {
      const newItems = new Map(state.items);
      const existing = newItems.get(event.data.productId);
      
      if (existing) {
        const newQuantity = existing.quantity - event.data.quantity;
        if (newQuantity <= 0) {
          newItems.delete(event.data.productId);
        } else {
          newItems.set(event.data.productId, {
            ...existing,
            quantity: newQuantity,
          });
        }
      }
 
      return {
        ...state,
        items: newItems,
        lastModified: event.timestamp,
      };
    }
 
    case 'ItemQuantityChanged': {
      const newItems = new Map(state.items);
      const existing = newItems.get(event.data.productId);
      
      if (existing) {
        newItems.set(event.data.productId, {
          ...existing,
          quantity: event.data.newQuantity,
        });
      }
 
      return {
        ...state,
        items: newItems,
        lastModified: event.timestamp,
      };
    }
 
    case 'CartCheckedOut':
      return {
        ...state,
        status: 'checked_out',
        lastModified: event.timestamp,
      };
 
    case 'CartAbandoned':
      return {
        ...state,
        status: 'abandoned',
        lastModified: event.timestamp,
      };
 
    default:
      // Unknown event type - log and continue
      console.warn(`Unknown cart event type: ${(event as any).type}`);
      return state;
  }
}
 
// Reconstituting a cart from events
function reconstitute(cartId: string, events: CartEvent[]): CartState {
  const initialState = createEmptyCartState(cartId);
  return foldEvents(events, initialState, applyCartEvent);
}
 
// Usage
const events: CartEvent[] = [
  { type: 'CartCreated', timestamp: new Date('2024-03-01'), data: { customerId: 'cust-1' } },
  { type: 'ItemAdded', timestamp: new Date('2024-03-01'), data: { productId: 'prod-1', productName: 'Widget', quantity: 2, unitPrice: 29.99 } },
  { type: 'ItemAdded', timestamp: new Date('2024-03-02'), data: { productId: 'prod-2', productName: 'Gadget', quantity: 1, unitPrice: 49.99 } },
  { type: 'ItemRemoved', timestamp: new Date('2024-03-03'), data: { productId: 'prod-1', quantity: 1 } },
];
 
const currentState = reconstitute('cart-123', events);
// Result: Cart with 1 Widget, 1 Gadget

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221

/**
 * Base class for event-sourced aggregates
 */
abstract class Aggregate<TEvent extends DomainEvent> {
  private uncommittedEvents: TEvent[] = [];
  private _version = -1;  // -1 = doesn't exist yet
  
  abstract readonly aggregateType: string;
  
  get id(): string {
    return this._id;
  }
  
  get version(): number {
    return this._version;
  }
 
  constructor(private readonly _id: string) {}
 
  /**
   * Apply an event to update internal state.
   * Called both during reconstitution and for new events.
   */
  protected abstract apply(event: TEvent): void;
 
  /**
   * Record a new event (typically called by command handlers).
   * Applies the event immediately and queues for persistence.
   */
  protected recordEvent(event: TEvent): void {
    // Apply to self immediately
    this.apply(event);
    
    // Queue for persistence
    this.uncommittedEvents.push(event);
  }
 
  /**
   * Get events that haven't been persisted yet.
   */
  getUncommittedEvents(): TEvent[] {
    return [...this.uncommittedEvents];
  }
 
  /**
   * Mark all uncommitted events as persisted.
   */
  clearUncommittedEvents(): void {
    this.uncommittedEvents = [];
  }
 
  /**
   * Reconstitute aggregate from stored events.
   */
  loadFromHistory(events: TEvent[]): void {
    for (const event of events) {
      this.apply(event);
      this._version++;
    }
  }
 
  /**
   * Set version explicitly (used by repository).
   */
  setVersion(version: number): void {
    this._version = version;
  }
}
 
/**
 * Concrete aggregate example: Order
 */
class OrderAggregate extends Aggregate<OrderEvent> {
  readonly aggregateType = 'Order';
  
  private customerId: string | null = null;
  private items: OrderItem[] = [];
  private status: OrderStatus = 'draft';
  private shippingAddress: Address | null = null;
  private total = 0;
 
  // Command: Place the order
  placeOrder(customerId: string, items: OrderItemData[]): void {
    // Invariant checks
    if (this.status !== 'draft') {
      throw new InvalidOperationError(
        `Cannot place order in ${this.status} status`
      );
    }
    if (items.length === 0) {
      throw new InvalidOperationError('Order must have at least one item');
    }
 
    // Calculate total
    const total = items.reduce(
      (sum, item) => sum + item.quantity * item.unitPrice,
      0
    );
 
    // Record the event
    this.recordEvent({
      eventType: 'OrderPlaced',
      aggregateId: this.id,
      timestamp: new Date(),
      payload: { customerId, items, total },
    });
  }
 
  // Command: Set shipping address
  setShippingAddress(address: Address): void {
    if (this.status === 'shipped' || this.status === 'delivered') {
      throw new InvalidOperationError(
        'Cannot change address for shipped orders'
      );
    }
 
    this.recordEvent({
      eventType: 'ShippingAddressSet',
      aggregateId: this.id,
      timestamp: new Date(),
      payload: { address },
    });
  }
 
  // Command: Ship the order
  ship(trackingNumber: string, carrier: string): void {
    if (this.status !== 'confirmed') {
      throw new InvalidOperationError(
        `Cannot ship order in ${this.status} status`
      );
    }
    if (!this.shippingAddress) {
      throw new InvalidOperationError('Shipping address not set');
    }
 
    this.recordEvent({
      eventType: 'OrderShipped',
      aggregateId: this.id,
      timestamp: new Date(),
      payload: { trackingNumber, carrier },
    });
  }
 
  // Event application - updates internal state
  protected apply(event: OrderEvent): void {
    switch (event.eventType) {
      case 'OrderPlaced':
        this.customerId = event.payload.customerId;
        this.items = event.payload.items.map(i => ({
          productId: i.productId,
          quantity: i.quantity,
          unitPrice: i.unitPrice,
        }));
        this.total = event.payload.total;
        this.status = 'pending_payment';
        break;
 
      case 'PaymentReceived':
        this.status = 'confirmed';
        break;
 
      case 'ShippingAddressSet':
        this.shippingAddress = event.payload.address;
        break;
 
      case 'OrderShipped':
        this.status = 'shipped';
        break;
 
      case 'OrderDelivered':
        this.status = 'delivered';
        break;
 
      case 'OrderCancelled':
        this.status = 'cancelled';
        break;
    }
  }
 
  // Getters for queries (if needed within aggregate)
  getStatus(): OrderStatus {
    return this.status;
  }
 
  getTotal(): number {
    return this.total;
  }
}
 
/**
 * Repository pattern for loading/saving aggregates
 */
class OrderRepository {
  constructor(private eventStore: EventStore) {}
 
  async load(orderId: string): Promise<OrderAggregate | null> {
    const stream = `order-${orderId}`;
    const result = await this.eventStore.readStream(stream);
    
    if (result.events.length === 0) {
      return null;
    }
 
    const order = new OrderAggregate(orderId);
    order.loadFromHistory(result.events as OrderEvent[]);
    order.setVersion(result.version);
    return order;
  }
 
  async save(order: OrderAggregate): Promise<void> {
    const uncommitted = order.getUncommittedEvents();
    
    if (uncommitted.length === 0) {
      return;
    }
 
    const stream = `order-${order.id}`;
    await this.eventStore.append(stream, order.version, uncommitted);
    order.clearUncommittedEvents();
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185

/**
 * Projection framework for building read models
 */
interface Projection<TReadModel> {
  readonly projectionName: string;
  
  // Handle a single event
  handle(event: StoredEvent): Promise<void>;
  
  // Build or rebuild from scratch
  rebuild(): Promise<void>;
  
  // Get current projection state
  getState(): Promise<TReadModel>;
}
 
/**
 * Example: Customer Orders Summary Projection
 * 
 * Maintains a denormalized view of order counts and totals per customer
 */
interface CustomerOrderSummary {
  customerId: string;
  totalOrders: number;
  totalSpent: number;
  lastOrderDate: Date | null;
  orderStatuses: Record<OrderStatus, number>;
}
 
class CustomerOrdersProjection implements Projection<Map<string, CustomerOrderSummary>> {
  readonly projectionName = 'customer-orders-summary';
  
  private state = new Map<string, CustomerOrderSummary>();
 
  constructor(
    private eventStore: EventStore,
    private checkpointStore: CheckpointStore,
    private readModelStore: ReadModelStore
  ) {}
 
  async handle(event: StoredEvent): Promise<void> {
    // Only care about order events
    if (!event.streamId.startsWith('order-')) {
      return;
    }
 
    switch (event.eventType) {
      case 'OrderPlaced': {
        const { customerId, total } = event.data as OrderPlacedData;
        const current = this.getCustomerSummary(customerId);
        
        await this.readModelStore.upsert('customer_orders', customerId, {
          ...current,
          totalOrders: current.totalOrders + 1,
          totalSpent: current.totalSpent + total,
          lastOrderDate: event.timestamp,
          orderStatuses: {
            ...current.orderStatuses,
            pending_payment: (current.orderStatuses.pending_payment ?? 0) + 1,
          },
        });
        break;
      }
 
      case 'PaymentReceived': {
        const { customerId } = await this.getOrderData(event.streamId);
        const current = this.getCustomerSummary(customerId);
        
        await this.readModelStore.upsert('customer_orders', customerId, {
          ...current,
          orderStatuses: {
            ...current.orderStatuses,
            pending_payment: Math.max(0, (current.orderStatuses.pending_payment ?? 0) - 1),
            confirmed: (current.orderStatuses.confirmed ?? 0) + 1,
          },
        });
        break;
      }
 
      case 'OrderCancelled': {
        // Similar status transition logic
        break;
      }
    }
 
    // Update checkpoint
    await this.checkpointStore.saveCheckpoint({
      subscriptionId: this.projectionName,
      position: event.globalPosition + 1,
      updatedAt: new Date(),
    });
  }
 
  async rebuild(): Promise<void> {
    console.log(`Rebuilding projection: ${this.projectionName}`);
    
    // Clear existing state
    await this.readModelStore.clear('customer_orders');
    await this.checkpointStore.saveCheckpoint({
      subscriptionId: this.projectionName,
      position: 0,
      updatedAt: new Date(),
    });
    
    // Replay all events
    let position = 0;
    const batchSize = 1000;
    
    while (true) {
      const result = await this.eventStore.readAll({
        fromPosition: position,
        count: batchSize,
      });
      
      if (result.events.length === 0) {
        break;
      }
      
      for (const event of result.events) {
        await this.handle(event);
      }
      
      position = result.events[result.events.length - 1].globalPosition + 1;
      console.log(`Projection rebuild progress: position ${position}`);
    }
    
    console.log(`Projection rebuild complete: ${this.projectionName}`);
  }
 
  async getState(): Promise<Map<string, CustomerOrderSummary>> {
    const all = await this.readModelStore.getAll<CustomerOrderSummary>('customer_orders');
    return new Map(all.map(item => [item.customerId, item]));
  }
 
  private getCustomerSummary(customerId: string): CustomerOrderSummary {
    return this.state.get(customerId) ?? {
      customerId,
      totalOrders: 0,
      totalSpent: 0,
      lastOrderDate: null,
      orderStatuses: {},
    };
  }
 
  private async getOrderData(streamId: string): Promise<{ customerId: string }> {
    // Read from stream or cached order data
    const result = await this.eventStore.readStream(streamId, { count: 1 });
    const orderPlaced = result.events.find(e => e.eventType === 'OrderPlaced');
    return { customerId: orderPlaced?.data.customerId };
  }
}
 
/**
 * Query handler that uses the projection
 */
class CustomerQueryHandler {
  constructor(private readModelStore: ReadModelStore) {}
 
  async getCustomerDashboard(customerId: string): Promise<CustomerDashboard> {
    const summary = await this.readModelStore.get<CustomerOrderSummary>(
      'customer_orders',
      customerId
    );
 
    if (!summary) {
      return {
        customerId,
        hasOrders: false,
        stats: null,
      };
    }
 
    return {
      customerId,
      hasOrders: true,
      stats: {
        orderCount: summary.totalOrders,
        totalSpent: summary.totalSpent,
        lastOrderDate: summary.lastOrderDate,
        pendingOrders: summary.orderStatuses.pending_payment ?? 0,
        completedOrders: summary.orderStatuses.delivered ?? 0,
      },
    };
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

/**
 * Synchronous projection update - coupled with write transaction
 */
class SynchronousProjectionUpdater {
  constructor(
    private eventStore: TransactionalEventStore,
    private projections: Projection[]
  ) {}
 
  async appendWithProjections(
    streamId: string,
    expectedVersion: number,
    events: NewEvent[]
  ): Promise<void> {
    await this.eventStore.withTransaction(async (tx) => {
      // 1. Append events (will fail if version mismatch)
      const storedEvents = await tx.append(streamId, expectedVersion, events);
      
      // 2. Update all projections (in same transaction)
      for (const projection of this.projections) {
        for (const event of storedEvents) {
          await projection.handle(event);
        }
      }
      
      // Transaction commits only if all succeed
    });
  }
}
 
/**
 * Asynchronous projection update - subscription-based
 */
class AsyncProjectionManager {
  private subscriptions = new Map<string, Subscription>();
 
  constructor(
    private eventStore: EventStore,
    private checkpointStore: CheckpointStore
  ) {}
 
  registerProjection(projection: Projection<unknown>): void {
    // Create catch-up subscription for this projection
    this.subscriptions.set(
      projection.projectionName,
      new ProjectionSubscription(
        this.eventStore,
        this.checkpointStore,
        projection
      )
    );
  }
 
  async startAll(): Promise<void> {
    for (const [name, subscription] of this.subscriptions) {
      console.log(`Starting projection: ${name}`);
      await subscription.start();
    }
  }
 
  async stopAll(): Promise<void> {
    for (const [name, subscription] of this.subscriptions) {
      await subscription.stop();
    }
  }
 
  async rebuildProjection(projectionName: string): Promise<void> {
    const subscription = this.subscriptions.get(projectionName);
    if (!subscription) {
      throw new Error(`Unknown projection: ${projectionName}`);
    }
    
    // Stop, rebuild, restart
    await subscription.stop();
    await subscription.projection.rebuild();
    await subscription.start();
  }
}
 
/**
 * Handling eventual consistency in the application layer
 */
class OrderCommandHandler {
  constructor(
    private orderRepo: OrderRepository,
    private readModelStore: ReadModelStore
  ) {}
 
  async placeOrder(command: PlaceOrderCommand): Promise<PlaceOrderResult> {
    // Execute command against aggregate
    const order = new OrderAggregate(command.orderId);
    order.placeOrder(command.customerId, command.items);
    await this.orderRepo.save(order);
 
    // Return command result - don't wait for projections
    return {
      orderId: command.orderId,
      status: 'placed',
      // Indicate to client that full details may take a moment
      message: 'Order placed successfully. Details will be available shortly.',
    };
  }
 
  async getOrderDetails(orderId: string): Promise<OrderDetails | null> {
    // First, try the read model (fast, but maybe stale)
    const cached = await this.readModelStore.get<OrderDetails>('orders', orderId);
    
    if (cached) {
      return cached;
    }
 
    // If not in read model yet, fall back to aggregate reconstitution
    // This handles the case where projections haven't caught up
    const order = await this.orderRepo.load(orderId);
    
    if (!order) {
      return null;
    }
 
    // Build response from aggregate (slower but always current)
    return order.toOrderDetails();
  }
}
 
/**
 * Consider-Then-React pattern for UI handling eventual consistency
 */
interface OrderConfirmationPage {
  // UI shows: "Your order #123 is being processed"
  // Poll or websocket to update when projection catches up
  orderId: string;
  status: 'processing' | 'confirmed';
  details?: OrderDetails;
  lastChecked: Date;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158

/**
 * Parallel rebuild strategy for zero-downtime projection migrations
 */
class ParallelRebuildManager {
  constructor(
    private eventStore: EventStore,
    private readModelStore: ReadModelStore
  ) {}
 
  async parallelRebuild(projection: Projection<unknown>): Promise<void> {
    const currentTable = `${projection.projectionName}_current`;
    const newTable = `${projection.projectionName}_rebuild`;
    const tempTable = `${projection.projectionName}_old`;
 
    console.log(`Starting parallel rebuild of ${projection.projectionName}`);
 
    // 1. Create new table with updated schema
    await this.readModelStore.createTable(newTable, projection.schema);
 
    // 2. Replay all events into new table
    const rebuildProjection = projection.withTable(newTable);
    let position = 0;
    let eventCount = 0;
 
    const startTime = Date.now();
    
    while (true) {
      const result = await this.eventStore.readAll({
        fromPosition: position,
        count: 1000,
      });
 
      if (result.events.length === 0) {
        break;
      }
 
      for (const event of result.events) {
        await rebuildProjection.handle(event);
        eventCount++;
      }
 
      position = result.events[result.events.length - 1].globalPosition + 1;
      
      // Progress logging
      if (eventCount % 10000 === 0) {
        const elapsed = (Date.now() - startTime) / 1000;
        console.log(
          `Rebuild progress: ${eventCount} events, ${Math.round(eventCount / elapsed)} events/sec`
        );
      }
    }
 
    // 3. Catch up any events written during rebuild
    const catchUpStart = position;
    console.log(`Catching up from position ${catchUpStart}`);
    
    // Brief pause to let any in-flight writes complete
    await sleep(1000);
    
    const catchUpResult = await this.eventStore.readAll({
      fromPosition: position,
      count: 10000,  // Catch-up should be small
    });
    
    for (const event of catchUpResult.events) {
      await rebuildProjection.handle(event);
    }
 
    // 4. Atomic table swap
    console.log('Performing atomic table swap');
    await this.readModelStore.withTransaction(async (tx) => {
      await tx.renameTable(currentTable, tempTable);
      await tx.renameTable(newTable, currentTable);
    });
 
    // 5. Cleanup old table (after verification period)
    console.log(`Rebuild complete. Old table ${tempTable} can be dropped after verification.`);
    
    // Optionally, schedule deferred cleanup
    // await this.scheduleCleanup(tempTable, { after: '1 hour' });
  }
}
 
/**
 * Handling projection version mismatches during rebuild
 */
interface ProjectionMetadata {
  version: number;  // Schema/logic version
  lastPosition: number;
  rebuildInProgress: boolean;
  rebuildStartedAt?: Date;
}
 
class VersionedProjection {
  static readonly CURRENT_VERSION = 3;
 
  constructor(
    private eventStore: EventStore,
    private metadataStore: MetadataStore
  ) {}
 
  async initialize(): Promise<void> {
    const metadata = await this.metadataStore.get<ProjectionMetadata>(
      'projection_meta',
      this.projectionName
    );
 
    if (!metadata) {
      // First time - full rebuild
      console.log(`Initializing projection ${this.projectionName} v${VersionedProjection.CURRENT_VERSION}`);
      await this.rebuild();
      return;
    }
 
    if (metadata.version < VersionedProjection.CURRENT_VERSION) {
      // Version mismatch - need rebuild
      console.log(
        `Projection version changed (${metadata.version} -> ${VersionedProjection.CURRENT_VERSION}). Rebuilding...`
      );
      await this.rebuild();
      return;
    }
 
    if (metadata.rebuildInProgress) {
      // Previous rebuild was interrupted
      console.log('Resuming interrupted rebuild');
      await this.rebuild(); // Or resume from metadata.lastPosition
      return;
    }
 
    // Normal startup - continue from checkpoint
    console.log(`Resuming from position ${metadata.lastPosition}`);
  }
 
  async rebuild(): Promise<void> {
    // Mark rebuild in progress
    await this.metadataStore.upsert('projection_meta', this.projectionName, {
      version: VersionedProjection.CURRENT_VERSION,
      lastPosition: 0,
      rebuildInProgress: true,
      rebuildStartedAt: new Date(),
    });
 
    try {
      // ... full rebuild logic ...
 
      // Mark rebuild complete
      await this.metadataStore.upsert('projection_meta', this.projectionName, {
        version: VersionedProjection.CURRENT_VERSION,
        lastPosition: finalPosition,
        rebuildInProgress: false,
      });
    } catch (error) {
      console.error('Rebuild failed:', error);
      throw error;
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

/**
 * Patterns for handling consistency in event-sourced systems
 */
 
// PATTERN 1: Read-Your-Writes with fallback
// When you need immediate consistency after a write
 
class ReadYourWritesHandler {
  constructor(
    private aggregateRepo: AggregateRepository,
    private readModelStore: ReadModelStore
  ) {}
 
  async getOrderAfterUpdate(orderId: string): Promise<OrderView> {
    // Try read model first (fast path)
    const readModel = await this.readModelStore.get<OrderView>('orders', orderId);
    
    // Check if read model is current enough
    // (e.g., compare lastUpdateCommand with what we just wrote)
    if (readModel && this.isFreshEnough(readModel)) {
      return readModel;
    }
    
    // Fallback to aggregate reconstitution (slow but consistent)
    const aggregate = await this.aggregateRepo.load(orderId);
    return aggregate.toView();
  }
 
  private isFreshEnough(readModel: OrderView): boolean {
    // Could check timestamp, version, or command correlation
    const maxLag = 5000; // 5 seconds
    return Date.now() - readModel.lastUpdated.getTime() < maxLag;
  }
}
 
// PATTERN 2: Saga for cross-aggregate operations
// When a business process spans multiple aggregates
 
interface OrderSagaState {
  orderId: string;
  customerId: string;
  total: number;
  step: 'started' | 'payment_reserved' | 'inventory_reserved' | 'completed' | 'compensating' | 'failed';
  failures: string[];
}
 
class OrderPlacementSaga {
  constructor(
    private sagaStore: SagaStore,
    private commandBus: CommandBus
  ) {}
 
  // Saga reacts to events and issues commands
  async handle(event: DomainEvent): Promise<void> {
    switch (event.eventType) {
      case 'OrderPlaced': {
        // Start the saga
        await this.sagaStore.create<OrderSagaState>({
          sagaId: `order-saga-${event.payload.orderId}`,
          state: {
            orderId: event.payload.orderId,
            customerId: event.payload.customerId,
            total: event.payload.total,
            step: 'started',
            failures: [],
          },
        });
        
        // Issue next command
        await this.commandBus.send({
          type: 'ReservePayment',
          customerId: event.payload.customerId,
          amount: event.payload.total,
          orderId: event.payload.orderId,
        });
        break;
      }
 
      case 'PaymentReserved': {
        const saga = await this.sagaStore.get<OrderSagaState>(
          `order-saga-${event.payload.orderId}`
        );
        
        await this.sagaStore.update(saga.sagaId, {
          step: 'payment_reserved',
        });
        
        // Next step: reserve inventory
        await this.commandBus.send({
          type: 'ReserveInventory',
          orderId: event.payload.orderId,
          items: saga.state.items,
        });
        break;
      }
 
      case 'InventoryReserved': {
        const saga = await this.sagaStore.get<OrderSagaState>(
          `order-saga-${event.payload.orderId}`
        );
        
        await this.sagaStore.update(saga.sagaId, {
          step: 'completed',
        });
        
        // Saga complete - trigger notification
        await this.commandBus.send({
          type: 'NotifyOrderConfirmed',
          orderId: event.payload.orderId,
          customerId: saga.state.customerId,
        });
        break;
      }
 
      case 'InventoryReservationFailed': {
        // Compensating action - release payment
        await this.compensate(event.payload.orderId, 'Inventory unavailable');
        break;
      }
    }
  }
 
  private async compensate(orderId: string, reason: string): Promise<void> {
    const saga = await this.sagaStore.get<OrderSagaState>(`order-saga-${orderId}`);
    
    await this.sagaStore.update(saga.sagaId, {
      step: 'compensating',
      failures: [...saga.state.failures, reason],
    });
 
    // Issue compensating commands based on completed steps
    if (saga.state.step === 'payment_reserved' || saga.state.step === 'inventory_reserved') {
      await this.commandBus.send({
        type: 'ReleasePaymentReservation',
        orderId,
        customerId: saga.state.customerId,
      });
    }
 
    // Mark order as failed
    await this.commandBus.send({
      type: 'FailOrder',
      orderId,
      reason,
    });
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165

/**
 * Performance optimization patterns
 */
 
// 1. Snapshot-based aggregate loading
interface SnapshotConfig {
  // Create snapshot every N events
  snapshotInterval: number;
  // Keep last N snapshots for fallback
  retainedSnapshots: number;
}
 
const DEFAULT_SNAPSHOT_CONFIG: SnapshotConfig = {
  snapshotInterval: 100,
  retainedSnapshots: 3,
};
 
class OptimizedAggregateRepository {
  constructor(
    private eventStore: EventStore,
    private snapshotStore: SnapshotStore,
    private config: SnapshotConfig = DEFAULT_SNAPSHOT_CONFIG
  ) {}
 
  async load<T extends Aggregate>(
    aggregateId: string,
    factory: AggregateFactory<T>
  ): Promise<T | null> {
    // Load snapshot if available
    const snapshot = await this.snapshotStore.getLatest(aggregateId);
    
    // Calculate events to replay
    const fromPosition = snapshot ? snapshot.position + 1 : 0;
    const events = await this.eventStore.readStream(aggregateId, { fromPosition });
    
    if (!snapshot && events.length === 0) {
      return null;
    }
 
    // Reconstitute from snapshot + events
    const aggregate = snapshot 
      ? factory.fromSnapshot(aggregateId, snapshot.state)
      : factory.create(aggregateId);
    
    aggregate.loadFromHistory(events);
 
    // Log warning if too many events since snapshot
    if (events.length > this.config.snapshotInterval * 2) {
      console.warn(
        `Aggregate ${aggregateId} has ${events.length} events since last snapshot. Consider snapshotting.`
      );
    }
 
    return aggregate;
  }
 
  async save<T extends Aggregate>(aggregate: T): Promise<void> {
    const events = aggregate.getUncommittedEvents();
    
    if (events.length === 0) return;
 
    await this.eventStore.append(
      aggregate.id,
      aggregate.version,
      events
    );
 
    // Check if we should create a snapshot
    const newVersion = aggregate.version + events.length;
    const eventsSinceSnapshot = await this.countEventsSinceSnapshot(aggregate.id);
    
    if (eventsSinceSnapshot >= this.config.snapshotInterval) {
      await this.createSnapshot(aggregate, newVersion);
    }
 
    aggregate.clearUncommittedEvents();
  }
 
  private async createSnapshot<T extends Aggregate>(
    aggregate: T,
    position: number
  ): Promise<void> {
    await this.snapshotStore.save({
      aggregateId: aggregate.id,
      aggregateType: aggregate.aggregateType,
      position,
      state: aggregate.getSnapshotState(),
      createdAt: new Date(),
    });
 
    // Cleanup old snapshots
    await this.snapshotStore.pruneOldSnapshots(
      aggregate.id,
      this.config.retainedSnapshots
    );
  }
}
 
// 2. Batch projection updates
class BatchingProjection {
  private pendingEvents: StoredEvent[] = [];
  private flushTimer: NodeJS.Timeout | null = null;
 
  constructor(
    private baseProjection: Projection<unknown>,
    private batchSize: number = 100,
    private flushIntervalMs: number = 1000
  ) {}
 
  async handle(event: StoredEvent): Promise<void> {
    this.pendingEvents.push(event);
 
    if (this.pendingEvents.length >= this.batchSize) {
      await this.flush();
    } else if (!this.flushTimer) {
      this.flushTimer = setTimeout(() => this.flush(), this.flushIntervalMs);
    }
  }
 
  private async flush(): Promise<void> {
    if (this.flushTimer) {
      clearTimeout(this.flushTimer);
      this.flushTimer = null;
    }
 
    if (this.pendingEvents.length === 0) return;
 
    const batch = this.pendingEvents;
    this.pendingEvents = [];
 
    // Process batch with single transaction
    await this.baseProjection.handleBatch(batch);
  }
}
 
// 3. Materialized view with TTL-based refresh
class TTLProjection {
  private cache = new Map<string, { data: unknown; expiresAt: number }>();
  
  constructor(
    private ttlMs: number,
    private computeFn: (id: string) => Promise<unknown>
  ) {}
 
  async get(id: string): Promise<unknown> {
    const cached = this.cache.get(id);
    
    if (cached && cached.expiresAt > Date.now()) {
      return cached.data;
    }
 
    const data = await this.computeFn(id);
    
    this.cache.set(id, {
      data,
      expiresAt: Date.now() + this.ttlMs,
    });
 
    return data;
  }
 
  invalidate(id: string): void {
    this.cache.delete(id);
  }
}