barrel_changes (barrel_docdb v1.1.1)

View Source

Changes feed API for barrel_docdb

Provides functions to track and query document changes in a database. Changes are ordered by HLC (Hybrid Logical Clock) timestamps for distributed ordering.

Summary

Functions

Count changes since a given HLC timestamp (exclusive)

Delete an old HLC entry (when document is updated)

Encode change to compact binary format. Format: <<DocIdLen:16, DocId/binary, RevLen:16, Rev/binary, Deleted:8, NumConflicts:16, HasDoc:8, [DocCbor/binary]>> Stores conflict count for quick check; optionally includes doc body for filtering.

Fold over changes since a given HLC timestamp (exclusive) Changes returned are strictly after the given HLC. Use 'first' to get all changes from the beginning.

Fold over changes with compact tuple format for efficiency. Callback receives {DocId, Hlc, Rev, Deleted, NumConflicts} tuples. Use for internal iteration; convert to maps at API boundary. Always uses long_scan mode for optimal sequential read performance.

Get a list of changes since an HLC timestamp

Get changes for a specific path pattern since HLC. Scans the path_hlc index directly for efficient filtered queries.

Get changes with continuation support for pagination. Accepts either an HLC timestamp, 'first', or a continuation token (binary). Returns changes along with continuation info for fetching the next batch.

Get the last HLC timestamp for a database First tries the fast path (metadata key), falls back to reverse iteration

Get the last sequence (opaque encoded HLC) for a database The sequence is an opaque binary that can be used for ordering.

Fast O(1) check if there are changes since a given HLC. Uses time-bucketed hints to avoid scanning the change log. Returns true if there might be changes, false if definitely no changes. Note: May return true even if no changes (false positive OK, false negative not OK).

Exact inverse of write_path_index_ops/3, for a change row being removed from the feed (timeline rewind): deletes the path_hlc rows and posting entries that the write created, deriving topics the same way (a tombstone's rows live under its own doc id, and the posting marker carries the row's own rev and deleted flag).

Return batch operation to update change bucket. Buckets store {MinHlc, MaxHlc, Count} in compact binary format. Format: <<MinHlc:12/binary, MaxHlc:12/binary, Count:32>>

Generate operations to update path index (remove old entries + add new). Used when a document is updated to maintain a current path index without stale entries.

Write a change entry for a document. Also updates change buckets for idle poll optimization.

Return batch operation to write a change entry. Use this to combine with other operations in a single write_batch. Also updates the last_hlc metadata for efficient get_last_seq lookups. Note: Does NOT update change buckets - call write_change/4 for full update.

Generate operations to index a change by all its paths. Creates entries under path_hlc/{db}/{topic}/{hlc} for exact matches, and prefix_changes/{db}/{prefix}/{bucket} posting lists for wildcard queries.

Types

att_info/0

-type att_info() ::
          #{name := binary(),
            content_type := binary(),
            length := non_neg_integer(),
            digest := binary(),
            chunked => boolean(),
            chunk_size => pos_integer(),
            chunk_count => pos_integer()}.

change/0

-type change() :: map().

changes_opts/0

-type changes_opts() ::
          #{include_docs => boolean(),
            limit => non_neg_integer(),
            descending => boolean(),
            style => main_only | all_docs,
            doc_ids => [docid()],
            paths => [binary()],
            query => barrel_query:query_spec(),
            channel => binary(),
            include_leaves => boolean()}.

changes_result/0

-type changes_result() ::
          #{changes := [change()], last_hlc := barrel_hlc:timestamp(), pending := non_neg_integer()}.

compact_change/0

-type compact_change() :: {docid(), barrel_hlc:timestamp(), binary(), boolean(), non_neg_integer()}.

compact_fold_fun/0

-type compact_fold_fun() ::
          fun((compact_change(), Acc :: term()) -> {ok, Acc :: term()} | {stop, Acc :: term()} | stop).

continuation_info/0

-type continuation_info() ::
          #{last_hlc := barrel_hlc:timestamp(), has_more := boolean(), continuation => binary()}.

db_config/0

-type db_config() :: #{path => string(), store => module(), atom() => term()}.

db_name/0

-type db_name() :: binary().

db_ref/0

-type db_ref() :: pid() | atom().

doc/0

-type doc() :: #{binary() => term()}.

doc_info/0

-type doc_info() :: #{id := docid(), rev := revid(), deleted := boolean(), revtree := revtree()}.

docid/0

-type docid() :: binary().

endpoint/0

-type endpoint() :: db_name() | {node(), db_name()} | {module(), term()}.

fold_fun/0

-type fold_fun() :: fun((change(), Acc :: term()) -> {ok, Acc :: term()} | {stop, Acc :: term()} | stop).

rep_options/0

-type rep_options() ::
          #{continuous => boolean(),
            since => seq_string(),
            filter => fun((doc()) -> boolean()),
            atom() => term()}.

rev_info/0

-type rev_info() ::
          #{id := revid(),
            parent := revid() | undefined,
            deleted := boolean(),
            attachments => #{binary() => att_info()}}.

revid/0

-type revid() :: binary().

revtree/0

-type revtree() :: #{revid() => rev_info()}.

seq/0

-type seq() :: barrel_hlc:timestamp().

seq_string/0

-type seq_string() :: binary().

view_name/0

-type view_name() :: binary().

view_result/0

-type view_result() :: #{key := term(), value := term(), id := docid()}.

Functions

count_changes_since(StoreRef, DbName, Since)

-spec count_changes_since(barrel_store_rocksdb:db_ref(), db_name(), barrel_hlc:timestamp()) ->
                             non_neg_integer().

Count changes since a given HLC timestamp (exclusive)

delete_old_change(StoreRef, DbName, OldHlc, DocId)

-spec delete_old_change(barrel_store_rocksdb:db_ref(), db_name(), barrel_hlc:timestamp(), docid()) -> ok.

Delete an old HLC entry (when document is updated)

encode_change(DocInfo)

-spec encode_change(#{id := binary(), rev := binary(), deleted => boolean(), _ => _}) -> binary().

Encode change to compact binary format. Format: <<DocIdLen:16, DocId/binary, RevLen:16, Rev/binary, Deleted:8, NumConflicts:16, HasDoc:8, [DocCbor/binary]>> Stores conflict count for quick check; optionally includes doc body for filtering.

fold_changes(StoreRef, DbName, Since, Fun, Acc)

-spec fold_changes(barrel_store_rocksdb:db_ref(),
                   db_name(),
                   barrel_hlc:timestamp() | first,
                   fold_fun(),
                   term()) ->
                      {ok, term(), barrel_hlc:timestamp()}.

Fold over changes since a given HLC timestamp (exclusive) Changes returned are strictly after the given HLC. Use 'first' to get all changes from the beginning.

fold_changes_compact(StoreRef, DbName0, Since, Fun, Acc)

-spec fold_changes_compact(barrel_store_rocksdb:db_ref(),
                           db_name(),
                           barrel_hlc:timestamp() | first,
                           compact_fold_fun(),
                           term()) ->
                              {ok, term(), barrel_hlc:timestamp()}.

Fold over changes with compact tuple format for efficiency. Callback receives {DocId, Hlc, Rev, Deleted, NumConflicts} tuples. Use for internal iteration; convert to maps at API boundary. Always uses long_scan mode for optimal sequential read performance.

get_changes(StoreRef, DbName, Since, Opts)

-spec get_changes(barrel_store_rocksdb:db_ref(),
                  db_name(),
                  barrel_hlc:timestamp() | first,
                  changes_opts()) ->
                     {ok, [change()], barrel_hlc:timestamp()} | {error, term()}.

Get a list of changes since an HLC timestamp

get_changes_by_path(StoreRef, DbName, PathPattern, Since, Opts)

-spec get_changes_by_path(barrel_store_rocksdb:db_ref(),
                          db_name(),
                          binary(),
                          barrel_hlc:timestamp() | first,
                          map()) ->
                             {ok, [change()], barrel_hlc:timestamp()}.

Get changes for a specific path pattern since HLC. Scans the path_hlc index directly for efficient filtered queries.

get_changes_chunked(StoreRef, DbName, ContinuationOrSince, Opts)

-spec get_changes_chunked(barrel_store_rocksdb:db_ref(),
                          db_name(),
                          barrel_hlc:timestamp() | first | binary(),
                          changes_opts()) ->
                             {ok, [change()], continuation_info()}.

Get changes with continuation support for pagination. Accepts either an HLC timestamp, 'first', or a continuation token (binary). Returns changes along with continuation info for fetching the next batch.

Example usage:

  %% First request
  {ok, Changes1, Info1} = get_changes_chunked(Store, Db, first, #{limit => 100}),
  %% Check if more changes available
  case maps:get(has_more, Info1) of
      true ->
          Continuation = maps:get(continuation, Info1),
          {ok, Changes2, Info2} = get_changes_chunked(Store, Db, Continuation, #{limit => 100});
      false ->
          done
  end.

get_last_hlc(StoreRef, DbName0)

Get the last HLC timestamp for a database First tries the fast path (metadata key), falls back to reverse iteration

get_last_seq(StoreRef, DbName)

-spec get_last_seq(barrel_store_rocksdb:db_ref(), db_name()) -> binary().

Get the last sequence (opaque encoded HLC) for a database The sequence is an opaque binary that can be used for ordering.

has_changes_since(StoreRef, DbName, Since)

-spec has_changes_since(barrel_store_rocksdb:db_ref(), db_name(), barrel_hlc:timestamp()) -> boolean().

Fast O(1) check if there are changes since a given HLC. Uses time-bucketed hints to avoid scanning the change log. Returns true if there might be changes, false if definitely no changes. Note: May return true even if no changes (false positive OK, false negative not OK).

remove_path_index_ops(DbName0, Hlc, DocInfo)

-spec remove_path_index_ops(db_name(),
                            barrel_hlc:timestamp(),
                            #{id := binary(), rev := binary(), deleted := boolean(), _ => _}) ->
                               [tuple()].

Exact inverse of write_path_index_ops/3, for a change row being removed from the feed (timeline rewind): deletes the path_hlc rows and posting entries that the write created, deriving topics the same way (a tombstone's rows live under its own doc id, and the posting marker carries the row's own rev and deleted flag).

update_change_bucket_ops(StoreRef, DbName, Hlc)

-spec update_change_bucket_ops(barrel_store_rocksdb:db_ref(), db_name(), barrel_hlc:timestamp()) ->
                                  [{put, binary(), binary()}].

Return batch operation to update change bucket. Buckets store {MinHlc, MaxHlc, Count} in compact binary format. Format: <<MinHlc:12/binary, MaxHlc:12/binary, Count:32>>

update_path_index_ops(DbName0, NewHlc, NewDocInfo, OldHlc, OldDoc)

-spec update_path_index_ops(db_name(),
                            barrel_hlc:timestamp(),
                            #{id := binary(), rev := binary(), deleted := boolean(), _ => _},
                            barrel_hlc:timestamp() | undefined,
                            map() | undefined) ->
                               [tuple()].

Generate operations to update path index (remove old entries + add new). Used when a document is updated to maintain a current path index without stale entries.

write_change(StoreRef, DbName, Hlc, DocInfo)

-spec write_change(barrel_store_rocksdb:db_ref(), db_name(), barrel_hlc:timestamp(), doc_info()) -> ok.

Write a change entry for a document. Also updates change buckets for idle poll optimization.

write_change_ops(DbName0, Hlc, DocInfo)

-spec write_change_ops(db_name(),
                       barrel_hlc:timestamp(),
                       #{id := binary(), rev := binary(), deleted := boolean(), _ => _}) ->
                          [{put, binary(), binary()}].

Return batch operation to write a change entry. Use this to combine with other operations in a single write_batch. Also updates the last_hlc metadata for efficient get_last_seq lookups. Note: Does NOT update change buckets - call write_change/4 for full update.

write_path_index_ops(DbName0, Hlc, DocInfo)

-spec write_path_index_ops(db_name(),
                           barrel_hlc:timestamp(),
                           #{id := binary(), rev := binary(), deleted := boolean(), _ => _}) ->
                              [{put, binary(), binary()} | {posting_append, binary(), binary()}].

Generate operations to index a change by all its paths. Creates entries under path_hlc/{db}/{topic}/{hlc} for exact matches, and prefix_changes/{db}/{prefix}/{bucket} posting lists for wildcard queries.