logdb_client

# logdb_client

Elixir client for the LogDB event bus: publish events, consume them with a durable
consumer and confirm explicitly (`ack`, `nack`, `term`, `progress`, or `defer` and confirm
later). Transports: WebSocket, HTTP, Erlang distribution, Hybrid. Reconnects on its own.

## Installation

```elixir
def deps do
  [
    {:logdb_client, "~> 0.3"}
  ]
end
```

## A consumer

```elixir
defmodule MyApp.Jobs do
  use LogDB.Client.Consumer

  @impl true
  def handle_event(_type, payload, _meta, state) do
    case MyApp.run(payload) do
      :ok -> {:ack, state}
      {:error, :busy} -> {:nack, state, delay_ms: 5_000}
      {:error, :bad_input} -> {:term, state, "bad input"}
    end
  end
end

# in your supervision tree; LOGDB_URL=logdb://<token>@bus.local/default/jobs?transport=ws
{MyApp.Jobs, name: :jobs, connection: System.fetch_env!("LOGDB_URL"), parser: LogDB.Client.Parser.Json}
```

A client that only publishes: `{LogDB.Client.Supervisor, name: :bus, connection: "logdb://<token>@bus.local"}`.
Several clients can sit in one supervisor: the child id is the client's name.

## Connection string

    logdb[s]://[token@]host[:port][/stream/consumer][?transport=…&node=…&client=…]

- `logdbs` — TLS (wss/https). Port: 4000, or 443 with TLS.
- `token` — the service token; or pass it as `token:` (if both are given they must match).
- `/stream/consumer` — the consumer to read, as `consumer: "stream/name"`.
- `transport` — `ws`, `http`, `hybrid` or `erlangd` (default: `http`).
- `node` — the bus node for erlangd: `logdb` (→ `logdb@host`) or a full `logdb@host`;
  `node_name` is the old name of the same parameter.
- `client` — this instance's name (`client_name:`).

An unknown parameter is an error. The same format is used by every LogDB client
(`docs/connection-string.md`). The token is hidden whenever the
config or the string is printed.

Instead of a string: `host:`, `port:`, `secure:`. Anything set both in the string and as an
option must agree.

## Options

`:name` (required), `:connection` or `:host`/`:port`/`:secure`, `:token`, `:transport`,
`:parser`, `:consumer` (or the old `:stream` + `:consumer_id`), `:node`, `:client_name`,
`:req_options` (Req, for HTTP), `:init_arg` (your data for the consumer's `init/1`).

Options are checked at start: a missing token or host, a wrong scheme, an unknown option
(with a hint: `:consumer_ld (did you mean :consumer_id?)`) — the client refuses to start and
says why. The client connects asynchronously; `LogDB.await_connected(name, timeout)` waits
for it when you need to.

## Calls and replies

    LogDB.publish(:bus, "rkp://jobs/run", %{"id" => 1}, stream: "default")
    LogDB.publish_batch(:bus, [%{"type" => "…", "payload" => …, "stream" => "default"}])
    LogDB.ack(:jobs, ids)  # also nack/3, term/3, progress/2
    LogDB.ping(:bus)

Every transport replies the same way: `{:ok, result}` or `{:error, %LogDB.Error{}}`
(`subscribe` and `ping`: `:ok` or `{:error, %LogDB.Error{}}`). `LogDB.Error` has a `code`
(`:disconnected`, `:timeout`, `:not_found`, `:invalid`, `:unauthorized`, …), a `message`
and `details`, and can be raised. A client that is not running or lost its connection gives
`{:error, %LogDB.Error{code: :disconnected}}`; the caller never crashes.

If `handle_event/4` raises, throws, exits or returns something else, the event is not
confirmed and `handle_error/2` gets the reason; the consumer keeps its state.

## Choosing the consumer

    consumer: "default/rkp_core"          # stream/name; the service is known from the token (LogDB with /v1)
    stream: "default", consumer_id: "…"   # the old pair, works with any LogDB

Without either the client only publishes. `transport: LogDB.Client.Transport.HTTP`
with a pull consumer receives events by long polling `/v1/events/poll`.

## Transports

- `LogDB.Client.Transport.WebSocket` — delivery and every command (publish, subscribe,
  ack/nack/term/progress, ping) over one socket; each command waits for the bus's reply.
- `LogDB.Client.Transport.Erlangd` — Erlang distribution; payload encoded by the parser.
- `LogDB.Client.Transport.HTTP` — publish and confirm; with a pull consumer, delivery by long polling.
- `LogDB.Client.Transport.Hybrid` — delivery over WS; writes over the same socket when LogDB
  speaks client API 1, over HTTP otherwise (kept for existing setups).

`LogDB.subscribe(name, "stream/name")` subscribes a running client to a consumer (LogDB with /v1).

## Client name

    client_name: "rkp_beam@node-2"   # default: the node name, or the host name

The client introduces itself to LogDB on every connection (`hello` over WS and erlangd,
`X-LogDB-Client` and the standard `User-Agent` header over HTTP), so the LogDB console shows which
instance of a service is connected.

## Tests

    mix test

Unit tests need nothing running. The live erlangd test (`test/live_erlangd_test.exs`)
needs a bus node started with a name and the same cookie:

    LOGDB_LIVE_CONNECTION="logdb://<host>:4000?node=<bus node name>" \
    LOGDB_LIVE_TOKEN=<service token> LOGDB_LIVE_CONSUMER_ID=<push consumer on <protocol>://**> \
    elixir --sname live_test -S mix test --only live