reckon_db_subscriptions (reckon_db v5.11.14)
View SourceSubscriptions API facade for reckon-db
Provides the public API for subscription operations: - subscribe: Create a new subscription - unsubscribe: Remove a subscription - get: Get a subscription by key - list: List all subscriptions - exists: Check if a subscription exists
Subscription types: - stream: Subscribe to all events in a specific stream - event_type: Subscribe to events of a specific type - event_pattern: Subscribe to events matching a pattern - event_payload: Subscribe to events with specific payload patterns
Summary
Functions
Acknowledge event delivery for a subscription
Check if a subscription exists
Get a subscription by key
List all subscriptions in the store
Re-establish subscriptions captured by snapshot_live/1 after the local tree was replaced by the cluster's.
Setup subscription tracking for a process
Subscriptions on this store whose subscriber is still alive.
Create a subscription with default options
Create a subscription with options
Names this VM's boot: 16 random bytes, drawn once per VM and kept in persistent_term. Nothing the VM can read about itself does this job: in a container the BEAM is OS pid 1 on every boot, and erlang:system_info(start_time) is the fixed value monotonic time starts at, so that pair named every boot of a container the same (reckon-db#12).
Remove a subscription by key
Remove a subscription by type, selector, and name
Types
-type event() :: #event{event_id :: binary(), event_type :: binary(), stream_id :: binary(), version :: non_neg_integer(), data :: map() | binary(), metadata :: map(), tags :: [binary()] | undefined, timestamp :: integer(), epoch_us :: integer(), data_content_type :: binary(), metadata_content_type :: binary(), prev_event_hash :: binary() | undefined, mac :: mac_value(), signature :: binary() | undefined}.
-type mac_value() :: {KeyId :: non_neg_integer(), MacBytes :: binary()} | {v2, KeyId :: non_neg_integer(), MacBytes :: binary()} | undefined.
-type subscribe_opts() :: #{subscription_name => binary(), pool_size => pos_integer(), start_from => non_neg_integer(), subscriber => pid()}.
-type subscription() :: #subscription{id :: binary(), type :: subscription_type(), selector :: binary() | map(), subscription_name :: binary(), subscriber_pid :: pid() | undefined, created_at :: integer(), pool_size :: pos_integer(), checkpoint :: non_neg_integer() | undefined, options :: map()}.
-type subscription_type() ::
stream | event_type | event_pattern | event_payload | by_stream | by_event_type |
by_event_pattern | by_event_payload | by_tags.
Functions
-spec ack(atom(), binary(), binary() | undefined, non_neg_integer()) -> ok | {error, term()}.
Acknowledge event delivery for a subscription
Updates the checkpoint for the subscription to track progress. This is typically called after successfully processing an event. The checkpoint allows subscriptions to resume from where they left off after a restart.
Parameters: StoreId - The store identifier SubscriptionName - Name of the subscription StreamId - ID of the stream the event came from (may be undefined for cross-stream) EventNumber - Version/position of the acknowledged event
Returns ok on success, or {error, Reason} if the subscription is not found.
Check if a subscription exists
-spec get(atom(), binary()) -> {ok, subscription()} | {error, not_found}.
Get a subscription by key
-spec list(atom()) -> {ok, [subscription()]} | {error, term()}.
List all subscriptions in the store
-spec restore(atom(), [subscription()]) -> ok.
Re-establish subscriptions captured by snapshot_live/1 after the local tree was replaced by the cluster's.
Each goes through subscribe/5. Absent from the cluster tree (the reset took it), it is created there: record, trigger, emitter pool, tracker notification, and its subscriber is caught up from position 0 of the cluster's history, which it has never seen (its own local history did not survive the join, so its previous checkpoint means nothing against the cluster's log). Already present with the same live subscriber (it was replicated before the reset), nothing needs doing.
Setup subscription tracking for a process
Joins the tracker group for subscriptions, allowing the process to receive notifications about subscription lifecycle events (created, updated, deleted).
The process will receive messages in the format: - {feature_created, subscriptions, Data} - {feature_updated, subscriptions, Data} - {feature_deleted, subscriptions, Data}
-spec snapshot_live(atom()) -> [subscription()].
Subscriptions on this store whose subscriber is still alive.
Taken right BEFORE an operation that resets the local Khepri tree (joining an existing cluster, the healer's reset-and-rejoin), so restore/2 can re-establish them in the cluster's tree afterwards. khepri_cluster:join/2 replaces the joiner's whole tree; a consumer that subscribed while its node was still a standalone cluster of one (the normal boot shape: it subscribes as soon as the store is up, discovery joins seconds later) would otherwise keep a live pid and a running emitter pool while its record and trigger are silently gone.
Bounded query (see SNAPSHOT_TIMEOUT); on failure nothing is captured rather than hanging the reset.
-spec subscribe(atom(), subscription_type(), binary() | map(), binary()) -> {ok, binary()} | {error, term()}.
Create a subscription with default options
Parameters: StoreId - The store identifier Type - Subscription type (stream, event_type, event_pattern, event_payload) Selector - The selector for matching events SubscriptionName - Human-readable name for the subscription
Returns {ok, SubscriptionKey} on success or {error, Reason} on failure.
-spec subscribe(atom(), subscription_type() | term(), term(), binary(), subscribe_opts()) -> {ok, binary()} | {error, term()}.
Create a subscription with options
Options: - pool_size: Number of emitter workers (default: 1) - start_from: Starting position for replay (default: 0) - subscriber: PID to receive events directly (default: undefined)
Type and Selector reach this from gateway clients as any term. A type that is not a subscription_type(), or a selector the type cannot use, is answered {error, {invalid_filter, Reason}} and nothing is stored.
-spec this_boot() -> binary().
Names this VM's boot: 16 random bytes, drawn once per VM and kept in persistent_term. Nothing the VM can read about itself does this job: in a container the BEAM is OS pid 1 on every boot, and erlang:system_info(start_time) is the fixed value monotonic time starts at, so that pair named every boot of a container the same (reckon-db#12).
reckon_db_app draws it at application start, before any subscriber can race for the first draw; a VM that subscribes without the application (a bare khepri store) draws it on first use.
Remove a subscription by key
-spec unsubscribe(atom(), subscription_type(), binary()) -> ok | {error, term()}.
Remove a subscription by type, selector, and name