Proper way of storing data with events

[mbajur]

Hey there! First of all, thank you for this great piece of software.

I'm not sure if that's a proper way to ask general best-practice questions but let's give it a shot. If not, please forgive me.

I'm building a system which receives a pretty complicated JSON object and needs to validate it async and, if it's valid, it needs to make a further work. Obviously that's a really simplified description of what's really going on.

As those incoming JSON objects are kind of like commands of what to do with actual models (Create/Update/Vote/etc), i decided to go for RES and test it out. Everythng goes smoothly so far but i'm unsure about one detail. Let's say steps are as follows:

1. JSON object comes in
2. ActivityCreated event is created in RES
3. JSON validation is being done using RES subscribers (async)
4. If Activity is successfull, ActivityAccepted event is created
5. otherwise, ActivityRejected event is created

Apps using that can subscribe to ActivityAccepted event and do further work. By "further work" i mean reading data from the JSON object persisted in step 2.

Right now, i subscribe to ActivityAccepted and gets the received JSON using the following code instead of storing the same JSON with ActivityAccepted event:

class OnActivityAccepted
  def call(event)
    id = event.causation_id
    Rails.configuration.event_store.read.event(id) # to fetch the contents of parent event
  end
end

As it works pretty well, is that a proper way of dealing with data in ES world? Should i pass the original JSON to the ActivityAccepted event (and any other further event)? Are there any best practices regarding that? Right now, the only contents of ActivityAccepted event are the state changed to accepted

Thanks in advance for any tips!

[swistak35]

There are at least few approaches for that :)

1. I assume that each json comes with some ID. If not, you can create one (for example by using SecureRandom.uuid) at the moment you receive the request with the activity. Then, when publishing event, you may publish that event in the stream specifically for that activity: event_store.publish(ActivityCreated.new(...), stream_name: "Activity$#{activity_id}"). Later on, you can publish ActivityAccepted in the same stream, and then in OnActivityAccepted you can use event_store.read.stream("Activity$#{event.data.fetch(:activity_id)}") to get all the events which are in this stream -- the first one will be ActivityAccepted :)
2. If in the moment when you receive the payload, you know what kind of "command" it will be, you may publish more specific event, like VoteRequested, and then do the similar stuff as above with streams, but you end up with more specific vocabulary used in your domain events. I can't tell which approach is better without seeing the code (if you or your company is interested we can take a look at your specific case).
3. If the flow you've made with async validation is only to make your http roundtrip faster and accept requests quickly, then you may just treat that payload as a command, but the one you want to execute asynchronously. Commands and Domain Events, are semantically very different, but in the end they both have similar characteristics and we may think of them as messages. We don't have for know any more explicit support for storing commands, but really, technically a lot of the things are the same and RES is good enough for storing Commands too (1, 2) :)

If the causation_id way solves your case, then it may be also acceptable approach. But personally I don't think I've ever used causation_id for purposes other than debugging. The problem in that approach may be that if later, you'll have some other (possibly optional) steps between ActivityCreated and ActivityAccepted, then finding the right ActivityCreated may not be so easy anymore.

[mbajur]

Hey @swistak35 thanks for this extended reply! Sadly, that's an open source project and i don't have budget to hire anyone for consultancy - but who knows, maybe one day :) If you are curious - i'm working on rails engine to make it easier to integrate existing apps with ActivityPub "fediverse" (protocol for federation behind Mastodon social network - https://activitypub.rocks/). It's model fits perfectly with Event Streaming IMO so i decided to give it a try.

Anyway - option no. 1 looks pretty nice (especially that i can query for specific events in such stream, not all of them) but one thing bothered me from first minutes of playing with RES - is that possible to specify to which stream event should be assigned in the AggregateRoot class ?

module ActivityPub
  class Activity
    def initialize(id)
      @id = id
      @state = :new
      @body = :new
    end

    def create(body)
      apply Events::ActivityCreated.new(data: {
        activity_id: @id,
        body:        body,
        state:       @state
      })
    end

    on Events::ActivityCreated do |raw_activity|
      @state = :created
    end
  end
end

based on the docs, the only place i can specify a stream(s) is a #publish method but AggregateRoot seems to skip that part.

Thank you!

[swistak35]

Check out: https://railseventstore.org/docs/app/

The event sourcing is described at this page. In particular:

stream_name = "Order$123"
order = AggregateRoot::Repository.new.load(Order.new, stream_name)

(and in the same way, when storing aggregate in #store method)

I'll close the issue for now :)