-spec ack_event(atom(), binary(), pid(), map()) -> ok.

Acknowledge receipt of an event by a subscriber

-spec append_events(atom(), binary(), list()) -> {ok, integer()} | {error, term()}.

Append events to a stream (auto-versioned)

-spec append_events(atom(), binary(), integer() | any, list()) ->
                       {ok, integer()} | {error, term()} | {error, {wrong_expected_version, integer()}}.

Append events to a stream with expected version

-spec build_causation_graph(atom(), binary()) -> {ok, map()} | {error, term()}.

Build a causation graph for visualization

-spec check_raft_log_consistency(atom()) -> {ok, map()} | {error, term()}.

Check Raft log consistency for a store

-spec create_link(atom(), map()) -> ok.

Create a new link

-spec delete_link(atom(), binary()) -> ok.

Delete a link

-spec delete_snapshot(atom(), binary(), binary(), non_neg_integer()) -> ok.

Delete a snapshot

-spec delete_stream(atom(), binary()) -> ok | {error, term()}.

Delete a stream and all its events

-spec event_type_summary(atom()) -> {ok, [map()]} | {error, term()}.

Census of event types in the store with counts.

-spec get_causation_chain(atom(), binary()) -> {ok, list()} | {error, term()}.

Get the full causation chain for an event

-spec get_cause(atom(), binary()) -> {ok, map()} | {error, term()}.

Get the event that caused another

-spec get_correlated(atom(), binary()) -> {ok, list()} | {error, term()}.

Get all events with the same correlation ID

-spec get_effects(atom(), binary()) -> {ok, list()} | {error, term()}.

Get events caused by an event

-spec get_events(atom(), binary(), integer(), integer(), forward | backward) ->
                    {ok, list()} | {error, term()}.

Get events from a stream

-spec get_link(atom(), binary()) -> {ok, map()} | {error, term()}.

Get a link by name

-spec get_memory_level(atom()) -> {ok, atom()} | {error, term()}.

Get current memory pressure level

-spec get_memory_stats(atom()) -> {ok, map()} | {error, term()}.

Get memory statistics

-spec get_schema(atom(), binary()) -> {ok, map()} | {error, term()}.

Get a schema by event type

-spec get_schema_version(atom(), binary()) -> {ok, integer()} | {error, term()}.

Get the version of a schema

-spec get_streams(atom()) -> {ok, list()} | {error, term()}.

Get all streams in a store

-spec get_subscription(atom(), binary()) -> {ok, map()} | {error, term()}.

Get a specific subscription by name

Returns the subscription details including the checkpoint.

-spec get_subscriptions(atom()) -> {ok, list()} | {error, term()}.

Get all subscriptions for a store

-spec get_version(atom(), binary()) -> {ok, integer()} | {error, term()}.

Get the current version of a stream

-spec get_workers(atom()) -> {ok, [worker_entry()]} | {error, term()}.

Get all registered workers for a store

-spec has_events(atom()) -> boolean().

Check if a store contains at least one event.

-spec health() -> {ok, map()}.

Get gateway health status

-spec link_info(atom(), binary()) -> {ok, map()} | {error, term()}.

Get detailed info about a link

-spec list_all_snapshots(atom()) -> {ok, [map()]} | {error, term()}.

List all snapshots across all streams in a store.

-spec list_links(atom()) -> {ok, list()} | {error, term()}.

List all links

-spec list_schemas(atom()) -> {ok, list()} | {error, term()}.

List all schemas

-spec list_snapshots(atom(), binary() | any, binary() | any) -> {ok, [map()]} | {error, term()}.

List snapshots

-spec list_store_subscriptions(atom()) -> {ok, [map()]} | {error, term()}.

List all subscriptions for a store with checkpoint positions.

-spec list_stores() -> {ok, list()} | {error, term()}.

List all managed stores in the cluster

-spec pick_worker([worker_entry()], node(), non_neg_integer()) ->
                     {ok, worker_entry(), non_neg_integer()} | {error, no_workers}.

Pure worker-selection policy behind select_worker/1.

Prefers workers on the caller's own node when any exist; falls back to cluster-wide round-robin only when no local worker is present. Extracted so the policy can be exercised without a running pg scope.

Why local-first

The worker registry is pg-based, so a call to get_workers/1 returns PIDs from every BEAM-connected node. That's the right pool for stores that form a shared Raft cluster — any worker writes into the same Khepri state machine and reads are cluster-wide.

For stores that stay local per node (autojoin=false), a write routed to a remote worker persists in THAT node's private store and is invisible to the caller's local store. Preferring the caller's own node keeps each daemon's writes on its own disk. Falling back to the remote pool preserves the cluster-wide behaviour when the local worker isn't registered yet (boot race) or when the store is genuinely clustered and the caller has no local presence.

-spec quick_health_check(atom()) -> {ok, map()} | {error, term()}.

Quick health check for a store

-spec read_all_global(atom(), non_neg_integer(), pos_integer()) -> {ok, list()} | {error, term()}.

Read all events across all streams sorted by epoch_us (global ordering).

This is used for catch-up subscriptions where a consumer needs to replay all historical events from a given offset.

Offset is the number of events to skip (not a version number). BatchSize controls how many events to return per call.

-spec read_by_event_types(atom(), [binary()], pos_integer()) -> {ok, list()} | {error, term()}.

Read events by type using native Khepri filtering

This uses the server-side read_by_event_types which performs efficient filtering at the database level rather than loading all events.

-spec read_by_tags(atom(), [binary()]) -> {ok, list()} | {error, term()}.

Read events by tags (default: ANY match, batch_size 1000).

Queries events across all streams that have matching tags. By default, returns events matching ANY of the provided tags (union).

-spec read_by_tags(atom(), [binary()], map()) -> {ok, list()} | {error, term()}.

Read events by tags with options.

Queries events across all streams that have matching tags. Tags are typically used for cross-stream querying in the process-centric model.

Options: - match: any (default) returns events matching ANY tag (union), all returns events matching ALL tags (intersection) - batch_size: Maximum events to return (default 1000)

-spec read_range(atom(), binary(), integer(), integer()) -> {ok, list()} | {error, term()}.

Read events in a time range

-spec read_range(atom(), binary(), integer(), integer(), map()) -> {ok, list()} | {error, term()}.

Read events in a time range with options

-spec read_snapshot(atom(), binary(), binary(), non_neg_integer()) -> {ok, map()} | {error, term()}.

Read a snapshot

-spec read_until(atom(), binary(), integer()) -> {ok, list()} | {error, term()}.

Read events up to a timestamp

-spec read_until(atom(), binary(), integer(), map()) -> {ok, list()} | {error, term()}.

Read events up to a timestamp with options

-spec record_snapshot(atom(), binary(), binary(), non_neg_integer(), map()) -> ok.

Record a snapshot

-spec register_schema(atom(), binary(), map()) -> ok.

Register a schema

-spec register_worker(atom()) -> ok | {error, term()}.

Register current process as a worker for a store

-spec register_worker(atom(), pid()) -> ok | {error, term()}.

Register a specific process as a worker for a store

-spec remove_subscription(atom(), atom(), binary() | map(), binary()) -> ok.

Remove a subscription

-spec save_subscription(atom(),
                        atom(),
                        binary() | map(),
                        binary(),
                        non_neg_integer(),
                        pid() | undefined) ->
                           ok.

Save a subscription

-spec scavenge(atom(), binary(), map()) -> {ok, map()} | {error, term()}.

Scavenge a stream (delete old events)

-spec scavenge_dry_run(atom(), binary(), map()) -> {ok, map()} | {error, term()}.

Dry-run scavenge (preview what would be deleted)

-spec scavenge_matching(atom(), binary(), map()) -> {ok, list()} | {error, term()}.

Scavenge streams matching a pattern

-spec start_link(atom(), binary()) -> ok.

Start a link

-spec stop_link(atom(), binary()) -> ok.

Stop a link

-spec store_stats(atom()) -> {ok, map()} | {error, term()}.

Aggregate statistics for a store.

-spec stream_backward(atom(), binary(), integer(), non_neg_integer()) -> {ok, list()} | {error, term()}.

Stream events backward from a version

-spec stream_forward(atom(), binary(), integer(), integer()) -> {ok, list()} | {error, term()}.

Stream events forward from a version

-spec stream_info(atom(), binary()) -> {ok, map()} | {error, term()}.

Detailed info for a single stream (timestamps, snapshot coverage).

-spec subscription_lag(atom(), binary()) -> {ok, map()} | {error, term()}.

Calculate lag for a specific subscription.

-spec unregister_schema(atom(), binary()) -> ok.

Unregister a schema

-spec unregister_worker(atom()) -> ok | {error, term()}.

Unregister current process as a worker for a store

-spec unregister_worker(atom(), pid()) -> ok | {error, term()}.

Unregister a specific process as a worker for a store

-spec upcast_events(atom(), list()) -> {ok, list()} | {error, term()}.

Upcast events to current schema version

-spec verify_cluster_consistency(atom()) -> {ok, map()} | {error, term()}.

Verify cluster consistency for a store

-spec verify_membership_consensus(atom()) -> {ok, map()} | {error, term()}.

Verify membership consensus for a store

-spec version_at(atom(), binary(), integer()) -> {ok, integer()} | {error, term()}.

Get stream version at a specific timestamp