Elixir SDK

Official SDK

This SDK is officially developed and maintained by the native web GmbH, the creators of EventSourcingDB. It is aligned with the latest platform features and receives regular updates.

Installation#

To install the SDK, add it to your dependencies in mix.exs:

Changed in 2.0

defp deps do
  [
    {:eventsourcingdb, "~> 1.0"}
  ]
end

Then run:

mix deps.get

You can also find this SDK on Hex.pm.

Capabilities#

This SDK intentionally focuses on application-level use cases. It provides core functionality for writing, reading, and observing events, and deliberately excludes operational concerns such as backup, restore, or health monitoring.

For a detailed explanation of the specific requirements behind each capability, see the Compliance Criteria.

Compatibility

Elixir SDK 1.3 runs with EventSourcingDB 1.1 and later.

Essentials

✓ Initialize the Client
✓ Ping the Server
✓ Verify API Token
Added in 1.3Broken in 1.3, fixed in 2.4✓ Detect a Stalled Connection

Writing Events

✓ Write Single Events
✓ Write Multiple Events
✓ Use isSubjectPristine
Requires EventSourcingDB 1.2✓ Use isSubjectPopulated
✓ Use isSubjectOnEventId
✓ Use isEventQlQueryTrue
✓ Include OpenTelemetry Trace Context

Reading Events

✓ Read Events by Subject
✓ Read Recursively
✓ Read Chronological and Anti-Chronological
✓ Use Boundaries
✓ Use fromLatestEvent

Using EventQL

✓ Execute EventQL Queries

Observing Events

✓ Observe Events By Subject
✓ Observe Recursively
✓ Use Boundaries
✓ Use fromLatestEvent

Metadata and Discovery

✓ Register Event Schemas
✓ List Subjects
✓ List Event Types
✓ List a Specific Event Type
✓ Verify Event Hashes
✓ Verify Event Signatures

Testcontainers Support

✓ Start and Stop EventSourcingDB in Tests
✓ Use Custom Configuration (e.g., API Token, Port, …)

Getting Started#

Create a client by providing the URL of your EventSourcingDB instance and the API token to use:

client = EventSourcingDB.Client.new("http://localhost:3000", "secret")

Then call the ping function to check whether the instance is reachable. If it is not, the function will return an error:

:ok = EventSourcingDB.ping(client)

Note that ping does not require authentication, so the call may succeed even if the API token is invalid.

Changed in 2.0If you want to verify the API token, call verify_api_token. If the token is invalid, the function will return an error:

:ok = EventSourcingDB.verify_api_token(client)

Using HTTP/2Added in 1.2#

By default, the client uses HTTP/1.1. To use HTTP/2, which EventSourcingDB offers over HTTPS, pass the protocol on to Req via the req_options of the client:

client = EventSourcingDB.Client.new(
  base_url: "https://localhost:4000",
  api_token: "secret",
  req_options: [connect_options: [protocols: [:http2]]]
)

Streams work over HTTP/2 as well, and stay open for as long as data arrives.

Configuring the Stall TimeoutAdded in 1.3Changed in 2.0#

A request fails once it waits for the server for 30 seconds without anything arriving: while connecting, while waiting for the response headers, or while waiting for the rest of the response. Everything that arrives restarts the 30 seconds, so this does not limit how long a request takes: a stream stays open for as long as events, rows, or heartbeats keep arriving. To wait for a different time, pass the stall_timeout option to the client, in milliseconds:

client = EventSourcingDB.Client.new(
  base_url: "http://localhost:3000",
  api_token: "secret",
  stall_timeout: 10_000
)

A request that stalls returns an EventSourcingDB.Errors.TransmissionError whose reason, such as a Req.TransportError, has the reason :timeout, and a stream that stalls while you enumerate it raises that error. The streams of observing events and running EventQL queries, which receive heartbeats, raise an EventSourcingDB.Errors.HeartbeatTimeout error instead.

Req retries a request that is safe to repeat after a transient error, such as a timeout over HTTP/1, by default three times, after about one, two, and four seconds. Among the requests of this client SDK, that is only ping, since it uses GET, so with the default stall timeout, a ping to a server that stalls fails only after about two minutes: four attempts of 30 seconds each, plus the seven seconds in between. To have it fail after the stall timeout, turn off Req's retries via req_options: [retry: false].

If you do not set stall_timeout, but pass a receive_timeout on to Req via req_options, every request other than observing events and running EventQL queries waits for that receive_timeout instead, as it always has, and connecting keeps Req's own timeout. A stall_timeout that you set takes precedence.

Note that over HTTP/2, a request with a single answer, such as writing events, must be answered completely within the stall timeout, since there Finch bounds the request as a whole rather than each wait for data. Streams are not affected.

Writing Events#

Call the write_events function and hand over a list with one or more events. You do not have to provide all event fields – some are automatically added by the server.

Specify source, subject, type, and data according to the CloudEvents format. Optionally, specify the trace context of OpenTelemetry as traceparent and tracestate.

The function returns the written events, including the fields added by the server:

event = %EventSourcingDB.EventCandidate{
  source: "https://library.eventsourcingdb.io",
  subject: "/books/42",
  type: "io.eventsourcingdb.library.book-acquired",
  data: %{
    "title" => "2001 – A Space Odyssey",
    "author" => "Arthur C. Clarke",
    "isbn" => "978-0756906788"
  }
}

case EventSourcingDB.write_events(client, [event]) do
  {:ok, events} -> # ...
  {:error, reason} -> # ...
end

Using the `IsSubjectPristine` precondition#

If you only want to write events in case a subject (such as /books/42) does not yet have any events, use the IsSubjectPristine precondition and pass it in a list as the third argument:

case EventSourcingDB.write_events(
  client,
  [event],
  [%EventSourcingDB.IsSubjectPristine{subject: "/books/42"}]
) do
  {:ok, events} -> # ...
  {:error, reason} -> # ...
end

Using the `IsSubjectPopulated` preconditionRequires EventSourcingDB 1.2#

If you only want to write events in case a subject (such as /books/42) already has at least one event, use the IsSubjectPopulated precondition and pass it in a list as the third argument:

case EventSourcingDB.write_events(
  client,
  [event],
  [%EventSourcingDB.IsSubjectPopulated{subject: "/books/42"}]
) do
  {:ok, events} -> # ...
  {:error, reason} -> # ...
end

Using the `IsSubjectOnEventId` precondition#

If you only want to write events in case the last event of a subject (such as /books/42) has a specific ID (e.g., 0), use the IsSubjectOnEventId precondition and pass it in a list as the third argument:

case EventSourcingDB.write_events(
  client,
  [event],
  [%EventSourcingDB.IsSubjectOnEventId{subject: "/books/42", event_id: "0"}]
) do
  {:ok, events} -> # ...
  {:error, reason} -> # ...
end

Note that according to the CloudEvents standard, event IDs must be of type string.

Using the `IsEventQLQueryTrue` precondition#

If you want to write events depending on an EventQL query, use the IsEventQLQueryTrue precondition:

case EventSourcingDB.write_events(
  client,
  [event],
  [%EventSourcingDB.IsEventQLQueryTrue{
    query: "FROM e IN events WHERE e.type == 'io.eventsourcingdb.library.book-borrowed' PROJECT INTO COUNT() < 10"
  }]
) do
  {:ok, events} -> # ...
  {:error, reason} -> # ...
end

Note that the query must return a single row with a single value, which is interpreted as a boolean.

Reading Events#

To read all events of a subject, call the read_events function with the subject and an options struct.

The function returns a stream from which you can retrieve one event at a time:

case EventSourcingDB.read_events(client, "/books/42") do
  {:ok, events} -> Enum.to_list(events)
  {:error, reason} -> # ...
end

Changed in 1.3Changed in 2.0If something goes wrong while you enumerate the stream, the stream closes the connection and raises the corresponding error: an EventSourcingDB.Errors.DBError if EventSourcingDB reports an error, an EventSourcingDB.Errors.TransmissionError if the connection fails or nothing arrives for the stall timeout, an EventSourcingDB.Errors.InvalidResponseType if the response contains an item of an unexpected type, or a Jason.DecodeError if the response is not valid JSON.

Added in 1.2Like all functions that return a stream, read_events receives the response as messages sent to the calling process, so enumerate the stream in that process. While you do, the stream leaves the other messages of the process, such as the calls and casts of a GenServer, in its mailbox.

Reading From Subjects Recursively#

If you want to read not only all the events of a subject, but also the events of all nested subjects, set the recursive option to true:

EventSourcingDB.read_events(
  client,
  "/books/42",
  %EventSourcingDB.ReadEventsOptions{recursive: true}
)

This also allows you to read all events ever written. To do so, provide / as the subject and set recursive to true, since all subjects are nested under the root subject.

Reading in Anti-Chronological Order#

By default, events are read in chronological order. To read in anti-chronological order, provide the order option and set it to :antichronological:

EventSourcingDB.read_events(
  client,
  "/books/42",
  %EventSourcingDB.ReadEventsOptions{
    recursive: false,
    order: :antichronological
  }
)

Note that you can also use :chronological to explicitly enforce the default order.

Specifying Bounds#

Sometimes you do not want to read all events, but only a range of events. For that, you can specify the lower_bound and upper_bound options – either one of them or even both at the same time.

Specify the ID and whether to include or exclude it, for both the lower and upper bound:

EventSourcingDB.read_events(
  client,
  "/books/42",
  %EventSourcingDB.ReadEventsOptions{
    recursive: false,
    lower_bound: %EventSourcingDB.BoundOptions{
      type: :inclusive,
      id: "100"
    },
    upper_bound: %EventSourcingDB.BoundOptions{
      type: :exclusive,
      id: "200"
    }
  }
)

Starting From the Latest Event of a Given Type#

To read starting from the latest event of a given type, provide the from_latest_event option and specify the subject, the type, and how to proceed if no such event exists.

Possible options are :read_nothing, which skips reading entirely, or :read_everything, which effectively behaves as if from_latest_event was not specified:

EventSourcingDB.read_events(
  client,
  "/books/42",
  %EventSourcingDB.ReadEventsOptions{
    recursive: false,
    from_latest_event: %EventSourcingDB.ReadFromLatestEventOptions{
      subject: "/books/42",
      type: "io.eventsourcingdb.library.book-borrowed",
      if_event_is_missing: :read_everything
    }
  }
)

Note that from_latest_event and lower_bound can not be provided at the same time.

Running EventQL Queries#

To run an EventQL query, call the run_eventql_query function and provide the query as argument. The function returns a stream:

case EventSourcingDB.run_eventql_query(client, "FROM e IN events PROJECT INTO e") do
  {:ok, rows} -> Enum.to_list(rows)
  {:error, reason} -> # ...
end

Note that each row returned by the stream matches the projection specified in your query.

Changed in 1.3Changed in 2.0While there are no rows to deliver, EventSourcingDB sends a heartbeat every second. If neither a row nor a heartbeat arrives for the stall timeout, 30 seconds by default, the stream closes the connection and raises an EventSourcingDB.Errors.HeartbeatTimeout error.

Added in 1.2Changed in 2.0If anything else goes wrong while you enumerate the stream, the stream closes the connection as well and raises the corresponding error: an EventSourcingDB.Errors.DBError if EventSourcingDB reports an error, an EventSourcingDB.Errors.TransmissionError if the connection fails, an EventSourcingDB.Errors.InvalidResponseType if the response contains an item of an unexpected type, or a Jason.DecodeError if the response is not valid JSON.

Observing Events#

To observe all events of a subject, call the observe_events function with the subject.

The function returns a stream from which you can retrieve one event at a time:

case EventSourcingDB.observe_events(client, "/books/42") do
  {:ok, events} -> Enum.to_list(events)
  {:error, reason} -> # ...
end

Changed in 1.3Changed in 2.0While there are no events to deliver, EventSourcingDB sends a heartbeat every second. If neither an event nor a heartbeat arrives for the stall timeout, 30 seconds by default, the stream closes the connection and raises an EventSourcingDB.Errors.HeartbeatTimeout error.

Observing From Subjects Recursively#

If you want to observe not only all the events of a subject, but also the events of all nested subjects, set the recursive option to true:

EventSourcingDB.observe_events(
  client,
  "/books/42",
  %EventSourcingDB.ObserveEventsOptions{
    recursive: true
  }
)

This also allows you to observe all events ever written. To do so, provide / as the subject and set recursive to true, since all subjects are nested under the root subject.

Specifying Bounds#

Sometimes you do not want to observe all events, but only a range of events. For that, you can specify the lower_bound option.

Specify the ID and whether to include or exclude it:

EventSourcingDB.observe_events(
  client,
  "/books/42",
  %EventSourcingDB.ObserveEventsOptions{
    recursive: false,
    lower_bound: %EventSourcingDB.BoundOptions{
      type: :inclusive,
      id: "100"
    }
  }
)

Starting From the Latest Event of a Given Type#

To observe starting from the latest event of a given type, provide the from_latest_event option and specify the subject, the type, and how to proceed if no such event exists.

Possible options are :wait_for_event, which waits for an event of the given type to happen, or :read_everything, which effectively behaves as if from_latest_event was not specified:

EventSourcingDB.observe_events(
  client,
  "/books/42",
  %EventSourcingDB.ObserveEventsOptions{
    recursive: false,
    from_latest_event: %EventSourcingDB.ObserveFromLatestEventOptions{
      subject: "/books/42",
      type: "io.eventsourcingdb.library.book-borrowed",
      if_event_is_missing: :read_everything
    }
  }
)

Note that from_latest_event and lower_bound can not be provided at the same time.

Registering an Event Schema#

To register an event schema, call the register_event_schema function and hand over an event type and the desired schema:

EventSourcingDB.register_event_schema(
  client,
  "io.eventsourcingdb.library.book-acquired",
  %{
    "type" => "object",
    "properties" => %{
      "title" => %{"type" => "string"},
      "author" => %{"type" => "string"},
      "isbn" => %{"type" => "string"}
    },
    "required" => [
      "title",
      "author",
      "isbn"
    ],
    "additionalProperties" => false
  }
)

Reading Subjects#

Changed in 2.0To list all subjects, call the read_subjects function with / as the base subject. The function returns a stream from which you can retrieve one subject at a time:

case EventSourcingDB.read_subjects(client, "/") do
  {:ok, subjects} -> Enum.to_list(subjects)
  {:error, reason} -> # ...
end

Changed in 2.0If you only want to list subjects within a specific branch, provide the desired base subject instead:

EventSourcingDB.read_subjects(client, "/books")

Reading Event Types#

Changed in 2.0To list all event types, call the read_event_types function. The function returns a stream from which you can retrieve one event type at a time:

case EventSourcingDB.read_event_types(client) do
  {:ok, event_types} -> Enum.to_list(event_types)
  {:error, reason} -> # ...
end

Reading a Specific Event TypeChanged in 2.0#

To read a specific event type, call the read_event_type function with the event type as an argument. The function returns the detailed event type, which includes the schema:

case EventSourcingDB.read_event_type(client, "io.eventsourcingdb.library.book-acquired") do
  {:ok, event_type} -> # ...
  {:error, reason} -> # ...
end

Verifying an Event's HashChanged in 1.3#

To verify the integrity of an event, call the Event.verify_hash function on the event. This recomputes the event's hash locally and compares it to the hash stored in the event. If the hashes differ, the function returns an EventSourcingDB.Errors.HashVerificationFailed error:

alias EventSourcingDB.Event

case Event.verify_hash(event) do
  :ok -> # hash is valid
  {:error, reason} -> # ...
end

The hash tells which version of the hash it is, and an event keeps the version it was written with: a hash that starts with esdb:hash:v2: is of version 2, which EventSourcingDB writes since version 1.3, and a hash without such a prefix is of version 1. The function verifies both. A hash whose prefix names a version that this client SDK does not know, such as esdb:hash:v3:, can not be verified, so the function returns an EventSourcingDB.Errors.UnsupportedHashVersion error, which names the version, rather than report the event as tampered with.

Both versions hash the data of the event exactly as EventSourcingDB sent it, byte for byte, which every event that the client SDK returns keeps for this purpose. For an event that you create yourself, for example with Event.new/1, only the decoded data is known, so the function encodes it again with Jason, which can differ from what EventSourcingDB sent, for example for &, <, and >, which EventSourcingDB escapes.

Events that the client SDK returns carry this text in the field __raw_data__, so comparing such an event with == against one built with Event.new/1 or a struct literal no longer matches, while pattern matching is unaffected.

Note that this only verifies the hash. If you also want to verify the signature, you can skip this step and call verify_signature directly, which performs a hash verification internally.

Verifying an Event's SignatureChanged in 1.3#

To verify the authenticity of an event, call the Event.verify_signature function on the event. This requires the public key that matches the private key used for signing on the server.

The function first verifies the event's hash, and then checks the signature. If any verification step fails, it returns an error, for example an EventSourcingDB.Errors.HashVerificationFailed error if the hash is invalid, an EventSourcingDB.Errors.UnsupportedHashVersion error if the hash is of a version that this client SDK does not know, or an EventSourcingDB.Errors.SignatureVerificationFailed error if the signature is invalid:

alias EventSourcingDB.Event

verification_key = # public key as Ed25519 binary

case Event.verify_signature(event, verification_key) do
  :ok -> # signature is valid
  {:error, reason} -> # ...
end

Using Testcontainers#

Follow the instructions to setup test containers for elixir.

Then you are ready to use the provided TestContainer in your tests:

Changed in 2.1

defmodule YourTest do
  alias EventSourcingDB.TestContainer
  use ExUnit.Case

  import Testcontainers.ExUnit

  container(:esdb, TestContainer.new())

  test "ping", %{esdb: esdb} do
    client = TestContainer.get_client(esdb)

    assert EventSourcingDB.ping(client) == :ok
  end
end

Configuring the Container Instance#

By default, TestContainer uses the latest tag of the official EventSourcingDB Docker image. To change that, call the with_image_tag function:

container(
  :esdb,
  TestContainer.new()
  |> TestContainer.with_image_tag("1.0.0")
)

Similarly, you can configure the port to use and the API token. Call the with_port or the with_api_token function respectively:

container(
  :esdb,
  TestContainer.new()
  |> TestContainer.with_port(4000)
  |> TestContainer.with_api_token("secret")
)

If you want to sign events, call the with_signing_key function. This generates a new signing and verification key pair inside the container:

container(
  :esdb,
  TestContainer.new()
  |> TestContainer.with_signing_key()
)

Changed in 1.1You can retrieve the private key (for signing) and the public key (for verifying signatures) once the container has been started:

signing_key = TestContainer.get_signing_key(esdb)
verification_key = TestContainer.get_verification_key(esdb)

The signing_key is the private key EventSourcingDB signs events with.

The verification_key can be passed to Event.verify_signature when verifying events read from the database.

Configuring the Client Manually#

In case you need to set up the client yourself, use the following functions to get details on the container:

TestContainer.get_base_url(esdb) returns the full URL of the container
TestContainer.get_mapped_port(esdb) returns the mapped port
TestContainer.get_api_token(esdb) returns the API token

Further Resources#

Repository: thenativeweb/eventsourcingdb-client-elixir
Registry: Hex.pm
License (MIT): License file
Issue tracker: GitHub Issues

Elixir SDK

Installation#

Capabilities#

Getting Started#

Using HTTP/2Added in 1.2#

Configuring the Stall TimeoutAdded in 1.3Changed in 2.0#

Writing Events#

Using the IsSubjectPristine precondition#

Using the IsSubjectPopulated preconditionRequires EventSourcingDB 1.2#

Using the IsSubjectOnEventId precondition#

Using the IsEventQLQueryTrue precondition#

Reading Events#

Reading From Subjects Recursively#

Reading in Anti-Chronological Order#

Specifying Bounds#

Starting From the Latest Event of a Given Type#

Running EventQL Queries#

Observing Events#

Observing From Subjects Recursively#

Specifying Bounds#

Starting From the Latest Event of a Given Type#

Registering an Event Schema#

Reading Subjects#

Reading Event Types#

Reading a Specific Event TypeChanged in 2.0#

Verifying an Event's HashChanged in 1.3#

Verifying an Event's SignatureChanged in 1.3#

Using Testcontainers#

Configuring the Container Instance#

Configuring the Client Manually#

Further Resources#

Using the `IsSubjectPristine` precondition#

Using the `IsSubjectPopulated` preconditionRequires EventSourcingDB 1.2#

Using the `IsSubjectOnEventId` precondition#

Using the `IsEventQLQueryTrue` precondition#