Forem: Oleg Borys

Event Sourcing with Rails from scratch

Oleg Borys — Fri, 22 Mar 2024 14:48:07 +0000

In the previous article Introduction to Event Sourcing and CQRS we got familiar with the main concepts of Event Sourcing and reviewed the cons and pros of this approach.

Implementing Event Sourcing in Rails can be a powerful way to handle complex business logic and maintain a reliable audit trail of changes to your application's state. Event Sourcing involves storing a sequence of events that represent changes to the state of your application over time.

Now, before we plummet into using Event Sourcing in Rails with help or battle-tested solutions, let's research the basics and learn how to implement things from scratch. However, be aware - you should think twice before using custom implementation (too many chances you going to miss some small detail and the consequences may be huge).
For these purposes let's create a simple application for listing rental advertisements on the website.

The process may be quite complicated, but for the sake of simplicity, let's assume it goes as below:

the ad listing is created
the content is updated
the listing goes published on the website
ad listing removed

Preparations

where we start building a path in our journey and getting familiar with events

So, let's create a new project for our implementation purposes first:

> rails new event_sourced_ads --database=postgresql --skip-test

Skipping tests, since I’m a rspec fan so we'll use that:

gem "rspec-rails", "~> 6.1.0"

Then bundle and rails generate rspec:install.

Now, once the initial setup is done we are going to use events. Let’s implement that part. And guess what we’re starting with:

dclass CreateEvents < ActiveRecord::Migration[7.1]
  def change
    create_table :events, id: :uuid do |t|
      t.string :stream_name
      t.string :event_type
      t.jsonb :data

      t.timestamps
    end
  end
end

We use UUID here for primary keys. When dealing with distributed systems and the need for worldwide uniqueness, opting for a UUID could be the optimal decision.

We need streams to separate events related to certain entities. Streams are needed to group events of a particular kind. In our case, I’m going to group events related to one single ad, so those can be easily fetched. Frankly speaking, it’s not a good idea to store stream names like that, since the events may be in different streams. For the sake of simplicity, let’s consider doing some evil (we’ll do more till the end).

We need some basic event class that we can publish and verify input with, also we need some way to use pub-sub (quite a crucial part). I’m excited about dry-rb stack - I can’t say it’s perfect, but it usually perfectly suits all my needs, so I'm turning on the imagination and seeing what it brings… ImaginationCompleted.publish(data: {idea: "Create BaseEvent", pub_sub: "KISS rails has one built-in"})

Who am I to argue with that 😇 Let’s start with a spec:

...
it "persists an event record" do
  expect { publish }.to change { Event.count }.by(1)
  expect(Event.last).to have_attributes(
    event_type: "FakeEvent",
    data: {"name" => "whatever"},
    stream_name: "123123"
  )
end

it "sends a notification" do
  allow(ActiveSupport::Notifications).to receive(:instrument)
  publish
  expect(ActiveSupport::Notifications).to have_received(:instrument).with(
    "FakeEvent", data: {name: "whatever"}, stream_name: "123123"
  )
end
...

and after some struggle, we come up with:

# lib/events/base_event.rb
module Events
  class BaseEvent
    class InvalidAttributes < StandardError; end

    class MissingContract < StandardError; end

    attr_reader :data

    def self.schema(&block)
      inner_schema = block.call
      define_method(:params_schema) do
        Dry::Schema.Params do
          required(:data).hash(inner_schema)
        end
      end
    end

    def self.publish(**args)
      new(**args.slice(:data)).publish(stream_name: args[:stream_name])
    end

    def initialize(**args)
      validate_input(args)
      @data = args[:data]
    end

    def publish(stream_name: nil)
      Event.create!(
        event_type: self.class.name, data:, stream_name:
      )
      ActiveSupport::Notifications.instrument(self.class.name, data:, stream_name:)
      self
    end

    def params_schema
      ->(_) { raise MissingContract, "Contract needs to be implemented" }
    end

    def validate_input(args)
      data_validation = params_schema.call(args)
      raise InvalidAttributes.new(data_validation.errors.to_h) if data_validation.errors.any?
    end
  end
end

And let’s try using our new pet. You know where to start…

RSpec.describe Events::AdCreated do
  describe ".publish" do
    subject(:publish) do
      described_class.publish(
        data: {title: "Some title", body: "Some description"},
        stream_name: "123456789",
      )
    end

    it "persists the event in database" do
      expect { publish }.to change { Event.count }.by(1)
      expect(Event.last).to have_attributes(
        event_type: "Events::AdCreated",
        data: {
          "title" => "Some title",
          "body" => "Some description"
        },
        stream_name: "123456789"
      )
    end
  end
end

and the event itself:

# lib/events/ad_created.rb
class Events::AdCreated < Events::BaseEvent
  schema do
    Dry::Schema.Params do
      required(:title).filled(:string)
      required(:body).filled(:string)
    end
  end
end

Aggregate part

the one where we learn to manipulate our ads

So, as user we should be able to create new ad, possibly modify that and publish. However, we shouldn’t be able to edit already published ad. So we need some consistency in actions and having corresponding event published after the action is executed. That’s where the aggregate comes to place.

So, we create AdAggregate class and start with test for the new instance:

# spec/services/ad_aggregate_spec.rb
RSpec.describe AdAggregate do
    it "has valid attributes on initialization" do
    expect(aggregate).to have_attributes(
      id: kind_of(String),
      state: :new
    )
  end
# app/services/ad_aggregate.rb
class AdAggregate
  attr_reader :id, :attributes, :state

  def initialize(id = nil)
    @id = id || SecureRandom.uuid
    @state = :new
  end
end

Next we need possibility to actually create new draft and have those attributes in the aggregate. Also we need to publish an event that the draft is created.

describe "#create_draft" do
  subject(:create_draft) { aggregate.create_draft(**attributes) }

  context "with valid attributes" do
    let(:attributes) { valid_attributes }

    it "updates attributes and state" do
      create_draft
      expect(aggregate).to have_attributes(
        attributes: {
          title: "Test title",
          body: "Test description"
        },
        state: :draft
      )
    end
  end
end

The one is easy to implement, but we face a problem here. We should be able to restore the state of the aggregate later when we want to apply next actions to that. The aggregate is supposed to be event sourced one. So we need a way to apply events to that and all we should actually do here is to apply an event

def create_draft(title:, body:)
  apply Events::AdCreated.new(data: {ad_id: id, title:, body:})
end

We need a handler in the aggregate to understand how we modify the attributes, how the state is changed and how we can restore the state of an aggregate from history of events (for this purpose we’ll create another class in a while 😉). Also, the events should be published when we store the aggregate. For these purposes, let’s add handler methods to explain how we want to modify aggregate’s state on event and common method that will also create a queue of unpublished events

  def unpublished_events
    @unpublished_events ||= []
  end

  def apply_event(event)
    send("apply_#{event.class.name.demodulize.underscore}", event)
  end

    private 

    def apply(event)
    unpublished_events << event
    apply_event(event)
  end

  def apply_ad_created(event)
    @state = :draft
    @attributes = event.data.slice(:title, :body)
  end

So, when the new event is applied we save that in a queue of unpublished events and call the corresponding handler. But what’s the sense of that without having the events stored? How to fetch previously created aggregate? We could implement that here in this class, though according Single Responsibility Principle, it’s definitely a work that someone else should do. That’s where we need a repository:

# frozen_string_literal: true

require "rails_helper"

RSpec.describe Repository do
  describe '.load' do
    subject(:load) { described_class.load(aggregate_class, stream_name) }

    let(:aggregate_class) { AdAggregate }
    let(:stream_name) { SecureRandom.uuid }

    context "without events" do
      it "loads new aggregate" do
        expect(load).to be_instance_of(aggregate_class).and have_attributes(
          id: stream_name,
          state: :new
        )
      end
    end

    context "with existing events" do
      context "when applying AdCreated event" do
        before do
          Event.create(
            event_type: "Events::AdCreated", stream_name:,
            data: {ad_id: stream_name, title: "title", body: "body"}
          )
        end

        it "applies event to aggregate" do
          expect(load).to be_a(AdAggregate).and have_attributes(
            id: stream_name,
            state: :draft,
            attributes: {
              title: "title",
              body: "body"
            }
          )
        end

        context "when applying AdPublished" do
          before do
            Event.create(
              event_type: "Events::AdPublished", stream_name:,
              data: {ad_id: stream_name, remote_id: "xosfjoj"}
            )
          end

          it "applies event to aggregate" do
            expect(load).to be_a(AdAggregate).and have_attributes(
              id: stream_name,
              state: :published,
              attributes: {
                title: "title",
                body: "body"
              }
            )
          end
        end
      end
    end
  end

  describe '.store' do
    subject(:store) { described_class.store(aggregate) }

    context "with unpublished events" do
      let(:aggregate) do
        instance_double(AdAggregate, id: stream_name, unpublished_events: [event])
      end
      let(:stream_name) { SecureRandom.uuid }
      let(:event) do
        Events::AdCreated.new(data: {ad_id: stream_name, title: "title", body: "body"})
      end

      it "publishes pending events" do
        expect { store }.to change { Event.count }.by(1)
        expect(Event.last).to have_attributes(
          stream_name:,
          event_type: "Events::AdCreated",
          data: {
            "ad_id" => stream_name,
            "title" => "title",
            "body" => "body"
          }
        )
      end
    end
  end
end

and the implementation of that is easy enough. I’ll omit description of that to save some precious space and time

module Repository
  extend self

  def load(aggregate_class, stream_name)
    events = Event.where(stream_name:).map do |event|
      event.event_type.constantize.new(data: event.data)
    end
    aggregate_class.new(stream_name).tap do |aggregate|
      events.each do |event|
        aggregate.apply_event(event)
      end
    end
  end

  def store(aggregate)
    aggregate.unpublished_events.each do |event|
      event.publish(stream_name: aggregate.id)
    end
  end
end

We do have a possibility to store the aggregate to load that from existing events. However, we are missing one of the main purposes for the aggregate. We should disallow editing already published ads, also we definitely can’t publish the same ad twice (well technically we can, but for sure that’s wrong). So, as usually:

describe "#update_content" do
  subject(:update_content) { aggregate.update_content(**new_attributes) }

  let(:aggregate) { described_class.new }
  let(:new_attributes) do
    {title: "Updated title", body: "Updated description"}
  end

  context "when ad is in draft state" do
    before { aggregate.create_draft(**valid_attributes) }

    it "updates ad attributes" do
      update_content
      expect(aggregate).to have_attributes(
        attributes: {
          title: "Updated title",
          body: "Updated description"
        },
        state: :draft
      )
    end
  end

  context "when ad is in published state" do
    before do
      aggregate.create_draft(**valid_attributes)
      aggregate.publish
    end

    it "raises an error" do
      expect { update_content }.to raise_error(described_class::AlreadyPublished)
    end
  end
end

describe "#publish" do
  subject(:publish) { aggregate.publish }

  let(:aggregate) { described_class.new }

  context "when ad is in draft state" do
    before { aggregate.create_draft(**valid_attributes) }

    it "updates state to published" do
      publish
      expect(aggregate.state).to eq(:published)
    end
  end

  context "when ad is in published state" do
    before do
      aggregate.create_draft(**valid_attributes)
      aggregate.publish
    end

    it "raises an error" do
      expect { publish }.to raise_error(described_class::AlreadyPublished)
    end
  end
end

You can check the implementation of the methods in the repository and it’d be a good idea to try implementing that by yourself 😉

CQRS part

the one where we get familiar with read models and presentation to users

Ok, pub is ready, now it’s time to ~~drink some beer~~ have sub part:

# config/initializers/event_listeners.rb
Rails.application.config.after_initialize do
  {
    AdEventListener: [
      Events::AdCreated,
    ]
  }.each do |listener, events|
    events.each { |event| ActiveSupport::Notifications.subscribe(event.to_s, listener.to_s.constantize) }
  end
end

So, here we are going to rule where events happen to be. In the example, AdEventListener will get the ActiveSupport event we broadcast with BaseEvent and send a call to our listener. Perfect… but not exactly what we need.

class ApplicationEventListener
  def self.call(event)
    public_send(
      "apply_#{event.name.demodulize.underscore}", **event.payload
    )
  end
end

and now we should be able to create listeners in a very convenient form:

class AdEventListener < ApplicationEventListener
  class << self
    def apply_ad_created(data:, stream_name:)
      Ad.create!(id: stream_name, **data)
    end
  end
end

🤔 …but stop, something’s wrong here. What’s Ad.create!? We don’t have that implemented… The part is omitted for a reason.

What we implemented above is a CQRS system and Ad is a read model. The structure of that is not important and should suit your needs. In this example project I’ve implemented Events::AdModified, Events::AdPublished, Events::AdRemoved. You can get familiar with the project

Retrospective part

the one where we look over what we did

We’ve just implemented an application using Event Sourcing from scratch. I definitely would recommend to stay away from self-made solutions in production. Several simplifications were made (but you may need those once your project grows). Anyway, it’s good to know what‘s inside the black-box (gem) you use.

In the upcoming articles we’re going to play with some recognized instruments to implement event sourced applications

Introduction to Event Sourcing and CQRS

Oleg Borys — Wed, 13 Dec 2023 23:36:17 +0000

In a galaxy far, far away, enter the saga of CQRS and event sourcing, where data updates unfold like an epic space opera. Jokes away, let's see what are those

Event sourcing is not a new term. Event Sourcing is a powerful paradigm for managing data in the ever-evolving software development landscape. While it might not be as widely known as other data management methods, Event Sourcing offers a unique approach that can benefit applications significantly.

Command Query Responsibility Segregation (CQRS) is an architectural paradigm that divides the responsibilities of read and write operations within a system.

At its heart is the notion that you can use a different model to update information than the model you use to read information Martin Fowler

In the realm of CQRS application architecture, the system is separated into two distinct segments. The first caters to the realms of updates and deletions – a territory we call the writing model. Concurrently, the second segment takes on the noble task of reading – aptly named the read model. Unlike the conventional CRUD approach, which relies on a single database, CQRS embraces two databases (or at least different tables). Each side is obsessed with the act of reading or writing.

To address scalability concerns, CQRS can be coupled with event sourcing. In this scenario, events generated by commands are stored in an event store. These events are subsequently asynchronously transmitted to a separate read data store, undergoing a transformation process to align with the read data model. This integration helps overcome some of the scalability challenges inherent in CQRS.

In this article, we will delve into the fundamentals of Event Sourcing coupled with CQRS, explore its advantages, and discuss when it might be better to avoid using this approach.

What is event-sourcing

If you work with technology, you must have come across it. It's a puissant tool used by many large organizations for data modeling. It can scale and meet the needs of the modern data processing industry.

Event sourcing is a compelling architectural pattern that might initially seem a bit eccentric. Instead of focusing on the state of your system, it keeps track of every change that happens. It's like holding a detailed diary of every emotional rollercoaster it goes through. All the events that are changing the state of your system are recorded, and such records serve as both a source for the current state and an audit trail of what has happened in the application over its lifetime.

Domain experts usually describe their systems as a collection of entities, which are containers for storing a state and events, representing changes to entities as a result of processing input data within various business processes. Events are often triggered by commands invocated by users, background processes, or integrations with external systems. The naming is usually defined by Ubiquitous Language.

Ubiquitous Language is a domain-driven design (DDD) concept, a software development approach. Imagine Ubiquitous Language as the magical Babelfish of software development. It's the secret code that lets developers, domain experts, and stakeholders all speak the same lingo, like a universal translator for geeks and business folks. With this shared language, the jargon barrier becomes a thing of the past, and everyone can boogie on the same wavelength.

Among existing practices to define the ubiquitous language, I'd emphasize Event-storming. However, sometimes it may not be the best option (team or resource constraints, limited initial knowledge, etc). Then, the process may be neglected, and developers apply the most widely used terms.

Many architectural patterns treat entities as a primary concept. These patterns describe how to store them, how to access them, and how to modify them. Within this architectural style, events are often "on the side": they are the consequences of entity changes. Unlike the traditional systems, with an event-sourcing approach, the events are considered the only source of truth.

At first glance, it may sound unusual, but most of the serious systems we know and interact with do not emphasize the concept of the current state or the final state (financial, banking, and many others). As Greg Young (creator of CQRS) said in one of his speeches, your bank account is not just a column in a table but the sum of all transactions that occurred with your bank account (renewals, write-offs, and recovery). For example, if you have a disagreement with your bank, the balance sheet and the bank confirm that the balance is 69 dollars, but your position is still 96. You won't hear a response like "The column states it's 69 dollars, and it's all you have".

Designing systems with a focus on events and event logs provides the following benefits:

It helps reduce impedance mismatches and the need for concept mapping, allowing technology teams to "speak the same language" (ubiquitous language) as the business when discussing the system.
Encourages separation of responsibility into commands and queries (command/query responsibility segregation), allowing you to optimize writing and reading independently of each other.
It provides temporality and a history of change, allowing questions to be answered about what the system looked like at specific points in the past and what events occurred before that point.

Key Concepts of Event Sourcing

Events

Events are the fundamental building blocks of Event Sourcing. They represent discrete changes in the state of an entity. Each event is immutable and contains information about what happened when it occurred and any relevant data associated with the change.

You can register as many events as you like. The name of the event should have clear semantics. Events are always talking about the past and can reveal with their names what has changed and how it has changed (for example, OrderCanceled, OrderItemAdded, ProductRemoved, OrderPlaced).

Event Store

The Event Store is the central repository for storing events in the order they occurred. It ensures that events are appended to the end of the log and provides methods to read, write, and query events.

Aggregates

Imagine Aggregates as a consistency boundary around a group of domain objects, such as entities and value objects. In Event Sourcing, Aggregates are reconstituted from a single fine-grained event stream (e.g., representing an order flow). During this operation, the current state of the aggregate is calculated so that it can be used to handle a command.

The state is a crucial part of an Aggregate toolkit. It's like their memory is wiped clean after every command unless they're into some fancy snapshotting business. Consider it a superhero's utility belt, equipped to deal with duplicate commands and other unexpected villains, thanks to at-least-once delivery guarantees. It's not playing movies; it's making decisions.

Projection

Projections are used to derive the current state of an entity or a view of the data by replaying events. Projections are separate from the Event Store and can be optimized for specific query needs.

An integral part of Event Sourcing is the concept of snapshots – intermediate state captures of an aggregate in the EventStore. Sometimes, to obtain the final state of an object, you need to replay many events, starting from the very first one. To optimize this process, snapshots are taken, and we will restore the final state not from the very first event in the system but from the latest snapshot. Despite the apparent benefit of this pattern, it should only be applied when obtaining the final state takes excessive time. Many might think that a downside is the exponential growth of data volume with this approach, but that's not the case. Events are storing not the complete data models but only the state changes. In reality, the slowdown in such a system is associated with the construction of aggregates from all domain events that have occurred until now. So, Event Sourcing is the storage of a series of events, and the data schema that reflects these events is a direct derivative of these events. The data schema in Event Sourcing systems is temporary and can be rebuilt or reconstructed from events at any time. Isn't it like a time machine?

Benefits of Event Sourcing: The Fun Side of Managing Data

Event sourcing coupled with CQRS promotes decentralized modification and reading of data. This architecture scales well and is suitable for systems that already work with event processing or want to migrate to such an architecture.

Complete Audit Trail: With Event Sourcing, you've got a complete record of your software's shenanigans. It's the ultimate way to check if your software's been up to no good or just having a few harmless adventures.
Temporal Querying: It's like going back in time to see what your software was thinking and doing at a specific moment. Wondering why it behaved oddly on a Tuesday six months ago? Event Sourcing has your back.
Parallel Processing: Event-sourcing turns your software into a multitasking genius. It can handle multiple tasks simultaneously, like a magician juggling flaming torches. Events are processed asynchronously, so your software can easily handle high loads.
Distributed Systems: In a distributed system, events can be processed asynchronously and independently, leading to improved scalability. Each component or microservice can consume and process events at its own pace without blocking others.
Flexibility: Event-sourcing is like a chameleon for data – it can adapt to different situations. It allows you to create various "disguises" for your data, tailored to specific needs. You could achieve the same with other approaches (like Postgres materialized views or building your own reports). However, event-sourcing allows you to achieve the same in a very natural way.
Fault Tolerance: Your software's history is safe and sound. In case of mishaps, you can simply turn back the clock and replay events to restore order in your software universe.

When to Avoid Event Sourcing: Not Every Party Needs a Diary

As much as we'd love to make Event Sourcing the life of the party, it's not always the best guest for every occasion. Here's when it's better to opt for something else:

Simplicity: If your application is as uncomplicated as a one-page novel, Event Sourcing might be like using a sledgehammer to crack a walnut. It's great for epic tales but overkill for short stories.
Performance: Event Sourcing might seem too leisurely for applications that demand real-time updates. Due to eventual consistency, the data may be slightly outmoded
Overhead: In read-focused systems, traditional databases may be more suited for your "need-it-now" attitude (building the projections would require some extra computing power)
Learning Curve: Implementing Event Sourcing can be like learning a new language. It's not something you can master overnight, so if your project is on a tight schedule or your team is new to this, be prepared for a learning adventure.

In short, shorter, shortest…

Event Sourcing is like the eccentric uncle who turns up quirky but surprisingly insightful at family gatherings. It offers a unique approach to managing data, providing complete auditability, temporal querying, scalability, flexibility, and fault tolerance. However, like any eccentric guest, it might not be the best fit for every occasion. So, choose Event Sourcing when the story is epic and complex, but feel free to opt for traditional methods when your tale is short and sweet. After all, the software world is diverse, and there's room for both diary keepers and straightforward note-takers. In the upcoming article, I'll share practical hints on implementing event-sourcing and CQRS in the Rails world.