feat: F5.1 server F keeps the shared documents

A new package, packages/server: Hocuspocus 3.4.4 on /sync, a project a
document, refused with a reason when the store has not got it; every update
stored in Postgres as it comes and compacted after 500 updates, a minute of
quiet and when the last person leaves (FOR UPDATE on the row written when
the project is made); the janitor runs only here, at loading and 250 ms
after a change that is not its own; readers change nothing; /healthz; JSON
log. On closing it takes no one new and closes the sockets too (Hocuspocus'
close is a message after which a client signs in again on the same socket).
The tests use real clients against a throwaway Postgres in Docker.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

packages/server/src/index.ts

@@ -0,0 +1,7 @@
+// Server F (plan §8): the shared documents, and later the simulation rooms,
+// the sign-in and the dashboard's service.
+export * from './log.js';
+export * from './persistence.js';
+export * from './server.js';
+export * from './store.js';
+export * from './sync.js';

packages/server/src/log.ts

@@ -0,0 +1,25 @@
+// The service's log: one JSON object a line, in English (it is for running
+// the service, not for its users; plan §8.2): the time, the level, what
+// happened, and the project, person and room it happened in.
+
+export type LogFields = Readonly<Record<string, unknown>>;
+
+export interface Log {
+  info(event: string, fields?: LogFields): void;
+  warn(event: string, fields?: LogFields): void;
+  error(event: string, fields?: LogFields): void;
+}
+
+export function jsonLog(write: (line: string) => void = (line) => process.stdout.write(`${line}\n`), now: () => Date = () => new Date()): Log {
+  const entry =
+    (level: string) =>
+    (event: string, fields: LogFields = {}) => {
+      write(JSON.stringify({ time: now().toISOString(), level, event, ...fields }));
+    };
+  return { info: entry('info'), warn: entry('warn'), error: entry('error') };
+}
+
+/** An error as the log writes it. */
+export function errorText(error: unknown): string {
+  return error instanceof Error ? error.message : String(error);
+}

packages/server/src/main.ts

@@ -0,0 +1,35 @@
+// Server F started from its environment: FILIUS_PORT (8080), DATABASE_URL.
+// Until the sign-in comes (F5.2) every connection is refused, unless
+// FILIUS_INSECURE_DEV=1 lets everyone write — on a developer's machine only.
+
+import { jsonLog } from './log.js';
+import { startServer } from './server.js';
+import { PostgresStore } from './store.js';
+
+const log = jsonLog();
+const url = process.env.DATABASE_URL;
+if (url === undefined) {
+  log.error('config.missing', { variable: 'DATABASE_URL' });
+  process.exit(1);
+}
+const insecure = process.env.FILIUS_INSECURE_DEV === '1';
+if (insecure) log.warn('auth.insecureDev', { note: 'every connection may write' });
+const store = PostgresStore.connect(url);
+await store.migrate();
+const server = await startServer({
+  port: Number(process.env.FILIUS_PORT ?? 8080),
+  store,
+  log,
+  authenticate: async () => {
+    if (insecure) return { readOnly: false };
+    throw new Error('no sign-in yet');
+  },
+});
+for (const signal of ['SIGTERM', 'SIGINT'] as const) {
+  process.once(signal, () => {
+    void server
+      .close()
+      .then(() => store.close())
+      .then(() => process.exit(0));
+  });
+}

packages/server/src/persistence.ts

@@ -0,0 +1,80 @@
+// One project's document kept while it is open: every update stored as it
+// comes, in order (a queue, so that a compaction never overtakes an update
+// before it), and compacted after COMPACT_AFTER_UPDATES updates or when the
+// service asks (a minute after the last change, and when the last person
+// leaves; plan §5.4). A compaction stores the document as it is and drops
+// the updates stored before it; one stored after it stays, and applying it
+// again on loading changes nothing (Yjs).
+
+import * as Y from 'yjs';
+import { errorText, type Log } from './log.js';
+import type { DocStore } from './store.js';
+
+export const COMPACT_AFTER_UPDATES = 500;
+/** The origin of what is loaded from the store: not stored again. */
+export const STORE_ORIGIN = 'store';
+
+export class DocumentPersistence {
+  private queue: Promise<void> = Promise.resolve();
+  private lastSeq = '0';
+  /** Updates stored since the last compaction (in the queue's order). */
+  private sinceCompaction = 0;
+  /** Updates queued since the last compaction was queued: what decides the next one. */
+  private queued = 0;
+
+  constructor(
+    private readonly store: DocStore,
+    private readonly project: string,
+    private readonly log: Log,
+  ) {}
+
+  /** The stored document into a Y.Doc; false without the project. */
+  async load(doc: Y.Doc): Promise<boolean> {
+    const stored = await this.store.load(this.project);
+    if (stored === undefined) return false;
+    Y.transact(
+      doc,
+      () => {
+        Y.applyUpdate(doc, stored.state);
+        for (const update of stored.updates) Y.applyUpdate(doc, update);
+      },
+      STORE_ORIGIN,
+    );
+    this.lastSeq = stored.lastSeq;
+    this.sinceCompaction = stored.updates.length;
+    this.queued = stored.updates.length;
+    return true;
+  }
+
+  private enqueue(work: () => Promise<void>): Promise<void> {
+    this.queue = this.queue.then(work).catch((error: unknown) => this.log.error('store.failed', { project: this.project, error: errorText(error) }));
+    return this.queue;
+  }
+
+  /** An update stored; the document compacted once enough have come. */
+  append(update: Uint8Array, doc: Y.Doc): Promise<void> {
+    const stored = this.enqueue(async () => {
+      this.lastSeq = await this.store.append(this.project, update);
+      this.sinceCompaction += 1;
+    });
+    this.queued += 1;
+    return this.queued >= COMPACT_AFTER_UPDATES ? this.compact(doc) : stored;
+  }
+
+  /** The document as it is now stored as the project's state, the updates before dropped. */
+  compact(doc: Y.Doc): Promise<void> {
+    this.queued = 0;
+    return this.enqueue(async () => {
+      if (this.sinceCompaction === 0) return;
+      const updates = this.sinceCompaction;
+      await this.store.compact(this.project, Y.encodeStateAsUpdate(doc), this.lastSeq);
+      this.sinceCompaction = 0;
+      this.log.info('doc.compacted', { project: this.project, updates });
+    });
+  }
+
+  /** Everything stored that was asked for. */
+  flush(): Promise<void> {
+    return this.enqueue(async () => {});
+  }
+}

packages/server/src/server.ts

@@ -0,0 +1,74 @@
+// Server F's front door (plan §8.2): one HTTP server — /healthz for the
+// proxy, the WebSocket /sync for the shared documents; anything else is not
+// found. Closed, it first takes no one new (a client lets go and comes back
+// at once, which would keep its document open), then lets everyone go,
+// waits until every document is stored and unloaded, and stops.
+
+import { createServer, type Server } from 'node:http';
+import type { AddressInfo, Socket } from 'node:net';
+import type { Hocuspocus } from '@hocuspocus/server';
+import { WebSocketServer } from 'ws';
+import type { Log } from './log.js';
+import { createSync, type SyncOptions } from './sync.js';
+
+export interface FiliusServer {
+  readonly port: number;
+  readonly sync: Hocuspocus;
+  close(): Promise<void>;
+}
+
+export interface ServerOptions extends SyncOptions {
+  readonly port?: number;
+  readonly host?: string;
+  readonly log: Log;
+}
+
+function refuse(socket: Socket, status: string): void {
+  socket.end(`HTTP/1.1 ${status}\r\nConnection: close\r\n\r\n`);
+}
+
+export async function startServer(options: ServerOptions): Promise<FiliusServer> {
+  const sync = createSync(options);
+  const sockets = new WebSocketServer({ noServer: true });
+  let closing = false;
+  const http: Server = createServer((request, response) => {
+    const path = new URL(request.url ?? '/', 'http://filius').pathname;
+    if (request.method === 'GET' && path === '/healthz') {
+      response.writeHead(200, { 'content-type': 'application/json' });
+      response.end(JSON.stringify({ status: 'ok', documents: sync.getDocumentsCount(), connections: sync.getConnectionsCount() }));
+      return;
+    }
+    response.writeHead(404, { 'content-type': 'application/json' });
+    response.end(JSON.stringify({ error: 'notFound' }));
+  });
+  http.on('upgrade', (request, socket, head) => {
+    const path = new URL(request.url ?? '/', 'http://filius').pathname;
+    if (path !== '/sync') {
+      refuse(socket as Socket, '404 Not Found');
+      return;
+    }
+    if (closing) {
+      refuse(socket as Socket, '503 Service Unavailable');
+      return;
+    }
+    sockets.handleUpgrade(request, socket, head, (ws) => sync.handleConnection(ws, request));
+  });
+  await new Promise<void>((resolve) => http.listen(options.port ?? 0, options.host ?? '0.0.0.0', resolve));
+  const port = (http.address() as AddressInfo).port;
+  options.log.info('server.listening', { port });
+  return {
+    port,
+    sync,
+    async close() {
+      closing = true;
+      sync.closeConnections();
+      // Hocuspocus' close is a message on the socket, after which a client signs in again on it: the sockets go too.
+      for (const socket of sockets.clients) socket.close(1001, 'service stopping');
+      // Each document is stored and unloaded once its last person has gone.
+      for (let waited = 0; sync.getDocumentsCount() > 0 && waited < 10_000; waited += 50) await new Promise((resolve) => setTimeout(resolve, 50));
+      sockets.close();
+      await new Promise<void>((resolve) => http.close(() => resolve()));
+      options.log.info('server.closed', { documents: sync.getDocumentsCount() });
+    },
+  };
+}

packages/server/src/store.ts

@@ -0,0 +1,109 @@
+// Where F keeps each project's document (plan §5.4): its compacted state in
+// filius_doc, the updates since then in filius_doc_update, in the order they
+// came. A project's row is written when the project is made, so that the
+// compaction's FOR UPDATE has a row to lock (a lock on a missing row locks
+// nothing; Scratch's lesson 6.13).
+
+import pg from 'pg';
+
+export interface StoredDocument {
+  readonly state: Uint8Array;
+  readonly updates: readonly Uint8Array[];
+  /** The last update's sequence number, "0" without one. */
+  readonly lastSeq: string;
+  readonly rev: number;
+}
+
+export interface DocStore {
+  /** Makes the tables, if they are not there. */
+  migrate(): Promise<void>;
+  /** A new project with its first state; false if it is there already. */
+  create(project: string, state: Uint8Array): Promise<boolean>;
+  /** A project's state and the updates since, undefined without the project. */
+  load(project: string): Promise<StoredDocument | undefined>;
+  /** An update kept; its sequence number. */
+  append(project: string, update: Uint8Array): Promise<string>;
+  /** A new state, which holds every update up to `upToSeq`: those go. */
+  compact(project: string, state: Uint8Array, upToSeq: string): Promise<void>;
+  /** How many updates wait for the next compaction. */
+  pending(project: string): Promise<number>;
+  remove(project: string): Promise<void>;
+  close(): Promise<void>;
+}
+
+const SCHEMA = `
+CREATE TABLE IF NOT EXISTS filius_doc (
+  project_uuid uuid PRIMARY KEY,
+  state bytea NOT NULL,
+  rev integer NOT NULL DEFAULT 0,
+  updated_at timestamptz NOT NULL DEFAULT now()
+);
+CREATE TABLE IF NOT EXISTS filius_doc_update (
+  project_uuid uuid NOT NULL REFERENCES filius_doc ON DELETE CASCADE,
+  seq bigserial NOT NULL,
+  update bytea NOT NULL,
+  PRIMARY KEY (project_uuid, seq)
+);
+`;
+
+const bytes = (buffer: Buffer) => new Uint8Array(buffer.buffer, buffer.byteOffset, buffer.byteLength);
+
+export class PostgresStore implements DocStore {
+  constructor(private readonly pool: pg.Pool) {}
+
+  static connect(url: string): PostgresStore {
+    return new PostgresStore(new pg.Pool({ connectionString: url }));
+  }
+
+  async migrate(): Promise<void> {
+    await this.pool.query(SCHEMA);
+  }
+
+  async create(project: string, state: Uint8Array): Promise<boolean> {
+    const result = await this.pool.query('INSERT INTO filius_doc (project_uuid, state) VALUES ($1, $2) ON CONFLICT DO NOTHING', [project, Buffer.from(state)]);
+    return result.rowCount === 1;
+  }
+
+  async load(project: string): Promise<StoredDocument | undefined> {
+    const doc = await this.pool.query<{ state: Buffer; rev: number }>('SELECT state, rev FROM filius_doc WHERE project_uuid = $1', [project]);
+    const row = doc.rows[0];
+    if (row === undefined) return undefined;
+    const updates = await this.pool.query<{ seq: string; update: Buffer }>('SELECT seq, update FROM filius_doc_update WHERE project_uuid = $1 ORDER BY seq', [project]);
+    return { state: bytes(row.state), updates: updates.rows.map((each) => bytes(each.update)), lastSeq: updates.rows.at(-1)?.seq ?? '0', rev: row.rev };
+  }
+
+  async append(project: string, update: Uint8Array): Promise<string> {
+    const result = await this.pool.query<{ seq: string }>('INSERT INTO filius_doc_update (project_uuid, update) VALUES ($1, $2) RETURNING seq', [project, Buffer.from(update)]);
+    return result.rows[0]!.seq;
+  }
+
+  async compact(project: string, state: Uint8Array, upToSeq: string): Promise<void> {
+    const client = await this.pool.connect();
+    try {
+      await client.query('BEGIN');
+      const locked = await client.query('SELECT 1 FROM filius_doc WHERE project_uuid = $1 FOR UPDATE', [project]);
+      if (locked.rowCount !== 1) throw new Error(`no document for project ${project}`);
+      await client.query('UPDATE filius_doc SET state = $2, rev = rev + 1, updated_at = now() WHERE project_uuid = $1', [project, Buffer.from(state)]);
+      await client.query('DELETE FROM filius_doc_update WHERE project_uuid = $1 AND seq <= $2', [project, upToSeq]);
+      await client.query('COMMIT');
+    } catch (error) {
+      await client.query('ROLLBACK');
+      throw error;
+    } finally {
+      client.release();
+    }
+  }
+
+  async pending(project: string): Promise<number> {
+    const result = await this.pool.query<{ count: string }>('SELECT count(*) FROM filius_doc_update WHERE project_uuid = $1', [project]);
+    return Number(result.rows[0]!.count);
+  }
+
+  async remove(project: string): Promise<void> {
+    await this.pool.query('DELETE FROM filius_doc WHERE project_uuid = $1', [project]);
+  }
+
+  async close(): Promise<void> {
+    await this.pool.end();
+  }
+}

packages/server/src/sync.ts

@@ -0,0 +1,113 @@
+// The shared documents (/sync, plan §5): Hocuspocus, a project a document.
+// A project is loaded from the store when its first person comes (a project
+// the store has not got is refused, with its reason: projects are made
+// through the service, F5.2), every
+// update stored as it comes, compacted a minute after the last change, when
+// the last person leaves and after COMPACT_AFTER_UPDATES updates. The
+// janitor (plan §5.3) runs only here, as a client of its own: once when the
+// document is loaded, then 250 ms after the last update that was not its
+// own; its repair goes to everyone like any update.
+
+import { Hocuspocus, type onAuthenticatePayload } from '@hocuspocus/server';
+import type * as Y from 'yjs';
+import { JANITOR_ORIGIN, repair } from '@mdg-filius/doc';
+import { errorText, type Log } from './log.js';
+import { DocumentPersistence } from './persistence.js';
+import type { DocStore } from './store.js';
+
+/** How long after the last change a document is compacted (Hocuspocus' debounce of onStoreDocument). */
+export const COMPACT_AFTER_IDLE_MILLISECONDS = 60_000;
+/** Compacted at the latest so long after the first change, even while changes keep coming. */
+export const COMPACT_AT_LATEST_MILLISECONDS = 600_000;
+/** The janitor's wait after an update (entprellt). */
+export const JANITOR_DELAY_MILLISECONDS = 250;
+
+/** What a connection may do; thrown, it is refused. */
+export interface Access {
+  readonly readOnly: boolean;
+  /** Who, for the log. */
+  readonly user?: string;
+}
+
+export interface SyncOptions {
+  readonly store: DocStore;
+  readonly log: Log;
+  readonly authenticate: (payload: onAuthenticatePayload) => Promise<Access>;
+  readonly compactAfterIdleMilliseconds?: number;
+}
+
+/** A refusal of a connection, with the reason the person's editor is told (Hocuspocus sends an error's `reason`). */
+export function refusal(reason: string): Error & { readonly reason: string } {
+  return Object.assign(new Error(reason), { reason });
+}
+
+/** A project's id: a UUID, as the dashboard makes them. */
+export const isProjectId = (name: string) => /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/.test(name);
+
+/** The janitor on a document; the function that ends it. */
+function startJanitor(doc: Y.Doc, project: string, log: Log): () => void {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const run = () => {
+    timer = undefined;
+    const repairs = repair(doc);
+    const count = repairs.danglingLinks.length + repairs.sharedPortLinks.length + repairs.madeFolders.length;
+    if (count > 0) log.info('janitor.repaired', { project, danglingLinks: repairs.danglingLinks.length, sharedPortLinks: repairs.sharedPortLinks.length, madeFolders: repairs.madeFolders.length });
+  };
+  const onUpdate = (_update: Uint8Array, origin: unknown) => {
+    if (origin === JANITOR_ORIGIN) return;
+    if (timer !== undefined) clearTimeout(timer);
+    timer = setTimeout(run, JANITOR_DELAY_MILLISECONDS);
+  };
+  run();
+  doc.on('update', onUpdate);
+  return () => {
+    if (timer !== undefined) clearTimeout(timer);
+    doc.off('update', onUpdate);
+  };
+}
+
+export function createSync({ store, log, authenticate, compactAfterIdleMilliseconds = COMPACT_AFTER_IDLE_MILLISECONDS }: SyncOptions): Hocuspocus {
+  const open = new Map<string, { readonly persistence: DocumentPersistence; stopJanitor?: () => void }>();
+  return new Hocuspocus({
+    quiet: true,
+    debounce: compactAfterIdleMilliseconds,
+    maxDebounce: COMPACT_AT_LATEST_MILLISECONDS,
+    async onAuthenticate(payload) {
+      if (!isProjectId(payload.documentName)) throw refusal('notAProject');
+      const access = await authenticate(payload);
+      payload.connectionConfig.readOnly = access.readOnly;
+      return access;
+    },
+    async onLoadDocument({ documentName, document }) {
+      const persistence = new DocumentPersistence(store, documentName, log);
+      if (!(await persistence.load(document))) {
+        log.warn('doc.unknown', { project: documentName });
+        throw refusal('unknownProject');
+      }
+      open.set(documentName, { persistence });
+      log.info('doc.loaded', { project: documentName });
+    },
+    async afterLoadDocument({ documentName, document }) {
+      const entry = open.get(documentName);
+      if (entry !== undefined) entry.stopJanitor = startJanitor(document, documentName, log);
+    },
+    async onChange({ documentName, document, update }) {
+      await open.get(documentName)?.persistence.append(update, document);
+    },
+    async onStoreDocument({ documentName, document }) {
+      await open.get(documentName)?.persistence.compact(document);
+    },
+    async afterUnloadDocument({ documentName }) {
+      const entry = open.get(documentName);
+      if (entry === undefined) return;
+      open.delete(documentName);
+      entry.stopJanitor?.();
+      try {
+        await entry.persistence.flush();
+      } catch (error) {
+        log.error('doc.unloadFailed', { project: documentName, error: errorText(error) });
+      }
+      log.info('doc.unloaded', { project: documentName });
+    },
+  });
+}