We’ve built GitHub - CargoSense/fable: Your events have a story to tell. · GitHub for that kind of thing at Cargoesnse, although sadly there’s basically 0 documentation right now.

The core idea is that in your application you often have “root” database tables and sort of auxilliary tables. As a simple example, we have a trips table, and then there are things like alarms, trip grants, trip statistics, and so forth. We wanted to emit events on a given trip, and ensure that all updates to the trips table itself and the associated alarms, grant, and statistics all came from those events.

Here is a basic example out of our actual code:

The basic concept is you have a simple mapping of event struct names to handler functions:

defmodule Maven.Events do
  use Fable.Events,
    repo: Maven.Repo

  alias Maven.{Accounts, Travel}

  def handlers() do
    %{
      Travel.TripStarted => &Travel.trip_started/2,
      Travel.TripEnded => &Travel.trip_ended/2,
      Travel.TripGrantIssued => &Travel.trip_grant_issued/2,
      Travel.TripGrantRevoked => &Travel.trip_grant_revoked/2,
       ...
    }
  end
end

And then your context functions emit an event:

def start_trip(current_user, animal, tracker, %{id: id} = attrs) do
    Repo.serial(%Trip{id: id}, fn trip ->
      with :ok <- current_user |> can(:start_trip, %{animal: animal}),
           :ok <- animal_available(animal, current_user),
           :ok <- tracker_registered(tracker),
           :ok <- tracker_available(tracker) do
        event = %TripStarted{
          trip_id: id,
          animal_id: animal.id,
          started_by_id: current_user.id,
          tracker_id: tracker.id,
          started_at: Map.get(attrs, :started_at, DateTime.utc_now())
        }

        Events.emit(trip, event)
      else
        error -> error
      end
    end)
  end

The %Trip{} struct is a 100% ordinary Ecto schema, it just has the addition of a field(:last_event_id, :integer, read_after_writes: true). This is used by Fable to guarantee that events are processed serially, and no events are skipped. The Trip{} row acts as basically the aggregate state in ES terms.

The %TripStarted{} event is just an ecto embedded schema, which gets written to the database when we call emit. Then Fable runs the specified handler functions, passing each the trip aggregate and the emitted event, and it’s this handler function’s job to go update the trip database row and any other associated rows that should react to this event:

def trip_started(trip, event) do
    attrs =
      event
      |> Map.from_struct()
      |> Map.put(:id, trip.id)

    trip_changes = Trip.changeset(trip, attrs)

    with {:ok, trip} <- Repo.insert(trip_changes) do
      create_initial_grants(trip, event.started_by_id)
      maybe_create_test_data(trip)
    end
  end

All of the above is done in a single database transaction.

Importantly, Fable does NOT require CQRS. You’re totally allowed to just query the trips table for read state too. It is absolutely event sourcing though, since all writes to the trips table and its associated tables start with an event and it guarantees that all events are processed in order for a single aggregate. And since it’s all just Ecto and postgres, so Ecto async tests work out of the box.

If you wanted to do it CQRS style though you totally could by making your event handler functions only update the aggregate row state, and then driving all your read model changes off of a process manager. Each Fable process manager is a GenServer backed by a database row that tracks progress through the event log, and thus you could use that to manage a separate set of read tables.

If there’s sufficient interest I’ll try to get some docs up.

Our team at Inflowmatix built our ES/CQRS into our platform using a combination of meta-programming, protocols and functional Elixir. We originally ran our platforms domain store (behind a protocol to allow switching) on Mnesia for around 18 months in Production. We used term_to_binary along with compression and made heavy use of snapshots of our Aggregate state to reduce Aggregate startup costs/times. Eventually though, as our most active Aggregates started to have a long history and frequent snapshots, the table fragmentation in Mnesia was getting heavy. Additionally the locking model at a table level for certain transactions caused frequent rollbacks and retries with lots of conflict resolutions, which eventually converge, but with heavy load can cause upstream timeouts on GenServers especially where a call was needed to assure the transaction completed.

Eventually after many small repairs and refactors we switched our store to Postgresql and we were able to use our protocol to write a splitter module that wrote to both stores for a while (and read from Postgresql) until we migrated the historic data in the background to Postgresql then switched the splitters behaviour to use only Postgresql.

Just some notes really to perhaps bear in mind.

I’m still chipping away at this whenever I get a moment, and thought that I’d provide a quick update on where I’m at.

I tried using many different options for storing events, all using a shared interface so that I could quickly switch between them. I’ve found some wonderful little features of both Elixir and Erlang and some cool libraries to help me along the way — it really has been a brilliant learning experience!

So far I’ve tried:

• Simple term_to_binary and File.open(x [:binary, :append])
• DETS
• Erlang’s disk_log
• Erlang’s file:consult for reading events back and a custom io_lib:format function to write them
• CubDB (using min_key and max_key for event number ranges was brilliant)

All had positives and negatives with disk_log probably coming out on top. It has a load of features that you really need for working with files already baked in and thoroughly tested. The deal breaker though was that it’s really hard to get events back out by their event number, as disk_log really doesn’t work with any keys, it literally just appends a given term to the end of the log.

Getting the events out required bringing all events into memory (ignoring disk_logs wonderful chunking feature) and filtering out events older than those we were interested in. This isn’t the worst thing in the world as I was planning on keeping logs partitioned by aggregate, meaning that they are unlikely to ever become so long that parsing the whole log becomes an issue as it’s incredibly fast - loading 100_000 average-sized events in approximately 75ms.

My concern was that I could end up with a compromised method for reading and writing the literal heart of this system.

—

The next phase in this experiment was yet more learning… Martin Kleppmann’s “Designing Data Intensive Applications”. I can honestly say that I learnt more reading this in a week than I did over the previous 9 months piecing together articles and documentation online. His closing conclusion of “turning the database inside out” is very in line with what I’m trying to accomplish here, although I definitely think he was talking at a much larger scale than I’m working at!

Importantly, the book highlighted many issues that I was unwittingly making with other parts of my system. I was neglecting a total order by writing all of my events in partitions based on aggregates. It wasn’t a deal-breaker though as I could accept partial order as a trade-off in this instance as causality isn’t an issue in my domain. However, these partitions did make my projections quite complex due to needing to track their event offsets (using consumer offsets over ack) for many, many, many streams. Not an issue but one that I decided that for now at least to avoid by opting for a single unified log with total order.

I’m using a very simple writer and index pattern from the book and at the moment it’s working very well. Sending a command to an aggregate, having it validated, events created, persisted and applied to the aggregate is taking on average around 150 micro-seconds. I still have to handle log rotation and snapshots, but these should be quite simple as I’m modelling my store as a series of GenStage (yet another thing I’ve learnt since starting this).

—

I’ve decided in many cases to opt for purposely naive approaches - whereas before they were just naive - in order to keep my system simple. My tests have shown that I’ve got a good amount of performance margin with which to make compromises such as the single log which is a potential bottleneck, but hugely simplifies the entire system. Equally, a single event stream, local-only Registry makes it a lot easier to work with and understand. I figure that it’s easier to make a system that already works well faster, than it is fix a fast system thats yet to work.

I have had crisis’ of confidence along the way — “Why aren’t you just using Commanded and PostgreSQL like a sensible person?” - “You’re totally out of your depth here!” - “Why even bother with event sourcing, CRUD could work here after all” — but overall I’m very happy with the progress I’ve made and the things I’ve learnt. I think I’m probably a little way off being able to build anything complex with it yet, but I’m enjoying the process.