uma_db_client

# Elixir Client for UmaDB

An Elixir gRPC client for [UmaDB](https://umadb.io)'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`:

```elixir
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:

```elixir
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:

```elixir
{: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`.

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

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

```elixir
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:

```elixir
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.

```elixir
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:

```elixir
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:

```elixir
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:

```elixir
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:

```elixir
{: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:

```elixir
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:

```elixir
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:

| Field | Notes |
| --- | --- |
| `position` | the event's global position |
| `type` | the event type |
| `tags` | list of tag strings |
| `data` | decoded payload, or raw bytes when no `:decode` is set |
| `uuid` | `nil` when the event carries no identifier |
| `tracking_info` | `%{source: ..., position: ...}`, or `nil` if none was recorded |
| `metadata` | a plain map, not `MetadataEntry` 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`:

```elixir
# 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:

```elixir
{: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 `uuid`s — appending the same
`uuid` twice without a condition simply stores two events. Idempotency comes from
combining `uuid`s 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 `uuid`s. 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 `uuid`s 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 `uuid`s 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:

```elixir
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 *different* `uuid` 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.

```elixir
{: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:
>
> ```elixir
> {: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:

```elixir
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):

```elixir
# 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.

```elixir
{: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:

```elixir
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.

```elixir
{: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`:

```elixir
{: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:

```elixir
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:

```elixir
{: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:

```elixir
{: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.

| Type | Fields | Builder |
| --- | --- | --- |
| `UmaDb.V1.Event` | `event_type`, `tags`, `data` (bytes), `uuid` (valid UUID or `""`), `metadata` | `Builder.event/1` |
| `UmaDb.V1.SequencedEvent` | `position`, `event`, `tracking_info` | — (returned by reads) |
| `UmaDbClient.Event` | `position`, `type`, `tags`, `data` (decoded), `uuid`, `tracking_info`, `metadata` (map) | — (returned by facade reads) |
| `UmaDb.V1.Query` | `items` — empty matches all events | `Builder.query/1` |
| `UmaDb.V1.QueryItem` | `types` (OR), `tags` (AND) | `Builder.query_item/2` |
| `UmaDb.V1.AppendCondition` | `fail_if_events_match`, `after` | `Builder.append_condition/1` |
| `UmaDb.V1.TrackingInfo` | `source`, `position` | `Builder.tracking_info/2` |

## Error Handling

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

```elixir
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:

```elixir
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:

```elixir
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:

```elixir
{: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`](#configuring-encode-and-decode), 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 `uuid`s 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.