diff --git a/apps/collab/package.json b/apps/collab/package.json index 37fbb5b..f4f08f8 100644 --- a/apps/collab/package.json +++ b/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" } } diff --git a/apps/collab/src/auth.test.ts b/apps/collab/src/auth.test.ts index 0096666..9e047f0 100644 --- a/apps/collab/src/auth.test.ts +++ b/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); diff --git a/apps/collab/src/index.ts b/apps/collab/src/index.ts index 2f63504..59718e7 100644 --- a/apps/collab/src/index.ts +++ b/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 { logger, tokenSecret: env.COLLAB_TOKEN_SECRET, pingDatabase: () => pingDatabase(pool), + persistence: new PostgresPagePersistence(pool), }); await server.listen(env.PORT); diff --git a/apps/collab/src/persistence.db.test.ts b/apps/collab/src/persistence.db.test.ts new file mode 100644 index 0000000..a60bea0 --- /dev/null +++ b/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 { + 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); + }); +}); diff --git a/apps/collab/src/persistence.test.ts b/apps/collab/src/persistence.test.ts new file mode 100644 index 0000000..7a01f38 --- /dev/null +++ b/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 { + 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> = []; + + 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 { + 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', + }); + }); +}); diff --git a/apps/collab/src/persistence.ts b/apps/collab/src/persistence.ts new file mode 100644 index 0000000..4ec723d --- /dev/null +++ b/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; + /** Persist the current `doc`, enforcing the document size ceiling. */ + store(pageId: string, doc: Y.Doc): Promise; + /** 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(); + + constructor(private readonly pool: Pool) {} + + async loadInto(pageId: string, doc: Y.Doc): Promise { + 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 { + 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); + } +} diff --git a/apps/collab/src/server.test.ts b/apps/collab/src/server.test.ts index 3199375..e177ac4 100644 --- a/apps/collab/src/server.test.ts +++ b/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. diff --git a/apps/collab/src/server.ts b/apps/collab/src/server.ts index dfe134b..2feb17c 100644 --- a/apps/collab/src/server.ts +++ b/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; + /** 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. diff --git a/apps/collab/src/testing/fake-persistence.ts b/apps/collab/src/testing/fake-persistence.ts new file mode 100644 index 0000000..aacac6b --- /dev/null +++ b/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(); + /** Page ids that should report as missing (to exercise the not-found path). */ + readonly missing = new Set(); + /** 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 { + 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 { + 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. + } +} diff --git a/apps/collab/src/testing/test-db.ts b/apps/collab/src/testing/test-db.ts new file mode 100644 index 0000000..b838606 --- /dev/null +++ b/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; diff --git a/apps/collab/src/yjs-content.ts b/apps/collab/src/yjs-content.ts new file mode 100644 index 0000000..96dc9d7 --- /dev/null +++ b/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), + }; +} diff --git a/apps/collab/vitest.config.ts b/apps/collab/vitest.config.ts new file mode 100644 index 0000000..f84987b --- /dev/null +++ b/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, + }, +}); diff --git a/apps/collab/vitest.global-setup.ts b/apps/collab/vitest.global-setup.ts new file mode 100644 index 0000000..dbb9748 --- /dev/null +++ b/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 { + 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 }, + ); +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f828853..b46993d 100644 --- a/pnpm-lock.yaml +++ b/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: