Event Sourcing with Rails from scratch
Tags: rails, eventdriven
Published: 2024-03-22 14:48:07

In the previous article **[Introduction to Event Sourcing and CQRS](https://www.smily.com/engineering/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:

```ruby
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:

```ruby
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:

```ruby
...
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:

```ruby
# 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…

```ruby
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:

```ruby
# 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:

```ruby
# 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.

```ruby
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

```ruby
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 

```ruby
  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:

```ruby
# 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 

```ruby
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:

```ruby
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:

```ruby
# 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.

```ruby
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:

```ruby
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](https://github.com/addicted2sounds/event_sourced_ads)

### 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