Copy change records to your own store

An observer is a backend process that joins a collaboration room with a SuperDoc SDK. It reads change records as they arrive and writes them to your own storage. Your collaboration server needs no SuperDoc code.

Before you start

Install the SDK for your language. See the Node.js, Python or Rust guide.

The observer opens the room the same way any SDK session does. It supports these providers:

`providerType`	Room	Authentication
`'hocuspocus'`	`url` and `documentId`	A token for this open request: `collaborationAuth: { type: 'token', token }`.
`'y-websocket'`	`url` and `documentId`	A token for this open request, as for Hocuspocus.
`'liveblocks'`	`roomId`	`authEndpoint` or `publicApiKey` in `collaboration`. There is no per-open token.

The samples below use Hocuspocus. For another provider, change the collaboration object and the authentication.

Give the observer read-only access where your server supports it. On Hocuspocus, mark the observer's connection read-only. The server template does this when its access check returns readOnly: true. A read-only connection writes nothing to the room, not even attribution keys. The observer makes no edits of its own.

Run the observer

Each sample keeps its history in local files under history/<documentId>/: records.jsonl, cursor.txt and a snapshots/ folder. Replace the file code with your database and keep the same steps.

Node.js

Save this as observe-room.ts:

import { appendFileSync, existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { SuperDocClient } from '@superdoc/sdk';

const SNAPSHOT_EVERY_RECORDS = 250;
const SNAPSHOT_EVERY_MS = 10 * 60 * 1000;

const documentId = process.env.DOCUMENT_ID ?? 'contract-123';
const token = process.env.OBSERVER_TOKEN;
if (!token) throw new Error('Set OBSERVER_TOKEN to a read-only token for the room.');

const historyDir = join('history', documentId);
const recordsFile = join(historyDir, 'records.jsonl');
const cursorFile = join(historyDir, 'cursor.txt');
mkdirSync(join(historyDir, 'snapshots'), { recursive: true });

const storedIds = new Set<string>(
  existsSync(recordsFile)
    ? readFileSync(recordsFile, 'utf8')
        .split('\n')
        .filter(Boolean)
        .map((line) => JSON.parse(line).id)
    : [],
);
let cursor = existsSync(cursorFile) ? readFileSync(cursorFile, 'utf8') : '';

let stopping = false;
process.once('SIGINT', () => {
  stopping = true;
});

const client = new SuperDocClient();

try {
  await client.connect();
  const doc = await client.open(
    {
      collaboration: {
        providerType: 'hocuspocus',
        url: process.env.COLLAB_URL ?? 'ws://127.0.0.1:1234',
        documentId,
        roomMode: 'join',
      },
    },
    { collaborationAuth: { type: 'token', token } },
  );

  const saveSnapshot = async () => {
    const snapshot = await doc.changes.snapshot({ includeState: true });
    const file = join(historyDir, 'snapshots', `${snapshot.at.replaceAll(':', '-')}.json`);
    writeFileSync(file, JSON.stringify(snapshot));
  };

  try {
    await saveSnapshot();
    let sinceSnapshot = 0;
    let lastSnapshotAt = Date.now();

    while (!stopping) {
      const result = await doc.changes.list({
        ...(cursor ? { after: cursor } : {}),
        waitMs: 30_000,
        includeUpdate: true,
      });

      for (const record of result.items) {
        if (storedIds.has(record.id)) continue;
        appendFileSync(recordsFile, `${JSON.stringify(record)}\n`);
        storedIds.add(record.id);
        sinceSnapshot += 1;
      }
      cursor = result.cursor ?? '';
      writeFileSync(cursorFile, cursor);

      const idle = result.items.length === 0 && sinceSnapshot > 0;
      const due =
        sinceSnapshot >= SNAPSHOT_EVERY_RECORDS ||
        (sinceSnapshot > 0 && Date.now() - lastSnapshotAt >= SNAPSHOT_EVERY_MS);
      if (result.truncated || idle || due) {
        await saveSnapshot();
        sinceSnapshot = 0;
        lastSnapshotAt = Date.now();
      }
    }
  } finally {
    await doc.close({ discard: true });
  }
} finally {
  await client.dispose();
}

Run it with a token for the room:

OBSERVER_TOKEN=... COLLAB_URL=wss://collab.example.com DOCUMENT_ID=contract-123 npx tsx observe-room.ts

Press Ctrl+C to stop. The observer finishes its current wait, which takes up to 30 seconds, then closes the document.

Python

Save this as observe_room.py:

import json
import os
import time
from pathlib import Path

from superdoc import SuperDocClient

SNAPSHOT_EVERY_RECORDS = 250
SNAPSHOT_EVERY_SECONDS = 10 * 60

document_id = os.environ.get("DOCUMENT_ID", "contract-123")
token = os.environ["OBSERVER_TOKEN"]

history_dir = Path("history") / document_id
records_file = history_dir / "records.jsonl"
cursor_file = history_dir / "cursor.txt"
(history_dir / "snapshots").mkdir(parents=True, exist_ok=True)

stored_ids = set()
if records_file.exists():
    for line in records_file.read_text().splitlines():
        if line:
            stored_ids.add(json.loads(line)["id"])
cursor = cursor_file.read_text() if cursor_file.exists() else ""


def save_snapshot(document):
    snapshot = document.changes.snapshot({"includeState": True})
    name = snapshot["at"].replace(":", "-") + ".json"
    (history_dir / "snapshots" / name).write_text(json.dumps(snapshot))


with SuperDocClient() as client:
    document = client.open(
        {
            "collaboration": {
                "providerType": "hocuspocus",
                "url": os.environ.get("COLLAB_URL", "ws://127.0.0.1:1234"),
                "documentId": document_id,
                "roomMode": "join",
            },
        },
        collaboration_auth={"type": "token", "token": token},
    )

    try:
        save_snapshot(document)
        since_snapshot = 0
        last_snapshot_at = time.monotonic()

        while True:
            params = {"waitMs": 30000, "includeUpdate": True}
            if cursor:
                params["after"] = cursor
            result = document.changes.list(params)

            with records_file.open("a") as records:
                for record in result["items"]:
                    if record["id"] in stored_ids:
                        continue
                    records.write(json.dumps(record) + "\n")
                    stored_ids.add(record["id"])
                    since_snapshot += 1
            cursor = result["cursor"] or ""
            cursor_file.write_text(cursor)

            idle = not result["items"] and since_snapshot > 0
            due = since_snapshot >= SNAPSHOT_EVERY_RECORDS or (
                since_snapshot > 0 and time.monotonic() - last_snapshot_at >= SNAPSHOT_EVERY_SECONDS
            )
            if result["truncated"] or idle or due:
                save_snapshot(document)
                since_snapshot = 0
                last_snapshot_at = time.monotonic()
    finally:
        document.close({"discard": True})

Run it with OBSERVER_TOKEN, COLLAB_URL and DOCUMENT_ID set, the same way as the Node.js sample. Ctrl+C stops it and closes the document. AsyncSuperDocClient takes the same calls with await.

Rust

Add observe_room.rs to your crate:

use std::{
    collections::HashSet,
    error::Error,
    fs,
    io::Write,
    path::PathBuf,
    time::{Duration, Instant},
};

use serde_json::{json, Value};
use superdoc_sdk::{
    generated::types::{
        DocChangesListBoundParams, DocChangesSnapshotBoundParams,
        DocCloseBoundParams, DocOpenParams,
    },
    wire::{Nullable, Number},
    Client, CollaborationAuth, Document, OpenOptions,
};

const SNAPSHOT_EVERY_RECORDS: usize = 250;
const SNAPSHOT_EVERY: Duration = Duration::from_secs(10 * 60);

pub struct FileHistory {
    dir: PathBuf,
    stored_ids: HashSet<String>,
    cursor: Option<String>,
}

impl FileHistory {
    pub fn open(document_id: &str) -> Result<Self, Box<dyn Error>> {
        let dir = PathBuf::from("history").join(document_id);
        fs::create_dir_all(dir.join("snapshots"))?;
        let mut stored_ids = HashSet::new();
        if let Ok(lines) = fs::read_to_string(dir.join("records.jsonl")) {
            for line in lines.lines().filter(|line| !line.is_empty()) {
                let record: Value = serde_json::from_str(line)?;
                if let Some(id) = record["id"].as_str() {
                    stored_ids.insert(id.to_owned());
                }
            }
        }
        let cursor = fs::read_to_string(dir.join("cursor.txt"))
            .ok()
            .filter(|cursor| !cursor.is_empty());
        Ok(Self {
            dir,
            stored_ids,
            cursor,
        })
    }

    fn append(&mut self, records: &[Value]) -> Result<usize, Box<dyn Error>> {
        let mut file = fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(self.dir.join("records.jsonl"))?;
        let mut added = 0;
        for record in records {
            let Some(id) = record["id"].as_str() else {
                continue;
            };
            if !self.stored_ids.insert(id.to_owned()) {
                continue;
            }
            writeln!(file, "{record}")?;
            added += 1;
        }
        Ok(added)
    }

    fn save_cursor(
        &mut self,
        cursor: Option<String>,
    ) -> Result<(), Box<dyn Error>> {
        fs::write(
            self.dir.join("cursor.txt"),
            cursor.as_deref().unwrap_or(""),
        )?;
        self.cursor = cursor;
        Ok(())
    }

    fn save_snapshot(&self, snapshot: &Value) -> Result<(), Box<dyn Error>> {
        let at = snapshot["at"]
            .as_str()
            .unwrap_or("snapshot")
            .replace(':', "-");
        fs::write(
            self.dir.join("snapshots").join(format!("{at}.json")),
            snapshot.to_string(),
        )?;
        Ok(())
    }
}

pub async fn observe_room(
    client: &Client,
    url: &str,
    document_id: &str,
    token: &str,
    should_stop: impl Fn() -> bool,
) -> Result<(), Box<dyn Error>> {
    let params: DocOpenParams = serde_json::from_value(json!({
        "collaboration": {
            "providerType": "hocuspocus",
            "url": url,
            "documentId": document_id,
            "roomMode": "join"
        }
    }))?;
    let options = OpenOptions {
        collaboration_auth: Some(CollaborationAuth::token(token)?),
        ..OpenOptions::default()
    };
    let document = client.open_with_options(&params, options).await?;
    let mut history = FileHistory::open(document_id)?;

    let result = copy_records(&document, &mut history, should_stop).await;

    document
        .close(&DocCloseBoundParams::new().with_discard(true))
        .await?;
    result
}

async fn save_snapshot(
    document: &Document,
    history: &FileHistory,
) -> Result<(), Box<dyn Error>> {
    let snapshot = document
        .api()
        .changes()
        .snapshot(
            &DocChangesSnapshotBoundParams::new().with_include_state(true),
        )
        .await?;
    history.save_snapshot(&serde_json::to_value(&snapshot)?)
}

async fn copy_records(
    document: &Document,
    history: &mut FileHistory,
    should_stop: impl Fn() -> bool,
) -> Result<(), Box<dyn Error>> {
    save_snapshot(document, history).await?;
    let mut since_snapshot = 0;
    let mut last_snapshot_at = Instant::now();

    while !should_stop() {
        let mut list = DocChangesListBoundParams::new()
            .with_wait_ms(Number::new(30000.0)?)
            .with_include_update(true);
        if let Some(after) = &history.cursor {
            list = list.with_after(after.clone());
        }
        let result = document.api().changes().list(&list).await?;

        let records = serde_json::to_value(&result.items)?;
        since_snapshot += history
            .append(records.as_array().map(Vec::as_slice).unwrap_or(&[]))?;
        history.save_cursor(match result.cursor {
            Nullable::Value(cursor) => Some(cursor),
            Nullable::Null => None,
        })?;

        let idle = result.items.is_empty() && since_snapshot > 0;
        let due = since_snapshot >= SNAPSHOT_EVERY_RECORDS
            || (since_snapshot > 0
                && last_snapshot_at.elapsed() >= SNAPSHOT_EVERY);
        if result.truncated || idle || due {
            save_snapshot(document, history).await?;
            since_snapshot = 0;
            last_snapshot_at = Instant::now();
        }
    }
    Ok(())
}

Call observe_room with a connected Client, as in the Rust SDK guide. It returns when should_stop returns true, and closes the document either way. The sample needs serde_json and tokio with the macros and rt features.

How the loop works

The three samples follow the same steps:

Join the room with roomMode: 'join'.
Take a snapshot. A session sees only the edits made after it joined, so this snapshot records the state it started from.
Call changes.list({ after: cursor, waitMs: 30000, includeUpdate: true }). The call returns as soon as a record after the cursor exists, or with no records after 30 seconds.
Append each record whose id is not stored yet. Use id as the primary key in your database and insert only when it is absent.
Replace the cursor with the one the result returns, then save it. Do this on every call, even when there are no records.
Take a snapshot every 250 records, every 10 minutes while the room is active, when a call returns nothing after activity, and when truncated is true.

These numbers are sample defaults. SuperDoc does not schedule snapshots for you.

What to keep

A record's update is the exact Yjs update behind it, base64 encoded. Keep it when you want to inspect or audit the bytes of an edit. An oversized record can arrive without its update and with part of its text left out.

A snapshot holds the revision, the cursor, the last change id, the time, the Yjs state vector and, with includeState, the room's full state as a base64 Yjs update. Use snapshots, not records, to rebuild a room's state. Updates that produce no record, such as SuperDoc's own maintenance writes, are not in your record list.

SuperDoc sessions keep deleted content. Content deleted before the observer joined is in a snapshot only if your server's Yjs documents keep deleted content (garbage collection off).

Restarts and gaps

Edits made while no observer was connected have no records. The new session starts empty and sees only what happens after it joins. The snapshot it takes first gives you the room's state at that point, and its state vector shows where the gap ends.

truncated: true also means the session dropped records before you read them, for example because the observer fell behind the session's retention window. The loop takes a snapshot when it sees it.

If a call throws, for example because the connection failed, close the document, open the room again and continue from your saved cursor. Run the observer under a supervisor that restarts it when it exits.

Push instead of poll

The SDK host also sends each batch of records as an event.change notification with params { sessionId, records }. Pushed records carry no update and no cursor. To keep a complete history, treat a notification as a signal to call changes.list from your cursor. This loop prints each record; store them as the polling samples do:

import { SuperDocCliError, SuperDocClient } from '@superdoc/sdk';

const token = process.env.OBSERVER_TOKEN;
if (!token) throw new Error('Set OBSERVER_TOKEN to a read-only token for the room.');

const client = new SuperDocClient();
const notifications = client.subscribe({ capacity: 256 });

try {
  await client.connect();
  const doc = await client.open(
    {
      collaboration: {
        providerType: 'hocuspocus',
        url: process.env.COLLAB_URL ?? 'ws://127.0.0.1:1234',
        documentId: process.env.DOCUMENT_ID ?? 'contract-123',
        roomMode: 'join',
      },
    },
    { collaborationAuth: { type: 'token', token } },
  );

  try {
    let cursor: string | null = null;
    for (;;) {
      const result = await doc.changes.list({ ...(cursor ? { after: cursor } : {}), includeUpdate: true });
      for (const record of result.items) console.log(record.id, record.actor?.id ?? 'unknown');
      cursor = result.cursor;
      if (result.items.length > 0) continue;

      try {
        const next = await notifications.next();
        if (next.done) break;
      } catch (error) {
        if (error instanceof SuperDocCliError && error.code === 'NOTIFICATIONS_LAGGED') continue;
        throw error;
      }
    }
  } finally {
    await doc.close({ discard: true });
  }
} finally {
  notifications.close();
  await client.dispose();
}

When a subscriber falls behind, the SDK drops the oldest notifications and raises NOTIFICATIONS_LAGGED. The loop above loses nothing, because it reads records from its cursor. A record can arrive by push and by changes.list, so keep deduplicating on id. event.remoteChange still reports only that something changed.

In Python, client.subscribe(capacity=256) returns a stream with recv(timeout=30), which returns None on timeout. In Rust, client.subscribe() returns a stream with recv().await. Both raise an error with code NOTIFICATIONS_LAGGED when they fall behind.

Polling with waitMs has no lag case, so it is the simpler loop to start with.

Run more than one observer

Run one observer per room. Sessions that receive the same update compute the same id, including for merges and moves. Yjs can merge updates, for example while a client reconnects, and a merged update forms one record with a new id that covers the same edits. Two observers can therefore store the same edit under different ids. Do not merge their histories by id alone.

Name versions

A named version is a pointer into your store. Keep:

{
  "name": "Sent to legal",
  "revision": "48",
  "cursor": "…",
  "snapshotRef": "history/contract-123/snapshots/2026-11-05T10-15-00.000Z.json",
  "createdBy": "alex",
  "createdAt": "2026-11-05T10:15:00Z"
}

revision and cursor come from changes.snapshot(). snapshotRef points at the stored state. Create a version when someone saves, on a timer, or when the room goes idle.

You can also export a DOCX at that point and store it next to the pointer. See Version history for DOCX versions. Restoring a version in a live room is a workflow your server owns. See Restore shared documents separately.

Next steps

Verify claims in the observer and store the result next to each record.
Persist the room itself with Save and restore a room.
Read authorship on your own server when you only need who sent each update.

On this page