# `Slap.Streams`
[🔗](https://github.com/mjrusso/slap/blob/slap_streams-v0.3.0/slap_streams/lib/slap/streams.ex#L1)

Durable Streams operations, in process (the HTTP layer and Yjs build on
these). Each stream lives on the shard `Slap.Streams.Cluster.shard_for(path)` and is
served by one process there. Writes are acknowledged only once
durable.

Offsets are integers here (`Slap.Streams.Offset` converts to and from the wire
format), and results are protocol outcomes, not HTTP responses.
Pass `cluster:` to use a module defined with `use Slap.Streams.Cluster,
otp_app: :my_app`. Run one Streams cluster per VM.

`PROTOCOL.md` sections below refer to the
[Durable Streams specification](https://github.com/durable-streams/durable-streams/blob/main/PROTOCOL.md).

## Errors shared by all calls

Invalid stream names, offsets, bodies and request fields return
`{:error, {:bad_request, reason}}`. Invalid control options (`:timeout`,
`:cluster`, `:placement_key`, read limits and waits) and unknown option
names raise `ArgumentError`. The `placement_key/1` path helper also raises
for a non-binary argument.

  * `{:error, :not_found}` - no such stream (404).
  * `{:error, :gone}` - soft-deleted (410).
  * `{:error, :unavailable}` - the shard is stopping or moved, or a write
    failed; retry (503).
  * `{:error, :timeout}` - no reply within `:timeout` (default 30 s). The
    request may still complete.
  * `{:error, {:bad_request, reason}}` - 400.

`reason` is one of:

  * `:invalid_path`, `:invalid_offset`, `:invalid_body`,
    `:invalid_content_type`, `:invalid_ttl`, `:invalid_expires_at`,
    `:invalid_closed`, `:invalid_close`, `:invalid_stream_seq`,
    `:invalid_producer`;
  * `:invalid_fork_source`, `:invalid_fork_offset`,
    `:invalid_fork_sub_offset`, `:fork_of_itself`,
    `:fork_offset_beyond_source`, `:fork_offset_trimmed`;
  * `:empty_body`, `:empty_array`, `:invalid_json`,
    `:ttl_and_expires_at`, `:new_epoch_must_start_at_zero`,
    `:trim_offset_beyond_tail`.

# `append_result`

```elixir
@type append_result() :: %{
  result: :appended | :duplicate | :closed,
  next_offset: non_neg_integer(),
  closed: boolean(),
  producer: {non_neg_integer(), non_neg_integer()} | nil
}
```

# `bad_request_reason`

```elixir
@type bad_request_reason() ::
  :empty_array
  | :empty_body
  | :fork_of_itself
  | :fork_offset_beyond_source
  | :fork_offset_trimmed
  | :invalid_body
  | :invalid_close
  | :invalid_closed
  | :invalid_content_type
  | :invalid_expires_at
  | :invalid_fork_offset
  | :invalid_fork_source
  | :invalid_fork_sub_offset
  | :invalid_json
  | :invalid_offset
  | :invalid_path
  | :invalid_producer
  | :invalid_stream_seq
  | :invalid_ttl
  | :new_epoch_must_start_at_zero
  | :trim_offset_beyond_tail
  | :ttl_and_expires_at
```

# `error`

```elixir
@type error() ::
  :bad_offset
  | :conflict
  | :content_type_mismatch
  | :deleted
  | :gone
  | :not_found
  | :overloaded
  | :offset_beyond_tail
  | :payload_too_large
  | :sealed
  | :source_gone
  | :source_not_found
  | :stream_seq_conflict
  | :timeout
  | :trimmed
  | :unavailable
  | {:bad_request, bad_request_reason()}
  | {:closed, non_neg_integer()}
  | {:producer_seq_gap, non_neg_integer(), non_neg_integer()}
  | {:stale_epoch, non_neg_integer()}
```

# `info`

```elixir
@type info() :: %{
  next_offset: non_neg_integer(),
  closed: boolean(),
  content_type: binary(),
  ttl_s: non_neg_integer() | nil,
  expires_at_ms: integer() | nil
}
```

# `path`

```elixir
@type path() :: binary()
```

# `read_result`

```elixir
@type read_result() :: %{
  messages: [{non_neg_integer(), binary()}],
  next_offset: non_neg_integer(),
  up_to_date: boolean(),
  closed: boolean(),
  content_type: binary()
}
```

# `append`

```elixir
@spec append(path(), binary(), keyword()) ::
  {:ok, append_result()} | {:error, error()}
```

Appends `body` (PROTOCOL.md §5.2). Options:

  * `:content_type` - checked against the stream's, if given.
  * `:close` - close the stream with this append (an empty body closes
    without appending, §5.3).
  * `:stream_seq` - the `Stream-Seq` header.
  * `:producer` - `{producer_id, epoch, seq}` (§5.2.1).

Returns `{:ok, %{result: r, next_offset: o, closed: c, producer: p}}` where
`r` is `:appended` (200 with a producer, else 204), `:duplicate` (204) or
`:closed` (a close-only request, 204), and `p` is `{epoch, seq}` to echo.

Errors besides the shared ones: `{:closed, next_offset}` (409 with
`Stream-Closed`), `:content_type_mismatch` (409), `:stream_seq_conflict`
(409), `{:stale_epoch, epoch}` (403), `{:producer_seq_gap, expected,
received}` (409).

# `await_wake`

```elixir
@spec await_wake(path(), Slap.Streams.Wait.t(), timeout()) ::
  :data | :closed | :deleted | :unavailable | :timeout
```

Waits up to `timeout` ms for the wake of a wait from `wait/4`: `:data`,
`:closed`, `:deleted`, `:unavailable` (the stream's server or its node
went down) or `:timeout` (the wait is then cancelled).

# `cancel_wait`

```elixir
@spec cancel_wait(path(), Slap.Streams.Wait.t(), keyword()) :: :ok | {:error, error()}
```

Cancels a wait from `wait/4`.

# `close`

```elixir
@spec close(path(), keyword()) :: {:ok, append_result()} | {:error, error()}
```

Closes the stream without appending: `append(path, "", close: true)`.

# `create`

```elixir
@spec create(path(), keyword()) ::
  {:ok, :created | :exists, info()} | {:error, error()}
```

Creates a stream (PROTOCOL.md §5.1). Options: `:content_type`, `:ttl_s`,
`:expires_at_ms`, `:closed`, `:body` (initial content).

A fork (§4.2) also takes `:forked_from` (the source's path), and optionally
`:fork_offset` (default: the source's tail) and `:fork_sub_offset`. The
source's data before the fork offset is copied into the fork.

Returns `{:ok, :created, info}` (201), `{:ok, :exists, info}` when a stream
with the same configuration exists (200), or `{:error, :conflict}` when one
with a different configuration does, or the path is soft-deleted (409).

Fork errors: `:source_not_found` (404), `:source_gone` (409),
`:content_type_mismatch` (409), `:payload_too_large` (the copy would
exceed `:max_fork_copy_bytes`, 413). A fork whose copy was interrupted is
`:unavailable` (503) until the create is retried, which finishes it.

# `delete`

```elixir
@spec delete(path(), keyword()) :: :ok | {:error, error()}
```

Deletes the stream (PROTOCOL.md §5.4). Returns `:ok` once durable.

# `head`

```elixir
@spec head(path(), keyword()) :: {:ok, info()} | {:error, error()}
```

The stream's durable metadata and tail (PROTOCOL.md §5.5).

# `list`

```elixir
@spec list(path(), keyword()) :: {:ok, [path()]} | {:error, error()}
```

The paths, in order, of the streams in `prefix`'s placement group
(`placement_key/1`) whose paths start with `prefix`: every stream whose
creation is durable and that is not deleted. A stream past its expiry is
listed until it is removed. Not part of the Durable Streams protocol.

# `placement_key`

```elixir
@spec placement_key(path()) :: String.t()
```

The key that places `path` on a shard: the path up to its first segment
that starts with `.`, or the whole path. Streams that differ only from
such a segment are one group on one shard, whether they are reached here
or over HTTP: a Yjs document's `.updates`, `.index` and
`.snapshots/<offset>_snapshot` (`slap_yjs`), for example. The
`:placement_key` option of the calls here overrides it.

# `read`

```elixir
@spec read(path(), non_neg_integer() | :start | :now, keyword()) ::
  {:ok, read_result()} | {:error, error()}
```

Reads from `offset` (an integer, `:start` for the first retained message,
or `:now`) up to the durable tail (PROTOCOL.md §5.6), at most about
`:max_bytes` (default 1 MiB; at least one message). The data is read in
the calling process.

Returns `{:ok, %{messages: [{offset, bytes}], next_offset, up_to_date,
closed, content_type}}`. For a JSON stream each message is one JSON value;
the HTTP layer joins them into a response array. `closed` is true only when the
read reached the end of a closed stream. An offset past the durable tail is
`{:error, :offset_beyond_tail}`.

With `wait: ms`, a read that finds no messages at the tail of an open
stream waits up to `ms` for data (long-poll), then reads again from the
tail; that read's result is returned, empty if nothing came. It reads
again after a timeout too, so that an empty result's `next_offset` and
`closed` come from one view of the stream. A stream deleted during the
wait is `{:error, :deleted}`; one whose server or node went down without
saying so is `{:error, :unavailable}`.

An explicit `:timeout` bounds the read, wait, and final read together. A
long-poll reserves time for the final read. If that read takes longer than
the remaining budget, the call returns `{:error, :timeout}` even when data
arrived; the caller can read again. Without `:timeout`, each request uses
its 30-second default and `:wait` bounds the wait.

# `seal`

```elixir
@spec seal(binary(), keyword()) :: :ok | {:error, error()}
```

Seals the placement group `group` for good: from when this returns, a
create of a stream in the group (`placement_key/1` of its path is
`group`), or a fork into it, fails with `{:error, :sealed}` (409), and
every stream created in it before is durable, so `list/2` returns it.
The group's existing streams are not changed. Not part of the Durable
Streams protocol.

# `trim`

```elixir
@spec trim(path(), non_neg_integer(), keyword()) :: :ok | {:error, error()}
```

Trims the stream before `offset`: reads from earlier offsets get
`{:error, :trimmed}` (410), and a background job deletes the data. An
offset past the tail is a bad request.

# `wait`

```elixir
@spec wait(path(), non_neg_integer() | :now) ::
  {:ok, :data | :closed | {:waiting, Slap.Streams.Wait.t()}} | {:error, error()}
```

Waits for data after `offset` (an integer or `:now`), for long-poll and
SSE. Returns `{:ok, :data}` or `{:ok, :closed}` at once if there is
something to report, else `{:ok, {:waiting, wait}}`: `pid` then gets
`{:slap_streams_wake, ref, reason}` once, where `ref` is
`Slap.Streams.Wait.ref(wait)` and `reason` is `:data`, `:closed`,
`:deleted` or `:unavailable`. Cancel with `cancel_wait/2`.

The third argument can be an options list; the waiting pid then defaults
to the caller.

When `pid` is the caller, the stream's server (perhaps on another node)
is monitored too, so that `await_wake/3` reports `:unavailable` if it or
its node goes down without saying so.

# `wait`

```elixir
@spec wait(path(), non_neg_integer() | :now, pid() | keyword()) ::
  {:ok, :data | :closed | {:waiting, Slap.Streams.Wait.t()}} | {:error, error()}
```

# `wait`

```elixir
@spec wait(path(), non_neg_integer() | :now, pid(), keyword()) ::
  {:ok, :data | :closed | {:waiting, Slap.Streams.Wait.t()}} | {:error, error()}
```

---

*Consult [api-reference.md](api-reference.md) for complete listing*