NODEDC_DESIGN_GUIDELINE/scripts/smoke-data-product-runtime.mjs

import assert from "node:assert/strict";
import { createHash } from "node:crypto";
import { once } from "node:events";
import { mkdtemp, mkdir, readFile, rm, writeFile, chmod } from "node:fs/promises";
import { createServer, request as httpRequest } from "node:http";
import { spawn } from "node:child_process";
import { tmpdir } from "node:os";
import { join } from "node:path";

import {
  FOUNDRY_BINDING_CATALOG_ACTION,
  FOUNDRY_BINDING_GRANT_SCHEMA_VERSION,
  FOUNDRY_BINDING_UPSERT_ACTION,
  FOUNDRY_BINDING_UPSERT_SCHEMA_VERSION,
  createFoundryBindingGrantToken,
  foundryBindingGrantFileName,
} from "../server/foundry-binding-api.mjs";

const root = new URL("..", import.meta.url).pathname;
const applicationId = "11111111-1111-4111-8111-111111111111";
const pageId = "map";
const bindingId = "fleet-positions";
const readerToken = `ndc_edprb_${"a".repeat(43)}`;
const targetKey = createHash("sha256").update(`${applicationId}/${pageId}/${bindingId}`, "utf8").digest("hex");
const runtimeDir = await mkdtemp(join(tmpdir(), "nodedc-foundry-runtime-"));
const grantDir = await mkdtemp(join(tmpdir(), "nodedc-foundry-reader-grants-"));
const bindingGrantDir = await mkdtemp(join(tmpdir(), "nodedc-foundry-binding-grants-"));
const bindingGrantToken = createFoundryBindingGrantToken();
let foundry;
let edp;
const pressure = { sent: 0, backpressured: false, closed: false };
const shutdownStream = { opened: false, closed: false };

function listen(server) {
  return new Promise((resolve) => {
    server.listen(0, "127.0.0.1", () => resolve(server.address().port));
  });
}

async function waitFor(url) {
  let lastError;
  for (let attempt = 0; attempt < 80; attempt += 1) {
    try {
      const response = await fetch(url, { cache: "no-store" });
      if (response.ok) return;
    } catch (error) {
      lastError = error;
    }
    await new Promise((resolve) => setTimeout(resolve, 50));
  }
  throw lastError || new Error("foundry_start_timeout");
}

async function waitUntil(predicate, label, timeoutMs = 5_000) {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    if (predicate()) return;
    await new Promise((resolve) => setTimeout(resolve, 25));
  }
  throw new Error(`${label}_timeout`);
}

function waitForDrainOrClose(response) {
  return new Promise((resolve) => {
    let settled = false;
    const settle = (value) => {
      if (settled) return;
      settled = true;
      response.off("drain", onDrain);
      response.off("close", onClose);
      response.off("error", onError);
      resolve(value);
    };
    const onDrain = () => settle(true);
    const onClose = () => settle(false);
    const onError = () => settle(false);
    response.once("drain", onDrain);
    response.once("close", onClose);
    response.once("error", onError);
  });
}

function openPausedStream(url) {
  return new Promise((resolve, reject) => {
    const request = httpRequest(url, { headers: { accept: "text/event-stream" } }, (response) => {
      response.pause();
      resolve({ request, response });
    });
    request.once("error", reject);
    request.end();
  });
}

function canonicalFact({ longitude, name }) {
  return {
    sourceId: "vehicle-001",
    semanticType: "map.moving_object",
    observedAt: "2026-07-15T12:00:00.000Z",
    receivedAt: "2026-07-15T12:00:01.000Z",
    attributes: { name, private: "must-not-reach-browser", token: "must-not-reach-browser" },
    geometry: { type: "Point", coordinates: [longitude, 55.75] },
  };
}

try {
  await mkdir(join(runtimeDir, "applications"), { recursive: true });
  const layout = JSON.parse(await readFile(join(root, "runtime-seed", "page-layouts", "map.json"), "utf8"));
  layout.dataProductBindings = [];
  await writeFile(join(runtimeDir, "applications", `${applicationId}.json`), `${JSON.stringify({
    schemaVersion: "0.1.0",
    id: applicationId,
    status: "draft",
    version: "0.1.0",
    metadata: { name: "Runtime test", slug: "runtime-test", description: "" },
    designProfile: { id: "default", version: "0.6.0", status: "published", theme: "dark" },
    pages: [{
      id: pageId,
      title: "Map",
      path: "/",
      template: { id: "map", version: "0.1.0" },
      navigation: { visible: true, label: "Map", order: 0 },
      features: { inspector: true, toolbar: true, assistant: false },
      layout: { map: layout },
    }],
    favicon: { source: "design-profile" },
    timestamps: { createdAt: "2026-07-15T12:00:00.000Z", updatedAt: "2026-07-15T12:00:00.000Z" },
  }, null, 2)}\n`, "utf8");
  await writeFile(join(grantDir, targetKey), `${readerToken}\n`, "utf8");
  await chmod(join(grantDir, targetKey), 0o400);
  const issuedAt = new Date(Date.now() - 60_000).toISOString();
  const expiresAt = new Date(Date.now() + 60 * 60_000).toISOString();
  const bindingGrantPath = join(bindingGrantDir, foundryBindingGrantFileName(bindingGrantToken));
  await writeFile(bindingGrantPath, `${JSON.stringify({
    schemaVersion: FOUNDRY_BINDING_GRANT_SCHEMA_VERSION,
    grantId: "foundry-runtime-smoke-grant",
    tokenHash: foundryBindingGrantFileName(bindingGrantToken),
    active: true,
    issuedAt,
    expiresAt,
    actorId: "engine-agent-smoke",
    ownerKey: "workspace-smoke",
    actions: [FOUNDRY_BINDING_CATALOG_ACTION, FOUNDRY_BINDING_UPSERT_ACTION],
    targets: [{
      applicationId,
      pageId,
      bindingId,
      dataProductId: "fleet.positions.current.v1",
      slotId: "points",
      semanticTypes: ["map.moving_object"],
      fieldProjection: ["name"],
    }],
  })}\n`, "utf8");
  await chmod(bindingGrantPath, 0o400);

  edp = createServer((request, response) => {
    assert.equal(request.headers.authorization, `Bearer ${readerToken}`);
    assert.equal(request.headers["x-nodedc-tenant-id"], undefined);
    assert.equal(request.headers["x-nodedc-connection-id"], undefined);
    if (request.url === "/internal/data-plane/v1/reader/data-products") {
      response.writeHead(200, { "content-type": "application/json" });
      response.end(JSON.stringify({
        ok: true,
        dataProducts: [{
          id: "fleet.positions.current.v1",
          version: "1.0.0",
          deliveryMode: "snapshot+patch",
          semanticTypes: ["map.moving_object"],
          active: true,
        }],
      }));
      return;
    }
    if (request.url === "/internal/data-plane/v1/data-products/fleet.positions.current.v1/snapshot") {
      response.writeHead(200, { "content-type": "application/json" });
      response.end(JSON.stringify({
        schemaVersion: "nodedc.data-product.snapshot/v1",
        dataProduct: { id: "fleet.positions.current.v1", version: "1.0.0" },
        generatedAt: "2026-07-15T12:00:01.000Z",
        cursor: "7",
        facts: [canonicalFact({ longitude: 37.61, name: "Vehicle 001" })],
      }));
      return;
    }
    if (request.url === "/internal/data-plane/v1/data-products/fleet.positions.current.v1/stream?after=1") {
      response.writeHead(409, { "content-type": "application/json" });
      response.end(JSON.stringify({ ok: false, error: "resync_required" }));
      return;
    }
    if (request.url === "/internal/data-plane/v1/data-products/fleet.positions.current.v1/stream?after=7") {
      response.writeHead(200, { "content-type": "text/event-stream" });
      response.write("event: nodedc.data-product.ready.v1\n");
      response.write(`data: ${JSON.stringify({ schemaVersion: "nodedc.data-product.ready/v1", dataProductId: "fleet.positions.current.v1", cursor: "7", emittedAt: "2026-07-15T12:00:01.000Z" })}\n\n`);
      response.write("id: 8\n");
      response.write("event: nodedc.data-product.patch.v1\n");
      response.write(`data: ${JSON.stringify({
        schemaVersion: "nodedc.data-product.patch/v1",
        dataProduct: { id: "fleet.positions.current.v1", version: "1.0.0" },
        cursor: "8",
        previousCursor: "7",
        emittedAt: "2026-07-15T12:00:02.000Z",
        operations: [{ op: "upsert", fact: canonicalFact({ longitude: 37.62, name: "Vehicle 001 updated" }) }],
      })}\n\n`);
      response.end();
      return;
    }
    if (request.url === "/internal/data-plane/v1/data-products/fleet.positions.current.v1/stream?after=9") {
      response.writeHead(200, { "content-type": "text/event-stream" });
      response.once("close", () => { pressure.closed = true; });
      void (async () => {
        const limit = 2_048;
        const largeName = "vehicle-position-".padEnd(16 * 1024, "x");
        for (let index = 0; index < limit && !response.destroyed; index += 1) {
          const cursor = String(10 + index);
          const previousCursor = String(9 + index);
          const frame = `id: ${cursor}\nevent: nodedc.data-product.patch.v1\ndata: ${JSON.stringify({
            schemaVersion: "nodedc.data-product.patch/v1",
            dataProduct: { id: "fleet.positions.current.v1", version: "1.0.0" },
            cursor,
            previousCursor,
            emittedAt: "2026-07-15T12:00:02.000Z",
            operations: [{ op: "upsert", fact: canonicalFact({ longitude: 37.62, name: largeName }) }],
          })}\n\n`;
          pressure.sent += 1;
          if (!response.write(frame)) {
            pressure.backpressured = true;
            if (!(await waitForDrainOrClose(response))) return;
          }
        }
        if (!response.destroyed) response.end();
      })();
      return;
    }
    if (request.url === "/internal/data-plane/v1/data-products/fleet.positions.current.v1/stream?after=10") {
      response.writeHead(200, { "content-type": "text/event-stream" });
      shutdownStream.opened = true;
      const heartbeat = setInterval(() => response.write(": keepalive\n\n"), 50);
      response.once("close", () => {
        clearInterval(heartbeat);
        shutdownStream.closed = true;
      });
      response.write(": keepalive\n\n");
      return;
    }
    response.writeHead(404).end();
  });
  const edpPort = await listen(edp);

  const foundryPortServer = createServer();
  const foundryPort = await listen(foundryPortServer);
  foundryPortServer.close();
  await once(foundryPortServer, "close");
  foundry = spawn(process.execPath, ["server/catalog-server.mjs"], {
    cwd: root,
    env: {
      ...process.env,
      NODE_ENV: "test",
      HOST: "127.0.0.1",
      PORT: String(foundryPort),
      FOUNDRY_RUNTIME_DIR: runtimeDir,
      NODEDC_FOUNDRY_AUTH_REQUIRED: "false",
      NODEDC_EXTERNAL_DATA_PLANE_INTERNAL_URL: `http://127.0.0.1:${edpPort}`,
      NODEDC_EXTERNAL_DATA_PLANE_READER_GRANTS_DIR: grantDir,
      NODEDC_FOUNDRY_BINDING_GRANTS_DIR: bindingGrantDir,
    },
    stdio: ["ignore", "pipe", "pipe"],
  });
  let stderr = "";
  foundry.stderr.on("data", (chunk) => { stderr += String(chunk); });
  await waitFor(`http://127.0.0.1:${foundryPort}/healthz`);

  const workloadHeaders = {
    authorization: `Bearer ${bindingGrantToken}`,
    "content-type": "application/json",
  };
  const catalogResponse = await fetch(`http://127.0.0.1:${foundryPort}/internal/foundry/v1/data-products`, {
    headers: { authorization: workloadHeaders.authorization },
  });
  assert.equal(catalogResponse.status, 200);
  assert.deepEqual(await catalogResponse.json(), {
    ok: true,
    dataProducts: [{ id: "fleet.positions.current.v1", semanticTypes: ["map.moving_object"] }],
  });
  const bindingCommand = {
    schemaVersion: FOUNDRY_BINDING_UPSERT_SCHEMA_VERSION,
    applicationId,
    pageId,
    idempotencyKey: "foundry-runtime-smoke-upsert",
    binding: {
      id: bindingId,
      dataProductId: "fleet.positions.current.v1",
      slotId: "points",
      semanticTypes: ["map.moving_object"],
      fieldProjection: ["name"],
    },
  };
  const bindingResponse = await fetch(`http://127.0.0.1:${foundryPort}/internal/foundry/v1/data-product-bindings`, {
    method: "POST",
    headers: workloadHeaders,
    body: JSON.stringify(bindingCommand),
  });
  assert.equal(bindingResponse.status, 200);
  assert.equal((await bindingResponse.json()).idempotency.replayed, false);
  const replayResponse = await fetch(`http://127.0.0.1:${foundryPort}/internal/foundry/v1/data-product-bindings`, {
    method: "POST",
    headers: workloadHeaders,
    body: JSON.stringify(bindingCommand),
  });
  assert.equal(replayResponse.status, 200);
  assert.equal((await replayResponse.json()).idempotency.replayed, true);

  const base = `http://127.0.0.1:${foundryPort}/api/applications/${applicationId}/pages/${pageId}/data-bindings/${bindingId}`;
  const snapshotResponse = await fetch(`${base}/snapshot`);
  assert.equal(snapshotResponse.status, 200);
  const snapshot = await snapshotResponse.json();
  assert.equal(snapshot.cursor, "7");
  assert.equal(snapshot.facts.length, 1);
  assert.deepEqual(snapshot.facts[0].attributes, { name: "Vehicle 001" });
  assert.equal(JSON.stringify(snapshot).includes("must-not-reach-browser"), false);

  const resyncResponse = await fetch(`${base}/stream?after=1`, { headers: { accept: "text/event-stream" } });
  assert.equal(resyncResponse.status, 200);
  assert.match(await resyncResponse.text(), /nodedc\.data-product\.resync-required\.v1/);

  const streamResponse = await fetch(`${base}/stream?after=7`, { headers: { accept: "text/event-stream" } });
  assert.equal(streamResponse.status, 200);
  const streamText = await streamResponse.text();
  assert.match(streamText, /event: nodedc\.data-product\.ready\.v1/);
  assert.match(streamText, /event: nodedc\.data-product\.patch\.v1/);
  assert.match(streamText, /"cursor":"8"/);
  assert.match(streamText, /Vehicle 001 updated/);
  assert.equal(streamText.includes("must-not-reach-browser"), false);
  assert.equal(stderr, "");

  await rm(join(grantDir, targetKey));
  const denied = await fetch(`${base}/snapshot`);
  assert.equal(denied.status, 403);
  assert.deepEqual(await denied.json(), { error: "data_product_reader_grant_not_found" });

  await writeFile(join(grantDir, targetKey), `${readerToken}\n`, "utf8");
  await chmod(join(grantDir, targetKey), 0o400);
  const paused = await openPausedStream(`${base}/stream?after=9`);
  await waitUntil(() => pressure.backpressured, "upstream_backpressure");
  await waitUntil(() => pressure.sent > 0 && pressure.sent < 2_048, "bounded_upstream_read");
  await new Promise((resolve) => setTimeout(resolve, 150));
  assert.ok(pressure.sent < 2_048, "Foundry must stop draining the upstream while the browser is backpressured");
  paused.response.destroy();
  paused.request.destroy();
  await waitUntil(() => pressure.closed, "upstream_close_after_browser_disconnect");

  const shutdownClient = await openPausedStream(`${base}/stream?after=10`);
  await waitUntil(() => shutdownStream.opened, "shutdown_stream_open");
  const exit = once(foundry, "exit");
  foundry.kill("SIGTERM");
  const [exitCode, exitSignal] = await Promise.race([
    exit,
    new Promise((_, reject) => setTimeout(() => reject(new Error("foundry_bounded_shutdown_timeout")), 2_000)),
  ]);
  assert.equal(exitCode, 0);
  assert.equal(exitSignal, null);
  await waitUntil(() => shutdownStream.closed, "upstream_close_after_shutdown");
  shutdownClient.response.destroy();
  shutdownClient.request.destroy();
  assert.equal(stderr, "");
  console.log("foundry data-product runtime BFF: ok");
} finally {
  if (foundry && !foundry.killed) {
    foundry.kill("SIGTERM");
    await once(foundry, "exit").catch(() => undefined);
  }
  if (edp) await new Promise((resolve) => edp.close(resolve));
  await rm(runtimeDir, { recursive: true, force: true });
  await rm(grantDir, { recursive: true, force: true });
  await rm(bindingGrantDir, { recursive: true, force: true });
}