dorfteich/apps/collab/src/server.ts

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<DatabaseProbe>;
  /** 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<CollabContext> {
      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();
    },
  });
}