Allow persistent projections to be built/updated with Active Record and Aggregate Root

[brentsnook]

We'd like to build Active Record model projections from our events to allow things to be queried efficiently from APIs etc without needing to rebuild the projection for each request. Has anyone considered building something like this on top of AggregateRoot to allow that to be mixed into an ActiveRecord model and have that model updated as new events are added to a stream? Or is there another way to do this?

This could look something like this (maybe fired from event handlers for the different events):

Order.find(uuid:).update_projection!

A projection record could store a polymorphic reference to the model, the stream it is projected from and the last seen event ID:

class Projection < ApplicationRecord
  belongs_to :record, polymorphic: true
  # stream : string 
  # last_event_id : uuid
end

The aggregate behaviour could then use it to apply only unseen events:

module AggregateRootActiveRecord # or ProjectedAggregateRoot?
  has_one :projection

  delegate :last_event_id, stream:, :last_event_id=, to: :projection
  after_initialize :add_projection

  def update_projection!
    ActiveRecord::Base.transaction do
      apply_unseen
      projection.update!(last_event_id:)
      save!
    end
  end

  private

  def apply_unseen
    AggregateRoot::Repository.new.load(self, stream)
  end

  def add_projection
    self.projection ||= Projection.new(
      record: self,
      stream: stream_name
    )
  end
end

And used in an AR model:

class Order < ApplicationRecord
  include AggregateRootActiveRecord

  on OrderSubmitted do |event|
    self.state = :submitted
    self.delivery_date = event.data.fetch(:delivery_date)
  end

  on OrderExpired do |event|
    self.state = :expired
  end

  private

  def stream_name
    "orders$#{uuid}"
  end
end

This would also require some changes to the AggregateRoot::Repository:

module AggregateRoot
  class Repository
    ...
    def load(aggregate, stream_name)
      event_store.read.stream(stream_name).after(aggregate.last_event_id).reduce do |_, ev|
        aggregate.apply(ev)
        aggregate.last_event_id = ev.event_id # or should #apply be doing this instead?
      end
      aggregate.version = aggregate.unpublished_events.count - 1
      aggregate
    end
  end
end

If there's some interest in this I can put up a spike PR, otherwise we'd probably look to implement something similar just internally.

[mostlyobvious]

That seems convenient, although it needs some clear conventions how things are tied together. I think it is more suitable to a framework rather than a general toolkit. Or perhaps it needs some escape hatches to rewire the conventions/configuration.

You want an ActiveRecord read model, re-built from a projection. Projection runs on a stream. That's first convention — parameterized matching of streams to read models.

On the other hand subscriptions trigger on events. Thus reactively updating the read model would require subscribing to all event types to be found in a projection stream. Perhaps that signals that a hypothetical read model DSL could declare event types it is interested in. It would have been easier to be updated on stream changes but it is what it is.

Not direct answers but good reads on read models from my former colleagues.
I also vaguely remember ecommerce having at some point an implementation of read-model update mechanism, perhaps @pjurewicz could help here more.

[brentsnook]

@mostlyobvious thanks for the reply, I can have a go at writing a separate proof of concept gem for this as a next step.

Here's a summary of what I'm thinking so far:

• add a module that can be easily mixed into an active record module (something like ActiveRecordProjection)
• build on AggregateRoot by having this module also include that one - I much prefer the on syntax to normal projections
• store a separate Projection record that tracks last seen event ID, stream and polymorphic reference to the AR record
• allow unseen events to be applied much in the same way as load in AggregateRoot::Repository but:
  • use after to only read/apply the events since the last seen
  • store a reference to the last seen/applied event on the projection
• allow asynchronous event handlers to automatically be registered to update the projections on each new event
  • override the on method to also register any event classes across all projections
  • register an single async event handler to update all projections for this active record model on registered events
    • store and use the deregistration callback to replace the event handler each time
      • this is the best way I could think of to ensure we can register a single handler for all events as well as possibly cater to lazy loading/reloading in dev
  • the event handler looks up streams via streams_of, finds all projections using them and then updates the projections

Use in a model could be something like:

class Order < ApplicationRecord
  include ActiveRecordProjection

  on OrderSubmitted do |event|
    self.state = :submitted
    self.delivery_date = event.data.fetch(:delivery_date)
  end

  on OrderExpired do |event|
    self.state = :expired
  end

  private

  def stream_name
    "orders$#{uuid}"
  end
end

Projection might look like:

class Projection < ApplicationRecord
  belongs_to :record, polymorphic: true
  # stream : string 
  # last_event_id : uuid

  def self.update_all(event)
    streams = Rails.event_store.streams_of(event.id).map(&:name)
    where(stream: streams).each(&:update_projection!)
  end

  def update_projection!
    ActiveRecord::Base.transaction do
      record.update_projection!
      save!
    end
  end
end

The mixin/concern for the AR model:

# or ProjectedAggregateRoot / ActiveRecordAggregateRoot / ActiveRecordProjection
module ActiveRecordProjection
  extend ActiveSupport::Concern

  included do
    include AggregateRoot
    has_one :projection

    delegate :last_event_id, stream:, :last_event_id=, to: :projection
    after_initialize :add_projection

    def self.on(*event_klasses, &block)
      ProjectedEventRegistry.register(event_klasses)
      super(event_klasses, &block)
    end
  end

  def update_projection!
    ActiveRecord::Base.transaction do
      apply_unseen
      save!
    end
  end

  private

  def apply_unseen
    event_store.read.stream(stream).after(last_event_id).reduce do |_, ev|
      apply(ev)
      self.last_event_id = ev.event_id
    end
  end

  def add_projection
    self.projection ||= Projection.new(
      record: self,
      stream:
    )
  end
end

Registering projected events:

class ProjectedEventRegistry
  @@event_classes = Set.new
  cattr_accessor :event_classes, :unsubscribe_handler

  def self.register(event_classes)
    self.event_classes.merge(event_classes)
    unsubscribe_handler&.call
    self.unsubscribe_handler = Rails.event_store.subscribe(UpdateProjections.new, to: self.event_classes)
  end
end

The update projections handler:

class UpdateProjections < ActiveJob::Base
  prepend RailsEventStore::AsyncHandler

  def perform(event)
    Projection.update_all!(event)
  end
end

Does this sound like it's on the right track?

[brentsnook]

I've now published a first cut of this to https://github.com/brentsnook/active_record_projection

[mostlyobvious]

allow unseen events to be applied much in the same way as load in AggregateRoot::Repository but:

• use after to only read/apply the events since the last seen
• store a reference to the last seen/applied event on the projection

This would work only in the following conditions:

• global stream with linearized writers

• named stream with optimistic concurrency control (using expected version with integer positions)

• any configuration on sqlite (only single writer thread there)

In other situations, there's a race condition coming from the fact that PostgreSQL and MySQL have multiple writers (MVCC). Global positions coming from auto-incremented primary key are not reliable for ordering due to the way how they're assigned across transactions and concurrent writers. This issue would manifest as skipped events under high concurrency.

The problem is very well described in an article from Oskar. There's a couple of possible solutions, all having some kind of drawback in legacy systems with high chance of long transactions.

As long as you stick to named streams with OCC (which is a default for Aggregate Root) you'll be fine. OCC will act as a linearizer in scope of a named stream, no event with an earlier position would appear. Positions in named streams coming from OCC are not global positions, thus they're safe.

[brentsnook]

The idea would be to use named streams and optimistic locking, I should add some detail about ordering and make that clear in the README though. Thanks for the article, it was an interesting read!