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

// Every event naturally carries audit information
interface AuditableEvent {
  // WHAT happened
  eventType: string;
  aggregateId: string;
  payload: unknown;
  
  // WHEN it happened
  timestamp: Date;
  
  // WHO initiated it (from command context)
  metadata: {
    userId?: string;           // Authenticated user
    impersonatingUser?: string; // Admin acting as user
    ipAddress?: string;         // Request origin
    userAgent?: string;         // Client information
    
    // WHY (business context)
    correlationId: string;      // Business transaction ID
    causationId: string;        // What triggered this event
    commandId?: string;         // Original command
    
    // Additional context
    requestId?: string;         // Request tracing
    sessionId?: string;         // User session
  };
}
 
// Audit query examples
class AuditQueryService {
  constructor(private eventStore: EventStore) {}
 
  /**
   * "Show me everything user X did on date Y"
   */
  async getUserActivityOnDate(
    userId: string,
    date: Date
  ): Promise<AuditableEvent[]> {
    const startOfDay = new Date(date);
    startOfDay.setHours(0, 0, 0, 0);
    
    const endOfDay = new Date(date);
    endOfDay.setHours(23, 59, 59, 999);
 
    const allEvents = await this.eventStore.readAll({
      fromTimestamp: startOfDay,
      toTimestamp: endOfDay,
    });
 
    return allEvents.events.filter(
      e => e.metadata?.userId === userId
    ) as AuditableEvent[];
  }
 
  /**
   * "What happened to order #123?"
   */
  async getEntityHistory(entityType: string, entityId: string): Promise<{
    events: AuditableEvent[];
    timeline: TimelineEntry[];
  }> {
    const stream = `${entityType}-${entityId}`;
    const result = await this.eventStore.readStream(stream);
 
    return {
      events: result.events as AuditableEvent[],
      timeline: result.events.map(e => ({
        timestamp: e.timestamp,
        action: e.eventType,
        actor: e.metadata?.userId ?? 'system',
        summary: this.summarizeEvent(e),
      })),
    };
  }
 
  /**
   * "Prove that balance was X on audit date Y"
   */
  async getStateAsOfDate<T>(
    aggregateId: string,
    date: Date,
    reducer: (events: StoredEvent[]) => T
  ): Promise<{
    state: T;
    provenBy: StoredEvent[];
    lastEventBeforeDate: StoredEvent;
  }> {
    const result = await this.eventStore.readStream(aggregateId);
    const eventsBeforeDate = result.events.filter(e => e.timestamp <= date);
    
    return {
      state: reducer(eventsBeforeDate),
      provenBy: eventsBeforeDate,
      lastEventBeforeDate: eventsBeforeDate[eventsBeforeDate.length - 1],
    };
  }
}

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

/**
 * Debugging tools enabled by event sourcing
 */
 
// Tool 1: Event Stream Inspector
class StreamInspector {
  constructor(private eventStore: EventStore) {}
 
  async inspect(streamId: string): Promise<void> {
    const result = await this.eventStore.readStream(streamId);
    
    console.log(`\n=== Stream: ${streamId} ===`);
    console.log(`Total events: ${result.events.length}`);
    console.log(`Current version: ${result.version}`);
    console.log(`\n--- Event Timeline ---`);
    
    for (const event of result.events) {
      console.log(`[${event.sequenceNumber}] ${event.timestamp.toISOString()}`);
      console.log(`    Type: ${event.eventType}`);
      console.log(`    Data: ${JSON.stringify(event.data, null, 2).slice(0, 200)}...`);
      console.log();
    }
  }
}
 
// Tool 2: Step-by-step state replay
class StateReplayer {
  async replayWithInspection<TState>(
    events: StoredEvent[],
    initialState: TState,
    apply: (state: TState, event: StoredEvent) => TState,
    inspector?: (step: number, event: StoredEvent, before: TState, after: TState) => void
  ): Promise<TState[]> {
    const stateHistory: TState[] = [initialState];
    let current = initialState;
    
    for (let i = 0; i < events.length; i++) {
      const event = events[i];
      const before = current;
      const after = apply(current, event);
      
      stateHistory.push(after);
      
      if (inspector) {
        inspector(i, event, before, after);
      }
      
      current = after;
    }
    
    return stateHistory;
  }
}
 
// Example debugging session
async function debugOrderIssue(orderId: string): Promise<void> {
  const eventStore = new EventStore();
  const inspector = new StreamInspector(eventStore);
  const replayer = new StateReplayer();
  
  // 1. Get the event stream
  const stream = `order-${orderId}`;
  const result = await eventStore.readStream(stream);
  
  console.log(`Debugging order ${orderId}`);
  console.log(`Found ${result.events.length} events`);
  
  // 2. Replay with step-by-step inspection
  const stateHistory = await replayer.replayWithInspection(
    result.events,
    createEmptyOrderState(orderId),
    applyOrderEvent,
    (step, event, before, after) => {
      console.log(`\n--- Step ${step + 1} ---`);
      console.log(`Event: ${event.eventType} at ${event.timestamp}`);
      console.log(`Before: status=${before.status}, total=${before.total}`);
      console.log(`After:  status=${after.status}, total=${after.total}`);
      
      // Check for anomalies
      if (after.total < 0) {
        console.error(`!!! ANOMALY: Negative total after ${event.eventType}`);
      }
      if (before.status === 'completed' && after.status !== 'completed') {
        console.error(`!!! ANOMALY: Status went backwards from completed`);
      }
    }
  );
  
  // 3. Compare to expected end state
  const finalState = stateHistory[stateHistory.length - 1];
  console.log(`\n=== Final State ===`);
  console.log(JSON.stringify(finalState, null, 2));
}
 
// Tool 3: Event diff between two streams
class StreamDiffer {
  static diff(
    events1: StoredEvent[],
    events2: StoredEvent[]
  ): EventDiff {
    const diff: EventDiff = {
      onlyIn1: [],
      onlyIn2: [],
      different: [],
    };
    
    const byId1 = new Map(events1.map(e => [e.eventId, e]));
    const byId2 = new Map(events2.map(e => [e.eventId, e]));
    
    for (const [id, event] of byId1) {
      if (!byId2.has(id)) {
        diff.onlyIn1.push(event);
      } else {
        const other = byId2.get(id)!;
        if (JSON.stringify(event.data) !== JSON.stringify(other.data)) {
          diff.different.push({ event1: event, event2: other });
        }
      }
    }
    
    for (const [id, event] of byId2) {
      if (!byId1.has(id)) {
        diff.onlyIn2.push(event);
      }
    }
    
    return diff;
  }
}
 
// Tool 4: What-if analysis
class WhatIfAnalyzer {
  async analyzeWithoutEvent(
    streamId: string,
    eventIdToSkip: string
  ): Promise<{ with: unknown; without: unknown; difference: unknown }> {
    const result = await this.eventStore.readStream(streamId);
    
    const allEvents = result.events;
    const filteredEvents = allEvents.filter(e => e.eventId !== eventIdToSkip);
    
    const stateWith = this.aggregate.reconstitute(allEvents);
    const stateWithout = this.aggregate.reconstitute(filteredEvents);
    
    return {
      with: stateWith,
      without: stateWithout,
      difference: this.computeDiff(stateWith, stateWithout),
    };
  }
}

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

/**
 * Event-driven integration patterns
 */
 
// Pattern 1: Event Router to multiple consumers
class EventRouter {
  private handlers = new Map<string, EventHandler[]>();
 
  registerHandler(eventType: string, handler: EventHandler): void {
    const existing = this.handlers.get(eventType) ?? [];
    this.handlers.set(eventType, [...existing, handler]);
  }
 
  async route(event: StoredEvent): Promise<void> {
    const handlers = this.handlers.get(event.eventType) ?? [];
    
    // Fan-out to all registered handlers
    await Promise.all(
      handlers.map(handler => 
        handler(event).catch(err => {
          console.error(`Handler error for ${event.eventType}:`, err);
          // Dead-letter or retry logic here
        })
      )
    );
  }
}
 
// Pattern 2: New consumer bootstrapping via replay
class NewConsumerBootstrap {
  constructor(
    private eventStore: EventStore,
    private checkpointStore: CheckpointStore
  ) {}
 
  async bootstrap(
    consumerId: string,
    handler: EventHandler,
    options: {
      startFrom?: Position;
      eventTypes?: string[];
      progressCallback?: (position: number, total: number) => void;
    } = {}
  ): Promise<void> {
    const startPos = options.startFrom ?? 0;
    const headPos = await this.eventStore.getHeadPosition();
    
    console.log(`Bootstrapping ${consumerId} from ${startPos} to ${headPos}`);
    
    let position = startPos;
    const batchSize = 1000;
    
    while (position <= headPos) {
      const result = await this.eventStore.readAll({
        fromPosition: position,
        count: batchSize,
        eventTypes: options.eventTypes,
      });
      
      if (result.events.length === 0) break;
      
      for (const event of result.events) {
        await handler(event);
      }
      
      position = result.events[result.events.length - 1].globalPosition + 1;
      
      // Save checkpoint periodically
      await this.checkpointStore.saveCheckpoint({
        subscriptionId: consumerId,
        position,
        updatedAt: new Date(),
      });
      
      if (options.progressCallback) {
        options.progressCallback(position, headPos);
      }
    }
    
    console.log(`Bootstrap complete for ${consumerId}`);
  }
}
 
// Pattern 3: Outbox-free publishing
// In event sourcing, events ARE the outbox
class EventPublisher {
  constructor(private subscription: Subscription) {}
 
  async startPublishing(
    target: ExternalMessageBroker,
    transform: (event: StoredEvent) => ExternalMessage
  ): Promise<void> {
    // Events are already durable in the event store
    // No separate outbox table needed!
    await this.subscription.start({
      onEvent: async (event) => {
        const message = transform(event);
        await target.publish(message);
      },
      onError: async (error, event) => {
        // Retry logic or dead-letter
        console.error('Publish failed:', error, event);
      },
    });
  }
}
 
// Pattern 4: Analytics streaming
class AnalyticsExporter {
  constructor(
    private eventStore: EventStore,
    private dataWarehouse: DataWarehouse
  ) {}
 
  async exportForAnalytics(): Promise<void> {
    const lastExported = await this.dataWarehouse.getLastExportedPosition();
    
    const result = await this.eventStore.readAll({
      fromPosition: lastExported + 1,
      count: 10000,
    });
    
    if (result.events.length === 0) return;
    
    // Transform events to analytics format
    const records = result.events.map(event => ({
      event_id: event.eventId,
      event_type: event.eventType,
      aggregate_id: event.aggregateId,
      timestamp: event.timestamp,
      user_id: event.metadata?.userId,
      data: event.data, // Flatten as needed for your schema
    }));
    
    // Batch insert into data warehouse
    await this.dataWarehouse.batchInsert('events_raw', records);
    
    // Update checkpoint
    await this.dataWarehouse.setLastExportedPosition(
      result.events[result.events.length - 1].globalPosition
    );
  }
}

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

/**
 * Strategies for handling eventual consistency
 */
 
// Strategy 1: Read-your-writes guarantee
class ConsistentReadService {
  constructor(
    private aggregateRepo: AggregateRepository,
    private readModelStore: ReadModelStore
  ) {}
 
  async getProfileWithConsistency(
    userId: string,
    expectedVersion?: number
  ): Promise<UserProfile> {
    const readModel = await this.readModelStore.get<UserProfile>(
      'profiles',
      userId
    );
    
    // If we have an expected version, wait or fallback
    if (expectedVersion !== undefined) {
      if (!readModel || readModel.version < expectedVersion) {
        // Option A: Wait briefly for projection
        const updated = await this.waitForProjection(userId, expectedVersion, 2000);
        if (updated) return updated;
        
        // Option B: Fall back to aggregate
        console.log(`Projection stale, falling back to aggregate for ${userId}`);
        const aggregate = await this.aggregateRepo.load(userId);
        return aggregate.toProfile();
      }
    }
    
    return readModel!;
  }
 
  private async waitForProjection(
    userId: string,
    expectedVersion: number,
    timeoutMs: number
  ): Promise<UserProfile | null> {
    const startTime = Date.now();
    const pollInterval = 100;
    
    while (Date.now() - startTime < timeoutMs) {
      const current = await this.readModelStore.get<UserProfile>('profiles', userId);
      if (current && current.version >= expectedVersion) {
        return current;
      }
      await sleep(pollInterval);
    }
    
    return null;
  }
}
 
// Strategy 2: Optimistic UI with confirmation
interface OptimisticUpdateResponse {
  success: boolean;
  // Return what the UI should show immediately
  optimisticState: Partial<UserProfile>;
  // Token to poll for confirmation
  confirmationToken: string;
  // Expected version after update
  expectedVersion: number;
}
 
class OptimisticUpdateHandler {
  async updateEmail(
    userId: string,
    newEmail: string
  ): Promise<OptimisticUpdateResponse> {
    // Execute command
    const result = await this.commandBus.send({
      type: 'ChangeEmail',
      userId,
      newEmail,
    });
    
    return {
      success: true,
      optimisticState: { email: newEmail },
      confirmationToken: result.correlationId,
      expectedVersion: result.newVersion,
    };
  }
}
 
// Frontend handling
async function handleEmailChange(newEmail: string): Promise<void> {
  // 1. Show optimistic update immediately
  displayEmail(newEmail);
  showProcessingIndicator();
  
  // 2. Send update
  const response = await api.updateEmail(userId, newEmail);
  
  // 3. Poll or subscribe for confirmation
  const confirmed = await waitForConfirmation(
    response.confirmationToken,
    response.expectedVersion,
    5000 // timeout
  );
  
  if (confirmed) {
    hideProcessingIndicator();
    showSuccessToast('Email updated');
  } else {
    // Fetch actual state and reconcile
    const actual = await api.getProfile(userId);
    displayEmail(actual.email);
    if (actual.email !== newEmail) {
      showWarningToast('Please verify your email change');
    }
  }
}
 
// Strategy 3: Synchronous projections for critical paths
class CriticalPathProjection {
  async handleWithImmediateConsistency(
    event: StoredEvent,
    transaction: Transaction
  ): Promise<void> {
    // Update projection in the same transaction as event write
    switch (event.eventType) {
      case 'PaymentReceived':
        await transaction.update('orders', event.aggregateId, {
          paymentStatus: 'paid',
          paidAt: event.timestamp,
          version: event.sequenceNumber,
        });
        break;
    }
  }
}

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

/**
 * Event schema evolution patterns
 */
 
// Pattern 1: Upcasting - Transform old versions to new on read
interface EventUpcaster {
  eventType: string;
  fromVersion: number;
  toVersion: number;
  upcast(event: StoredEvent): StoredEvent;
}
 
class UpcasterRegistry {
  private upcasters = new Map<string, EventUpcaster[]>();
 
  register(upcaster: EventUpcaster): void {
    const key = `${upcaster.eventType}:${upcaster.fromVersion}`;
    const existing = this.upcasters.get(key) ?? [];
    this.upcasters.set(key, [...existing, upcaster]);
  }
 
  upcast(event: StoredEvent): StoredEvent {
    const version = event.schemaVersion ?? 1;
    let current = event;
    let currentVersion = version;
    
    while (true) {
      const key = `${event.eventType}:${currentVersion}`;
      const upcasters = this.upcasters.get(key);
      
      if (!upcasters || upcasters.length === 0) {
        break; // No more upcasters
      }
      
      for (const upcaster of upcasters) {
        current = upcaster.upcast(current);
        currentVersion = upcaster.toVersion;
      }
    }
    
    return current;
  }
}
 
// Example upcasters
const orderPlacedV1ToV2: EventUpcaster = {
  eventType: 'OrderPlaced',
  fromVersion: 1,
  toVersion: 2,
  upcast(event) {
    // V1 had flat address fields; V2 has nested address object
    const data = event.data as OrderPlacedV1Data;
    return {
      ...event,
      schemaVersion: 2,
      data: {
        ...data,
        // Convert flat fields to nested structure
        shippingAddress: {
          street: data.shippingStreet ?? '',
          city: data.shippingCity ?? '',
          postalCode: data.shippingZip ?? '',
          country: data.shippingCountry ?? 'US',
        },
        // Remove old flat fields (or keep for compatibility)
      } as OrderPlacedV2Data,
    };
  },
};
 
const orderPlacedV2ToV3: EventUpcaster = {
  eventType: 'OrderPlaced',
  fromVersion: 2,
  toVersion: 3,
  upcast(event) {
    // V3 added giftOptions
    const data = event.data as OrderPlacedV2Data;
    return {
      ...event,
      schemaVersion: 3,
      data: {
        ...data,
        giftOptions: {
          isGift: false,
          wrapStyle: null,
          giftMessage: null,
        },
      } as OrderPlacedV3Data,
    };
  },
};
 
// Pattern 2: Defensive deserialization
function deserializeOrderPlaced(event: StoredEvent): OrderPlacedData {
  const raw = event.data as Record<string, unknown>;
  const version = event.schemaVersion ?? 1;
  
  // Always return consistent shape, regardless of version
  return {
    orderId: raw.orderId as string,
    customerId: raw.customerId as string,
    items: raw.items as OrderItem[],
    total: raw.total as number,
    // Handle fields that may be missing in old versions
    shippingAddress: raw.shippingAddress 
      ? raw.shippingAddress as Address
      : {
          // Reconstruct from old flat fields if present
          street: (raw.shippingStreet as string) ?? '',
          city: (raw.shippingCity as string) ?? '',
          postalCode: (raw.shippingZip as string) ?? '',
          country: (raw.shippingCountry as string) ?? 'UNKNOWN',
        },
    giftOptions: raw.giftOptions as GiftOptions | undefined,
    currency: (raw.currency as string) ?? 'USD', // Defaulted field
    createdAt: new Date(raw.createdAt as string),
  };
}
 
// Pattern 3: Typed event handlers with version awareness
class VersionAwareProjection {
  handle(event: StoredEvent): void {
    // Upcast to latest version
    const upcasted = this.upcasterRegistry.upcast(event);
    
    switch (event.eventType) {
      case 'OrderPlaced':
        this.handleOrderPlaced(deserializeOrderPlaced(upcasted));
        break;
      // ... other handlers
    }
  }
  
  private handleOrderPlaced(data: OrderPlacedData): void {
    // Always works with latest schema shape
    // No version checks in business logic
  }
}

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

/**
 * Storage management strategies
 */
 
// Strategy 1: Tiered storage with archival
class TieredEventStore {
  constructor(
    private hotStore: EventStore,     // Fast SSD, recent events
    private warmStore: EventStore,    // Standard storage, older events
    private coldStore: ArchivalStore, // S3/Glacier, archived events
    private config: TieringConfig
  ) {}
 
  async readStream(
    streamId: string,
    options?: ReadOptions
  ): Promise<ReadResult> {
    // Try hot store first
    const hotEvents = await this.hotStore.readStream(streamId, options);
    
    if (this.coversFullHistory(hotEvents, streamId)) {
      return hotEvents;
    }
    
    // Need to fetch from warm/cold storage
    const warmEvents = await this.warmStore.readStream(streamId, {
      ...options,
      toPosition: hotEvents.events[0]?.sequenceNumber ?? Number.MAX_SAFE_INTEGER,
    });
    
    return {
      events: [...warmEvents.events, ...hotEvents.events],
      version: hotEvents.version,
    };
  }
 
  // Background job: move old events to cooler storage
  async archiveOldEvents(): Promise<void> {
    const cutoffDate = new Date();
    cutoffDate.setDate(cutoffDate.getDate() - this.config.hotRetentionDays);
    
    const eventsToArchive = await this.hotStore.readAll({
      toTimestamp: cutoffDate,
      count: 10000,
    });
    
    // Move to warm store
    await this.warmStore.appendBatch(eventsToArchive.events);
    
    // Mark as archived in hot store (or delete if warm is durable)
    await this.hotStore.markArchived(eventsToArchive.events.map(e => e.eventId));
  }
}
 
// Strategy 2: Aggregate-aware snapshotting
class AdaptiveSnapshotting {
  constructor(
    private snapshotStore: SnapshotStore,
    private metricsService: MetricsService
  ) {}
 
  async shouldSnapshot(
    aggregateId: string,
    eventCount: number,
    lastSnapshotAge: number
  ): Promise<boolean> {
    // Dynamic thresholds based on aggregate access patterns
    const accessFrequency = await this.metricsService.getAccessFrequency(aggregateId);
    
    // High-traffic aggregates: snapshot more frequently
    if (accessFrequency > 100) { // 100 reads/hour
      return eventCount >= 20 || lastSnapshotAge > 3600000; // 1 hour
    }
    
    // Medium traffic
    if (accessFrequency > 10) {
      return eventCount >= 50 || lastSnapshotAge > 86400000; // 1 day
    }
    
    // Low traffic: standard snapshotting
    return eventCount >= 100;
  }
}
 
// Strategy 3: Event compaction (use with caution!)
interface CompactionConfig {
  // Only compact events older than this
  minAge: number;
  // Compact if more than this many events
  minEvents: number;
  // Event types that are safe to compact
  compactableTypes: string[];
}
 
class EventCompaction {
  /**
   * CAUTION: Compaction loses granularity!
   * Only use for specific, well-understood scenarios
   * (e.g., summarizing daily metrics into weekly)
   */
  async compact(
    streamId: string,
    config: CompactionConfig
  ): Promise<void> {
    const events = await this.eventStore.readStream(streamId);
    
    // Find events eligible for compaction
    const cutoffDate = new Date(Date.now() - config.minAge);
    const oldEvents = events.events.filter(
      e => e.timestamp < cutoffDate && 
           config.compactableTypes.includes(e.eventType)
    );
    
    if (oldEvents.length < config.minEvents) {
      return; // Not enough to compact
    }
    
    // Create summary event
    const summary = this.summarizeEvents(oldEvents);
    
    // Write summary to new stream (preserving original)
    await this.eventStore.append(
      `${streamId}:compacted`,
      'any',
      [summary]
    );
    
    console.log(
      `Compacted ${oldEvents.length} events into 1 summary for ${streamId}`
    );
  }
}