@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).

@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).

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

Cancels a wait from wait/4.

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

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

@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.

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

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

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

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

@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.

@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.

@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.

@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.

@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.

@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.