Aggregate state snapshots

[commanded]

Aggregate state snapshots

[drozzy]

I've done this on my own. It is not hard, just tricky.

Yes, you need to callbacks on the aggregate:

• save_to_memento
• load_from_memento(memento)

These would in the same place as the apply method is right now.

For snapshot stream, I use AggName-snapshots-AggID. So for aggregate Foo with id 123 it would be Food-snapshots-123.

Loading snapshots

The process that does "loading" from snapshots and events I call the hydrator, and here is there relevant piece of code (I'm sure your implementation is much more elegant, but you get the gist):

    def do_hydrate_from_snapshot(%Hydrator{snapshot_stream_id: id}=state) do
       state2 = case EvStore.read_events_backward(id, -1, 1) do
         {:error, :NoStream, _} -> state
         {:ok, %{events: events}} -> 
            [snapshot] = Adaptor.from_external(events)
            hydrate_from_snapshot(snapshot, state)
       end
       # Now start hydrating from events
       send self(), :hydrate_from_events
       {:noreply, state2}
    end

    def hydrate_from_snapshot(%Snapshot{memento: m, event_number: n}, %Hydrator{agg_mod: agg_mod, agg_pid: agg_pid}=state) do
       :ok = agg_mod.load_from_memento(agg_pid, m)
       %{state | last_event: n, last_snapshot_event: n}
    end

Basically, you want to get the "last" event. If there is none, e.g. no snapshot available, just continue hydrating from events. If there is a snapshot, deserialize it and give it to aggregate.

Saving snapshots

For saving the snapshots, I set a periodic timer:

:timer.send_after(@save_period, :save)

and then check if it is time to save the snapshot (in this case @snapshot_period 1000 - so snapshot about every thousand events - this works out very well, for Gregg's ES because 5K is when it starts to get slow):

def save_snapshot(%{agg_pid: agg_pid, 
     agg_mod: agg_mod, 
     last_event: last_event, 
     snapshot_stream_id: snapshot_stream_id,
     last_snapshot_event: last_snapshot_event}=state) do
     
     case (last_event - last_snapshot_event) > @snapshot_period do
       true -> 
         {:ok, memento} = agg_mod.save_to_memento(agg_pid)
         event = %Snapshot{event_number: last_event, memento: memento}
         external = Adaptor.to_external([event])
         {:ok, _} = EvStore.append_events(snapshot_stream_id, external, -2)
         %{state | last_snapshot_event: last_event}
       false ->
         state
     end
   end

[slashdotdash]

@drozzy Thanks for the feedback.