Circuit Trace Generators

import { createStreamingCircuitField } from "./circuit-field-stream.js";

const scope = globalThis;
const pendingEnsures = [];
const pumpChannel = new MessageChannel();
const retainedTraceMeta = new Map();

let field;
let unsubscribe;
let initialized = false;
let pumpScheduled = false;
let highestTarget = 0;
let publiclyPruned = 0;
let fatal = false;

function serializeError(error) {
  return {
    name: error?.name || "Error",
    message: error?.message || String(error),
    stack: error?.stack || "",
  };
}

function post(type, payload = {}) {
  scope.postMessage({ type, ...payload });
}

function publicStats() {
  return {
    ...field.stats,
    retainedTraces: retainedTraceMeta.size,
    totalPruned: publiclyPruned,
  };
}

function postFatal(error, phase) {
  if (fatal) return;
  fatal = true;
  pumpScheduled = false;
  post("fatal", {
    phase,
    error: serializeError(error),
  });
  unsubscribe?.();
  field?.destroy();
  pendingEnsures.length = 0;
  retainedTraceMeta.clear();
}

function settleEnsures() {
  if (!field) return;
  const frontier = field.generatedThrough;
  let writeIndex = 0;
  highestTarget = 0;

  for (let index = 0; index < pendingEnsures.length; index += 1) {
    const request = pendingEnsures[index];
    if (request.target <= frontier) {
      post("ensured", {
        requestId: request.requestId,
        target: request.target,
        generatedThrough: frontier,
        stats: publicStats(),
      });
      continue;
    }
    pendingEnsures[writeIndex] = request;
    writeIndex += 1;
    highestTarget = Math.max(highestTarget, request.target);
  }
  pendingEnsures.length = writeIndex;
}

function schedulePump() {
  if (
    pumpScheduled
    || fatal
    || !field
    || field.generatedThrough >= highestTarget
  ) return;
  pumpScheduled = true;
  pumpChannel.port2.postMessage(0);
}

function pumpOneChunk() {
  pumpScheduled = false;
  if (fatal || !field) return;

  try {
    settleEnsures();
    if (field.generatedThrough >= highestTarget) return;

    const stats = field.stats;
    const nextChunkBoundary = (
      (stats.chunks + 1)
      * stats.chunkRows
      * stats.pitch
    );
    field.ensureThrough(nextChunkBoundary);
    settleEnsures();
    schedulePump();
  } catch (error) {
    postFatal(error, "generation");
  }
}

pumpChannel.port1.onmessage = pumpOneChunk;

function initialize(message) {
  if (initialized) throw new Error("Worker has already been initialized");
  initialized = true;
  field = createStreamingCircuitField(message.options || {});
  unsubscribe = field.onAppend((traces) => {
    for (const trace of traces) retainedTraceMeta.set(trace.id, trace.maxY);
    post("append", {
      traces,
      generatedThrough: field.generatedThrough,
      stats: publicStats(),
    });
    // postMessage has already cloned the batch. Keep only the routing
    // occupancy and lightweight pruning metadata in this thread.
    field.pruneBefore(Number.POSITIVE_INFINITY);
  });
  post("ready", {
    stats: publicStats(),
    width: field.width,
    generatedThrough: field.generatedThrough,
  });
}

function ensureThrough(message) {
  if (!field) throw new Error("Worker is not initialized");
  const target = Math.max(0, Number(message.target) || 0);
  pendingEnsures.push({
    requestId: message.requestId,
    target,
  });
  highestTarget = Math.max(highestTarget, target);
  settleEnsures();
  schedulePump();
}

function pruneBefore(message) {
  if (!field) throw new Error("Worker is not initialized");
  const boundary = Number(message.worldY) || 0;
  const removedIds = [];
  for (const [id, maxY] of retainedTraceMeta) {
    if (maxY >= boundary) continue;
    removedIds.push(id);
    retainedTraceMeta.delete(id);
  }
  publiclyPruned += removedIds.length;
  post("pruned", {
    requestId: message.requestId,
    removedIds,
    stats: publicStats(),
  });
}

function destroy() {
  fatal = true;
  pumpScheduled = false;
  unsubscribe?.();
  field?.destroy();
  pendingEnsures.length = 0;
  retainedTraceMeta.clear();
  pumpChannel.port1.close();
  pumpChannel.port2.close();
  scope.close?.();
}

scope.onmessage = (event) => {
  if (fatal) return;
  const message = event.data || {};
  try {
    if (message.type === "init") initialize(message);
    else if (message.type === "ensure") ensureThrough(message);
    else if (message.type === "prune") pruneBefore(message);
    else if (message.type === "destroy") destroy();
    else throw new Error(`Unknown worker message: ${String(message.type)}`);
  } catch (error) {
    postFatal(error, message.type || "message");
  }
};