@inkly/crdt

CRDT transport helpers for inkly -- relay Yjs / Automerge binary updates and awareness over rooms, with debounced coalescing.

This package is not a CRDT. It treats Uint8Array payloads as opaque blobs and relays them unchanged. Any CRDT library that emits binary updates (Yjs, Automerge, ...) is supported.

Why

Collaborative editors emit many small updates in rapid succession. Sending each one as a separate WebSocket message wastes bandwidth. @inkly/crdt coalesces bursts of updates into a single network message per room per peer, reducing message volume without changing the semantics of applying the changes.

Awareness updates (cursor positions, online status) are relayed immediately -- they are ephemeral and must not be delayed.

References

• Yjs binary updates: Y.encodeStateAsUpdate / Y.applyUpdate / doc.on("update", (u: Uint8Array) => ...) -- https://github.com/yjs/yjs and https://docs.yjs.dev
• Automerge binary changes: Automerge.getLastLocalChange / Automerge.applyChanges / Automerge.save / Automerge.load -- https://automerge.org

Install

pnpm add @inkly/crdt

Yjs is an optional peer dependency. Install it separately if you use Yjs:

pnpm add yjs

Quickstart

Server

import { inkly } from "@inkly/core";
import { installCrdt, crdtContract } from "@inkly/crdt";

const app = inkly(crdtContract);

// Coalesce updates for 20ms, relay awareness immediately.
const crdt = installCrdt(app, {
  debounceMs: 20,
  maxWaitMs: 200,
});

app.on("open", (peer) => {
  peer.join("doc:main"); // add peers to the document room
});

// On shutdown: flush pending batches.
crdt.dispose();

Client (Yjs)

import * as Y from "yjs";
import { createClient } from "@inkly/client";
import { bindDoc, toBase64, fromBase64, crdtContract } from "@inkly/crdt";

const doc = new Y.Doc();
const client = createClient(crdtContract, { url: "ws://localhost:3000" });

let myPeerId: string | undefined; // populate from your auth/session

const REMOTE = "inkly-remote"; // tag for remotely-applied updates

const binding = bindDoc(doc, {
  origin: REMOTE,
  debounceMs: 20,
  send: (updates) =>
    updates.forEach((u) =>
      void client.actions["crdt:update"]({ room: "doc:main", data: toBase64(u) })
    ),
  applyRemote: (u) => Y.applyUpdate(doc, u, REMOTE),
});

// Receive batched updates from the server.
client.on("crdt:update", ({ from, updates }) => {
  if (from === myPeerId) return; // filter own echo (see note below)
  binding.receive(updates.map(fromBase64));
});

// Teardown:
// binding.dispose();

Client filters own from

app.to(room).emit() sends to all room members including the origin peer. Every crdt:update and crdt:awareness event therefore carries a from: peer.id field. Clients must skip events where from equals their own peer id to avoid echo-applying their own changes:

client.on("crdt:update", ({ from, updates }) => {
  if (from === myPeerId) return; // skip own echo
  // ...
});

Binary encoding vs msgpack

inkly's default JSON codec cannot round-trip a raw Uint8Array. This package therefore defaults to base64 strings (+33% size) which work everywhere.

If your app uses a binary codec (e.g. @inkly/msgpack), set binary: true:

installCrdt(app, { binary: true });

| Codec | binary option | Wire format | Overhead | |---|---|---|---| | jsonCodec (default) | false (default) | base64 string[] | +33% | | @inkly/msgpack | true | Uint8Array[] | none |

Security: rooms and cross-room writes

The target room is supplied by the client on every action, but membership is decided by the server (peer.join(...), usually derived from peer.context). By default these are decoupled, so a peer can write CRDT bytes into any room name it likes — including a document room it never joined. (It still cannot read that room; fan-out only reaches members.)

If rooms are an authorization boundary (e.g. one room per document/tenant), enable the membership guard:

installCrdt(app, { requireMembership: true });

A relay into a room the peer has not joined (!peer.rooms.has(room)) is then rejected with a FORBIDDEN error and nothing is fanned out. Correct clients are unaffected — they join a room before collaborating in it. Default is false to preserve the open relay model.

API

toBase64(bytes: Uint8Array): string

Encode a Uint8Array as a base64 string. Handles all byte values 0x00-0xFF.

fromBase64(b64: string): Uint8Array

Decode a base64 string to Uint8Array. Inverse of toBase64.

coalesce<T>(onFlush, opts): Coalescer<T>

Create a debounce/coalesce buffer. Pushes items into a batch; flushes after debounceMs of idle, or after maxWaitMs from the first item. Timers are unref'd on Node.js so they do not keep the process alive.

interface Coalescer<T> {
  push(item: T): void;
  flush(): void;       // immediate flush
  dispose(): void;     // cancel timers, discard buffer
  readonly pending: number;
}

crdtContract

Ready-made inkly contract fragment. Spread into your own contract or pass directly to inkly(). Declares:

• Actions (client -> server): crdt:update, crdt:awareness -- both accept { room: string; data: string }
• Events (server -> client): crdt:update with { from: string; updates: string[] }, crdt:awareness with { from: string; data: string }

installCrdt(app, options?): { dispose() }

Register crdt:update and crdt:awareness action handlers on an inkly app. Returns { dispose() } to flush + clean up all coalescers.

interface CrdtOptions {
  debounceMs?: number;  // default 20
  maxWaitMs?:  number;  // default 200
  binary?:     boolean; // default false
}

Note on typing: just pass your real inkly app — installCrdt has a generic overload (installCrdt<C, Ctx>(app: InklyApp<C, Ctx>, options?)) that accepts it with no cast. Compose crdtContract into your app's contract so the crdt:update / crdt:awareness actions validate at runtime. The exported CrdtApp structural interface is only needed if you build a custom (non-inkly) relay host and want to type-check against it.

bindDoc(doc, opts): { receive(updates); dispose() }

Bind a CRDT document to an inkly transport. Subscribes to local doc updates, coalesces them, and calls opts.send. receive(updates) applies remote updates via opts.applyRemote. Works with Yjs Y.Doc out of the box.

interface BindDocOptions {
  origin?:     unknown;                        // tag to skip echo-applied updates
  debounceMs?: number;                         // default 20
  send:        (updates: Uint8Array[]) => void;
  applyRemote: (update: Uint8Array)   => void;
}

Caveats

• Origin exclusion is client-side. The server broadcasts to the whole room (including the sender). Clients must check from === myPeerId.
• base64 overhead. Default encoding adds ~33% to update size. Use @inkly/msgpack + binary: true to eliminate it.
• No merge. The server relays updates verbatim. Clients receive an ordered array and apply them in sequence; CRDT convergence is the library's job.
• No persistence. Joining peers do not receive historical state. Add a persistence layer (e.g. load full state on open and send via peer.send) if late-joiner sync is required.

See docs/overview.md and examples/yjs-server.ts.