mdg-filius/packages/sim/src/room.ts

// A simulation room (plan §6): one engine for one project, the clients that
// watch it, and what goes to them. The same room runs on F for everyone in a
// project and in a web worker for one (LocalSimHost); it takes the
// document's snapshot when the simulation starts, so that changes made while
// it runs come in at the next start, as in Java.
//
// Time: the engine's clock follows the wall clock, a millisecond for a
// millisecond, as Java's threads do; the speed is Java's slider, the delay of
// a cable (RuntimeSpeed), not a time lapse — unlike Swift's tick (plan §6.3).
// The engine counts whole milliseconds, as Java's clock does; the wall
// clock's fractions are carried to the next tick, so that no time is lost.
// A tick catches up at most MAXIMUM_CATCH_UP_MILLISECONDS: after a tab slept,
// the simulation goes on from where it was rather than running the missed
// minutes in one go.
//
// The engine outlives a stop: the capture stays, as Java's Lauscher does, and
// the next start goes on at the time it stopped, so that the capture's times
// keep rising (plan §6.2). So do the command lines' windows, as Java's do
// (TerminalWindow).

import * as Y from 'yjs';
import { DEFAULT_SPEED_PERCENT, NetworkRuntimeEngine, captureRows, interfaceKey, stableGeneratedMacAddress, type CapturedFrame, type CaptureRow, type Id, type LogLine } from '@mdg-filius/core';
import { docSnapshot } from '@mdg-filius/doc';
import {
  parseClientMessage,
  SIM_PROTOCOL_VERSION,
  type AppKind,
  type ClientMessage,
  type ErrorCode,
  type HostMessage,
  type NodeSettings,
  type SimMode,
  type Topic,
  type TraceFrame,
} from '@mdg-filius/protocol';
import { TerminalWindow } from './terminalWindow.js';

export const TICK_MILLISECONDS = 200;
export const MAXIMUM_CATCH_UP_MILLISECONDS = 2_000;

export interface RoomOptions {
  /** The wall clock, in milliseconds. */
  readonly now: () => number;
  /** A seed for each start. */
  readonly seed: () => number;
}

interface Client {
  readonly send: (message: HostMessage) => void;
  readonly topics: Set<Topic>;
  /** Per card, the number of the last frame sent. */
  readonly traceSent: Map<Id, number>;
  /** Per program window, how many of its program's lines it has; none while the program does not run (a stop empties it). */
  readonly appSent: Map<string, number>;
  /** Per computer, the settings last sent (as text). */
  readonly nodeSent: Map<Id, string>;
}

/** A program with a window: its lines. */
interface WithLog {
  readonly log: { readonly all: readonly LogLine<unknown>[] };
}

const message = <T extends Omit<HostMessage, 'v'>>(fields: T) => ({ v: SIM_PROTOCOL_VERSION, ...fields }) as unknown as HostMessage;

export class SimulationRoom {
  private readonly engine: NetworkRuntimeEngine;
  private mode: SimMode = 'design';
  private speed = DEFAULT_SPEED_PERCENT;
  private readonly clients = new Map<number, Client>();
  private nextClient = 1;
  private lastWall: number | undefined;
  /** The cable each card is plugged into, as the simulation started. */
  private linkOfPort = new Map<Id, Id>();
  /** Per card, how many frames it held at the last tick. */
  private framesSeen = new Map<Id, number>();
  private lastActivity: string[] = [];
  /** Each computer's command line window, kept over stops. */
  private readonly terminalWindows = new Map<Id, TerminalWindow>();

  constructor(
    private readonly doc: Y.Doc,
    private readonly options: RoomOptions,
  ) {
    this.engine = new NetworkRuntimeEngine(options.seed(), undefined, { wallClock: options.now });
  }

  /** A new client; it gets the room's state at once. */
  connect(send: (message: HostMessage) => void): number {
    const id = this.nextClient++;
    this.clients.set(id, { send, topics: new Set(), traceSent: new Map(), appSent: new Map(), nodeSent: new Map() });
    send(this.state());
    return id;
  }

  disconnect(client: number): void {
    this.clients.delete(client);
  }

  /** How many clients the room has (F stops a room nobody watches). */
  get clientCount(): number {
    return this.clients.size;
  }

  get currentMode(): SimMode {
    return this.mode;
  }

  /** A client's message, as it came over the wire. */
  receive(clientId: number, raw: unknown): void {
    const client = this.clients.get(clientId);
    if (client === undefined) return;
    const parsed = parseClientMessage(raw);
    if ('error' in parsed) {
      const reqId = typeof (raw as { reqId?: unknown } | null)?.reqId === 'string' ? (raw as { reqId: string }).reqId : undefined;
      client.send(message({ kind: 'error', code: parsed.error.code, detail: parsed.error.detail, ...(reqId === undefined ? {} : { reqId }) }));
      return;
    }
    this.handle(client, parsed.message);
  }

  private handle(client: Client, input: ClientMessage): void {
    switch (input.kind) {
      case 'sub':
        for (const topic of input.topics) {
          client.topics.add(topic);
          if (topic.startsWith('trace:')) client.traceSent.delete(topic.slice('trace:'.length));
          if (topic.startsWith('app:')) client.appSent.delete(topic);
          if (topic.startsWith('node:')) client.nodeSent.delete(topic.slice('node:'.length));
        }
        this.flush([client], true);
        return;
      case 'unsub':
        for (const topic of input.topics) client.topics.delete(topic);
        return;
      case 'start':
        this.start();
        return;
      case 'stop':
        this.stop();
        return;
      case 'speed':
        this.speed = Math.min(100, Math.max(1, Math.trunc(input.value)));
        this.engine.setSimulationSpeed(this.speed);
        this.broadcast(this.state());
        return;
      case 'input': {
        let failure: ErrorCode | undefined = 'engineFailed';
        if (!this.guarded(() => (failure = this.act(input)))) return;
        client.send(failure === undefined ? message({ kind: 'ack', reqId: input.reqId }) : message({ kind: 'error', code: failure, reqId: input.reqId }));
        this.flush([...this.clients.values()], false);
        return;
      }
    }
  }

  private state(): HostMessage {
    return message({ kind: 'state', mode: this.mode, speed: this.speed, t: this.engine.state.currentTimeMilliseconds });
  }

  private broadcast(output: HostMessage): void {
    for (const client of this.clients.values()) client.send(output);
  }

  start(): void {
    if (this.mode === 'simulation') return;
    const snapshot = docSnapshot(this.doc);
    this.engine.setSimulationSpeed(this.speed);
    if (!this.guarded(() => this.engine.handle({ type: 'start', snapshot, seed: this.options.seed(), initialTimeMilliseconds: this.engine.state.currentTimeMilliseconds }))) return;
    this.linkOfPort = new Map(snapshot.links.flatMap((link) => [[link.sourcePortId, link.id] as const, [link.targetPortId, link.id] as const]));
    for (const node of snapshot.nodes) {
      const terminal = this.engine.terminal(node.id);
      if (terminal === undefined) continue;
      const window = this.terminalWindows.get(node.id);
      if (window === undefined) this.terminalWindows.set(node.id, new TerminalWindow(terminal));
      else window.attach(terminal);
    }
    this.mode = 'simulation';
    this.lastWall = this.options.now();
    this.broadcast(this.state());
    this.flush([...this.clients.values()], false);
  }

  stop(): void {
    if (this.mode === 'design') return;
    this.engine.handle({ type: 'stop' });
    this.mode = 'design';
    this.lastWall = undefined;
    this.broadcast(this.state());
    this.flush([...this.clients.values()], false);
  }

  /** Lets the simulation run to the wall clock's time, then sends what came of it. */
  tick(): void {
    if (this.mode !== 'simulation') return;
    const now = this.options.now();
    const behind = Math.max(0, now - (this.lastWall ?? now));
    // Whole milliseconds, as the engine counts them; the wall clock's fraction is kept for the next tick.
    const elapsed = Math.min(MAXIMUM_CATCH_UP_MILLISECONDS, Math.floor(behind));
    this.lastWall = behind > MAXIMUM_CATCH_UP_MILLISECONDS ? now : (this.lastWall ?? now) + elapsed;
    if (!this.guarded(() => this.engine.handle({ type: 'advance', toMilliseconds: this.engine.state.currentTimeMilliseconds + elapsed }))) return;
    this.flush([...this.clients.values()], false);
  }

  /**
   * Runs a step of the engine. When the core throws (a part it has not got
   * yet), the room tells every client and goes back to design mode instead of
   * dying with its worker; false then.
   */
  private guarded(step: () => void): boolean {
    try {
      step();
      return true;
    } catch (error) {
      this.broadcast(message({ kind: 'error', code: 'engineFailed', detail: error instanceof Error ? error.message : String(error) }));
      if (this.engine.state.phase === 'running') this.engine.handle({ type: 'stop' });
      this.mode = 'design';
      this.lastWall = undefined;
      this.broadcast(this.state());
      this.flush([...this.clients.values()], false);
      return false;
    }
  }

  /** An action in a program's window; the failure, if any. */
  private act(input: Extract<ClientMessage, { kind: 'input' }>): ErrorCode | undefined {
    if (this.mode !== 'simulation') return 'notRunning';
    if (!this.engine.state.topologySnapshot.nodes.some((node) => node.id === input.node)) return 'unknownNode';
    if (input.app !== 'terminal') return 'unknownAction';
    // A window has a program while the simulation runs (checked above), from its first start on.
    const window = this.terminalWindows.get(input.node);
    if (window === undefined || this.engine.terminal(input.node) === undefined) return 'unknownApp';
    switch (input.action) {
      case 'enter':
        if (typeof input.args.line !== 'string') return 'badArgs';
        window.enter(input.args.line);
        return undefined;
      case 'interrupt':
        window.interrupt();
        return undefined;
      default:
        return 'unknownAction';
    }
  }

  private program(node: Id, app: AppKind): WithLog | undefined {
    switch (app) {
      case 'terminal':
        return this.engine.terminal(node);
      case 'webBrowser':
        return this.engine.webBrowser(node);
      case 'webServer':
        return this.engine.webServer(node);
      case 'emailClient':
        return this.engine.emailClient(node);
      case 'emailServer':
        return this.engine.emailServer(node);
      case 'echoServer':
        return this.engine.echoServer(node);
      case 'simpleClient':
        return this.engine.simpleClient(node);
      case 'gnutella':
        return this.engine.gnutella(node);
    }
  }

  /** What the clients are owed: the cables' activity, new frames of the cards and new lines of the windows they watch. */
  private flush(clients: readonly Client[], subscribing: boolean): void {
    if (!subscribing) this.updateActivity();
    for (const window of this.terminalWindows.values()) window.pull();
    const rowsCache = new Map<Id, TraceFrame[]>();
    const traceFrames = (portId: Id, after: number): TraceFrame[] => {
      const frames = this.engine.capture.frames(portId).filter((frame) => frame.number > after);
      if (frames.length === 0) return [];
      if (after === 0 && rowsCache.has(portId)) return rowsCache.get(portId)!;
      const result = traceFramesOf(frames);
      if (after === 0) rowsCache.set(portId, result);
      return result;
    };
    for (const client of clients) {
      for (const topic of client.topics) {
        if (topic === 'activity') {
          if (subscribing) client.send(message({ kind: 'activity', links: this.lastActivity }));
        } else if (topic.startsWith('node:')) {
          const node = topic.slice('node:'.length);
          const settings = this.nodeSettings(node);
          if (settings === undefined) continue;
          const text = JSON.stringify(settings);
          if (client.nodeSent.get(node) === text) continue;
          client.nodeSent.set(node, text);
          client.send(message({ kind: 'node', node, settings }));
        } else if (topic.startsWith('trace:')) {
          const portId = topic.slice('trace:'.length);
          const reset = !client.traceSent.has(portId);
          const frames = traceFrames(portId, client.traceSent.get(portId) ?? 0);
          if (frames.length === 0 && !reset) continue;
          client.traceSent.set(portId, frames.at(-1)?.number ?? client.traceSent.get(portId) ?? 0);
          client.send(message({ kind: 'trace', portId, frames, reset }));
        } else {
          const [, node, app] = /^app:([^/]+)\/(.+)$/.exec(topic)!;
          const sent = client.appSent.get(topic);
          if (app === 'terminal') {
            // The window's lines, kept over stops: a client goes on from those it has.
            const lines = this.terminalWindows.get(node!)?.lines ?? [];
            const from = sent ?? 0;
            if (sent !== undefined && lines.length === from) continue;
            client.appSent.set(topic, lines.length);
            client.send(message({ kind: 'app', node: node!, app: 'terminal', from, lines: lines.slice(from) }));
            continue;
          }
          const program = this.program(node!, app as AppKind);
          if (program === undefined) {
            // A window of a program that is not running: it starts empty with the next start.
            if (sent !== undefined || subscribing) client.send(message({ kind: 'app', node: node!, app: app as AppKind, from: 0, lines: [] }));
            client.appSent.delete(topic);
            continue;
          }
          // The window goes on from the lines it has; after a stop (none sent) it begins anew.
          const from = sent ?? 0;
          const lines = program.log.all.slice(from);
          if (sent !== undefined && lines.length === 0) continue;
          client.appSent.set(topic, program.log.all.length);
          client.send(message({ kind: 'app', node: node!, app: app as AppKind, from, lines }));
        }
      }
    }
  }

  /**
   * A device's first card as the engine runs it (a computer's settings, which
   * DHCP may change); undefined for a device the simulation never had.
   */
  private nodeSettings(id: Id): NodeSettings | undefined {
    const snapshot = this.engine.state.topologySnapshot;
    const node = snapshot.nodes.find((each) => each.id === id);
    if (node === undefined) return undefined;
    const port = node.ports[0];
    const device = snapshot.deviceConfigurations.get(id);
    const card = port === undefined ? undefined : snapshot.interfaceConfigurations.get(interfaceKey(id, port.id));
    return {
      ipAddress: device?.ipAddress ?? card?.ipAddress ?? '',
      subnetMask: device?.subnetMask ?? card?.subnetMask ?? '',
      defaultGateway: device?.defaultGateway ?? '',
      dnsServer: device?.dnsServer ?? '',
      macAddress: port === undefined ? '' : (port.macAddress ?? stableGeneratedMacAddress(port.id)),
    };
  }

  /** The cables whose cards got frames since the last tick; sent when that changes. */
  private updateActivity(): void {
    const links = new Set<string>();
    for (const portId of this.engine.capture.portIds()) {
      const frames = this.engine.capture.frames(portId);
      const count = frames.at(-1)?.number ?? 0;
      if (count > (this.framesSeen.get(portId) ?? 0)) {
        const link = this.linkOfPort.get(portId);
        if (link !== undefined) links.add(link);
      }
      this.framesSeen.set(portId, count);
    }
    const activity = [...links].sort();
    if (activity.join() === this.lastActivity.join()) return;
    this.lastActivity = activity;
    for (const client of this.clients.values()) if (client.topics.has('activity')) client.send(message({ kind: 'activity', links: activity }));
  }
}

/** A card's frames as the capture window shows them, both views. */
export function traceFramesOf(frames: readonly CapturedFrame[]): TraceFrame[] {
  const byNumber = (rows: readonly CaptureRow[]) => {
    const grouped = new Map<number, CaptureRow[]>();
    for (const row of rows) grouped.set(row.number, [...(grouped.get(row.number) ?? []), row]);
    return grouped;
  };
  const rows = byNumber(captureRows(frames, { withAddresses: true }));
  const layered = byNumber(captureRows(frames, { withAddresses: false }));
  return frames.map((frame) => ({ number: frame.number, rows: rows.get(frame.number) ?? [], layeredRows: layered.get(frame.number) ?? [] }));
}