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 firstStream = { opened: false, closed: false };
const shutdownStream = { opened: false, closed: false };
const internalAccessToken = "foundry-runtime-smoke-internal-access";

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 (await predicate()) return;
    await new Promise((resolve) => setTimeout(resolve, 25));
  }
  throw new Error(`${label}_timeout`);
}

async function readStreamUntil(response, pattern, timeoutMs = 5_000) {
  assert.equal(response.status, 200);
  assert.ok(response.body);
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let text = "";
  const deadline = Date.now() + timeoutMs;
  try {
    while (Date.now() < deadline) {
      const next = await Promise.race([
        reader.read(),
        new Promise((_, reject) => setTimeout(() => reject(new Error("stream_read_timeout")), 250)),
      ]);
      if (next.done) break;
      text += decoder.decode(next.value, { stream: true });
      if (pattern.test(text)) return text;
    }
    throw new Error("stream_pattern_timeout");
  } finally {
    await reader.cancel().catch(() => undefined);
    reader.releaseLock();
  }
}

function readHttpStreamUntil(url, pattern, timeoutMs = 5_000) {
  return new Promise((resolve, reject) => {
    let text = "";
    const timer = setTimeout(() => {
      request.destroy();
      reject(new Error("http_stream_pattern_timeout"));
    }, timeoutMs);
    const request = httpRequest(url, { headers: { accept: "text/event-stream" } }, (response) => {
      response.setEncoding("utf8");
      response.on("data", (chunk) => {
        text += chunk;
        if (!pattern.test(text)) return;
        clearTimeout(timer);
        response.destroy();
        request.destroy();
        resolve(text);
      });
      response.once("error", (error) => {
        if (pattern.test(text)) return;
        clearTimeout(timer);
        reject(error);
      });
    });
    request.once("error", (error) => {
      if (pattern.test(text)) return;
      clearTimeout(timer);
      reject(error);
    });
    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/history?from=2026-07-15T12%3A00%3A00.000Z&to=2026-07-15T12%3A05%3A00.000Z&resolutionMs=60000&limit=1000&sourceIds=vehicle-001") {
      response.writeHead(200, { "content-type": "application/json" });
      response.end(JSON.stringify({
        schemaVersion: "nodedc.data-product.history/v1",
        dataProduct: { id: "fleet.positions.current.v1", version: "1.0.0" },
        generatedAt: "2026-07-15T12:05:01.000Z",
        query: {
          from: "2026-07-15T12:00:00.000Z",
          to: "2026-07-15T12:05:00.000Z",
          resolutionMs: 60000,
          sourceIds: ["vehicle-001"],
          order: "asc",
        },
        facts: [{
          ...canonicalFact({ longitude: 37.61, name: "Vehicle 001 historical" }),
          bucketStart: "2026-07-15T12:00:00.000Z",
        }],
        nextCursor: "opaque_history_cursor",
      }));
      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" });
      firstStream.opened = true;
      const heartbeat = setInterval(() => response.write(": keepalive\n\n"), 50);
      const closeFirstStream = () => {
        clearInterval(heartbeat);
        firstStream.closed = true;
      };
      response.once("close", closeFirstStream);
      request.socket.once("close", closeFirstStream);
      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`);
      return;
    }
    if (request.url === "/internal/data-plane/v1/data-products/fleet.positions.current.v1/stream?after=8") {
      response.writeHead(200, { "content-type": "text/event-stream" });
      shutdownStream.opened = true;
      const heartbeat = setInterval(() => response.write(": keepalive\n\n"), 50);
      const closeShutdownStream = () => {
        clearInterval(heartbeat);
        shutdownStream.closed = true;
      };
      response.once("close", closeShutdownStream);
      request.socket.once("close", closeShutdownStream);
      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_INTERNAL_ACCESS_TOKEN: internalAccessToken,
      FOUNDRY_MCP_URL: `http://127.0.0.1:${foundryPort}/api/mcp`,
      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 entitlementResponse = await fetch(`http://127.0.0.1:${foundryPort}/api/ai-workspace/entitlements`, {
    method: "POST",
    headers: { authorization: `Bearer ${internalAccessToken}`, "content-type": "application/json" },
    body: JSON.stringify({
      schemaVersion: "ai-workspace.entitlement-request.v1",
      appId: "module-foundry",
      owner: { id: "user_root", key: "user_root" },
    }),
  });
  assert.equal(entitlementResponse.status, 200);
  const entitlement = await entitlementResponse.json();
  const mcpAuthorization = entitlement.appGrants["module-foundry"].mcpServers[0].httpHeaders.Authorization;
  let mcpRequestId = 0;
  const mcpRequest = async (method, params = {}) => {
    const response = await fetch(`http://127.0.0.1:${foundryPort}/api/mcp`, {
      method: "POST",
      headers: {
        authorization: mcpAuthorization,
        "content-type": "application/json",
        "mcp-protocol-version": "2025-06-18",
      },
      body: JSON.stringify({ jsonrpc: "2.0", id: ++mcpRequestId, method, params }),
    });
    assert.equal(response.status, 200);
    const payload = await response.json();
    assert.equal(payload.error, undefined);
    return payload.result;
  };
  const initialized = await mcpRequest("initialize", { protocolVersion: "2025-06-18", capabilities: {}, clientInfo: { name: "runtime-smoke", version: "1" } });
  assert.equal(initialized.serverInfo.version, "0.5.0");
  const listedTools = await mcpRequest("tools/list");
  assert.equal(listedTools.tools.length, 17);
  assert.ok(listedTools.tools.some((tool) => tool.name === "foundry_apply_map_data_product_consumer"));
  assert.ok(listedTools.tools.some((tool) => tool.name === "foundry_save_map_page_view_state"));
  assert.ok(listedTools.tools.some((tool) => tool.name === "foundry_remove_map_pin_binding"));

  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 consumerTarget = { applicationId, pageId, bindingId };
  const planCall = await mcpRequest("tools/call", {
    name: "foundry_plan_map_data_product_consumer",
    arguments: consumerTarget,
  });
  const consumerPlan = planCall.structuredContent;
  assert.equal(consumerPlan.action, "create");
  assert.match(consumerPlan.planId, /^fcp1_/);
  assert.equal(JSON.stringify(consumerPlan).includes(readerToken), false);
  assert.equal(JSON.stringify(consumerPlan).includes(`127.0.0.1:${edpPort}`), false);
  const applyCall = await mcpRequest("tools/call", {
    name: "foundry_apply_map_data_product_consumer",
    arguments: { ...consumerTarget, planId: consumerPlan.planId, idempotencyKey: "runtime-smoke-consumer-apply" },
  });
  assert.equal(applyCall.structuredContent.consumer.cursor, "7");
  assert.equal(applyCall.structuredContent.consumer.subjectCount, 1);

  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 historyResponse = await fetch(`${base}/history?from=2026-07-15T12%3A00%3A00.000Z&to=2026-07-15T12%3A05%3A00.000Z&resolutionMs=60000&sourceIds=vehicle-001&limit=1000`);
  assert.equal(historyResponse.status, 200);
  const history = await historyResponse.json();
  assert.equal(history.schemaVersion, "nodedc.data-product.history/v1");
  assert.equal(history.facts.length, 1);
  assert.equal(history.facts[0].bucketStart, "2026-07-15T12:00:00.000Z");
  assert.deepEqual(history.facts[0].attributes, { name: "Vehicle 001 historical" });
  assert.equal(history.nextCursor, "opaque_history_cursor");
  assert.equal(JSON.stringify(history).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 streamText = await readHttpStreamUntil(`${base}/stream?after=7`, /"cursor":"8"/);
  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);
  await waitUntil(async () => {
    const call = await mcpRequest("tools/call", {
      name: "foundry_get_map_data_product_consumer_status",
      arguments: consumerTarget,
    });
    return call.structuredContent.consumer.viewerCount === 0;
  }, "viewer_lease_release");
  await waitUntil(async () => {
    const call = await mcpRequest("tools/call", {
      name: "foundry_get_map_data_product_consumer_status",
      arguments: consumerTarget,
    });
    return call.structuredContent.consumer.upstreamStreamCount === 0;
  }, "shared_upstream_manager_release");
  await waitUntil(() => firstStream.closed, "shared_upstream_close_after_last_viewer");

  const statusCall = await mcpRequest("tools/call", {
    name: "foundry_get_map_data_product_consumer_status",
    arguments: consumerTarget,
  });
  assert.equal(statusCall.structuredContent.consumer.cursor, "8");
  assert.equal(statusCall.structuredContent.consumer.subjectCount, 1);
  assert.equal(statusCall.structuredContent.consumer.metrics.patchCommits, 1);
  assert.equal(statusCall.structuredContent.consumer.viewerCount, 0);
  const acceptanceCall = await mcpRequest("tools/call", {
    name: "foundry_accept_map_data_product_consumer",
    arguments: { ...consumerTarget, timeoutMs: 1000, minSubjectCount: 1, minPatchCount: 0 },
  });
  assert.equal(acceptanceCall.structuredContent.accepted, true);
  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 shutdownClient = await fetch(`${base}/stream?after=8`, { headers: { accept: "text/event-stream" } });
  assert.equal(shutdownClient.status, 200);
  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");
  await shutdownClient.body?.cancel().catch(() => undefined);
  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 });
}