Add collab document persistence hooks and content-cache refresh (#35)

The collaboration server becomes the writer of page state (ADR 0003,
realtime-collaboration.md §lifecycle):

- onLoadDocument reconstructs a page's Y.Doc from PostgreSQL by applying
  `pages.ydoc_state` and then every `page_updates` row in order, so a page
  with a long update log loads correctly.
- onStoreDocument persists debounced (2 s, max 30 s): it appends the delta
  since the last flush to `page_updates`, periodically merges the log back
  into `ydoc_state` (inline threshold; the session-aware compaction of idle
  pages remains the separate job, #40), refreshes `page_content_cache`
  (plain text / Markdown / HTML / outline via the shared derivation, #24),
  bumps `pages.updated_at`, and keeps `Attachment.pageId` pointed at the
  embedding page (#31). Each flush runs in one transaction and its duration
  is logged.
- The document size ceiling (MAX_PAGE_DOCUMENT_BYTES) is enforced on store:
  an oversize document is not persisted and the clients are notified with a
  stateless error so they can revert.

Persistence is an injected port (PagePersistence): the Postgres
implementation is covered by a DB-backed test (store/load round-trip,
content-cache refresh, a 1000-entry update log, size-ceiling rejection,
not-found), and the hook wiring — two-client sync, survival across a server
restart, and the size-ceiling stateless notification — by an integration
test using an in-memory fake. The collab package gains its own vitest setup
that provisions an isolated `_collab` test database.

The REST `PUT /pages/:id/state` write path stays in place for now and is
retired (410) together with switching the editor to live collaboration in
#36, so the deployed editor is never left unable to save between the two
deploys.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PGdhRiwU1WRL4XxJfZYipY

apps/collab/package.json

@@ -17,14 +17,17 @@
     "@dorfteich/shared": "workspace:*",
     "@hocuspocus/server": "^4.3.0",
     "pg": "^8.16.0",
-    "pino": "^9.6.0"
+    "pino": "^9.6.0",
+    "prosemirror-model": "^1.25.9",
+    "y-prosemirror": "^1.3.7",
+    "yjs": "^13.6.31"
   },
   "devDependencies": {
     "@hocuspocus/provider": "^4.3.0",
     "@types/node": "^26.1.0",
     "@types/pg": "^8.11.0",
+    "prisma": "^6.3.0",
     "tsx": "^4.19.0",
-    "vitest": "^3.0.0",
-    "yjs": "^13.6.0"
+    "vitest": "^3.0.0"
   }
 }

apps/collab/src/auth.test.ts

@@ -7,6 +7,7 @@ import * as Y from 'yjs';
 
 import { createCollabServer } from './server.js';
 import { freePort } from './testing/free-port.js';
+import { InMemoryPagePersistence } from './testing/fake-persistence.js';
 
 const secret = 'integration-test-secret-32-chars!!';
 const logger = pino({ enabled: false });
@@ -31,6 +32,7 @@ describe('collab authentication', () => {
       logger,
       tokenSecret: secret,
       pingDatabase: async () => ({ ok: true }),
+      persistence: new InMemoryPagePersistence(),
     });
     const port = await freePort();
     await server.listen(port);

apps/collab/src/index.ts

@@ -2,6 +2,7 @@ import { collabEnvSchema, parseEnv } from '@dorfteich/shared';
 
 import { createPool, pingDatabase } from './db.js';
 import { createLogger } from './logger.js';
+import { PostgresPagePersistence } from './persistence.js';
 import { createCollabServer } from './server.js';
 
 /**
@@ -20,6 +21,7 @@ async function bootstrap(): Promise<void> {
     logger,
     tokenSecret: env.COLLAB_TOKEN_SECRET,
     pingDatabase: () => pingDatabase(pool),
+    persistence: new PostgresPagePersistence(pool),
   });
 
   await server.listen(env.PORT);

apps/collab/src/persistence.db.test.ts

@@ -0,0 +1,152 @@
+import { randomUUID } from 'node:crypto';
+
+import { editorSchema } from '@dorfteich/shared';
+import { Pool } from 'pg';
+import { prosemirrorJSONToYXmlFragment } from 'y-prosemirror';
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+import * as Y from 'yjs';
+
+import { PostgresPagePersistence } from './persistence.js';
+import { deriveContentFromDoc } from './yjs-content.js';
+import { collabTestDatabaseUrlOrUndefined } from './testing/test-db.js';
+
+const url = collabTestDatabaseUrlOrUndefined;
+
+/** A Y.Doc whose "default" XmlFragment holds a single paragraph of `text`. */
+function makeDoc(text: string): Y.Doc {
+  const doc = new Y.Doc();
+  const paragraph = editorSchema.node('paragraph', null, text ? [editorSchema.text(text)] : []);
+  const pmDoc = editorSchema.node('doc', null, [paragraph]);
+  prosemirrorJSONToYXmlFragment(editorSchema, pmDoc.toJSON(), doc.getXmlFragment('default'));
+  return doc;
+}
+
+/**
+ * A minimal valid Yjs state with no document content. Seeding pages with an
+ * empty state (rather than an empty paragraph) keeps these SQL round-trip tests
+ * from merging two independent doc lineages — a test artifact, not something the
+ * real lifecycle produces, where the editor always edits the loaded document.
+ */
+function emptyState(): Buffer {
+  const doc = new Y.Doc();
+  const update = Buffer.from(Y.encodeStateAsUpdate(doc));
+  doc.destroy();
+  return update;
+}
+
+describe.skipIf(!url)('PostgresPagePersistence (DB-backed)', () => {
+  let pool: Pool;
+  const userId = randomUUID();
+  const pondId = randomUUID();
+  const createdPageIds: string[] = [];
+
+  async function createPage(): Promise<string> {
+    const id = randomUUID();
+    await pool.query(
+      `INSERT INTO pages (id, pond_id, title, slug, ydoc_state, sort_key, created_by, updated_at)
+       VALUES ($1, $2, 'Test', $3, $4, 'a0', $5, now())`,
+      [id, pondId, `p-${id.slice(0, 8)}`, emptyState(), userId],
+    );
+    createdPageIds.push(id);
+    return id;
+  }
+
+  beforeAll(async () => {
+    pool = new Pool({ connectionString: url });
+    await pool.query(
+      'INSERT INTO users (id, username, email, display_name) VALUES ($1, $2, $3, $4)',
+      [userId, `collab-${userId.slice(0, 8)}`, `${userId}@example.test`, 'Collab Tester'],
+    );
+    await pool.query(
+      `INSERT INTO ponds (id, slug, name, type, owner_id, updated_at)
+       VALUES ($1, $2, 'Collab Pond', 'PERSONAL', $3, now())`,
+      [pondId, `collab-pond-${pondId.slice(0, 8)}`, userId],
+    );
+  });
+
+  afterAll(async () => {
+    if (createdPageIds.length > 0) {
+      // page_updates and page_content_cache cascade on page delete.
+      await pool.query('DELETE FROM pages WHERE id = ANY($1::text[])', [createdPageIds]);
+    }
+    await pool.query('DELETE FROM ponds WHERE id = $1', [pondId]);
+    await pool.query('DELETE FROM users WHERE id = $1', [userId]);
+    await pool.end();
+  });
+
+  it('stores a document and reloads identical content, refreshing the cache', async () => {
+    const pageId = await createPage();
+    const persistence = new PostgresPagePersistence(pool);
+
+    const doc = makeDoc('Hello from the DB test');
+    const result = await persistence.store(pageId, doc);
+    expect(result.outcome).toBe('stored');
+
+    const cache = await pool.query<{ markdown: string; plain_text: string }>(
+      'SELECT markdown, plain_text FROM page_content_cache WHERE page_id = $1',
+      [pageId],
+    );
+    expect(cache.rows[0]?.markdown).toContain('Hello from the DB test');
+    expect(cache.rows[0]?.plain_text).toContain('Hello from the DB test');
+
+    const reloaded = new Y.Doc();
+    const loaded = await persistence.loadInto(pageId, reloaded);
+    expect(loaded).toBe(true);
+    expect(deriveContentFromDoc(reloaded).markdown).toBe(deriveContentFromDoc(doc).markdown);
+  });
+
+  it('reconstructs a document from a long update log (1000 entries)', async () => {
+    const pageId = await createPage();
+
+    // Build 1000 independent Yjs updates, each a single character inserted into
+    // a scratch text type, then bulk-insert them as the page's update log.
+    const master = new Y.Doc();
+    let vector = Y.encodeStateVector(master);
+    const seqs: number[] = [];
+    const updates: Buffer[] = [];
+    for (let i = 0; i < 1000; i += 1) {
+      master.getText('log').insert(i, 'x');
+      seqs.push(i);
+      updates.push(Buffer.from(Y.encodeStateAsUpdate(master, vector)));
+      vector = Y.encodeStateVector(master);
+    }
+    await pool.query(
+      `INSERT INTO page_updates (id, page_id, seq, update)
+       SELECT gen_random_uuid(), $1, s, u FROM unnest($2::int[], $3::bytea[]) AS t(s, u)`,
+      [pageId, seqs, updates],
+    );
+
+    const persistence = new PostgresPagePersistence(pool);
+    const reloaded = new Y.Doc();
+    const loaded = await persistence.loadInto(pageId, reloaded);
+    expect(loaded).toBe(true);
+    expect(reloaded.getText('log').toString()).toBe('x'.repeat(1000));
+  });
+
+  it('rejects an oversize document without persisting anything', async () => {
+    const pageId = await createPage();
+    const persistence = new PostgresPagePersistence(pool);
+
+    const doc = makeDoc('');
+    doc.getText('bloat').insert(0, 'a'.repeat(6 * 1024 * 1024)); // > 5 MiB ceiling
+
+    const result = await persistence.store(pageId, doc);
+    expect(result.outcome).toBe('too_large');
+
+    const updates = await pool.query('SELECT 1 FROM page_updates WHERE page_id = $1', [pageId]);
+    expect(updates.rowCount).toBe(0);
+    const cache = await pool.query('SELECT 1 FROM page_content_cache WHERE page_id = $1', [pageId]);
+    expect(cache.rowCount).toBe(0);
+  });
+
+  it('reports not-found for a page that does not exist', async () => {
+    const persistence = new PostgresPagePersistence(pool);
+    const missingId = randomUUID();
+
+    const stored = await persistence.store(missingId, makeDoc('orphan'));
+    expect(stored.outcome).toBe('not_found');
+
+    const loaded = await persistence.loadInto(missingId, new Y.Doc());
+    expect(loaded).toBe(false);
+  });
+});

apps/collab/src/persistence.test.ts

@@ -0,0 +1,108 @@
+import { HocuspocusProvider } from '@hocuspocus/provider';
+import { signCollabToken } from '@dorfteich/shared/token-crypto';
+import { pino } from 'pino';
+import { afterEach, describe, expect, it } from 'vitest';
+import * as Y from 'yjs';
+
+import { createCollabServer } from './server.js';
+import { freePort } from './testing/free-port.js';
+import { InMemoryPagePersistence } from './testing/fake-persistence.js';
+
+const secret = 'persistence-test-secret-32-chars!!';
+const logger = pino({ enabled: false });
+
+/** Poll `predicate` until it is true or the timeout elapses. */
+async function waitFor(predicate: () => boolean, timeoutMs = 5000): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (predicate()) return;
+    await new Promise((resolve) => setTimeout(resolve, 20));
+  }
+  throw new Error('timed out waiting for condition');
+}
+
+describe('collab persistence hooks', () => {
+  const cleanups: Array<() => Promise<void> | void> = [];
+
+  afterEach(async () => {
+    // Tear down in reverse order (providers before their server).
+    for (const cleanup of cleanups.splice(0).reverse()) await cleanup();
+  });
+
+  async function startServer(persistence: InMemoryPagePersistence): Promise<string> {
+    const server = createCollabServer({
+      version: 'test',
+      logger,
+      tokenSecret: secret,
+      pingDatabase: async () => ({ ok: true }),
+      persistence,
+    });
+    const port = await freePort();
+    await server.listen(port);
+    cleanups.push(() => server.destroy());
+    return `ws://127.0.0.1:${port}`;
+  }
+
+  function connect(url: string, pageId: string): { provider: HocuspocusProvider; doc: Y.Doc } {
+    const doc = new Y.Doc();
+    const token = signCollabToken({ userId: `u-${pageId}`, pageId, mode: 'rw' }, secret, 60);
+    const provider = new HocuspocusProvider({ url, name: pageId, document: doc, token });
+    cleanups.push(() => provider.destroy());
+    return { provider, doc };
+  }
+
+  it('syncs edits between two read-write clients on the same page', async () => {
+    const persistence = new InMemoryPagePersistence();
+    const url = await startServer(persistence);
+    const pageId = '11111111-1111-1111-1111-111111111111';
+
+    const a = connect(url, pageId);
+    const b = connect(url, pageId);
+
+    a.doc.getText('t').insert(0, 'hello from A');
+
+    await waitFor(() => b.doc.getText('t').toString() === 'hello from A');
+    expect(b.doc.getText('t').toString()).toBe('hello from A');
+  });
+
+  it('persists edits so they survive a collab-server restart', async () => {
+    const persistence = new InMemoryPagePersistence();
+    const pageId = '22222222-2222-2222-2222-222222222222';
+
+    const url1 = await startServer(persistence);
+    const first = connect(url1, pageId);
+    first.doc.getText('t').insert(0, 'durable content');
+
+    // Disconnecting the only client triggers an immediate final store.
+    await waitFor(() => first.provider.isSynced);
+    await first.provider.destroy();
+    await waitFor(() => persistence.storeCalls > 0);
+
+    // A brand-new server instance reusing the same persistence must reload it.
+    const url2 = await startServer(persistence);
+    const second = connect(url2, pageId);
+    await waitFor(() => second.doc.getText('t').toString() === 'durable content');
+    expect(second.doc.getText('t').toString()).toBe('durable content');
+  });
+
+  it('rejects an oversize document and notifies the client via a stateless error', async () => {
+    const persistence = new InMemoryPagePersistence();
+    persistence.sizeLimit = 10; // any real edit exceeds this
+    const url = await startServer(persistence);
+    const pageId = '33333333-3333-3333-3333-333333333333';
+
+    const { provider, doc } = connect(url, pageId);
+    let statelessPayload: string | undefined;
+    provider.on('stateless', ({ payload }: { payload: string }) => {
+      statelessPayload = payload;
+    });
+
+    doc.getText('t').insert(0, 'this document is over the ceiling');
+
+    await waitFor(() => statelessPayload !== undefined);
+    expect(JSON.parse(statelessPayload!)).toMatchObject({
+      type: 'error',
+      code: 'page_document_too_large',
+    });
+  });
+});

apps/collab/src/persistence.ts

@@ -0,0 +1,182 @@
+import { MAX_PAGE_DOCUMENT_BYTES } from '@dorfteich/shared';
+import type { Pool } from 'pg';
+import * as Y from 'yjs';
+
+import { deriveContentFromDoc } from './yjs-content.js';
+
+/** Outcome of a store attempt, surfaced to the caller for logging/notification. */
+export interface StoreResult {
+  outcome: 'stored' | 'too_large' | 'not_found';
+  /** Size of the full merged state in bytes (for logging and the size ceiling). */
+  bytes: number;
+  /** Wall-clock duration of the flush, in milliseconds (ADR: measure flushes). */
+  durationMs: number;
+  /** Whether this flush merged the update log back into `pages.ydoc_state`. */
+  merged: boolean;
+}
+
+/**
+ * The persistence port the collab hooks depend on (ADR 0003). Kept as an
+ * interface so the server hooks can be integration-tested with an in-memory
+ * fake, while the Postgres implementation is exercised by a DB-backed test.
+ */
+export interface PagePersistence {
+  /**
+   * Apply the persisted state of `pageId` into `doc`. Returns `false` when the
+   * page does not exist or is trashed — the caller then keeps the empty doc.
+   */
+  loadInto(pageId: string, doc: Y.Doc): Promise<boolean>;
+  /** Persist the current `doc`, enforcing the document size ceiling. */
+  store(pageId: string, doc: Y.Doc): Promise<StoreResult>;
+  /** Release per-document bookkeeping when Hocuspocus unloads the document. */
+  forget(pageId: string): void;
+}
+
+/**
+ * When the append log for an open document grows past this many rows, a store
+ * flush merges it back into `pages.ydoc_state` and truncates the log. This
+ * bounds the cost of `loadInto` for long-lived sessions; the heavier,
+ * session-aware compaction of idle pages is the separate maintenance job (#40,
+ * default threshold 500 in realtime-collaboration.md).
+ */
+const INLINE_MERGE_THRESHOLD = 200;
+
+export class PostgresPagePersistence implements PagePersistence {
+  /**
+   * The Yjs state vector last persisted for each open document, so each store
+   * appends only the delta since the previous flush. Seeded by `loadInto` and
+   * cleared by `forget`; a missing entry safely falls back to storing the full
+   * state as the delta.
+   */
+  private readonly lastStoredVector = new Map<string, Uint8Array>();
+
+  constructor(private readonly pool: Pool) {}
+
+  async loadInto(pageId: string, doc: Y.Doc): Promise<boolean> {
+    const pageRow = await this.pool.query<{ ydoc_state: Buffer }>(
+      'SELECT ydoc_state FROM pages WHERE id = $1 AND deleted_at IS NULL',
+      [pageId],
+    );
+    const stored = pageRow.rows[0];
+    if (!stored) return false;
+
+    Y.applyUpdate(doc, new Uint8Array(stored.ydoc_state));
+
+    const updates = await this.pool.query<{ update: Buffer }>(
+      'SELECT update FROM page_updates WHERE page_id = $1 ORDER BY seq ASC',
+      [pageId],
+    );
+    for (const row of updates.rows) {
+      Y.applyUpdate(doc, new Uint8Array(row.update));
+    }
+
+    this.lastStoredVector.set(pageId, Y.encodeStateVector(doc));
+    return true;
+  }
+
+  async store(pageId: string, doc: Y.Doc): Promise<StoreResult> {
+    const start = performance.now();
+    // Capture everything from the live doc synchronously, before any await, so
+    // concurrent inbound updates cannot change what this flush persists.
+    const full = Y.encodeStateAsUpdate(doc);
+    const durationOf = (): number => performance.now() - start;
+
+    if (full.byteLength > MAX_PAGE_DOCUMENT_BYTES) {
+      return {
+        outcome: 'too_large',
+        bytes: full.byteLength,
+        durationMs: durationOf(),
+        merged: false,
+      };
+    }
+
+    const previousVector = this.lastStoredVector.get(pageId);
+    const delta = previousVector ? Y.encodeStateAsUpdate(doc, previousVector) : full;
+    const derived = deriveContentFromDoc(doc);
+    const nextVector = Y.encodeStateVector(doc);
+
+    const client = await this.pool.connect();
+    try {
+      await client.query('BEGIN');
+
+      // Lock the page row for the duration of the flush: this serialises seq
+      // allocation and guards against storing to a page trashed mid-session.
+      const page = await client.query<{ pond_id: string }>(
+        'SELECT pond_id FROM pages WHERE id = $1 AND deleted_at IS NULL FOR UPDATE',
+        [pageId],
+      );
+      const pageMeta = page.rows[0];
+      if (!pageMeta) {
+        await client.query('ROLLBACK');
+        return {
+          outcome: 'not_found',
+          bytes: full.byteLength,
+          durationMs: durationOf(),
+          merged: false,
+        };
+      }
+      const pondId = pageMeta.pond_id;
+
+      const seqRow = await client.query<{ seq: number }>(
+        'SELECT COALESCE(MAX(seq) + 1, 0) AS seq FROM page_updates WHERE page_id = $1',
+        [pageId],
+      );
+      const seq = seqRow.rows[0]?.seq ?? 0;
+      await client.query(
+        'INSERT INTO page_updates (id, page_id, seq, update) VALUES (gen_random_uuid(), $1, $2, $3)',
+        [pageId, seq, Buffer.from(delta)],
+      );
+
+      const merged = seq + 1 >= INLINE_MERGE_THRESHOLD;
+      if (merged) {
+        await client.query('UPDATE pages SET ydoc_state = $2, updated_at = now() WHERE id = $1', [
+          pageId,
+          Buffer.from(full),
+        ]);
+        await client.query('DELETE FROM page_updates WHERE page_id = $1', [pageId]);
+      } else {
+        await client.query('UPDATE pages SET updated_at = now() WHERE id = $1', [pageId]);
+      }
+
+      await client.query(
+        `INSERT INTO page_content_cache (page_id, plain_text, markdown, html, outline, updated_at)
+         VALUES ($1, $2, $3, $4, $5::jsonb, now())
+         ON CONFLICT (page_id) DO UPDATE
+           SET plain_text = EXCLUDED.plain_text,
+               markdown = EXCLUDED.markdown,
+               html = EXCLUDED.html,
+               outline = EXCLUDED.outline,
+               updated_at = now()`,
+        [
+          pageId,
+          derived.plainText,
+          derived.markdown,
+          derived.html,
+          JSON.stringify(derived.outline),
+        ],
+      );
+
+      if (derived.imageFileIds.length > 0) {
+        // Keep Attachment.pageId pointed at the page embedding the file (#31),
+        // scoped to the pond so a client cannot claim another pond's file.
+        await client.query(
+          'UPDATE attachments SET page_id = $1 WHERE id = ANY($2::text[]) AND pond_id = $3',
+          [pageId, derived.imageFileIds, pondId],
+        );
+      }
+
+      await client.query('COMMIT');
+      this.lastStoredVector.set(pageId, nextVector);
+      return { outcome: 'stored', bytes: full.byteLength, durationMs: durationOf(), merged };
+    } catch (error) {
+      await client.query('ROLLBACK').catch(() => undefined);
+      throw error;
+    } finally {
+      client.release();
+    }
+  }
+
+  forget(pageId: string): void {
+    this.lastStoredVector.delete(pageId);
+  }
+}

apps/collab/src/server.test.ts

@@ -3,6 +3,7 @@ import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest';
 
 import { createCollabServer } from './server.js';
 import { freePort } from './testing/free-port.js';
+import { InMemoryPagePersistence } from './testing/fake-persistence.js';
 import type { CollabHealthReport, DatabaseProbe } from './health.js';
 
 // A silent logger; these tests assert behaviour, not log output.
@@ -20,6 +21,7 @@ describe('collab server', () => {
       logger,
       tokenSecret: 'server-test-secret-32-characters!',
       pingDatabase: probe,
+      persistence: new InMemoryPagePersistence(),
     });
     // Hocuspocus' listen(port) ignores a falsy port (0 → default 80), so bind
     // an explicit OS-assigned free port instead.

apps/collab/src/server.ts

@@ -1,8 +1,10 @@
 import { Server } from '@hocuspocus/server';
+import { MAX_PAGE_DOCUMENT_BYTES } from '@dorfteich/shared';
 import { verifyCollabToken } from '@dorfteich/shared/token-crypto';
 import type { Logger } from 'pino';
 
 import { buildHealthReport, isHealthRequest, type DatabaseProbe } from './health.js';
+import type { PagePersistence } from './persistence.js';
 
 /** Per-connection context returned by onAuthenticate and used by later hooks. */
 export interface CollabContext {
@@ -18,16 +20,29 @@ export interface CollabServerDeps {
   tokenSecret: string;
   /** Probe used by the `/healthz` endpoint. Injected so it is easy to test. */
   pingDatabase: () => Promise<DatabaseProbe>;
+  /** Loads and persists page documents against PostgreSQL (#35). */
+  persistence: PagePersistence;
 }
 
 /**
- * The Hocuspocus collaboration server (ADR 0003). This skeleton (issue #33)
- * wires structured connection logging and a `/healthz` endpoint; it does not
- * yet authenticate connections (issue #34) or persist documents (issue #35),
- * so any WebSocket handshake is currently accepted.
+ * Stateless message the server broadcasts when a document exceeds the size
+ * ceiling on store: the client surfaces this and reverts the offending edit.
+ */
+export interface CollabErrorMessage {
+  type: 'error';
+  code: 'page_document_too_large';
+  limitBytes: number;
+}
+
+/**
+ * The Hocuspocus collaboration server (ADR 0003). It authenticates every
+ * connection with the api-minted token (#34) and is the writer of page state:
+ * `onLoadDocument` reconstructs the document from PostgreSQL and
+ * `onStoreDocument` persists it debounced, refreshing the derived content
+ * cache (#35).
  */
 export function createCollabServer(deps: CollabServerDeps): Server {
-  const { version, logger, tokenSecret, pingDatabase } = deps;
+  const { version, logger, tokenSecret, pingDatabase, persistence } = deps;
 
   return new Server({
     name: 'dorfteich-collab',
@@ -36,6 +51,10 @@ export function createCollabServer(deps: CollabServerDeps): Server {
     // index.ts owns graceful shutdown (it also closes the db pool); don't let
     // Hocuspocus install its own signal handlers that would call process.exit.
     stopOnSignals: false,
+    // Debounced persistence (realtime-collaboration.md §lifecycle): store 2 s
+    // after the last change, and at least every 30 s under continuous editing.
+    debounce: 2000,
+    maxDebounce: 30000,
 
     /**
      * Authorize every connection with the token the api minted after its own
@@ -85,6 +104,65 @@ export function createCollabServer(deps: CollabServerDeps): Server {
       );
     },
 
+    /**
+     * Reconstruct the document from PostgreSQL (`ydoc_state` merged with the
+     * `page_updates` log) on first open. A page that no longer exists loads as
+     * an empty document; the token check already gates real access (#34).
+     */
+    async onLoadDocument({ documentName, document }) {
+      const loaded = await persistence.loadInto(documentName, document);
+      logger.debug(
+        { event: 'document.load', documentName, loaded },
+        loaded ? 'document loaded' : 'document not found, starting empty',
+      );
+      return document;
+    },
+
+    /**
+     * Persist the document (debounced by the config above) and refresh the
+     * derived `page_content_cache`. An oversize document is rejected without
+     * persisting and the offending clients are notified via a stateless
+     * message so they can revert (realtime-collaboration.md failure modes).
+     */
+    async onStoreDocument({ documentName, document }) {
+      const result = await persistence.store(documentName, document);
+      if (result.outcome === 'too_large') {
+        const message: CollabErrorMessage = {
+          type: 'error',
+          code: 'page_document_too_large',
+          limitBytes: MAX_PAGE_DOCUMENT_BYTES,
+        };
+        document.broadcastStateless(JSON.stringify(message));
+        logger.warn(
+          { event: 'document.store.rejected', documentName, bytes: result.bytes },
+          'document exceeds size ceiling; update not persisted',
+        );
+        return;
+      }
+      if (result.outcome === 'not_found') {
+        logger.warn(
+          { event: 'document.store.skipped', documentName },
+          'document no longer exists; update not persisted',
+        );
+        return;
+      }
+      logger.debug(
+        {
+          event: 'document.store',
+          documentName,
+          bytes: result.bytes,
+          durationMs: Math.round(result.durationMs),
+          merged: result.merged,
+        },
+        'document persisted',
+      );
+    },
+
+    /** Drop the per-document store bookkeeping once Hocuspocus unloads it. */
+    async afterUnloadDocument({ documentName }) {
+      persistence.forget(documentName);
+    },
+
     async onRequest({ request, response }) {
       if (!isHealthRequest(request.method, request.url)) {
         // Not a health probe: let Hocuspocus handle the request.

apps/collab/src/testing/fake-persistence.ts

@@ -0,0 +1,43 @@
+import * as Y from 'yjs';
+
+import type { PagePersistence, StoreResult } from '../persistence.js';
+
+/**
+ * In-memory {@link PagePersistence} for tests: it keeps the last merged state
+ * per page so a "server restart" (destroy + recreate around the same instance)
+ * still reloads content, without needing a database. The Postgres implementation
+ * is covered separately by `persistence.db.test.ts`.
+ */
+export class InMemoryPagePersistence implements PagePersistence {
+  private readonly states = new Map<string, Uint8Array>();
+  /** Page ids that should report as missing (to exercise the not-found path). */
+  readonly missing = new Set<string>();
+  /** Overridable size ceiling so a test can trip it without a 5 MiB document. */
+  sizeLimit = Number.POSITIVE_INFINITY;
+  /** store() call count, for assertions. */
+  storeCalls = 0;
+
+  async loadInto(pageId: string, doc: Y.Doc): Promise<boolean> {
+    if (this.missing.has(pageId)) return false;
+    const state = this.states.get(pageId);
+    if (state) Y.applyUpdate(doc, state);
+    return true;
+  }
+
+  async store(pageId: string, doc: Y.Doc): Promise<StoreResult> {
+    this.storeCalls += 1;
+    const full = Y.encodeStateAsUpdate(doc);
+    if (full.byteLength > this.sizeLimit) {
+      return { outcome: 'too_large', bytes: full.byteLength, durationMs: 0, merged: false };
+    }
+    if (this.missing.has(pageId)) {
+      return { outcome: 'not_found', bytes: full.byteLength, durationMs: 0, merged: false };
+    }
+    this.states.set(pageId, full);
+    return { outcome: 'stored', bytes: full.byteLength, durationMs: 0, merged: false };
+  }
+
+  forget(): void {
+    // Nothing to release for the in-memory fake.
+  }
+}

apps/collab/src/testing/test-db.ts

@@ -0,0 +1,17 @@
+/**
+ * The DB-backed collab tests run against their own database, derived from
+ * `TEST_DATABASE_URL` by suffixing the database name with `_collab`. This keeps
+ * them isolated from the api's DB tests, which share `TEST_DATABASE_URL` and run
+ * concurrently under `pnpm -r test` (both a schema-push race and a data race
+ * would otherwise be possible). The vitest global setup creates and migrates
+ * this database; the tests skip themselves when `TEST_DATABASE_URL` is absent.
+ */
+export function collabTestDatabaseUrl(base: string): string {
+  const url = new URL(base);
+  url.pathname = url.pathname.replace(/\/([^/]+)$/, (_full, db: string) => `/${db}_collab`);
+  return url.toString();
+}
+
+export const collabTestDatabaseUrlOrUndefined = process.env.TEST_DATABASE_URL
+  ? collabTestDatabaseUrl(process.env.TEST_DATABASE_URL)
+  : undefined;

apps/collab/src/yjs-content.ts

@@ -0,0 +1,69 @@
+import {
+  docToHtml,
+  docToMarkdown,
+  docToPlainText,
+  editorSchema,
+  extractOutline,
+  type OutlineEntry,
+} from '@dorfteich/shared';
+import { Node } from 'prosemirror-model';
+import { yXmlFragmentToProseMirrorRootNode } from 'y-prosemirror';
+import * as Y from 'yjs';
+
+/**
+ * The Yjs XmlFragment name the editor binds to — TipTap's collaboration
+ * extension defaults to "default" (#25). api (`apps/api/src/pages/yjs-content.ts`),
+ * web, and collab must all agree on this or Yjs states become unreadable across
+ * them. This file deliberately mirrors the api's derivation (the shared
+ * functions come from `@dorfteich/shared`, #24); the two were kept separate on
+ * purpose rather than abstracted prematurely (see the M3 handoff).
+ */
+const FRAGMENT_NAME = 'default';
+
+/** Thrown for state bytes that are not a well-formed Yjs update for this schema. */
+export class InvalidPageStateError extends Error {}
+
+function docFromDoc(ydoc: Y.Doc): Node {
+  try {
+    return yXmlFragmentToProseMirrorRootNode(ydoc.getXmlFragment(FRAGMENT_NAME), editorSchema);
+  } catch (error) {
+    throw new InvalidPageStateError(error instanceof Error ? error.message : 'invalid Yjs state');
+  }
+}
+
+export interface DerivedPageContent {
+  plainText: string;
+  markdown: string;
+  html: string;
+  outline: OutlineEntry[];
+  /** fileIds of every `image` node embedded in the document — keeps
+   * `Attachment.pageId` pointed at the page that embeds the file (issue #31),
+   * mirroring the api's REST save path. */
+  imageFileIds: string[];
+}
+
+function imageFileIdsOf(doc: Node): string[] {
+  const ids: string[] = [];
+  doc.descendants((node) => {
+    if (node.type.name === 'image' && typeof node.attrs.fileId === 'string') {
+      ids.push(node.attrs.fileId);
+    }
+  });
+  return ids;
+}
+
+/**
+ * Decode a live Yjs document into the derived representations stored in
+ * `page_content_cache` (issue #23/#35), using the shared editor schema (#24)
+ * so the cache matches exactly what the api derives from the same state.
+ */
+export function deriveContentFromDoc(ydoc: Y.Doc): DerivedPageContent {
+  const doc = docFromDoc(ydoc);
+  return {
+    plainText: docToPlainText(doc),
+    markdown: docToMarkdown(doc),
+    html: docToHtml(doc),
+    outline: extractOutline(doc),
+    imageFileIds: imageFileIdsOf(doc),
+  };
+}

apps/collab/vitest.config.ts

@@ -0,0 +1,11 @@
+import { defineConfig } from 'vitest/config';
+
+export default defineConfig({
+  test: {
+    environment: 'node',
+    globalSetup: './vitest.global-setup.ts',
+    // The DB-backed persistence suite shares one database; run files serially
+    // so they cannot race each other (mirrors the api's vitest config).
+    fileParallelism: false,
+  },
+});

apps/collab/vitest.global-setup.ts

@@ -0,0 +1,50 @@
+import { execFileSync } from 'node:child_process';
+import { createRequire } from 'node:module';
+import { dirname, resolve } from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+import { Client } from 'pg';
+
+import { collabTestDatabaseUrl } from './src/testing/test-db.js';
+
+const require = createRequire(import.meta.url);
+const here = dirname(fileURLToPath(import.meta.url));
+
+/**
+ * Prepare the isolated collab test database (see `test-db.ts`): create it if
+ * missing, then push the current Prisma schema (owned by the api package) into
+ * it. Runs once per test run; a no-op when `TEST_DATABASE_URL` is unset, in
+ * which case the DB-backed suite skips itself.
+ */
+export default async function globalSetup(): Promise<void> {
+  const base = process.env.TEST_DATABASE_URL;
+  if (!base) return;
+
+  const target = collabTestDatabaseUrl(base);
+  const dbName = decodeURIComponent(new URL(target).pathname.slice(1));
+
+  const admin = new Client({ connectionString: base });
+  await admin.connect();
+  try {
+    const exists = await admin.query('SELECT 1 FROM pg_database WHERE datname = $1', [dbName]);
+    if (exists.rowCount === 0) {
+      // Identifier can't be parameterised; dbName is derived from our own env.
+      await admin.query(`CREATE DATABASE "${dbName}"`);
+    }
+  } finally {
+    await admin.end();
+  }
+
+  execFileSync(
+    process.execPath,
+    [
+      require.resolve('prisma/build/index.js'),
+      'db',
+      'push',
+      '--skip-generate',
+      '--schema',
+      resolve(here, '../api/prisma/schema.prisma'),
+    ],
+    { env: { ...process.env, DATABASE_URL: target }, stdio: 'inherit', cwd: here },
+  );
+}

pnpm-lock.yaml

@@ -156,6 +156,15 @@ importers:
       pino:
         specifier: ^9.6.0
         version: 9.14.0
+      prosemirror-model:
+        specifier: ^1.25.9
+        version: 1.25.9
+      y-prosemirror:
+        specifier: ^1.3.7
+        version: 1.3.7(prosemirror-model@1.25.9)(prosemirror-state@1.4.4)(prosemirror-view@1.42.0)(y-protocols@1.0.7(yjs@13.6.31))(yjs@13.6.31)
+      yjs:
+        specifier: ^13.6.31
+        version: 13.6.31
     devDependencies:
       '@hocuspocus/provider':
         specifier: ^4.3.0
@@ -166,15 +175,15 @@ importers:
       '@types/pg':
         specifier: ^8.11.0
         version: 8.20.0
+      prisma:
+        specifier: ^6.3.0
+        version: 6.19.3(typescript@5.9.3)
       tsx:
         specifier: ^4.19.0
         version: 4.23.0
       vitest:
         specifier: ^3.0.0
         version: 3.2.6(@types/node@26.1.0)(jiti@2.7.0)(jsdom@26.1.0)(terser@5.48.0)(tsx@4.23.0)
-      yjs:
-        specifier: ^13.6.0
-        version: 13.6.31
 
   apps/web:
     dependencies: