Skip to content
Prague
Improve this page

Tutorial: Build an Order Service

You will build a small order-lookup service end to end: define the models, register their caches with the Kafka integration, wait for the initial load, query and join, put the hot path behind a frozen query, react to live shipment updates, and expose health checks and an HTTP endpoint. Every code block is a region of BuildAnOrderService.cs, which compiles with the repository. Steps that need a broker are plain methods; every other step runs as a test, so dotnet test samples/Prague.Samples --filter FullyQualifiedName~Tutorial passes without Kafka.

Prerequisites: the packages from Installation, .NET 9 or 10, and the Quick Start if [DataCache] is new to you. A broker is needed only for the steps marked broker.

1. Define the models

The service uses the domain every page of these docs shares: Order, Customer, OrderLine, Product and Shipment. Order is the one that matters most here: it carries the topic template, a foreign key to Customer, a Range index on CreatedAtUnixMs and a named sort.

public enum OrderStatus {
    Pending,
    Paid,
    Shipped,
    Cancelled,
}

[DataCache]                                                     // emits OrderCache; [DataCache("OrdersCache")] renames it
[DataCacheTopic("orders.[v:env]")]                              // Kafka topic; [v:env] resolves from KafkaCachesOptions.Vars
[DataCacheSort("ByCreatedAt", "CreatedAtUnixMs:-1", "OrderId:1")] // emits OrderCache.ByCreatedAtComparer (newest first)
public partial class Order {                                    // must be partial
    [DataCacheKey]
    public required string OrderId { get; init; }

    [DataCacheForeignKey<Customer>(DataCacheJoinType.ManyToOne)]   // many orders → one customer; also indexes CustomerId
    [DataCacheGlobalLastUpdateIndex<CustomerActivityIndex>]        // every order write stamps its customer's activity
    public required int CustomerId { get; init; }

    [DataCacheIndex(DataCacheIndexType.Range)]                     // sorted — Gte / Lt / … queries
    public required long CreatedAtUnixMs { get; init; }

    [DataCacheIndex(DataCacheIndexType.Many)]                      // 1:N — many orders share a status
    public required OrderStatus Status { get; init; }

    [DataCacheHasValueIndex]                                       // key set: WithSettledAtUnixMs() = settled (non-null)
    public long? SettledAtUnixMs { get; init; }

    public required decimal Total { get; init; }
}

The other four are in samples/Prague.Samples/Models. What you get for each: a generated XxxCache with index storage, AddOrUpdate, Query() and Prepare(); CacheEquals and Clone() on the entity; and, from the foreign keys, JoinWithCustomer(), JoinWithOrderLine() and JoinWithShipment() on OrderCache. Nothing to implement by hand.

What to expect: after a build, OrderCache, CustomerCache, OrderLineCache, ShipmentCache and ProductCache exist in your namespace, and Customer carries CustomerActivityIndex, a global last-update index that both Customer and Order report into.

2. Register the caches (broker)

AddKafkaCaches takes the connection options and a builder that lists which caches to consume. Topic names are templates: [v:name] is filled from Vars, [e:NAME] from an environment variable. Order declares its template on the class; the others get theirs here.

public static void AddOrderCaches(this IServiceCollection services) {
    services.AddKafkaCaches(
        o => {
            o.BootstrapServers = "kafka:9092";
            o.Vars["env"] = "prod";                              // fills [v:env] in every topic template
        },
        b => {
            b.AddCache<OrderCache, string, Order>();            // topic "orders.[v:env]" from [DataCacheTopic] → "orders.prod"
            b.AddCache<CustomerCache, int, Customer>("customers.[v:env]");   // explicit template
            b.AddCache<OrderLineCache, string, OrderLine>("order-lines.[v:env]");
            b.AddCache<ShipmentCache, string, Shipment>("shipments.[v:env]")
                .WithAfterHandler<ShipmentNotifier>();          // step 7
        });

    services.AddSingleton<CustomerOrdersPage>();            // step 6: one frozen query for the process
}

ShipmentNotifier and CustomerOrdersPage are written in steps 6 and 7. Registering OrderCache also pulls ProductCache into the registry (through OrderLine.ProductId) so joins resolve; it is not consumed because nothing calls AddCache for it, and it is not a DI service either. Only the caches you AddCache are injectable.

What to expect: building the container contacts no broker, so this part is testable as-is. The test RegistrationResolvesCachesWithoutABroker resolves OrderCache and asserts orders.Topic == "orders.prod". Consumption starts with the host.

3. Wait for the load (broker)

The hosted service AddKafkaCaches registers replays every topic to its end-of-partition before its StartAsync returns, so app.Run() does not accept requests on empty caches. If your host starts services concurrently, or you built the provider yourself, await the load explicitly.

public static async Task WaitForCachesAsync(IHost host) {
    // Option A — a hosted service is registered for you: its StartAsync returns only after every cache has
    // replayed its topic to EOF, so host.RunAsync() / app.RunAsync() do not serve before the data is there.
    // Option B — somewhere else (a host that starts services concurrently, a test that built the provider):
    await host.DataCachesLoadCompletion();

    // Option C — anything that holds the registry can await the same completion:
    var registry = host.Services.GetRequiredService<IDataCacheRegistry>();
    await registry.LoadingCompletion;
}

What to expect: a log line per cache with its load time and size; the readiness check from step 8 flips to Healthy at the same moment. The replay and the live phase are described in Kafka Integration.

4. Seed and query

From here the tutorial runs without a broker: it builds the same registry AddKafkaCaches would build and writes a few fixtures with AddOrUpdate, which is also what the consumer calls per message.

var registry = new DataCacheRegistryBuilder()           // in the service: AddKafkaCaches builds this for you
    .Register<OrderCache>()                              // pulls in Customer, OrderLine and Shipment through the FKs
    .Build();
var orders = registry.GetCache<OrderCache>();
var customers = registry.GetCache<CustomerCache>();
var lines = registry.GetCache<OrderLineCache>();

customers.AddOrUpdate(new Customer { Id = 42, Email = "ada@example.com", Country = "GB", Tier = CustomerTier.Gold, Name = "Ada" });
customers.AddOrUpdate(new Customer { Id = 7, Email = "bob@example.com", Country = "US", Tier = CustomerTier.Standard, Name = "Bob" });

orders.AddOrUpdate(NewOrder("ORD-001", 42, Now - 1 * DayMs, OrderStatus.Paid, 120m));
orders.AddOrUpdate(NewOrder("ORD-002", 42, Now - 2 * DayMs, OrderStatus.Pending, 80m));
orders.AddOrUpdate(NewOrder("ORD-003", 42, Now - 40 * DayMs, OrderStatus.Shipped, 15m));
orders.AddOrUpdate(NewOrder("ORD-004", 7, Now - 3 * DayMs, OrderStatus.Paid, 60m));

lines.AddOrUpdate(new OrderLine { LineId = "ORD-001/1", OrderId = "ORD-001", ProductId = 1, Quantity = 2, UnitPrice = 50m });
lines.AddOrUpdate(new OrderLine { LineId = "ORD-001/2", OrderId = "ORD-001", ProductId = 2, Quantity = 1, UnitPrice = 20m });

Each WithXxx call is a lane on an index; lanes are intersected, and only the survivors see Where. Count() never materializes rows.

var thirtyDaysAgo = Now - 30 * DayMs;

using var recent = orders.Query()
    .WithCustomerId(42)                                 // Many lane (from the foreign key)
    .WithCreatedAtUnixMs(q => q.Gte(thirtyDaysAgo))     // Range lane: intersected with the first
    .Where(o => o.Total > 100m)                         // residual filter on the survivors only
    .ExecutePooled();

var openCount = orders.Query().WithStatus(OrderStatus.Pending).Count();   // no materialization

Or takes two or more branches, each a lane set of its own, and unions them before the intersection with the outer lanes.

// CustomerId = 42 AND (Status = Pending OR Status = Paid)
using var actionable = orders.Query()
    .WithCustomerId(42)
    .Or(b => b.WithStatus(OrderStatus.Pending),
        b => b.WithStatus(OrderStatus.Paid))
    .ExecutePooled();

SortBounded with skip/take selects the page without sorting the rest; TotalCount is the size before paging. OrderCache.ByCreatedAtComparer comes from the [DataCacheSort] on Order: newest first, OrderId as the tie-break, so the order is total and pages never overlap.

using var page = orders.Query()
    .WithCustomerId(42)
    .SortBounded(OrderCache.ByCreatedAtComparer)        // from [DataCacheSort("ByCreatedAt", "CreatedAtUnixMs:-1", "OrderId:1")]
    .ExecutePooled(skip: 0, take: 2);                   // selects the page, never sorts the rest

Console.WriteLine($"showing {page.Count} of {page.TotalCount}");

What to expect: recent holds ORD-001 only (ORD-002 fails the Total filter, ORD-003 the date); actionable holds ORD-001 and ORD-002; the page prints showing 2 of 3. ExecutePooled hands back a buffer from a pool and the using returns it. Query Engine covers plan selection and the terminals; Sorted Paging covers Sort versus SortBounded.

5. Join orders with customers and lines

The foreign keys on Order and OrderLine generate the join methods. JoinWithCustomer() is many-to-one, so each row carries a Customer?; JoinWithOrderLine() follows OrderLine.OrderId back, so each row carries a QueryResults<OrderLine>.

using var rows = orders.Query()
    .WithCustomerId(42)
    .JoinWithCustomer()                                 // ManyToOne forward: row.Right is Customer?
    .JoinWithOrderLine()                                // 1:N reverse: row.Right2 is QueryResults<OrderLine>
    .ExecutePooled();                                   // one Dispose returns every buffer, lines included

foreach (var row in rows)
    Console.WriteLine($"{row.Left.OrderId} for {row.Right?.Name ?? "unknown"}: {row.Right2.Count} line(s)");

What to expect: three rows for customer 42; ORD-001 has two lines, the others none (an empty, not null, Right2). One Dispose on the outer result returns every nested buffer. Inner joins, chained joins and the leak rules are in Joins.

6. Freeze the hot path

The endpoint in step 8 answers one shape of question many times: orders of customer X with status Y, newest first, one page. Plan it once with Prepare<TArgs>() and BuildFrozen(); each execution then costs no allocation and no planning. The class below is what the endpoint injects.

public sealed class CustomerOrdersPage {
    public readonly record struct Args(int CustomerId, OrderStatus Status);

    private readonly FrozenQuery<Args, Order> _page;

    public CustomerOrdersPage(OrderCache orders) {
        _page = orders.Prepare<Args>()
            .WithCustomerId(static a => a.CustomerId)          // parameterized lane
            .WithStatus(static a => a.Status)
            .SortBounded(OrderCache.ByCreatedAtComparer)        // newest first, OrderId breaks ties: a total order
            .BuildFrozen();                                     // planned once; the only allocation
    }

    public OrderDto[] Get(int customerId, OrderStatus status, int skip, int take) {
        using var rows = _page.ExecutePooled(new Args(customerId, status), skip, take);   // 0 B, any thread
        var dtos = new OrderDto[rows.Count];
        for (var i = 0; i < rows.Count; i++)
            dtos[i] = new OrderDto(rows[i].OrderId, rows[i].Total, rows[i].Status.ToString());
        return dtos;                                            // the pooled buffer is returned before we leave
    }
}

public sealed record OrderDto(string OrderId, decimal Total, string Status);

Two rules the code follows: lanes take static selectors over the argument struct, and nothing pooled leaves the method; the rows are copied into DTOs before the buffer goes back.

What to expect: FrozenQueryServesPages gets ORD-001 for (42, Paid), ORD-002 for (42, Pending) and Bob's single order for (7, Paid). Without a sort a frozen result has no row-order guarantee; this one is sorted, so pages match the eager query exactly. Plans, executors and Explain() are in Prepared Queries.

7. React to live updates

An ICacheAfterHandler<TKey, TValue> runs after a message has been applied to the cache during the live phase; the initial replay never calls it. UpdateType tells you what happened: Add, Update (with oldValue), Delete (with oldValue), Same (a re-emit of the resident value) or Filtered (rejected by a filter: no key, no values).

public sealed class ShipmentNotifier : ICacheAfterHandler<string, Shipment> {
    private readonly List<string> _notified = new();

    public IReadOnlyList<string> Notified => _notified;

    public ValueTask Handle(UpdateType updateType, string key, Shipment? newValue, Shipment? oldValue) {
        // Live phase only: the initial replay of the topic never reaches this method.
        if (updateType is UpdateType.Add or UpdateType.Update)
            _notified.Add($"{newValue!.OrderId} shipped with {newValue.Carrier}");
        return ValueTask.CompletedTask;                         // no I/O here; this runs on the cache's live worker
    }
}

It was registered in step 2 with .WithAfterHandler<ShipmentNotifier>(). Handlers run sequentially on the cache's own live worker, so keep them short and hand real work to a queue. A unit test drives one directly:

var notifier = new ShipmentNotifier();
var shipment = new Shipment { ShipmentId = "SHP-1", OrderId = "ORD-001", Carrier = "DHL", ShippedAtUnixMs = Now };

// The consumer calls Handle on the live phase; a unit test calls it directly.
await notifier.Handle(UpdateType.Add, shipment.ShipmentId, shipment, null);
await notifier.Handle(UpdateType.Same, shipment.ShipmentId, shipment, null);     // a re-emit: ignored

What to expect: one notification; the Same call is ignored. React to changes covers each UpdateType, equality and the factory registration.

8. Health checks and an HTTP endpoint

Two checks ship with Prague.Kafka: liveness (the consumer is polling and no handler is stuck) and readiness (loaded, assigned, brokers up). Map them on separate paths, and map Degraded to 503; the default returns 200.

public static void AddOrderHealth(this IServiceCollection services) {
    services.AddHealthChecks()
        .AddPragueKafkaLiveness(tags: ["live"])              // Unhealthy: consumer dead or stalled
        .AddPragueKafkaReadiness(tags: ["ready"]);           // Degraded: still loading, no partition, brokers down
}

public static void MapOrderHealth(this IEndpointRouteBuilder app) {
    app.MapHealthChecks("/health/live", new HealthCheckOptions { Predicate = r => r.Tags.Contains("live") });
    app.MapHealthChecks("/health/ready", new HealthCheckOptions {
        Predicate = r => r.Tags.Contains("ready"),
        ResultStatusCodes = {
            [HealthStatus.Healthy] = StatusCodes.Status200OK,
            [HealthStatus.Degraded] = StatusCodes.Status503ServiceUnavailable,   // the default maps Degraded → 200
            [HealthStatus.Unhealthy] = StatusCodes.Status503ServiceUnavailable,
        },
    });
}

The endpoint injects the frozen query from step 6 and the generated CustomerCache, which AddKafkaCaches registered as a singleton.

public static void MapOrderEndpoints(this IEndpointRouteBuilder app) {
    app.MapGet("/customers/{customerId:int}/orders",
        (int customerId, OrderStatus status, int skip, int take, CustomerOrdersPage page)
            => Results.Ok(page.Get(customerId, status, skip, take)));   // DTOs: nothing pooled leaves the handler

    app.MapGet("/customers/{customerId:int}", (int customerId, CustomerCache customers)
        => customers.Cache.TryGet(customerId, out var customer) ? Results.Ok(customer) : Results.NotFound());
}

Program.cs is the sum of the steps (broker):

public static WebApplication BuildOrderService(string[] args) {
    var builder = WebApplication.CreateBuilder(args);
    builder.Services.AddOrderCaches();                      // step 2
    builder.Services.AddOrderHealth();                      // step 8

    var app = builder.Build();
    app.MapOrderHealth();
    app.MapOrderEndpoints();                                // step 8
    return app;                                             // app.Run(): serving starts after the caches loaded
}

What to expect: EndpointServesDtosFromTheCache hosts the endpoints in-process over the seeded caches and gets ORD-001 from GET /customers/42/orders?status=Paid&skip=0&take=10, Ada from GET /customers/42 and a 404 for GET /customers/999. With a broker, GET /health/ready answers 503 until step 3 completes and 200 afterwards. Health and metrics explains each failure reason and the OpenTelemetry instruments.

Where to go next

Prague documentation