Projections and Read Models — Building Query Views

In event sourcing, the event stream is the write model.
You cannot query it efficiently for display purposes.

"Show all active Warfarin prescriptions for Ward 4B" requires:
  → Scanning every event stream for every prescription
  → Replaying all events to find current status
  → Filtering by medication name and ward

This is impossibly slow for read queries.

Projections solve this:
  → Listen to events as they are appended
  → Maintain a denormalised, queryable read model (a table, a document)
  → Update the read model incrementally as events arrive

Result:
  Write: append events to the event store (fast, append-only)
  Read:  query the projection table (fast, pre-computed, indexed)

The projection table is a derived view — it can always be rebuilt
by replaying all events through the projection logic.

// Simplest approach: update the read model in the same DB transaction as the event append
// Works when event store and read model are in the same database

public sealed class PrescriptionApprovedProjection
{
    private readonly DbConnection _conn;

    public async Task HandleAsync(
        PrescriptionApprovedEvent @event,
        SqlTransaction transaction,
        CancellationToken ct)
    {
        await _conn.ExecuteAsync("""
            UPDATE prescriptions.prescription_summaries
            SET    status          = 'Approved',
                   approved_at     = @approvedAt,
                   approved_by     = @approvedBy
            WHERE  id = @prescriptionId
            """,
            new
            {
                prescriptionId = @event.PrescriptionId,
                approvedAt     = @event.ApprovedAt,
                approvedBy     = @event.ApprovedBy
            },
            transaction);
    }
}

// Called immediately after appending the event, within the same transaction:
await _eventStore.AppendAsync(streamId, new[] { approvedEvent }, expectedVersion, ct);
await _projection.HandleAsync(approvedEvent, currentTransaction, ct);
await transaction.CommitAsync(ct);
// Both event and read model update are atomic.

// For projections that can be eventually consistent:
// A background worker reads new events and updates the read model

public sealed class ProjectionWorker : BackgroundService
{
    private readonly IEventStore           _eventStore;
    private readonly PrescriptionProjection _projection;
    private readonly IProjectionCheckpoint  _checkpoint;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            var lastProcessed = await _checkpoint.GetAsync("prescription-projection");

            var newEvents = await _eventStore.ReadAllFromAsync(
                fromPosition: lastProcessed + 1, stoppingToken);

            foreach (var recorded in newEvents)
            {
                await _projection.HandleAsync(recorded, stoppingToken);
                await _checkpoint.SaveAsync("prescription-projection", recorded.GlobalPosition);
            }

            // Poll for new events every 500ms
            await Task.Delay(500, stoppingToken);
        }
    }
}

// Checkpoint table: stores the last processed global event position per projection
CREATE TABLE projections.checkpoints (
    projection_name  NVARCHAR(200) NOT NULL PRIMARY KEY,
    last_position    BIGINT        NOT NULL DEFAULT -1,
    updated_at       DATETIME2     NOT NULL DEFAULT SYSUTCDATETIME()
);

// A projection handles multiple event types and maintains one read table

public sealed class PrescriptionSummaryProjection
{
    private readonly DbConnection _conn;

    public Task HandleAsync(RecordedEvent recorded, CancellationToken ct)
    {
        var @event = EventDeserializer.Deserialize(recorded);

        return @event switch
        {
            PrescriptionCreatedEvent e   => OnCreated(e, ct),
            PrescriptionApprovedEvent e  => OnApproved(e, ct),
            PrescriptionSuspendedEvent e => OnSuspended(e, ct),
            PrescriptionExpiredEvent e   => OnExpired(e, ct),
            _                           => Task.CompletedTask
        };
    }

    private Task OnCreated(PrescriptionCreatedEvent e, CancellationToken ct) =>
        _conn.ExecuteAsync("""
            INSERT INTO prescriptions.prescription_summaries
                (id, patient_id, medication_name, dose_amount, dose_unit, status, created_at)
            VALUES
                (@id, @patientId, @medicationName, @doseAmount, @doseUnit, 'Draft', @createdAt)
            """,
            new { e.Id, e.PatientId, e.MedicationName, e.DoseAmount, e.DoseUnit, e.CreatedAt });

    private Task OnApproved(PrescriptionApprovedEvent e, CancellationToken ct) =>
        _conn.ExecuteAsync("""
            UPDATE prescriptions.prescription_summaries
            SET    status = 'Approved', approved_at = @approvedAt
            WHERE  id = @id
            """,
            new { e.Id, e.ApprovedAt });
}

// When projection logic changes, rebuild from scratch by replaying all events
// Safe because the event store is immutable — events never change

public sealed class ProjectionRebuilder
{
    public async Task RebuildAsync(
        string                projectionName,
        IProjectionHandler    projection,
        CancellationToken     ct)
    {
        // 1. Truncate the read model table
        await _conn.ExecuteAsync(
            $"TRUNCATE TABLE prescriptions.prescription_summaries");

        // 2. Reset checkpoint
        await _checkpoint.ResetAsync(projectionName);

        // 3. Read all events from position 0
        long position = 0;
        await foreach (var batch in _eventStore.ReadAllInBatchesAsync(batchSize: 1000, ct))
        {
            foreach (var recorded in batch)
                await projection.HandleAsync(recorded, ct);

            await _checkpoint.SaveAsync(projectionName, batch.Last().GlobalPosition);
        }
    }
}

// Projection rebuild can be run:
//   → On startup in a migration-style command
//   → As a CLI command during deployment
//   → In a feature flag-gated background job while the old projection stays live

The same events can feed multiple independent projections:

PrescriptionCreatedEvent feeds:
  → PrescriptionSummaryProjection    (for the prescription list UI)
  → PatientPrescriptionHistoryProjection  (for the patient timeline UI)
  → PharmacyDispenseQueueProjection  (for the pharmacy dashboard)
  → MhraAuditProjection              (for regulatory reporting)

Each projection is an independent read model, optimised for its consumer.
Adding a new projection requires no changes to the write side.
The event store replays the same events through a new projection.

This is the key benefit: the write model is append-only and never changes.
New requirements only add new projections, not new columns or schema changes.