865 lines
35 KiB
JavaScript
865 lines
35 KiB
JavaScript
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 });
|
|
}
|
|
});
|
|
|
|
test("Data Product replacement bootstraps a successor reader generation before revoking its predecessor", async () => {
|
|
const stateDir = await mkdtemp(join(tmpdir(), "foundry-consumer-generation-"));
|
|
const input = { applicationId: target.application.id, pageId: target.page.id, bindingId: target.binding.id };
|
|
let currentTarget = structuredClone(target);
|
|
let failV2Snapshot = true;
|
|
let failPredecessorRevoke = true;
|
|
const grants = new Map([[1, "fleet.positions.current.v1"]]);
|
|
const events = [];
|
|
const product = (id) => ({
|
|
id,
|
|
version: id.endsWith(".v2") ? "2.0.0" : "1.0.0",
|
|
deliveryMode: "snapshot+patch",
|
|
semanticTypes: ["map.moving_object"],
|
|
active: true,
|
|
});
|
|
const manager = createFoundryDataProductConsumerManager({
|
|
stateDir,
|
|
dataPlaneUrl: "http://edp.test",
|
|
resolveTarget: async () => structuredClone(currentTarget),
|
|
readReaderToken: async (_applicationId, _pageId, _bindingId, generation = 1) => {
|
|
return grants.has(generation) ? `ndc_edprb_generation-${generation}` : null;
|
|
},
|
|
inspectReaderGrant: async (resolvedTarget, { generation = 1 } = {}) => {
|
|
const desiredProductId = resolvedTarget.binding.dataProductId;
|
|
return {
|
|
product: product(desiredProductId),
|
|
readerGrantAction: grants.get(generation) === desiredProductId ? "reuse" : "ensure",
|
|
readerGrantGeneration: generation,
|
|
};
|
|
},
|
|
ensureReaderGrant: async (resolvedTarget, { generation }) => {
|
|
grants.set(generation, resolvedTarget.binding.dataProductId);
|
|
events.push(`ensure-g${generation}`);
|
|
return { ensured: true, generation };
|
|
},
|
|
revokeReaderGrant: async (_resolvedTarget, { generation }) => {
|
|
if (generation === 1 && failPredecessorRevoke) {
|
|
failPredecessorRevoke = false;
|
|
events.push("revoke-g1-failed");
|
|
throw Object.assign(new Error("foundry_reader_grant_provisioner_unavailable"), { statusCode: 503 });
|
|
}
|
|
grants.delete(generation);
|
|
events.push(`revoke-g${generation}`);
|
|
return { revoked: true, generation };
|
|
},
|
|
resolvePolicy: (resolvedProduct) => ({
|
|
id: resolvedProduct.id.endsWith(".v2") ? "map-moving-object-current-v2" : "map-moving-object-current-v1",
|
|
version: resolvedProduct.version,
|
|
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 = {}) => {
|
|
const dataProductId = decodeURIComponent(new URL(request).pathname.split("/").at(-2));
|
|
const generation = Number(String(options.headers.authorization).match(/generation-(\d+)/)?.[1]);
|
|
assert.equal(grants.get(generation), dataProductId);
|
|
events.push(`snapshot-${dataProductId}-g${generation}`);
|
|
if (dataProductId.endsWith(".v2") && failV2Snapshot) {
|
|
throw Object.assign(new Error("data_product_runtime_unavailable"), { statusCode: 503 });
|
|
}
|
|
return Response.json({
|
|
...snapshot(dataProductId.endsWith(".v2") ? "20" : "10"),
|
|
dataProduct: { id: dataProductId, version: dataProductId.endsWith(".v2") ? "2.0.0" : "1.0.0" },
|
|
});
|
|
},
|
|
idleStopMs: 10,
|
|
staleSweepMs: 10_000,
|
|
});
|
|
try {
|
|
const createPlan = await manager.plan(input);
|
|
assert.equal(createPlan.configuration.readerGrantGeneration, 1);
|
|
const created = await manager.apply({ ...input, planId: createPlan.planId });
|
|
assert.equal(created.consumer.product.id, "fleet.positions.current.v1");
|
|
assert.equal(created.consumer.readerGrantGeneration, 1);
|
|
|
|
currentTarget.binding.dataProductId = "fleet.positions.current.v2";
|
|
currentTarget.binding.fieldProjection = [
|
|
"display_name", "availability_state", "motion_state", "position_state", "freshness_state",
|
|
];
|
|
const replacePlan = await manager.plan(input);
|
|
assert.equal(replacePlan.action, "replace");
|
|
assert.equal(replacePlan.configuration.readerGrantAction, "ensure");
|
|
assert.equal(replacePlan.configuration.readerGrantGeneration, 2);
|
|
assert.ok(replacePlan.effects.includes("ensure-successor-target-scoped-reader-grant"));
|
|
await assert.rejects(
|
|
manager.apply({ ...input, planId: replacePlan.planId }),
|
|
/data_product_runtime_unavailable/,
|
|
);
|
|
const preserved = await manager.status(input);
|
|
assert.equal(preserved.consumer.product.id, "fleet.positions.current.v1");
|
|
assert.equal(preserved.consumer.readerGrantGeneration, 1);
|
|
assert.equal(grants.get(1), "fleet.positions.current.v1");
|
|
assert.equal(grants.get(2), "fleet.positions.current.v2");
|
|
assert.equal(events.includes("revoke-g1"), false);
|
|
|
|
failV2Snapshot = false;
|
|
const retryPlan = await manager.plan(input);
|
|
assert.equal(retryPlan.configuration.readerGrantGeneration, 2);
|
|
await assert.rejects(
|
|
manager.apply({ ...input, planId: retryPlan.planId }),
|
|
/foundry_reader_grant_provisioner_unavailable/,
|
|
);
|
|
const pendingFinalization = await manager.status(input);
|
|
assert.equal(pendingFinalization.consumer.product.id, "fleet.positions.current.v2");
|
|
assert.equal(pendingFinalization.consumer.readerGrantGeneration, 2);
|
|
assert.equal(grants.has(1), true);
|
|
const finalizePlan = await manager.plan(input);
|
|
assert.equal(finalizePlan.action, "finalize-reader-grant-rotation");
|
|
assert.equal(finalizePlan.effects.includes("bootstrap-scoped-snapshot"), false);
|
|
assert.ok(finalizePlan.effects.includes("revoke-predecessor-reader-grant"));
|
|
const replaced = await manager.apply({ ...input, planId: finalizePlan.planId });
|
|
assert.equal(replaced.consumer.product.id, "fleet.positions.current.v2");
|
|
assert.equal(replaced.consumer.readerGrantGeneration, 2);
|
|
assert.equal(grants.has(1), false);
|
|
assert.equal(grants.get(2), "fleet.positions.current.v2");
|
|
assert.ok(events.lastIndexOf("snapshot-fleet.positions.current.v2-g2") < events.indexOf("revoke-g1"));
|
|
} finally {
|
|
await manager.shutdown();
|
|
await rm(stateDir, { recursive: true, force: true });
|
|
}
|
|
});
|
|
|
|
const movingObjectV4Policy = {
|
|
id: "map-moving-object-current-v4",
|
|
version: "4.0.0",
|
|
dataProductId: "fleet.positions.current.v4",
|
|
productVersion: "4.0.0",
|
|
staleAfterMs: null,
|
|
terminalStatuses: ["inactive"],
|
|
statusContract: {
|
|
attribute: "signal_state",
|
|
allowedValues: ["active", "inactive"],
|
|
missing: "reject",
|
|
freshness: "none",
|
|
},
|
|
removeMode: "canonical-tombstone-or-snapshot-rebase",
|
|
};
|
|
|
|
const movingObjectV4Product = {
|
|
id: "fleet.positions.current.v4",
|
|
version: "4.0.0",
|
|
deliveryMode: "snapshot+patch",
|
|
semanticTypes: ["map.moving_object"],
|
|
active: true,
|
|
};
|
|
|
|
function movingObjectV4Target(fieldProjection = ["display_name", "signal_state", "movement_state", "speed_kph"]) {
|
|
return {
|
|
application: { id: "44444444-4444-4444-8444-444444444444" },
|
|
page: { id: "map" },
|
|
binding: {
|
|
id: "fleet-current-v4",
|
|
dataProductId: movingObjectV4Product.id,
|
|
slotId: "points",
|
|
delivery: "snapshot+patch",
|
|
semanticTypes: ["map.moving_object"],
|
|
fieldProjection,
|
|
},
|
|
};
|
|
}
|
|
|
|
function movingObjectV4Snapshot(signalState = "active") {
|
|
return {
|
|
schemaVersion: "nodedc.data-product.snapshot/v1",
|
|
dataProduct: { id: movingObjectV4Product.id, version: movingObjectV4Product.version },
|
|
generatedAt: "2026-07-20T12:00:01.000Z",
|
|
cursor: "40",
|
|
facts: [{
|
|
sourceId: "gelios-unit-001",
|
|
semanticType: "map.moving_object",
|
|
observedAt: "2026-01-01T00:00:00.000Z",
|
|
receivedAt: "2026-07-20T12:00:00.000Z",
|
|
attributes: {
|
|
display_name: "Unit 001",
|
|
signal_state: signalState,
|
|
movement_state: "stopped",
|
|
speed_kph: 0,
|
|
},
|
|
geometry: { type: "Point", coordinates: [37.61, 55.75] },
|
|
}],
|
|
};
|
|
}
|
|
|
|
test("v4 consumer preserves the closed signal and movement states without inventing freshness", async () => {
|
|
const stateDir = await mkdtemp(join(tmpdir(), "foundry-consumer-v4-"));
|
|
const currentTarget = movingObjectV4Target();
|
|
let signalState = "active";
|
|
const manager = createFoundryDataProductConsumerManager({
|
|
stateDir,
|
|
dataPlaneUrl: "http://edp.test",
|
|
resolveTarget: async () => structuredClone(currentTarget),
|
|
readReaderToken: async () => "ndc_edprb_v4-reader-capability",
|
|
inspectReaderGrant: async () => ({ product: movingObjectV4Product, readerGrantAction: "reuse", readerGrantGeneration: 1 }),
|
|
resolvePolicy: () => movingObjectV4Policy,
|
|
sanitizeSnapshot: (value) => value,
|
|
sanitizePatch: (value) => value,
|
|
fetchImpl: async (request, options = {}) => {
|
|
assert.equal(options.headers.authorization, "Bearer ndc_edprb_v4-reader-capability");
|
|
assert.ok(new URL(request).pathname.endsWith("/snapshot"));
|
|
return Response.json(movingObjectV4Snapshot(signalState));
|
|
},
|
|
now: () => Date.parse("2026-07-20T12:00:10.000Z"),
|
|
});
|
|
try {
|
|
const input = { applicationId: currentTarget.application.id, pageId: currentTarget.page.id, bindingId: currentTarget.binding.id };
|
|
const plan = await manager.plan(input);
|
|
assert.deepEqual(plan.configuration.policy.statusContract.allowedValues, ["active", "inactive"]);
|
|
assert.equal(plan.configuration.policy.staleAfterMs, null);
|
|
const active = await manager.apply({ ...input, planId: plan.planId });
|
|
assert.equal(active.consumer.subjects[0].status, "active");
|
|
assert.equal(active.consumer.metrics.staleTransitions, 0);
|
|
assert.equal((await manager.snapshot(currentTarget)).facts[0].attributes.movement_state, "stopped");
|
|
|
|
signalState = "inactive";
|
|
const refreshPlan = await manager.plan(input);
|
|
const inactive = await manager.apply({ ...input, planId: refreshPlan.planId });
|
|
assert.equal(inactive.consumer.subjects[0].status, "inactive");
|
|
assert.equal(inactive.consumer.metrics.staleTransitions, 0);
|
|
} finally {
|
|
await manager.shutdown();
|
|
await rm(stateDir, { recursive: true, force: true });
|
|
}
|
|
});
|
|
|
|
test("v4 consumer rejects bindings that omit signal_state and snapshots outside the closed contract", async () => {
|
|
const omittedStateDir = await mkdtemp(join(tmpdir(), "foundry-consumer-v4-omitted-"));
|
|
const omittedTarget = movingObjectV4Target(["display_name", "movement_state"]);
|
|
const omittedManager = createFoundryDataProductConsumerManager({
|
|
stateDir: omittedStateDir,
|
|
dataPlaneUrl: "http://edp.test",
|
|
resolveTarget: async () => structuredClone(omittedTarget),
|
|
readReaderToken: async () => "ndc_edprb_v4-reader-capability",
|
|
inspectReaderGrant: async () => ({ product: movingObjectV4Product, readerGrantAction: "reuse", readerGrantGeneration: 1 }),
|
|
resolvePolicy: () => movingObjectV4Policy,
|
|
sanitizeSnapshot: (value) => value,
|
|
sanitizePatch: (value) => value,
|
|
});
|
|
try {
|
|
await assert.rejects(
|
|
omittedManager.plan({ applicationId: omittedTarget.application.id, pageId: omittedTarget.page.id, bindingId: omittedTarget.binding.id }),
|
|
/data_product_consumer_status_field_not_projected/,
|
|
);
|
|
} finally {
|
|
await omittedManager.shutdown();
|
|
await rm(omittedStateDir, { recursive: true, force: true });
|
|
}
|
|
|
|
const invalidStateDir = await mkdtemp(join(tmpdir(), "foundry-consumer-v4-invalid-"));
|
|
const invalidTarget = movingObjectV4Target();
|
|
const invalidManager = createFoundryDataProductConsumerManager({
|
|
stateDir: invalidStateDir,
|
|
dataPlaneUrl: "http://edp.test",
|
|
resolveTarget: async () => structuredClone(invalidTarget),
|
|
readReaderToken: async () => "ndc_edprb_v4-reader-capability",
|
|
inspectReaderGrant: async () => ({ product: movingObjectV4Product, readerGrantAction: "reuse", readerGrantGeneration: 1 }),
|
|
resolvePolicy: () => movingObjectV4Policy,
|
|
sanitizeSnapshot: (value) => value,
|
|
sanitizePatch: (value) => value,
|
|
fetchImpl: async () => Response.json(movingObjectV4Snapshot("unknown")),
|
|
});
|
|
try {
|
|
const input = { applicationId: invalidTarget.application.id, pageId: invalidTarget.page.id, bindingId: invalidTarget.binding.id };
|
|
const plan = await invalidManager.plan(input);
|
|
await assert.rejects(
|
|
invalidManager.apply({ ...input, planId: plan.planId }),
|
|
/data_product_consumer_fact_status_invalid/,
|
|
);
|
|
const failed = await invalidManager.status(input);
|
|
assert.equal(failed.consumer.runtimeState, "error");
|
|
assert.equal(failed.consumer.subjectCount, 0);
|
|
} finally {
|
|
await invalidManager.shutdown();
|
|
await rm(invalidStateDir, { recursive: true, force: true });
|
|
}
|
|
});
|
|
|
|
const zoneV2Projection = [
|
|
"display_name",
|
|
"geometry_kind",
|
|
"max_speed_kph",
|
|
"schedule_timezone",
|
|
"applies_to_couriers",
|
|
"applies_to_kicksharing",
|
|
];
|
|
|
|
const zoneV2Policy = {
|
|
id: "map-zone-current-v2",
|
|
version: "2.0.0",
|
|
dataProductId: "map.zones.current.v2",
|
|
productVersion: "2.0.0",
|
|
freshness: "none",
|
|
staleAfterMs: null,
|
|
terminalStatuses: [],
|
|
consumerContract: {
|
|
ontologyRevision: "ontology.map.zone.v1",
|
|
deliveryMode: "snapshot+patch",
|
|
semanticTypes: ["map.zone"],
|
|
fieldProjection: zoneV2Projection,
|
|
fieldTypes: {
|
|
display_name: "string",
|
|
geometry_kind: "string",
|
|
max_speed_kph: "number",
|
|
schedule_timezone: "string",
|
|
applies_to_couriers: "boolean",
|
|
applies_to_kicksharing: "boolean",
|
|
},
|
|
geometryTypes: ["Polygon", "MultiPolygon"],
|
|
subjectIdentity: "semantic-type+source-id",
|
|
snapshotMode: "atomic-replace",
|
|
patchMode: "atomic",
|
|
},
|
|
removeMode: "canonical-tombstone-or-snapshot-rebase",
|
|
};
|
|
|
|
const zoneV2Product = {
|
|
id: "map.zones.current.v2",
|
|
version: "2.0.0",
|
|
ontologyRevision: "ontology.map.zone.v1",
|
|
deliveryMode: "snapshot+patch",
|
|
semanticTypes: ["map.zone"],
|
|
fields: [...zoneV2Projection, "geometry", "source_revision"],
|
|
fieldContracts: {
|
|
display_name: { type: "string", required: true },
|
|
geometry_kind: { type: "string", required: true },
|
|
max_speed_kph: { type: "number", required: false },
|
|
schedule_timezone: { type: "string", required: false },
|
|
applies_to_couriers: { type: "boolean", required: false },
|
|
applies_to_kicksharing: { type: "boolean", required: false },
|
|
},
|
|
active: true,
|
|
};
|
|
|
|
function zoneV2Target(fieldProjection = zoneV2Projection) {
|
|
return {
|
|
application: { id: "55555555-5555-4555-8555-555555555555" },
|
|
page: { id: "map" },
|
|
binding: {
|
|
id: "depttrans-pmd-slow-zones",
|
|
dataProductId: zoneV2Product.id,
|
|
slotId: "points",
|
|
delivery: "snapshot+patch",
|
|
semanticTypes: ["map.zone"],
|
|
fieldProjection: [...fieldProjection],
|
|
},
|
|
};
|
|
}
|
|
|
|
function zoneFact({ sourceId = "depttrans-zone-001", geometryType = "Polygon", maxSpeedKph = 20 } = {}) {
|
|
const polygon = [[[37.60, 55.75], [37.61, 55.75], [37.61, 55.76], [37.60, 55.75]]];
|
|
return {
|
|
sourceId,
|
|
semanticType: "map.zone",
|
|
observedAt: "2026-07-21T10:00:00.000Z",
|
|
receivedAt: "2026-07-21T10:00:01.000Z",
|
|
attributes: {
|
|
display_name: "Slow zone 001",
|
|
geometry_kind: "polygon",
|
|
max_speed_kph: maxSpeedKph,
|
|
schedule_timezone: "Europe/Moscow",
|
|
applies_to_couriers: true,
|
|
applies_to_kicksharing: true,
|
|
},
|
|
geometry: geometryType === "MultiPolygon"
|
|
? { type: "MultiPolygon", coordinates: [polygon] }
|
|
: geometryType === "Polygon"
|
|
? { type: "Polygon", coordinates: polygon }
|
|
: { type: geometryType, coordinates: [37.61, 55.75] },
|
|
};
|
|
}
|
|
|
|
function zoneSnapshot({ cursor = "903", version = "2.0.0", facts = [zoneFact()] } = {}) {
|
|
return {
|
|
schemaVersion: "nodedc.data-product.snapshot/v1",
|
|
dataProduct: { id: zoneV2Product.id, version },
|
|
generatedAt: "2026-07-21T10:00:02.000Z",
|
|
cursor,
|
|
facts,
|
|
};
|
|
}
|
|
|
|
function zonePatch(cursor, previousCursor, operations, version = "2.0.0") {
|
|
return {
|
|
schemaVersion: "nodedc.data-product.patch/v1",
|
|
dataProduct: { id: zoneV2Product.id, version },
|
|
cursor,
|
|
previousCursor,
|
|
emittedAt: "2026-07-21T10:00:03.000Z",
|
|
operations,
|
|
};
|
|
}
|
|
|
|
test("zone v2 consumer pins provider-neutral contract and commits Polygon/MultiPolygon patches by stable subject identity", async () => {
|
|
const stateDir = await mkdtemp(join(tmpdir(), "foundry-consumer-zone-v2-"));
|
|
const currentTarget = zoneV2Target();
|
|
const streamCounters = { open: 0, closed: 0 };
|
|
const manager = createFoundryDataProductConsumerManager({
|
|
stateDir,
|
|
dataPlaneUrl: "http://edp.test",
|
|
resolveTarget: async () => structuredClone(currentTarget),
|
|
readReaderToken: async () => "ndc_edprb_zone-reader-capability",
|
|
inspectReaderGrant: async () => ({ product: zoneV2Product, readerGrantAction: "reuse", readerGrantGeneration: 1 }),
|
|
resolvePolicy: () => zoneV2Policy,
|
|
sanitizeSnapshot: (value) => value,
|
|
sanitizePatch: (value) => value,
|
|
fetchImpl: async () => Response.json(zoneSnapshot()),
|
|
openStream: async ({ signal }) => sseResponse(
|
|
zonePatch("904", "903", [{ op: "upsert", fact: zoneFact({ geometryType: "MultiPolygon", maxSpeedKph: 15 }) }]),
|
|
signal,
|
|
streamCounters,
|
|
),
|
|
now: () => Date.parse("2026-07-21T10:00:10.000Z"),
|
|
idleStopMs: 10,
|
|
reconnectMinMs: 10,
|
|
reconnectMaxMs: 20,
|
|
});
|
|
try {
|
|
const input = { applicationId: currentTarget.application.id, pageId: currentTarget.page.id, bindingId: currentTarget.binding.id };
|
|
const plan = await manager.plan(input);
|
|
assert.equal(plan.configuration.product.version, "2.0.0");
|
|
assert.deepEqual(plan.configuration.policy.consumerContract.semanticTypes, ["map.zone"]);
|
|
assert.deepEqual(plan.configuration.policy.consumerContract.fieldProjection, zoneV2Projection);
|
|
assert.deepEqual(plan.configuration.policy.consumerContract.geometryTypes, ["Polygon", "MultiPolygon"]);
|
|
assert.equal(JSON.stringify(plan).includes("http://edp.test"), false);
|
|
assert.equal(JSON.stringify(plan).includes("ndc_edprb_"), false);
|
|
|
|
const applied = await manager.apply({ ...input, planId: plan.planId });
|
|
assert.equal(applied.consumer.cursor, "903");
|
|
assert.equal(applied.consumer.subjectCount, 1);
|
|
assert.equal(applied.consumer.subjects[0].subjectId, "depttrans-zone-001");
|
|
assert.equal(applied.consumer.subjects[0].status, "active");
|
|
assert.equal(applied.consumer.metrics.staleTransitions, 0);
|
|
|
|
const lease = await manager.subscribe({ ...currentTarget, after: "903" }, () => undefined);
|
|
await waitUntil(async () => (await manager.status(input)).consumer.cursor === "904", "zone_patch_cursor_904");
|
|
const patched = await manager.snapshot(currentTarget);
|
|
assert.equal(patched.facts.length, 1);
|
|
assert.equal(patched.facts[0].sourceId, "depttrans-zone-001");
|
|
assert.equal(patched.facts[0].geometry.type, "MultiPolygon");
|
|
assert.equal(patched.facts[0].attributes.max_speed_kph, 15);
|
|
assert.equal((await manager.status(input)).consumer.upstreamStreamCount, 1);
|
|
lease.release();
|
|
await waitUntil(() => streamCounters.closed === 1, "zone_stream_closed");
|
|
await waitUntil(async () => (await manager.status(input)).consumer.runtimeState === "idle", "zone_stream_idle");
|
|
} finally {
|
|
await manager.shutdown();
|
|
await rm(stateDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 20 });
|
|
}
|
|
});
|
|
|
|
test("zone v2 consumer fails closed on projection, product, envelope and geometry contract drift", async () => {
|
|
const cases = [
|
|
{
|
|
name: "projection",
|
|
target: zoneV2Target(zoneV2Projection.slice(0, -1)),
|
|
product: zoneV2Product,
|
|
policy: zoneV2Policy,
|
|
snapshot: zoneSnapshot(),
|
|
planError: /data_product_consumer_contract_mismatch/,
|
|
},
|
|
{
|
|
name: "field-type",
|
|
target: zoneV2Target(),
|
|
product: { ...zoneV2Product, fieldContracts: { ...zoneV2Product.fieldContracts, max_speed_kph: { type: "string" } } },
|
|
policy: zoneV2Policy,
|
|
snapshot: zoneSnapshot(),
|
|
planError: /data_product_consumer_contract_mismatch/,
|
|
},
|
|
{
|
|
name: "product-version",
|
|
target: zoneV2Target(),
|
|
product: { ...zoneV2Product, version: "2.0.1" },
|
|
policy: zoneV2Policy,
|
|
snapshot: zoneSnapshot({ version: "2.0.1" }),
|
|
planError: /data_product_consumer_policy_not_found/,
|
|
},
|
|
{
|
|
name: "envelope-version",
|
|
target: zoneV2Target(),
|
|
product: zoneV2Product,
|
|
policy: zoneV2Policy,
|
|
snapshot: zoneSnapshot({ version: "2.0.1" }),
|
|
applyError: /data_product_snapshot_product_mismatch/,
|
|
},
|
|
{
|
|
name: "point-geometry",
|
|
target: zoneV2Target(),
|
|
product: zoneV2Product,
|
|
policy: zoneV2Policy,
|
|
snapshot: zoneSnapshot({ facts: [zoneFact({ geometryType: "Point" })] }),
|
|
applyError: /data_product_consumer_fact_contract_invalid/,
|
|
},
|
|
];
|
|
|
|
for (const variant of cases) {
|
|
const stateDir = await mkdtemp(join(tmpdir(), `foundry-consumer-zone-v2-${variant.name}-`));
|
|
const manager = createFoundryDataProductConsumerManager({
|
|
stateDir,
|
|
dataPlaneUrl: "http://edp.test",
|
|
resolveTarget: async () => structuredClone(variant.target),
|
|
readReaderToken: async () => "ndc_edprb_zone-reader-capability",
|
|
inspectReaderGrant: async () => ({ product: variant.product, readerGrantAction: "reuse", readerGrantGeneration: 1 }),
|
|
resolvePolicy: () => variant.policy,
|
|
sanitizeSnapshot: (value) => value,
|
|
sanitizePatch: (value) => value,
|
|
fetchImpl: async () => Response.json(variant.snapshot),
|
|
});
|
|
const input = { applicationId: variant.target.application.id, pageId: variant.target.page.id, bindingId: variant.target.binding.id };
|
|
try {
|
|
if (variant.planError) {
|
|
await assert.rejects(manager.plan(input), variant.planError);
|
|
continue;
|
|
}
|
|
const plan = await manager.plan(input);
|
|
await assert.rejects(manager.apply({ ...input, planId: plan.planId }), variant.applyError);
|
|
const failed = await manager.status(input);
|
|
assert.equal(failed.consumer.subjectCount, 0);
|
|
assert.equal(failed.consumer.runtimeState, "error");
|
|
} finally {
|
|
await manager.shutdown();
|
|
await rm(stateDir, { recursive: true, force: true });
|
|
}
|
|
}
|
|
});
|