import { Server, type Hocuspocus } 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'; import type { SessionRegistry } from './session-registry.js'; import type { VersionStore } from './version-store.js'; /** Per-connection context returned by onAuthenticate and used by later hooks. */ export interface CollabContext { /** `null` for an anonymous visitor on a public page (issue #53). */ userId: string | null; mode: 'rw' | 'ro'; } export interface CollabServerDeps { /** Version string surfaced in health responses (image build arg). */ version: string; logger: Logger; /** Secret shared with the api that signs the collaboration tokens (#34). */ 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; /** * Advertises which pages have a live session so the compaction job skips * them (#40). Optional: a server without it simply records no sessions, * which keeps the unit/integration server tests free of a DB dependency. */ sessionRegistry?: SessionRegistry; /** * Creates automatic version snapshots and tracks contributors (#41). * Optional for the same reason as `sessionRegistry`. */ versionStore?: VersionStore; /** * Maintenance mode during an in-app restore (issue #103): while true, new * connections are refused so no session comes up mid-restore. Optional — * servers without the listener (tests) never enter maintenance. */ isMaintenanceActive?: () => boolean; } /** * 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; } /** Hocuspocus' "Reset Connection" WebSocket close code — the client reconnects. */ const RESET_CONNECTION_CODE = 4205; /** * Force every live session on a document to re-validate access *now* (issue * #53). Hocuspocus' own `closeConnections` only sends an application-level * close message: the client detaches the document but keeps the socket open and * re-checks access lazily, after its ~30s message timeout. This closes the * underlying WebSocket instead, so the client reconnects and re-authenticates * with a freshly-minted token within seconds — turning a downgraded editor * read-only, or dropping a reader whose access was revoked, right away * (permissions.md §Performance: "revoking write access closes live sessions"). */ export function closeDocumentConnections(hocuspocus: Hocuspocus, documentName: string): void { const document = hocuspocus.documents.get(documentName); if (!document) return; for (const connection of document.getConnections()) { connection.webSocket.close(RESET_CONNECTION_CODE, 'Reset Connection'); } } /** * 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, persistence, sessionRegistry, versionStore } = deps; return new Server({ name: 'dorfteich-collab', // Suppress Hocuspocus' ASCII start screen; we emit our own pino logs. quiet: true, // 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 * permission check (#34, ADR 0003). Rejecting (throwing) closes the socket. * The document name is the page id, so a token issued for one page cannot * open another. A `ro` token connects but its inbound document updates are * dropped server-side via Hocuspocus' read-only connection flag. */ async onAuthenticate({ documentName, token, connectionConfig }): Promise { if (deps.isMaintenanceActive?.()) { logger.info( { event: 'auth.rejected', documentName, reason: 'maintenance' }, 'collab authentication rejected', ); throw new Error('maintenance'); } const result = await verifyCollabToken(token, tokenSecret); if (!result.valid) { logger.info( { event: 'auth.rejected', documentName, reason: result.reason }, 'collab authentication rejected', ); throw new Error('unauthorized'); } if (result.claims.pageId !== documentName) { logger.info( { event: 'auth.rejected', documentName, reason: 'page_mismatch' }, 'collab authentication rejected', ); throw new Error('unauthorized'); } if (result.claims.mode === 'ro') { connectionConfig.readOnly = true; } logger.debug( { event: 'auth.ok', documentName, userId: result.claims.userId, mode: result.claims.mode }, 'collab connection authenticated', ); return { userId: result.claims.userId, mode: result.claims.mode }; }, async onConnect({ documentName, socketId }) { logger.info( { event: 'connection.open', documentName, socketId }, 'collaboration connection opened', ); }, /** Attribute each change to the editing user for version contributor sets (#41). */ async onChange({ documentName, context }) { if (context?.userId) versionStore?.recordContributor(documentName, context.userId); }, async onDisconnect({ documentName, socketId, clientsCount, document }) { logger.info( { event: 'connection.close', documentName, socketId, clientsCount }, 'collaboration connection closed', ); // The editing session ended: snapshot a version if anything changed (#41). if (clientsCount === 0) { await versionStore?.onSessionEnd(documentName, document); } }, /** * 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); // Advertise the open session so the compaction job skips this page (#40). sessionRegistry?.markOpen(documentName); // Start the active-editing interval clock for automatic versions (#41). versionStore?.noteOpened(documentName); 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', ); // Flush contributors and take an interval snapshot if one is due (#41). await versionStore?.onStore(documentName, document); }, /** Drop the per-document store bookkeeping once Hocuspocus unloads it. */ async afterUnloadDocument({ documentName }) { persistence.forget(documentName); // The session ended; let the compaction job consider this page again (#40). sessionRegistry?.markClosed(documentName); versionStore?.forget(documentName); }, async onRequest({ request, response }) { if (!isHealthRequest(request.method, request.url)) { // Not a health probe: let Hocuspocus handle the request. return; } const database = await pingDatabase(); const { httpStatus, body } = buildHealthReport(version, database); if (!database.ok) { logger.warn( { event: 'health.failed', detail: database.detail }, 'health check failed: database unreachable', ); } response.writeHead(httpStatus, { 'Content-Type': 'application/json' }); response.end(JSON.stringify(body)); // Hocuspocus sends a default 200 unless the onRequest chain rejects with // a falsy reason; the response is already written, so stop the chain. return Promise.reject(); }, }); }