Elixir Client for UmaDB

An Elixir gRPC client for UmaDB's DCB (Dynamic Consistency Boundary) event store service. It wraps the generated UmaDb.V1.DCB.Stub with a small, ergonomic API for appending events, reading and subscribing to event streams, querying by type and tags, tracking consumer positions, and optimistic concurrency control — plus UmaDbClient.Builder convenience constructors for the underlying proto message structs.

Installation

Add uma_db_client to your list of dependencies in mix.exs:

def deps do
  [
    {:uma_db_client, "~> 0.7.7"}
  ]
end

Documentation is published on HexDocs at https://hexdocs.pm/uma_db_client.

Application setup

The underlying gRPC client requires GRPC.Client.Supervisor to be running. Add it to your application's supervision tree — connect/2 raises without it:

children = [
  {GRPC.Client.Supervisor, []}
]

Supervisor.start_link(children, strategy: :one_for_one, name: MyApp.Supervisor)

In scripts, tests, or iex, you can start it manually instead:

{:ok, _pid} = DynamicSupervisor.start_link(strategy: :one_for_one, name: GRPC.Client.Supervisor)

Connecting to UmaDB

UmaDbClient.connect/2 takes a "host:port" target and returns a gRPC channel. Options are passed straight through to GRPC.Stub.connect/2.

{:ok, channel} = UmaDbClient.connect("localhost:50051")

For TLS, build a GRPC.Credential and pass it as :cred:

cred = GRPC.Credential.new(ssl: [cacertfile: "/path/to/ca.pem"])
{:ok, channel} = UmaDbClient.connect("myserver:443", cred: cred)

Note: the server certificate really is verified against the given CA, but a verification failure surfaces from connect/2 as {:error, :timeout} rather than a TLS-specific error. If a TLS connection times out, check that the cacertfile matches the CA that signed the server's certificate.

Close the channel when you're done:

GRPC.Stub.disconnect(channel)

Appending Events

UmaDbClient.append/3 writes one or more events atomically and returns the position of the last appended event. Use UmaDbClient.Builder.event/1 to construct events — data is raw bytes, so serialize however you like. Only :type is required.

alias UmaDbClient.Builder

event =
  Builder.event(
    type: "OrderPlaced",
    tags: ["order:1"],
    data: Jason.encode!(%{id: 1, total: 99})
  )

{:ok, position} = UmaDbClient.append(channel, [event])

All events in a single call are committed together:

events = [
  Builder.event(type: "OrderPlaced", tags: ["order:1"], data: Jason.encode!(%{id: 1})),
  Builder.event(type: "PaymentReceived", tags: ["order:1"], data: Jason.encode!(%{amount: 99}))
]

{:ok, position} = UmaDbClient.append(channel, events)

Attach metadata as a map or keyword list of string pairs:

event =
  Builder.event(
    type: "OrderPlaced",
    tags: ["order:1"],
    data: Jason.encode!(%{id: 1}),
    metadata: %{"correlation_id" => "abc-123", "user" => "alice"}
  )

Configuring encode and decode

Encoding and decoding payloads at every call site gets repetitive. use UmaDbClient to configure both once and get a client facade that applies them:

defmodule MyApp.DB do
  use UmaDbClient, encode: &Jason.encode!/1, decode: &Jason.decode!/1
end

:data is then a plain term on the way in, and already decoded on the way out:

{:ok, channel} = MyApp.DB.connect("localhost:50051")

event = MyApp.DB.event(type: "OrderPlaced", tags: ["order:1"], data: %{id: 1})
{:ok, position} = MyApp.DB.append(channel, [event])

{:ok, stream} = MyApp.DB.read(channel)

Enum.each(stream, fn %UmaDbClient.Event{position: position, type: type, data: data} ->
  IO.inspect({position, type, data})
end)

Both options are optional and independent — encode only, decode only, or neither all work. Each accepts a one-argument function, a {module, function} tuple, or a module, exporting encode!/1 and decode!/1 respectively, so use UmaDbClient, encode: Jason, decode: Jason works too.

For values that are already encoded, :raw_data bypasses the encoder:

event = MyApp.DB.event(type: "Snapshot", raw_data: <<1, 2, 3>>)

Passing both :data and :raw_data raises ArgumentError.

The facade also delegates append/3, head/1, get_tracking_info/2, connect/2 and the builders (query/1, query_item/2, append_condition/1, tracking_info/2), so it can replace UmaDbClient and UmaDbClient.Builder entirely:

query = MyApp.DB.query([MyApp.DB.query_item(["OrderPlaced"], ["order:1"])])
{:ok, stream} = MyApp.DB.read(channel, query: query)

UmaDbClient.Event

Facade reads and subscriptions yield UmaDbClient.Event structs rather than the generated proto structs:

An event recorded without a payload comes back as data: nil. The decoder is never handed an empty binary, since decoders like Jason.decode!/1 raise on one.

UmaDbClient.read/2 and UmaDbClient.subscribe/2 are unchanged and still return UmaDb.V1.SequencedEvent structs. The single-use caveat above applies to facade streams too.

Optimistic concurrency

An AppendCondition makes the append fail if events matching a query already exist — the core of DCB's consistency model. Build one with Builder.append_condition/1:

# Fail if this order was already placed.
query = Builder.query([Builder.query_item(["OrderPlaced"], ["order:1"])])
condition = Builder.append_condition(fail_if_events_match: query)

case UmaDbClient.append(channel, [event], condition: condition) do
  {:ok, position} ->
    {:ok, position}

  {:error, reason} ->
    # The condition matched — another writer got there first.
    {:error, {:conflict, reason}}
end

Use :after to only consider events recorded after a position you've already seen — the usual read-decide-write cycle:

{:ok, head} = UmaDbClient.head(channel)

condition =
  Builder.append_condition(
    fail_if_events_match: query,
    after: head
  )

{:ok, position} = UmaDbClient.append(channel, [event], condition: condition)

Idempotent retries

The server does not enforce uniqueness of event uuids — appending the same uuid twice without a condition simply stores two events. Idempotency comes from combining uuids with an append condition: when a conditional append would fail, the server first checks whether the events that trip the condition are the same events being submitted, comparing their uuids. If they match, it treats the call as a retry and returns the original commit position instead of a conflict.

Two requirements follow from how this is implemented:

• Every event in the append must carry a uuid. If any one of them is missing, the check is skipped entirely and the retry fails as a conflict.
• The uuids are compared in order. The server takes the first N events matching the condition — where N is the number of events you submitted — and compares their uuids positionally against yours; all N must match. Submitting the same events in a different order is therefore treated as a conflict.

This makes a retry after an ambiguous failure (a timeout, say, where you don't know whether the first attempt landed) safe to issue blindly.

Set the :uuid option to enable it. It must be a valid UUID string — the server rejects anything else with deserialization error: Invalid UUID in Event, so an application-specific key like "order-1" will not work:

event =
  Builder.event(
    type: "OrderPlaced",
    tags: ["order:1"],
    data: Jason.encode!(%{id: 1}),
    uuid: "550e8400-e29b-41d4-a716-446655440000"
  )

query = Builder.query([Builder.query_item(["OrderPlaced"], ["order:1"])])
condition = Builder.append_condition(fail_if_events_match: query)

# Both calls return {:ok, same_position}; only one event is stored.
{:ok, position} = UmaDbClient.append(channel, [event], condition: condition)
{:ok, ^position} = UmaDbClient.append(channel, [event], condition: condition)

Retrying with the same condition but a differentuuid is treated as a genuine conflict and fails with an integrity error.

Reading Events

UmaDbClient.read/2 returns a lazy Enumerable of UmaDb.V1.SequencedEvent structs. Events are fetched from the server as you iterate.

{:ok, stream} = UmaDbClient.read(channel)

Enum.each(stream, fn %UmaDb.V1.SequencedEvent{position: position, event: event} ->
  IO.puts("#{position}: #{event.event_type}")
end)

The stream is single-use. It is backed by a live gRPC server stream, so it can only be enumerated once — a second pass (for example Enum.count/1 followed by Enum.to_list/1) blocks forever waiting for data that will never arrive. Enumerate once and keep the result:

{:ok, stream} = UmaDbClient.read(channel)
events = Enum.to_list(stream)
count = length(events)

Filter with a query. Within a QueryItem, types match as OR and tags match as AND; multiple items in a Query are OR'd together:

query =
  Builder.query([
    Builder.query_item(["OrderPlaced", "OrderCancelled"], ["order:1"])
  ])

{:ok, stream} = UmaDbClient.read(channel, query: query)
events = Enum.to_list(stream)

Additional options — :start (inclusive), :backwards, :limit, and :batch_size (events per server response):

# The 10 most recent events.
{:ok, stream} = UmaDbClient.read(channel, backwards: true, limit: 10)

# Everything from position 100 onward, in batches of 500.
{:ok, stream} = UmaDbClient.read(channel, start: 100, batch_size: 500)

Subscribing

UmaDbClient.subscribe/2 returns a lazy Enumerable that first catches up on recorded events and then continues yielding new ones as they arrive.

{:ok, stream} = UmaDbClient.subscribe(channel, query: query, after: last_position)

Enum.each(stream, fn %UmaDb.V1.SequencedEvent{position: position, event: event} ->
  handle_event(position, event)
end)

This blocks indefinitely — the stream only ends when the server closes it or an error occurs. Run it in a dedicated process:

Task.start_link(fn ->
  {:ok, stream} = UmaDbClient.subscribe(channel, after: last_position)
  Enum.each(stream, &handle_event/1)
end)

Getting the Head Position

UmaDbClient.head/1 returns the position of the last recorded event, or nil when the log is empty.

{:ok, position} = UmaDbClient.head(channel)

Tracking Consumer Positions

UmaDB can store a cursor for a named source, so a consumer can resume where it left off. Read it with UmaDbClient.get_tracking_info/2:

{:ok, position} = UmaDbClient.get_tracking_info(channel, "projection:orders")

Advance the cursor atomically as part of an append by passing :tracking_info — this is what makes exactly-once processing possible, since the events and the cursor commit together:

tracking = Builder.tracking_info("projection:orders", source_position)

{:ok, position} =
  UmaDbClient.append(channel, [event], tracking_info: tracking)

Since UmaDB 0.7.0 the cursor recorded with an append is also returned when the event is read back, so a consumer can see which upstream position an event was committed with:

{:ok, stream} = UmaDbClient.read(channel)

Enum.each(stream, fn %UmaDb.V1.SequencedEvent{tracking_info: tracking} ->
  IO.inspect(tracking)  # %UmaDb.V1.TrackingInfo{} or nil
end)

Through the facade it arrives as a plain map:

{:ok, stream} = MyApp.DB.read(channel)

Enum.each(stream, fn %UmaDbClient.Event{tracking_info: tracking} ->
  IO.inspect(tracking)  # %{source: "projection:orders", position: 41} or nil
end)

Data Types

All types are the generated UmaDb.V1.* structs. UmaDbClient.Builder provides constructors, but you can always build the structs directly.

Error Handling

All API functions return {:ok, result} or {:error, reason}, where reason is typically a GRPC.RPCError:

case UmaDbClient.append(channel, [event], condition: condition) do
  {:ok, position} -> {:ok, position}
  {:error, %GRPC.RPCError{status: status, message: message}} -> {:error, {status, message}}
end

read/2 and subscribe/2 are the exception: they return {:ok, stream} immediately, but an error that occurs mid-stream is raised as a GRPC.RPCError while iterating. Wrap iteration if you need to recover:

try do
  Enum.each(stream, &handle_event/1)
rescue
  e in GRPC.RPCError -> Logger.error("stream failed: #{e.message}")
end

Server-side failures are categorized by UmaDb.V1.ErrorResponse.ErrorType — IO, SERIALIZATION, INTEGRITY, CORRUPTION, INTERNAL, AUTHENTICATION, and INVALID_ARGUMENT. A failed append condition arrives as a GRPC.RPCError with status 9 (failed precondition) and a message beginning integrity error: condition failed, naming the event that matched.

Complete Example

Read current state, decide, then append conditionally so a concurrent writer can't slip in between:

alias UmaDbClient.Builder

{:ok, channel} = UmaDbClient.connect("localhost:50051")

# 1. Capture the current head to scope the condition.
{:ok, head} = UmaDbClient.head(channel)

# 2. Read what already happened for this order.
query = Builder.query([Builder.query_item([], ["order:1"])])
{:ok, stream} = UmaDbClient.read(channel, query: query)
history = Enum.to_list(stream)

# 3. Decide, based on that history.
if Enum.any?(history, &(&1.event.event_type == "OrderPlaced")) do
  {:error, :already_placed}
else
  # 4. Append, failing if anything matching arrived since we read.
  condition =
    Builder.append_condition(
      fail_if_events_match: query,
      after: head
    )

  event =
    Builder.event(type: "OrderPlaced", tags: ["order:1"], data: Jason.encode!(%{id: 1}))

  case UmaDbClient.append(channel, [event], condition: condition) do
    {:ok, position} -> {:ok, position}
    {:error, reason} -> {:error, {:conflict, reason}}
  end
end

A consumer that processes events and checkpoints its position in the same transaction:

{:ok, last} = UmaDbClient.get_tracking_info(channel, "projection:orders")

{:ok, stream} = UmaDbClient.subscribe(channel, after: last)

Enum.each(stream, fn %UmaDb.V1.SequencedEvent{position: position, event: event} ->
  derived = handle_event(event)

  # Commit the derived event and the cursor together.
  UmaDbClient.append(channel, [derived],
    tracking_info: Builder.tracking_info("projection:orders", position)
  )
end)

Notes and Limitations

• Calls are blocking. Each function call blocks the calling process until the server responds. subscribe/2 blocks for as long as the stream is open.
• data is raw bytes on the wire. The protocol carries an opaque byte string, so UmaDbClient.Builder.event/1 and UmaDbClient.read/2 leave the choice of format to you — pick one (JSON, protobuf, :erlang.term_to_binary/1) and use it consistently across writers and readers. To stop repeating it at every call site, configure it once with use UmaDbClient, which encodes on append and decodes on read.
• Read and subscribe streams are single-use. They are backed by a live gRPC stream, so enumerating one twice blocks forever. Enumerate once and keep the result.
• Event uuids must be valid UUIDs, and are not enforced to be unique. A non-UUID string is rejected with deserialization error: Invalid UUID in Event, so an application key like "order-1" will not work. The server will store the same valid uuid twice quite happily; it only has an effect when paired with an append condition.
• API-key authentication is not supported. The client does not thread per-call metadata through to the underlying stub. TLS via :cred is supported.
• read/2 does not surface the server's head position.ReadResponse.head is dropped when the batched responses are flattened into a single event stream; use head/1 instead.