Step 4: Show a statement

Alice wants a statement: every deposit and withdrawal, with the balance after it. The account cannot answer that. It decides commands and keeps only what its rules need, which is the owner and the balance. This step adds the other half of the split from the overview: a read model, a table shaped for one screen, which a projection keeps up to date from the journal.

The same feature in a CRUD application

A CRUD application writes the statement row in the same transaction as the balance, and the statement query reads the table that the write code changed:

BEGIN;
UPDATE accounts SET balance = balance - 30 WHERE id = 'alice';
INSERT INTO transactions (account, entry, amount)
VALUES ('alice', 'Withdrawal', -30);
COMMIT;
-- The statement reads the rows the write code inserted.
SELECT entry, amount FROM transactions WHERE account = 'alice';

Every code path that changes a balance must also insert the transaction row. In FCQRS, the account stores only its event. One projection writes the statement from the journal, and because the journal holds every event, the statement can be rebuilt from it.

Create the read model

module Account =
    open FCQRS.Common

    // What a caller can ask an account to do.
    type AccountCommand =
        | Open of owner: string
        | Deposit of amount: decimal
        | Withdraw of amount: decimal

    // What the account replies. Rejected is a reply only: it is never stored.
    type AccountEvent =
        | Opened of owner: string
        | Deposited of amount: decimal
        | Withdrawn of amount: decimal
        | Rejected of reason: string

    // What the account knows now, rebuilt from its events.
    type AccountState = { Owner: string option; Balance: decimal }

    // The state before the account's first event.
    let initial = { Owner = None; Balance = 0m }

    // Chooses what to do with a command, based on the current state.
    let decide (command: Command<AccountCommand>) (state: AccountState) =
        match command.CommandDetails, state.Owner with
        | Open _, Some _ -> DeferEvent(Rejected "The account is already open")
        | Open owner, None -> PersistEvent(Opened owner)
        | _, None -> DeferEvent(Rejected "The account is not open")
        | (Deposit amount | Withdraw amount), _ when amount <= 0m ->
            DeferEvent(Rejected "The amount must be positive")
        | Deposit amount, _ -> PersistEvent(Deposited amount)
        | Withdraw amount, _ when amount > state.Balance ->
            DeferEvent(Rejected $"Insufficient funds: {state.Balance} available")
        | Withdraw amount, _ -> PersistEvent(Withdrawn amount)

    // Applies one event to the state. A rejection changes nothing.
    let fold (event: Event<AccountEvent>) (state: AccountState) =
        match event.EventDetails with
        | Opened owner -> { state with Owner = Some owner }
        | Deposited amount -> { state with Balance = state.Balance + amount }
        | Withdrawn amount -> { state with Balance = state.Balance - amount }
        | Rejected _ -> state
module Statement =
    open System.Data.Common
    open System.Threading.Tasks
    open Akka.Persistence.Query
    open Dapper
    open FCQRS.Common
    open Account

// The read model: one row per stored event, with the balance after it.
let createTable (connection: DbConnection) =
    connection.Execute
        "CREATE TABLE IF NOT EXISTS statement (
             account TEXT NOT NULL,
             version INTEGER NOT NULL,
             entry TEXT NOT NULL,
             amount NUMERIC NOT NULL,
             balance NUMERIC NOT NULL,
             PRIMARY KEY (account, version))"
    |> ignore

// The read model: one row per stored event, with the balance after it.
public static void CreateTable(DbConnection connection) =>
    connection.Execute(
        """
        CREATE TABLE IF NOT EXISTS statement (
            account TEXT NOT NULL,
            version INTEGER NOT NULL,
            entry TEXT NOT NULL,
            amount NUMERIC NOT NULL,
            balance NUMERIC NOT NULL,
            PRIMARY KEY (account, version))
        """);

The statement is an ordinary table. version is the version of the event that produced the row, so each event adds at most one row.

Write the projection

// Adds a row whose balance continues from the account's previous row.
let private addRow =
    "INSERT INTO statement (account, version, entry, amount, balance)
     SELECT @Account, @Version, @Entry, @Amount,
            COALESCE((SELECT balance FROM statement WHERE account = @Account
                      ORDER BY version DESC LIMIT 1), 0) + @Amount"

// FCQRS calls this for each stored event, inside a transaction it commits.
let handle (connection: DbConnection) (transaction: DbTransaction)
           (envelope: EventEnvelope) =
    task {
        match envelope.Event with
        // Only account events go on a statement; Sender is the account's ID.
        | :? Event<AccountEvent> as stored ->
            let add (entry: string) (amount: decimal) =
                let row =
                    {| Account = string stored.Sender.Value
                       Version = envelope.SequenceNr
                       Entry = entry
                       Amount = amount |}
                connection.ExecuteAsync(addRow, row, transaction) :> Task
            match stored.EventDetails with
            | Opened owner -> do! add $"Opened for {owner}" 0m
            | Deposited amount -> do! add "Deposit" amount
            | Withdrawn amount -> do! add "Withdrawal" -amount
            // A rejection is a reply only; the journal never holds one.
            | Rejected _ -> ()
        | _ -> ()
    }
    :> Task

// Adds a row whose balance continues from the account's previous row.
const string AddRow =
    """
    INSERT INTO statement (account, version, entry, amount, balance)
    SELECT @Account, @Version, @Entry, @Amount,
           COALESCE((SELECT balance FROM statement WHERE account = @Account
                     ORDER BY version DESC LIMIT 1), 0) + @Amount
    """;

// FCQRS calls this for each stored event, inside a transaction it commits.
public static async Task Handle(
    DbConnection connection, DbTransaction transaction, EventEnvelope envelope)
{
    // Only account events go on a statement; Sender is the account's ID.
    if (envelope.Event is not Event<AccountEvent> { Sender: { } sender } stored)
        return;

    Task Add(string entry, decimal amount)
    {
        var row = new
        {
            Account = sender.Value.ToString(),
            Version = envelope.SequenceNr,
            Entry = entry,
            Amount = amount
        };
        return connection.ExecuteAsync(AddRow, row, transaction);
    }

    await (stored.EventDetails switch
    {
        Opened opened => Add($"Opened for {opened.Owner}", 0m),
        Deposited deposited => Add("Deposit", deposited.Amount),
        Withdrawn withdrawn => Add("Withdrawal", -withdrawn.Amount),
        // A rejection is a reply only; the journal never holds one.
        Rejected => Task.CompletedTask
    });
}

FCQRS calls the handler for each stored event and passes it a connection and a transaction. It commits the handler's writes together with the projection's progress. If the handler fails, neither is committed, and the event is handled again when the projection restarts.

The handler receives events from every aggregate, so it keeps only account events. envelope.Event is the stored event, Sender is the ID of the account that stored it, and envelope.SequenceNr is its version. Rejected never arrives, because the journal never holds it.

Each row's balance continues from the account's previous row. That is correct because a projection receives one account's events in version order. Across accounts it follows the journal's numbers, which on PostgreSQL can differ from the order writes commit, so a handler that combines several accounts must not depend on which account's event arrives first.

Register the projection

open System
open System.Data.Common
open System.IO
open Microsoft.Data.Sqlite
open Microsoft.Extensions.Configuration
open Microsoft.Extensions.Logging
open FCQRS.Actor
open FCQRS.Common
open FCQRS.FSharp
open FCQRS.ProjectionStorage
open FCQRS.Projections
open Account
let database = Path.Combine(AppContext.BaseDirectory, "accounts.db")
let connectionString = $"Data Source={database}"

// Startup and the account registration are the same as in step 2.
let logging = LoggerFactory.Create(fun _ -> ())
let configuration = ConfigurationBuilder().Build()
let connection = Fcqrs.connect DBType.Sqlite connectionString
let api = Fcqrs.actor configuration logging (Some connection) "accounts"

let accounts =
    Fcqrs.aggregate api
        { Name = "Account"
          Initial = initial
          Decide = decide
          Fold = fold
          Snapshots = Default
          Passivation = PassivationPolicy.Default }

Fcqrs.wireSagaStarters api []

// The statement table lives in the same SQLite file as the journal.
do
    use connection = new SqliteConnection(connectionString)
    Statement.createTable connection

// FCQRS passes each new journal event to Statement.handle, and commits the
// handler's rows and the projection's progress in one transaction.
let store =
    SqlProjectionStore(
        ProjectionSqlDialect.Sqlite,
        Func<DbConnection>(fun () -> new SqliteConnection(connectionString)))
let options = TransactionalProjectionOptions("Statement", store)
let statement = Fcqrs.transactionalProjection api options Statement.handle

// The statement table lives in the same SQLite file as the journal.
using (var connection = new SqliteConnection(connectionString))
    Statement.CreateTable(connection);

// FCQRS passes each new journal event to Statement.Handle, and commits the
// handler's rows and the projection's progress in one transaction.
var store = new SqlProjectionStore(
    ProjectionSqlDialect.Sqlite, () => new SqliteConnection(connectionString));
var options = new TransactionalProjectionOptions("Statement", store);

var builder = Host.CreateApplicationBuilder();
builder.Logging.ClearProviders();
builder.Services.AddFcqrs(connectionString, "accounts")
    .AddAggregate<Account>()
    .AddTransactionalProjection(options, Statement.Handle);
using var host = builder.Build();
await host.StartAsync();

"Statement" names the projection's progress. FCQRS records, for each account, the version of the last event this projection committed, and resumes from there after a restart. Give each read model its own name. In C#, AddTransactionalProjection registers the projection with the host, and the program gets it as IProjection.

Send commands and wait for the statement

let describe event =
    match event with
    | Opened owner -> $"Opened for {owner}"
    | Deposited amount -> $"Deposited {amount}"
    | Withdrawn amount -> $"Withdrew {amount}"
    | Rejected reason -> $"Rejected: {reason}"

let alice = Fcqrs.aggregateId "alice"

// Send a command and wait until the statement includes the event it stored.
let send command =
    let reply =
        Fcqrs.sendAwaiting
            statement accounts (Fcqrs.newCid ()) alice command (fun _ -> true)
        |> Async.RunSynchronously
    let description = describe reply.EventDetails
    printfn $"{description} (version {reply.Version})"

send (Open "Alice")
send (Deposit 100m)
send (Withdraw 30m)
send (Deposit 50m)
send (Withdraw 500m)

// Sends commands to accounts and returns the event each one replied with.
var accounts = host.Services
    .GetRequiredService<Handler<AccountCommand, AccountEvent>>();
// Publishes a notification after it commits each event.
var statement = host.Services.GetRequiredService<IProjection>();
var alice = Values.CreateAggregateId("alice");

// Send a command and wait until the statement includes the event it stored.
async Task Send(AccountCommand command)
{
    var cid = Values.NewCID();
    // Subscribe first: a notification sent before the subscription is lost.
    using var projected = statement.SubscribeForFirst(cid);
    var reply = await accounts(_ => true, cid, alice, command);
    // A rejection is not stored, so no notification comes for it.
    if (reply.Journaled?.Value != false)
        await projected.Task.WaitAsync(TimeSpan.FromSeconds(30));
    var description = Describe(reply.EventDetails);
    Console.WriteLine($"{description} (version {reply.Version})");
}

await Send(new Open("Alice"));
await Send(new Deposit(100m));
await Send(new Withdraw(30m));
await Send(new Deposit(50m));
await Send(new Withdraw(500m));

The account replies as soon as it stores the event. The projection writes the statement row afterwards, in its own transaction, so a query right after the reply can miss the newest row.

A correlation ID labels one request. FCQRS copies it to the events the command causes, and the projection publishes a notification carrying it after it commits each event. Fcqrs.sendAwaiting subscribes to the command's correlation ID, sends the command, and waits for that notification. The C# Send spells out the same steps.

• Subscribe before sending. A notification published before the subscription exists is missed, and the wait would last until its timeout.
• A rejection has no notification. It is not stored, so the projection never sees it. sendAwaiting returns right away when the reply was not stored; the C# code checks Journaled.
• The wait is bounded. sendAwaiting raises TimeoutException after the command timeout, 30 seconds by default. The C# code uses WaitAsync for the same bound.

The last argument, fun _ -> true (_ => true in C#), chooses which reply completes the send. Each command here produces one reply, so it accepts any.

Waiting covers this projection only. Another read model, or a system that reads the journal, can still be behind. Subscriptions live in memory for the duration of the request; they are not a queue a client can reconnect to. To wait for every event stored so far instead of one request's events, call CatchUpAsync on the projection, as Catch up projections shows.

Query the statement

// The statement is an ordinary table: read it with SQL.
let printStatement () =
    use connection = new SqliteConnection(connectionString)
    connection.Open()
    use query = connection.CreateCommand()
    query.CommandText <-
        "SELECT version, entry, amount, balance FROM statement
         WHERE account = 'alice' ORDER BY version"
    use rows = query.ExecuteReader()
    printfn "\nStatement for alice:"
    printfn "  version  entry              amount  balance"
    while rows.Read() do
        let version, entry = rows.GetInt64 0, rows.GetString 1
        let amount, balance = rows.GetDecimal 2, rows.GetDecimal 3
        printfn $"  {version,7}  {entry,-18}{amount,7}{balance,9}"

printStatement ()

// The statement is an ordinary table: read it with SQL.
void PrintStatement()
{
    using var connection = new SqliteConnection(connectionString);
    connection.Open();
    using var query = connection.CreateCommand();
    query.CommandText =
        """
        SELECT version, entry, amount, balance FROM statement
        WHERE account = 'alice' ORDER BY version
        """;
    using var rows = query.ExecuteReader();
    Console.WriteLine();
    Console.WriteLine("Statement for alice:");
    Console.WriteLine("  version  entry              amount  balance");
    while (rows.Read())
    {
        var (version, entry) = (rows.GetInt64(0), rows.GetString(1));
        var (amount, balance) = (rows.GetDecimal(2), rows.GetDecimal(3));
        Console.WriteLine($"  {version,7}  {entry,-18}{amount,7}{balance,9}");
    }
}

PrintStatement();

Reading a read model is a plain SQL query. The table already has the shape the screen needs, so the query does not fold events or join tables.

Run it

From samples/accounts:

dotnet run --project 4-show-a-statement/fsharp

dotnet run --project 4-show-a-statement/csharp

Both programs print:

Opened for Alice (version 1)
Deposited 100 (version 2)
Withdrew 30 (version 3)
Deposited 50 (version 4)
Rejected: Insufficient funds: 120 available (version 4)

Statement for alice:
  version  entry              amount  balance
        1  Opened for Alice        0        0
        2  Deposit               100      100
        3  Withdrawal            -30       70
        4  Deposit                50      120

The rejected withdrawal has no row: the projection only sees stored events.

Run it again

Rejected: The account is already open (version 4)
Deposited 100 (version 5)
Withdrew 30 (version 6)
Deposited 50 (version 7)
Rejected: Insufficient funds: 240 available (version 7)

Statement for alice:
  version  entry              amount  balance
        1  Opened for Alice        0        0
        2  Deposit               100      100
        3  Withdrawal            -30       70
        4  Deposit                50      120
        5  Deposit               100      220
        6  Withdrawal            -30      190
        7  Deposit                50      240

The projection resumed after version 4, the last event it had committed for Alice, so rows 1 to 4 were not written again. The primary key would reject a repeated row.

Rebuild the statement

The statement holds nothing that the journal does not, so it can be rebuilt. Stop the program, delete the statement's rows and the projection's progress, and run it again:

DELETE FROM statement;
DELETE FROM fcqrs_projection_progress
WHERE projection_name = 'Statement';

The projection starts again from each account's first event and writes every row, then continues with the new commands. Rebuild this way after fixing a bug in the handler or changing the table. Rebuild a read model covers rebuilding beside a model that must stay available.

Next

Step 5: Transfer money moves money from Alice to Bob. A transfer changes two accounts, and each account decides only with its own state, so the transfer needs a workflow that coordinates them: a saga.