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

// Snapshot structure and storage
 
interface Snapshot<TState> {
  // Identity
  snapshotId: string;              // Unique ID for this snapshot
  aggregateId: string;             // Which aggregate this is a snapshot of
  aggregateType: string;           // Type for validation and routing
  
  // Version tracking
  version: number;                 // Event stream version at snapshot time
  schemaVersion: number;           // State schema version for compatibility
  
  // Timing
  createdAt: Date;                 // When snapshot was created
  
  // The actual state
  state: TState;                   // Serialized aggregate state
  
  // Integrity
  checksum?: string;               // Hash for corruption detection
}
 
// Example snapshot for a bank account
interface BankAccountState {
  accountId: string;
  balance: number;
  status: 'active' | 'frozen' | 'closed';
  overdraftLimit: number;
  transactionCount: number;
  lastTransactionAt?: Date;
}
 
const exampleSnapshot: Snapshot<BankAccountState> = {
  snapshotId: 'snap_1234567890',
  aggregateId: 'BankAccount-acct_abc123',
  aggregateType: 'BankAccount',
  version: 5000,                   // State as of event #5000
  schemaVersion: 3,                // Third iteration of state schema
  createdAt: new Date('2024-01-15T10:00:00Z'),
  state: {
    accountId: 'acct_abc123',
    balance: 15750.42,
    status: 'active',
    overdraftLimit: 500,
    transactionCount: 5000,
    lastTransactionAt: new Date('2024-01-15T09:45:00Z'),
  },
  checksum: 'sha256:a1b2c3d4...',
};
 
// Snapshot store interface
interface SnapshotStore {
  // Save a new snapshot
  save<TState>(snapshot: Snapshot<TState>): Promise<void>;
  
  // Get the latest snapshot for an aggregate
  getLatest<TState>(aggregateId: string): Promise<Snapshot<TState> | null>;
  
  // Get snapshot at or before a specific version
  getAtVersion<TState>(
    aggregateId: string, 
    maxVersion: number
  ): Promise<Snapshot<TState> | null>;
  
  // Delete old snapshots (retain only N most recent)
  prune(aggregateId: string, keepCount: number): Promise<number>;
}
 
// Loading with snapshot
async function loadAggregate<TState, TEvent>(
  aggregateId: string,
  eventStore: EventStore,
  snapshotStore: SnapshotStore,
  initialState: TState,
  apply: (state: TState, event: TEvent) => TState
): Promise<{ state: TState; version: number }> {
  
  // 1. Try to load the latest snapshot
  const snapshot = await snapshotStore.getLatest<TState>(aggregateId);
  
  let currentState: TState;
  let fromVersion: number;
  
  if (snapshot) {
    // Start from snapshot
    currentState = snapshot.state;
    fromVersion = snapshot.version;
    console.log(`Loaded snapshot at version ${fromVersion}`);
  } else {
    // No snapshot, start from scratch
    currentState = initialState;
    fromVersion = 0;
  }
  
  // 2. Load events since snapshot
  const events = await eventStore.readStream<TEvent>(aggregateId, {
    fromVersion: fromVersion + 1,
  });
  
  console.log(`Replaying ${events.length} events from version ${fromVersion + 1}`);
  
  // 3. Apply events to reach current state
  for (const event of events) {
    currentState = apply(currentState, event);
  }
  
  const finalVersion = fromVersion + events.length;
  
  return { state: currentState, version: finalVersion };
}

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

// Snapshot creation strategies
 
// Strategy 1: Every N events
class PeriodicSnapshotter {
  constructor(
    private snapshotStore: SnapshotStore,
    private interval: number = 100 // Snapshot every 100 events
  ) {}
  
  shouldSnapshot(previousVersion: number, currentVersion: number): boolean {
    // Snapshot if we've crossed an interval boundary
    const previousInterval = Math.floor(previousVersion / this.interval);
    const currentInterval = Math.floor(currentVersion / this.interval);
    return currentInterval > previousInterval;
  }
  
  async maybeSnapshot<TState>(
    aggregate: { id: string; state: TState; version: number },
    previousVersion: number
  ): Promise<void> {
    if (this.shouldSnapshot(previousVersion, aggregate.version)) {
      await this.snapshotStore.save({
        snapshotId: generateId(),
        aggregateId: aggregate.id,
        aggregateType: aggregate.constructor.name,
        version: aggregate.version,
        schemaVersion: getCurrentSchemaVersion(),
        createdAt: new Date(),
        state: aggregate.state,
      });
    }
  }
}
 
// Strategy 2: Time-based
class TimeBasedSnapshotter {
  private lastSnapshotTime = new Map<string, Date>();
  
  constructor(
    private snapshotStore: SnapshotStore,
    private minIntervalMs: number = 3600000 // 1 hour minimum
  ) {}
  
  shouldSnapshot(aggregateId: string): boolean {
    const lastTime = this.lastSnapshotTime.get(aggregateId);
    if (!lastTime) return true; // Never snapshotted
    
    const elapsed = Date.now() - lastTime.getTime();
    return elapsed >= this.minIntervalMs;
  }
  
  async maybeSnapshot<TState>(
    aggregate: { id: string; state: TState; version: number }
  ): Promise<void> {
    if (this.shouldSnapshot(aggregate.id)) {
      await this.snapshotStore.save({
        snapshotId: generateId(),
        aggregateId: aggregate.id,
        aggregateType: aggregate.constructor.name,
        version: aggregate.version,
        schemaVersion: getCurrentSchemaVersion(),
        createdAt: new Date(),
        state: aggregate.state,
      });
      this.lastSnapshotTime.set(aggregate.id, new Date());
    }
  }
}
 
// Strategy 3: Adaptive based on replay cost
class AdaptiveSnapshotter {
  constructor(
    private snapshotStore: SnapshotStore,
    private maxReplayEvents: number = 500
  ) {}
  
  async shouldSnapshot(
    aggregateId: string,
    currentVersion: number
  ): Promise<boolean> {
    const latestSnapshot = await this.snapshotStore.getLatest(aggregateId);
    const snapshotVersion = latestSnapshot?.version ?? 0;
    const eventsSinceSnapshot = currentVersion - snapshotVersion;
    
    return eventsSinceSnapshot >= this.maxReplayEvents;
  }
}
 
// Strategy 4: Background snapshotting via subscription
class BackgroundSnapshotter {
  constructor(
    private eventStore: EventStore,
    private snapshotStore: SnapshotStore,
    private snapshotThreshold: number = 100
  ) {}
  
  async start(): Promise<void> {
    // Subscribe to all events
    await this.eventStore.subscribeToAll(async (event) => {
      await this.checkAndSnapshot(event.aggregateId, event.streamPosition);
    });
  }
  
  private async checkAndSnapshot(
    aggregateId: string,
    currentVersion: number
  ): Promise<void> {
    const latestSnapshot = await this.snapshotStore.getLatest(aggregateId);
    const snapshotVersion = latestSnapshot?.version ?? 0;
    
    if (currentVersion - snapshotVersion >= this.snapshotThreshold) {
      // Create snapshot asynchronously
      await this.createSnapshot(aggregateId, currentVersion);
    }
  }
  
  private async createSnapshot(
    aggregateId: string,
    targetVersion: number
  ): Promise<void> {
    // Load aggregate up to target version
    const events = await this.eventStore.readStream(aggregateId, {
      maxCount: targetVersion,
    });
    
    const state = events.reduce(applyEvent, initialState);
    
    await this.snapshotStore.save({
      snapshotId: generateId(),
      aggregateId,
      aggregateType: inferAggregateType(aggregateId),
      version: targetVersion,
      schemaVersion: getCurrentSchemaVersion(),
      createdAt: new Date(),
      state,
    });
  }
}
 
// Strategy 5: On-demand / lazy snapshotting
class LazySnapshotter {
  constructor(
    private snapshotStore: SnapshotStore,
    private replayThreshold: number = 200
  ) {}
  
  // Only snapshot after a slow load
  async snapshotIfSlowLoad<TState>(
    aggregate: { id: string; state: TState; version: number },
    eventsReplayed: number,
    loadTimeMs: number
  ): Promise<void> {
    // Snapshot if we replayed many events or load was slow
    if (eventsReplayed >= this.replayThreshold || loadTimeMs > 100) {
      await this.snapshotStore.save({
        snapshotId: generateId(),
        aggregateId: aggregate.id,
        aggregateType: aggregate.constructor.name,
        version: aggregate.version,
        schemaVersion: getCurrentSchemaVersion(),
        createdAt: new Date(),
        state: aggregate.state,
      });
    }
  }
}

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

// Snapshot schema evolution and migration
 
// Version-aware snapshot loading
interface SnapshotMigrator<TState> {
  currentVersion: number;
  migrate(snapshot: Snapshot<unknown>): TState;
}
 
class BankAccountMigrator implements SnapshotMigrator<BankAccountState> {
  currentVersion = 3;
  
  migrate(snapshot: Snapshot<unknown>): BankAccountState {
    let state = snapshot.state as Record<string, unknown>;
    let version = snapshot.schemaVersion;
    
    // Apply migrations sequentially
    if (version === 1) {
      state = this.migrateV1ToV2(state);
      version = 2;
    }
    
    if (version === 2) {
      state = this.migrateV2ToV3(state);
      version = 3;
    }
    
    if (version !== this.currentVersion) {
      throw new Error(
        `Cannot migrate from version ${snapshot.schemaVersion} to ${this.currentVersion}`
      );
    }
    
    return state as BankAccountState;
  }
  
  // V1 -> V2: Added 'overdraftLimit' field
  private migrateV1ToV2(state: Record<string, unknown>): Record<string, unknown> {
    return {
      ...state,
      overdraftLimit: 0, // Default value for existing accounts
    };
  }
  
  // V2 -> V3: Renamed 'balance' to 'currentBalance', added 'currency'
  private migrateV2ToV3(state: Record<string, unknown>): Record<string, unknown> {
    return {
      ...state,
      currentBalance: state.balance,
      currency: 'USD', // Assume USD for legacy accounts
    };
  }
}
 
// Safe snapshot loading with migration
async function loadWithMigration<TState>(
  aggregateId: string,
  snapshotStore: SnapshotStore,
  eventStore: EventStore,
  migrator: SnapshotMigrator<TState>,
  initialState: TState,
  apply: (state: TState, event: unknown) => TState
): Promise<{ state: TState; version: number }> {
  
  const snapshot = await snapshotStore.getLatest<unknown>(aggregateId);
  
  let state: TState;
  let fromVersion: number;
  
  if (snapshot) {
    if (snapshot.schemaVersion === migrator.currentVersion) {
      // Current version, no migration needed
      state = snapshot.state as TState;
    } else if (snapshot.schemaVersion < migrator.currentVersion) {
      // Old version, migrate it
      console.log(
        `Migrating snapshot from v${snapshot.schemaVersion} to v${migrator.currentVersion}`
      );
      state = migrator.migrate(snapshot);
      
      // Optionally: save migrated snapshot
      await snapshotStore.save({
        ...snapshot,
        schemaVersion: migrator.currentVersion,
        state,
      });
    } else {
      // Snapshot is from the FUTURE? This is a problem.
      throw new Error(
        `Snapshot version ${snapshot.schemaVersion} is newer than ` +
                                        `current version ${migrator.currentVersion}. ` +
                                    `Is this service running old code?`
      );
    }
    fromVersion = snapshot.version;
  } else {
    state = initialState;
    fromVersion = 0;
  }
  
  // Replay events since snapshot
  const events = await eventStore.readStream(aggregateId, {
    fromVersion: fromVersion + 1,
  });
  
  for (const event of events) {
    state = apply(state, event);
  }
  
  return { state, version: fromVersion + events.length };
}
 
// Alternative: Invalidate old snapshots instead of migrating
class InvalidatingSnapshotStore implements SnapshotStore {
  constructor(
    private baseStore: SnapshotStore,
    private currentSchemaVersion: number
  ) {}
  
  async getLatest<TState>(aggregateId: string): Promise<Snapshot<TState> | null> {
    const snapshot = await this.baseStore.getLatest<TState>(aggregateId);
    
    if (snapshot && snapshot.schemaVersion !== this.currentSchemaVersion) {
      // Old schema version, treat as if no snapshot exists
      // Full replay will create a snapshot with new schema
      console.log(
        `Ignoring stale snapshot (v${snapshot.schemaVersion}) for ${aggregateId}`
      );
      return null;
    }
    
    return snapshot;
  }
  
  save<TState>(snapshot: Snapshot<TState>): Promise<void> {
    // Always save with current version
    return this.baseStore.save({
      ...snapshot,
      schemaVersion: this.currentSchemaVersion,
    });
  }
  
  // ... other methods delegate to baseStore
}

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

// Snapshot verification and recovery
 
// Checksums for corruption detection
function createSnapshotWithChecksum<TState>(
  aggregateId: string,
  state: TState,
  version: number
): Snapshot<TState> {
  const stateJson = JSON.stringify(state);
  const checksum = crypto
    .createHash('sha256')
    .update(stateJson)
    .digest('hex');
  
  return {
    snapshotId: generateId(),
    aggregateId,
    aggregateType: inferType(aggregateId),
    version,
    schemaVersion: getCurrentSchemaVersion(),
    createdAt: new Date(),
    state,
    checksum: `sha256:${checksum}`,
  };
}
 
function verifySnapshotChecksum<TState>(snapshot: Snapshot<TState>): boolean {
  if (!snapshot.checksum) return true; // No checksum to verify
  
  const stateJson = JSON.stringify(snapshot.state);
  const expectedChecksum = crypto
    .createHash('sha256')
    .update(stateJson)
    .digest('hex');
  
  return snapshot.checksum === `sha256:${expectedChecksum}`;
}
 
// Verification during load
async function loadWithVerification<TState>(
  aggregateId: string,
  snapshotStore: SnapshotStore,
  eventStore: EventStore,
  initialState: TState,
  apply: (state: TState, event: unknown) => TState
): Promise<{ state: TState; version: number }> {
  
  const snapshot = await snapshotStore.getLatest<TState>(aggregateId);
  
  if (snapshot) {
    // Verify checksum
    if (!verifySnapshotChecksum(snapshot)) {
      console.error(`Snapshot corruption detected for ${aggregateId}`);
      metrics.increment('snapshot.corruption.detected');
      
      // Fall back to full replay
      return loadWithoutSnapshot(aggregateId, eventStore, initialState, apply);
    }
  }
  
  // Normal load path...
  return loadWithSnapshot(snapshot, aggregateId, eventStore, initialState, apply);
}
 
// Periodic snapshot verification job
async function verifyAllSnapshots(
  snapshotStore: SnapshotStore,
  eventStore: EventStore
): Promise<VerificationReport> {
  const report: VerificationReport = {
    totalChecked: 0,
    valid: 0,
    corrupted: 0,
    stale: 0,
    errors: [],
  };
  
  // Iterate all snapshots
  for await (const snapshot of snapshotStore.listAll()) {
    report.totalChecked++;
    
    try {
      // 1. Check checksum
      if (!verifySnapshotChecksum(snapshot)) {
        report.corrupted++;
        report.errors.push({
          aggregateId: snapshot.aggregateId,
          issue: 'checksum_mismatch',
        });
        continue;
      }
      
      // 2. Verify snapshot matches replay (expensive but thorough)
      const replayedState = await replayFromScratch(
        snapshot.aggregateId,
        eventStore,
        snapshot.version
      );
      
      if (!deepEqual(snapshot.state, replayedState)) {
        report.stale++;
        report.errors.push({
          aggregateId: snapshot.aggregateId,
          issue: 'state_mismatch',
          snapshotVersion: snapshot.version,
        });
        continue;
      }
      
      report.valid++;
      
    } catch (error) {
      report.errors.push({
        aggregateId: snapshot.aggregateId,
        issue: 'verification_error',
        error: (error as Error).message,
      });
    }
  }
  
  return report;
}
 
// Automatic snapshot regeneration for corrupted/stale snapshots
async function regenerateSnapshot(
  aggregateId: string,
  snapshotStore: SnapshotStore,
  eventStore: EventStore
): Promise<void> {
  console.log(`Regenerating snapshot for ${aggregateId}`);
  
  // Full replay to current version
  const stream = await eventStore.readStream(aggregateId);
  const state = stream.events.reduce(applyEvent, initialState);
  
  // Create new, verified snapshot
  const snapshot = createSnapshotWithChecksum(
    aggregateId,
    state,
    stream.version
  );
  
  await snapshotStore.save(snapshot);
  console.log(`Snapshot regenerated at version ${stream.version}`);
}

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

// Snapshot retention and pruning strategies
 
interface RetentionPolicy {
  // How many snapshots to keep per aggregate
  keepLatestCount: number;
  
  // Maximum age for any snapshot
  maxAgeDays: number;
  
  // Keep snapshots at specific intervals for history
  keepAtIntervals?: {
    dailyCount: number;    // Keep N daily snapshots
    weeklyCount: number;   // Keep N weekly snapshots
    monthlyCount: number;  // Keep N monthly snapshots
  };
}
 
const defaultRetentionPolicy: RetentionPolicy = {
  keepLatestCount: 2,      // Keep 2 most recent
  maxAgeDays: 90,          // Delete anything older than 90 days
};
 
// Pruning implementation
class SnapshotPruner {
  constructor(
    private snapshotStore: SnapshotStore,
    private policy: RetentionPolicy
  ) {}
  
  async pruneAggregate(aggregateId: string): Promise<PruneResult> {
    const snapshots = await this.snapshotStore.listByAggregate(
      aggregateId,
      { orderBy: 'createdAt', direction: 'desc' }
    );
    
    const toDelete: string[] = [];
    const toKeep: string[] = [];
    const now = Date.now();
    
    for (let i = 0; i < snapshots.length; i++) {
      const snapshot = snapshots[i];
      const ageInDays = (now - snapshot.createdAt.getTime()) / (1000 * 60 * 60 * 24);
      
      // Always keep the N most recent
      if (i < this.policy.keepLatestCount) {
        toKeep.push(snapshot.snapshotId);
        continue;
      }
      
      // Delete if older than max age
      if (ageInDays > this.policy.maxAgeDays) {
        toDelete.push(snapshot.snapshotId);
        continue;
      }
      
      // Otherwise delete (we only keep keepLatestCount)
      toDelete.push(snapshot.snapshotId);
    }
    
    // Perform deletions
    for (const snapshotId of toDelete) {
      await this.snapshotStore.delete(snapshotId);
    }
    
    return {
      aggregateId,
      originalCount: snapshots.length,
      deletedCount: toDelete.length,
      keptCount: toKeep.length,
    };
  }
  
  async pruneAll(): Promise<PruneSummary> {
    const summary: PruneSummary = {
      aggregatesPruned: 0,
      snapshotsDeleted: 0,
      snapshotsKept: 0,
      bytesFreed: 0,
    };
    
    const aggregateIds = await this.snapshotStore.listAggregateIds();
    
    for (const aggregateId of aggregateIds) {
      const result = await this.pruneAggregate(aggregateId);
      summary.aggregatesPruned++;
      summary.snapshotsDeleted += result.deletedCount;
      summary.snapshotsKept += result.keptCount;
    }
    
    return summary;
  }
}
 
// Scheduled pruning job
class ScheduledSnapshotPruner {
  private timer: NodeJS.Timer | null = null;
  
  constructor(
    private pruner: SnapshotPruner,
    private intervalMs: number = 24 * 60 * 60 * 1000 // Daily
  ) {}
  
  start(): void {
    console.log('Starting scheduled snapshot pruning...');
    
    // Run immediately on start
    this.runPruning();
    
    // Schedule regular runs
    this.timer = setInterval(() => this.runPruning(), this.intervalMs);
  }
  
  stop(): void {
    if (this.timer) {
      clearInterval(this.timer);
      this.timer = null;
    }
  }
  
  private async runPruning(): Promise<void> {
    console.log('Running snapshot pruning...');
    const startTime = Date.now();
    
    try {
      const summary = await this.pruner.pruneAll();
      
      console.log(
        `Snapshot pruning complete: ` +
                    `${summary.aggregatesPruned} aggregates, ` +
                  `${summary.snapshotsDeleted} deleted, ` +
                  `${summary.snapshotsKept} kept, ` +
                  `${Date.now() - startTime}ms`
      );
      
      metrics.gauge('snapshot.count', summary.snapshotsKept);
      metrics.increment('snapshot.pruned', summary.snapshotsDeleted);
      
    } catch (error) {
      console.error('Snapshot pruning failed:', error);
      metrics.increment('snapshot.pruning.error');
    }
  }
}