-spec by_event_pattern(term()) ->
                          khepri_evf:tree_event_filter() |
                          {error,
                           empty_selector | {not_a_map, term()} | {unknown_event_fields, [term()]}}.

Create a filter for an event_pattern subscription.

The selector is a map whose keys are #event{} field names (event_type, stream_id, version, data, metadata, tags, ...). An event matches when every field given matches: a map-valued field (data, metadata) matches when every key given has that value, any other field when it is equal. See matches/3.

Khepri's data-match condition cannot express that over the stored #event{} record (a map pattern never matches a tuple), so the trigger fires for every new event and the subscription's trigger proc applies matches/3 before delivering, the same predicate catch-up uses.

Refused: a selector that is not a map, an empty map, and a map with a key that is not an #event{} field.

-spec by_event_payload(term()) ->
                          khepri_evf:tree_event_filter() | {error, empty_selector | {not_a_map, term()}}.

Create a filter for an event_payload subscription.

The selector is a map over the event's data: an event matches when every key given has that value in its data. Like by_event_pattern/1, the trigger fires for every new event and matches/3 decides. Refused: a selector that is not a map, and an empty map.

-spec by_event_type(term()) ->
                       khepri_evf:tree_event_filter() | {error, empty_selector | {not_a_binary, term()}}.

Create a filter for events of a specific type

Uses #event{} record pattern matching since events are stored as records. Returns {error, empty_selector} for an empty binary and {error, {not_a_binary, Selector}} for anything else that is not one.

-spec by_stream(term()) ->
                   khepri_evf:tree_event_filter() |
                   {error, empty_selector | reckon_db_stream_path:validation_error()}.

Build a Khepri tree-event filter for subscriptions of type stream (a.k.a. by_stream).

Selector grammar

• <<"$all">> — the only recognised wildcard sentinel. Matches every event in every stream in the store.
• Any other non-empty binary — a literal stream id; the filter matches events on that exact stream only.

Stream id conventions

The filter doesn't enforce stream-id shape (it just treats the selector as a path component), but reckon-db has two formats:

• **User streams** — <prefix>-<hex>, where prefix is [A-Za-z]+ and hex is [A-Fa-f0-9]+ (typically a UUIDv7 with dashes stripped). Example: account-018f6a7b8c9d4abc8901234567890abc.
• **System streams** — $<namespace>:<name>, where the $ prefix marks "reckon-db-managed, not user data" and <name> is intentionally human-readable. Example: $link:high-value-orders from reckon_db_links. Same role as EventStoreDB's $ce- / $et- / $by_* prefixes.

Both formats subscribe identically — the filter doesn't care. The $ namespace is only special at the sentinel level ($all).

Returns {error, empty_selector} for an empty binary, and reckon_db_stream_path:validate/1's error for an id the grammar rejects; every other input builds a filter.

The malformed case used to raise from the path builder, and it raised after the subscription record had already been written: the caller's cleanup branch never ran, so a poisoned subscription stayed in the store and raised again on the next restore or trigger re-register (reckon-db#4). Returning the error lets that branch delete it.

-spec by_tags(term()) ->
                 khepri_evf:tree_event_filter() | {error, empty_selector | {not_a_tag_list, term()}}.

Create a filter for a by_tags subscription.

An event matches when it carries every tag given. Khepri's pattern matching has no list membership, so the trigger fires for every new event and matches/3 decides, live as on catch-up. For queries rather than subscriptions, use reckon_db_streams:read_by_tags/4. Refused: anything but a non-empty list of binaries.

-spec event_type_pattern(binary()) -> tuple().

A Khepri data-match pattern for a stored #event{} of this type: the record's tag, the type, and `_'' in every other field.

It is a match pattern, not an event, so it is built as a tuple: #event{event_type = T, _ =_'}' says the same thing in record syntax and violates every other field's declared type.

-spec matches(subscription_type(), binary() | map() | [binary()], event()) -> boolean().

In-memory predicate matching an #event{} record against a subscription's (Type, Selector) pair.

The Khepri event filters above are evaluated by Khepri itself on live triggers, but the catch-up replay path bypasses Khepri's trigger engine — it reads the global stream and delivers events directly to the subscriber. Without this predicate, every subscription with start_from = 0 would receive the entire log regardless of its declared selector. Use this to filter the catch-up batch before sending.

Tag membership uses set inclusion: the event must carry ALL of the requested tags (the all match of read_by_tags/4). event_pattern and event_payload are described at by_event_pattern/1 and by_event_payload/1.

The live path applies this too: the trigger proc of every subscription checks it before broadcasting, so the two paths agree by construction (reckon-db#8, #11).