Event Store

Verified by tests

AppendAndWaitEventStoreDecoratorTests, SecurityContextEventStoreDecoratorTests, SyncTrackingEventStoreDecoratorTests, InMemoryEventStoreTests, EventStoreAppendBatchTests — library CI run #31657041675 (2026-08-13)

The IEventStore interface provides append-only event storage for event sourcing patterns. It supports stream-based storage with automatic sequence numbering, polymorphic reads, and synchronous verification via AppendAndWaitAsync.

Overview

Whizbang's event store is designed for:

• Append-only storage - Events are immutable once written
• Stream-based organization - Events grouped by aggregate ID
• Polymorphic reads - Read multiple event types from a stream
• AOT compatibility - Generic methods avoid reflection
• Sync verification - Wait for perspectives to process events

Basic Usage

Appending Events

Appending Events

public class OrderHandler {
  private readonly IEventStore _eventStore;

  public async Task CreateOrder(CreateOrderCommand cmd) {
    var orderId = Guid.NewGuid();
    var evt = new OrderCreatedEvent(orderId, cmd.CustomerId, cmd.Items);

    // Append to stream identified by orderId
    await _eventStore.AppendAsync(orderId, evt);
  }
}

Reading Events

Reading Events

// Read all events of a specific type
await foreach (var envelope in _eventStore.ReadAsync<OrderCreatedEvent>(orderId, fromSequence: 0)) {
  var evt = envelope.Payload;
  // Process event
}

// Read from a specific event ID
await foreach (var envelope in _eventStore.ReadAsync<OrderEvent>(orderId, fromEventId: lastProcessedId)) {
  // Process events after the checkpoint
}

AppendAndWaitAsync

The AppendAndWaitAsync method appends an event and waits for a perspective to process it. This enables synchronous-feeling APIs over event sourcing.

AppendAndWaitEventStoreDecorator

The AppendAndWaitAsync functionality is provided by the AppendAndWaitEventStoreDecorator, which wraps the base event store implementation. This decorator:

• Waits for a specific perspective via IPerspectiveSyncAwaiter
• Waits for ALL perspectives via IEventCompletionAwaiter (single-type-parameter overload)
• Times out gracefully if perspective processing takes too long
• Returns sync results with timing and event count details

The decorator is automatically applied when using DecorateEventStoreWithSyncTracking(), which is called by data providers during registration.

Usage

Usage

var syncResult = await _eventStore.AppendAndWaitAsync<OrderCreatedEvent, OrderProjection>(
    streamId: orderId,
    message: new OrderCreatedEvent(orderId, customerId, items),
    timeout: TimeSpan.FromSeconds(10));

// Event has been appended AND OrderProjection has processed it

Parameters

Parameter	Type	Description
streamId	Guid	The stream/aggregate ID
message	TMessage	The event to append
timeout	TimeSpan?	Maximum wait time (default: 30 seconds)
onWaiting	Action<SyncWaitingContext>?	Optional callback invoked when waiting begins (only if there are events to wait for)
onDecisionMade	Action<SyncDecisionContext>?	Optional callback always invoked when the sync decision is made, regardless of outcome
cancellationToken	CancellationToken	Cancellation token

Waiting for All Perspectives

The single-type-parameter overload AppendAndWaitAsync<TMessage> waits for all registered perspectives to process the event, rather than a specific one:

Wait for All Perspectives

var syncResult = await _eventStore.AppendAndWaitAsync(
    streamId: orderId,
    message: new OrderCreatedEvent(orderId, customerId, items),
    timeout: TimeSpan.FromSeconds(10));

// Event has been appended AND all perspectives have processed it

This uses IEventCompletionAwaiter internally to wait for all perspectives.

SyncResult

Returns a SyncResult with the outcome:

SyncResult

var result = await eventStore.AppendAndWaitAsync<OrderCreatedEvent, OrderProjection>(
    streamId, evt);

switch (result.Outcome) {
  case SyncOutcome.Synced:
    Console.WriteLine($"Synced {result.EventsAwaited} events in {result.ElapsedTime}");
    break;
  case SyncOutcome.TimedOut:
    Console.WriteLine("Event appended but perspective sync timed out");
    break;
  case SyncOutcome.NoPendingEvents:
    Console.WriteLine("No events to process");
    break;
}

When to Use

Use AppendAndWaitAsync when:

• You need to verify a specific projection processed an event before returning
• Implementing request-response patterns over event sourcing
• Building APIs that need immediate consistency with a particular read model

For higher-level patterns that wait for ALL perspectives, see LocalInvokeAndSyncAsync.

Security Context Propagation

When appending events with the message-only overload (AppendAsync<TMessage>(streamId, message)), the event store automatically propagates security context from the ambient scope.

This happens via the SecurityContextEventStoreDecorator:

Security Context Propagation

// Security context from ScopeContextAccessor.CurrentContext is auto-propagated
await _eventStore.AppendAsync(orderId, new OrderCreatedEvent(...));

// The envelope will contain SecurityContext with UserId and TenantId

This ensures: - Audit trails - Events record who performed the action - Multi-tenancy - Events are tagged with tenant ID - Traceability - Security context flows through the event chain

When Context is Propagated

Security context is propagated when: 1. ScopeContextAccessor.CurrentContext contains an ImmutableScopeContext 2. The context has ShouldPropagate = true

When Context is Propagated

// Context is set by middleware or scope initialization
var extraction = new SecurityExtraction {
  Scope = new PerspectiveScope { UserId = "user-123", TenantId = "tenant-456" },
  // ...
};
ScopeContextAccessor.CurrentContext = new ImmutableScopeContext(extraction, shouldPropagate: true);

// Events now include this security context
await _eventStore.AppendAsync(orderId, evt);

IEventStore Interface

IEventStore Interface

public interface IEventStore {
  // Append with envelope (full control)
  Task AppendAsync<TMessage>(Guid streamId, MessageEnvelope<TMessage> envelope, CancellationToken ct = default);

  // Append with message (auto-creates envelope with security context)
  Task AppendAsync<TMessage>(Guid streamId, TMessage message, CancellationToken ct = default)
      where TMessage : notnull;

  // Append and wait for specific perspective sync
  Task<SyncResult> AppendAndWaitAsync<TMessage, TPerspective>(
      Guid streamId, TMessage message, TimeSpan? timeout = null,
      Action<SyncWaitingContext>? onWaiting = null,
      Action<SyncDecisionContext>? onDecisionMade = null,
      CancellationToken ct = default)
      where TMessage : notnull
      where TPerspective : class;

  // Append and wait for ALL perspectives to process
  Task<SyncResult> AppendAndWaitAsync<TMessage>(
      Guid streamId, TMessage message, TimeSpan? timeout = null,
      Action<SyncWaitingContext>? onWaiting = null,
      Action<SyncDecisionContext>? onDecisionMade = null,
      CancellationToken ct = default)
      where TMessage : notnull;

  // Read by sequence number
  IAsyncEnumerable<MessageEnvelope<TMessage>> ReadAsync<TMessage>(
      Guid streamId, long fromSequence, CancellationToken ct = default);

  // Read after event ID (checkpoint-based)
  IAsyncEnumerable<MessageEnvelope<TMessage>> ReadAsync<TMessage>(
      Guid streamId, Guid? fromEventId, CancellationToken ct = default);

  // Read multiple event types
  IAsyncEnumerable<MessageEnvelope<IEvent>> ReadPolymorphicAsync(
      Guid streamId, Guid? fromEventId, IReadOnlyList<Type> eventTypes, CancellationToken ct = default);

  // Get events between checkpoints (single type)
  Task<List<MessageEnvelope<TMessage>>> GetEventsBetweenAsync<TMessage>(
      Guid streamId, Guid? afterEventId, Guid upToEventId, CancellationToken ct = default);

  // Get events between checkpoints (polymorphic - multiple event types)
  Task<List<MessageEnvelope<IEvent>>> GetEventsBetweenPolymorphicAsync(
      Guid streamId, Guid? afterEventId, Guid upToEventId,
      IReadOnlyList<Type> eventTypes, CancellationToken ct = default);

  // Get last sequence number (-1 if the stream is empty)
  Task<long> GetLastSequenceAsync(Guid streamId, CancellationToken ct = default);

  // Batch append across (potentially different) streams.
  // Default implementation loops AppendAsync serially; backends override for bulk performance.
  Task AppendBatchAsync<TMessage>(
      IReadOnlyList<(Guid streamId, MessageEnvelope<TMessage> envelope)> entries,
      CancellationToken ct = default);

  // Resolve the commit_sequence stamped on an event (null if not stamped / not tracked)
  Task<long?> GetCommitSequenceAsync(Guid eventId, CancellationToken ct = default);

  // Deserialize raw stream event data (drain mode) into typed envelopes
  List<MessageEnvelope<IEvent>> DeserializeStreamEvents(
      IReadOnlyList<StreamEventData> streamEvents,
      IReadOnlyList<Type> eventTypes);
}

Decorator Stack

Whizbang applies decorators to enhance event store functionality:

graph TB
    A["IEventStore (your code calls this)"]
    B["AppendAndWaitEventStoreDecorator (enables AppendAndWaitAsync)"]
    C["SyncTrackingEventStoreDecorator (tracks events for sync)"]
    D["SecurityContextEventStoreDecorator (propagates security context)"]
    E["UpcastingEventStoreDecorator (transforms stored events on read)"]
    F["Base IEventStore (e.g., EFCoreEventStore, DapperEventStore)"]

    A --> B --> C --> D --> E --> F

    style A fill:#d4edda,stroke:#28a745
    style F fill:#fff3cd,stroke:#ffc107

The upcasting layer only participates when upcasters are registered — see Event Upcasting.

These decorators are automatically applied when using DecorateEventStoreWithSyncTracking(), which is called by data providers.

Registration

Event stores are registered by data providers:

Registration

services.AddWhizbang()
    .WithEFCore<MyDbContext>()
    .WithDriver.Postgres;  // Registers EFCoreEventStore with decorators

The decorators are applied automatically by DecorateEventStoreWithSyncTracking().

Related Documentation

• Dispatcher - LocalInvokeAndSyncAsync for request-response patterns
• Perspectives - Read model projections
• Event Store Query - Querying events
• Message Security - Security context