@loro-dev/streams-crdt

Transport layer for synchronizing CRDT state over Loro Streams (Durable Streams).

Core Concepts

Two-class model

| Class | Responsibility | | --- | --- | | StreamsCrdt | Transport runtime bound to one stream URL. Handles auth, reconnect, bootstrap, SSE/long-poll, binary framing, and remote cursor persistence. | | CrdtAdapter | Binding for one local CRDT instance. Handles snapshot export/import, version tracking, local update export, remote update application, and optional isolated 410 recovery inference. |

Operations

| Method | Purpose | | --- | --- | | createStream() | Creates the stream on the server. Returns { created: true } on first call, { created: false } if it already exists. | | deleteStream() | Deletes the stream. Closes any active join() first. | | sync() | One-shot bootstrap/catch-up cycle. Syncs remote state into local CRDT and pushes any local changes to the server. Does not enter live mode. | | catchup() | Pull-only remote catch-up. Applies remote stream updates to the local CRDT but does not intentionally append new local updates. If live reads from join() are already healthy, returns immediately unless { force: true } is passed. | | join() | Initial sync plus live subscription. Internally performs the same sync as sync(), then enters a persistent SSE/long-poll read loop and starts forwarding local writes. No need to call sync() before join(). | | close() | Closes any active join and releases resources. If this instance was using appendWriteOnly(), that write-only path is also torn down and cannot be used again on the same instance. |

join() includes an initial sync. You do not need to call sync() before join(). When join() resolves, the local CRDT is already caught up with the server and the live subscription is active.

Error handling — Result, not exceptions

All transport operations return a Result<T, TransportError> instead of throwing. Callers must check .ok before accessing the value:

const result = await transport.join();
if (!result.ok) {
  // result.error is a TransportError with a discriminated `code` field.
  switch (result.error.code) {
    case "stream_not_found":
      // 404 — the stream doesn't exist yet. Create it first.
      break;
    case "auth_failed":
      // 401/403 from the server — refresh credentials and retry.
      break;
    case "auth_provider_error":
      // Your auth callback threw. The live loop will retry automatically.
      // For permanent logout, return undefined from the callback instead.
      break;
    case "server_error":
      // 5xx — transient backend failure. The live loop retries automatically;
      // for one-shot calls (sync/createStream/etc.) you decide whether to
      // retry. `retryAfterMs` honors the server's Retry-After hint.
      break;
    case "network_error":
    case "timeout":
    case "initial_sync_timeout":
      // Transient — safe to retry.
      break;
    case "protocol_error":
    case "internal_error":
      // Non-retryable: bad server response or local runtime/adapter failure.
      break;
    default:
      console.error("transport error:", result.error);
  }
  return;
}

const sub = result.value; // TransportSubscription

Every error carries a retryable flag so callers can decide whether to retry without hard-coding error codes:

if (!result.ok && result.error.retryable) {
  // safe to retry after a delay
}

Surfacing errors to end users

HTTP-derived variants (server_error, auth_failed, protocol_error, stream_not_found, gone) carry the response details so apps can show actionable messages without parsing the message string:

const err = result.error;
if (err.code === "server_error") {
  toast.error(`Server error ${err.status}${err.requestId ? ` (req: ${err.requestId})` : ""}`);
  console.error("backend body:", err.body);
  // err.retryAfterMs may suggest a backoff if the server set Retry-After.
}

| Code | status | message | body | Other | | ---------------------- | -------- | --------- | ------ | ------------------------------ | | stream_not_found | 404 | yes | yes | — | | auth_failed | 401/403 | yes | yes | — | | protocol_error | yes | yes | yes | — | | gone | 410 | yes | yes | — | | server_error | 5xx | yes | yes | requestId?, retryAfterMs? | | timeout | — | yes | — | phase: "connect" \| "poll" | | initial_sync_timeout | — | yes | — | operation, timeoutMs, elapsedMs | | network_error | — | yes | — | — | | auth_provider_error | — | yes | — | cause? (original thrown value) | | internal_error | — | yes | — | — | | apply_incomplete | — | yes | — | reason: "loro_pending_updates" |

Operation-level retries vs. live-mode retries

• sync(), catchup(), createStream(), deleteStream(), appendWriteOnly() return after a single attempt. The application is responsible for retrying retryable errors.
• After join() succeeds, the live read/write loops automatically reconnect on every retryable: true error using exponential backoff ([0, 0.5s, 1s, 2s, 4s, 8s, 15s, 30s] with ±20% jitter; cap 30s). When a 5xx response includes Retry-After, the runtime waits at least that long. Non-retryable errors transition the subscription to status: "error"; call transport.rejoin() after fixing the cause (e.g., refreshing the auth token) to reset retry state and reconnect.

Initial sync deadlines

Initial bootstrap/catch-up work is allowed to run longer than ordinary request connect and poll deadlines. The transport uses reconnectConfig for this:

const transport = new StreamsCrdt({
  streamUrl,
  adapter,
  auth,
  reconnectConfig: {
    // Logs a debug diagnostic when initial sync is still pending; does not abort.
    initialSyncSlowAfterMs: 30_000,
    // Aborts initial bootstrap/catch-up with `initial_sync_timeout`.
    initialSyncHardTimeoutMs: 120_000,
  },
});

join(), sync(), pull-only catchup(), and 410 bootstrap recovery all use the initial-sync deadline path when they need to bootstrap or catch up from a stored cursor. In StreamsCrdt constructor options, set initialSyncHardTimeoutMs to 0 or Infinity to disable the hard cap. Because constructor reconnectConfig is merged over defaults, omitting the field keeps the default hard cap. Set initialSyncSlowAfterMs the same way to disable the slow diagnostic.

Why Result instead of throw? Transport failures (network errors, auth expiry, missing streams) are expected at runtime, not exceptional. Wrapping them in Result makes the error path explicit in the type system so callers cannot accidentally ignore failures with a missing try/catch.

Migration notes

For most callers nothing changes — if (!result.ok) { handle(result.error); } keeps working, and result.error.retryable is still the right gate for "safe to retry?".

Only adjust if any of these apply:

• Your auth callback throws on permanent failure (e.g., user is logged out). Throwing is now treated as a transient error and retried in live mode. Return undefined instead so the server returns 401 and you get the terminal auth_failed. See "What if the auth callback throws?" below.
• You construct TransportError literals (e.g., in tests or fixtures). stream_not_found now requires status: 404 + message; gone now requires status: 410. TypeScript will flag these.
• You exhaustively switch on code with a : never default. Add a case for auth_provider_error and initial_sync_timeout (and optionally server_error if you handled 5xx via unknown/internal_error before).
• You used "message" in error to defensively narrow. All variants now carry message; the guard is no longer needed.

Quick Start (Loro)

import { LoroDoc } from "loro-crdt";
import {
  createLoroDocAdapter,
  StreamsCrdt,
} from "@loro-dev/streams-crdt/loro";

const doc = new LoroDoc();
const transport = new StreamsCrdt({
  streamUrl: "https://streams-api.loro.dev/ds/my-bucket/doc-1",
  auth: async () => "<gateway-jwt>",
  adapter: createLoroDocAdapter(doc),
  // Optional TTL applied when the stream is created.
  streamTtlSeconds: 60 * 60 * 24 * 30,
  // Enable automatic snapshot upload so new readers bootstrap faster.
  snapshotUpload: { canUpload: async () => true },
});

// Create the stream (idempotent).
const created = await transport.createStream();
if (!created.ok) throw created.error;

// Join enters live collaborative editing mode.
// This performs an initial sync automatically before entering live mode.
const joined = await transport.join({
  onStatusChange(status) {
    console.log("room status:", status);
  },
});
if (!joined.ok) throw joined.error;

const sub = joined.value;

// The doc is now syncing in real time.
// Call sub.unsubscribe() or transport.close() to leave.

createLoroDocAdapter(doc) requires an attached LoroDoc. If the document was checked out or detached, call checkoutToLatest() or attach() before passing it to StreamsCrdt.

Request shard origins

Browsers can still show head-of-line blocking symptoms when large bootstrap/catch-up downloads and normal app requests contend on the same connection. shardUrls lets the transport replace only the request origin for selected operations while preserving the stream path and query:

const transport = new StreamsCrdt({
  streamUrl,
  auth: async () => "<gateway-jwt>",
  adapter: createLoroDocAdapter(doc),
  shardUrls: {
    bootstrap: [
      "https://control-a.api.streams-api-x.loro.dev",
      "https://control-b.api.streams-api-x.loro.dev",
    ],
    catchup: [
      "https://control-a.api.streams-api-x.loro.dev",
      "https://control-b.api.streams-api-x.loro.dev",
    ],
    largePost: [
      "https://write-a.api.streams-api-x.loro.dev",
      "https://write-b.api.streams-api-x.loro.dev",
    ],
    other: [
      "https://live-a.api.streams-api-x.loro.dev",
      "https://live-b.api.streams-api-x.loro.dev",
    ],
    largePostMinBytes: 64 * 1024,
  },
});

Shard entries must be origin URLs with no path, query, or hash. Each transport instance rotates through the configured origins per operation, with a randomized starting point so multiple tabs are less likely to all begin on the same shard. Bootstrap applies to GET <stream>/bootstrap; catch-up applies to non-SSE offset reads, including long-poll fallback. largePost applies to non-empty append POST bodies at or above largePostMinBytes bytes. other applies when no operation-specific pool routes the request, covering live SSE, create/delete, head, snapshot upload, and append POSTs not routed by largePost. The default threshold is 0, so every non-empty append POST can use largePost when that pool is configured. Verify actual connection separation with browser NetLog: browsers may coalesce compatible HTTP/2 or HTTP/3 origins when DNS, certificate, and transport settings allow it.

Loro pending imports and cursor safety

Loro can accept an update whose dependencies are missing and keep it as a pending import. Pending imports are not represented in doc.version() and are not included in snapshots exported from that document. For that reason, the Loro adapter must not persist a RemoteCursor until all pending imports are resolved and the local document is complete at the cursor offset.

The transport may continue reading with an in-memory volatile cursor to find the missing dependencies. If the server reports up-to-date while Loro still has unresolved imports, the client must bootstrap again and only save the bootstrap cursor after the batch import is clean. Snapshot upload is skipped while Loro has unresolved imports.

See specs/loro-pending.md for the full behavior and implementation checklist.

Quick Start (Loro EphemeralStore)

Use a separate EphemeralStreamCrdt for presence-like Loro EphemeralStore data. It bootstraps through the first SSE event, posts local ephemeral updates, and never republishes state imported from other peers. If a client needs to keep its own presence alive, update that client's own keys with store.set() so the store emits a fresh local update.

Ephemeral live delivery is latest-state oriented. The server may keep only a tiny live fanout queue for current SSE subscribers and may disconnect slow subscribers instead of buffering every intermediate cursor or presence update. The client reconnects and receives a fresh bootstrap, so the local store converges to current state without requiring offset replay.

import { EphemeralStore } from "loro-crdt";
import {
  EphemeralStreamCrdt,
  EphemeralStoreAdaptor,
} from "@loro-dev/streams-crdt/loro";

const store = new EphemeralStore();
const docStreamUrl = "https://streams-api.loro.dev/ds/my-bucket/doc-1";

const presence = new EphemeralStreamCrdt({
  streamUrl: `${docStreamUrl}?ephemeral=presence`,
  auth: async () => "<gateway-jwt>",
  adaptor: EphemeralStoreAdaptor(store),
});

const joinedPresence = await presence.join();
if (!joinedPresence.ok) throw joinedPresence.error;

Ephemeral stream URL format

An ephemeral stream URL is the durable stream URL for the same document plus a channel query parameter:

{baseUrl}/ds/{bucketId}/{streamId}?ephemeral={channel}

For example, if the durable Flock document stream is:

https://streams-api.loro.dev/ds/my-bucket/flock-cm9vbS0x

then the matching presence stream is:

https://streams-api.loro.dev/ds/my-bucket/flock-cm9vbS0x?ephemeral=presence

Use the same bucketId and streamId as the durable StreamsCrdt room. The ephemeral value names a side-channel for that room, such as presence, cursor, or selection. Different channels under the same durable stream are separate in-memory rooms. The compatibility spelling ?awareness=presence is also accepted, but new code should use ?ephemeral=presence.

When an app exposes a room link such as /rooms/design-review-42, the link normally maps to one durable stream id first, and both durable and ephemeral sync derive from that same id:

const roomId = "design-review-42";
const durableStreamUrl = new URL(
  `ds/my-bucket/${encodeURIComponent(`flock-${roomId}`)}`,
  "https://streams-api.loro.dev/proxy/",
).toString();

const presenceStreamUrl = new URL(durableStreamUrl);
presenceStreamUrl.searchParams.set("ephemeral", "presence");

const flockTransport = new StreamsCrdt({
  streamUrl: durableStreamUrl,
  auth,
  adapter: createFlockAdapter(flock),
});

const presenceTransport = new EphemeralStreamCrdt({
  streamUrl: presenceStreamUrl.toString(),
  auth,
  adaptor: EphemeralStoreAdaptor(store),
});

The ds/... path is relative so a gateway path prefix such as /proxy/ is preserved. Validate dynamic bucket and stream segments with isValidRillId() before constructing the URL; this rejects ./.. before URL normalization.

EphemeralStreamCrdt does not create or delete the durable stream. It posts updates to the ephemeral URL and opens SSE on the same URL with live=sse. Ephemeral state is not persisted with the Flock document stream; it is scoped by the URL path plus channel and is meant for current live state only.

Quick Start (Flock)

import { Flock } from "@loro-dev/flock-wasm";
import {
  createFlockAdapter,
  StreamsCrdt,
} from "@loro-dev/streams-crdt/flock";

const flock = new Flock();
const transport = new StreamsCrdt({
  streamUrl: "https://streams-api.loro.dev/ds/my-bucket/room-1",
  auth: async () => "<gateway-jwt>",
  adapter: createFlockAdapter(flock),
});

await transport.createStream();
const joined = await transport.join();
if (!joined.ok) throw joined.error;

const sub = joined.value;

When to Use catchup(), sync(), and join()

Use catchup() when you only want to pull remote updates into the local CRDT:

// Pull remote changes without uploading local pending changes.
const caughtUp = await transport.catchup();
if (!caughtUp.ok) throw caughtUp.error;
// caughtUp.value.cursor contains the replay progress after the pull.

catchup() is safe to call while a room is live. When join() has a healthy SSE/long-poll read loop, catchup() returns immediately because the live loop is already catching up continuously. Pass { force: true } to issue one explicit non-SSE offset read anyway, for example after app foreground or device wake:

await transport.catchup({ force: true });

If the live read side of the room is reconnecting, disconnected, or in error, catchup() skips the live backoff path and tries one immediate non-SSE catch-up read from the room's current cursor. A successful call updates the local cursor and nudges the live loops to reconnect, but it does not by itself guarantee that local pending writes have reached the server.

Use sync() when you need a one-shot bidirectional sync without entering live mode:

// Offline-first: pull remote changes, upload local changes, then work offline.
const synced = await transport.sync();
if (!synced.ok) throw synced.error;
// synced.value.cursor contains the replay progress after sync.

Use sync() or subscription.waitUntilSynced() when the caller needs local pending changes uploaded to the server.

Use join() for real-time collaboration. It does everything sync() does and then keeps the connection alive for continuous bidirectional sync:

const joined = await transport.join();

Delete Stream

const deleted = await transport.deleteStream();
if (!deleted.ok) throw deleted.error;

console.log(deleted.value.deleted); // true when deleted now, false when already missing

deleteStream() closes any active join first. When the configured remoteCursorStore implements delete(streamUrl) (the built-in stores do), the transport also clears the stored replay cursor for that stream.

Auth

auth may be:

• a static string
• a sync callback returning a string
• an async callback returning a string

StreamsCrdt asks for a token before each request. If a request comes back with 401 or 403, it calls auth one more time with { reason: "unauthorized", status, previousToken } and retries that request once.

This lets the caller keep a cached token in the normal path, and only refresh after the server rejects it:

let token = readCachedGatewayJwt();

const transport = new StreamsCrdt({
  streamUrl,
  adapter,
  auth: async (context) => {
    if (context?.reason === "unauthorized") {
      token = await refreshGatewayJwt();
    }
    return token;
  },
});

If your callback already returns a fresh token every time, you can ignore the context argument.

What if the auth callback throws?

Throwing from the callback is treated as a transient failure. The runtime wraps the error as auth_provider_error (retryable: true) so a hiccupping auth backend doesn't terminate the live read/write loops — they retry under the same exponential backoff used for server_error and network_error.

For the "user is permanently logged out" case, return undefined (or null/empty string) instead of throwing. The request goes out without an Authorization header, the server returns 401, and you receive the non-retryable auth_failed instead — which the application typically routes to a sign-in page.

auth: async () => {
  try {
    return await fetchTokenFromAuthBackend(); // network → throw → retry
  } catch (e) {
    if (isPermanentLogoutError(e)) {
      return undefined; // → server 401 → auth_failed (terminal)
    }
    throw e; // → auth_provider_error (retryable)
  }
}

End-to-End Encryption (E2EE)

e2ee enables end-to-end encryption for CRDT update batches and snapshots before their bytes are appended to Durable Streams. Auth stays separate: gateway tokens authorize HTTP requests, while document keys protect document bytes from the server and unauthorized readers.

E2EE at a Glance

1. What the application provides

The application injects a PayloadProtectionProvider, not a raw key. At the streams-crdt API boundary, the provider supplies three things:

• seal() encrypts with the current write key and a fresh nonce, then returns an opaque authenticated header plus provider-defined sealed bytes.
• open() parses that header, resolves the corresponding historical read key, authenticates the ciphertext, and returns plaintext.
• maxSealOverheadBytes declares the maximum provider-added size so streams-crdt can batch updates without speculatively sealing or consuming a nonce.

The provider owns the audited AEAD implementation, nonce generation, sealed layout, opaque header schema, logical-room AAD, current write key/epoch, and historical key lookup. streams-crdt never receives a key directly and does not implement MLS, device identity, epoch state, key storage, or product policy.

2. What streams-crdt does

streams-crdt places that provider at the common durable payload boundary:

local CRDT payload
  -> batch and freeze
  -> provider.seal()
  -> streams-crdt envelope and transport framing
  -> Durable Streams

Durable Streams payload
  -> remove transport framing and parse the envelope
  -> provider.open()
  -> clear CRDT payload
  -> apply to Loro or Flock

The same boundary covers updates, snapshots, bootstrap, catch-up, live SSE/long-poll, 410 recovery, and write-only appends. Frozen update retries reuse the same ciphertext instead of calling seal() and generating a second nonce. Protected reads authenticate before adapter apply or cursor persistence and fail closed on any envelope or provider error.

streamUrl remains an opaque transport URL and is not cryptographic identity. The provider must bind a stable logical room identity itself when cross-room replay protection is required. This boundary covers only durable StreamsCrdt payloads: EphemeralStreamCrdt presence, awareness, cursor, selection, and display-name payloads are outside it.

3. What gets encoded

Every protected update batch and snapshot uses the same envelope:

protectedEnvelope =
  "LSCE"[4]
  || version[1]
  || payloadKind[1]
  || providerHeaderLength[2]
  || reserved[2]
  || providerHeader[H]
  || sealed[provider-defined]

The fixed envelope header is 10 bytes. providerHeader is 1–512 bytes, so the prefix before sealed is 10 + H, or 11–522 bytes. The provider header is visible but authenticated; streams-crdt does not parse it.

For example, a provider using XChaCha20-Poly1305 commonly chooses this sealed layout:

sealed = nonce[24] || ciphertext[plaintext.length] || authenticationTag[16]

The 24-byte nonce is the random 192-bit XChaCha nonce. The 16-byte tag is the 128-bit Poly1305 authentication tag. XChaCha ciphertext has the same length as its plaintext, so this provider layout adds 24 + 16 = 40 bytes. This layout is an example, not a format imposed by streams-crdt.

The two durable payload classes then differ only in their surrounding framing:

update append body = u32be(protectedEnvelope.length) || protectedEnvelope
snapshot PUT body  = protectedEnvelope

If H is the provider-header length, a direct XChaCha layout adds the following bytes relative to the plaintext passed to seal():

| Payload | Size calculation | Added bytes | | --- | --- | ---: | | Update batch | outer length 4 + envelope 10 + header H + nonce/tag 40 | 54 + H | | Snapshot | envelope 10 + header H + nonce/tag 40 | 50 + H |

For a 16-byte provider header, that is 70 bytes per encrypted update append and 66 bytes per encrypted snapshot. That provider's declared maxSealOverheadBytes must cover H + 40; the fixed 10-byte envelope and the update-only 4-byte outer length are accounted for separately by streams-crdt. Compression may independently change the snapshot plaintext size before sealing. The 40-byte streams-crdt AAD domain tag is reconstructed locally and authenticated, not serialized as another 40 bytes on the wire.

The server can still observe the stream URL, provider header, offsets, total payload sizes, timing, and auth metadata. It cannot read the CRDT payload or, within one encrypted update batch, its update count and individual update sizes.

Minimal Provider Setup

The cryptographic and header helpers in this example are application pseudocode, not exports from @loro-dev/streams-crdt. An application should implement them with an audited AEAD library and an unambiguous binary encoding. The only streams-crdt API being implemented here is PayloadProtectionProvider.

import {
  PayloadProtectionError,
  type PayloadProtectionProvider,
} from "@loro-dev/streams-crdt/loro";

// This provider belongs to one application room session and captures that
// session's finalized write epoch and stable room AAD. Header parsing remains
// provider-owned. frameAad() must encode its inputs unambiguously.
const roomAad = encodeApplicationRoomIdentity(roomId);
const provider: PayloadProtectionProvider = {
  // header + random 24-byte nonce + 16-byte tag, conservatively bounded.
  maxSealOverheadBytes: 256,

  async seal({ plaintext, additionalData }) {
    const header = encodeLodyHeader({
      epoch: roomSession.writeEpoch,
      keyId: roomSession.writeKeyId,
    });
    const nonce = crypto.getRandomValues(new Uint8Array(24));
    const ciphertextAndTag = await xchacha20poly1305Seal({
      key: roomSession.writeKey,
      nonce,
      plaintext,
      aad: frameAad(roomAad, additionalData(header)),
    });
    return { header, sealed: concatBytes(nonce, ciphertextAndTag) };
  },
  async open({ sealed, header, additionalData }) {
    const historical = decodeLodyHeader(header);
    // Resolve from logical room + provider-owned epoch/key id.
    const key = await roomKeys.readKey(roomId, historical);
    if (key == null) {
      throw new PayloadProtectionError("missing_read_key");
    }
    return await xchacha20poly1305Open({
      key,
      nonce: sealed.slice(0, 24),
      ciphertextAndTag: sealed.slice(24),
      aad: frameAad(roomAad, additionalData),
    });
  },
};

const transport = new StreamsCrdt({
  streamUrl: "https://streams-api.loro.dev/ds/my-bucket/doc-1",
  adapter: createLoroDocAdapter(doc),
  e2ee: {
    provider,
    // Defaults shown explicitly:
    readPolicy: "encrypted-only",
    writePolicy: "encrypt",
  },
});

Provider Contract

The package's exported TypeScript declarations are the source of truth. The relevant contract, defined in src/types.ts, is equivalent to this abbreviated definition:

type MaybePromise<T> = T | Promise<T>;
type PayloadProtectionKind = "update_batch" | "snapshot";

interface PayloadProtectionContext {
  readonly protocol: "loro-streams-crdt-payload-protection";
  readonly version: 2;
  readonly kind: PayloadProtectionKind;
}

interface PayloadProtectionProvider {
  readonly maxSealOverheadBytes: number;

  seal(input: {
    readonly plaintext: Uint8Array;
    readonly context: PayloadProtectionContext;
    readonly additionalData: (header: Uint8Array) => Uint8Array;
  }): MaybePromise<{
    readonly header: Uint8Array;
    readonly sealed: Uint8Array;
  }>;

  open(input: {
    readonly sealed: Uint8Array;
    readonly header: Uint8Array;
    readonly context: PayloadProtectionContext;
    readonly additionalData: Uint8Array;
  }): MaybePromise<Uint8Array>;
}

interface E2eeOptions {
  readonly provider?: PayloadProtectionProvider;
  readonly readPolicy?: "encrypted-only" | "allow-plaintext";
  readonly writePolicy?: "encrypt" | "plaintext";
}

The provider is optional only when writePolicy is "plaintext"; protected reads still need one to open encrypted history.

The inputs have these exact meanings:

| Field | Meaning | | --- | --- | | plaintext in seal() | For update_batch, an inner length-prefixed batch containing one or more CRDT update blobs. For snapshot, the adapter snapshot after optional snapshotCodec.compress. | | context | Stable streams-crdt protocol/version/payload-kind context. It contains no URL, bucket, stream, room, epoch, or key id. | | additionalData(header) in seal() | Builds the exact streams-crdt-owned AAD for the final provider header. It must be called exactly once. | | header | Provider-owned, plaintext metadata used to select an algorithm or historical key. It must be 1–512 bytes and must be authenticated. | | sealed | Provider-owned non-empty bytes, normally nonce plus ciphertext and authentication tag. | | additionalData in open() | The same streams-crdt-owned AAD bytes produced for that envelope at write time. |

For cross-room replay protection, use keys isolated per logical room or combine application room AAD and streams-crdt AAD without ambiguity. For example, use domain tags plus length-prefixed fields; do not use plain string concatenation where two input pairs can produce the same bytes. seal() must return the exact header passed to additionalData(header).

The sealed layout stays provider-owned. For example, it may be a random 192-bit XChaCha20-Poly1305 nonce followed by ciphertext and tag; streams-crdt does not parse the nonce or enforce an algorithm suite.

seal() runs exactly once after a logical append is frozen for retry, so a retry reuses the same ciphertext. It does not define an epoch transition barrier. For a Lody write-epoch change, drain pending appends with waitUntilSynced(), close the old room transport, finalize the epoch change, then create a replacement room transport/provider. Only that room session must be rebuilt, not the whole repo. The replacement provider may still resolve old headers dynamically in open().

Encrypted streams still use application/octet-stream. The server stores opaque bodies and never parses the protected envelope. Update appends retain their outer item-length framing. Snapshot PUT bodies contain the snapshot envelope directly; they do not use the update append's outer item frame.

Payload Lifecycle and Framing

E2EE wraps the existing transport payloads; it does not replace Durable Streams HTTP, CAS, offset, multipart, or live-event framing.

Update write path

One seal() call protects one frozen append batch, not one individual CRDT update. Before freezing an append, the transport may merge FIFO queued batches while the estimated encoded body remains within its 256 KiB append budget. It then performs this pipeline:

adapter export/subscription
  -> clearUpdates[]                         one or more opaque CRDT update blobs
  -> encodeItems(clearUpdates)              inner update-batch plaintext
  -> provider.seal(kind = "update_batch")  exactly once for this frozen append
  -> protectedEnvelope
  -> encodeItems([protectedEnvelope])       outer Durable Streams append frame
  -> POST/CAS body

encodeItems() is a sequence of unsigned 32-bit big-endian length prefixes:

innerUpdateBatch =
  u32be(update_1.length) || update_1
  || u32be(update_2.length) || update_2
  || ...

protectedEnvelope =
  fixedEnvelopeHeader || providerHeader || providerSealedBytes

encryptedAppendBody =
  u32be(protectedEnvelope.length) || protectedEnvelope

The provider receives innerUpdateBatch as seal().plaintext. The outer length prefix is added only after sealing so streams-crdt readers can recover the protected item from the stored append. The server still treats the whole append body as opaque and does not learn the number or individual sizes of the CRDT updates inside the batch. Once the protected append and producer sequence are frozen, retries reuse the same bytes and do not call seal() or generate another nonce.

With writePolicy: "plaintext", there is no inner encrypted batch or protected envelope: the append body is simply encodeItems(clearUpdates).

Update read path

Bootstrap tails, catch-up responses, long-poll responses, and SSE events all reuse the same reverse path:

Durable Streams append bytes
  -> decodeItems(outer append frame)
  -> detect and parse protectedEnvelope
  -> provider.open(kind = "update_batch")
  -> innerUpdateBatch plaintext
  -> decodeItems(innerUpdateBatch)
  -> clearUpdates[]
  -> adapter.applyRemoteUpdates(clearUpdates, remoteVersion)

An encrypted outer item must open to one valid inner encodeItems() payload. It may expand to several CRDT update blobs before the adapter is called. Under allow-plaintext, a non-envelope outer item is passed to the adapter as one clear update. Under encrypted-only, that same item fails before adapter apply or cursor persistence.

Snapshot write and read paths

Snapshots use the same provider envelope but not the update batch's inner or outer item framing:

adapter.exportSnapshot()
  -> snapshotCodec.compress?()
  -> provider.seal(kind = "snapshot")
  -> protectedEnvelope
  -> PUT snapshot body directly

bootstrap snapshot part
  -> parse protectedEnvelope
  -> provider.open(kind = "snapshot")
  -> snapshotCodec.decompress?()
  -> adapter.applySnapshot()

The provider therefore sees the optionally compressed snapshot bytes as seal().plaintext. A selected snapshot is exported, compressed, and sealed once before its PUT; an HTTP-level resend of that prepared request reuses the same body. A later scheduled snapshot may export and seal newer state again. Snapshot offset headers and bootstrap multipart boundaries remain outside the envelope because the service must read them.

Bootstrap, catch-up, live, and recovery

All durable paths converge on the two update hooks and two snapshot hooks described above:

| Path | Processing order | | --- | --- | | Initial bootstrap | Open/decompress/apply the snapshot when present, then open and apply the retained update tail. | | sync() / catchup() | Deframe and open every returned update batch before adapter apply. | | join() live SSE or long-poll | Use the same update read path for every event; there is no live-specific plaintext bypass. | | 410 recovery | Re-bootstrap through the same snapshot and update hooks before resuming live reads. | | Local join/sync writes | Merge, freeze, seal, and append through the same update write path. | | appendWriteOnly() | Uses the same frozen update write path but performs no remote read or snapshot upload. | | Snapshot upload | Compress, seal, and PUT through the same snapshot write path. |

Protected reads complete authentication and decoding before applySnapshot(), applyRemoteUpdates(), local durability hooks, or remote cursor persistence. Any failure therefore leaves the corresponding remote cursor unadvanced.

Provider Envelope and AAD

Provider envelope v2 is:

magic               4 bytes   "LSCE"
version             1 byte    0x02
kind                1 byte    0x01 update_batch, 0x02 snapshot
provider_header_len 2 bytes   unsigned big-endian; 1..512
reserved            2 bytes   both 0
provider_header     provider_header_len bytes, opaque
sealed              remaining bytes, provider-defined and non-empty

The common AAD is an unambiguous byte sequence:

UTF8("loro-streams-crdt-payload-protection/v2\0")
|| exact envelope bytes from magic through provider_header

This streams-crdt-owned AAD binds protocol/version, payload kind, and the provider's entire opaque header. Algorithm, epoch, key-id, application room, and stream identity schemas are not part of the streams-crdt protocol. A provider that needs cross-room replay protection must capture a stable logical room identity and combine it unambiguously with additionalData, as in the example above. Do not use a gateway origin as room identity unless moving that origin is intentionally a cryptographic migration.

The service does not parse the v2 envelope. It remains inside the application payload: append/CAS headers, offsets, multipart boundaries, and snapshot offsets stay outside so Durable Streams can route and retain bytes.

maxSealOverheadBytes must be a safe integer from 0 through 4096 and must bound header.length + sealed.length - plaintext.length. The transport uses that bound to merge queued CRDT batches without speculative sealing or nonce use; actual output above the declaration fails closed.

Draft Compatibility

Provider v2 is still unreleased. Ciphertext produced by earlier revisions of this draft branch used URL-derived AAD and is not compatible with the final caller-owned AAD contract; discard or recreate that draft data rather than adding a production dual-reader for an unpublished format.

Mode Matrix

e2ee supports a few distinct shapes. Choose one explicitly:

| Use case | readPolicy | writePolicy | Required protection config | Remote plaintext accepted? | Local writes sent as | | --- | --- | --- | --- | --- | --- | | Existing plaintext stream | omitted | omitted | none | Yes | plaintext | | Provider private reader/writer | encrypted-only (default) | encrypt (default) | streamUrl + provider | No | provider v2 | | Mixed provider reader/writer | allow-plaintext | encrypt | streamUrl + provider | Yes | provider v2 | | Public write-only writer | allow-plaintext | plaintext | none | Yes | plaintext via appendWriteOnly() | | Provider reader without writes | encrypted-only | plaintext | streamUrl + provider | No | plaintext if local writes are exported |

There is no separate transport-level read-only mode. The last row is only appropriate when the local CRDT is treated as read-only or the caller's auth rejects writes. writePolicy: "plaintext" avoids requiring a write key; it does not suppress local export attempts by itself.

Adapter Compatibility: Loro and Flock

e2ee is adapter-agnostic at the transport layer: the same options shape works with createLoroDocAdapter(doc) and createFlockAdapter(flock). Encryption wraps the CRDT payload bytes that the adapter already exports.

That also means one stream must use exactly one CRDT family:

• A stream written with the Loro adapter must be read with the Loro adapter.
• A stream written with the Flock adapter must be read with the Flock adapter.
• Do not mix Loro and Flock replicas against the same stream, whether payload protection is enabled or not. The decrypted inner update bytes and snapshots are CRDT-specific and are not cross-compatible.

The examples below use separate stream URLs and room-bound providers:

const streamUrl = "https://streams-api.loro.dev/ds/my-bucket/doc-1";
const flockStreamUrl = "https://streams-api.loro.dev/ds/my-bucket/flock-1";
const loroProvider = createRoomProvider("doc-1");
const flockProvider = createRoomProvider("flock-1");

Protected Loro room:

import { LoroDoc } from "loro-crdt";
import {
  createLoroDocAdapter,
  StreamsCrdt,
} from "@loro-dev/streams-crdt/loro";

const doc = new LoroDoc();
const loroTransport = new StreamsCrdt({
  streamUrl,
  adapter: createLoroDocAdapter(doc),
  e2ee: {
    provider: loroProvider,
  },
});

Protected Flock room:

import { Flock } from "@loro-dev/flock-wasm";
import {
  createFlockAdapter,
  StreamsCrdt,
} from "@loro-dev/streams-crdt/flock";

const flock = new Flock();
const flockTransport = new StreamsCrdt({
  streamUrl: flockStreamUrl,
  adapter: createFlockAdapter(flock),
  e2ee: {
    provider: flockProvider,
  },
});

If one application stack uses Loro and another uses Flock, give them separate streams. Reusing the same key material across those streams is an application decision; it does not make the CRDT payloads interoperable.

Private E2EE Room

This is the default protected mode: all retained updates and snapshots are encrypted, and readers reject any plaintext payload that appears in the stream.

const privateReaderWriter = new StreamsCrdt({
  streamUrl,
  adapter: createLoroDocAdapter(doc),
  e2ee: {
    provider,
  },
});

Mixed Plaintext and Encrypted Updates

Some applications intentionally allow public writers to append CRDT updates but not read private history. Configure readers explicitly before accepting those plaintext updates:

const privateReaderWriter = new StreamsCrdt({
  streamUrl,
  adapter: createLoroDocAdapter(doc),
  e2ee: {
    readPolicy: "allow-plaintext",
    writePolicy: "encrypt",
    provider,
  },
});

Plaintext writers should opt in just as explicitly:

const publicWriter = new StreamsCrdt({
  streamUrl,
  adapter: createLoroDocAdapter(publicDoc),
  auth: async () => getWriteOnlyToken(),
  e2ee: {
    readPolicy: "allow-plaintext",
    writePolicy: "plaintext",
  },
});

publicDoc.getText("comments").insert(0, "public note");
const appended = await publicWriter.appendWriteOnly();

Behavior and Validation Rules

• If e2ee is present and no policies are specified, the defaults are readPolicy: "encrypted-only" and writePolicy: "encrypt".
• streamUrl is required, remains opaque, and is never parsed to invent a security identity.
• streams-crdt does not bind ciphertext to streamUrl. The provider/application must authenticate a stable logical room identity or use keys isolated per room. A provider configured for the wrong room must fail authentication.
• PayloadProtectionProviderConfig (alias E2eeProviderConfig) is a scope-free { provider, readPolicy?, writePolicy? } shape suitable for a repo to forward directly. Set payloadProtectionRequired: true once protection is enabled at repo level so a missing room config throws instead of becoming plaintext. This flag checks config presence only; it does not override an explicit allow-plaintext / plaintext policy such as public write-only mode.
• Lody rekey is an application-owned room-session barrier: drain, close, switch the finalized write epoch, and create a new room transport. Historical reads may resolve provider headers dynamically.
• appendWriteOnly() is the write-only path. It only posts local CRDT batches that this transport observed after construction. It does not bootstrap, catch up, join SSE, upload snapshots, apply remote updates, or persist a remote cursor.
• appendWriteOnly() must not be mixed with sync() or join() on the same transport instance.
• After close(), appendWriteOnly() is permanently unavailable on that instance because the write-only subscription has been released. Create a new StreamsCrdt instance for later write-only use.
• Snapshot uploads are encrypted when writePolicy is "encrypt". Snapshot upload is rejected when writePolicy is "plaintext"; a public writer that cannot read private state should not publish full-document snapshots.
• Protected read failures stop before applySnapshot(), applyRemoteUpdates(), local durability hooks, or remote cursor persistence.
• A recognized protected envelope never falls back to plaintext after an authentication failure, unknown version/key/epoch, wrong scope/kind, or tamper. Under encrypted-only, bytes without the envelope magic are also rejected as plaintext_forbidden. allow-plaintext intentionally accepts non-envelope bytes and is therefore not an encrypted-only room policy.
• Keep historical keys until retained protected updates and snapshots no longer reference them.
• Provider batching uses only declared bounded overhead. It never calls seal() while estimating a merge and validates the actual frozen output.

If e2ee is configured and readPolicy is left at the default "encrypted-only", remote plaintext updates fail with code: "payload_protection_error" and reason: "plaintext_forbidden".

E2EE Errors

E2EE-related transport failures surface as code: "payload_protection_error" with one of these reasons:

| Reason | Meaning | | --- | --- | | plaintext_forbidden | A reader configured as encrypted-only saw plaintext data in the stream. | | missing_read_key | The provider cannot resolve its opaque header to a historical read key. | | decrypt_failed | The provider AEAD open failed. Common causes are a wrong key, wrong scope, or tampered bytes. | | invalid_envelope | The payload was not a supported streams-crdt protected envelope/version. | | wrong_payload_kind | An encrypted snapshot was used where an update batch was expected, or the reverse. | | encrypt_failed | Local write-side encryption could not proceed because the key/config/input was invalid. |

Callers usually treat these as operator or configuration errors, not transient network failures. They are always returned as retryable: false. The public TransportError carries only the stable reason and a stable redacted message; it never copies provider causes, key ids, or epochs into the default error path.

E2EE does not hide stream URLs, the opaque provider header, offsets, total append/snapshot sizes, timing, or auth metadata. It hides the count and individual sizes of updates within one encrypted append batch, but it is not a traffic-analysis defense. It also does not by itself detect a malicious server rolling back or forking history. In mixed allow-plaintext mode, no format can distinguish deliberate plaintext from ciphertext whose magic was stripped; use encrypted-only for a no-downgrade room.

Remote Cursor Store

The RemoteCursorStore persists replay progress metadata only — it does not store any CRDT update data. A remote cursor contains:

• nextOffset — where the transport should resume reading from
• serverLowerBoundVersion — the CRDT version that the server data represents
• streamUrl — identity of the stream

Important: cursor-only persistence is unsafe

If RemoteCursorStore.load() returns a saved cursor, the transport resumes from nextOffset and assumes the caller has already restored the matching local CRDT state represented by serverLowerBoundVersion.

If the local CRDT was reset to empty but the cursor was persisted, the transport will do delta-only catch-up from nextOffset — applying those deltas against an empty document will fail or produce an invalid state.

Rule of thumb

| Local CRDT state | Cursor store | Safe? | | --- | --- | --- | | Not persisted (ephemeral) | InMemoryRemoteCursorStore (default) | Yes | | Persisted (IndexedDB, SQLite, etc.) | IndexedDbRemoteCursorStore + beforeRemoteCursorSave hook | Yes | | Not persisted | IndexedDbRemoteCursorStore | No — data loss risk |

If you do not persist the local CRDT payload across reloads, keep the remote cursor ephemeral too. The default InMemoryRemoteCursorStore is the safe choice.

Hooking Local Durability

Use beforeRemoteCursorSave to couple local payload persistence with remote cursor advancement. The transport advances the cursor only after:

1. Remote data was applied locally
2. The durability hook succeeded
3. The cursor store write succeeded

const transport = new StreamsCrdt({
  streamUrl: "https://streams-api.loro.dev/ds/my-bucket/doc-1",
  adapter: createLoroDocAdapter(doc),
  beforeRemoteCursorSave: async ({ cursor, source }) => {
    // Persist local CRDT state before cursor advances.
    // `source` is "local" | "remote" | "bootstrap".
    await persistLocalPayload(doc);
    console.log(source, cursor.nextOffset);
  },
});

persistLocalPayload(doc) must make the local durable state correspond to the cursor that is about to be stored. On the next app start, restore that local state before calling sync() or join().

Persistent Cursor Example

Only use a persisted RemoteCursorStore such as IndexedDbRemoteCursorStore when you also persist and restore the local CRDT state for the same stream.

import { LoroDoc } from "loro-crdt";
import {
  createLoroDocAdapter,
  IndexedDbRemoteCursorStore,
  StreamsCrdt,
} from "@loro-dev/streams-crdt/loro";

// Step 1: restore local CRDT state first.
const doc = (await restorePersistedDoc("doc-1")) ?? new LoroDoc();

const transport = new StreamsCrdt({
  streamUrl: "https://streams-api.loro.dev/ds/my-bucket/doc-1",
  auth: async () => "<gateway-jwt>",
  adapter: createLoroDocAdapter(doc),
  // Step 2: use a persistent cursor store.
  remoteCursorStore: new IndexedDbRemoteCursorStore({
    dbName: "my-app-ds-cursors",
  }),
  // Step 3: persist local state before cursor advances.
  beforeRemoteCursorSave: async () => {
    await persistLocalPayload("doc-1", doc);
  },
});

// join() handles initial sync + live mode.
const joined = await transport.join();

The required ordering is:

1. Restore the local CRDT state first
2. Let the transport apply remote updates
3. Persist the updated local CRDT state (via beforeRemoteCursorSave)
4. Only then persist the remote cursor

If you cannot provide steps 1 and 3, do not use a persisted remote cursor store.

Snapshot Codec Hooks

Use snapshotCodec when you want transport to transform full snapshots during upload and bootstrap:

const transport = new StreamsCrdt({
  streamUrl,
  adapter: createLoroDocAdapter(doc),
  snapshotCodec: {
    // Sync or async are both allowed.
    compress: async (snapshot) => snapshot,
    decompress: (snapshot) => snapshot,
  },
});

Behavior

• snapshotCodec.compress(snapshot) runs right before a snapshot is uploaded to Loro Streams
• snapshotCodec.decompress(snapshot) runs right after a non-empty snapshot is downloaded from Loro Streams and right before adapter.applySnapshot(...)
• Both hooks must be provided together
• Both hooks may return either Uint8Array or Promise<Uint8Array>
• These hooks affect full snapshots only. Incremental update batches are unchanged

Compatibility

Transport does not version or negotiate snapshot codecs for you.

If you enable snapshotCodec, you must make sure decompress can still read any older snapshots that are already stored for that stream. This is especially important when:

• existing snapshots were uploaded without compression
• you change the compression format later
• you roll out the codec gradually across clients

@loro-dev/streams-crdt/zstd

@loro-dev/streams-crdt/zstd is a built-in helper entry point backed by @bokuweb/zstd-wasm.

pnpm add @loro-dev/streams-crdt

Then wire the exported hooks directly into snapshotCodec:

import { LoroDoc } from "loro-crdt";
import {
  createLoroDocAdapter,
  StreamsCrdt,
} from "@loro-dev/streams-crdt/loro";
import {
  compress,
  decompress,
} from "@loro-dev/streams-crdt/zstd";

const doc = new LoroDoc();
const transport = new StreamsCrdt({
  streamUrl: "https://streams-api.loro.dev/ds/my-bucket/doc-1",
  adapter: createLoroDocAdapter(doc),
  snapshotCodec: { compress, decompress },
});

@loro-dev/streams-crdt/zstd exports:

• compress(snapshot) — async; uses an extra Worker when available and falls back to the current thread otherwise
• compressInWorker(snapshot) — async; worker-only compression helper
• compressOnCurrentThread(snapshot) — async; current-thread compression helper
• decompress(snapshot) — async; current-thread decompression helper

Automatic Snapshot Upload

The transport can upload snapshots automatically during join() when callers opt in with snapshotUpload. This feature is disabled by default — if you do not pass snapshotUpload, no snapshots will ever be uploaded.

const transport = new StreamsCrdt({
  streamUrl,
  adapter: createLoroDocAdapter(doc),
  // Enable automatic snapshot upload:
  snapshotUpload: {
    canUpload: async () => true,
    // debounceMs: 10_000,                 // default: 10 seconds
    // minBytesSinceRemoteSnapshot: 102400, // default: 100 KB
  },
});

Configuration

| Option | Default | Description | | --- | --- | --- | | canUpload | (required) | Authorization gate called before each upload attempt. Return false to skip. | | debounceMs | 10_000 (10 s) | Debounce window after the last local write before attempting upload. | | minBytesSinceRemoteSnapshot | 102400 (100 KB) | Minimum byte delta between the last remote snapshot offset and the current stream tail. Upload is skipped when the delta is smaller. |

Behavior

• Automatic snapshot upload is considered only during join(), never sync()
• The feature is opt-in: callers must pass snapshotUpload with at least canUpload to enable it
• Successful local appends while join() is active refresh a debounce window (default 10 s) for a possible snapshot upload check
• If a remote apply advances the inferred remote version during that window, the transport treats that as another writer becoming active and skips that snapshot attempt
• At debounce expiry, the transport may enqueue a best-effort maybeUploadSnapshot task onto the same serialized operation queue used for local append and remote apply
• When that task executes, it fetches the latest remote snapshot offset, uses the latest confirmed local cursor.nextOffset as the candidate snapshot_offset, and uploads only if current_tail_offset - remote_snapshot_offset exceeds the configured byte threshold

Entry Points

| Import path | What it adds | | --- | --- | | @loro-dev/streams-crdt | Transport core, cursor stores, ID helpers, all shared types | | @loro-dev/streams-crdt/loro | Everything above + createLoroDocAdapter(doc) | | @loro-dev/streams-crdt/flock | Everything above + createFlockAdapter(flock) | | @loro-dev/streams-crdt/zstd | Snapshot compress / decompress hooks backed by @bokuweb/zstd-wasm |

Public API Reference

Root Entry Point

• StreamsCrdt — transport runtime class
• StreamsCrdtOptions — constructor options
• CrdtAdapter — adapter contract interface
• IsolatedCrdtAdapter — optional apply-only adapter used for safe 410 recovery
• CrdtUpdateBatch — local update batch type
• CrdtApplyOutcome, CrdtApplyReturn, CrdtUnresolvedSpan — adapter apply result types used to report incomplete imports such as Loro pending updates
• Result — Rust-style Ok | Err result
• StreamsAuthProvider — auth callback type
• StreamsCrdtShardUrlsOptions — optional origin pools for request routing
• TransportError — discriminated error union apply_incomplete is returned when remote data was read but the CRDT adapter still has unresolved dependencies and the cursor cannot be saved.
• TransportCatchupParams — catchup() options
• TransportCreateStreamSuccess, TransportDeleteStreamSuccess, TransportCatchupSuccess, TransportSyncSuccess — result payloads
• TransportJoinParams — join() options
• TransportSubscription — active live subscription handle
• TransportRoomStatus — "joined" | "reconnecting" | "disconnected" | "error"
• SnapshotCodec, SnapshotTransformHook — full snapshot encode/decode hooks
• WriteOnlyAppendResult — result returned by appendWriteOnly()
• E2eeError — alias for PayloadProtectionError
• E2eeOptions, E2eeProvider, E2eeProviderConfig, E2eeReadPolicy, E2eeWritePolicy — E2EE configuration types
• PayloadProtectionProvider, PayloadProtectionProviderConfig, PayloadProtectionOptions — provider implementation and room configuration
• PayloadProtectionSealInput, PayloadProtectionSealResult, PayloadProtectionOpenInput — exact seal() / open() byte contracts
• PayloadProtectionContext, PayloadProtectionKind, PayloadProtectionReadPolicy, PayloadProtectionWritePolicy — stable protocol context and policy types
• PayloadProtectionError, PayloadProtectionFailureReason — provider and transport failure classification
• SnapshotUploadOptions — snapshot upload configuration
• RemoteCursor — replay progress metadata
• RemoteCursorStore — abstract cursor storage interface
• RemoteCursorSaveSource — "local" | "remote" | "bootstrap"
• BeforeRemoteCursorSaveContext, BeforeRemoteCursorSaveHook — durability hook types
• InMemoryRemoteCursorStore — ephemeral cursor store (default, always safe)
• IndexedDbRemoteCursorStore — persistent cursor store (requires local CRDT persistence)
• IndexedDbRemoteCursorStoreOptions — IndexedDB store config
• createInitialRemoteCursor(...) — seed a cursor store
• isValidRillId(...) — stream ID validation helper

Loro Entry Point

@loro-dev/streams-crdt/loro re-exports everything from the root plus:

• createLoroDocAdapter(doc) — creates a CrdtAdapter for one LoroDoc

Flock Entry Point

@loro-dev/streams-crdt/flock re-exports everything from the root plus:

• createFlockAdapter(flock) — creates a CrdtAdapter for one Flock replica
• VersionVector, ExportBundle — re-exported from @loro-dev/flock-wasm

Zstd Entry Point

@loro-dev/streams-crdt/zstd exports:

• compress(snapshot) — async snapshot compression helper
• compressInWorker(snapshot) — async worker-only snapshot compression helper
• compressOnCurrentThread(snapshot) — async current-thread snapshot compression helper
• decompress(snapshot) — async snapshot decompression helper

Notes

• createStreamIfMissing defaults to false and only affects the initial sync/join bootstrap
• streamTtlSeconds is a non-negative safe integer in seconds. It is sent only when the transport creates the stream, either through createStream() or automatic creation with createStreamIfMissing.
• deleteStream() returns { deleted: false } when the target stream is already missing
• sync() and join() return Result objects instead of implicitly creating missing streams
• join() resolves only after the initial replay has succeeded; without a stored remote cursor it bootstraps once, then prefers ordinary live SSE and falls back sticky to long-poll
• One transport instance maps to one DS stream
• One adapter instance maps to one local CRDT instance
• Built-in remote cursor stores implement delete(streamUrl) so deleteStream() can reset replay state cleanly

Repository-only Scripts

The published package only includes built runtime artifacts under dist/. For repository verification against the hosted backend, run:

• pnpm --dir packages/streams-crdt run test:e2e:streams-api