2. Wiring and running it

Chapter 1 produced a document aggregate as pure functions. This chapter registers those functions with FCQRS, stores their events in SQLite, projects the events into query data, and sends a command through the complete path. Continue in the same Program.fs.

Course position: chapter 1 defined a pure aggregate. This chapter runs it and follows one stored event into a query model. By the end you will understand the journal, projection offset, correlation id, and subscribe-before-send ordering from one working request.

This chapter adds three pieces around chapter 1's aggregate. Each failwith message names the section that implements it.

let buildApi () : IActor =
    failwith "create the actor system"

let handle (offset: int64) (message: obj) : unit =
    failwith "build the query side"

let run () : Async<unit> =
    failwith "send a command and read your own write"

// "Create the actor system": the host-builder replaces buildApi.
var builder = WebApplication.CreateBuilder(args);

// "Build the query side": the projection handler.
void Handle(long offset, object message) =>
    throw new NotImplementedException("build the query side");

// "Send a command and read your own write": top-level statements after Build().

Two paths, on purpose: writing and reading

The application has two paths:

                decide / fold
  command  ---------------------->  JOURNAL   (append-only events; the truth)
                                       |
                                       v
                                   projection
                                       |
                                       v
                                  read model   (a derived view; disposable)

Commands flow through the aggregate and append events to the journal. Projections consume those events and update read models. Application queries use the read model, not the aggregate state or journal.

Motivation: Two paths let query shapes evolve without widening the aggregate, and let business rules evolve without redesigning every query. The stored events are the stable handoff between them.

Create the actor system

Fcqrs.actor builds the Akka.NET system. The connection configures the journal, query journal, and snapshot store. The embedded defaults configure serialization, sharding, and a one-node cluster.

let buildApi () : IActor =
    let config = ConfigurationBuilder().Build()
    let loggerFactory = LoggerFactory.Create(fun _ -> ())
    let connection =
        Fcqrs.connect FCQRS.Actor.DBType.Sqlite "Data Source=tutorial.db;"
    Fcqrs.actor config loggerFactory (Some connection) "tutorial"

// C#: the DI host-builder creates the actor system and registers the aggregate;
// config and logger come from the container, so there's no explicit buildApi.
var builder = WebApplication.CreateBuilder(args);
builder.Services
    .AddFcqrs("Data Source=tutorial.db;", "tutorial")
    .AddAggregate<DocumentAggregate, DocumentState, DocumentCommand, DocumentEvent>();

The empty IConfiguration accepts every embedded default. User configuration is merged over those defaults when an application needs different Akka.NET settings. tutorial.db outlives the process, so the next run can recover the aggregate and replay the projection. See Configure the database for other providers.

Build the query side

This tutorial uses an in-memory dictionary for query data. The projection starts at offset zero and replays all stored Updated events on every run. A production projection instead persists its read model and, in the same database transaction, the offset of the last event it applied. The shared transaction is what makes a crash safe: with separate writes, an offset saved first skips an event forever, and data saved first applies an event twice on restart. The in-memory version here keeps the event-to-query transformation visible.

let readModel = ConcurrentDictionary<string, Document.Root>()

let handle (_offset: int64) (message: obj) =
    match message with
    | :? Event<Document.Event> as event ->
        match event.EventDetails with
        | Document.Updated document -> readModel[document.Id.ToString()] <- document
    | _ -> ()

// C#: the same in-memory read model and projection.
var readModel = new ConcurrentDictionary<string, Document>();

void Handle(long offset, object message)
{
    if (message is Event<DocumentEvent> { EventDetails: DocumentEvent.Updated updated })
        readModel[updated.Document.Id.ToString()] = updated.Document;
}

builder.Services.AddProjection((offset, ev) => Handle(offset, ev));

Projection.single publishes an aggregate event to subscribers after handle returns. A caller can therefore wait until this projection has applied a specific command's event. See Add a projection for transactional offset storage.

Send a command and read your own write

The aggregate acknowledges its stored event before the projection necessarily updates the read model. A query sent immediately after the command can therefore return the previous view. FCQRS carries a correlation id (CID) through the command, event, and projection notification:

  mint a CID
      |
      v
  SUBSCRIBE to that CID    <-- before sending, so the answer can't slip past
      |
      v
  send command  -->  aggregate  -->  journal  -->  projection re-publishes  -->  your wait wakes

Subscribe before sending. Subscribing afterward creates a race in which the projection can publish the notification before the subscription exists. .Send accepts the CID, aggregate id, command, and a predicate selecting the aggregate reply.

let run () =
    async {
        let api = buildApi ()

        let documents =
            Fcqrs.aggregate api
                { Name = "Document"
                  Initial = Document.initial
                  Decide = Document.decide
                  Fold = Document.fold
                  Snapshots = Default }

        Fcqrs.wireSagaStarters api []

        let subs = Fcqrs.projection api (Projection.single 0 handle)

        let cid = Fcqrs.newCid ()
        let id = Fcqrs.aggregateId "11111111-1111-1111-1111-111111111111"

        // Subscribe to this CID *before* sending so the confirmation can't be missed.
        use awaiter = subs.Subscribe(cid, 1)

        match Document.Root.TryCreate(Guid "11111111-1111-1111-1111-111111111111", "Welcome", "draft") with
        | Error e -> printfn "rejected: %s" e
        | Ok doc ->
            let! event =
                documents.Send cid id (Document.CreateOrUpdate doc)
                    (fun e ->
                        match e with
                        | Document.Updated _ -> true)

            do! awaiter.Task |> Async.AwaitTask
            let projected = readModel[doc.Id.ToString()]
            printfn "saved version %A; query returned '%s'" event.Version projected.Title.Value
    }

// C#: resolve the command handler + subscription from DI, then the same
// subscribe-before-send, read-your-writes flow.
using var app = builder.Build();
await app.StartAsync();
var documents = app.Services.GetRequiredService<Handler<DocumentCommand, DocumentEvent>>();
var subs = app.Services.GetRequiredService<ISubscribe>();

var cid = Values.NewCID();
var id = Values.CreateAggregateId("11111111-1111-1111-1111-111111111111");

using var awaiter = subs.SubscribeForFirst(cid);   // subscribe BEFORE sending
if (Document.TryCreate(Guid.Parse("11111111-1111-1111-1111-111111111111"), "Welcome", "draft", out var doc, out var err))
{
    var ev = await documents(
        e => e is DocumentEvent.Updated,
        cid, id, new DocumentCommand.CreateOrUpdate(doc));
    await awaiter.Task;
    Console.WriteLine($"saved version {ev.Version}; query returned '{readModel[doc.Id.ToString()].Title}'");
}

await app.StopAsync();

Wire it to your entry point:

[<EntryPoint>]
let main _ =
    run () |> Async.RunSynchronously
    0

// Nothing to wire: the top-level statements are the entry point, and the
// StartAsync and StopAsync calls shown above form its lifecycle.

Run it, then run it again

Run the program twice without deleting tutorial.db:

dotnet run
# saved version 1; query returned 'Welcome'

dotnet run
# saved version 2; query returned 'Welcome'

The second process starts with empty memory. FCQRS replays the first Updated event through fold, then handles the new command and stores version 2. The projection also starts with an empty dictionary and replays from offset zero before applying the new event.

Delete tutorial.db* and run again to start a new event history at version 1. In production, deleting a read model and resetting its offset is a rebuild operation; deleting the journal discards the source of truth.

What you now understand

A command produced a stored event. The aggregate recovered its state by replaying stored events, and a separate projection rebuilt query data from the same history. Subscribing to the CID before sending coordinated the query with that projection.

Common mistakes

• Querying without waiting for the required projection. The read model may still contain the old view.
• Subscribing after sending. The notification can pass before the subscription exists.
• Starting a durable projection at offset zero on every run. Persist the offset with the read-model update instead.
• Forgetting Fcqrs.wireSagaStarters. Wire an empty list when the application has no sagas.

Continue the learning path

Next, add a saga for a publication rule that spans a document and its URL slug. The next chapter assumes you understand the two paths and the projection timing shown here.

For optional depth after chapter 3, The read side explains transactional offsets and rebuilds, while Add a projection is the short production recipe.