mdg-filius/packages/server/src/simRooms.ts

// The simulation rooms on F (plan §6): one room for a project, shared by
// everyone in it, held by one of a few threads (worker_threads; the one
// 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. 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, restart, set the speed, hold "Pakete verwerfen", clear
// the capture or act in a window (WRITING_KINDS; else `readOnly`). A frozen
// project's simulation does not start, nor start anew (`frozen`); freezing stops a running
// one, what it left written back first.
//
// When the last person leaves, the room stays LINGER_MILLISECONDS (a tab
// changed, a tablet locked, a short drop of the network: plan §6.2), then
// stops — what the simulation left written back — and ends. A thread that
// fails ends its rooms: their people are let go (SIM_ROOM_FAILED_CODE) and
// find a new room, in design mode, when they come back.

import { availableParallelism } from 'node:os';
import { Worker } from 'node:worker_threads';
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';

export const LINGER_MILLISECONDS = 60_000;

/** Who is in a room. */
export interface SimPerson {
  readonly user: string;
  readonly login: string;
  readonly name: string;
  /** A write grant: may change the room. */
  readonly write: boolean;
}

export interface SimRoomsOptions {
  readonly hocuspocus: Hocuspocus;
  readonly blobs: BlobStore;
  readonly log: Log;
  /** The bundled room thread (dist/simWorker.mjs). */
  readonly workerUrl: URL;
  /** How many threads hold rooms; all cores but one by default. */
  readonly threads?: number;
  readonly lingerMilliseconds?: number;
}

interface Thread {
  worker: Worker;
  readonly rooms: Set<Room>;
}

interface Room {
  readonly project: string;
  readonly thread: Thread;
  readonly direct: DirectConnection;
  readonly doc: Y.Doc;
  readonly clients: Map<number, { readonly socket: WebSocket; readonly person: SimPerson }>;
  readonly forward: (update: Uint8Array, origin: unknown) => void;
  /** What waits for the thread's answer to `stop` and `close`. */
  readonly waiting: { stopped: (() => void)[]; closed: (() => void)[] };
  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[];

/** What a socket sent while F looked the person up, held until the room hears it. */
export interface HeldMessages {
  readonly messages: readonly RawData[];
  /** Stops holding: the room's listener takes over. */
  release(): void;
}

/** Holds a socket's messages from now on. */
export function holdMessages(socket: WebSocket): HeldMessages {
  const messages: RawData[] = [];
  const hold = (data: RawData) => messages.push(data);
  socket.on('message', hold);
  return { messages, release: () => socket.off('message', hold) };
}

const text = (data: RawData) => (Array.isArray(data) ? Buffer.concat(data) : Buffer.from(data as ArrayBuffer)).toString('utf8');

export class SimRooms {
  private readonly threads: Thread[] = [];
  private readonly rooms = new Map<string, Promise<Room>>();
  private readonly ending = new Map<string, Promise<void>>();
  private nextClient = 1;
  private closing = false;

  constructor(private readonly options: SimRoomsOptions) {
    const count = options.threads ?? Math.max(1, availableParallelism() - 1);
    for (let index = 0; index < count; index++) this.threads.push(this.spawn());
  }

  private spawn(): Thread {
    const thread: Thread = { worker: new Worker(this.options.workerUrl), rooms: new Set() };
    thread.worker.on('message', (message: FromRoomThread) => this.fromThread(thread, message));
    thread.worker.on('error', (error) => this.options.log.error('sim.threadError', { error: errorText(error) }));
    thread.worker.on('exit', (code) => {
      if (!this.closing) this.threadFailed(thread, code);
    });
    return thread;
  }

  private threadFailed(thread: Thread, code: number): void {
    this.options.log.error('sim.threadFailed', { code, rooms: thread.rooms.size });
    for (const room of thread.rooms) {
      for (const { socket } of room.clients.values()) socket.close(SIM_ROOM_FAILED_CODE, 'roomFailed');
      this.forget(room);
      void room.direct.disconnect();
    }
    this.threads[this.threads.indexOf(thread)] = this.spawn();
  }

  private post(room: Room, message: ToRoomThread): void {
    room.thread.worker.postMessage(message);
  }

  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':
        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));
        return;
      case 'stopped':
      case 'closed':
        room.waiting[message.type].shift()?.();
        return;
    }
  }

  /** The room of a project, made when there is none (its document loaded; refused if F does not know the project). */
  private async roomOf(project: string, frozen: boolean): Promise<Room> {
    // A room ending is let go first: its document's direct connection must not meet Hocuspocus unloading it.
    await this.ending.get(project);
    let made = this.rooms.get(project);
    if (made === undefined) {
      made = this.open(project, frozen);
      this.rooms.set(project, made);
      made.catch(() => this.rooms.delete(project));
    }
    return made;
  }

  private async open(project: string, frozen: boolean): Promise<Room> {
    const direct = await this.options.hocuspocus.openDirectConnection(project, { room: true });
    const thread = this.threads.reduce((fewest, each) => (each.rooms.size < fewest.rooms.size ? each : fewest));
    const doc = direct.document!;
    const room: Room = {
      project,
      thread,
      direct,
      doc,
      clients: new Map(),
      forward: (update, origin) => {
        if (origin !== room) this.post(room, { type: 'doc', room: project, update });
      },
      waiting: { stopped: [], closed: [] },
      frozen,
      linger: undefined,
      ended: false,
      taken: Promise.resolve(),
    };
    thread.rooms.add(room);
    this.post(room, { type: 'open', room: project, state: Y.encodeStateAsUpdate(doc) });
    doc.on('update', room.forward);
    this.options.log.info('sim.roomOpened', { project });
    return room;
  }

  /**
   * A person in a project's room: the socket's messages go to the room, the
   * room's to the socket, with what it sent before (`held`, while F looked
   * the person up). Rejected if F does not know the project.
   */
  async join(socket: WebSocket, project: string, person: SimPerson, frozen: boolean, held: HeldMessages = holdMessages(socket)): Promise<void> {
    const room = await this.roomOf(project, frozen);
    // Ended while it was made (its project deleted, its thread failed): the next room.
    if (room.ended) return this.join(socket, project, person, frozen, held);
    // Gone while the room was made: a room no one came to ends as one everyone left.
    if (socket.readyState !== socket.OPEN) {
      held.release();
      if (room.clients.size === 0 && room.linger === undefined && !room.ended) this.lingerOn(room);
      return;
    }
    const client = this.nextClient++;
    room.clients.set(client, { socket, person });
    if (room.linger !== undefined) clearTimeout(room.linger);
    room.linger = undefined;
    this.post(room, { type: 'connect', room: project, client, name: person.name });
    this.options.log.info('sim.joined', { project, user: person.login, write: person.write });
    const receive = (data: RawData) => this.receive(room, client, person, data);
    held.release();
    for (const data of held.messages) receive(data);
    socket.on('message', receive);
    socket.on('close', () => this.leave(room, client));
  }

  private receive(room: Room, client: number, person: SimPerson, data: RawData): void {
    const answer = (message: Omit<Extract<HostMessage, { kind: 'error' }>, 'v'>) => room.clients.get(client)?.socket.send(JSON.stringify({ v: 1, ...message }));
    let message: unknown;
    try {
      message = JSON.parse(text(data));
    } catch {
      answer({ kind: 'error', code: 'badMessage', detail: 'not JSON' });
      return;
    }
    const { kind, reqId } = (typeof message === 'object' && message !== null ? message : {}) as { kind?: unknown; reqId?: unknown };
    const request = typeof reqId === 'string' ? { reqId } : {};
    if (WRITING_KINDS.includes(kind as ClientMessageKind)) {
      if (!person.write) {
        answer({ kind: 'error', code: 'readOnly', ...request });
        return;
      }
      if ((kind === 'start' || kind === 'restart') && room.frozen) {
        answer({ kind: 'error', code: 'frozen', ...request });
        return;
      }
      if (kind === 'start' || kind === 'stop' || kind === 'restart') this.options.log.info(`sim.${kind}`, { project: room.project, user: person.login });
    }
    this.post(room, { type: 'client', room: room.project, client, message });
  }

  private leave(room: Room, client: number): void {
    if (!room.clients.delete(client) || room.ended) return;
    this.post(room, { type: 'disconnect', room: room.project, client });
    if (room.clients.size === 0) this.lingerOn(room);
  }

  private lingerOn(room: Room): void {
    room.linger = setTimeout(() => void this.end(room, true), this.options.lingerMilliseconds ?? LINGER_MILLISECONDS);
  }

  /** The room no longer reachable: a new one is made for whoever comes. */
  private forget(room: Room): void {
    room.ended = true;
    if (room.linger !== undefined) clearTimeout(room.linger);
    room.doc.off('update', room.forward);
    room.thread.rooms.delete(room);
    void this.rooms.get(room.project)?.then(
      (current) => {
        if (current === room) this.rooms.delete(room.project);
      },
      () => {},
    );
    for (const resolve of [...room.waiting.stopped, ...room.waiting.closed]) resolve();
  }

  private ask(room: Room, message: Extract<ToRoomThread, { type: 'stop' | 'close' }>): Promise<void> {
    return new Promise((resolve) => {
      room.waiting[message.type === 'stop' ? 'stopped' : 'closed'].push(resolve);
      this.post(room, message);
    });
  }

  /** Ends a room: stopped first with `writeBack` (what it left in the document), else as it is; its document let go. */
  private end(room: Room, writeBack: boolean): Promise<void> {
    if (room.ended) return this.ending.get(room.project) ?? Promise.resolve();
    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 });
    })();
    room.ended = true;
    this.ending.set(room.project, done);
    void done.finally(() => {
      if (this.ending.get(room.project) === done) this.ending.delete(room.project);
    });
    return done;
  }

  private async existing(project: string): Promise<Room | undefined> {
    const room = await this.rooms.get(project)?.catch(() => undefined);
    return room === undefined || room.ended ? undefined : room;
  }

  /** A project frozen (its running simulation stopped, what it left written back) or thawed. */
  async freeze(project: string, frozen: boolean): Promise<void> {
    const room = await this.existing(project);
    if (room === undefined) return;
    room.frozen = frozen;
    if (frozen) await this.ask(room, { type: 'stop', room: project });
  }

  /** A project deleted: its room ends without writing back, its people are let go; how many. */
  async drop(project: string): Promise<number> {
    const room = await this.existing(project);
    if (room === undefined) return 0;
    const sockets = [...room.clients.values()].map(({ socket }) => socket);
    await this.end(room, false);
    return this.letGo(sockets, 'deleted');
  }

  /** A person's grant ended, to one project or to all: their sockets let go; how many. */
  disconnect(user: string, project?: string): number {
    const sockets: WebSocket[] = [];
    for (const thread of this.threads)
      for (const room of thread.rooms) {
        if (project !== undefined && room.project !== project) continue;
        for (const { socket, person } of room.clients.values()) if (person.user === user) sockets.push(socket);
      }
    return this.letGo(sockets, 'revoked');
  }

  private letGo(sockets: readonly WebSocket[], reason: SyncClosing): number {
    for (const socket of sockets) socket.close(SIM_CLOSED_CODE, reason);
    return sockets.length;
  }

  /** How many rooms, people in them, and rooms each thread holds. */
  stats(): { readonly rooms: number; readonly people: number; readonly threads: readonly number[] } {
    const rooms = this.threads.flatMap((thread) => [...thread.rooms]);
    return { rooms: rooms.length, people: rooms.reduce((sum, room) => sum + room.clients.size, 0), threads: this.threads.map((thread) => thread.rooms.size) };
  }

  /** The threads' workers (for the stats' tests: one may be ended to see the rooms go). */
  get workers(): readonly Worker[] {
    return this.threads.map((thread) => thread.worker);
  }

  /** Every room stopped and ended (what it left written back), every person let go, the threads ended. */
  async close(): Promise<void> {
    const rooms = this.threads.flatMap((thread) => [...thread.rooms]);
    await Promise.all(rooms.map((room) => this.end(room, true)));
    this.closing = true;
    for (const room of rooms) for (const { socket } of room.clients.values()) socket.close(1001, 'service stopping');
    await Promise.all(this.threads.map((thread) => thread.worker.terminate()));
  }
}