Tyto.Materializer 0.1.0-alpha.11

Tyto.Materializer

A high-performance, resilient projection engine for building Materialized Views in Event-Driven Architectures and CQRS systems.

Tyto.Materializer transforms a stream of domain events into query-optimized read models (Views). Unlike standard caching mechanisms (Last-Write-Wins), it enforces strong data consistency using optimistic concurrency control, ensuring your read models never get overwritten by stale data, even under high concurrency.

🚀 Key Features

• Event-Driven & Reactive: Seamlessly integrates with message buses to update views in near real-time.
• Optimistic Concurrency: Uses Version Vectors to handle race conditions automatically.
• Class-Based Projectors: Cleanly separate your projection logic into testable classes (IViewProjector) with full Dependency Injection support.
• Auto-Discovery: Automatically registers all projection rules implemented by a class using Reflection, keeping your startup configuration clean.
• Resilient: Built-in retry policies with exponential backoff for handling concurrency conflicts.
• Storage Agnostic: Plug-and-play storage providers (Redis, EF Core, InMemory).

📦 Installation

dotnet add package Tyto.Materializer

> Note: You will likely need a storage provider package as well, such as Tyto.Materializer.Redis.

⚡ Getting Started

1. Define Your View

Views are simple POCO classes that implement IProjectableView. This interface adds a Version property which the library uses to track state changes and prevent race conditions.

using Tyto.Materializer;

public class UserBalanceView : IProjectableView
{
    // The unique key for this view (e.g., UserId)
    public Guid UserId { get; set; } 
    
    public string FullName { get; set; } = string.Empty;
    public decimal CurrentBalance { get; set; }
    public DateTime LastUpdated { get; set; }
    
    // Managed automatically by Tyto
    public ProjectableVersion Version { get; set; }
}

2. Define a Projector

Instead of writing logic inside configuration files, define your projection logic in clean, testable classes implementing IViewProjector<TView, TEvent>. You can inject repositories, loggers, or other services here.

using Tyto.Materializer;

public class UserBalanceProjector : 
    IViewProjector<UserBalanceView, UserRegistered>,
    IViewProjector<UserBalanceView, MoneyDeposited>
{
    private readonly ILogger<UserBalanceProjector> _logger;

    public UserBalanceProjector(ILogger<UserBalanceProjector> logger)
    {
        _logger = logger;
    }

    public ValueTask ProjectAsync(UserBalanceView view, UserRegistered @event, CancellationToken ct)
    {
        // Tyto automatically handles creating the view instance if it doesn't exist.
        view.UserId = @event.UserId;
        view.FullName = @event.Name;
        view.CurrentBalance = 0;
        view.LastUpdated = DateTime.UtcNow;
        
        _logger.LogInformation("Projection created for user {Id}", @event.UserId);
        return ValueTask.CompletedTask;
    }

    public ValueTask ProjectAsync(UserBalanceView view, MoneyDeposited @event, CancellationToken ct)
    {
        view.CurrentBalance += @event.Amount;
        view.LastUpdated = DateTime.UtcNow;
        return ValueTask.CompletedTask;
    }
}

3. Declare it on the endpoint that receives the events

A view belongs to an endpoint: it is projected on that endpoint, and nowhere else.

tyto.Endpoints(endpoints => endpoints.Add("accounts", endpoint =>
{
    endpoint.ListenOn("rabbit", "accounts");

    endpoint.Materialize(views =>
    {
        views.ForView<UserBalanceView>()
            // 1. Identify the Key: How to extract the View ID from an Event
            .IdentifiedAs<Guid>(id => id.From<UserRegistered>(e => e.UserId))
            
            // 2. Configure Storage: Choose where to save the view
            .Store.UseRedis("localhost:6379,abortConnect=false") 
            
            // 3. Register Projectors: Use the class-based approach
            // This single line automatically registers BOTH UserRegistered and MoneyDeposited
            // handlers defined in the UserBalanceProjector class.
            .ProjectsWith<UserBalanceProjector>();
    });
}));

In a module, the same call goes inside module.Endpoint(...). A host that does not run the module registers nothing for it, so the view — and the DbContext it lives in — is never resolved there.

4. Composite Keys (Optional)

If your view has a composite key (such as multi-tenant partitions (ApplicationId, RecipientId)), define a record struct as your key type:

public readonly record struct RecipientKey(long ApplicationId, string RecipientId);

public class RecipientView : IProjectableView
{
    public long ApplicationId { get; set; }
    public string RecipientId { get; set; } = string.Empty;
    public string Name { get; set; } = string.Empty;
    public ProjectableVersion Version { get; set; }
}

endpoint.Materialize(views => views
    .ForView<RecipientView>()
    .IdentifiedAs<RecipientKey>(id => {
        id.From<RecipientCreated>(e => new RecipientKey(e.ApplicationId, e.RecipientId));
        id.From<RecipientUpdated>(e => new RecipientKey(e.ApplicationId, e.RecipientId));
    })
    .ProjectsWith<RecipientProjector>());

• The properties of RecipientKey are automatically mapped by name to the properties of RecipientView.
• In Entity Framework Core, declare HasKey(v => new { v.ApplicationId, v.RecipientId }). Tyto validates that all key components exist in the EF Core model at startup and queries them in the model's exact primary key order.
• A strongly-typed id is one key, not a composite. When every selector reads a property the view has under the same name and of the key's own type — id.From<OrderPlaced>(e => e.OrderId) with OrderId OrderId on the view, where record struct OrderId(Guid Value) — it is a single column, as it always was. Only a key the selectors build (e => new RecipientKey(...)) is split into components.
• An aggregate view's partition is always one property; a composite partition is refused at startup.

A projection is a handler

Materialize adds one handler per event type to the endpoint. That is the whole mechanism, and it is why a view gets everything a handler gets: the message's scope, the endpoint's transaction, its inbox, its retries and its dead-letter queue. An event routed to two endpoints moves each endpoint's own views, once each. Because handlers always run inside the endpoint's middleware, it makes no difference whether Materialize is called before or after UseTransactions.

A view projects an event once per endpoint: two Projects<TEvent> for the same view and event — in one Materialize call or in two — are refused at startup. Inside a transaction they would share one DbContext, and the second would update a row against a version the database never had. Put both changes in one projection.

(tyto.AddMaterializer(...) used to register views for the whole host, as a middleware beside the handlers. Every endpoint then projected every view — one module's view was saved on another module's endpoint, outside that endpoint's transaction — and because the middleware sat outside the transaction, a redelivery the inbox refused was projected anyway. It has been removed.)

🏗️ Architecture & Concurrency

In a distributed system, multiple events might try to update the same view at the exact same time. Tyto handles this robustly:

1. Read: The Materializer loads the current View (e.g., Version 1).
2. Project: It applies the logic from your Projector.
3. Atomic Save: It attempts to save the new state (Version 2).
4. Conflict Detection: If the underlying store detects that the version has changed (e.g., someone else saved Version 2 before we could), it throws a ConcurrencyException.
5. Retry: Tyto automatically catches this exception, reloads the fresh data (Version 2), re-applies your projection logic, and attempts to save again.

This guarantees that a concurrent writer cannot overwrite a change it never saw. Note what it does not guarantee on its own: the version counts how many times the row has been written, not which events those writes were, so a redelivered event passes it cleanly — the row is read, the event applied a second time, and the write is consistent with a version nobody else touched.

Applying each event once, and in order

A broker that delivers at least once will redeliver, and two partitions will hand you event 3 before event 2. Tell the view how to read an event's position in its entity's stream and both are handled:

public sealed class UserBalanceView : IOrderedProjectableView {
    public Guid UserId { get; set; }
    public ProjectableVersion Version { get; set; } = ProjectableVersion.Initial;
    public long AppliedSequence { get; set; }   // how far this view has been brought
    public decimal Balance { get; set; }
}

materializer.ForView<UserBalanceView>()
    .IdentifiedAs<Guid>(id => id.FromEventsImplementing<IUserEvent>(e => e.UserId))
    .EnsureEventOrder(e => ((IUserEvent)e).StreamVersion)
    .Projects<MoneyDeposited>((view, e) => view.Balance += e.Amount);

• An event the view has already applied is discarded — counted on tyto.materializer.replays.total, not applied twice.
• An event that would skip one is refused with ProjectionOutOfOrderException, so the message fails and the broker delivers it again; by then the missing event has usually been handled. The broker is the buffer, and nothing is stashed in a table of its own.
• AppliedSequence lives on the view, so what a view has seen and what it says commit together.

It is opt-in: a transport that guarantees order and never redelivers pays nothing for a guard it does not need.

A failed projection fails the message

Projection used to catch and log every exception. A conflict that outlived its retries, an unreachable store, a save the transaction forbids — each left the read model one change behind for good, with a log line as the only trace. It now propagates: a read model quietly wrong is worse than a message visibly stuck, because only one of the two is ever noticed.

Inside a transactional endpoint

With Entity Framework Core, keep the view in the DbContext the endpoint's transaction runs on:

endpoint.UseTransactions<AppDbContext>(tx => tx.UseInbox());
endpoint.Materialize(views => views
    .ForView<UserBalanceView>()
    .IdentifiedAs<Guid>(id => id.From<MoneyDeposited>(e => e.UserId))
    .Store.UseEntityFrameworkCore().InContext<AppDbContext>()
    .Projects<MoneyDeposited>((view, e) => view.Balance += e.Amount));

The projection handler runs in the message's scope, so the store is handed the transaction's own context: the view is written into it and committed by its single SaveChanges, alongside the inbox row for the event that moved it — or not at all. There is nothing to opt into (tx.UseMaterializer() is now an obsolete no-op).

A view stored in some other DbContext than the transaction's is refused with an InvalidOperationException rather than saved on its own: it would commit whether or not the message did, be projected twice on a redelivery and be left behind by a rollback. Outside a message — a replay, an HTTP request — the store saves for itself as before.

Aggregate views: many events, no single owner

A view that many independent events add to — storage per bucket, uploads per day — has no owner. As an IProjectableView it is one row they all fight over: under concurrency the version check turns most of them into conflicts, and the retries into a storm. An aggregate view is instead one row per event, carrying that event's delta, summed when it is read. An insert has nothing to conflict with.

public sealed class BucketStorageEntry : IAggregateViewEntry {
    public long Id { get; set; }
    public string BucketName { get; set; } = "";      // the partition
    public string EventId { get; set; } = "";         // the message id: the same on every redelivery
    public DateTimeOffset CreatedAt { get; set; }
    public long Bytes { get; set; }                   // this event's delta, not a running total
}

endpoint.Materialize(views => views
    .ForAggregateView<BucketStorageEntry>()
    .PartitionedBy<string>(p => {
        p.From<MediaUploaded>(e => e.BucketName);
        p.From<MediaDeleted>(e => e.BucketName);
    })
    .Store.UseEntityFrameworkCore().InContext<AnalyticsDbContext>()
    .ProjectsWith<BucketStorageProjector>());

// OnModelCreating — required, and checked when the store is first used:
modelBuilder.Entity<BucketStorageEntry>(e => {
    e.HasKey(x => x.Id);
    e.HasIndex(x => new { x.BucketName, x.EventId }).IsUnique();
});

// The query side sums:
long total = await db.Set<BucketStorageEntry>().Where(x => x.BucketName == "avatars").SumAsync(x => x.Bytes);

public sealed class BucketStorageProjector :
    IAggregateProjector<BucketStorageEntry, MediaUploaded>,
    IAggregateProjector<BucketStorageEntry, MediaDeleted> {

    public ValueTask ProjectAsync(BucketStorageEntry entry, MediaUploaded e, CancellationToken ct) {
        entry.Bytes = e.FileSize;        // this upload's contribution
        return ValueTask.CompletedTask;
    }

    public ValueTask ProjectAsync(BucketStorageEntry entry, MediaDeleted e, CancellationToken ct) {
        entry.Bytes = -e.FileSize;
        return ValueTask.CompletedTask;
    }
}

• The projector is handed a fresh entry every time — set the delta. That is why it is an IAggregateProjector, not an IViewProjector: the shape is the same, but entry.Bytes += ... is right for a view and silently wrong here.
• A redelivered event is not counted twice. Nothing here is ever updated, so a second copy that got in would be wrong for good. An event whose message id is already in its partition is skipped; two copies racing each other are settled by the unique index — one commits, the other fails its message, and its redelivery finds the entry already there.
• ConcurrencyException cannot happen: there is no version.
• Stores: Entity Framework Core and in-memory. Redis does not support aggregate views yet.
• CreatedAt is there for a later compaction that folds old rows into a snapshot; there is none yet.

🧩 Advanced Usage

Several projector classes for one view

ProjectsWith<TProjector>() registers the events a class implements IViewProjector<TView, TEvent> for, so different classes can handle different events of the same view:

.ProjectsWith<UserRegistrationProjector>()   // IViewProjector<UserBalanceView, UserRegistered>
.ProjectsWith<DepositProjector>()            // IViewProjector<UserBalanceView, MoneyDeposited>

A class that implements it for no event of the view is refused at startup, and so is one event claimed by two classes. The projector gets the message's cancellation token.

(.Projects<TView, TKey, TProjector>() and .Projects<TView, TKey, TEvent, TProjector>() still compile, marked obsolete: the view and key types they ask for are already known.)

In-Memory Provider

For Unit Testing, use the In-Memory provider. It simulates serialization and concurrency conflicts to ensure your logic is production-ready.

.Store.UseInMemory()

Dependency Injection

Because Projectors are resolved from the DI container, you can use any registered service (Repositories, Anti-Corruption Layers, etc.) within your ProjectAsync methods.

public class OrderProjector(IExchangeRateService rateService) : IViewProjector<OrderView, OrderPlaced>
{
    public async ValueTask ProjectAsync(OrderView view, OrderPlaced @event, CancellationToken ct)
    {
        var rate = await rateService.GetCurrentRateAsync(ct);
        view.Total = @event.Amount * rate;
    }
}