Event Sourcing in Rust

Rust is an incredible language. Event sourcing is a wonderful pattern. Combine the two and you have something beautiful.

Event sourcing & CQRS can be implemented in a so many different ways and I believe no one can really become an expert at the art of it. But getting started doesn’t have to be hard!

Sadly, there’s many existing libraries which lived and died in a short period of time, and a few slightly active ones but only being updated once month or more.

The purpose of this blog post, is to introduce a project I’ve been working on full time called Thalo.

GitHub - thalo-rs/thalo: Event sourcing for microservices.

It’s a Rust library providing everything you need to build event sourced systems.

Currently, Thalo provides:

• Core crate with basic aggregate, event, event store & event stream traits, along with some derive macros
• Testing library (Given-When-Then)
• Postgres, in-memory & file event stores
• Kafka event stream

Example cli event store

The gif above is from an example on the Thalo repository, with commands being send from a client to a server through gRPC with an event store is in-memory.

You can run this example by cloning the project and running the client & server in separate terminal tabs.

$ git clone git@github.com:thalo-rs/thalo.git && cd thalo
$ cargo run -p example-protobuf --bin server
# In separate terminal tab
$ cargo run -p example-protobuf --bin client

Writing the aggregates isn’t much work thanks to the derive macros provided by Thalo, but I still felt like something was missing. I’ve looked into CloudEvents and AsyncAPI as options for defining aggregate schemas, but they don’t seem to be great at doing so.

ESDL

Event-sourcing Schema Definition Language is a schema language for defining aggregates, commands, events & types which can be used to generate code in Rust greatly simplifying boilerplate for aggregates.

GitHub - thalo-rs/esdl: Event Sourcing Schema Definition Language

Let me introduce you to the syntax.

bank-account.esdl

It’s quite self explanatory if you’re familiar with GraphQL, but let’s go through it line by line.

• aggregate BankAccount {
  An esdl file always must define exactly one aggregate, in our case named BankAccount.
• open_account(initial_balance: Float!): OpenedAccount!
  We define a command called open_account which takes a required float for the initial balance, and results in an OpenedAccount event.
  Note the ! this means the type is required and cannot be undefined / null / None .
• deposited_funds(amount: Float!): DepositedFunds!
withdrew_funds(amount: Float!): WithdrewFunds!
  As described above, these are also commands which take an amount and return an event.
• event OpenedAccount {
  Here we define the OpenedAccount event and it’s available fields.
• initial_balance: Float!
  The opened account event has an initial balance with a required float.
• … the rest is quite self explanatory.

At this point we have an awesome Rust library for event sourcing, and a schema language for defining aggregates. If we combine the two, we can use the esdl file for the schema, and Rust for the implementation through code generation.

To generate Rust code from this esdl file, we can use a build.rs file and compile it using the esdl crate.

// build.rs
fn main() -> Result<(), Box<dyn std::error::Error>> {
    esdl::configure()
        .add_schema_file("./bank-account.esdl")?
        .compile()?;
    Ok(())
}

And in our Rust code, we can import it using the macro provided by Thalo.

thalo::include_aggregate!("BankAccount");

This will include the generated code which includes:

• trait BankAccountCommand
• enum BankAccountEvent
• struct OpenedAccountEvent
• struct DepositedFundsEvent
• struct WithdrewFundsEvent

From this, we can begin to implement the commands.

use thalo::aggregate::{Aggregate, TypeId};
#[derive(Aggregate, Clone, Debug, Default, PartialEq, TypeId)]
pub struct BankAccount {
    id: String,
    opened: bool,
    balance: f64,
}
impl BankAccountCommand for BankAccount {
    type Error = ();
    fn open_account(&self, initial_balance: f64)
        -> Result<OpenedAccountEvent, Self::Error> { ... }
    fn deposit_funds(&self, amount: f64)
        -> Result<DepositedFundsEvent, Self::Error> { ... }
    fn withdraw_funds(&self, amount: f64)
        -> Result<WithdrewFundsEvent, Self::Error> { ... }
}

And then the apply function to update aggregate state.

fn apply(bank_account: &mut BankAccount, event: BankAccountEvent) {
    use BankAccountEvent::*;
    match event {
        OpenedAccount(OpenedAccountEvent { initial_balance }) => {},
        DepositedFunds(DepositedFundsEvent { amount }) => {},
        WithdrewFunds(WithdrewFundsEvent { amount }) => {},
    }
}

Thalo & ESDL will give you a head start in writing clean & consistent aggregates for your event sourced systems.

I’m actively working on Thalo full time and hope to release something substantial in the coming months, so stay tuned!

If you’d like to get in touch with me, I have links on my Github profile, or you can become a member of the Thalo Discord server.
https://discord.gg/4Cq8NnPYPA

Ari Seyhun — Github