# Copy change records to your own store

> Run an observer that joins a room with an SDK and writes each change record to your database.



An **observer** is a backend process that joins a collaboration room with a SuperDoc SDK. It reads [change records](/editor/collaboration/record-changes) as they arrive and writes them to your own storage. Your collaboration server needs no SuperDoc code.

## Before you start [#before-you-start]

Install the SDK for your language. See the [Node.js](/agents/automation/node-sdk), [Python](/agents/automation/python-sdk) or [Rust](/agents/automation/rust-sdk) 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](/editor/collaboration/deploy-a-server#check-access) 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 [#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 [#nodejs]

Save this as `observe-room.ts`:

```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:

```bash
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 [#python]

Save this as `observe_room.py`:

```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 [#rust]

Add `observe_room.rs` to your crate:

```rs
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](/agents/automation/rust-sdk). 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 [#how-the-loop-works]

The three samples follow the same steps:

1. Join the room with `roomMode: 'join'`.
2. Take a snapshot. A session sees only the edits made after it joined, so this snapshot records the state it started from.
3. 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.
4. 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.
5. Replace the cursor with the one the result returns, then save it. Do this on every call, even when there are no records.
6. 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 [#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](/editor/collaboration/record-changes#oversized-records) 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 [#restarts-and-gaps]

> **A cursor belongs to the session that issued it (warning)**
>
> Never build or compare cursors. After the observer restarts, its saved cursor comes from an earlier session. The first
> call then returns from the oldest record the new session holds, with `truncated: true`. The loop stores those records,
> adopts the new cursor and takes a snapshot. The next call is normal again.


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](/editor/collaboration/record-changes#which-sessions-record). 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 [#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:

```ts
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-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 [#name-versions]

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

```json
{
  "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](/editor/version-history) for DOCX versions. Restoring a version in a live room is a workflow your server owns. See [Restore shared documents separately](/editor/version-history#restore-shared-documents-separately).

## Next steps [#next-steps]

* [Verify claims](/editor/collaboration/verify-claims) in the observer and store the result next to each record.
* Persist the room itself with [Save and restore a room](/editor/collaboration/save-and-restore-a-room).
* [Read authorship on your own server](/editor/collaboration/read-authorship-on-your-server) when you only need who sent each update.