feat: F5.5 large binary files as blobs on server F

From 16 KB a picture's or binary file's entry names its bytes by their
SHA-256 instead of holding them; the bytes are on F, content-addressed
(GET/POST /blobs with the session's grant, from the editor's origin, at
most 16 MiB). A document's blob side says where they are: the editor's
fetches and uploads, the room thread's asks F's main thread, and a start
waits for the ones asked for; without a side everything stays in the
document. Text stays Y.Text. A blob whose bytes are not there is never
written back empty. A blob F has not got yet is asked for again a few
times, and a running room reads late bytes.

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

packages/server/src/blobRoutes.ts

@@ -0,0 +1,72 @@
+// The editor's way to the blobs (blobs.ts): GET /blobs/<sha>?project=… for
+// whoever has a grant to the project, POST /blobs?project=… (the bytes as
+// the body) for whoever may write in it — not in a frozen project, and only
+// from the editor's origin (a page elsewhere under the same domain would
+// bring the person's cookie along). The answer to a POST is the name F gave
+// the bytes. For the editor's origins F answers CORS, so that a page served
+// from elsewhere (the development page) can reach it with its cookie.
+
+import type { IncomingMessage, ServerResponse } from 'node:http';
+import { BLOB_MAX_BYTES, type BlobStore } from './blobs.js';
+import type { Log } from './log.js';
+import type { SessionStore } from './sessions.js';
+import { cookieValue, SESSION_COOKIE } from './sync.js';
+import { isUuid } from './ticket.js';
+
+export interface BlobRouteOptions {
+  readonly blobs: BlobStore;
+  readonly sessions: SessionStore;
+  readonly log: Log;
+  readonly origins: readonly string[];
+}
+
+export async function handleBlobs(request: IncomingMessage, response: ServerResponse, url: URL, { blobs, sessions, log, origins }: BlobRouteOptions): Promise<void> {
+  const origin = request.headers.origin;
+  const allowed = origin !== undefined && origins.includes(origin);
+  const headers: Record<string, string> = allowed ? { 'access-control-allow-origin': origin, 'access-control-allow-credentials': 'true', vary: 'Origin' } : {};
+  const send = (status: number, value: object) => {
+    response.writeHead(status, { ...headers, 'content-type': 'application/json', 'cache-control': 'no-store' });
+    response.end(JSON.stringify(value));
+  };
+  if (request.method === 'OPTIONS') {
+    response.writeHead(allowed ? 204 : 403, { ...headers, 'access-control-allow-methods': 'GET, POST', 'access-control-allow-headers': 'content-type', 'access-control-max-age': '600' });
+    response.end();
+    return;
+  }
+  const project = url.searchParams.get('project');
+  if (!isUuid(project)) return send(400, { error: 'notAProject' });
+  const key = cookieValue(request.headers.cookie, SESSION_COOKIE);
+  const access = key === undefined ? ({ kind: 'notSignedIn' } as const) : await sessions.access(key, project);
+  if (access.kind !== 'granted') return send(403, { error: access.kind });
+  const name = /^\/blobs\/([^/]+)$/.exec(url.pathname)?.[1];
+  if (request.method === 'GET' && name !== undefined) {
+    const bytes = (await blobs.get([name])).get(name);
+    if (bytes === undefined) return send(404, { error: 'unknownBlob' });
+    // A blob's bytes never change under its name.
+    response.writeHead(200, { ...headers, 'content-type': 'application/octet-stream', 'content-length': String(bytes.length), 'cache-control': 'private, max-age=31536000, immutable' });
+    response.end(bytes);
+    return;
+  }
+  if (request.method === 'POST' && url.pathname === '/blobs') {
+    if (!allowed) {
+      log.warn('blob.foreignOrigin', { project, origin: origin ?? null });
+      return send(403, { error: 'foreignOrigin' });
+    }
+    if (access.level !== 'write') return send(403, { error: 'readOnly' });
+    if (access.frozen) return send(403, { error: 'frozen' });
+    const parts: Buffer[] = [];
+    let size = 0;
+    for await (const part of request as AsyncIterable<Buffer>) {
+      size += part.length;
+      if (size > BLOB_MAX_BYTES) {
+        log.warn('blob.tooLarge', { project, user: access.person.login });
+        return send(413, { error: 'tooLarge' });
+      }
+      parts.push(part);
+    }
+    const sha = await blobs.put(Buffer.concat(parts));
+    log.info('blob.kept', { project, user: access.person.login, sha, size });
+    return send(201, { sha });
+  }
+  send(404, { error: 'notFound' });
+}

packages/server/src/blobs.ts

@@ -0,0 +1,52 @@
+// The large binary files of the projects' computers (plan §5.1,
+// packages/doc's blobs.ts), content-addressed: named by the SHA-256 of
+// their bytes, kept once however many projects hold them. F computes the
+// name itself; what a client says a blob is called is not believed.
+//
+// Not yet: forgetting a blob no project names any more (they stay; the
+// commits keep theirs in the dashboard, plan §10.1).
+
+import { createHash } from 'node:crypto';
+import type pg from 'pg';
+
+/** A blob at most (a picture or a file put into a simulated computer). */
+export const BLOB_MAX_BYTES = 16 * 1024 * 1024;
+
+export interface BlobStore {
+  migrate(): Promise<void>;
+  /** Bytes kept; their name. */
+  put(bytes: Uint8Array): Promise<string>;
+  /** The blobs of these names that F has. */
+  get(shas: readonly string[]): Promise<Map<string, Uint8Array>>;
+}
+
+export const shaOf = (bytes: Uint8Array) => createHash('sha256').update(bytes).digest('hex');
+
+const SCHEMA = `
+CREATE TABLE IF NOT EXISTS filius_blob (
+  sha text PRIMARY KEY,
+  bytes bytea NOT NULL,
+  size integer NOT NULL,
+  created_at timestamptz NOT NULL DEFAULT now()
+);
+`;
+
+export class PostgresBlobs implements BlobStore {
+  constructor(private readonly pool: pg.Pool) {}
+
+  async migrate(): Promise<void> {
+    await this.pool.query(SCHEMA);
+  }
+
+  async put(bytes: Uint8Array): Promise<string> {
+    const sha = shaOf(bytes);
+    await this.pool.query('INSERT INTO filius_blob (sha, bytes, size) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING', [sha, Buffer.from(bytes), bytes.length]);
+    return sha;
+  }
+
+  async get(shas: readonly string[]): Promise<Map<string, Uint8Array>> {
+    if (shas.length === 0) return new Map();
+    const result = await this.pool.query<{ sha: string; bytes: Buffer }>('SELECT sha, bytes FROM filius_blob WHERE sha = ANY($1)', [shas]);
+    return new Map(result.rows.map((row) => [row.sha, new Uint8Array(row.bytes.buffer, row.bytes.byteOffset, row.bytes.byteLength)]));
+  }
+}

packages/server/src/index.ts

@@ -1,6 +1,8 @@
 // Server F (plan §8): the shared documents, the simulation rooms, the
 // sign-in and the dashboard's service.
 export * from './auth.js';
+export * from './blobRoutes.js';
+export * from './blobs.js';
 export * from './log.js';
 export * from './persistence.js';
 export * from './server.js';

packages/server/src/main.ts

@@ -7,6 +7,7 @@
 // machine without HTTPS). The rooms' thread is the bundle beside this one
 // (dist/simWorker.mjs).
 
+import { PostgresBlobs } from './blobs.js';
 import { jsonLog } from './log.js';
 import { startServer } from './server.js';
 import { PostgresSessions } from './sessions.js';
@@ -23,12 +24,15 @@ const secureCookie = process.env.FILIUS_INSECURE_COOKIE !== '1';
 if (!secureCookie) log.warn('auth.insecureCookie', { note: 'the session cookie goes over plain HTTP' });
 const store = PostgresStore.connect(config.DATABASE_URL!);
 const sessions = new PostgresSessions(store.pool);
+const blobs = new PostgresBlobs(store.pool);
 await store.migrate();
 await sessions.migrate();
+await blobs.migrate();
 const server = await startServer({
   port: Number(process.env.FILIUS_PORT ?? 8080),
   store,
   sessions,
+  blobs,
   log,
   ticketSecret: config.FILIUS_TICKET_SECRET!,
   serviceKey: config.FILIUS_SERVICE_KEY!,

packages/server/src/server.ts

@@ -1,6 +1,7 @@
 // Server F's front door (plan §8.2): one HTTP server — /healthz for the
 // proxy, /__auth for the editor's frame (ticket in, cookie out), /svc/* for
-// the dashboard (with the service key), the WebSockets /sync for the shared
+// the dashboard (with the service key), /blobs for the large binary files
+// of the projects' computers, the WebSockets /sync for the shared
 // documents and /sim?project=… for the simulation rooms (the session's grant
 // decides there as at /sync; a refusal closes the socket with its reason);
 // anything else is not found. A WebSocket from a page of another
@@ -18,6 +19,8 @@ import type { AddressInfo, Socket } from 'node:net';
 import { SIM_REFUSED_CODE, type SyncRefusal } from '@mdg-filius/protocol';
 import { WebSocketServer, type WebSocket } from 'ws';
 import { exchangeTicket } from './auth.js';
+import { handleBlobs } from './blobRoutes.js';
+import type { BlobStore } from './blobs.js';
 import { errorText, type Log } from './log.js';
 import { handleService } from './service.js';
 import { holdMessages, SimRooms } from './simRooms.js';
@@ -46,6 +49,7 @@ export interface ServerOptions extends SyncOptions {
   readonly secureCookie: boolean;
   /** The bundled room thread (dist/simWorker.mjs). */
   readonly workerUrl: URL;
+  readonly blobs: BlobStore;
   readonly simThreads?: number;
   readonly lingerMilliseconds?: number;
   readonly pingMilliseconds?: number;
@@ -65,6 +69,7 @@ export async function startServer(options: ServerOptions): Promise<FiliusServer>
   const { hocuspocus } = sync;
   const sims = new SimRooms({
     hocuspocus,
+    blobs: options.blobs,
     log: options.log,
     workerUrl: options.workerUrl,
     ...(options.simThreads === undefined ? {} : { threads: options.simThreads }),
@@ -105,6 +110,7 @@ export async function startServer(options: ServerOptions): Promise<FiliusServer>
     if (request.method === 'GET' && path === '/healthz') json(response, 200, { status: 'ok', documents: hocuspocus.getDocumentsCount(), connections: hocuspocus.getConnectionsCount(), rooms: sims.stats().rooms });
     else if (request.method === 'GET' && path === '/__auth') await exchangeTicket(request, response, options);
     else if (path.startsWith('/svc/')) await handleService(request, response, path, { ...options, sync, sims });
+    else if (path === '/blobs' || path.startsWith('/blobs/')) await handleBlobs(request, response, new URL(request.url ?? '/', 'http://filius'), options);
     else json(response, 404, { error: 'notFound' });
   };
   const http: Server = createServer((request, response) => {

packages/server/src/simChannel.ts

@@ -4,10 +4,16 @@
 // rooms, each with a replica of its project's document: the document's
 // updates go both ways, a person's messages go in, the room's come out,
 // each for one person (a room tells its people apart; the main thread
-// numbers them).
+// numbers them). A room's large binary files (blobs) are F's: the thread
+// asks for the ones it has not got, and gives F the ones the simulation
+// makes before the document's change that names them.
 
 import type { HostMessage } from '@mdg-filius/protocol';
 
+/** How often, and how long apart, a thread asks again for a blob F has not got (yet). */
+export const BLOB_ASKS = 5;
+export const BLOB_ASK_AGAIN_MILLISECONDS = 1_000;
+
 export type ToRoomThread =
   /** A room for a project, its document as it is now. */
   | { readonly type: 'open'; readonly room: string; readonly state: Uint8Array }
@@ -19,10 +25,16 @@ export type ToRoomThread =
   /** The simulation stopped (what it left written back first); answered by `stopped`. */
   | { readonly type: 'stop'; readonly room: string }
   /** The room ended — stopped first with `writeBack`, else as it is (its project is gone); answered by `closed`. */
-  | { readonly type: 'close'; readonly room: string; readonly writeBack: boolean };
+  | { readonly type: 'close'; readonly room: string; readonly writeBack: boolean }
+  /** The blobs asked for: those F has, and the names of those it has not. */
+  | { readonly type: 'blobs'; readonly room: string; readonly blobs: readonly { readonly sha: string; readonly bytes: Uint8Array }[]; readonly missing: readonly string[] };
 
 export type FromRoomThread =
   | { readonly type: 'doc'; readonly room: string; readonly update: Uint8Array }
   | { readonly type: 'host'; readonly room: string; readonly client: number; readonly message: HostMessage }
   | { readonly type: 'stopped'; readonly room: string }
+  /** Blobs the document names that the thread has not got. */
+  | { readonly type: 'needBlobs'; readonly room: string; readonly shas: readonly string[] }
+  /** A blob the simulation made (before the change that names it). */
+  | { readonly type: 'blob'; readonly room: string; readonly bytes: Uint8Array }
   | { readonly type: 'closed'; readonly room: string };

packages/server/src/simRooms.ts

@@ -3,7 +3,11 @@
 // with the fewest rooms takes a new one). A room keeps its project's
 // document open on F (a direct connection: what the simulation leaves goes
 // into the document even when no one has it open) and gets every change of
-// it; what the room writes comes back as the room's own change.
+// it; what the room writes comes back as the room's own change. The blobs
+// its document names come from F's store when the thread asks; a blob the
+// simulation makes is kept before the change that names it is applied (the
+// room's messages are taken in order), so that no one is told of a file
+// whose bytes are not there.
 //
 // Who may do what: everyone in the project watches; only with a write grant
 // does one start, stop, set the speed, hold "Pakete verwerfen", clear the
@@ -23,6 +27,7 @@ import type { DirectConnection, Hocuspocus } from '@hocuspocus/server';
 import * as Y from 'yjs';
 import { SIM_CLOSED_CODE, SIM_ROOM_FAILED_CODE, WRITING_KINDS, type ClientMessageKind, type HostMessage, type SyncClosing } from '@mdg-filius/protocol';
 import type WebSocket from 'ws';
+import type { BlobStore } from './blobs.js';
 import { errorText, type Log } from './log.js';
 import type { FromRoomThread, ToRoomThread } from './simChannel.js';
 
@@ -39,6 +44,7 @@ export interface SimPerson {
 
 export interface SimRoomsOptions {
   readonly hocuspocus: Hocuspocus;
+  readonly blobs: BlobStore;
   readonly log: Log;
   /** The bundled room thread (dist/simWorker.mjs). */
   readonly workerUrl: URL;
@@ -64,6 +70,8 @@ interface Room {
   frozen: boolean;
   linger: ReturnType<typeof setTimeout> | undefined;
   ended: boolean;
+  /** The thread's changes and blobs, taken in order. */
+  taken: Promise<void>;
 }
 
 type RawData = Buffer | ArrayBuffer | Buffer[];
@@ -124,9 +132,30 @@ export class SimRooms {
   private fromThread(thread: Thread, message: FromRoomThread): void {
     const room = [...thread.rooms].find((each) => each.project === message.room);
     if (room === undefined) return;
+    const inOrder = (work: () => Promise<void> | void) => {
+      room.taken = room.taken.then(work).catch((error: unknown) => this.options.log.error('sim.blobFailed', { project: room.project, error: errorText(error) }));
+    };
     switch (message.type) {
       case 'doc':
-        Y.applyUpdate(room.doc, message.update, room);
+        inOrder(() => Y.applyUpdate(room.doc, message.update, room));
+        return;
+      case 'blob':
+        inOrder(async () => void (await this.options.blobs.put(message.bytes)));
+        return;
+      case 'needBlobs':
+        void this.options.blobs.get(message.shas).then(
+          (found) =>
+            this.post(room, {
+              type: 'blobs',
+              room: room.project,
+              blobs: [...found].map(([sha, bytes]) => ({ sha, bytes })),
+              missing: message.shas.filter((sha) => !found.has(sha)),
+            }),
+          (error: unknown) => {
+            this.options.log.error('sim.blobFailed', { project: room.project, error: errorText(error) });
+            this.post(room, { type: 'blobs', room: room.project, blobs: [], missing: message.shas });
+          },
+        );
         return;
       case 'host':
         room.clients.get(message.client)?.socket.send(JSON.stringify(message.message));
@@ -168,6 +197,7 @@ export class SimRooms {
       frozen,
       linger: undefined,
       ended: false,
+      taken: Promise.resolve(),
     };
     thread.rooms.add(room);
     this.post(room, { type: 'open', room: project, state: Y.encodeStateAsUpdate(doc) });
@@ -267,6 +297,8 @@ export class SimRooms {
     const done = (async () => {
       this.rooms.delete(room.project);
       await this.ask(room, { type: 'close', room: room.project, writeBack });
+      // What it wrote back is in the document before the document is let go.
+      await room.taken;
       this.forget(room);
       await room.direct.disconnect();
       this.options.log.info('sim.roomClosed', { project: room.project, writeBack });

packages/server/src/simWorker.ts

@@ -1,12 +1,18 @@
 // A thread of server F that holds simulation rooms (simChannel.ts): each
 // room a replica of its project's document and the room that reads it, all
 // ticking together every TICK_MILLISECONDS. Bundled on its own
-// (dist/simWorker.mjs), started by the main thread.
+// (dist/simWorker.mjs), started by the main thread. The blobs a room's
+// document names are asked of the main thread as they come into it; a start
+// waits until the ones asked for have come (or F said it has not got them),
+// so that the simulation starts with the files' bytes. One F has not got —
+// yet: the editor that made it may still be taking it there — is asked for
+// again a few times; when it comes, the running simulation reads it.
 
 import { parentPort } from 'node:worker_threads';
 import * as Y from 'yjs';
+import { attachBlobs, memoryBlobSide, missingBlobs } from '@mdg-filius/doc';
 import { SimulationRoom, TICK_MILLISECONDS } from '@mdg-filius/sim';
-import type { FromRoomThread, ToRoomThread } from './simChannel.js';
+import { BLOB_ASK_AGAIN_MILLISECONDS, BLOB_ASKS, type FromRoomThread, type ToRoomThread } from './simChannel.js';
 
 /** The origin of what came from the main thread: not sent back. */
 const MAIN_ORIGIN = 'main';
@@ -16,21 +22,43 @@ interface HeldRoom {
   readonly room: SimulationRoom;
   /** The main thread's number of a person → the room's. */
   readonly clients: Map<number, number>;
+  readonly blobs: ReturnType<typeof memoryBlobSide>;
+  /** Blobs asked for and not answered yet. */
+  readonly asked: Set<string>;
+  /** Blobs F said it has not got: asked again a while later, a few times (`misses` counts). */
+  readonly unavailable: Set<string>;
+  readonly misses: Map<string, number>;
+  /** Starts waiting for the blobs asked for. */
+  readonly waiting: { readonly client: number; readonly message: unknown }[];
 }
 
 const port = parentPort!;
 const rooms = new Map<string, HeldRoom>();
 const post = (message: FromRoomThread) => port.postMessage(message);
 
+/** Asks the main thread for the blobs the room's document names and the room has not got. */
+function askForBlobs(name: string, held: HeldRoom): void {
+  const shas = missingBlobs(held.doc).filter((sha) => !held.asked.has(sha) && !held.unavailable.has(sha));
+  if (shas.length === 0) return;
+  for (const sha of shas) held.asked.add(sha);
+  post({ type: 'needBlobs', room: name, shas });
+}
+
+const isStart = (message: unknown) => typeof message === 'object' && message !== null && (message as { kind?: unknown }).kind === 'start';
+
 port.on('message', (message: ToRoomThread) => {
   if (message.type === 'open') {
     const doc = new Y.Doc();
+    const blobs = memoryBlobSide((_sha, bytes) => post({ type: 'blob', room: message.room, bytes }));
+    attachBlobs(doc, blobs);
     Y.applyUpdate(doc, message.state, MAIN_ORIGIN);
     doc.on('update', (update: Uint8Array, origin: unknown) => {
       if (origin !== MAIN_ORIGIN && rooms.get(message.room)?.doc === doc) post({ type: 'doc', room: message.room, update });
     });
     const room = new SimulationRoom(doc, { now: () => performance.now(), seed: () => Math.floor(Math.random() * 2 ** 31) });
-    rooms.set(message.room, { doc, room, clients: new Map() });
+    const held: HeldRoom = { doc, room, clients: new Map(), blobs, asked: new Set(), unavailable: new Set(), misses: new Map(), waiting: [] };
+    rooms.set(message.room, held);
+    askForBlobs(message.room, held);
     return;
   }
   const held = rooms.get(message.room);
@@ -38,7 +66,32 @@ port.on('message', (message: ToRoomThread) => {
   switch (message.type) {
     case 'doc':
       Y.applyUpdate(held.doc, message.update, MAIN_ORIGIN);
+      askForBlobs(message.room, held);
       return;
+    case 'blobs': {
+      held.blobs.arrived(new Map(message.blobs.map(({ sha, bytes }) => [sha, bytes])));
+      for (const { sha } of message.blobs) held.asked.delete(sha);
+      if (message.blobs.length > 0) held.room.blobsArrived();
+      for (const sha of message.missing) {
+        held.asked.delete(sha);
+        const asks = (held.misses.get(sha) ?? 0) + 1;
+        held.misses.set(sha, asks);
+        held.unavailable.add(sha);
+        if (asks < BLOB_ASKS)
+          setTimeout(() => {
+            if (rooms.get(message.room) !== held) return;
+            held.unavailable.delete(sha);
+            askForBlobs(message.room, held);
+          }, BLOB_ASK_AGAIN_MILLISECONDS);
+      }
+      if (held.asked.size === 0) {
+        for (const { client, message: waiting } of held.waiting.splice(0)) {
+          const roomClient = held.clients.get(client);
+          if (roomClient !== undefined) held.room.receive(roomClient, waiting);
+        }
+      }
+      return;
+    }
     case 'connect':
       held.clients.set(
         message.client,
@@ -47,7 +100,9 @@ port.on('message', (message: ToRoomThread) => {
       return;
     case 'client': {
       const client = held.clients.get(message.client);
-      if (client !== undefined) held.room.receive(client, message.message);
+      if (client === undefined) return;
+      if (isStart(message.message) && held.asked.size > 0) held.waiting.push({ client: message.client, message: message.message });
+      else held.room.receive(client, message.message);
       return;
     }
     case 'disconnect': {