Handle missing events cases through new methodwith_aggregate!

[laertispappas]

Hello,

Thank you for your work on rails event store. I have a quick question regarding the #with_aggregate method on https://github.com/RailsEventStore/rails_event_store/blob/master/aggregate_root/lib/aggregate_root/repository.rb#L22

It's not clear on the callers how they should require the aggregate to already exist on the stream. Something similar to .find method in active record. Would it make sense to extend the repo above and include equivelant ! methods to take into account that there should be at least one event on the stream otherwise error out. It is not clear in general from the docs how to handle missing events.

Thank you in advance.

[mostlyobvious]

Just to make sure I understand:

• you'd like to be able to initialize an aggregate instance, make it produce an event as a result of some action and persist it via aggregate root repository as usual
• load an aggregate instance from persisted state only if there is any state, otherwise error out when using with_aggregate! variant of AggregateRoot::Repository#with_aggregate

Is this correct?

I haven't yet personally found the need to implement such behaviour. This influence comes from the perspective of CQRS and using:

• aggregates for the write model (protecting business invariants and producing events as a result of an action)
• read models for the read model (populated from events, usually projected into database tables)

I think it could be useful when you're not segregating reads from writes and need such aggregate instance with internal state exposed for display.

Can you please tell me more about your use case or perhaps show some code samples?

[laertispappas]

@pawelpacana thank you for your input. I understand the separation between the read and write models, what is not clear to me is how to handle a missing aggregate without the need to check internal details like version. A small context on my side I use the library to create a central CMS and use projections to project the content from the central event store to "different environments".

Consider the following snippet

repository.with_aggregate!(Catalog::SomeEntity, command.uid) do |entity|
  # I should not call the external service with an a `new aggregate here` i need to make sure the aggregate is persisted. 
  # this is why I need the `load!` method
  service.sync!(entity)
  entity.sync_success!(...)
  Result::Success.new(entity)
end

I need to make sure that with_aggregate! loads an existing entity and that exists on the store so I implemented this:

  def with_aggregate!(aggregate_class, aggregate_id)
    load!(aggregate_class, aggregate_id).then do |aggregate|
      yield(aggregate).tap do
        repository.store(aggregate, stream_name(aggregate_class, aggregate_id))
      end
    end
  end

  def load!(aggregate_class, aggregate_id)
    load(aggregate_class, aggregate_id).tap do |aggregate|
      raise AggregateNotFoundError unless aggregate.persisted?
    end
  end

  def load(aggregate_class, aggregate_id)
    repository.load(
      aggregate_class.new(uid: aggregate_id),
      stream_name(aggregate_class, aggregate_id)
    )
  end


class EventSourcedEntity
  def persisted?
    version > -1
  end
end

My main concern is the implementation of def persisted? where I need to check the version to decide if the aggregate exists or not. I was curious if you guys have something similar encounter and how you handle that?

Thank you in advance.

[mostlyobvious]

There's no other way to check if stream exists other than checking if it's empty. And I don't see other option to verify if an aggregate instance "exists" other than checking its stream.

This could be the implementation that short-circuits already in the repository:

  class MyRepository < AggregateRoot::Repository
    AggregateNotFoundError = Class.new(StandardError)
    
    def with_aggregate!(aggregate, stream_name, &block)
      raise AggregateNotFoundError if stream_empty?(stream_name)
      with_aggregate(aggregate, stream_name, &block)
    end

    private
    
    def stream_empty?(stream_name)
      event_store.read.stream(stream_name).count.zero?
    end
  end

at a cost of additional query:

SELECT COUNT(*) FROM "event_store_events_in_streams" INNER JOIN "event_store_events" ON "event_store_events"."event_id" = "event_store_events_in_streams"."event_id" WHERE "event_store_events_in_streams"."stream" = $1  [["stream", "..."]]

The other option would be to introduce load! which acts on array of just read events:

rails_event_store/aggregate_root/lib/aggregate_root/repository.rb

Line 10 in e424c3e

event_store.read.stream(stream_name).reduce { |_, ev| aggregate.apply(ev) }

.

[laertispappas]

@pawelpacana Alright buddy thank you for your feedback, much appreciated.

[laertispappas]

@pawelpacana Rethinking this. This won't work for deleted entities (soft deletes) So I ended up doing the following:

  def find!(cmd)
    cmd.aggregate_class.new(id: cmd.id).tap do |aggregate|
      repository.load(aggregate, aggregate.stream_name)
      raise AggregateNotFoundError if stream_empty?(aggregate.stream_name) || aggregate.deleted?
    end
  end

deleted method on the aggregate is a helper predicate to check if something is deleted:

In general, I think we still need a convenient supported method to know if an aggregate is persisted or not and not rely on the callers to implement it.