@spec append(
  event_stream :: Enumerable.t(),
  connection :: Spear.Connection.t(),
  stream_name :: String.t(),
  opts :: Keyword.t()
) ::
  :ok
  | {:ok, AppendResp.t()}
  | {:error, reason :: Spear.ExpectationViolation.t() | any()}

Appends an enumeration of events to an EventStoreDB stream

event_stream is an enumerable which may either be a collection of Spear.Event.t/0 structs or more low-level Spear.Records.Streams.append_resp/0 records. In cases where the enumerable produces Spear.Event.t/0 structs, they will be lazily mapped to Spear.Records.Streams.append_req/0 records before being encoded to wire data.

See the Writing Events guide for more information about writing events.

Options

• :expect - (default: :any) the expectation to set on the status of the stream. The write will fail if the expectation fails. See Spear.ExpectationViolation for more information about expectations.
• :timeout - (default: 5_000 - 5s) the GenServer timeout for calling the RPC.
• :raw? - (default: false) a boolean which controls whether the return signature should be a simple :ok | {:error, any()} or {:ok, AppendResp.t()} | {:error, any()}. This can be used to extract metadata and information from the append response which is not available through the simplified return API, such as the stream's revision number after writing the events.
• :credentials - (default: nil) a two-tuple {username, password} to use as credentials for the request. This option overrides any credentials set in the connection configuration, if present. See the Security guide for more details.

Examples

iex> [Spear.Event.new("es_supported_clients", %{})]
...> |> Spear.append(conn, expect: :exists)
:ok
iex> [Spear.Event.new("es_supported_clients", %{})]
...> |> Spear.append(conn, expect: :empty)
{:error, %Spear.ExpectationViolation{current: 1, expected: :empty}}

@spec append_batch(
  event_stream :: Enumerable.t(),
  connection :: Spear.Connection.t(),
  request_id :: reference() | :new,
  stream_name :: String.t(),
  opts :: Keyword.t()
) :: {:ok, batch_id :: String.t(), request_id :: reference()} | {:error, term()}

Appends an enumeration of events to an EventStoreDB stream

BatchAppend is a feature added in EventStoreDB version 21.6.0 which aims to optimize append throughput. It works like a persistent subscription in reverse: the client sends chunks of events to write and once the events have been committed, the client receives an acknowledgement message.

BatchAppends have a life cycle: a new batch append request is created by passing the :new atom as the request_id argument and a request is concluded by calling Spear.cancel_subscription/2 on the request_id.

See the Writing Events guide for more information about batching and when to prefer the append_batch/5 function over append/4.

Fragmentation

By default, each invocation of append_batch/5 sends a single protobuf message across the wire for all events in the event_stream argument. If the event_stream argument is very large (> ~1MB), this may be undesirable from a networking perspective.

The :done? flag can be set to false to mark a batch as a fragment, which prevents the EventStoreDB from attempting to commit any events sent for the batch until a fragment with the :done? flag set to true is sent. Each batch of events after the first fragment must pass the initial fragment's batch_id in the :batch_id option. A set of fragments can be concluded by sending a final append_batch/5 with the :done? option set to true.

Options

• :done? - (default: true) controls whether the current chunk of events being written is complete. See the "Fragmentation" section above for more details.
• :batch_id - (default: Spear.Uuid.uuid_v4()) the unique ID of the batch of events being appended. This must be passed the batch_id returned by the first request which sets :done? to false until a final fragment is appended (done?: true). See the "Fragmentation" section above for more details.
• :expect - (default: :any) the expectation to set on the status of the stream. The write will fail if the expectation fails. See Spear.ExpectationViolation for more information about expectations.
• :send_ack_to - (default: self()) a process or process name which should receive acknowledgement messages detailing whether a batch has succeeded or failed to be committed by the deadline.
• :raw? - (default: false) a boolean which controls whether messages emitted to the :send_ack_to process are decoded from Spear.Records.Streams.batch_append_resp/0 records. Spear preserves most of the information when decoding the record into the Spear.BatchAppendResult.t/0 struct, but it discards duplicated information such as stream name and expected revision. Setting the :raw? flag allows one to decode the record however they wish.
• :credentials - (default: nil) a two-tuple {username, password} to use as credentials for the request. This option overrides any credentials set in the connection configuration, if present. See the Security guide for more details.
• :deadline - (default: nil) the deadline for the batch to be appended. This is like :timeout but is interpreted on the EventStoreDB server. This may be a DateTime or a tuple of {seconds, nanos} since 1970 (smeared) but may be a tuple {:duration, seconds, nanos} when using EventStoreDB version 21.10.5 or higher.
• :timeout - (default: 5_000) the timeout for the initial call to open the batch request.

Examples

iex> {:ok, first_batch_id, request_id} =
...>   Spear.append_batch(first_batch, conn, :new, first_stream_name)
{:ok, "496ba076-098f-4108-a2ca-c73f7b94c06f", #Reference<0.1691282361.2492989441.105913>}
iex> receive do
...>   %Spear.BatchAppendResult{request_id: ^request_id, batch_id: ^first_batch_id, result: result} ->
...>     result
...> end
:ok
iex> {:ok, second_batch_id, ^request_id} =
...>   Spear.append_batch(second_batch, conn, batch_id, second_stream_name)
{:ok, "3a1ed972-716c-4679-9cc7-c23c3544e538", #Reference<0.1691282361.2492989441.105913>}
iex> receive(do: (%Spear.BatchAppendResult{request_id: ^request_id, batch_id: ^second_batch_id, result: result}) -> result))
:ok
iex> {:ok, third_batch_id, ^request_id} =
...>   Spear.append_batch(third_batch, conn, batch_id, third_stream_name)
{:ok, "b4eb1330-8c59-48d0-8a0b-2df33672cc0b", #Reference<0.1691282361.2492989441.105913>}
iex> receive(do: (%Spear.BatchAppendResult{request_id: ^request_id, batch_id: ^third_batch_id, result: result}) -> result))
:ok
iex> Spear.cancel_subscription(conn, request_id)
:ok

@spec append_batch_stream(
  batch_stream :: Enumerable.t(),
  connection :: Spear.Connection.t()
) ::
  Enumerable.t()

A convenience wrapper around append_batch/5 for transforming a stream into a batch append operation.

The append_batch/5 function provides fine-grained control over the batch append feature. This function transforms an input stream of batches to apply the append_batch/5 on all of them, making sure to clean up the request once the batch is finished.

The expected batch_stream is an enumerable with each element follows the format

{stream_name :: String.t(), events :: [Spear.Event.t()]}
# or
{stream_name :: String.t(), events :: [Spear.Event.t()], opts :: Keyword.t()}

Where opts can be any option below. The options below are applied to each call to append_batch/5 when provided, except :credentials which is only applied when specified on the first batch.

The resulting stream must be run (with an Enum function or Stream.run/1). Each element is mapped to the acknowledgement Spear.BatchAppendResult.t/0 responses for each batch attempting to be appended.

Options

• :expect - (default: :any) the expectation to set on the status of the stream. The write will fail if the expectation fails. See Spear.ExpectationViolation for more information about expectations.
• :raw? - (default: false) a boolean which controls whether messages emitted to the :send_ack_to process are decoded from Spear.Records.Streams.batch_append_resp/0 records. Spear preserves most of the information when decoding the record into the Spear.BatchAppendResult.t/0 struct, but it discards duplicated information such as stream name and expected revision. Setting the :raw? flag allows one to decode the record however they wish.
• :credentials - (default: nil) a two-tuple {username, password} to use as credentials for the request. This option overrides any credentials set in the connection configuration, if present. See the Security guide for more details.
• :timeout - (default: 5_000) the timeout for the initial call to open the batch request. After the first batch, this argument instead controls how long each append_batch/5 operation will await an acknowledgement.

Examples

batch_stream
|> Spear.append_batch_stream(conn)
|> Enum.reduce_while(:ok, fn ack, _acc ->
  case ack.result do
    :ok -> {:cont, :ok}
    {:error, reason} -> {:halt, {:error, reason}}
  end
end)
#=> :ok

@spec delete_stream(
  connection :: Spear.Connection.t(),
  stream_name :: String.t(),
  opts :: Keyword.t()
) :: :ok | {:error, any()}

Deletes an EventStoreDB stream

EventStoreDB supports two kinds of stream deletions: soft-deletes and tombstones. By default this function will perform a soft-delete. Pass the tombstone?: true option to tombstone the stream.

Soft-deletes make the events in the specified stream no longer accessible through reads. A scavenge operation will reclaim the disk space taken by any soft-deleted events. New events may be written to a soft-deleted stream. When reading soft-deleted streams, :from options of :start and :end will behave as expected, but all events in the stream will have revision numbers off-set by the number of deleted events.

Tombstoned streams may not be written to ever again. Attempting to write to a tombstoned stream will fail with a gRPC :failed_precondition error

iex> [Spear.Event.new("delete_test", %{})] |> Spear.append(conn, "delete_test_0")
:ok
iex> Spear.delete_stream(conn, "delete_test_0", tombstone?: true)
:ok
iex> [Spear.Event.new("delete_test", %{})] |> Spear.append(conn, "delete_test_0")
{:error,
 %Spear.Grpc.Response{
   data: "",
   message: "Event stream 'delete_test_0' is deleted.",
   status: :failed_precondition,
   status_code: 9
 }}

Options

• :tombstone? - (default: false) controls whether the stream is soft-deleted or tombstoned.
• :timeout - (default: 5_000 - 5s) the time allowed to block while waiting for the EventStoreDB to delete the stream.
• :expect - (default: :any) the expected state of the stream when performing the deletion. See append/4 and Spear.ExpectationViolation for more information.
• :credentials - (default: nil) a two-tuple {username, password} to use as credentials for the request. This option overrides any credentials set in the connection configuration, if present. See the Security guide for more details.

Examples

iex> Spear.append(events, conn, "my_stream")
:ok
iex> Spear.delete_stream(conn, "my_stream")
:ok
iex> Spear.stream!(conn, "my_stream") |> Enum.to_list()
[]

@spec get_stream_metadata(
  connection :: Spear.Connection.t(),
  stream :: String.t(),
  opts :: Keyword.t()
) :: {:ok, Spear.StreamMetadata.t()} | {:error, any()}

Queries the metadata for a stream

Note that the stream argument is passed through meta_stream/1 before being read. It is not necessary to call that function on the stream name before passing it as stream.

If no metadata has been set on a stream {:error, :unset} is returned.

Options

Under the hood, get_stream_metadata/3 uses read_stream/3 and all options are passed directly to that function. These options are overridden, however, and cannot be changed:

• :direction
• :from
• :max_count
• :raw?

Examples

iex> Spear.get_stream_metadata(conn, "my_stream")
{:error, :unset}
iex> Spear.get_stream_metadata(conn, "some_stream_with_max_count")
{:ok, %Spear.StreamMetadata{max_count: 50_000, ..}}

@spec read_stream(Spear.Connection.t(), String.t() | :all, Keyword.t()) ::
  {:ok, event_stream :: Enumerable.t()} | {:error, any()}

Reads a chunk of events from an EventStoreDB stream into an enumerable

Unlike stream!/3, this function will only read one chunk of events at a time specified by the :max_count option. This function also does not raise in cases of error, instead returning an ok- or error-tuple.

If the stream_name EventStoreDB stream does not exist (is empty) and the gRPC request succeeds for this function, {:ok, []} will be returned.

Options

• :from - (default: :start) the EventStoreDB stream revision from which to read. Valid values include :start, :end, any non-negative integer representing the event number revision in the stream and events. Event numbers are inclusive (e.g. reading from 0 will first return the event with revision 0 in the stream, if one exists). :start and :end are treated as inclusive (e.g. :start will return the first event in the stream). Events (either Spear.Event or ReadResp records) can also be supplied and will be treated as inclusive.
• :direction - (default: :forwards) the direction in which to read the EventStoreDB stream. Valid values include :forwards and :backwards. Reading the EventStoreDB stream forwards will return events in the order in which they were written to the EventStoreDB; reading backwards will return events in the opposite order.
• :filter - (default: nil) the server-side filter to apply. This option is only valid if the stream_name is :all. See Spear.Filter for more information. This feature requires EventStoreDB vTODO+.
• :resolve_links? - (default: true) whether or not to request that link references be resolved. See the moduledocs for more information about link resolution.
• :max_count - (default: 42) the maximum number of events to read from the EventStoreDB stream. Any positive integer is valid. Even if the stream is longer than this :max_count option, only :max_count events will be returned from this function. :infinity is not a valid value for :max_count. Use stream!/3 for an enumerable which reads an EventStoreDB stream in its entirety in chunked network requests.
• :timeout - (default: 5_000 - 5s) the time allowed for the read of the single chunk of events in the EventStoreDB stream. Note that the gRPC request which reads events from the EventStoreDB is front-loaded in this function: the :timeout covers the time it takes to read the events. The timeout may be exceeded
• :raw?: - (default: false) controls whether or not the enumerable event_stream is decoded to Spear.Event structs from their raw ReadResp output. Setting raw?: true prevents this transformation and leaves each event as a Spear.Records.Streams.read_resp/0 record. See Spear.Event.from_read_response/2 for more information.
• :credentials - (default: nil) a two-tuple {username, password} to use as credentials for the request. This option overrides any credentials set in the connection configuration, if present. See the Security guide for more details.

Timing and Timeouts

The gRPC request which reads events from the EventStoreDB is front-loaded in this function: this function returns immediately after receiving all data off the wire from the network request. This means that the :timeout option covers the gRPC request and response time but not any time spend decoding the response (see the Enumeration Characteristics section below for more details on how the enumerable decodes messages).

The default timeout of 5s may not be enough time in cases of either reading very large numbers of events or reading events with very large bodies.

Note that up to the :max_count of events is returned from this call depending on however many events are in the EventStoreDB stream being read. When tuning the :timeout option, make sure to test against a stream which is at least as long as :max_count events.

Enumeration Characteristics

The event_stream Enumerable.t/0 returned in the success case of this function is a wrapper around the bytes received from the gRPC response. Note that when the {:ok, event_stream} is returned, the gRPC request has already concluded.

This offers only marginal performance improvement: an enumerable is returned mostly for consistency in the Spear API.

Examples

# say we have 5 events in the stream "es_supported_clients"
iex> {:ok, events} = Spear.read_stream(conn, "es_supported_clients", max_count: 2)
iex> events |> Enum.count()
2
iex> {:ok, events} = Spear.read_stream(conn, "es_supported_clients", max_count: 10)
iex> events |> Enum.count()
5
iex> events |> Enum.take(1)
[
  %Spear.Event{
    body: %{"languages" => ["typescript", "javascript"], "runtime" => "NodeJS"},
    id: "1fc908c1-af32-4d06-a9bd-3bf86a833fdf",
    metadata: %{..},
    type: "grpc-client"
  }
]

@spec set_stream_metadata(
  connection :: Spear.Connection.t(),
  stream :: String.t(),
  metadata :: Spear.StreamMetadata.t(),
  opts :: Keyword.t()
) :: :ok | {:error, any()}

Sets a stream's metadata

Note that the stream argument is passed through meta_stream/1 before being read. It is not necessary to call that function on the stream name before passing it as stream.

Options

This function uses append/4 under the hood. All options are passed to the opts argument of append/4.

Examples

# only allow admins to read, write, and delete the stream (or stream metadata)
iex> metadata = %Spear.StreamMetadata{acl: Spear.Acl.admins_only()}
iex> Spear.set_stream_metadata(conn, stream, metadata)
:ok

@spec stream!(
  connection :: Spear.Connection.t(),
  stream_name :: String.t() | :all,
  opts :: Keyword.t()
) :: event_stream :: Enumerable.t()

Collects an EventStoreDB stream into an enumerable

This function may raise in cases where the gRPC requests fail to read events from the EventStoreDB (in cases of timeout or unavailability).

This function does not raise if a stream does not exist (is empty), instead returning an empty enumerable [].

connection may be any valid GenServer name (including PIDs) for a process running the Spear.Connection GenServer.

stream_name can be any stream, existing or not, including projected streams such as category streams or event-type streams. The :all atom may be passed as stream_name to read all events in the EventStoreDB.

Options

• :from - (default: :start) the EventStoreDB stream revision from which to read. Valid values include :start, :end, any non-negative integer representing the event number revision in the stream and events. Event numbers are inclusive (e.g. reading from 0 will first return the event with revision 0 in the stream, if one exists). :start and :end are treated as inclusive (e.g. :start will return the first event in the stream). Events (either Spear.Event or ReadResp records) can also be supplied and will be treated as inclusive.
• :direction - (default: :forwards) the direction in which to read the EventStoreDB stream. Valid values include :forwards and :backwards. Reading the EventStoreDB stream forwards will return events in the order in which they were written to the EventStoreDB; reading backwards will return events in the opposite order.
• :filter - (default: nil) the server-side filter to apply. This option is only valid if the stream_name is :all. See Spear.Filter for more information. This feature requires EventStoreDB vTODO+.
• :resolve_links? - (default: true) whether or not to request that link references be resolved. See the moduledocs for more information about link resolution.
• :chunk_size - (default: 128) the number of events to read from the EventStoreDB at a time. Any positive integer is valid. See the enumeration characteristics section below for more information about how :chunk_size works and how to tune it.
• :timeout - (default: 5_000 - 5s) the time allowed for the read of a single chunk of events in the EventStoreDB stream. This time is not cumulative: an EventStoreDB stream 100 events long which takes 5s to read each chunk may be read in chunks of 20 events culumaltively in 25s. A timeout of 5_001ms would not raise a timeout error in that scenario (assuming the chunk read consistently takes <= 5_000 ms).
• :raw?: - (default: false) controls whether or not the enumerable event_stream is decoded to Spear.Event structs from their raw Spear.Records.Streams.read_resp/0 output. Setting raw?: true prevents this transformation and leaves each event as a ReadResp record. See Spear.Event.from_read_response/2 for more information.
• :credentials - (default: nil) a two-tuple {username, password} to use as credentials for the request. This option overrides any credentials set in the connection configuration, if present. See the Security guide for more details.

Enumeration Characteristics

The event_stream Enumerable.t/0 returned by this function initially contains a buffer of bytes from the first read of the stream stream_name. This buffer potentially contains up to :chunk_size messages when run. The enumerable is written as a formula which abstracts away the chunking nature of the gRPC requests, however, so even though the EventStoreDB stream is read in chunks (per the :chunk_size option), the entire EventStoreDB stream can be read by running the enumeration (e.g. with Enum.to_list/1). Note that the stream will make a gRPC request to read more events whenever the buffer runs dry with up to :chunk_size messages filling the buffer on each request.

:chunk_size is difficult to tune as it causes a tradeoff between (gRPC) request duration and number of messages added to the buffer. A higher :chunk_size may hydrate the buffer with more events and reduce the number of gRPC requests needed to read an entire stream, but it also increases the number of messages that will be sent over the network per request which could decrease reliability. Generally speaking, a lower :chunk_size is appropriate for streams in which the events are large and a higher :chunk_size is appropriate for streams with many small events. Manual tuning and trial-and-error can be used to find a performant :chunk_size setting for any individual environment.

Examples

iex> Spear.stream!(MyConnection, "es_supported_clients", chunk_size: 1) |> Enum.take(1)
[
  %Spear.Event{
    body: %{"languages" => ["typescript", "javascript"], "runtime" => "NodeJS"},
    id: "1fc908c1-af32-4d06-a9bd-3bf86a833fdf",
    metadata: %{..},
    type: "grpc-client"
  }
]
# say we have 5 events in the "es_supported_clients" stream
iex> Spear.stream!(MyConnection, "es_supported_clients", chunk_size: 3) |> Enum.count()
5

@spec subscribe(
  connection :: Spear.Connection.t(),
  subscriber :: pid() | GenServer.name(),
  stream_name :: String.t() | :all,
  opts :: Keyword.t()
) :: {:ok, subscription_reference :: reference()} | {:error, any()}

Subscribes a process to an EventStoreDB stream

Unlike read_stream/3 or stream!/3, this function does not return an enumerable. Instead the subscriber process is signed up to receive messages for subscription events. Events are emitted in order as info messages with the signature

Spear.Event.t() | Spear.Filter.Checkpoint.t()

or if the raw?: true option is provided, Spear.Records.Streams.read_resp/0 records will be returned in the shape of

{subscription :: reference(), Spear.Records.Streams.read_resp()}

This function will block the caller until the subscription has been confirmed by the EventStoreDB.

When the subscription is terminated, the subscription process will receive a message in the form of {:eos, subscription, reason}. {:eos, subscription, :closed} is emitted when the connection between EventStoreDB and subscriber is severed and {:eos, subscription, :dropped} is emitted when the EventStoreDB explicitly drops a subscription. If this message is received, the subscription is considered to be concluded and the subscription process must re-subscribe from the last received event or checkpoint to resume the subscription. subscription is the reference returned by this function.

Events can be correlated to their subscription via the subscription reference returned by this function. The subscription reference is included in Spear.Event.metadata.subscription, Spear.Filter.Checkpoint.subscription, and in the {:eos, subscription, reason} tuples as noted above.

Subscriptions can be gracefully shut down with Spear.cancel_subscription/3. The subscription will be cancelled by the connection process if the subscriber process exits.

Options

• :from - (default: :start) the EventStoreDB stream revision from which to read. Valid values include :start, :end, any non-negative integer representing the event number revision in the stream and events. Event numbers are exclusive (e.g. reading from 0 will first return the event numbered 1 in the stream, if one exists). :start and :end are treated as inclusive (e.g. :start will return the first event in the stream). Events and checkpoints (Spear.Event.t/0, ReadResp records, or Spear.Filter.Checkpoint.t/0) can also be supplied and will be treated as exclusive.
• :filter - (default: nil) the server-side filter to apply. This option is only valid if the stream_name is :all. See Spear.Filter for more information.
• :resolve_links? - (default: true) whether or not to request that link references be resolved. See the moduledocs for more information about link resolution.
• :timeout - (default: 5_000) the time to wait for the EventStoreDB to confirm the subscription request.
• :raw? - (default: false) controls whether the events are sent as raw ReadResp records or decoded into Spear.Event.t/0s
• :credentials - (default: nil) a two-tuple {username, password} to use as credentials for the request. This option overrides any credentials set in the connection configuration, if present. See the Security guide for more details.

Examples

# say there are 3 events in the EventStoreDB stream "my_stream"
iex> {:ok, sub} = Spear.subscribe(conn, self(), "my_stream", from: 0)
{:ok, #Reference<0.1160763861.3015180291.51238>}
iex> flush
%Spear.Event{} # second event
%Spear.Event{} # third event
:ok
iex> Spear.cancel_subscription(conn, sub)
:ok

iex> {:ok, sub} = Spear.subscribe(conn, self(), :all, filter: Spear.Filter.exclude_system_events())
iex> flush
%Spear.Filter.Checkpoint{}
%Spear.Filter.Checkpoint{}
%Spear.Event{}
%Spear.Event{}
%Spear.Filter.Checkpoint{}
%Spear.Event{}
%Spear.Filter.Checkpoint{}
:ok
iex> GenServer.call(conn, :close)
{:ok, :closed}
iex> flush
{:eos, #Reference<0.1160763861.3015180291.51238>, :closed}