Projections & Subscriptions

The write side (aggregates) and the read side (projections) are separate. The write side is optimized for consistency: one aggregate, one stream, one transaction. The read side is optimized for queries: denormalized, shaped for specific use cases, eventually consistent.

What is a projection?

A projection subscribes to an event stream and builds a read model. It's a function:

The same events that an aggregate emits can feed dozens of independent projections, each building a different view:

Document store projection

The most common pattern: one read-model document per aggregate.

public class BookReadModel : IDocument
{
    public string Id { get; set; }
    public string Title { get; set; }
    public string Author { get; set; }
    public bool IsBorrowed { get; set; }
    public string? BorrowerId { get; set; }
}

public class BookProjection : SqlDocumentStoreProjection<BookEvent, BookReadModel>
{
    public BookProjection(DbConnection connection, ICheckpointStore checkpointStore)
        : base(connection, checkpointStore) { }

    public override SubscriptionId SubscriptionId => new("book-projection");

    protected override BookReadModel Apply(
        BookReadModel? current, EventEnvelope<BookEvent> envelope)
    {
        var doc = current ?? new BookReadModel { Id = envelope.AggregateId.Value };

        return envelope.Event switch
        {
            BookAdded e => doc with { Title = e.Title, Author = e.Author },
            BookBorrowed e => doc with { IsBorrowed = true, BorrowerId = e.BorrowerId },
            BookReturned _ => doc with { IsBorrowed = false, BorrowerId = null },
            _ => doc
        };
    }
}

Apply receives the current document (or null for a new aggregate) and returns the updated document. Loading, saving, and checkpoint tracking are handled by the framework.

Enrichment projection

Sometimes a projection also needs to emit integration events to notify other systems:

public class BookIntegrationProjection
    : SqlDocumentEnrichmentProjection<BookEvent, BookReadModel>
{
    public BookIntegrationProjection(
        DbConnection connection,
        ICheckpointStore checkpointStore,
        IEventEmitter emitter)
        : base(connection, checkpointStore, emitter) { }

    public override SubscriptionId SubscriptionId => new("book-integration");

    protected override (BookReadModel, IEnumerable<object>?) Apply(
        BookReadModel? current, EventEnvelope<BookEvent> envelope)
    {
        var doc = current ?? new BookReadModel { Id = envelope.AggregateId.Value };

        return envelope.Event switch
        {
            BookBorrowed e => (
                doc with { IsBorrowed = true },
                new[] { new BookBorrowedIntegrationEvent(doc.Title, e.BorrowerId) }),
            _ => (doc, null)
        };
    }
}

The enrichment projection reads domain events and produces integration events that other bounded contexts consume. See modular monolith for where this fits in the architecture.

Custom subscriptions

For use cases that do not fit document-per-aggregate (sending emails, calling external APIs, updating search indexes), implement ISubscription<TEvent> directly:

public class OrderNotificationSubscription : ISubscription<OrderEvent>
{
    public SubscriptionId SubscriptionId => new("order-notifications");

    public Task Prepare() => Task.CompletedTask;

    public async Task Handle(EventEnvelope<OrderEvent> evt)
    {
        if (evt.Event is OrderPlaced placed)
        {
            await _emailService.SendOrderConfirmation(
                evt.AggregateId.Value, placed.ProductId);
        }
    }
}

Custom subscriptions still get checkpointing. The framework tracks the position after each successful Handle call and persists it according to the BatchSize in your subscription options. If the service restarts, it resumes from the last checkpoint. You do not need to manage checkpoints yourself.

Registration

builder.Services.AddSubscription<BookProjection, BookEvent>(options =>
{
    options.BatchSize = 50;
    options.PollDelay = TimeSpan.FromMilliseconds(200);
    options.MaxErrorDelay = TimeSpan.FromSeconds(60);
});

Subscriptions run as IHostedService. They start with the application and process events continuously in the background.

Eventual consistency

Projections are eventually consistent. After an event is appended, there is a brief delay (typically milliseconds, controlled by PollDelay) before projections process it.

This means projections can run on different nodes, a broken projection does not affect writes, you can reset a checkpoint and rebuild from scratch, and you can add a projection years later and backfill from history.

If you need immediate read-after-write consistency, read from the aggregate directly.

Checkpoints

Each subscription tracks its position in the event stream via a checkpoint: the position of the last event it successfully processed. On restart, it resumes from that checkpoint rather than replaying the entire stream.

// SQLite
services.AddSingleton<ICheckpointStore>(sp =>
    new SqliteCheckpointStore(sp.GetRequiredService<DbConnection>()));

// SQL Server
services.AddSingleton<ICheckpointStore>(sp =>
    new MsSqlCheckpointStore(sp.GetRequiredService<DbConnection>()));

// PostgreSQL
services.AddSingleton<ICheckpointStore>(sp =>
    new PostgresCheckpointStore(sp.GetRequiredService<DbConnection>()));

To rebuild a projection, delete its checkpoint. On next startup, it processes every event from the beginning and reconstructs the read model from scratch.

Distributed locking

In multi-instance deployments where the same subscription runs on multiple nodes, you need to ensure only one instance processes events at a time. Nagare provides lock providers for this:

// SQL Server advisory lock
services.AddSingleton<ILockProvider>(sp =>
    new MsSqlLockProvider(sp.GetRequiredService<DbConnection>()));

// PostgreSQL advisory lock
services.AddSingleton<ILockProvider>(sp =>
    new PostgresLockProvider(sp.GetRequiredService<DbConnection>()));

Without a lock provider, Nagare uses NoopLockProvider, which is fine for single-instance deployments.

Rebuilding projections

Projections are disposable. When the schema changes, a bug is fixed, or a new field is added, rebuild from scratch.

If your projection implements IRebuildableProjection (the built-in document store projections do), the framework handles this:

var rebuilder = serviceProvider.GetRequiredService<ProjectionRebuilder>();
await rebuilder.Rebuild<BookProjection, BookEvent>();

This truncates the projection's data, resets its checkpoint to the beginning, and the subscription runner reprocesses every event from position zero. The projection rebuilds itself.

For custom subscriptions, implement IRebuildableProjection:

public class SearchIndexProjection : ISubscription<BookEvent>, IRebuildableProjection
{
    public async Task Truncate()
    {
        await _searchClient.DeleteIndex("books");
        await _searchClient.CreateIndex("books");
    }

    // ... Handle, Prepare, etc.
}

Dead letter handling

By default, subscriptions retry forever. If an event causes a handler to fail, the subscription backs off exponentially and keeps trying. No events are ever skipped.

For subscriptions where skipping a poisoned event is acceptable, enable dead-letter handling:

builder.Services.AddSubscription<BookProjection, BookEvent>(options =>
{
    options.DeadLetter = new DeadLetterOptions
    {
        MaxRetries = 5,
        InitialRetryDelay = TimeSpan.FromSeconds(1),
        MaxRetryDelay = TimeSpan.FromSeconds(30)
    };
});

After MaxRetries failures, the event is stored in the dead letter store and the subscription advances past it. The dead letter store records the subscription ID, position, event type, payload, error, and retry count.

You also need to register a dead letter store implementation:

// In-memory (for tests)
builder.Services.AddSingleton<IDeadLetterStore>(new InMemoryDeadLetterStore());

Use dead-letter handling for read-model projections where a single malformed event should not block all downstream processing. A dashboard projection can tolerate a gap.

Do not use it for subscriptions that synchronize with external systems where idempotency depends on processing every event in order. If you are syncing inventory to a warehouse system, skipping an event means the systems diverge silently. Keep the default retry-forever behavior.

Live subscriptions (tap mode)

Not all subscriptions need the full event history. For notifications, SSE broadcasting, and real-time side effects, you want to start from "now" and only process new events.

// Live subscription — skips all existing events, only processes new ones
builder.Services.AddLiveSubscription<NotificationHandler, VisitEvent>();

When the subscription starts and has no saved checkpoint, it jumps to the current head position in the event store and begins processing from there. Events that were appended before the subscription started are never delivered.

Use this for:

• Push notifications to mobile apps
• SSE/WebSocket event broadcasting
• Real-time alerting and dashboards
• Audit event forwarding to external systems

Do not use for:

• Read model projections (they need full history to build their state)
• Integration event publishing (consumers may need to replay)
• Any subscription where missing historical events would cause incorrect state

How it works

Live subscriptions use a separate LiveSubscriptionRunner that is fundamentally different from the regular SubscriptionRunner:

• No checkpoint store — position is tracked in memory only. The runner doesn't even accept an ICheckpointStore.
• No dead letters — failed events are logged and skipped. Acceptable for notifications; not for projections.
• No resume — on restart, the runner starts from the current head again. Events between shutdown and restart are not replayed.
• No history — existing events in the journal are skipped entirely on first start.

Configuring live subscriptions

AddLiveSubscription accepts the same SubscriptionOptions for BatchSize and PollDelay:

builder.Services.AddLiveSubscription<AlertHandler, VisitEvent>(options =>
{
    options.BatchSize = 100;
    options.PollDelay = TimeSpan.FromMilliseconds(50);
});

On PostgreSQL, use AddPostgresLiveSubscription which combines the live runner with LISTEN/NOTIFY for sub-10ms event-to-notification latency:

// PostgreSQL: live subscription + LISTEN/NOTIFY = near-instant delivery
builder.Services.AddPostgresLiveSubscription<NotificationHandler, VisitEvent>();

See store-adapters § PostgreSQL-specific features for details on LISTEN/NOTIFY.

Design guidelines

1. One projection, one purpose. Do not try to serve the dashboard and the search index from the same projection.

2. Projections are disposable. They can be deleted and rebuilt at any time.

3. Handle all events. Even events that do not affect your read model still advance the checkpoint. The _ => doc catch-all handles this.

4. Use the envelope. EventEnvelope<TEvent> gives you AggregateId, Position, Version, CreatedAt, and Metadata beyond just the event itself.

5. Keep projections simple. If a projection needs complex logic or external API calls, it is doing too much. Split it, or move the complexity into a custom subscription.

6. Idempotency is built-in. If a projection crashes mid-batch, it restarts from the last checkpoint and reprocesses. The document store upsert handles deduplication.

---

Next: Read Models