Orkestra.EventStore.InMemory (orkestra v0.2.0)

In-memory EventStore adapter backed by an Agent. For tests and local development without EventStoreDB.

Agent State

The Agent holds a map with the following keys:

:streams — map of stream_id => [stored_event()] for per-stream events
:global_counter — non-negative integer; incremented on each appended event to assign gap-free monotonic :global_position values (D-01)
:subscribers — list of {ref, pid, stream_or_all} tuples registered via subscribe_from_position/3. The stream_or_all is the subscribed stream id (or :all) and is used to filter live delivery; ref is the handle returned to the caller and accepted by unsubscribe/1.
:global_events — list of all events in global-append order, each extended with :global_position

Subscriber Delivery (D-03)

When a subscriber registers via subscribe_from_position/3, the adapter:

Atomically registers the subscriber pid and snapshots global_events inside a single Agent.get_and_update, preventing races with concurrent appends (RESEARCH.md Pitfall 3).
Replays snapshotted events with global_position > from_position (exclusive, matching Spear's from: semantics — Pitfall 1).
On each subsequent append_events/3, pushes new events to each registered subscriber in order, filtered by the subscriber's subscribed stream (:all subscribers receive every event; a per-stream subscriber receives only events stamped with its stream_id).

This push model mirrors EventStoreDB's subscription model so the Phase 2 Projector GenServer can code against a single delivery interface.

Note: the Agent is a named singleton. InMemory subscription tests must use async: false and start the adapter per-test via start_supervised/1.

Summary

Functions

child_spec(arg)

Returns a specification to start this module under a supervisor.

reset!(name \\ __MODULE__)

Resets all stored events and subscriber state. Useful in test setup.

start_link(opts \\ [])

Starts the InMemory adapter. Accepts an optional :name in opts.

subscribe_from_position(stream_id_or_all, from_position, subscriber)

Subscribes subscriber to receive events from stream_id_or_all starting after from_position (exclusive).

unsubscribe(ref)

Removes the subscription identified by ref so the subscriber stops receiving live events.

Functions

child_spec(arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

reset!(name \\ MODULE)

@spec reset!(atom()) :: :ok

Resets all stored events and subscriber state. Useful in test setup.

start_link(opts \\ [])

@spec start_link(keyword()) :: Agent.on_start()

Starts the InMemory adapter. Accepts an optional :name in opts.

subscribe_from_position(stream_id_or_all, from_position, subscriber)

@spec subscribe_from_position(
  Orkestra.EventStore.stream_id() | :all,
  integer(),
  pid()
) ::
  {:ok, reference()} | {:error, term()}

Subscribes subscriber to receive events from stream_id_or_all starting after from_position (exclusive).

Atomically registers the subscriber and snapshots existing events inside a single Agent.get_and_update, then replays the snapshot to the subscriber (filtered to stream_id_or_all). Subsequent calls to append_events/3 push new events to all registered subscribers in order, filtered by each subscriber's subscribed stream.

Returns {:ok, subscription_ref}. The returned ref is a real handle: pass it to unsubscribe/1 to stop delivery and remove the subscriber from state.

unsubscribe(ref)

@spec unsubscribe(reference()) :: :ok

Removes the subscription identified by ref so the subscriber stops receiving live events.

Returns :ok whether or not a matching subscription existed (idempotent).