Rickten.Projector 1.3.0

Rickten.Projector

Lightweight library for building event-sourced projections (read models) with declarative filtering and checkpoint management.

Features

• ✅ Simple Projection Interface - IProjection<TView> for building read models
• ✅ Declarative Filtering - [Projection] attribute with aggregate and event type filters
• ✅ Efficient Queries - Leverages IEventStore.LoadAllAsync filtering
• ✅ Checkpoint Management - Automatic catch-up from last processed version
• ✅ Full Event Context - Access to StreamEvent with metadata in projections
• ✅ Flexible Checkpointing - Manual rebuild or automatic catch-up modes
• ✅ Filter Validation - Runtime validation that received events match configured filters

Installation

dotnet add package Rickten.Projector

Quick Start

1. Define Your View Model

public record ActiveSessionsView
{
    public HashSet<string> ActiveSessions { get; init; } = [];
    public int TotalStarted { get; init; }
    public int TotalCompleted { get; init; }
}

2. Create a Projection

[Projection("ActiveSessions",
    AggregateTypes = ["SessionReview"],
    EventTypes = ["SessionReview.SessionStarted.v1", "SessionReview.SessionCompleted.v1", "SessionReview.SessionCancelled.v1"])]
public class ActiveSessionsProjection : Projection<ActiveSessionsView>
{
    public override ActiveSessionsView InitialView() => new();

    protected override ActiveSessionsView ApplyEvent(
        ActiveSessionsView view,
        StreamEvent streamEvent)
    {
        return streamEvent.Event switch
        {
            SessionReviewEvent.SessionStarted started => view with
            {
                ActiveSessions = view.ActiveSessions.Add(started.SessionId),
                TotalStarted = view.TotalStarted + 1
            },

            SessionReviewEvent.SessionCompleted completed => view with
            {
                ActiveSessions = view.ActiveSessions.Remove(completed.SessionId),
                TotalCompleted = view.TotalCompleted + 1
            },

            SessionReviewEvent.SessionCancelled cancelled => view with
            {
                ActiveSessions = view.ActiveSessions.Remove(cancelled.SessionId)
            },

            _ => view
        };
    }
}

3. Project Events

Option A: Manual Rebuild (from scratch or specific version)

var eventStore = ...; // IEventStore
var projection = new ActiveSessionsProjection();

// Rebuild from beginning
var (view, lastVersion) = await ProjectionRunner.RebuildAsync(
    eventStore,
    projection);

Console.WriteLine($"Active sessions: {view.ActiveSessions.Count}");
Console.WriteLine($"Last version: {lastVersion}");

Option B: Catch-Up with Checkpoints (automatic checkpoint management)

var eventStore = ...; // IEventStore
var projectionStore = ...; // IProjectionStore
var projection = new ActiveSessionsProjection();

// Loads last checkpoint, processes new events, saves updated checkpoint
var (view, version) = await ProjectionRunner.CatchUpAsync(
    eventStore,
    projectionStore,
    projection);

Console.WriteLine($"Active sessions: {view.ActiveSessions.Count}");
Console.WriteLine($"Checkpoint version: {version}");

Core Concepts

IProjection<TView>

The fundamental interface for projections:

public interface IProjection<TView>
{
    TView InitialView();
    TView Apply(TView view, StreamEvent streamEvent);
}

Projection<TView> Base Class

Abstract base class with attribute-based filtering:

public abstract class Projection<TView> : IProjection<TView>
{
    public string ProjectionName { get; }
    public string[]? AggregateTypeFilter { get; }
    public string[]? EventTypeFilter { get; }

    public abstract TView InitialView();
    protected abstract TView ApplyEvent(TView view, StreamEvent streamEvent);
}

[Projection] Attribute

Optional attribute for metadata and filtering:

[Projection("ProjectionName",
    AggregateTypes = ["Aggregate1", "Aggregate2"],
    EventTypes = ["Aggregate1.Event1.v1", "Aggregate2.Event2.v1"],
    Description = "What this projection does")]

Properties:

• Name (required) - Projection identifier for checkpointing
• AggregateTypes (optional) - Filter by aggregate types (uses IEventStore.LoadAllAsync)
• EventTypes (optional) - Filter by event types using wire-name format: {Aggregate}.{Name}.v{Version} (uses IEventStore.LoadAllAsync)
• Description (optional) - Documentation

Filter Behavior:

• Filters are passed to IEventStore.LoadAllAsync for efficient querying
• EventTypes must match stored wire names from [Event] attributes, not short event names
• Runtime validation ensures received events match filters
• Throws InvalidOperationException if mismatch detected
• null filters mean "all events"

ProjectionRunner

Static utility methods for projection operations:

RebuildAsync - Rebuild projection from scratch:

public static Task<(TView View, long LastGlobalPosition)> RebuildAsync<TView>(
    IEventStore eventStore,
    IProjection<TView> projection,
    long fromGlobalPosition = 0,
    CancellationToken cancellationToken = default);

CatchUpAsync - Automatic checkpoint management:

public static Task<(TView View, long GlobalPosition)> CatchUpAsync<TView>(
    IEventStore eventStore,
    IProjectionStore projectionStore,
    IProjection<TView> projection,
    string? projectionName = null,
    CancellationToken cancellationToken = default);

Access to Event Metadata

Projections receive the full StreamEvent with metadata:

protected override MyView ApplyEvent(MyView view, StreamEvent streamEvent)
{
    // Access event
    var @event = streamEvent.Event;

    // Access stream information
    var streamType = streamEvent.StreamPointer.Stream.StreamType;
    var version = streamEvent.StreamPointer.Version;

    // Access metadata
    var metadata = streamEvent.Metadata;
    var correlationId = metadata.FirstOrDefault(m => m.Key == "CorrelationId")?.Value;

    // Use metadata in projection logic
    return view with { /* ... */ };
}

Design Principles

1. Projection-Controlled - Each projection decides how to use metadata and checkpoints
2. Efficient Filtering - Leverage store-level filtering via [Projection] attribute
3. Simple Mechanics - Just rebuild or catch-up, no background services
4. Flexible Checkpointing - Manual or automatic, projection decides
5. Filter Validation - Runtime checks ensure query/filter consistency

Error Handling

• Filter Mismatch: Throws InvalidOperationException if received event doesn't match filters
• Missing Projection Name: Throws ArgumentException in CatchUpAsync if name not provided
• Event Processing: Projection code handles event-specific errors

When to Use Each Mode

Use RebuildAsync when:

• Building projections for the first time
• Need to rebuild from scratch (data corruption, schema change)
• Testing projections
• Don't need checkpoint management
• Want full control over persistence

Use CatchUpAsync when:

• Production projections with checkpoint management
• Want automatic "catch up to current" behavior
• Need to resume from last processed version
• Prefer ProjectionStore to manage checkpoints

Examples

Simple Event Counter

public record EventCountView(int Count);

[Projection("EventCount")]
public class EventCountProjection : Projection<EventCountView>
{
    public override EventCountView InitialView() => new(0);

    protected override EventCountView ApplyEvent(
        EventCountView view,
        StreamEvent streamEvent)
    {
        return view with { Count = view.Count + 1 };
    }
}

Aggregate-Specific Projection

[Projection("UserStats", AggregateTypes = ["User"])]
public class UserStatsProjection : Projection<UserStatsView>
{
    // Only processes events from "User" aggregate
}

Event-Specific Projection

// EventTypes must use wire-name format: {Aggregate}.{Name}.v{Version}
[Projection("CompletedSessions",
    EventTypes = ["SessionReview.SessionCompleted.v1"])]
public class CompletedSessionsProjection : Projection<CompletedView>
{
    // Only processes SessionCompleted events from SessionReview aggregate
    // Wire name matches [Event("SessionReview", "SessionCompleted", 1)]
}

Using Metadata

EventMetadata values are stored as object?, but after round-trip through storage, they materialize as JsonElement rather than their original CLR types. Use the safe typed extension methods:

protected override AuditView ApplyEvent(AuditView view, StreamEvent streamEvent)
{
    // Use extension methods for safe, typed access
    var timestamp = streamEvent.Metadata.GetDateTime("Timestamp");
    var userId = streamEvent.Metadata.GetString("UserId");
    var correlationId = streamEvent.Metadata.GetGuid("CorrelationId");
    var count = streamEvent.Metadata.GetInt32("Count");

    // Use metadata in projection logic
    return view with { /* ... */ };
}

Available Extension Methods:

• GetString(key) - Returns string?
• GetDateTime(key) - Returns DateTime?
• GetGuid(key) - Returns Guid?
• GetInt32(key) - Returns int?
• GetInt64(key) - Returns long?
• GetDecimal(key) - Returns decimal?
• GetDouble(key) - Returns double?
• GetBoolean(key) - Returns bool?

All methods return null if the key is not found or the value is null.

Relationship to Other Packages

• Rickten.EventStore - Provides IEventStore and IProjectionStore interfaces
• Rickten.Aggregator - Write-side aggregate patterns (commands → events → state)
• Rickten.Projector - Read-side projection patterns (events → views)

Projections and aggregates are complementary:

• Aggregates enforce business rules and produce events
• Projections build optimized read models from those events

For more information, see the main repository.