feat(foundry): add managed data consumers and agent settings

2026-07-19 15:03:54 +03:00
parent 5f583caa05
commit aac44d057f
35 changed files with 4534 additions and 243 deletions
@@ -0,0 +1,336 @@
+import assert from "node:assert/strict";
+import { mkdtemp, readFile, readdir, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import test from "node:test";
+import { createFoundryDataProductConsumerManager } from "./foundry-data-product-consumer.mjs";
+
+const target = {
+  application: { id: "11111111-1111-4111-8111-111111111111" },
+  page: { id: "map" },
+  binding: {
+    id: "fleet-current",
+    dataProductId: "fleet.positions.current.v1",
+    slotId: "points",
+    delivery: "snapshot+patch",
+    semanticTypes: ["map.moving_object"],
+    fieldProjection: ["display_name", "operational_status"],
+  },
+};
+
+function fact({ longitude = 37.61, observedAt = "2026-07-19T10:00:00.000Z" } = {}) {
+  return {
+    sourceId: "vehicle-001",
+    semanticType: "map.moving_object",
+    observedAt,
+    receivedAt: observedAt,
+    attributes: { display_name: "Vehicle 001", operational_status: "active" },
+    geometry: { type: "Point", coordinates: [longitude, 55.75] },
+  };
+}
+
+function snapshot(cursor = "7", facts = [fact()]) {
+  return {
+    schemaVersion: "nodedc.data-product.snapshot/v1",
+    dataProduct: { id: "fleet.positions.current.v1", version: "1.0.0" },
+    generatedAt: "2026-07-19T10:00:01.000Z",
+    cursor,
+    facts,
+  };
+}
+
+function patch(cursor, previousCursor, operations) {
+  return {
+    schemaVersion: "nodedc.data-product.patch/v1",
+    dataProduct: { id: "fleet.positions.current.v1", version: "1.0.0" },
+    cursor,
+    previousCursor,
+    emittedAt: "2026-07-19T10:00:02.000Z",
+    operations,
+  };
+}
+
+function sseResponse(value, signal, counters) {
+  const encoder = new TextEncoder();
+  return new Response(new ReadableStream({
+    start(controller) {
+      counters.open += 1;
+      controller.enqueue(encoder.encode(`event: nodedc.data-product.patch.v1\ndata: ${JSON.stringify(value)}\n\n`));
+      const close = () => {
+        counters.closed += 1;
+        try { controller.close(); } catch { /* already closed */ }
+      };
+      signal.addEventListener("abort", close, { once: true });
+    },
+  }), { status: 200, headers: { "content-type": "text/event-stream" } });
+}
+
+async function waitUntil(check, label, timeoutMs = 2_000) {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (await check()) return;
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+  throw new Error(`timeout:${label}`);
+}
+
+test("server-owned consumer persists cursor, shares streams, transitions stale and removes canonically", async () => {
+  const stateDir = await mkdtemp(join(tmpdir(), "foundry-consumer-"));
+  let nowMs = Date.parse("2026-07-19T10:00:10.000Z");
+  let currentSnapshot = snapshot();
+  const streamPatches = [
+    patch("8", "7", [{ op: "upsert", fact: fact({ longitude: 37.62 }) }]),
+    patch("9", "8", [{
+      op: "remove",
+      sourceId: "vehicle-001",
+      semanticType: "map.moving_object",
+      removedAt: "2026-07-19T10:01:30.000Z",
+      reason: "tombstone",
+    }]),
+  ];
+  const counters = { catalog: 0, snapshot: 0, stream: 0, open: 0, closed: 0 };
+  const fetchImpl = async (input, options = {}) => {
+    const url = new URL(input);
+    assert.equal(options.headers.authorization, "Bearer ndc_edprb_test-reader-capability");
+    assert.equal(url.origin, "http://edp.test");
+    if (url.pathname.endsWith("/reader/data-products")) {
+      counters.catalog += 1;
+      return Response.json({
+        ok: true,
+        dataProducts: [{
+          id: "fleet.positions.current.v1",
+          version: "1.0.0",
+          deliveryMode: "snapshot+patch",
+          semanticTypes: ["map.moving_object"],
+          active: true,
+        }],
+      });
+    }
+    if (url.pathname.endsWith("/snapshot")) {
+      counters.snapshot += 1;
+      return Response.json(currentSnapshot);
+    }
+    if (url.pathname.endsWith("/stream")) {
+      counters.stream += 1;
+      const next = streamPatches.shift();
+      assert.ok(next, "unexpected extra upstream stream");
+      assert.equal(url.searchParams.get("after"), next.previousCursor);
+      return sseResponse(next, options.signal, counters);
+    }
+    throw new Error(`unexpected_url:${url}`);
+  };
+  const sanitizeSnapshot = (value) => value;
+  const sanitizePatch = (value) => value;
+  const manager = createFoundryDataProductConsumerManager({
+    stateDir,
+    dataPlaneUrl: "http://edp.test",
+    resolveTarget: async () => target,
+    readReaderToken: async () => "ndc_edprb_test-reader-capability",
+    resolvePolicy: (product) => ({
+      id: "map-moving-object-current-v1",
+      version: "1.0.0",
+      dataProductId: product.id,
+      productVersion: product.version,
+      staleAfterMs: 60_000,
+      terminalStatuses: ["inactive", "no-position", "no_position"],
+      removeMode: "canonical-tombstone-or-snapshot-rebase",
+    }),
+    sanitizeSnapshot,
+    sanitizePatch,
+    fetchImpl,
+    now: () => nowMs,
+    idleStopMs: 20,
+    staleSweepMs: 10_000,
+    reconnectMinMs: 10,
+    reconnectMaxMs: 20,
+  });
+
+  try {
+    const input = { applicationId: target.application.id, pageId: target.page.id, bindingId: target.binding.id };
+    const plan = await manager.plan(input);
+    assert.equal(plan.action, "create");
+    assert.match(plan.planId, /^fcp1_/);
+    assert.equal(JSON.stringify(plan).includes("ndc_edprb_"), false);
+    assert.equal(JSON.stringify(plan).includes("http://edp.test"), false);
+
+    const applied = await manager.apply({ ...input, planId: plan.planId });
+    assert.equal(applied.consumer.cursor, "7");
+    assert.equal(applied.consumer.subjectCount, 1);
+    assert.equal(applied.consumer.runtimeState, "idle");
+
+    const events = [];
+    const first = await manager.subscribe({ ...target, after: "7" }, (event) => events.push(event));
+    const second = await manager.subscribe({ ...target, after: "7" }, (event) => events.push(event));
+    await waitUntil(async () => (await manager.status(input)).consumer.cursor === "8", "patch_cursor_8");
+    assert.equal(counters.stream, 1, "two viewers must share one upstream stream");
+    assert.equal((await manager.status(input)).consumer.subjectCount, 1);
+    assert.equal(events.filter((event) => event.event === "nodedc.data-product.patch.v1").length, 2, "one committed patch is fanned out once per viewer");
+
+    nowMs = Date.parse("2026-07-19T10:01:30.001Z");
+    const stale = await manager.status(input);
+    assert.equal(stale.consumer.subjects[0].status, "stale");
+    assert.equal(stale.consumer.metrics.staleTransitions, 1);
+
+    first.release();
+    second.release();
+    await waitUntil(() => counters.closed === 1, "last_viewer_closes_upstream");
+
+    const removeEvents = [];
+    const third = await manager.subscribe({ ...target, after: "8" }, (event) => removeEvents.push(event));
+    await waitUntil(async () => (await manager.status(input)).consumer.cursor === "9", "remove_cursor_9");
+    assert.equal((await manager.status(input)).consumer.subjectCount, 0);
+    assert.equal((await manager.status(input)).consumer.metrics.canonicalRemovals, 1);
+    assert.ok(removeEvents.some((event) => event.data?.operations?.some((operation) => operation.op === "remove")));
+    third.release();
+    await waitUntil(() => counters.closed === 2, "remove_stream_closed");
+
+    currentSnapshot = snapshot("9", []);
+    const refreshPlan = await manager.plan(input);
+    await manager.apply({ ...input, planId: refreshPlan.planId });
+    const empty = await manager.snapshot(target);
+    assert.deepEqual(empty.facts, []);
+
+    const rolledBack = await manager.rollback(input);
+    assert.equal(rolledBack.snapshotPreserved, true);
+    assert.equal(rolledBack.consumer.runtimeState, "stopped");
+    await assert.rejects(() => manager.snapshot(target), /data_product_consumer_stopped/);
+
+    const stateFiles = await readdir(stateDir);
+    assert.equal(stateFiles.length, 1);
+    const persisted = await readFile(join(stateDir, stateFiles[0]), "utf8");
+    assert.equal(persisted.includes("ndc_edprb_"), false);
+    assert.equal(persisted.includes("http://edp.test"), false);
+  } finally {
+    await manager.shutdown();
+    await rm(stateDir, { recursive: true, force: true });
+  }
+});
+
+test("restart restores the committed cursor and transient stream errors preserve subjects", async () => {
+  const stateDir = await mkdtemp(join(tmpdir(), "foundry-consumer-restart-"));
+  const input = { applicationId: target.application.id, pageId: target.page.id, bindingId: target.binding.id };
+  let streamAttempts = 0;
+  const fetchImpl = async (request, options = {}) => {
+    const url = new URL(request);
+    assert.equal(options.headers.authorization, "Bearer ndc_edprb_test-reader-capability");
+    if (url.pathname.endsWith("/reader/data-products")) {
+      return Response.json({ ok: true, dataProducts: [{
+        id: "fleet.positions.current.v1",
+        version: "1.0.0",
+        deliveryMode: "snapshot+patch",
+        semanticTypes: ["map.moving_object"],
+        active: true,
+      }] });
+    }
+    if (url.pathname.endsWith("/snapshot")) return Response.json(snapshot());
+    if (url.pathname.endsWith("/stream")) {
+      streamAttempts += 1;
+      return Response.json({ ok: false, error: "temporary_failure" }, { status: 503 });
+    }
+    throw new Error(`unexpected_url:${url}`);
+  };
+  const createManager = () => createFoundryDataProductConsumerManager({
+    stateDir,
+    dataPlaneUrl: "http://edp.test",
+    resolveTarget: async () => target,
+    readReaderToken: async () => "ndc_edprb_test-reader-capability",
+    resolvePolicy: (product) => ({
+      id: "map-moving-object-current-v1",
+      version: "1.0.0",
+      dataProductId: product.id,
+      productVersion: product.version,
+      staleAfterMs: 60_000,
+      terminalStatuses: ["inactive", "no-position", "no_position"],
+      removeMode: "canonical-tombstone-or-snapshot-rebase",
+    }),
+    sanitizeSnapshot: (value) => value,
+    sanitizePatch: (value) => value,
+    fetchImpl,
+    now: () => Date.parse("2026-07-19T10:00:10.000Z"),
+    idleStopMs: 10,
+    staleSweepMs: 10_000,
+    reconnectMinMs: 10,
+    reconnectMaxMs: 20,
+  });
+  const first = createManager();
+  let second;
+  try {
+    const plan = await first.plan(input);
+    await first.apply({ ...input, planId: plan.planId });
+    await first.shutdown();
+
+    second = createManager();
+    await second.resumePersisted();
+    const restored = await second.status(input);
+    assert.equal(restored.consumer.cursor, "7");
+    assert.equal(restored.consumer.subjectCount, 1);
+    const lease = await second.subscribe({ ...target, after: "7" }, () => undefined);
+    await waitUntil(async () => (await second.status(input)).consumer.metrics.reconnects > 0, "transient_reconnect");
+    lease.release();
+    const afterFailure = await second.status(input);
+    assert.equal(afterFailure.consumer.subjectCount, 1);
+    assert.equal(afterFailure.consumer.cursor, "7");
+    assert.ok(streamAttempts >= 1);
+  } finally {
+    await first.shutdown();
+    await second?.shutdown();
+    await rm(stateDir, { recursive: true, force: true });
+  }
+});
+
+test("exact apply can ensure a missing target-scoped reader grant without exposing it in the plan", async () => {
+  const stateDir = await mkdtemp(join(tmpdir(), "foundry-consumer-grant-"));
+  const input = { applicationId: target.application.id, pageId: target.page.id, bindingId: target.binding.id };
+  const product = {
+    id: "fleet.positions.current.v1",
+    version: "1.0.0",
+    deliveryMode: "snapshot+patch",
+    semanticTypes: ["map.moving_object"],
+    active: true,
+  };
+  let grantReady = false;
+  let ensureCount = 0;
+  const manager = createFoundryDataProductConsumerManager({
+    stateDir,
+    dataPlaneUrl: "http://edp.test",
+    resolveTarget: async () => target,
+    readReaderToken: async () => grantReady ? "ndc_edprb_managed-reader-capability" : null,
+    inspectReaderGrant: async () => ({ product, readerGrantAction: grantReady ? "reuse" : "ensure" }),
+    ensureReaderGrant: async () => {
+      ensureCount += 1;
+      grantReady = true;
+      return { ensured: true };
+    },
+    resolvePolicy: (resolvedProduct) => ({
+      id: "map-moving-object-current-v1",
+      version: "1.0.0",
+      dataProductId: resolvedProduct.id,
+      productVersion: resolvedProduct.version,
+      staleAfterMs: 60_000,
+      terminalStatuses: ["inactive", "no-position", "no_position"],
+      removeMode: "canonical-tombstone-or-snapshot-rebase",
+    }),
+    sanitizeSnapshot: (value) => value,
+    sanitizePatch: (value) => value,
+    fetchImpl: async (request, options = {}) => {
+      assert.equal(options.headers.authorization, "Bearer ndc_edprb_managed-reader-capability");
+      assert.ok(new URL(request).pathname.endsWith("/snapshot"));
+      return Response.json(snapshot());
+    },
+  });
+  try {
+    const plan = await manager.plan(input);
+    assert.equal(plan.configuration.readerGrantAction, "ensure");
+    assert.ok(plan.effects.includes("ensure-target-scoped-reader-grant"));
+    assert.equal(JSON.stringify(plan).includes("capability"), false);
+    const applied = await manager.apply({ ...input, planId: plan.planId });
+    assert.equal(ensureCount, 1);
+    assert.equal(applied.consumer.subjectCount, 1);
+    const refresh = await manager.plan(input);
+    assert.equal(refresh.configuration.readerGrantAction, "reuse");
+    assert.equal(refresh.effects.includes("ensure-target-scoped-reader-grant"), false);
+  } finally {
+    await manager.shutdown();
+    await rm(stateDir, { recursive: true, force: true });
+  }
+});