kurrentdb_erlang

Erlang runtime backend for the sans-IO KurrentDB package.

This package takes the request builders and response decoders from kurrentdb and runs them on Erlang using gleam_hackney and gleam_otp. A Connection owns a temporary-worker supervisor. Each operation starts a short-lived worker process that builds its request from the operation configuration, sends it to KurrentDB, decodes the response, and sends the result back to the caller.

Unary operations such as appends and deletes return a Task. Use await to receive the result.

let client = kurrentdb.new("localhost", 2113, kurrentdb.TlsDisabled)
let assert Ok(connection) =
  kurrentdb_erlang.start(client, hackney.configure(), option.None)

let event =
  append_to_stream.json_event(
    uuid: uuid.v7(),
    event_type: "OrderPlaced",
    data: json.object([]),
  )

let task =
  kurrentdb_erlang.append_to_stream(
    connection,
    stream: "orders-1",
    events: [event],
    config: append_to_stream.configure(),
  )

kurrentdb_erlang.await(task, within: 5000)

Streaming operations return a Stream. Use receive repeatedly to consume messages and close when you no longer need the stream. Callback-style event subscriptions return a Subscription and call your function for each event until closed.

Types

Connection

opaque

Handle to a running KurrentDB Erlang backend supervisor.

The connection does not represent a single TCP connection. It is an OTP factory supervisor used to start temporary workers for each operation.

pub opaque type Connection

Error

Errors returned by Erlang backend operations.

The type is parameterised by the operation-specific error type. For example Task(append_to_stream.Append, append_to_stream.ResponseError) resolves to Result(append_to_stream.Append, Error(append_to_stream.ResponseError)).

pub type Error(operation_error) {
  TransportError(hackney.Error)
  OperationError(operation_error)
  GrpcError(kurrentdb.GrpcError)
  StreamTimeout
}

Constructors

```
TransportError(hackney.Error)
```
The HTTP transport failed before an operation response could be decoded.
```
OperationError(operation_error)
```
The KurrentDB operation returned an operation-specific failure.
```
GrpcError(kurrentdb.GrpcError)
```
A shared gRPC or protobuf error occurred while decoding a stream.
```
StreamTimeout
```
await timed out before a task result was available.

Stream

opaque

Handle to a streaming read or subscription.

Use receive to consume StreamMessage values and close to stop the worker process. Streams are used for reads and subscriptions where callers want to pull messages themselves.

pub opaque type Stream

StreamMessage

Messages returned by pull-based streaming operations.

pub type StreamMessage {
  ReadMessage(read_stream.ReadMessage)
  ReadEvent(read_stream.ReadEvent)
  StreamFinished
  StreamFailed(Error(read_stream.ResponseError))
}

Constructors

```
ReadMessage(read_stream.ReadMessage)
```
Any read message decoded from the KurrentDB stream.
```
ReadEvent(read_stream.ReadEvent)
```
Convenience message emitted when a ReadMessage contains a read event.

The stream also emits the original ReadMessage(read_stream.ReadEvent(...)), so callers can choose either the full message stream or this event-only shortcut.
```
StreamFinished
```
The server closed the stream and the final gRPC frame decoder state was valid.
```
StreamFailed(Error(read_stream.ResponseError))
```
The stream failed due to transport, gRPC, or operation decoding error.

Subscription

opaque

Handle to a callback-style event subscription.

Use close_subscription to stop the worker process. Subscriptions are used by the *_events functions that call a supplied callback for every decoded read_stream.ReadEvent.

pub opaque type Subscription

Task

opaque

Handle to an asynchronous unary operation.

Use await to receive the operation result. Tasks are returned by operations that have a single final result, such as append, delete, tombstone, and metadata reads.

pub opaque type Task(value, operation_error)

WorkerMessage

opaque

Internal worker control messages.

This type is public because it appears in the typed OTP supervisor name used by start, supervised, and from_name. Application code normally does not construct values of this type directly.

pub opaque type WorkerMessage

WorkerStart

opaque

Internal worker start data.

Each variant contains the full operation information needed to construct and send the request in the worker process. This type is public because named supervisors need its type in their Name, but application code normally uses the high-level functions instead of constructing workers manually.

pub opaque type WorkerStart

Values

append_to_stream

pub fn append_to_stream(
  connection: Connection,
  stream stream: String,
  events events: List(append_to_stream.Event),
  config config: append_to_stream.Configuration,
) -> Task(append_to_stream.Append, append_to_stream.ResponseError)

Append events to a stream.

The operation is started immediately in a temporary worker and the returned Task can be awaited for the append result. The connection’s configured kurrentdb.Client and hackney.Configuration are used by the worker.

await

pub fn await(
  task: Task(value, operation_error),
  within timeout: Int,
) -> Result(value, Error(operation_error))

Await the result of a task.

Returns Error(StreamTimeout) if no result arrives within timeout milliseconds. The worker continues independently if the caller times out.

close

pub fn close(stream: Stream) -> Nil

Close a pull-based stream.

Closing asks the stream worker to stop. It is safe to call after a stream has finished or failed.

close_subscription

pub fn close_subscription(subscription: Subscription) -> Nil

Close a callback-style subscription.

Closing asks the subscription worker to stop and prevents further callback invocations from that worker.

delete_stream

pub fn delete_stream(
  connection: Connection,
  stream stream: String,
  config config: delete_stream.Configuration,
) -> Task(delete_stream.Delete, delete_stream.ResponseError)

Soft-delete a stream.

Soft-deleted streams may be recreated later depending on server behavior. Use tombstone_stream when the stream should be permanently deleted.

from_name

pub fn from_name(
  name: process.Name(
    factory_supervisor.Message(
      WorkerStart,
      process.Subject(WorkerMessage),
    ),
  ),
) -> Connection

Create a Connection handle for a named backend supervisor.

If no supervisor has been started with this name, operations using the returned handle will fail when they try to start a worker.

get_stream_metadata

pub fn get_stream_metadata(
  connection: Connection,
  client: kurrentdb.Client,
  stream stream: String,
  config config: read_stream.Configuration,
) -> Task(
  stream_metadata.StreamMetadata,
  get_stream_metadata.ResponseError,
)

Read stream metadata from the stream’s metadata stream.

This starts a worker that reads from $$<stream>, collects the read messages, and decodes the first metadata event into StreamMetadata.

read_all

pub fn read_all(
  connection: Connection,
  client: kurrentdb.Client,
  config config: read_all.Configuration,
) -> Stream

Start a pull-based read of $all.

Use receive to consume messages. Filters, direction, position, and maximum count are configured with read_all.Configuration.

read_stream

pub fn read_stream(
  connection: Connection,
  client: kurrentdb.Client,
  stream stream_name: String,
  config config: read_stream.Configuration,
) -> Stream

Start a pull-based read of one stream.

Use receive to consume StreamMessage values. A finite read eventually yields StreamFinished when the server closes the response cleanly.

receive

pub fn receive(
  stream: Stream,
  within timeout: Int,
) -> Result(StreamMessage, Nil)

Receive the next message from a stream.

Returns Error(Nil) if no message arrives within timeout milliseconds. Stream failures are delivered as Ok(StreamFailed(error)) because they are messages produced by the stream worker.

set_stream_metadata

pub fn set_stream_metadata(
  connection: Connection,
  stream stream: String,
  metadata metadata: stream_metadata.StreamMetadata,
  uuid uuid: uuid.Uuid,
  config config: append_to_stream.Configuration,
) -> Task(append_to_stream.Append, append_to_stream.ResponseError)

Set stream metadata by appending a $metadata event.

The uuid is used as the event id for the metadata event. Metadata is written to the stream’s metadata stream ($$<stream>) using the supplied append configuration.

start

pub fn start(
  client: kurrentdb.Client,
  http_config: hackney.Configuration,
  name: option.Option(
    process.Name(
      factory_supervisor.Message(
        WorkerStart,
        process.Subject(WorkerMessage),
      ),
    ),
  ),
) -> Result(Connection, actor.StartError)

Start a backend supervisor.

client supplies the KurrentDB endpoint and credentials. http_config is the gleam_hackney configuration used by every worker started under this connection. Pass option.Some(name) to register the supervisor under an OTP name, or option.None to start it anonymously.

Most long-running applications should prefer supervised and add the child specification to their supervision tree.

subscribe_to_all

pub fn subscribe_to_all(
  connection: Connection,
  client: kurrentdb.Client,
  config config: subscribe_to_all.Configuration,
) -> Stream

Subscribe to $all and receive messages by polling.

The returned stream remains active until the server closes it, an error occurs, or close is called.

subscribe_to_all_events

pub fn subscribe_to_all_events(
  connection: Connection,
  client: kurrentdb.Client,
  config config: subscribe_to_all.Configuration,
  on_event on_event: fn(read_stream.ReadEvent) -> Nil,
) -> Subscription

Subscribe to $all and call a function for each event.

Non-event subscription messages, such as confirmations and checkpoints, are ignored by this callback API. Use subscribe_to_all if you need the full message stream.

subscribe_to_stream

pub fn subscribe_to_stream(
  connection: Connection,
  client: kurrentdb.Client,
  stream stream_name: String,
  config config: subscribe_to_stream.Configuration,
) -> Stream

Subscribe to one stream and receive messages by polling.

The returned stream remains active until the server closes it, an error occurs, or close is called.

subscribe_to_stream_events

pub fn subscribe_to_stream_events(
  connection: Connection,
  client: kurrentdb.Client,
  stream stream_name: String,
  config config: subscribe_to_stream.Configuration,
  on_event on_event: fn(read_stream.ReadEvent) -> Nil,
) -> Subscription

Subscribe to one stream and call a function for each event.

Non-event subscription messages, such as confirmations and checkpoints, are ignored by this callback API. Use subscribe_to_stream if you need the full message stream.

supervised

pub fn supervised(
  http_config: hackney.Configuration,
  client: kurrentdb.Client,
  name: process.Name(
    factory_supervisor.Message(
      WorkerStart,
      process.Subject(WorkerMessage),
    ),
  ),
) -> supervision.ChildSpecification(Connection)

Create a child specification for a named backend supervisor.

Use this when integrating the backend into an OTP supervision tree. If the supervisor’s return value is not available where operations are issued, use from_name with the same name to create a Connection handle.

tombstone_stream

pub fn tombstone_stream(
  connection: Connection,
  stream stream: String,
  config config: tombstone_stream.Configuration,
) -> Task(
  tombstone_stream.Tombstone,
  tombstone_stream.ResponseError,
)

Permanently delete a stream.

Tombstoned streams cannot be recreated. The returned task resolves to the tombstone result or an operation/transport error.