Orkestra.EventStore.EventStoreDB.SubscriptionRelay (orkestra v0.2.3)

Copy Markdown View Source

A per-subscription relay process that adapts Spear's raw subscription delivery to the Orkestra.EventStore subscriber contract.

Spear.subscribe/4 pushes Spear.Event.t/0 structs (plus Spear.Filter.Checkpoint.t/0, {:caught_up, ref}, {:fell_behind, ref} and {:eos, ref, reason} control messages) directly to the subscribing process. The Orkestra projector, however, is written against a single delivery contract shared with Orkestra.EventStore.InMemory: it receives only Orkestra.EventStore.stored_event_with_position/0 maps (maps carrying a :global_position key) and has no handle_info/2 clause for Spear's control structs — an unmatched %Spear.Filter.Checkpoint{} would crash it.

This relay closes that gap. It becomes the Spear subscriber, and for each message it receives it:

  • transforms a %Spear.Event{} into a stored_event_with_position() map (via Orkestra.EventStore.EventStoreDB.to_stored_event/1, the same mapping used by load_events/1,2) and forwards it to the real subscriber — so the projector sees exactly the shape InMemory delivers;
  • silently drops every Spear control message (checkpoints, caught_up, fell_behind) — they carry no domain event;
  • stops (cancelling the underlying Spear subscription) when it receives an {:eos, _, reason} end-of-stream, or when the real subscriber dies (it Process.monitor/1s the subscriber), so no subscription is leaked.

For an :all subscription the relay passes resolve_links?: false and filter: Spear.Filter.exclude_system_events() to Spear.subscribe/4. With EventStoreDB standard projections enabled, an event is also surfaced through system link streams ($ce-*, $et-*) that share the original event's commit_position; resolve_links?: true (Spear's default) would therefore deliver the same event several times, inflating checkpoints and causing re-processing. Disabling link resolution and excluding $-prefixed system streams delivers each event exactly once.

Note that orkestra's own snapshot-<stream> streams do not start with $, so their events still reach an :all subscriber. That is intentional: the projector's handler is expected to ignore event types it does not recognise (advancing the checkpoint without dead-lettering them).

Summary

Functions

start(connection, stream_id_or_all, from, subscriber)

Starts a relay linking a Spear subscription on connection to subscriber.

stop(pid)

Cancels the relay's Spear subscription and stops the relay process.

Functions

start(connection, stream_id_or_all, from, subscriber)

@spec start(
  Spear.Connection.t(),
  Orkestra.EventStore.stream_id() | :all,
  term(),
  pid()
) ::
  {:ok, pid()} | {:error, term()}

Starts a relay linking a Spear subscription on connection to subscriber.

Returns {:ok, relay_pid} once the Spear subscription is confirmed, or {:error, reason} if Spear.subscribe/4 fails. The returned pid is the opaque subscription handle: pass it to Orkestra.EventStore.EventStoreDB.unsubscribe/1 to tear the subscription down (this is also done automatically when subscriber dies).

stop(pid)

@spec stop(pid()) :: :ok

Cancels the relay's Spear subscription and stops the relay process.

Idempotent and safe to call with a dead pid.