npm package: @loro-dev/streams-client
Description: Stream-scoped Durable Streams client for TypeScript.
Author: Loro Team
Latest version: 0.3.1

# `@loro-dev/streams-client`

TypeScript client for a single Durable Stream URL.

Use this package when you already know the stream URL and want to:

- create or delete the stream
- append bytes or close the stream
- read from an offset
- tail new data with SSE or long-poll
- bootstrap from a snapshot, then continue from the returned offset

This package is protocol-level on purpose. It does not encode CRDT updates for
you or manage CRDT state in memory.

## Install

```bash
pnpm add @loro-dev/streams-client
```

Runtime requirements:

- modern browsers and worker runtimes are supported
- Node.js `>=18` is supported
- if your runtime does not expose standard Web APIs like `fetch`,
  `Headers`, `ReadableStream`, and `DOMException`, you need to provide
  compatible polyfills before using this package

## Start Here

The main entry is:

```ts
import { StreamsClient } from "@loro-dev/streams-client";
```

Create one client per stream URL:

```ts
const client = new StreamsClient({
  url: "https://streams-api.loro.dev/ds/my-bucket/my-stream",
  auth: async () => accessToken,
});
```

Notes:

- `url` is the full stream URL, not just `baseUrl` or `bucketId`.
- `auth` should return the raw token string. The client adds `Bearer ` for you.
- Import entry: `@loro-dev/streams-client`
- Repo source entry: `src/index.ts`
- Main client implementation in this repo: `src/client.ts`

## 30-Second Example

```ts
import { StreamsClient } from "@loro-dev/streams-client";

const client = new StreamsClient({
  url: "https://streams-api.loro.dev/ds/my-bucket/my-stream",
});

const created = await client.create({
  contentType: "application/json",
});
if (!created.ok) throw created.result;

const appended = await client.append({
  part: {
    contentType: "application/json",
    body: JSON.stringify({ type: "hello", value: 1 }),
  },
});
if (!appended.ok) throw appended.result;

const read = await client.read({ offset: "-1" });
if (!read.ok) throw read.result;
console.log(read.result.payload.json());

for await (const event of client.live({ offset: appended.result.nextOffset })) {
  if (event.type === "data") {
    console.log(event.payload.json());
  }
  if (event.type === "eof") {
    break;
  }
}
```

## What You Usually Need

| Task                             | API                                      | What it returns                                                 | Spec                                                                                                                 |
| -------------------------------- | ---------------------------------------- | --------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------- |
| Create or ensure a stream exists | `client.create()`                        | `created`, `contentType`, `nextOffset`, `closed`                | [`docs/ds-protocol.md` -> 5.1 Create Stream](./docs/ds-protocol.md#51-create-stream)                                 |
| Append data                      | `client.append()`                        | `nextOffset`, `closed`                                          | [`docs/ds-protocol.md` -> 5.2 Append to Stream](./docs/ds-protocol.md#52-append-to-stream)                           |
| Close the stream                 | `client.close()`                         | same shape as append                                            | [`docs/ds-protocol.md` -> 5.3 Close Stream](./docs/ds-protocol.md#53-close-stream)                                   |
| Read current stream metadata     | `client.head()`                          | `contentType`, `nextOffset`, `closed`, optional snapshot info   | [`docs/ds-protocol.md` -> 5.5 Stream Metadata](./docs/ds-protocol.md#55-stream-metadata)                             |
| Read once from an offset         | `client.read()`                          | payload plus `nextOffset`                                       | [`docs/ds-protocol.md` -> 5.6 Read Stream - Catch-up](./docs/ds-protocol.md#56-read-stream---catch-up)               |
| Do one long-poll round trip      | `client.readOnce({ live: "long-poll" })` | `200` with payload or `204` with no payload                     | [`docs/ds-protocol.md` -> 5.7 Read Stream - Live (Long-poll)](./docs/ds-protocol.md#57-read-stream---live-long-poll) |
| Tail the stream continuously     | `client.live()`                          | async iterable of `data`, `up_to_date`, `eof`, reconnect events | [`docs/ds-protocol.md` -> 5.7 and 5.8](./docs/ds-protocol.md#57-read-stream---live-long-poll)                        |
| Open a raw SSE session           | `client.openSseSession()`                | lower-level SSE session object                                  | [`docs/ds-protocol.md` -> 5.8 Read Stream - Live (SSE)](./docs/ds-protocol.md#58-read-stream---live-sse)             |
| Delete the stream                | `client.delete()`                        | `deleted`                                                       | [`docs/ds-protocol.md` -> 5.4 Delete Stream](./docs/ds-protocol.md#54-delete-stream)                                 |

For most callers:

- use `read()` for one catch-up read
- use `live()` for long-running tailing
- use `openSseSession()` only if you need raw SSE control

## Result Shape

Every network method returns:

```ts
type Result<T, E> = { ok: true; result: T } | { ok: false; result: E };
```

That means your normal flow is:

```ts
const result = await client.append(...);
if (!result.ok) {
  console.error(result.result);
  return;
}

console.log(result.result.nextOffset);
```

If you need request and response details for debugging, pass
`includeMeta: true`.

## Payloads and Offsets

Read APIs return a `StreamPart`:

```ts
const read = await client.read({ offset: "-1" });
if (!read.ok) throw read.result;

read.result.payload.body;
read.result.payload.text();
read.result.payload.json();
```

Offset rules:

- use `-1` to read from the earliest retained data
- use `now` to start from the current tail
- otherwise, reuse the server-returned `nextOffset`
- treat offsets as opaque strings

Spec: [`docs/ds-protocol.md` -> 6. Offsets](./docs/ds-protocol.md#6-offsets)

## Live Reads

`client.live()` is the high-level live API:

```ts
for await (const event of client.live({ offset: "now" })) {
  if (event.type === "data") {
    console.log(event.payload.text());
  }
}
```

Defaults:

- `mode` defaults to `"auto"`
- `"auto"` tries SSE first and falls back to long-poll when needed

Spec:

- [`docs/ds-protocol.md` -> 5.7 Read Stream - Live (Long-poll)](./docs/ds-protocol.md#57-read-stream---live-long-poll)
- [`docs/ds-protocol.md` -> 5.8 Read Stream - Live (SSE)](./docs/ds-protocol.md#58-read-stream---live-sse)

## Snapshot and Bootstrap

If your stream supports snapshots, these are the key methods:

| Task                                        | API                                    | Spec                                                                                                   |
| ------------------------------------------- | -------------------------------------- | ------------------------------------------------------------------------------------------------------ |
| Read latest visible snapshot                | `client.readLatestSnapshot()`          | [`docs/ds-extensions.md` -> 2.3 Read Latest Snapshot](./docs/ds-extensions.md#23-read-latest-snapshot) |
| Read a specific snapshot                    | `client.readSnapshot(offset)`          | [`docs/ds-extensions.md` -> 2.5 Read Snapshot](./docs/ds-extensions.md#25-read-snapshot)               |
| Publish a snapshot                          | `client.putSnapshot({ offset, part })` | [`docs/ds-extensions.md` -> 2.2 Publish Snapshot](./docs/ds-extensions.md#22-publish-snapshot)         |
| Delete a snapshot                           | `client.deleteSnapshot(offset)`        | [`docs/ds-extensions.md` -> 2.6 Delete Snapshot](./docs/ds-extensions.md#26-delete-snapshot)           |
| Get snapshot + retained updates in one call | `client.bootstrap()`                   | [`docs/ds-extensions.md` -> 3. Bootstrap](./docs/ds-extensions.md#3-bootstrap)                         |

Typical recovery flow:

1. Call `client.bootstrap()`
2. Apply `snapshot` if it exists
3. Apply `updates` in order
4. Continue with `client.live({ offset: bootstrap.result.nextOffset })`

If a normal read returns `410 Gone`, you usually need `bootstrap()` before you
continue. Spec: [`docs/ds-extensions.md` -> 3.4 Compatibility](./docs/ds-extensions.md#34-compatibility)

## Producer Coordination

If you use idempotent writers, pass `producer` to `append()` or `close()`:

```ts
await client.append({
  part: {
    contentType: "application/octet-stream",
    body: bytes,
  },
  producer: {
    producerId: "writer-1",
    epoch: 1,
    seq: 42,
  },
});
```

Spec: [`docs/ds-protocol.md` -> 5.2.1 Idempotent Producers](./docs/ds-protocol.md#521-idempotent-producers)

## Other Exports

These helpers are useful when you need them, but most users only need
`StreamsClient`:

- `createStreamPart(contentType, body)`: build a `StreamPart` from bytes
- `normalizeStreamBody(body)`: normalize string, `Uint8Array`, `ArrayBuffer`, or view into `Uint8Array`
- `DEFAULT_RETRY_CONFIG`
- `DEFAULT_TIMEOUT_CONFIG`

## Spec Map

This package ships the protocol docs in `docs/`, so after install you can read:

- `node_modules/@loro-dev/streams-client/docs/ds-protocol.md`
- `node_modules/@loro-dev/streams-client/docs/ds-extensions.md`

Use this map when you need exact wire semantics:

- stream URL shape and bucket rules:
  [`docs/ds-extensions.md` -> 1. Buckets](./docs/ds-extensions.md#1-buckets)
- create, append, close, delete, head, read, long-poll, SSE:
  [`docs/ds-protocol.md`](./docs/ds-protocol.md)
- offsets and JSON mode:
  [`docs/ds-protocol.md` -> 6. Offsets](./docs/ds-protocol.md#6-offsets),
  [`docs/ds-protocol.md` -> 7.1 JSON Mode](./docs/ds-protocol.md#71-json-mode)
- snapshots and bootstrap:
  [`docs/ds-extensions.md` -> 2. Snapshots](./docs/ds-extensions.md#2-snapshots),
  [`docs/ds-extensions.md` -> 3. Bootstrap](./docs/ds-extensions.md#3-bootstrap)