pass data to simulation from simulacrum service

Author: jbolda

packages/server/bin/run-simulation-child.ts

@@ -1,11 +1,49 @@
 #!/usr/bin/env node
-import { main, suspend, until } from "effection";
+import { main, suspend, until, type Operation } from "effection";
 import { pathToFileURL } from "node:url";
 import type {
   FoundationSimulator,
   FoundationSimulatorListening,
 } from "@simulacrum/foundation-simulator";
 
+function guardedFactory(
+  factory: Function
+): (initData?: unknown) => Promise<FoundationSimulator<any>> {
+  return async function startSimulation(initData?: unknown) {
+    const sim = await factory(initData);
+    if ("listen" in sim && typeof sim.listen === "function") {
+      return sim as FoundationSimulator<any>;
+    }
+    throw new Error("factory did not return a simulator instance");
+  };
+}
+
+function* normalizeSimulatorFactory(url: string) {
+  try {
+    const mod: unknown = yield* until(import(url));
+
+    // dynamically import module has to be an object if correctly resolved
+    if (mod && typeof mod === "object") {
+      const m = mod;
+
+      // export default as factory
+      if ("default" in m && typeof m.default === "function") {
+        const factory = m.default;
+        return guardedFactory(factory);
+      }
+
+      // export named 'simulation' as factory
+      if ("simulation" in m && typeof m.simulation === "function") {
+        const factory = m.simulation;
+        return guardedFactory(factory);
+      }
+    }
+  } catch (err) {
+    // no-op - will throw in fall through below
+  }
+  throw new Error("no factory or simulator instance found in module");
+}
+
 main(function* () {
   const args = process.argv.slice(2);
   if (args.length < 1) {
@@ -15,37 +53,37 @@ main(function* () {
   const modulePath = args[0];
 
   // Resolve and import module inside the operation
-  let mod: any;
-  try {
-    const url =
-      modulePath.startsWith("./") || modulePath.startsWith("/")
-        ? pathToFileURL(modulePath).href
-        : modulePath;
-    mod = yield* until(import(url));
-  } catch (err) {
-    throw new Error(`failed to import module: ${String(err)}`);
-  }
+  const url =
+    modulePath.startsWith("./") || modulePath.startsWith("/")
+      ? pathToFileURL(modulePath).href
+      : modulePath;
+  const factory = yield* normalizeSimulatorFactory(url);
 
-  const exportNames = ["default", "simulation"];
-  let factory: Function | undefined = undefined;
-  for (const name of exportNames) {
-    if (name in mod && typeof mod[name] === "function") {
-      factory = mod[name];
-      break;
+  let simulacrumPort: number | undefined = undefined;
+  // parse optional flags after modulePath
+  for (let i = 1; i < args.length; i++) {
+    if (args[i] === "--simulacrum-port") {
+      simulacrumPort = Number(args[i + 1]);
+      i++;
     }
   }
-  // fallback: module itself is a function
-  if (!factory && typeof mod === "function") factory = mod;
 
-  if (!factory) {
-    throw new Error(`no factory function found in module: ${modulePath}`);
+  // if present fetch the data chunk and pass it to the factory
+  let initData: JSON | undefined = undefined;
+  if (typeof simulacrumPort === "number" && !Number.isNaN(simulacrumPort)) {
+    try {
+      const res = yield* until(
+        fetch(`http://127.0.0.1:${simulacrumPort}/data`)
+      );
+      initData = yield* until(res.json());
+    } catch (err) {
+      // ignore fetch failures
+      console.error("failed to fetch simulacrum data:", err);
+    }
   }
 
-  let sim = factory() as FoundationSimulator<any>;
-
-  if (!sim || typeof sim.listen !== "function") {
-    throw new Error("factory did not return a simulator with .listen()");
-  }
+  // invoke factory; it may return a simulator instance or a Promise thereof
+  const sim = yield* until(factory(initData));
 
   let listening: FoundationSimulatorListening<any> | undefined = undefined;
   try {

packages/server/example/services/gen-sim-factory.ts

@@ -11,25 +11,31 @@ import {
 export function simulation(
   port: number = 3301,
   startDelay: number = 10
-): FoundationSimulator<any> {
-  const factory = createFoundationSimulationServer({
-    port,
-    extendRouter(router) {
-      router.get("/status", (_req, res) => {
-        res.status(200).send("ok");
-      });
-    },
-  })();
+): (initData?: unknown) => FoundationSimulator<any> {
+  return (initData?: unknown) => {
+    if (initData) console.log("simulation received init data:", initData);
+    const factory = createFoundationSimulationServer({
+      port,
+      extendRouter(router) {
+        router.get("/status", (_req, res) => {
+          res.status(200).send("ok");
+        });
+        router.get("/init-data", (_req, res) => {
+          res.status(200).json({ data: initData ?? null });
+        });
+      },
+    })();
 
-  return {
-    async listen(
-      ...args: Parameters<FoundationSimulator<any>["listen"]>
-    ): Promise<any> {
-      if (startDelay > 0) {
-        await new Promise((resolve) => setTimeout(resolve, startDelay));
-      }
-      // delegate to underlying factory listen
-      return factory.listen(...args);
-    },
+    return {
+      async listen(
+        ...args: Parameters<FoundationSimulator<any>["listen"]>
+      ): Promise<any> {
+        if (startDelay > 0) {
+          await new Promise((resolve) => setTimeout(resolve, startDelay));
+        }
+        // delegate to underlying factory listen
+        return factory.listen(...args);
+      },
+    };
   };
 }

packages/server/example/simulation-graph.ts

@@ -14,7 +14,9 @@ const servicesMap = {
   },
 };
 
-export const services = useServiceGraph(servicesMap);
+export const services = useServiceGraph(servicesMap, {
+  globalData: { exampleKey: "exampleValue" },
+});
 
 export function example(opts: { duration?: number } = {}) {
   return (function* () {

packages/server/src/data-service.ts

@@ -0,0 +1,82 @@
+import { call, resource, type Operation } from "effection";
+import { createServer } from "node:http";
+import { stdout } from "./logging.ts";
+
+export type DataServiceOptions = Record<string, unknown> | undefined;
+
+export function startDataService(
+  data: DataServiceOptions = {}
+): Operation<{ port: number }> {
+  return resource(function* (provide) {
+    const server = createServer((req, res) => {
+      try {
+        const url = new URL(req.url ?? "", `http://127.0.0.1`);
+        const pathname = url.pathname;
+
+        // GET /data -> whole object
+        if (
+          req.method === "GET" &&
+          (pathname === "/data" || pathname === "/")
+        ) {
+          const body = JSON.stringify(data || {});
+          res.writeHead(200, {
+            "content-type": "application/json",
+            "content-length": String(Buffer.byteLength(body)),
+          });
+          res.end(body);
+          return;
+        }
+
+        // GET /data/<key> -> value or 404
+        if (req.method === "GET" && pathname.startsWith("/data/")) {
+          const key = decodeURIComponent(pathname.replace(/^\/data\//, ""));
+          if (!key) {
+            res.writeHead(400);
+            res.end();
+            return;
+          }
+
+          const value = (data as any)?.[key];
+          if (value === undefined) {
+            res.writeHead(404, { "content-type": "text/plain" });
+            res.end("not found");
+            return;
+          }
+
+          const body = JSON.stringify(value);
+          res.writeHead(200, {
+            "content-type": "application/json",
+            "content-length": String(Buffer.byteLength(body)),
+          });
+          res.end(body);
+          return;
+        }
+
+        // unknown endpoint
+        res.writeHead(404, { "content-type": "text/plain" });
+        res.end("not found");
+      } catch (err) {
+        res.writeHead(500, { "content-type": "text/plain" });
+        res.end(String(err));
+      }
+    });
+
+    // listen on ephemeral port bound to localhost
+    yield* call(() => server.listen());
+
+    const address = server.address();
+    const port =
+      typeof address === "object" && address !== null && "port" in address
+        ? address.port
+        : 0;
+
+    yield* stdout(`data service: started on port ${port}`);
+
+    try {
+      yield* provide({ port });
+    } finally {
+      yield* call(() => server.close());
+      yield* stdout(`data service: stopped on port ${port}`);
+    }
+  });
+}

packages/server/src/services.ts

@@ -1,15 +1,19 @@
 import {
   type Operation,
+  type Stream,
+  type WithResolvers,
   resource,
   spawn,
   withResolvers,
   each,
-  type Stream,
-  type WithResolvers,
+  createContext,
 } from "effection";
 
 import { type ServiceUpdate, useWatcher } from "./watch.ts";
 import { stdout } from "./logging.ts";
+import { startDataService } from "./data-service.ts";
+
+export const SimulacrumEndpoint = createContext<number>("SimulacrumEndpoint");
 
 export type ServiceDefinition<
   S,
@@ -64,7 +68,11 @@ export function useServiceGraph<
   T extends MaybeSimulation
 >(
   services: S,
-  options?: { watch?: boolean; watchDebounce?: number }
+  options?: {
+    globalData?: Record<string, unknown>;
+    watch?: boolean;
+    watchDebounce?: number;
+  }
 ): (subset?: string[] | string) => Operation<ServiceGraph<S, T>> {
   return (subset?: string[] | string) =>
     resource(function* (provide) {
@@ -123,6 +131,16 @@ export function useServiceGraph<
           )}`
         );
       }
+      // track service ports (when services expose one)
+      const servicePorts = new Map<string, number>();
+
+      const dataServiceProvided = yield* startDataService(
+        options?.globalData ?? {}
+      );
+      servicePorts.set("simulacrum", dataServiceProvided.port);
+      // set the SimulacrumEndpoint in this operation scope so children started
+      // in this graph can access the port via context
+      yield* SimulacrumEndpoint.set(dataServiceProvided.port);
 
       const watcher = yield* useWatcher(
         effectiveServices,
@@ -147,9 +165,6 @@ export function useServiceGraph<
         }
       }
 
-      // track service ports (when services expose one)
-      const servicePorts = new Map<string, number>();
-
       function bumpService(service: string) {
         const task = status.get(service);
         if (!task) throw new Error(`missing status for service '${service}'`);

packages/server/src/simulation.ts

@@ -6,6 +6,7 @@ import type {
   FoundationSimulator,
   FoundationSimulatorListening,
 } from "@simulacrum/foundation-simulator";
+import { SimulacrumEndpoint } from "./services.ts";
 
 /**
  * Helper to start a foundation simulation server factory
@@ -42,15 +43,20 @@ export function useChildSimulation<L extends object = Record<string, unknown>>(
   modulePath: string
 ): Operation<FoundationSimulatorListening<L>> {
   return resource(function* (provide) {
-    const cmd = [
+    // attempt to read the simulacrum port from context; if not present, continue without it
+    const port = yield* SimulacrumEndpoint.get();
+
+    const parts = [
       "node",
       "--import",
       "tsx",
       "./bin/run-simulation-child.ts",
       modulePath,
-    ]
-      .map((s) => (s.includes(" ") ? `'${s}'` : s))
-      .join(" ");
+    ];
+    if (typeof port === "number") {
+      parts.push("--simulacrum-port", String(port));
+    }
+    const cmd = parts.map((s) => (s.includes(" ") ? `'${s}'` : s)).join(" ");
 
     const process = yield* exec(cmd);
 

packages/server/test/data-service.test.ts

@@ -0,0 +1,31 @@
+import { it } from "node:test";
+import assert from "node:assert";
+import { run, sleep, until } from "effection";
+import { useServiceGraph } from "../src/services.ts";
+
+it("starts data service and serves configured data", async () => {
+  await run(function* () {
+    const runGraph = yield* useServiceGraph(
+      {},
+      { globalData: { a: 1, nested: { b: 2 } } }
+    )();
+
+    let port: string | number | undefined = undefined;
+    for (let i = 0; i < 50 && !port; i++) {
+      if (runGraph && runGraph.servicePorts) {
+        port = runGraph.servicePorts.get("simulacrum");
+        if (typeof port === "number") break;
+      }
+      yield* sleep(10);
+    }
+
+    assert.ok(
+      typeof port === "number",
+      "data service port should be registered on servicePorts"
+    );
+
+    const res = yield* until(fetch(`http://127.0.0.1:${port}/data`));
+    const json = yield* until(res.json());
+    assert.deepStrictEqual(json, { a: 1, nested: { b: 2 } });
+  });
+});