@durable-streams/y-durable-streams

Yjs provider for Durable Streams - sync Yjs documents over HTTP with automatic server-side compaction and optional awareness (presence) support.

Overview

This package provides a Yjs provider that syncs documents using the Yjs Durable Streams Protocol. Unlike WebSocket-based providers, it uses standard HTTP (SSE by default, with long-polling as an alternative) plus automatic server-side compaction, making it simpler to deploy and scale.

Key benefits:

• No WebSocket infrastructure - Works with standard HTTP load balancers and CDNs
• Automatic compaction - Server manages document snapshots to keep sync fast
• Snapshot webhooks - Wake external workers to persist the latest snapshot
• Scalable - Stateless server design, documents stored in durable streams
• Presence support - Optional awareness for cursors, selections, and user status

Installation

npm install @durable-streams/y-durable-streams yjs y-protocols lib0

Quick Start

import { YjsProvider } from "@durable-streams/y-durable-streams"
import * as Y from "yjs"
import { Awareness } from "y-protocols/awareness"

const doc = new Y.Doc()
const awareness = new Awareness(doc)

const provider = new YjsProvider({
  doc,
  baseUrl: "http://localhost:4438/v1/yjs/my-service",
  docId: "my-document",
  awareness,
})

provider.on("synced", (synced) => {
  console.log("Synced:", synced)
})

Usage

Document Only (No Presence)

const provider = new YjsProvider({
  doc,
  baseUrl: "http://localhost:4438/v1/yjs/my-service",
  docId: "my-document",
})

With Authentication

const provider = new YjsProvider({
  doc,
  baseUrl: "http://localhost:4438/v1/yjs/my-service",
  docId: "my-document",
  awareness,
  headers: {
    Authorization: "Bearer your-token",
  },
})

Manual Connection

const provider = new YjsProvider({
  doc,
  baseUrl,
  docId,
  connect: false, // Don't connect automatically
})

// Set up listeners first
provider.on("synced", handleSync)
provider.on("error", handleError)

// Then connect
await provider.connect()

Event Handling

// Sync state changes
provider.on("synced", (synced: boolean) => {
  if (synced) {
    console.log("Document is synced with server")
  }
})

// Connection status changes
provider.on("status", (status: YjsProviderStatus) => {
  console.log("Status:", status) // "disconnected" | "connecting" | "connected"
})

// Error handling
provider.on("error", (error: Error) => {
  console.error("Provider error:", error)
})

Cleanup

// Disconnect temporarily
provider.disconnect()

// Reconnect
await provider.connect()

// Destroy permanently
provider.destroy()

API

YjsProvider

class YjsProvider {
  constructor(options: YjsProviderOptions)

  // Properties
  readonly doc: Y.Doc
  readonly synced: boolean
  readonly connected: boolean
  readonly connecting: boolean

  // Methods
  connect(): Promise<void>
  disconnect(): Promise<void>
  destroy(): void

  // Events
  on(event: "synced", handler: (synced: boolean) => void): void
  on(event: "status", handler: (status: YjsProviderStatus) => void): void
  on(event: "error", handler: (error: Error) => void): void
}

Options

interface YjsProviderOptions {
  doc: Y.Doc
  baseUrl: string // Yjs server URL, e.g. "http://localhost:4438/v1/yjs/my-service"
  docId: string // Document identifier (may contain forward slashes)
  awareness?: Awareness // Optional awareness for presence
  headers?: HeadersRecord // Optional auth headers (static strings or () => string)
  liveMode?: "sse" | "long-poll" // Live update transport (default: "sse")
  connect?: boolean // Auto-connect on construction (default: true)
}

Server

The package includes a Yjs server that implements the protocol. For development/testing:

import { YjsServer } from "@durable-streams/y-durable-streams/server"

const server = new YjsServer({
  port: 4438,
  dsServerUrl: "http://localhost:4437", // Durable streams server
})

await server.start()
console.log(`Yjs server running at ${server.url}`)

Snapshot webhooks

Subscribe a webhook to be woken whenever a newer snapshot is available for a matching document:

await fetch(
  `${server.url}/v1/yjs/my-service/__ds/subscriptions/snapshot-archive`,
  {
    method: "PUT",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({
      type: "webhook",
      events: ["snapshot.available"],
      document_pattern: "projects/**",
      webhook: { url: "https://archive.example/hooks/yjs" },
      lease_ttl_ms: 30_000,
    }),
  }
)

document_pattern is relative to the service's document namespace. * matches one path segment and ** matches recursively. The underlying Durable Streams server must support webhook subscriptions.

Webhook requests use the standard Durable Streams subscription payload and signature. A notification means one or more newer snapshots are available; it does not carry the snapshot bytes and may coalesce multiple compactions. For each pending .index stream, derive the document path and fetch its latest snapshot with ?offset=snapshot. Return { "done": true } only after the snapshot has been persisted successfully. New subscriptions observe future snapshots and do not replay the snapshot that was current when they were created.

Conformance Tests

The package includes conformance tests to verify Yjs server implementations. By default, tests run against local test servers. To test against an external server:

# Run tests against an external Yjs server
YJS_CONFORMANCE_URL=http://localhost:4438/v1/yjs/test pnpm vitest run --project y-durable-streams

# Run tests with local test servers (default)
pnpm vitest run --project y-durable-streams

Note: The "Server Restart" test is skipped when using an external URL since it requires starting/stopping local servers.

Server Protocol API

For the complete protocol specification, see YJS-PROTOCOL.md.

Base URL Structure

Each document is accessed via a single URL with query parameters:

{baseUrl}/docs/{docPath}?{queryParams}

Where:

• baseUrl is typically http://host:port/v1/yjs/{service}
• docPath can include forward slashes (e.g., project/chapter-1)

Key Operations

Snapshot Discovery

GET {baseUrl}/docs/{docPath}?offset=snapshot

Returns a 307 redirect to either:

• ?offset={N}_snapshot if a snapshot exists
• ?offset=-1 if no snapshot (read from beginning)

Read Snapshot

GET {baseUrl}/docs/{docPath}?offset={N}_snapshot

Returns binary Yjs snapshot with stream-next-offset header indicating where to continue reading updates.

Read/Write Updates

GET  {baseUrl}/docs/{docPath}?offset={N}&live=true
POST {baseUrl}/docs/{docPath}

• Read: Get updates from offset, optionally with live=true for long-polling
• Write: POST raw Yjs update bytes (server handles lib0 framing)

Awareness (Presence)

GET  {baseUrl}/docs/{docPath}?awareness=default&offset=now&live=true
POST {baseUrl}/docs/{docPath}?awareness=default

Named awareness streams via query parameter. Uses SSE for real-time delivery.

Compaction

The server automatically compacts documents when updates exceed a threshold:

1. Read current state (snapshot + updates)
2. Create new snapshot at current offset
3. Update internal index stream
4. Delete old snapshot

Compaction is transparent to clients - existing connections continue uninterrupted.

Error Responses

How It Works

The provider connects to a Yjs server which manages document storage using durable streams:

1. Snapshot Discovery - Client requests ?offset=snapshot, server redirects to current snapshot or beginning
2. Snapshot Loading - Binary Yjs state with stream-next-offset header for where to continue
3. Live Updates - Long-poll for incremental updates from the offset
4. Awareness - Optional SSE stream for presence (cursors, selections, user info)

The server automatically compacts documents when updates exceed a threshold, creating new snapshots. This keeps initial sync fast for new clients. The protocol uses a single URL per document with query parameters for different operations.

License

Apache-2.0