mesh-tools/src/main.ts

// The runnable entrypoint. Three modes:
//
//   mesh-tools                    serve — bind the broker and serve the assigned modules until
//                                 stopped. A module entrypoint that subscribes to events (on("#"))
//                                 starts consuming as it is imported, so this also runs consumers.
//   mesh-tools emit TYPE [JSON]   emit one event onto the mesh and exit — an operable primitive,
//                                 and what an events test uses to put a message on the wire.
//   mesh-tools prepare            bring this module's state to the shape this version needs and exit
//                                 — the runtime's answer to the word the mesh asks every module
//                                 (novox/hq ADR 0135). The entrypoints come from MESH_PREPARE, which
//                                 the module's own image names beside MESH_TOOL_MODULES.
//   mesh-tools run ENTRYPOINT     run one compiled module entrypoint to completion and exit — the
//                                 runtime side of a run-once step (novox/hq ADR 0052). It imports
//                                 the given entrypoint, whose top-level code does its work — seed a
//                                 store, migrate, health-gate — and awaits it. It does NOT connect
//                                 to the broker: a first-boot step runs offline, before the module
//                                 has anything to talk to, and the host gates the container that
//                                 depends on it on this process exiting 0.
//
// The broker, in order of preference:
//   MESH_BROKER_FILE   a sealed {url, fingerprint} the mesh delivered (novox/hq ADR 0043) — an
//                      amqps account scoped to this module. Preferred: a module holds its own.
//   MESH_BROKER_URL    a plain URL, for the bootstrap/admin case before a module has an account.
//   MESH_TOOL_MODULES  /path/a,/path/b,…   compiled module entrypoints (serve mode)
//   MESH_PREPARE       /path/a,/path/b,…   compiled entrypoints that prepare this module's state
//   MESH_MODULE / MESH_NODE                the identity stamped onto emitted events (ADR 0042)

import { readFileSync } from "node:fs";
import { pathToFileURL } from "node:url";
import { connectNats, fatalBrokerReason as fatalNatsReason, type Credential } from "./broker-nats.js";
import { runTools } from "./runtime.js";

/** The credential this process connected with, for what it says beyond the connection (ADR 0159). */
let lastCredential: Credential | undefined;
import { invokeTool } from "@novox/mesh-sdk/tools";
import { useBroker } from "@novox/mesh-sdk/messaging";
import type { Broker } from "@novox/mesh-sdk/messaging";
import { emit } from "@novox/mesh-sdk/events";

/**
 * Connect the way this process is meant to: with its sealed credential if the mesh gave it one, and
 * over the plain bootstrap URL otherwise. A scoped module assumes the foundation's exchanges exist —
 * its account may not declare them (ADR 0043).
 */
// fatalBrokerReasonFor is the reason a connection failure is final rather than "not yet", for
// whichever bus this runtime is on — each transport knows its own refusals.
function fatalBrokerReasonFor(err: unknown): string | null {
  return fatalNatsReason(err);
}

async function connectBroker(): Promise<Broker> {
  const file = process.env.MESH_BROKER_FILE;
  if (file) {
    let credential: Credential;
    try {
      credential = JSON.parse(readFileSync(file, "utf8")) as Credential;
      lastCredential = credential;
    } catch (err) {
      console.error(`mesh-tools: cannot read the broker credential at ${file}: ${err}`);
      process.exit(1);
    }
    if (!credential.url) {
      console.error(`mesh-tools: ${file} carries no url — it is not a broker credential`);
      process.exit(1);
    }
    // The mesh scoped this account to a node and module; take the runtime's identity from the
    // credential so its queue and the events it emits match what the mesh authorised, no matter
    // what the environment says.
    if (credential.node) process.env.MESH_NODE = credential.node;
    if (credential.module) process.env.MESH_MODULE = credential.module;
    // **The credential names the bus.** A module moved to the bus being built was handed a
    // credential for it — `nats://…` with user, password and fingerprint beside the address — and
    // nothing else in its environment changed (design 25; novox/hq design 28 task 5.2). The scheme
    // is enough to know which bus to speak; a runtime that always dialled the old one would keep
    // serving and answer nobody.
    // One bus (novox/hq ADR 0131, design 28 task 5.5): the credential names it, and it is this.
    return connectNats(credential);
  }
  const url = process.env.MESH_BROKER_URL;
  if (!url) {
    console.error(
      "mesh-tools: set MESH_BROKER_FILE (a sealed credential) or MESH_BROKER_URL — there is no broker to reach",
    );
    process.exit(1);
  }
  return connectNats({ url });
}

/**
 * Connect for serve mode, retrying while the broker is merely not reachable yet. At startup that
 * is the NORMAL case, not a failure: a container comes up in seconds and the overlay tunnel a
 * moment later (novox/hq issue 058). Exiting instead delegated the retry to the container
 * runtime, which read as a crash-loop to every restart-counting health check and every person
 * watching. Retried indefinitely, aloud: the dependency appears or somebody reads why not.
 *
 * A failure that waiting cannot fix (see fatalBrokerReason) is thrown at once rather than retried —
 * a permanent fault masquerading as "not reachable yet" is the silent non-progress this whole
 * change exists to remove. Missing-file and empty-URL configuration errors exit inside
 * connectBroker before they reach here; a malformed URL and a refused login are caught here.
 */
async function connectBrokerPatiently(): Promise<Broker> {
  for (let delay = 2_000; ; delay = Math.min(delay * 2, 30_000)) {
    try {
      return await connectBroker();
    } catch (err) {
      const fatal = fatalBrokerReasonFor(err);
      if (fatal !== null) {
        console.error(`mesh-tools: ${fatal} — waiting will not fix this; giving up`);
        throw err;
      }
      const why = err instanceof Error ? err.message : String(err);
      // A little jitter so every module that was up when the broker bounced does not retry in
      // lockstep and stampede it as it recovers.
      const wait = delay + Math.floor(Math.random() * 1_000);
      console.error(`mesh-tools: the broker is not reachable yet (${why}); retrying in ${Math.round(wait / 1000)}s`);
      await new Promise((r) => setTimeout(r, wait));
    }
  }
}

async function serve(): Promise<void> {
  const moduleEntrypoints = (process.env.MESH_TOOL_MODULES ?? "")
    .split(",")
    .map((s) => s.trim())
    .filter(Boolean);

  const broker = await connectBrokerPatiently();
  const stop = await runTools({ broker, moduleEntrypoints, credential: lastCredential });

  const shutdown = async (): Promise<void> => {
    stop();
    await broker.close();
    process.exit(0);
  };
  process.on("SIGTERM", () => void shutdown());
  process.on("SIGINT", () => void shutdown());
}

async function emitOnce(type: string, bodyJson: string): Promise<void> {
  let body: unknown = {};
  if (bodyJson) {
    try {
      body = JSON.parse(bodyJson);
    } catch {
      console.error(`mesh-tools emit: body is not JSON: ${bodyJson}`);
      process.exit(1);
    }
  }
  const broker = await connectBroker();
  useBroker(() => broker);
  // emit awaits the broker's publish confirm (ADR 0042), so the event is accepted before we close.
  await emit(type, body);
  await broker.close();
}

async function invokeOnce(module: string, tool: string, argsJson: string): Promise<void> {
  let args: Record<string, unknown> = {};
  if (argsJson) {
    try {
      args = JSON.parse(argsJson) as Record<string, unknown>;
    } catch {
      console.error(`mesh-tools invoke: args are not JSON: ${argsJson}`);
      process.exit(1);
    }
  }
  const broker = await connectBroker();
  const result = await invokeTool(broker, module, tool, args);
  process.stdout.write(JSON.stringify(result) + "\n");
  await broker.close();
}

/**
 * Run one compiled module entrypoint to completion — the runtime side of a run-once step
 * (novox/hq ADR 0052). Importing it runs its top-level code and awaits any top-level await, so this
 * returns only once the step's own code has finished; a step that throws rejects here and the
 * process exits non-zero, which is how the host knows the step did not complete and must not start
 * the container it gates. No broker is connected — a first-boot seed or migration runs offline.
 */
async function runEntry(entrypoint: string): Promise<void> {
  if (!entrypoint) {
    console.error("mesh-tools run <entrypoint> — a compiled module entrypoint path is required");
    process.exit(1);
  }
  // A file URL, not a bare path: dynamic import of an absolute path is not portable, and the
  // entrypoint the manifest names is an absolute path inside the image.
  await import(pathToFileURL(entrypoint).href);
}

/**
 * Bring this module's state to the shape this version needs, and exit — the runtime's answer to the
 * one word the mesh asks every module (novox/hq ADR 0135).
 *
 * The entrypoints come from `MESH_PREPARE`, which a module's own image sets beside the entrypoints it
 * already lists there: the module knows which of its files prepares its state, and nothing else could.
 * Each is imported in the order given, to completion, with no broker — preparation runs before the
 * version that would use it, so there is nothing yet to talk to.
 *
 * **An empty list is a failure, not a no-op.** The mesh only asks this of a module whose manifest says
 * it prepares something; a module that says so and names nothing has been built wrong, and exiting 0
 * would let that version serve against a state nobody shaped.
 */
async function prepareState(): Promise<void> {
  const named = (process.env.MESH_PREPARE ?? "")
    .split(",")
    .map((entry) => entry.trim())
    .filter((entry) => entry !== "");
  if (named.length === 0) {
    console.error(
      "mesh-tools prepare: this module was asked to prepare its state and its image names nothing " +
        "to do it with — set MESH_PREPARE to the compiled entrypoint(s) that prepare it, the way " +
        "MESH_TOOL_MODULES names the ones it serves",
    );
    process.exit(1);
  }
  for (const entrypoint of named) {
    console.log(`[mesh-tools] preparing with ${entrypoint}`);
    await runEntry(entrypoint);
  }
  console.log(`[mesh-tools] prepared: ${named.length} entrypoint(s) ran to completion`);
}

async function main(): Promise<void> {
  const [command, ...rest] = process.argv.slice(2);
  if (command === "run") {
    await runEntry(rest[0] ?? "");
    return;
  }
  if (command === "prepare") {
    await prepareState();
    return;
  }
  if (command === "invoke") {
    const [module, tool] = rest;
    if (!module || !tool) {
      console.error("mesh-tools invoke <module> <tool> [json-args] — a module and tool are required");
      process.exit(1);
    }
    await invokeOnce(module, tool, rest[2] ?? "");
    return;
  }
  if (command === "emit") {
    const type = rest[0];
    if (!type) {
      console.error("mesh-tools emit <type> [json-body] — a routing key is required");
      process.exit(1);
    }
    await emitOnce(type, rest[1] ?? "");
    return;
  }
  await serve();
}

void main();