Cratis Chronicle reducer batch methods

A batch method in Cratis Chronicle’s .NET client updates a read model from one partition’s delivered events in a single call. Use it when the transition needs to consider several events together rather than handle each event separately.

Chronicle delivers events in pages. A partition groups events for one event source, such as an account, and the client reduces each partition separately. The batch contains the observed event types from that partition in the current page, not its entire history. Your method folds those events in sequence order with the current read model and returns the next state.

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.Observation;

[EventType]
public record BatchMoneyDeposited(decimal Amount);

[EventType]
public record BatchMoneyWithdrawn(decimal Amount);

public record AccountBalance(decimal Balance)
{
    [Observes<BatchMoneyDeposited>]
    [Observes<BatchMoneyWithdrawn>]
    public static AccountBalance? Reduce(EventBatch events, AccountBalance? current, BatchContext context) =>
        events.Aggregate(current ?? new AccountBalance(0), (account, e) => e.Event switch
        {
            BatchMoneyDeposited deposited => account with { Balance = account.Balance + deposited.Amount },
            BatchMoneyWithdrawn withdrawn => account with { Balance = account.Balance - withdrawn.Amount },
            _ => account
        });
}

View C# snippet source on GitHub

Kotlin

import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.observation.BatchContext
import io.cratis.chronicle.observation.EventBatch
import io.cratis.chronicle.observation.Observes
import io.cratis.chronicle.observation.Reducer

@EventType
data class BatchAccountOpened(val owner: String)

@EventType
data class BatchMoneyDeposited(val amount: Double)

@EventType
data class BatchMoneyWithdrawn(val amount: Double)

data class BatchAccount(val owner: String, val balance: Double)

@Reducer
class BatchAccountReducer {
    @Observes(BatchAccountOpened::class)
    @Observes(BatchMoneyDeposited::class)
    @Observes(BatchMoneyWithdrawn::class)
    fun reduce(events: EventBatch, current: BatchAccount?, context: BatchContext): BatchAccount? =
        events.fold(current) { account, e ->
            when (val event = e.event) {
                is BatchAccountOpened -> BatchAccount(event.owner, 0.0)
                is BatchMoneyDeposited -> account?.copy(balance = account.balance + event.amount)
                is BatchMoneyWithdrawn -> account?.copy(balance = account.balance - event.amount)
                else -> account
            }
        }
}

View Kotlin snippet source on GitHub

Java

import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.observation.BatchContext;
import io.cratis.chronicle.observation.EventAndContext;
import io.cratis.chronicle.observation.EventBatch;
import io.cratis.chronicle.observation.Observes;
import io.cratis.chronicle.observation.Reducer;

@EventType
record BatchAccountOpened(String owner) {}

@EventType
record BatchMoneyDeposited(double amount) {}

@EventType
record BatchMoneyWithdrawn(double amount) {}

record BatchAccount(String owner, double balance) {}

@Reducer
class BatchAccountReducer {
    @Observes(BatchAccountOpened.class)
    @Observes(BatchMoneyDeposited.class)
    @Observes(BatchMoneyWithdrawn.class)
    public BatchAccount reduce(EventBatch events, BatchAccount current, BatchContext context) {
        BatchAccount account = current;
        for (EventAndContext e : events) {
            if (e.getEvent() instanceof BatchAccountOpened opened) {
                account = new BatchAccount(opened.owner(), 0);
            } else if (e.getEvent() instanceof BatchMoneyDeposited deposited && account != null) {
                account = new BatchAccount(account.owner(), account.balance() + deposited.amount());
            } else if (e.getEvent() instanceof BatchMoneyWithdrawn withdrawn && account != null) {
                account = new BatchAccount(account.owner(), account.balance() - withdrawn.amount());
            }
        }
        return account;
    }
}

View Java snippet source on GitHub

Elixir

defmodule BatchAccountOpened do
  use Chronicle.Events.EventType, id: "batch-account-opened"

  defstruct owner: ""
end

defmodule BatchMoneyDeposited do
  use Chronicle.Events.EventType, id: "batch-money-deposited"

  defstruct amount: 0
end

defmodule BatchMoneyWithdrawn do
  use Chronicle.Events.EventType, id: "batch-money-withdrawn"

  defstruct amount: 0
end

defmodule BatchAccount do
  use Chronicle.ReadModels.ReadModel

  defstruct owner: "", balance: 0
end

defmodule BatchAccountReducer do
  use Chronicle.Reducers.Reducer, model: BatchAccount

  # Called once per account in a page, with its events and its stored state.
  @observes BatchAccountOpened
  @observes BatchMoneyDeposited
  @observes BatchMoneyWithdrawn
  def fold(events, current, _context) do
    Enum.reduce(events, current, fn
      {%BatchAccountOpened{owner: owner}, _context}, _account ->
        %BatchAccount{owner: owner}

      {%BatchMoneyDeposited{amount: amount}, _context}, account ->
        %{account | balance: account.balance + amount}

      {%BatchMoneyWithdrawn{amount: amount}, _context}, account ->
        %{account | balance: account.balance - amount}
    end)
  end
end

View Elixir snippet source on GitHub

TypeScript

TypeScript has no self-reducing read models, so the batch method lives on a reducer class.

// Copyright (c) Cratis. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

import { BatchContext, EventBatch, eventType, observes, reducer } from '@cratis/chronicle';
import { field } from '@cratis/fundamentals';

@eventType()
class BatchMoneyDeposited {
    @field(Number) readonly amount: number;

    constructor(amount: number = 0) {
        this.amount = amount;
    }
}

@eventType()
class BatchMoneyWithdrawn {
    @field(Number) readonly amount: number;

    constructor(amount: number = 0) {
        this.amount = amount;
    }
}

class AccountBalance {
    @field(Number) balance = 0;
}

@reducer('account-balance', undefined, AccountBalance)
class AccountBalanceReducer {
    @observes(BatchMoneyDeposited)
    @observes(BatchMoneyWithdrawn)
    reduce(events: EventBatch, current: AccountBalance | undefined, _context: BatchContext): AccountBalance {
        return events.events.reduce((account, entry) => {
            if (entry.is(BatchMoneyDeposited)) return { balance: account.balance + entry.event.amount };
            if (entry.is(BatchMoneyWithdrawn)) return { balance: account.balance - entry.event.amount };
            return account;
        }, current ?? { balance: 0 });
    }
}

The batch method is called once per event source in a page, with that event source’s events and its stored state. Return undefined to remove the read model. Declare @batchSize({ maxConcurrency }) on the reducer to limit how many event sources of a page are reduced at the same time; the default is the number of processors.

View TypeScript snippet source on GitHub

Declaring a batch method

The account example applies deposits and withdrawals to the balance so far. Each item exposes the event through Event, allowing the method to choose the transition by event type.

A batch method takes an EventBatch first and the nullable current read model second, and returns the read model or a task of it. The current model is null before the partition has state; returning null removes the read model. Declare each observed event type with [Observes<TEvent>]. Optional parameters can receive BatchContext, CancellationToken, and services.

On a self-reducing read model the batch method is a public static method named Reduce. On a separate reducer implementing IReducerFor<TReadModel> it can have any name.

A reducer has per-event methods or one batch method, never both, because splitting its events between methods would break the fold. Chronicle refuses to register a reducer that declares both, more than one batch method, or a batch method that does not take and return the read model.

Reporting outcomes

Throwing or calling Fail on the batch fails the partition without writing its new state. Calling Defer requests delivery again after the delay you supply, without failing the partition or writing state. Deferral requires kernel support; without it, the client reports a failure instead.

Other partitions in the page can still succeed. The client reduces partitions in parallel, while keeping each partition’s events in order. Keep state changes local to the partition rather than relying on another partition finishing first.

For the per-event alternative and replay behavior, continue with reducer event processing.