suaveplan/openspec/changes/event-sourcing-completion/specs/event-sourcing/spec.md

Capability Spec Delta: event-sourcing (event-sourcing-completion)

Package: @suaveplan/event-sourcing Status: PROPOSED

Every requirement below carries a **Verified by:** line at authoring time. (See openspec/AGENTS.md Section 13.)

ADDED Requirements

Requirement: Transactional outbox pattern

The package SHALL expose an EventOutbox interface and a MemoryEventOutbox reference implementation that lets callers persist domain events alongside aggregate writes and forward them to external consumers in a separate, retry-safe step. The interface SHALL define append(events), relay(publisher), pendingCount(), failedCount(), and cleanup(beforeIso). Outbox entries SHALL preserve insertion order, carry a per-row attempts counter, and transition through pending → published (success) or pending → failed (retry exhaustion).

Scenario: append preserves insertion order

• WHEN three events are appended via outbox.append([e1, e2, e3])
• THEN a subsequent relay(publisher) invokes publisher with the events in the same [e1, e2, e3] order

Scenario: relay marks rows published on success

• WHEN relay is called and the supplied publisher resolves
• THEN every drained row transitions from pending to published AND relay() returns the count of relayed events AND pendingCount() decreases by that count

Scenario: relay increments attempts on publisher failure

• WHEN relay is called and the publisher rejects
• THEN every drained row's attempts counter increments AND rows below the retry cap remain pending AND rows at the retry cap transition to failed AND relay() returns 0

Scenario: invalid OutboxConfig raises OUTBOX_TABLE_INIT_FAILED

• WHEN an outbox is constructed with a non-Zod-conformant OutboxConfig (e.g. negative batchSize)
• THEN the constructor throws EventSourcingError with code ES.OUTBOX_TABLE_INIT_FAILED and the offending field in context.field

Verified by: packages/events/event-sourcing/src/outbox/event-outbox.test.ts::"preserves insertion order across append and relay" AND packages/events/event-sourcing/src/outbox/event-outbox.test.ts::"marks rows published on successful relay" AND packages/events/event-sourcing/src/outbox/event-outbox.test.ts::"increments attempts on publisher failure" AND packages/events/event-sourcing/src/outbox/event-outbox.spec-conformance.test.ts::"raises OUTBOX_TABLE_INIT_FAILED on invalid config"

---

Requirement: Outbox relay worker with cancellation

The package SHALL expose an OutboxRelay worker class that drains pending outbox rows on a configurable interval (default 1000 ms) using an injectable Clock from @suaveplan/types. The worker SHALL accept an AbortSignal; when the signal aborts mid-drain, the worker SHALL reject pending publishes with EventSourcingError(ES.OUTBOX_RELAY_INTERRUPTED) and stop scheduling further iterations.

Scenario: AbortSignal terminates the relay loop

• WHEN an OutboxRelay is started with an AbortController's signal and controller.abort() fires mid-drain
• THEN the active relay() call rejects with EventSourcingError(ES.OUTBOX_RELAY_INTERRUPTED) AND no further Clock.schedule calls are made

Scenario: injected Clock controls poll interval

• WHEN an OutboxRelay is constructed with a fake Clock advancing 100 ms per tick and pollIntervalMs: 500
• THEN the relay loop fires once every five fake ticks AND OTel counter outbox.relayed increments by events.length per successful drain

Verified by: packages/events/event-sourcing/src/outbox/relay.test.ts::"rejects pending publish with OUTBOX_RELAY_INTERRUPTED when signal aborts" AND packages/events/event-sourcing/src/outbox/relay.test.ts::"polls on the injected Clock interval"

---

Requirement: Database-backed event store with optimistic concurrency

The package SHALL expose a DatabaseEventStore<TEvent extends DomainEvent> class that satisfies the EventStore interface and persists events through a DatabaseAdapter from @suaveplan/db. Every append(aggregateId, events, expectedVersion) SHALL run inside adapter.transaction(); if the current max version does not match expectedVersion, the store SHALL raise EventSourcingError(ES.CONCURRENCY_CONFLICT) and roll back the transaction. The events table SHALL be auto-created when autoCreate: true (default).

Scenario: optimistic-concurrency conflict raises CONCURRENCY_CONFLICT

• WHEN append("agg-1", events, expectedVersion: 0) is called and the stored stream is already at version 2
• THEN the call rejects with EventSourcingError(ES.CONCURRENCY_CONFLICT) AND context contains { aggregateId: "agg-1", expectedVersion: 0, currentVersion: 2 } AND no rows are inserted

Scenario: append + load round-trip

• WHEN events are appended successfully and getEvents("agg-1") is called
• THEN the returned array preserves insertion order AND each event's payload and metadata round-trip through the configured serializer

Scenario: getEventsByType honours limit

• WHEN getEventsByType("OrderPlaced", limit: 5) is called
• THEN at most 5 events are returned AND they are ordered by created_at ascending

Scenario: table auto-create gated on autoCreate flag

• WHEN the store is constructed with autoCreate: false and the underlying table does not exist
• THEN append rejects with EventSourcingError(ES.DATABASE_TABLE_INIT_FAILED) rather than silently creating the table

Verified by: packages/events/event-sourcing/src/store/database-event-store.test.ts::"raises CONCURRENCY_CONFLICT when expectedVersion does not match" AND packages/events/event-sourcing/src/store/database-event-store.test.ts::"appends and loads in stream order" AND packages/events/event-sourcing/src/store/database-event-store.test.ts::"limits getEventsByType results" AND packages/events/event-sourcing/src/store/database-event-store.spec-conformance.test.ts::"gates table creation on autoCreate flag"

---

Requirement: Database-backed snapshot store

The package SHALL expose a DatabaseSnapshotStore<TState> class that persists the latest snapshot per aggregate via a DatabaseAdapter. The store SHALL accept optional serializer / deserializer config so non-JSON state types round-trip. The shouldSnapshot(eventsSinceLastSnapshot) method SHALL return true when eventsSinceLastSnapshot >= interval.

Scenario: latest snapshot per aggregate

• WHEN two snapshots are saved for aggregateId: "agg-1" (versions 5 and 10)
• THEN load("agg-1") returns the version-10 snapshot AND a third call to save updates rather than inserts

Scenario: serializer round-trips non-JSON state

• WHEN the store is configured with a MessagePack serializer / deserializer pair and a Map-typed state is saved
• THEN load(aggregateId) returns a Map with the same entries

Scenario: delete returns false for unknown aggregate

• WHEN delete("never-saved") is called
• THEN the result is false AND no SQL DELETE is dispatched

Verified by: packages/events/event-sourcing/src/store/database-snapshot-store.test.ts::"returns the latest snapshot per aggregate" AND packages/events/event-sourcing/src/store/database-snapshot-store.test.ts::"round-trips non-JSON state via injected serializer" AND packages/events/event-sourcing/src/store/database-snapshot-store.test.ts::"returns false when deleting an unknown aggregate"

---

Requirement: Event replayer with progress / pause / resume / cancel

The package SHALL expose an EventReplayer class that replays events from an EventStore through a handler with batch processing, progress callbacks, pause / resume / cancel control, and AbortSignal support. ReplayFilter SHALL narrow by eventTypes, aggregateId, fromDate, toDate. ReplayProgress SHALL expose total, processed, percentage, paused, completed. The terminal onProgress tick SHALL carry completed: true.

Scenario: replay invokes handler in stream order

• WHEN replay("agg-1", handler) is called against a stream of 100 events
• THEN handler is invoked exactly 100 times in stream order AND the final onProgress tick has processed: 100, percentage: 100, completed: true

Scenario: filter narrows by eventTypes

• WHEN replay("agg-1", handler, { eventTypes: ["OrderPlaced"] }) is called against a stream that mixes OrderPlaced and OrderShipped
• THEN handler is invoked only for OrderPlaced events AND total reflects the filtered count

Scenario: pause halts processing until resume

• WHEN pause() is called mid-replay and resume() is called 50 ms later
• THEN the handler stops being invoked between pause and resume AND a paused: true progress tick is emitted at the pause boundary

Scenario: AbortSignal mid-batch rejects with REPLAY_CANCELLED

• WHEN replay("agg-1", handler, undefined, signal) is in flight and the signal aborts
• THEN the call rejects with EventSourcingError(ES.REPLAY_CANCELLED) AND in-flight handler invocations are not awaited past the abort point

Scenario: invalid filter rejected up front

• WHEN replay("agg-1", handler, { fromDate: "not-an-iso-date" }) is called
• THEN the call rejects with EventSourcingError(ES.REPLAY_FILTER_INVALID) and context.fromDate is the offending value

Verified by: packages/events/event-sourcing/src/replay/event-replayer.test.ts::"invokes handler in stream order across batches" AND packages/events/event-sourcing/src/replay/event-replayer.test.ts::"narrows replayed events by eventTypes filter" AND packages/events/event-sourcing/src/replay/event-replayer.test.ts::"halts processing between pause and resume" AND packages/events/event-sourcing/src/replay/event-replayer.spec-conformance.test.ts::"rejects with REPLAY_CANCELLED when signal aborts mid-batch" AND packages/events/event-sourcing/src/replay/event-replayer.spec-conformance.test.ts::"rejects invalid filter with REPLAY_FILTER_INVALID"

---

Requirement: Event bus with subscription filter and dead-letter sink

The package SHALL expose an EventBus class implementing typed publish/subscribe over DomainEvent. subscribe(handler, options?) SHALL accept a SubscriptionFilter (eventTypes, aggregateIds, aggregateType) and an optional group string; multiple subscribers in the same group receive events in round-robin order, while ungrouped subscribers all receive every matching event. Handler exceptions SHALL be routed to a DeadLetterSink with the original cause preserved on EventSourcingError(ES.EVENT_HANDLER_FAILED). Subscription IDs SHALL come from an injectable IdGenerator so tests can pin them.

Scenario: subscribe and publish round-trip

• WHEN a handler subscribes without a filter and an event is published
• THEN the handler is invoked exactly once with the published event

Scenario: filter narrows by aggregateType

• WHEN a subscriber's filter sets aggregateType: "Order" and an Account event is published
• THEN the handler is not invoked AND subscriberCount is unchanged

Scenario: group round-robin distribution

• WHEN three subscribers join group "workers" and four events are published
• THEN subscriber A handles events 1 and 4, subscriber B handles event 2, subscriber C handles event 3

Scenario: handler exception routes to DeadLetterSink

• WHEN a subscribed handler throws synchronously and a MemoryDeadLetterSink is configured
• THEN the sink captures the failed event AND the captured entry's cause.code === ES.EVENT_HANDLER_FAILED AND cause.context.subscriptionId matches the failing subscription

Scenario: unsubscribe removes record

• WHEN subscription.unsubscribe() is called
• THEN subscription.active === false AND subscriberCount decreases by 1 AND subsequent publishes do not invoke the handler

Verified by: packages/events/event-sourcing/src/subscription/event-bus.test.ts::"invokes handler exactly once on publish without filter" AND packages/events/event-sourcing/src/subscription/event-bus.test.ts::"narrows delivery by aggregateType filter" AND packages/events/event-sourcing/src/subscription/event-bus.test.ts::"distributes events round-robin within a group" AND packages/events/event-sourcing/src/subscription/event-bus.spec-conformance.test.ts::"routes handler exceptions to DeadLetterSink with EVENT_HANDLER_FAILED cause" AND packages/events/event-sourcing/src/subscription/event-bus.test.ts::"removes record when unsubscribe is called" AND packages/events/event-sourcing/src/subscription/dead-letter.test.ts::"captures failed events in insertion order"

---

Requirement: Snapshot manager with reducer and pruning

The package SHALL expose a SnapshotManager<TState> class that automates snapshot creation at a configurable interval and prunes per-aggregate history to maxSnapshots. load(aggregateId, initialState, reducer) SHALL fetch the latest snapshot, replay tail events through reducer, and return an AggregateLoadResult<TState> carrying state, version, restoredFromSnapshot, eventsReplayed. takenAt timestamps SHALL come from an injectable Clock.

Scenario: load rebuilds state from snapshot plus tail events

• WHEN a snapshot exists at version 10 and 3 newer events are in the store
• THEN load(...) invokes the reducer 3 times (not 13) AND result.eventsReplayed === 3 AND result.restoredFromSnapshot === true

Scenario: auto-create snapshot at configured interval

• WHEN load replays >= interval events past the last snapshot
• THEN SnapshotManager calls snapshotStore.save with the rebuilt state AND a snapshot.created lifecycle event fires AND getSnapshotHistory(aggregateId) includes the new entry

Scenario: pruning enforces maxSnapshots

• WHEN snapshots maxSnapshots + 2 exist for one aggregate
• THEN the oldest 2 are pruned from the in-memory history AND getSnapshotHistory(aggregateId).length === maxSnapshots

Scenario: reducer exception raises SNAPSHOT_HANDLER_FAILED

• WHEN the supplied reducer throws while replaying tail events
• THEN load rejects with EventSourcingError(ES.SNAPSHOT_HANDLER_FAILED) AND cause is the original exception

Scenario: injected Clock stamps takenAt deterministically

• WHEN a fake Clock returning a fixed instant is injected and a snapshot is created
• THEN the saved snapshot's takenAt equals the fake instant

Verified by: packages/events/event-sourcing/src/snapshot/snapshot-manager.test.ts::"rebuilds state from snapshot plus tail events" AND packages/events/event-sourcing/src/snapshot/snapshot-manager.test.ts::"auto-creates snapshot at the configured interval" AND packages/events/event-sourcing/src/snapshot/snapshot-manager.test.ts::"prunes per-aggregate history to maxSnapshots" AND packages/events/event-sourcing/src/snapshot/snapshot-manager.spec-conformance.test.ts::"raises SNAPSHOT_HANDLER_FAILED when reducer throws" AND packages/events/event-sourcing/src/snapshot/snapshot-manager.spec-conformance.test.ts::"stamps takenAt from the injected Clock"

---

Requirement: Versioned upcaster chain with schema-version detection

The package SHALL expose an UpcasterChain class that holds an ordered registry of EventUpcaster entries and applies them in fromVersion order during upcast(event). detectSchemaVersion(event) SHALL read metadata.schemaVersion, defaulting to 1 when absent. createUpcastedEvent and createRenamedEvent SHALL stamp schemaVersion: targetVersion on the returned event's metadata. createFieldRenameUpcaster, createDefaultFieldUpcaster, and createRemoveFieldUpcaster SHALL produce upcasters that handle a single (eventType, fromVersion) step. Duplicate (eventType, fromVersion) registrations SHALL raise EventSourcingError(ES.UPCASTER_DUPLICATE).

Scenario: chain walks v1 to v3 in order

• WHEN upcasters for (OrderPlaced, 1→2) and (OrderPlaced, 2→3) are registered out of order
• THEN chain.upcast(v1Event) returns a v3 event AND the upcasters fire in fromVersion order regardless of registration order

Scenario: detectSchemaVersion defaults to 1

• WHEN an event has no metadata.schemaVersion
• THEN detectSchemaVersion(event) returns 1

Scenario: getUpcastersForType filters by eventType

• WHEN the chain holds upcasters for OrderPlaced and OrderShipped
• THEN getUpcastersForType("OrderPlaced") returns only the OrderPlaced entries in fromVersion order

Scenario: createFieldRenameUpcaster round-trip

• WHEN createFieldRenameUpcaster("OrderPlaced", 1, 2, { customer: "customerId" }) is registered and applied to a v1 payload { customer: "c-1", total: 100 }
• THEN the upcasted payload is { customerId: "c-1", total: 100 } AND metadata.schemaVersion === 2

Scenario: duplicate registration rejected

• WHEN two upcasters are registered with the same (eventType, fromVersion) pair
• THEN the second register() call raises EventSourcingError(ES.UPCASTER_DUPLICATE) and the chain remains at one entry for that key

Verified by: packages/events/event-sourcing/src/upcasting/chain.test.ts::"walks the chain in fromVersion order regardless of registration order" AND packages/events/event-sourcing/src/upcasting/chain.spec-conformance.test.ts::"detectSchemaVersion defaults to 1 when metadata is absent" AND packages/events/event-sourcing/src/upcasting/chain.test.ts::"filters upcasters by eventType" AND packages/events/event-sourcing/src/upcasting/chain.spec-conformance.test.ts::"createFieldRenameUpcaster renames the field and stamps schemaVersion" AND packages/events/event-sourcing/src/upcasting/chain.spec-conformance.test.ts::"raises UPCASTER_DUPLICATE on collision"

---

Requirement: Backwards-compatible flat upcaster API delegates to UpcasterChain

The package SHALL preserve the existing flat registerUpcaster(options), clearUpcasters(), and upcastEvents(events) exports. These functions SHALL delegate to a single module-scoped UpcasterChain instance so consumers calling either API observe the same registry.

Scenario: flat register and class chain see the same upcasters

• WHEN registerUpcaster({ eventType: "OrderPlaced", fromVersion: 1, toVersion: 2, upcast }) is called
• THEN the module-scoped UpcasterChain.getUpcasters() includes that entry AND a subsequent upcastEvents([v1Event]) invocation produces the same v2 result as chain.upcastAll([v1Event])

Scenario: clearUpcasters resets the shared chain

• WHEN clearUpcasters() is called
• THEN both upcastEvents([anyEvent]) and chain.getUpcasters() reflect an empty registry

Verified by: packages/events/event-sourcing/src/upcasting/chain.test.ts::"flat registerUpcaster delegates to the module chain" AND packages/events/event-sourcing/src/upcasting/chain.test.ts::"clearUpcasters empties the shared chain"