feat(data-plane): add provider contracts and ontology delivery

This commit is contained in:
Codex
2026-07-16 02:23:34 +03:00
parent e527812826
commit 569b8762e6
84 changed files with 11170 additions and 70 deletions
+32
View File
@@ -0,0 +1,32 @@
FROM node:20-alpine AS deps
WORKDIR /workspace
COPY packages/external-provider-contract ./packages/external-provider-contract
COPY services/external-data-plane/package.json services/external-data-plane/package-lock.json ./services/external-data-plane/
WORKDIR /workspace/services/external-data-plane
RUN npm ci --omit=dev
FROM node:20-alpine AS runner
ENV NODE_ENV=production
ENV PORT=18106
WORKDIR /app
COPY --from=deps /workspace/services/external-data-plane/package.json /workspace/services/external-data-plane/package-lock.json ./
COPY --from=deps /workspace/services/external-data-plane/node_modules ./node_modules
COPY --from=deps /workspace/packages/external-provider-contract /packages/external-provider-contract
COPY services/external-data-plane/src ./src
COPY services/external-data-plane/definitions ./definitions
# A dedicated numeric identity can read only the EDP provisioning secret mount;
# it is intentionally not the shared Node/Map Gateway uid/gid (1000).
RUN addgroup -S -g 11006 nodedc-edp && adduser -S -D -H -u 11006 -G nodedc-edp nodedc-edp
USER 11006:11006
EXPOSE 18106
CMD ["node", "src/server.mjs"]
@@ -0,0 +1,30 @@
{
"id": "fleet.positions.current.v1",
"version": "1.0.0",
"ontologyRevision": "ontology.map.moving_object.v1",
"deliveryMode": "snapshot+patch",
"semanticTypes": [
"map.moving_object"
],
"fields": [
"course_degrees",
"display_name",
"elevation_meters",
"geometry",
"hdop",
"horizontal_accuracy_meters",
"object_kind",
"operational_status",
"position_source",
"position_valid",
"quality_flags",
"satellite_count",
"speed_kph"
],
"history": {
"mode": "sampled",
"intervalMs": 60000,
"strategy": "latest-per-entity-per-bucket",
"retentionDays": 90
}
}
File diff suppressed because it is too large Load Diff
+18
View File
@@ -0,0 +1,18 @@
{
"name": "@nodedc/external-data-plane",
"version": "0.1.0",
"private": true,
"type": "module",
"scripts": {
"start": "node src/server.mjs",
"dev": "node --watch src/server.mjs",
"check": "node --check src/server.mjs && node --check src/schema.mjs && node --check src/config.mjs && node --check src/intake-policy.mjs && node --check src/writer-binding.mjs && node --check src/reader-binding.mjs && node --check src/data-product-policy.mjs && node --check src/data-product-delivery.mjs && node --check src/definitions.mjs",
"test": "node test/config.test.mjs && node test/intake-policy.test.mjs && node test/writer-binding.test.mjs && node test/reader-binding.test.mjs && node test/data-product-policy.test.mjs && node test/definitions.test.mjs",
"test:integration": "node test/data-product-delivery.integration.test.mjs && node test/api.integration.test.mjs"
},
"dependencies": {
"@nodedc/external-provider-contract": "file:../../packages/external-provider-contract",
"express": "^5.2.1",
"pg": "^8.18.0"
}
}
@@ -0,0 +1,75 @@
import { readFileSync } from "node:fs";
export function readConfig(env = process.env) {
const provisionerApiEnabled = boolean(env.EXTERNAL_DATA_PLANE_PROVISIONING_ENABLED, false);
const config = {
port: integer(env.PORT, 18106, 1, 65535),
databaseUrl: required(env.EXTERNAL_DATA_PLANE_DATABASE_URL, "EXTERNAL_DATA_PLANE_DATABASE_URL"),
databasePoolSize: integer(env.EXTERNAL_DATA_PLANE_DATABASE_POOL_SIZE, 10, 1, 50),
internalAccessToken: optional(env.NODEDC_INTERNAL_ACCESS_TOKEN),
provisionerApiEnabled,
provisionerAccessToken: provisionerApiEnabled ? secretFile(env.EXTERNAL_DATA_PLANE_PROVISIONER_TOKEN_FILE) : "",
rawRetentionDays: integer(env.EXTERNAL_DATA_PLANE_RAW_RETENTION_DAYS, 14, 1, 3650),
maxBatchBytes: integer(env.EXTERNAL_DATA_PLANE_MAX_BATCH_BYTES, 5 * 1024 * 1024, 1024, 50 * 1024 * 1024),
maxFactsPerPublish: integer(env.EXTERNAL_DATA_PLANE_MAX_FACTS_PER_PUBLISH, 5000, 1, 100_000),
maxAttributesBytesPerFact: integer(env.EXTERNAL_DATA_PLANE_MAX_ATTRIBUTES_BYTES_PER_FACT, 64 * 1024, 256, 1024 * 1024),
maxPatchOperations: integer(env.EXTERNAL_DATA_PLANE_MAX_PATCH_OPERATIONS, 500, 1, 5000),
maxPatchBytes: integer(env.EXTERNAL_DATA_PLANE_MAX_PATCH_BYTES, 256 * 1024, 16 * 1024, 5 * 1024 * 1024),
patchRetentionMs: integer(env.EXTERNAL_DATA_PLANE_PATCH_RETENTION_MS, 60 * 60 * 1000, 60 * 1000, 7 * 24 * 60 * 60 * 1000),
receiptRetentionMs: integer(env.EXTERNAL_DATA_PLANE_RECEIPT_RETENTION_MS, 7 * 24 * 60 * 60 * 1000, 24 * 60 * 60 * 1000, 90 * 24 * 60 * 60 * 1000),
retentionDeleteLimit: integer(env.EXTERNAL_DATA_PLANE_RETENTION_DELETE_LIMIT, 10_000, 100, 100_000),
streamHeartbeatMs: integer(env.EXTERNAL_DATA_PLANE_STREAM_HEARTBEAT_MS, 20_000, 5_000, 60_000),
streamPollMs: integer(env.EXTERNAL_DATA_PLANE_STREAM_POLL_MS, 1_000, 250, 10_000),
maxReaderStreams: integer(env.EXTERNAL_DATA_PLANE_MAX_READER_STREAMS, 10, 1, 1000),
writerBindingMaxTtlDays: integer(env.EXTERNAL_DATA_PLANE_WRITER_BINDING_MAX_TTL_DAYS, 90, 1, 365),
maxFutureSkewSeconds: integer(env.EXTERNAL_DATA_PLANE_MAX_FUTURE_SKEW_SECONDS, 300, 0, 86400),
retentionSweepMs: integer(env.EXTERNAL_DATA_PLANE_RETENTION_SWEEP_MS, 60 * 60 * 1000, 60 * 1000, 24 * 60 * 60 * 1000),
legacyIntakeEnabled: boolean(env.EXTERNAL_DATA_PLANE_LEGACY_INTAKE_ENABLED, false),
};
if (config.internalAccessToken && config.provisionerAccessToken && config.internalAccessToken === config.provisionerAccessToken) {
throw new Error("provisioner_token_must_differ_from_internal_token");
}
return config;
}
function required(value, name) {
const normalized = optional(value);
if (!normalized) throw new Error(`${name}_required`);
return normalized;
}
function optional(value) {
const normalized = String(value ?? "").trim();
return normalized || "";
}
function integer(value, fallback, min, max) {
const candidate = optional(value);
if (!candidate) return fallback;
const number = Number.parseInt(candidate, 10);
if (!Number.isInteger(number) || number < min || number > max) throw new Error(`invalid_integer:${candidate}`);
return number;
}
function boolean(value, fallback) {
const normalized = optional(value).toLowerCase();
if (!normalized) return fallback;
if (normalized === "true") return true;
if (normalized === "false") return false;
throw new Error(`invalid_boolean:${normalized}`);
}
function secretFile(value) {
const path = optional(value);
if (!path) return "";
let secret = "";
try {
secret = readFileSync(path, "utf8").trim();
} catch {
throw new Error("external_data_plane_provisioner_token_file_unreadable");
}
if (!/^[A-Za-z0-9_-]{48,256}$/.test(secret)) {
throw new Error("external_data_plane_provisioner_token_file_invalid");
}
return secret;
}
@@ -0,0 +1,527 @@
import { createHash, randomUUID } from "node:crypto";
import {
DATA_PRODUCT_PATCH_SCHEMA_VERSION,
DATA_PRODUCT_SNAPSHOT_SCHEMA_VERSION,
} from "@nodedc/external-provider-contract";
export async function loadDataProductDefinition(db, dataProductId, { activeOnly = true } = {}) {
const result = await db.query(
`select id, version, ontology_revision as "ontologyRevision",
delivery_mode as "deliveryMode", semantic_types as "semanticTypes",
fields, history_policy as "historyPolicy", active,
created_at as "createdAt", updated_at as "updatedAt"
from external_data_plane_products
where id = $1 ${activeOnly ? "and active = true" : ""}`,
[dataProductId],
);
return result.rows[0] || null;
}
export async function persistDataProductDefinition(db, definition) {
const result = await db.query(
`insert into external_data_plane_products (
id, version, ontology_revision, delivery_mode, semantic_types, fields, history_policy
) values ($1, $2, $3, $4, $5::jsonb, $6::jsonb, $7::jsonb)
on conflict (id) do update set
active = true,
updated_at = now()
where external_data_plane_products.version = excluded.version
and external_data_plane_products.ontology_revision = excluded.ontology_revision
and external_data_plane_products.delivery_mode = excluded.delivery_mode
and external_data_plane_products.semantic_types = excluded.semantic_types
and external_data_plane_products.fields = excluded.fields
and external_data_plane_products.history_policy = excluded.history_policy
returning id, version, ontology_revision as "ontologyRevision",
delivery_mode as "deliveryMode", semantic_types as "semanticTypes",
fields, history_policy as "historyPolicy", active,
created_at as "createdAt", updated_at as "updatedAt"`,
[
definition.id,
definition.version,
definition.ontologyRevision,
definition.deliveryMode,
JSON.stringify(definition.semanticTypes),
JSON.stringify(definition.fields),
JSON.stringify(definition.history),
],
);
if (!result.rowCount) throw deliveryError("data_product_version_is_immutable", 409);
return result.rows[0];
}
export async function persistDataProductPublish(pool, batch, definition, {
maxPatchOperations = 500,
maxPatchBytes = 256 * 1024,
} = {}) {
assertPublishMatchesDefinition(batch, definition);
const requestFingerprint = publishFingerprint(batch);
const client = await pool.connect();
try {
await client.query("begin");
const batchId = randomUUID();
const insertedBatch = await client.query(
`insert into external_data_plane_batches (
id, tenant_id, connection_id, provider_id, data_product_id, contract_version,
ontology_revision, run_id, sequence, idempotency_key, received_at, fact_count,
request_fingerprint
) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
on conflict (tenant_id, connection_id, provider_id, data_product_id, idempotency_key)
do nothing
returning id`,
[
batchId, batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
batch.contract.dataProductId, batch.contract.version, batch.contract.ontologyRevision,
batch.batch.runId, batch.batch.sequence, batch.batch.idempotencyKey,
batch.batch.receivedAt, batch.facts.length, requestFingerprint,
],
);
if (!insertedBatch.rowCount) {
const existing = await client.query(
`select id as "batchId", fact_count as "publishedFactCount",
current_updated_count as "currentUpdatedCount",
history_inserted_count as "historyInsertedCount",
patch_operation_count as "patchOperationCount",
delivery_cursor::text as cursor, received_at as "acceptedAt",
request_fingerprint as "requestFingerprint"
from external_data_plane_batches
where tenant_id = $1 and connection_id = $2 and provider_id = $3
and data_product_id = $4 and idempotency_key = $5`,
scopeValues(batch, batch.batch.idempotencyKey),
);
if (!existing.rowCount || existing.rows[0].requestFingerprint !== requestFingerprint) {
throw deliveryError("idempotency_key_reused", 409);
}
await client.query("commit");
return { ...normalizeReceipt(existing.rows[0]), idempotent: true };
}
const records = batch.facts.map((fact) => ({
source_id: fact.sourceId,
semantic_type: fact.semanticType,
observed_at: fact.observedAt,
received_at: batch.batch.receivedAt,
attributes: fact.attributes || {},
geometry: fact.geometry || null,
fingerprint: fingerprint(fact),
}));
const updated = await client.query(
`insert into external_data_plane_current (
tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type,
observed_at, received_at, attributes, geometry, fingerprint, updated_at
)
select $1, $2, $3, $4, item.source_id, item.semantic_type,
item.observed_at, item.received_at, coalesce(item.attributes, '{}'::jsonb),
case when item.geometry is null then null
else ST_SetSRID(ST_MakePoint(
(item.geometry->'coordinates'->>0)::double precision,
(item.geometry->'coordinates'->>1)::double precision
), 4326)::geography end,
item.fingerprint, now()
from jsonb_to_recordset($5::jsonb) as item(
source_id text, semantic_type text, observed_at timestamptz, received_at timestamptz,
attributes jsonb, geometry jsonb, fingerprint text
)
on conflict (tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type)
do update set
observed_at = excluded.observed_at,
received_at = excluded.received_at,
attributes = excluded.attributes,
geometry = excluded.geometry,
fingerprint = excluded.fingerprint,
updated_at = now()
where excluded.observed_at > external_data_plane_current.observed_at
or (excluded.observed_at = external_data_plane_current.observed_at
and excluded.fingerprint <> external_data_plane_current.fingerprint)
returning source_id as "sourceId", semantic_type as "semanticType",
observed_at as "observedAt", received_at as "receivedAt", attributes,
case when geometry is null then null
else jsonb_build_object('type', 'Point', 'coordinates', jsonb_build_array(
ST_X(geometry::geometry), ST_Y(geometry::geometry)
)) end as geometry,
fingerprint`,
[
batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
batch.contract.dataProductId, JSON.stringify(records),
],
);
const historyInsertedCount = await persistHistory(client, batch, definition.historyPolicy, records);
const operations = updated.rows.map((fact) => ({ op: "upsert", fact: publicFact(fact) }));
const chunks = definition.deliveryMode === "snapshot+patch"
? chunkOperations(operations, maxPatchOperations, maxPatchBytes)
: [];
const cursor = chunks.length ? await persistPatchChunks(client, batch, definition, batchId, chunks) : await currentCursor(client, batch);
await client.query(
`update external_data_plane_batches set
current_updated_count = $2,
history_inserted_count = $3,
patch_operation_count = $4,
delivery_cursor = $5
where id = $1`,
[batchId, updated.rowCount, historyInsertedCount, operations.length, cursor],
);
await client.query("commit");
return normalizeReceipt({
batchId,
idempotent: false,
publishedFactCount: batch.facts.length,
currentUpdatedCount: updated.rowCount,
historyInsertedCount,
patchOperationCount: operations.length,
cursor: String(cursor),
acceptedAt: batch.batch.receivedAt,
});
} catch (error) {
await client.query("rollback");
throw error;
} finally {
client.release();
}
}
export async function readDataProductSnapshot(pool, binding, definition, { limit = 5000 } = {}) {
const client = await pool.connect();
try {
await client.query("begin isolation level repeatable read read only");
const cursor = await currentCursor(client, binding, definition.id);
const rows = await client.query(
`select source_id as "sourceId", semantic_type as "semanticType",
observed_at as "observedAt", received_at as "receivedAt", attributes,
case when geometry is null then null
else jsonb_build_object('type', 'Point', 'coordinates', jsonb_build_array(
ST_X(geometry::geometry), ST_Y(geometry::geometry)
)) end as geometry
from external_data_plane_current
where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4
order by source_id asc, semantic_type asc
limit $5`,
[binding.tenantId, binding.connectionId, binding.providerId, definition.id, limit + 1],
);
if (rows.rowCount > limit) throw deliveryError("data_product_snapshot_limit_exceeded", 413);
await client.query("commit");
return {
schemaVersion: DATA_PRODUCT_SNAPSHOT_SCHEMA_VERSION,
dataProduct: { id: definition.id, version: definition.version },
generatedAt: new Date().toISOString(),
cursor: String(cursor),
facts: rows.rows.map(publicFact),
};
} catch (error) {
await client.query("rollback");
throw error;
} finally {
client.release();
}
}
export async function readPatchEvents(pool, binding, definition, after, { limit = 100 } = {}) {
const client = await pool.connect();
try {
await client.query("begin isolation level repeatable read read only");
const current = await currentCursor(client, binding, definition.id);
if (after > current) throw deliveryError("resync_required", 409);
const floor = await patchFloor(client, binding, definition.id);
if (after < floor - 1n) throw deliveryError("resync_required", 409);
const rows = await client.query(
`select cursor::text, previous_cursor::text as "previousCursor",
operations, emitted_at as "emittedAt"
from external_data_plane_patch_outbox
where tenant_id = $1 and connection_id = $2 and provider_id = $3
and data_product_id = $4 and cursor > $5
order by cursor asc limit $6`,
[binding.tenantId, binding.connectionId, binding.providerId, definition.id, after.toString(), limit],
);
await client.query("commit");
return rows.rows.map((row) => ({
schemaVersion: DATA_PRODUCT_PATCH_SCHEMA_VERSION,
dataProduct: { id: definition.id, version: definition.version },
previousCursor: row.previousCursor,
cursor: row.cursor,
emittedAt: new Date(row.emittedAt).toISOString(),
operations: row.operations,
}));
} catch (error) {
await client.query("rollback");
throw error;
} finally {
client.release();
}
}
export async function prunePatchOutbox(db, { retentionMs, limit = 10_000 }) {
const cutoff = new Date(Date.now() - retentionMs);
const result = await db.query(
`with expired as (
select tenant_id, connection_id, provider_id, data_product_id, cursor
from external_data_plane_patch_outbox
where emitted_at < $1
order by emitted_at asc
limit $2
)
delete from external_data_plane_patch_outbox as target
using expired
where target.tenant_id = expired.tenant_id
and target.connection_id = expired.connection_id
and target.provider_id = expired.provider_id
and target.data_product_id = expired.data_product_id
and target.cursor = expired.cursor`,
[cutoff, limit],
);
return result.rowCount;
}
export async function pruneDataProductHistory(db, { limit = 10_000 } = {}) {
const result = await db.query(
`with expired as (
select history.tenant_id, history.connection_id, history.provider_id,
history.data_product_id, history.source_id, history.semantic_type, history.bucket_start
from external_data_plane_history as history
join external_data_plane_products as product on product.id = history.data_product_id
where history.bucket_start < now() - (
greatest(1, coalesce((product.history_policy->>'retentionDays')::integer, 1)) * interval '1 day'
)
order by history.bucket_start asc
limit $1
)
delete from external_data_plane_history as target
using expired
where target.tenant_id = expired.tenant_id
and target.connection_id = expired.connection_id
and target.provider_id = expired.provider_id
and target.data_product_id = expired.data_product_id
and target.source_id = expired.source_id
and target.semantic_type = expired.semantic_type
and target.bucket_start = expired.bucket_start`,
[limit],
);
return result.rowCount;
}
export async function pruneBatchReceipts(db, { retentionMs, limit = 10_000 }) {
const cutoff = new Date(Date.now() - retentionMs);
const result = await db.query(
`with expired as (
select batch.id
from external_data_plane_batches as batch
where batch.created_at < $1
and not exists (select 1 from external_data_plane_patch_outbox as patch where patch.batch_id = batch.id)
and not exists (select 1 from external_data_plane_raw_envelopes as raw where raw.batch_id = batch.id)
order by batch.created_at asc
limit $2
)
delete from external_data_plane_batches as target
using expired
where target.id = expired.id`,
[cutoff, limit],
);
return result.rowCount;
}
async function persistHistory(client, batch, policy, facts) {
if (!facts.length || policy?.mode === "none") return 0;
const intervalMs = policy?.mode === "sampled" ? Number(policy.intervalMs) : 1;
const records = facts.map((fact) => ({
...fact,
bucket_start: new Date(Math.floor(new Date(fact.observed_at).getTime() / intervalMs) * intervalMs).toISOString(),
}));
const result = await client.query(
`insert into external_data_plane_history (
tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type,
bucket_start, observed_at, received_at, attributes, geometry, fingerprint, updated_at
)
select $1, $2, $3, $4, item.source_id, item.semantic_type,
item.bucket_start, item.observed_at, item.received_at, coalesce(item.attributes, '{}'::jsonb),
case when item.geometry is null then null
else ST_SetSRID(ST_MakePoint(
(item.geometry->'coordinates'->>0)::double precision,
(item.geometry->'coordinates'->>1)::double precision
), 4326)::geography end,
item.fingerprint, now()
from jsonb_to_recordset($5::jsonb) as item(
source_id text, semantic_type text, bucket_start timestamptz,
observed_at timestamptz, received_at timestamptz,
attributes jsonb, geometry jsonb, fingerprint text
)
on conflict (tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type, bucket_start)
do update set
observed_at = excluded.observed_at,
received_at = excluded.received_at,
attributes = excluded.attributes,
geometry = excluded.geometry,
fingerprint = excluded.fingerprint,
updated_at = now()
where excluded.observed_at >= external_data_plane_history.observed_at`,
[
batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
batch.contract.dataProductId,
JSON.stringify(records),
],
);
return result.rowCount;
}
async function persistPatchChunks(client, batch, definition, batchId, chunks) {
const endCursorResult = await client.query(
`insert into external_data_plane_delivery_state (
tenant_id, connection_id, provider_id, data_product_id, current_cursor
) values ($1, $2, $3, $4, $5)
on conflict (tenant_id, connection_id, provider_id, data_product_id)
do update set
current_cursor = external_data_plane_delivery_state.current_cursor + excluded.current_cursor,
updated_at = now()
returning current_cursor`,
[
batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
definition.id, chunks.length,
],
);
const end = BigInt(endCursorResult.rows[0].current_cursor);
const start = end - BigInt(chunks.length) + 1n;
for (let index = 0; index < chunks.length; index += 1) {
const cursor = start + BigInt(index);
await client.query(
`insert into external_data_plane_patch_outbox (
tenant_id, connection_id, provider_id, data_product_id,
cursor, previous_cursor, batch_id, operations
) values ($1, $2, $3, $4, $5, $6, $7, $8::jsonb)`,
[
batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
definition.id, cursor.toString(), (cursor - 1n).toString(), batchId,
JSON.stringify(chunks[index]),
],
);
}
return end;
}
async function currentCursor(db, scope, explicitProductId) {
const dataProductId = explicitProductId || scope.contract.dataProductId;
const result = await db.query(
`select current_cursor from external_data_plane_delivery_state
where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4`,
[scope.tenantId || scope.source.tenantId, scope.connectionId || scope.source.connectionId,
scope.providerId || scope.source.providerId, dataProductId],
);
return result.rowCount ? BigInt(result.rows[0].current_cursor) : 0n;
}
async function patchFloor(db, binding, dataProductId) {
const result = await db.query(
`select min(cursor) as floor from external_data_plane_patch_outbox
where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4`,
[binding.tenantId, binding.connectionId, binding.providerId, dataProductId],
);
if (result.rows[0]?.floor !== null && result.rows[0]?.floor !== undefined) return BigInt(result.rows[0].floor);
const current = await currentCursor(db, binding, dataProductId);
return current + 1n;
}
function scopeValues(batch, tail) {
return [
batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
batch.contract.dataProductId, tail,
];
}
function normalizeReceipt(value) {
return {
batchId: value.batchId,
idempotent: value.idempotent === true,
publishedFactCount: Number(value.publishedFactCount || 0),
currentUpdatedCount: Number(value.currentUpdatedCount || 0),
historyInsertedCount: Number(value.historyInsertedCount || 0),
patchOperationCount: Number(value.patchOperationCount || 0),
cursor: String(value.cursor || "0"),
acceptedAt: new Date(value.acceptedAt).toISOString(),
};
}
function publicFact(fact) {
const value = {
sourceId: fact.sourceId,
semanticType: fact.semanticType,
observedAt: new Date(fact.observedAt).toISOString(),
receivedAt: new Date(fact.receivedAt).toISOString(),
attributes: fact.attributes || {},
};
if (fact.geometry) value.geometry = fact.geometry;
return value;
}
function chunkOperations(values, maxOperations, maxBytes) {
const result = [];
let current = [];
let currentBytes = 2;
for (const value of values) {
const serializedBytes = Buffer.byteLength(JSON.stringify(value));
if (serializedBytes + 2 > maxBytes) throw deliveryError("data_product_patch_operation_size_exceeded", 413);
const separatorBytes = current.length ? 1 : 0;
if (current.length && (current.length >= maxOperations || currentBytes + separatorBytes + serializedBytes > maxBytes)) {
result.push(current);
current = [];
currentBytes = 2;
}
current.push(value);
currentBytes += (current.length > 1 ? 1 : 0) + serializedBytes;
}
if (current.length) result.push(current);
return result;
}
function fingerprint(value) {
return createHash("sha256").update(stableJson(value)).digest("hex");
}
function publishFingerprint(batch) {
return fingerprint({
schemaVersion: batch.schemaVersion,
source: batch.source,
contract: batch.contract,
batch: {
runId: batch.batch.runId,
sequence: batch.batch.sequence,
idempotencyKey: batch.batch.idempotencyKey,
},
facts: batch.facts,
});
}
export function assertPublishMatchesDefinition(batch, definition) {
const semanticTypes = new Set(Array.isArray(definition?.semanticTypes) ? definition.semanticTypes : []);
const fields = new Set(Array.isArray(definition?.fields) ? definition.fields : []);
const entityKeys = new Set();
for (const fact of batch?.facts || []) {
if (!semanticTypes.has(fact.semanticType)) throw deliveryError("data_product_semantic_type_forbidden", 422);
const entityKey = `${fact.sourceId}\u0000${fact.semanticType}`;
if (entityKeys.has(entityKey)) throw deliveryError("data_product_duplicate_entity_key", 422);
entityKeys.add(entityKey);
for (const attribute of Object.keys(fact.attributes || {})) {
if (!fields.has(attribute) && !fields.has(`attributes.${attribute}`)) {
throw deliveryError("data_product_field_forbidden", 422);
}
}
if (fact.geometry !== undefined && !geometryDeclared(fields)) {
throw deliveryError("data_product_geometry_forbidden", 422);
}
}
}
function geometryDeclared(fields) {
return fields.has("geometry")
|| fields.has("coordinates")
|| (fields.has("longitude") && fields.has("latitude"));
}
function stableJson(value) {
if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`;
if (value && typeof value === "object") {
return `{${Object.keys(value).sort().map((key) => `${JSON.stringify(key)}:${stableJson(value[key])}`).join(",")}}`;
}
return JSON.stringify(value);
}
function deliveryError(code, status) {
return Object.assign(new Error(code), { status, code });
}
@@ -0,0 +1,118 @@
const IDENTIFIER = /^[a-z][a-z0-9._:-]{2,127}$/;
const SEMVER = /^\d+\.\d+\.\d+(?:[-+][a-z0-9.-]+)?$/i;
const DELIVERY_MODES = new Set(["snapshot", "snapshot+patch", "query"]);
const HISTORY_MODES = new Set(["none", "all", "sampled"]);
const DEFINITION_KEYS = new Set([
"id", "version", "ontologyRevision", "deliveryMode", "semanticTypes", "fields", "history",
]);
const HISTORY_KEYS = new Set(["mode", "intervalMs", "retentionDays", "strategy"]);
export function normalizeDataProductDefinition(value) {
if (!isPlainObject(value) || !hasOnlyKeys(value, DEFINITION_KEYS)) {
throw policyError("data_product_definition_invalid");
}
const id = identifier(value.id);
const version = string(value.version);
const ontologyRevision = identifier(value.ontologyRevision);
const deliveryMode = string(value.deliveryMode);
const semanticTypes = identifierSet(
value.semanticTypes,
"data_product_definition_semantic_types_invalid",
"data_product_definition_semantic_types_duplicate",
);
const fields = identifierSet(
value.fields,
"data_product_definition_fields_invalid",
"data_product_definition_fields_duplicate",
);
if (!id || !SEMVER.test(version) || !ontologyRevision || !DELIVERY_MODES.has(deliveryMode)) {
throw policyError("data_product_definition_identity_invalid");
}
if (!semanticTypes.length || !fields.length) throw policyError("data_product_definition_shape_invalid");
const history = normalizeHistoryPolicy(value.history);
return Object.freeze({ id, version, ontologyRevision, deliveryMode, semanticTypes, fields, history });
}
export function normalizeHistoryPolicy(value = { mode: "none" }) {
if (!isPlainObject(value) || !hasOnlyKeys(value, HISTORY_KEYS)) throw policyError("history_policy_invalid");
const mode = string(value.mode);
if (!HISTORY_MODES.has(mode)) throw policyError("history_policy_mode_invalid");
const retentionDays = integer(value.retentionDays, mode === "none" ? 1 : 90, 1, 3650);
if (mode === "none") {
if (value.intervalMs !== undefined || value.strategy !== undefined) throw policyError("history_policy_none_has_sampling_fields");
return Object.freeze({ mode, retentionDays });
}
if (mode === "all") {
if (value.intervalMs !== undefined || value.strategy !== undefined) throw policyError("history_policy_all_has_sampling_fields");
return Object.freeze({ mode, retentionDays });
}
const intervalMs = integer(value.intervalMs, 60_000, 1_000, 24 * 60 * 60 * 1000);
const strategy = string(value.strategy || "latest-per-entity-per-bucket");
if (strategy !== "latest-per-entity-per-bucket") throw policyError("history_policy_strategy_invalid");
return Object.freeze({ mode, intervalMs, strategy, retentionDays });
}
export function safeDataProductDefinition(row) {
return {
id: row.id,
version: row.version,
ontologyRevision: row.ontologyRevision,
deliveryMode: row.deliveryMode,
semanticTypes: array(row.semanticTypes),
fields: array(row.fields),
history: isPlainObject(row.historyPolicy) ? row.historyPolicy : {},
active: row.active === true,
createdAt: iso(row.createdAt),
updatedAt: iso(row.updatedAt),
};
}
function policyError(code) {
return Object.assign(new Error(code), { status: 400, code });
}
function identifier(value) {
const normalized = string(value);
return IDENTIFIER.test(normalized) ? normalized : "";
}
function identifierSet(value, invalidCode, duplicateCode) {
if (!Array.isArray(value)) throw policyError(invalidCode);
const normalized = value.map((entry) => {
const result = identifier(entry);
if (!result) throw policyError(invalidCode);
return result;
});
if (new Set(normalized).size !== normalized.length) throw policyError(duplicateCode);
return Object.freeze(normalized.sort());
}
function integer(value, fallback, min, max) {
const number = value === undefined ? fallback : Number(value);
if (!Number.isInteger(number) || number < min || number > max) throw policyError("history_policy_number_invalid");
return number;
}
function string(value) {
return typeof value === "string" ? value.trim() : "";
}
function array(value) {
return Array.isArray(value) ? value : [];
}
function iso(value) {
return value ? new Date(value).toISOString() : undefined;
}
function isPlainObject(value) {
return Boolean(value) && typeof value === "object" && !Array.isArray(value);
}
function hasOnlyKeys(value, allowed) {
return Object.keys(value).every((key) => allowed.has(key));
}
@@ -0,0 +1,43 @@
import { lstat, readFile, readdir } from "node:fs/promises";
import { basename, join, resolve } from "node:path";
import { fileURLToPath } from "node:url";
import { normalizeDataProductDefinition } from "./data-product-policy.mjs";
import { persistDataProductDefinition } from "./data-product-delivery.mjs";
const bundledDefinitionsDir = fileURLToPath(new URL("../definitions/", import.meta.url));
export async function loadDataProductDefinitions(directory = bundledDefinitionsDir) {
const root = resolve(directory);
const entries = (await readdir(root, { withFileTypes: true }))
.filter((entry) => entry.name.endsWith(".json"))
.sort((left, right) => left.name.localeCompare(right.name));
const definitions = [];
for (const entry of entries) {
if (!entry.isFile() || entry.isSymbolicLink()) throw new Error("data_product_definition_file_invalid");
const path = join(root, entry.name);
const metadata = await lstat(path);
if (!metadata.isFile() || metadata.isSymbolicLink() || metadata.size < 2 || metadata.size > 256 * 1024) {
throw new Error("data_product_definition_file_invalid");
}
let value;
try {
value = JSON.parse(await readFile(path, "utf8"));
} catch {
throw new Error("data_product_definition_json_invalid");
}
const definition = normalizeDataProductDefinition(value);
if (`${definition.id}.json` !== basename(path)) throw new Error("data_product_definition_filename_mismatch");
definitions.push(definition);
}
if (new Set(definitions.map((definition) => definition.id)).size !== definitions.length) {
throw new Error("data_product_definition_duplicate");
}
return Object.freeze(definitions);
}
export async function reconcileDataProductDefinitions(db, directory = bundledDefinitionsDir) {
const definitions = await loadDataProductDefinitions(directory);
for (const definition of definitions) await persistDataProductDefinition(db, definition);
return definitions;
}
@@ -0,0 +1,37 @@
export function assertBatchTimeBounds(batch, { now = new Date(), maxFutureSkewSeconds = 300 } = {}) {
const acceptedAt = validDate(now, "intake_policy_clock_invalid");
const maxFutureAt = new Date(acceptedAt.getTime() + Number(maxFutureSkewSeconds) * 1000);
if (!Number.isInteger(maxFutureSkewSeconds) || maxFutureSkewSeconds < 0) {
throw intakePolicyError("intake_policy_future_skew_invalid");
}
const receivedAt = validDate(batch?.batch?.receivedAt, "batch_received_at_invalid");
if (receivedAt > maxFutureAt) throw intakePolicyError("batch_received_at_too_far_in_future");
for (const fact of batch?.facts || []) {
const observedAt = validDate(fact?.observedAt, "fact_observed_at_invalid");
if (observedAt > maxFutureAt) throw intakePolicyError("fact_observed_at_too_far_in_future");
}
}
/**
* Retention is based on the server's acceptance time, never on provider or L2
* timestamps supplied in a batch.
*/
export function rawRetentionExpiry({ now = new Date(), rawRetentionDays }) {
const acceptedAt = validDate(now, "intake_policy_clock_invalid");
if (!Number.isInteger(rawRetentionDays) || rawRetentionDays < 1) {
throw intakePolicyError("raw_retention_days_invalid");
}
return new Date(acceptedAt.getTime() + rawRetentionDays * 24 * 60 * 60 * 1000);
}
export function intakePolicyError(code) {
return Object.assign(new Error(code), { status: 422, code });
}
function validDate(value, code) {
const parsed = value instanceof Date ? new Date(value.getTime()) : new Date(String(value ?? ""));
if (Number.isNaN(parsed.getTime())) throw intakePolicyError(code);
return parsed;
}
@@ -0,0 +1,90 @@
import { createHash, randomBytes } from "node:crypto";
const IDENTIFIER = /^[a-z][a-z0-9._:-]{2,127}$/;
const TOKEN_PREFIX = "ndc_edprb_";
const SECRET_LIKE_KEY = /(token|secret|password|authorization|access[_-]?token|refresh[_-]?token|api[_-]?key)/i;
const REQUEST_KEYS = new Set(["source", "allowedDataProductIds", "expiresAt"]);
const SOURCE_KEYS = new Set(["tenantId", "connectionId", "providerId"]);
export function createReaderToken() {
return `${TOKEN_PREFIX}${randomBytes(32).toString("base64url")}`;
}
export function hashReaderToken(token) {
return createHash("sha256").update(String(token), "utf8").digest("hex");
}
export function normalizeReaderBindingRequest(value, { now = new Date(), maxTtlDays = 90 } = {}) {
if (!isPlainObject(value) || !isPlainObject(value.source)) throw readerError("reader_binding_request_invalid");
if (containsSecretLikeKey(value)) throw readerError("reader_binding_request_secret_material_forbidden");
if (!hasOnlyKeys(value, REQUEST_KEYS) || !hasOnlyKeys(value.source, SOURCE_KEYS)) {
throw readerError("reader_binding_request_fields_invalid");
}
const tenantId = identifier(value.source.tenantId);
const connectionId = identifier(value.source.connectionId);
const providerId = identifier(value.source.providerId);
const allowedDataProductIds = uniqueIdentifiers(value.allowedDataProductIds);
if (!tenantId || !connectionId || !providerId || !allowedDataProductIds.length) {
throw readerError("reader_binding_scope_invalid");
}
const expiresAt = new Date(String(value.expiresAt || ""));
const maxExpiresAt = new Date(now.getTime() + maxTtlDays * 24 * 60 * 60 * 1000);
if (Number.isNaN(expiresAt.getTime()) || expiresAt <= now || expiresAt > maxExpiresAt) {
throw readerError("reader_binding_expiry_invalid");
}
return Object.freeze({ tenantId, connectionId, providerId, allowedDataProductIds, expiresAt: expiresAt.toISOString() });
}
export function assertReaderProduct(binding, dataProductId, now = new Date()) {
if (!isPlainObject(binding) || binding.active !== true || new Date(binding.expiresAt) <= now) {
throw readerError("reader_binding_inactive", 401);
}
const normalized = identifier(dataProductId);
if (!normalized || !uniqueIdentifiers(binding.allowedDataProductIds).includes(normalized)) {
throw readerError("reader_binding_data_product_forbidden", 403);
}
return normalized;
}
export function safeReaderBinding(binding) {
return {
id: binding.id,
tenantId: binding.tenantId,
connectionId: binding.connectionId,
providerId: binding.providerId,
allowedDataProductIds: uniqueIdentifiers(binding.allowedDataProductIds),
active: binding.active === true,
expiresAt: new Date(binding.expiresAt).toISOString(),
createdAt: binding.createdAt ? new Date(binding.createdAt).toISOString() : undefined,
rotatedAt: binding.rotatedAt ? new Date(binding.rotatedAt).toISOString() : undefined,
revokedAt: binding.revokedAt ? new Date(binding.revokedAt).toISOString() : undefined,
};
}
function readerError(code, status = 400) {
return Object.assign(new Error(code), { status, code });
}
function identifier(value) {
const normalized = typeof value === "string" ? value.trim() : "";
return IDENTIFIER.test(normalized) ? normalized : "";
}
function uniqueIdentifiers(value) {
if (!Array.isArray(value)) return [];
return [...new Set(value.map(identifier).filter(Boolean))];
}
function containsSecretLikeKey(value) {
if (Array.isArray(value)) return value.some(containsSecretLikeKey);
if (!isPlainObject(value)) return false;
return Object.entries(value).some(([key, child]) => SECRET_LIKE_KEY.test(key) || containsSecretLikeKey(child));
}
function isPlainObject(value) {
return Boolean(value) && typeof value === "object" && !Array.isArray(value);
}
function hasOnlyKeys(value, allowed) {
return Object.keys(value).every((key) => allowed.has(key));
}
+206
View File
@@ -0,0 +1,206 @@
export async function migrate(pool) {
await pool.query("create extension if not exists timescaledb");
await pool.query("create extension if not exists postgis");
await pool.query(`
create table if not exists external_data_plane_products (
id text primary key,
version text not null,
ontology_revision text not null,
delivery_mode text not null,
semantic_types jsonb not null,
fields jsonb not null,
history_policy jsonb not null,
active boolean not null default true,
created_at timestamptz not null default now(),
updated_at timestamptz not null default now(),
check (jsonb_typeof(semantic_types) = 'array' and jsonb_array_length(semantic_types) > 0),
check (jsonb_typeof(fields) = 'array' and jsonb_array_length(fields) > 0)
)
`);
await pool.query(`
create table if not exists external_data_plane_batches (
id uuid primary key,
tenant_id text not null,
connection_id text not null,
provider_id text not null,
data_product_id text not null,
contract_version text not null,
ontology_revision text not null,
run_id text not null,
sequence integer not null,
idempotency_key text not null,
received_at timestamptz not null,
fact_count integer not null default 0,
inserted_fact_count integer not null default 0,
created_at timestamptz not null default now(),
unique (tenant_id, connection_id, provider_id, data_product_id, idempotency_key)
)
`);
await pool.query("create index if not exists external_data_plane_batches_scope_idx on external_data_plane_batches (tenant_id, connection_id, data_product_id, received_at desc)");
await pool.query("alter table external_data_plane_batches add column if not exists current_updated_count integer not null default 0");
await pool.query("alter table external_data_plane_batches add column if not exists history_inserted_count integer not null default 0");
await pool.query("alter table external_data_plane_batches add column if not exists patch_operation_count integer not null default 0");
await pool.query("alter table external_data_plane_batches add column if not exists delivery_cursor bigint");
await pool.query("alter table external_data_plane_batches add column if not exists request_fingerprint text");
await pool.query(`
create table if not exists external_data_plane_raw_envelopes (
id uuid primary key,
batch_id uuid not null unique references external_data_plane_batches(id) on delete cascade,
payload_hash text not null,
content_type text not null,
payload jsonb,
payload_ref text,
payload_bytes integer,
received_at timestamptz not null,
expires_at timestamptz not null,
check (payload is not null or payload_ref is not null)
)
`);
await pool.query("create index if not exists external_data_plane_raw_expiry_idx on external_data_plane_raw_envelopes (expires_at)");
await pool.query(`
create table if not exists external_data_plane_writer_bindings (
id uuid primary key,
token_hash text not null unique,
tenant_id text not null,
connection_id text not null,
provider_id text not null,
allowed_data_product_ids jsonb not null,
expires_at timestamptz not null,
active boolean not null default true,
created_at timestamptz not null default now(),
rotated_at timestamptz,
revoked_at timestamptz,
check (
case when jsonb_typeof(allowed_data_product_ids) = 'array'
then jsonb_array_length(allowed_data_product_ids) > 0
else false
end
)
)
`);
await pool.query("create index if not exists external_data_plane_writer_bindings_active_idx on external_data_plane_writer_bindings (active, expires_at)");
await pool.query(`
create table if not exists external_data_plane_reader_bindings (
id uuid primary key,
token_hash text not null unique,
tenant_id text not null,
connection_id text not null,
provider_id text not null,
allowed_data_product_ids jsonb not null,
expires_at timestamptz not null,
active boolean not null default true,
created_at timestamptz not null default now(),
rotated_at timestamptz,
revoked_at timestamptz,
check (
case when jsonb_typeof(allowed_data_product_ids) = 'array'
then jsonb_array_length(allowed_data_product_ids) > 0
else false
end
)
)
`);
await pool.query("create index if not exists external_data_plane_reader_bindings_active_idx on external_data_plane_reader_bindings (active, expires_at)");
await pool.query(`
create table if not exists external_data_plane_facts (
id uuid not null,
batch_id uuid not null references external_data_plane_batches(id) on delete cascade,
tenant_id text not null,
connection_id text not null,
provider_id text not null,
data_product_id text not null,
source_id text not null,
semantic_type text not null,
observed_at timestamptz not null,
received_at timestamptz not null,
attributes jsonb not null default '{}'::jsonb,
geometry geography(Point, 4326),
fingerprint text not null,
primary key (id, observed_at),
unique (tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type, observed_at, fingerprint)
)
`);
await pool.query("select create_hypertable('external_data_plane_facts', 'observed_at', if_not_exists => true, migrate_data => true)");
await pool.query("create index if not exists external_data_plane_facts_source_time_idx on external_data_plane_facts (tenant_id, connection_id, data_product_id, source_id, observed_at desc)");
await pool.query("create index if not exists external_data_plane_facts_geometry_idx on external_data_plane_facts using gist (geometry)");
await pool.query(`
create table if not exists external_data_plane_current (
tenant_id text not null,
connection_id text not null,
provider_id text not null,
data_product_id text not null,
source_id text not null,
semantic_type text not null,
observed_at timestamptz not null,
received_at timestamptz not null,
attributes jsonb not null default '{}'::jsonb,
geometry geography(Point, 4326),
fingerprint text not null,
updated_at timestamptz not null default now(),
primary key (tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type)
)
`);
await pool.query("create index if not exists external_data_plane_current_scope_idx on external_data_plane_current (tenant_id, connection_id, data_product_id, observed_at desc)");
await pool.query("create index if not exists external_data_plane_current_geometry_idx on external_data_plane_current using gist (geometry)");
await pool.query(`
create table if not exists external_data_plane_delivery_state (
tenant_id text not null,
connection_id text not null,
provider_id text not null,
data_product_id text not null,
current_cursor bigint not null default 0,
updated_at timestamptz not null default now(),
primary key (tenant_id, connection_id, provider_id, data_product_id)
)
`);
await pool.query(`
create table if not exists external_data_plane_patch_outbox (
tenant_id text not null,
connection_id text not null,
provider_id text not null,
data_product_id text not null,
cursor bigint not null,
previous_cursor bigint not null,
batch_id uuid not null references external_data_plane_batches(id) on delete cascade,
operations jsonb not null,
emitted_at timestamptz not null default now(),
primary key (tenant_id, connection_id, provider_id, data_product_id, cursor),
check (jsonb_typeof(operations) = 'array' and jsonb_array_length(operations) > 0)
)
`);
await pool.query("create index if not exists external_data_plane_patch_outbox_retention_idx on external_data_plane_patch_outbox (emitted_at)");
await pool.query(`
create table if not exists external_data_plane_history (
tenant_id text not null,
connection_id text not null,
provider_id text not null,
data_product_id text not null,
source_id text not null,
semantic_type text not null,
bucket_start timestamptz not null,
observed_at timestamptz not null,
received_at timestamptz not null,
attributes jsonb not null default '{}'::jsonb,
geometry geography(Point, 4326),
fingerprint text not null,
updated_at timestamptz not null default now(),
primary key (
tenant_id, connection_id, provider_id, data_product_id,
source_id, semantic_type, bucket_start
)
)
`);
await pool.query("select create_hypertable('external_data_plane_history', 'bucket_start', if_not_exists => true, migrate_data => true)");
await pool.query("create index if not exists external_data_plane_history_source_time_idx on external_data_plane_history (tenant_id, connection_id, data_product_id, source_id, bucket_start desc)");
await pool.query("create index if not exists external_data_plane_history_geometry_idx on external_data_plane_history using gist (geometry)");
}
+822
View File
@@ -0,0 +1,822 @@
import express from "express";
import { createHash, randomUUID, timingSafeEqual } from "node:crypto";
import { createServer } from "node:http";
import { Pool } from "pg";
import { validateDataProductPublish, validateIntakeBatch } from "@nodedc/external-provider-contract";
import { readConfig } from "./config.mjs";
import {
loadDataProductDefinition,
assertPublishMatchesDefinition,
persistDataProductDefinition,
persistDataProductPublish,
pruneBatchReceipts,
pruneDataProductHistory,
prunePatchOutbox,
readDataProductSnapshot,
readPatchEvents,
} from "./data-product-delivery.mjs";
import { normalizeDataProductDefinition, safeDataProductDefinition } from "./data-product-policy.mjs";
import { reconcileDataProductDefinitions } from "./definitions.mjs";
import { assertBatchTimeBounds, rawRetentionExpiry } from "./intake-policy.mjs";
import {
assertReaderProduct,
createReaderToken,
hashReaderToken,
normalizeReaderBindingRequest,
safeReaderBinding,
} from "./reader-binding.mjs";
import { migrate } from "./schema.mjs";
import {
createWriterToken,
hashWriterToken,
materializeDataProductPublish,
materializeWriterBoundBatch,
normalizeWriterBindingRequest,
safeWriterBinding,
} from "./writer-binding.mjs";
const config = readConfig();
const pool = new Pool({ connectionString: config.databaseUrl, max: config.databasePoolSize });
const app = express();
const httpServer = createServer(app);
let retentionSweepTimer = null;
let lastRetentionSweepAt = null;
const activeReaderStreams = new Map();
const activeStreamResponses = new Set();
let shuttingDown = false;
let shutdownPromise = null;
app.disable("x-powered-by");
app.use(express.json({ limit: config.maxBatchBytes }));
app.get("/healthz", asyncRoute(async (_req, res) => {
await pool.query("select 1");
res.json({
ok: true,
service: "nodedc-external-data-plane",
database: "ready",
internalApiConfigured: Boolean(config.internalAccessToken),
providerLogic: "absent",
commandTransport: "absent",
writerBindings: "supported",
readerBindings: "supported",
dataProductDelivery: "snapshot+durable-patch",
legacyIntake: config.legacyIntakeEnabled ? "migration-only" : "disabled",
writerBindingProvisioning: config.provisionerApiEnabled ? "enabled" : "disabled",
rawRetentionSweep: {
mode: "server-scheduled",
lastSweepAt: lastRetentionSweepAt,
},
});
}));
app.put("/internal/data-plane/v1/data-products/:dataProductId", requireProvisionerApi, asyncRoute(async (req, res) => {
const definition = normalizeDataProductDefinition({ ...req.body, id: req.params.dataProductId });
const saved = await persistDataProductDefinition(pool, definition);
res.json({ ok: true, dataProduct: safeDataProductDefinition(saved) });
}));
app.post("/internal/data-plane/v1/data-products/:dataProductId/publish", requireWriterBinding, asyncRoute(async (req, res) => {
if (hasScopeHeaders(req)) throw httpError(400, "data_product_publish_scope_headers_forbidden");
const validation = validateDataProductPublish(req.body, {
maxFacts: config.maxFactsPerPublish,
maxAttributesBytes: config.maxAttributesBytesPerFact,
});
if (!validation.ok) throw httpError(422, validation.errors[0] || "invalid_data_product_publish");
const dataProductId = requireIdentifier(req.params.dataProductId, "data_product_id_invalid");
const definition = await loadDataProductDefinition(pool, dataProductId);
if (!definition) throw httpError(404, "data_product_not_found");
const batch = materializeDataProductPublish(req.body, req.writerBinding, definition, dataProductId);
const canonicalValidation = validateIntakeBatch(batch);
if (!canonicalValidation.ok) throw httpError(422, "materialized_publish_invalid");
assertBatchTimeBounds(batch, { maxFutureSkewSeconds: config.maxFutureSkewSeconds });
const receipt = await persistDataProductPublish(pool, batch, definition, {
maxPatchOperations: config.maxPatchOperations,
maxPatchBytes: config.maxPatchBytes,
});
res.status(receipt.idempotent ? 200 : 201).json({ ok: true, ...receipt });
}));
app.get("/internal/data-plane/v1/writer/data-products", requireWriterBinding, asyncRoute(async (req, res) => {
const dataProducts = await listGrantedProducts(req.writerBinding);
res.json({ ok: true, dataProducts });
}));
app.get("/internal/data-plane/v1/reader/data-products", requireReaderBinding, asyncRoute(async (req, res) => {
const dataProducts = await listGrantedProducts(req.readerBinding);
res.json({ ok: true, dataProducts });
}));
app.post("/internal/data-plane/v1/intake", requireLegacyIntake, requireInternalApi, asyncRoute(async (req, res) => {
const validation = validateIntakeBatch(req.body);
if (!validation.ok) throw httpError(422, "invalid_intake_batch");
assertBatchTimeBounds(req.body, { maxFutureSkewSeconds: config.maxFutureSkewSeconds });
await assertLegacyBatchProduct(req.body);
const scope = requireScope(req);
if (scope.tenantId !== req.body.source.tenantId || scope.connectionId !== req.body.source.connectionId) {
throw httpError(403, "scope_mismatch");
}
const result = await persistBatch(req.body);
res.status(result.idempotent ? 200 : 201).json({ ok: true, ...result });
}));
app.post("/internal/data-plane/v1/intake/writer-bound", requireLegacyIntake, requireWriterBinding, asyncRoute(async (req, res) => {
const batch = materializeWriterBoundBatch(req.body, req.writerBinding, {
hasScopeHeaders: hasScopeHeaders(req),
});
const validation = validateIntakeBatch(batch);
if (!validation.ok) throw httpError(422, "invalid_intake_batch");
assertBatchTimeBounds(batch, { maxFutureSkewSeconds: config.maxFutureSkewSeconds });
await assertLegacyBatchProduct(batch);
const result = await persistBatch(batch);
res.status(result.idempotent ? 200 : 201).json({ ok: true, ...result });
}));
app.post("/internal/data-plane/v1/writer-bindings", requireProvisionerApi, asyncRoute(async (req, res) => {
const policy = normalizeWriterBindingRequest(req.body, {
maxTtlDays: config.writerBindingMaxTtlDays,
});
await assertRegisteredProductIds(policy.allowedDataProductIds);
const token = createWriterToken();
const bindingId = randomUUID();
const result = await pool.query(
`insert into external_data_plane_writer_bindings (
id, token_hash, tenant_id, connection_id, provider_id,
allowed_data_product_ids, expires_at
) values ($1, $2, $3, $4, $5, $6::jsonb, $7)
returning id, tenant_id as "tenantId", connection_id as "connectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
expires_at as "expiresAt", active, created_at as "createdAt",
rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
[
bindingId,
hashWriterToken(token),
policy.tenantId,
policy.connectionId,
policy.providerId,
JSON.stringify(policy.allowedDataProductIds),
policy.expiresAt,
],
);
// The token is intentionally returned exactly once to a trusted provisioner.
// It must be placed directly into an opaque Engine credential reference and
// must never be logged, stored in L2 graph data or shown to a consumer.
sendOneTimeCapability(res, 201, { ok: true, writerBinding: safeWriterBinding(result.rows[0]), token });
}));
app.post("/internal/data-plane/v1/writer-bindings/:bindingId/rotate", requireProvisionerApi, asyncRoute(async (req, res) => {
const bindingId = requireUuid(req.params.bindingId, "writer_binding_id_invalid");
const token = createWriterToken();
const result = await pool.query(
`update external_data_plane_writer_bindings
set token_hash = $2, rotated_at = now()
where id = $1 and active = true and expires_at > now()
returning id, tenant_id as "tenantId", connection_id as "connectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
expires_at as "expiresAt", active, created_at as "createdAt",
rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
[bindingId, hashWriterToken(token)],
);
if (!result.rowCount) throw httpError(404, "writer_binding_not_found_or_inactive");
sendOneTimeCapability(res, 200, { ok: true, writerBinding: safeWriterBinding(result.rows[0]), token });
}));
app.post("/internal/data-plane/v1/writer-bindings/:bindingId/revoke", requireProvisionerApi, asyncRoute(async (req, res) => {
const bindingId = requireUuid(req.params.bindingId, "writer_binding_id_invalid");
const result = await pool.query(
`update external_data_plane_writer_bindings
set active = false, revoked_at = now()
where id = $1 and active = true
returning id, tenant_id as "tenantId", connection_id as "connectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
expires_at as "expiresAt", active, created_at as "createdAt",
rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
[bindingId],
);
if (!result.rowCount) throw httpError(404, "writer_binding_not_found_or_inactive");
res.json({ ok: true, writerBinding: safeWriterBinding(result.rows[0]) });
}));
app.post("/internal/data-plane/v1/reader-bindings", requireProvisionerApi, asyncRoute(async (req, res) => {
const policy = normalizeReaderBindingRequest(req.body, {
maxTtlDays: config.writerBindingMaxTtlDays,
});
await assertRegisteredProductIds(policy.allowedDataProductIds);
const token = createReaderToken();
const bindingId = randomUUID();
const result = await pool.query(
`insert into external_data_plane_reader_bindings (
id, token_hash, tenant_id, connection_id, provider_id,
allowed_data_product_ids, expires_at
) values ($1, $2, $3, $4, $5, $6::jsonb, $7)
returning id, tenant_id as "tenantId", connection_id as "connectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
expires_at as "expiresAt", active, created_at as "createdAt",
rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
[
bindingId,
hashReaderToken(token),
policy.tenantId,
policy.connectionId,
policy.providerId,
JSON.stringify(policy.allowedDataProductIds),
policy.expiresAt,
],
);
sendOneTimeCapability(res, 201, { ok: true, readerBinding: safeReaderBinding(result.rows[0]), token });
}));
app.post("/internal/data-plane/v1/reader-bindings/:bindingId/rotate", requireProvisionerApi, asyncRoute(async (req, res) => {
const bindingId = requireUuid(req.params.bindingId, "reader_binding_id_invalid");
const token = createReaderToken();
const result = await pool.query(
`update external_data_plane_reader_bindings
set token_hash = $2, rotated_at = now()
where id = $1 and active = true and expires_at > now()
returning id, tenant_id as "tenantId", connection_id as "connectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
expires_at as "expiresAt", active, created_at as "createdAt",
rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
[bindingId, hashReaderToken(token)],
);
if (!result.rowCount) throw httpError(404, "reader_binding_not_found_or_inactive");
sendOneTimeCapability(res, 200, { ok: true, readerBinding: safeReaderBinding(result.rows[0]), token });
}));
app.post("/internal/data-plane/v1/reader-bindings/:bindingId/revoke", requireProvisionerApi, asyncRoute(async (req, res) => {
const bindingId = requireUuid(req.params.bindingId, "reader_binding_id_invalid");
const result = await pool.query(
`update external_data_plane_reader_bindings
set active = false, revoked_at = now()
where id = $1 and active = true
returning id, tenant_id as "tenantId", connection_id as "connectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
expires_at as "expiresAt", active, created_at as "createdAt",
rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
[bindingId],
);
if (!result.rowCount) throw httpError(404, "reader_binding_not_found_or_inactive");
res.json({ ok: true, readerBinding: safeReaderBinding(result.rows[0]) });
}));
app.get("/internal/data-plane/v1/data-products/:dataProductId/snapshot", requireReaderBinding, asyncRoute(async (req, res) => {
if (hasScopeHeaders(req)) throw httpError(400, "data_product_read_scope_headers_forbidden");
const dataProductId = assertReaderProduct(req.readerBinding, req.params.dataProductId);
const definition = await loadDataProductDefinition(pool, dataProductId);
if (!definition) throw httpError(404, "data_product_not_found");
const limit = boundedLimit(req.query.limit, 5000, 1, 5000);
const snapshot = await readDataProductSnapshot(pool, req.readerBinding, definition, { limit });
res.json(snapshot);
}));
app.get("/internal/data-plane/v1/data-products/:dataProductId/stream", requireReaderBinding, asyncRoute(async (req, res) => {
if (shuttingDown) throw httpError(503, "service_shutting_down");
if (hasScopeHeaders(req)) throw httpError(400, "data_product_read_scope_headers_forbidden");
const dataProductId = assertReaderProduct(req.readerBinding, req.params.dataProductId);
const definition = await loadDataProductDefinition(pool, dataProductId);
if (!definition) throw httpError(404, "data_product_not_found");
if (definition.deliveryMode !== "snapshot+patch") throw httpError(409, "data_product_stream_not_supported");
const streamCount = activeReaderStreams.get(req.readerBinding.id) || 0;
if (streamCount >= config.maxReaderStreams) throw httpError(429, "reader_stream_limit_exceeded");
activeReaderStreams.set(req.readerBinding.id, streamCount + 1);
activeStreamResponses.add(res);
let poll = null;
let heartbeat = null;
let polling = false;
let heartbeatWriting = false;
let cleanedUp = false;
const cleanup = () => {
if (cleanedUp) return;
cleanedUp = true;
if (poll) clearInterval(poll);
if (heartbeat) clearInterval(heartbeat);
activeStreamResponses.delete(res);
const remaining = Math.max(0, (activeReaderStreams.get(req.readerBinding.id) || 1) - 1);
if (remaining) activeReaderStreams.set(req.readerBinding.id, remaining);
else activeReaderStreams.delete(req.readerBinding.id);
};
req.once("close", cleanup);
res.once("close", cleanup);
try {
let cursor = parseCursor(req.get("last-event-id") || req.query.after || "0");
const initialEvents = await readPatchEvents(pool, req.readerBinding, definition, cursor);
if (res.destroyed || cleanedUp) return;
res.status(200);
res.set({
"Cache-Control": "no-cache, no-transform",
"Content-Type": "text/event-stream",
Connection: "keep-alive",
"X-Accel-Buffering": "no",
});
res.flushHeaders();
await writeSseFrame(res, ": nodedc-data-product-stream\n\n");
await writeSseFrame(res, `event: nodedc.data-product.ready.v1\ndata: ${JSON.stringify({
schemaVersion: "nodedc.data-product.ready/v1",
dataProductId,
cursor: cursor.toString(),
emittedAt: new Date().toISOString(),
})}\n\n`);
for (const event of initialEvents) {
await writePatchEvent(res, event);
cursor = BigInt(event.cursor);
}
const pump = async () => {
if (polling || res.writableEnded || res.destroyed || shuttingDown) return;
polling = true;
try {
await assertReaderStreamAccess(req.readerBinding, dataProductId);
const events = await readPatchEvents(pool, req.readerBinding, definition, cursor);
for (const event of events) {
await writePatchEvent(res, event);
cursor = BigInt(event.cursor);
}
} catch (error) {
const code = safeErrorCode(error);
if (!res.writableEnded && !res.destroyed) {
await writeSseFrame(res, `event: error\ndata: ${JSON.stringify({ ok: false, error: code })}\n\n`).catch(() => {});
res.end();
}
} finally {
polling = false;
}
};
poll = setInterval(() => { void pump(); }, config.streamPollMs);
poll.unref();
heartbeat = setInterval(() => {
if (polling || heartbeatWriting || res.writableEnded || res.destroyed) return;
heartbeatWriting = true;
void writeSseFrame(res, `: heartbeat ${Date.now()}\n\n`)
.catch(() => { if (!res.writableEnded) res.end(); })
.finally(() => { heartbeatWriting = false; });
}, config.streamHeartbeatMs);
heartbeat.unref();
} catch (error) {
cleanup();
if (res.headersSent) {
if (!res.writableEnded) res.end();
return;
}
throw error;
}
}));
app.get("/internal/data-plane/v1/data-products/:dataProductId/current", requireLegacyIntake, requireInternalApi, asyncRoute(async (req, res) => {
const scope = requireScope(req);
const dataProductId = String(req.params.dataProductId || "");
if (!isIdentifier(dataProductId)) throw httpError(400, "data_product_id_invalid");
const limit = boundedLimit(req.query.limit, 200, 1, 1000);
const rows = await pool.query(
`select
provider_id as "providerId", data_product_id as "dataProductId",
source_id as "sourceId", semantic_type as "semanticType",
observed_at as "observedAt", received_at as "receivedAt", attributes,
case when geometry is null then null
else jsonb_build_object('type', 'Point', 'coordinates', jsonb_build_array(
ST_X(geometry::geometry), ST_Y(geometry::geometry)
)) end as geometry
from external_data_plane_current
where tenant_id = $1 and connection_id = $2 and data_product_id = $3
order by observed_at desc, source_id asc
limit $4`,
[scope.tenantId, scope.connectionId, dataProductId, limit],
);
res.json({
ok: true,
dataProductId,
tenantId: scope.tenantId,
connectionId: scope.connectionId,
facts: rows.rows,
});
}));
app.get("/internal/data-plane/v1/status", requireLegacyIntake, requireInternalApi, asyncRoute(async (req, res) => {
const scope = requireScope(req);
const result = await pool.query(
`select count(*)::integer as "currentFactCount", max(received_at) as "lastReceivedAt"
from external_data_plane_current where tenant_id = $1 and connection_id = $2`,
[scope.tenantId, scope.connectionId],
);
res.json({ ok: true, ...scope, ...result.rows[0], providerLogic: "absent", commandTransport: "absent" });
}));
app.use((error, _req, res, _next) => {
const status = Number(error?.status || 500);
const publicStatus = status >= 400 && status < 600 ? status : 500;
console.error(JSON.stringify({ event: "external_data_plane_error", error: safeErrorCode(error), status: publicStatus }));
res.status(publicStatus).json({ ok: false, error: publicStatus >= 500 ? "internal_error" : safeErrorCode(error) });
});
await migrate(pool);
const bundledDataProducts = await reconcileDataProductDefinitions(pool);
retentionSweepTimer = setInterval(() => {
void sweepRetention().catch((error) => {
console.error(JSON.stringify({ event: "external_data_plane_retention_sweep_failed", error: safeErrorCode(error) }));
});
}, config.retentionSweepMs);
retentionSweepTimer.unref();
httpServer.listen(config.port, "0.0.0.0", () => {
console.log(`NODE.DC External Data Plane listening on http://0.0.0.0:${config.port}`);
console.log(JSON.stringify({ event: "external_data_plane_definitions_ready", count: bundledDataProducts.length }));
setImmediate(() => {
void sweepRetention().catch((error) => {
console.error(JSON.stringify({ event: "external_data_plane_retention_sweep_failed", error: safeErrorCode(error) }));
});
});
});
process.on("SIGTERM", shutdown);
process.on("SIGINT", shutdown);
async function persistBatch(batch) {
const client = await pool.connect();
try {
await client.query("begin");
const existing = await client.query(
`select id, fact_count as "factCount", inserted_fact_count as "insertedFactCount"
from external_data_plane_batches
where tenant_id = $1 and connection_id = $2 and provider_id = $3
and data_product_id = $4 and idempotency_key = $5
for update`,
[
batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
batch.contract.dataProductId, batch.batch.idempotencyKey,
],
);
if (existing.rowCount) {
await client.query("commit");
return { batchId: existing.rows[0].id, idempotent: true, ...existing.rows[0] };
}
const batchId = randomUUID();
await client.query(
`insert into external_data_plane_batches (
id, tenant_id, connection_id, provider_id, data_product_id, contract_version,
ontology_revision, run_id, sequence, idempotency_key, received_at, fact_count
) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`,
[
batchId, batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
batch.contract.dataProductId, batch.contract.version, batch.contract.ontologyRevision,
batch.batch.runId, batch.batch.sequence, batch.batch.idempotencyKey,
batch.batch.receivedAt, batch.facts.length,
],
);
if (batch.raw) await persistRawEnvelope(client, batchId, batch);
const records = batch.facts.map((fact) => ({
id: randomUUID(),
source_id: fact.sourceId,
semantic_type: fact.semanticType,
observed_at: fact.observedAt,
attributes: fact.attributes || {},
geometry: fact.geometry || null,
fingerprint: hash(fact),
}));
const inserted = await client.query(
`insert into external_data_plane_facts (
id, batch_id, tenant_id, connection_id, provider_id, data_product_id,
source_id, semantic_type, observed_at, received_at, attributes, geometry, fingerprint
)
select item.id::uuid, $1, $2, $3, $4, $5,
item.source_id, item.semantic_type, item.observed_at, $6,
coalesce(item.attributes, '{}'::jsonb),
case when item.geometry is null then null
else ST_SetSRID(ST_MakePoint(
(item.geometry->'coordinates'->>0)::double precision,
(item.geometry->'coordinates'->>1)::double precision
), 4326)::geography end,
item.fingerprint
from jsonb_to_recordset($7::jsonb) as item(
id text, source_id text, semantic_type text, observed_at timestamptz,
attributes jsonb, geometry jsonb, fingerprint text
)
on conflict do nothing
returning id`,
[
batchId, batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
batch.contract.dataProductId, batch.batch.receivedAt, JSON.stringify(records),
],
);
await client.query(
`insert into external_data_plane_current (
tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type,
observed_at, received_at, attributes, geometry, fingerprint, updated_at
)
select tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type,
observed_at, received_at, attributes, geometry, fingerprint, now()
from external_data_plane_facts where batch_id = $1
on conflict (tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type)
do update set
observed_at = excluded.observed_at,
received_at = excluded.received_at,
attributes = excluded.attributes,
geometry = excluded.geometry,
fingerprint = excluded.fingerprint,
updated_at = now()
where excluded.observed_at >= external_data_plane_current.observed_at`,
[batchId],
);
await client.query(
"update external_data_plane_batches set inserted_fact_count = $2 where id = $1",
[batchId, inserted.rowCount],
);
await client.query("commit");
return { batchId, idempotent: false, factCount: batch.facts.length, insertedFactCount: inserted.rowCount };
} catch (error) {
await client.query("rollback");
throw error;
} finally {
client.release();
}
}
async function persistRawEnvelope(client, batchId, batch) {
const serialized = batch.raw.payload === undefined ? null : JSON.stringify(batch.raw.payload);
const expiresAt = rawRetentionExpiry({ rawRetentionDays: config.rawRetentionDays });
await client.query(
`insert into external_data_plane_raw_envelopes (
id, batch_id, payload_hash, content_type, payload, payload_ref, payload_bytes, received_at, expires_at
) values ($1, $2, $3, $4, $5::jsonb, $6, $7, $8, $9)`,
[
randomUUID(), batchId, batch.raw.hash || hash(batch.raw.payload), batch.raw.contentType, serialized, batch.raw.ref || null,
serialized ? Buffer.byteLength(serialized) : null, batch.batch.receivedAt, expiresAt,
],
);
}
function requireInternalApi(req, _res, next) {
if (!config.internalAccessToken) return next(httpError(503, "internal_api_not_configured"));
const value = bearerToken(req);
if (!value || !safeEqual(value, config.internalAccessToken)) return next(httpError(401, "unauthorized"));
return next();
}
function requireLegacyIntake(_req, _res, next) {
if (!config.legacyIntakeEnabled) return next(httpError(410, "legacy_intake_disabled"));
return next();
}
function requireProvisionerApi(req, _res, next) {
if (!config.provisionerApiEnabled) return next(httpError(503, "provisioner_api_disabled"));
if (!config.provisionerAccessToken) return next(httpError(503, "provisioner_api_not_configured"));
const value = bearerToken(req);
if (!value || !safeEqual(value, config.provisionerAccessToken)) return next(httpError(401, "provisioner_unauthorized"));
return next();
}
function requireWriterBinding(req, _res, next) {
return resolveWriterBinding(req)
.then((binding) => {
req.writerBinding = binding;
next();
})
.catch(next);
}
function requireReaderBinding(req, _res, next) {
return resolveReaderBinding(req)
.then((binding) => {
req.readerBinding = binding;
next();
})
.catch(next);
}
async function resolveWriterBinding(req) {
const token = bearerToken(req);
if (!token) throw httpError(401, "writer_binding_unauthorized");
const result = await pool.query(
`select id, tenant_id as "tenantId", connection_id as "connectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
expires_at as "expiresAt", active, created_at as "createdAt",
rotated_at as "rotatedAt", revoked_at as "revokedAt"
from external_data_plane_writer_bindings
where token_hash = $1 and active = true and expires_at > now()`,
[hashWriterToken(token)],
);
if (!result.rowCount) throw httpError(401, "writer_binding_unauthorized");
return result.rows[0];
}
async function resolveReaderBinding(req) {
const token = bearerToken(req);
if (!token) throw httpError(401, "reader_binding_unauthorized");
const result = await pool.query(
`select id, token_hash as "tokenHash", tenant_id as "tenantId", connection_id as "connectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
expires_at as "expiresAt", active, created_at as "createdAt",
rotated_at as "rotatedAt", revoked_at as "revokedAt"
from external_data_plane_reader_bindings
where token_hash = $1 and active = true and expires_at > now()`,
[hashReaderToken(token)],
);
if (!result.rowCount) throw httpError(401, "reader_binding_unauthorized");
return result.rows[0];
}
async function assertReaderStreamAccess(binding, dataProductId) {
const result = await pool.query(
`select binding.id
from external_data_plane_reader_bindings as binding
join external_data_plane_products as product
on product.id = $3 and product.active = true
where binding.id = $1 and binding.token_hash = $2
and binding.active = true and binding.expires_at > now()
and binding.allowed_data_product_ids ? $3`,
[binding.id, binding.tokenHash, dataProductId],
);
if (!result.rowCount) throw httpError(401, "reader_stream_access_revoked");
}
async function assertLegacyBatchProduct(batch) {
const definition = await loadDataProductDefinition(pool, batch.contract.dataProductId);
if (!definition) throw httpError(422, "legacy_data_product_not_registered");
if (definition.version !== batch.contract.version || definition.ontologyRevision !== batch.contract.ontologyRevision) {
throw httpError(422, "legacy_data_product_contract_mismatch");
}
assertPublishMatchesDefinition(batch, definition);
}
async function listGrantedProducts(binding) {
const allowed = Array.isArray(binding.allowedDataProductIds) ? binding.allowedDataProductIds : [];
if (!allowed.length) return [];
const result = await pool.query(
`select id, version, ontology_revision as "ontologyRevision",
delivery_mode as "deliveryMode", semantic_types as "semanticTypes",
fields, history_policy as "historyPolicy", active,
created_at as "createdAt", updated_at as "updatedAt"
from external_data_plane_products
where active = true and id = any($1::text[])
order by id asc`,
[allowed],
);
return result.rows.map(safeDataProductDefinition);
}
async function assertRegisteredProductIds(dataProductIds) {
const result = await pool.query(
"select id from external_data_plane_products where active = true and id = any($1::text[])",
[dataProductIds],
);
const registered = new Set(result.rows.map((row) => row.id));
if (dataProductIds.some((id) => !registered.has(id))) throw httpError(422, "binding_data_product_not_registered");
}
function requireScope(req) {
const tenantId = String(req.get("x-nodedc-tenant-id") || "");
const connectionId = String(req.get("x-nodedc-connection-id") || "");
if (!isIdentifier(tenantId) || !isIdentifier(connectionId)) throw httpError(400, "scope_headers_required");
return { tenantId, connectionId };
}
function hasScopeHeaders(req) {
return Object.hasOwn(req.headers, "x-nodedc-tenant-id") || Object.hasOwn(req.headers, "x-nodedc-connection-id");
}
function bearerToken(req) {
return String(req.get("authorization") || "").replace(/^Bearer\s+/i, "");
}
function requireUuid(value, code) {
const normalized = String(value || "");
if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(normalized)) {
throw httpError(400, code);
}
return normalized;
}
function requireIdentifier(value, code) {
const normalized = String(value || "");
if (!isIdentifier(normalized)) throw httpError(400, code);
return normalized;
}
function parseCursor(value) {
const normalized = String(value ?? "");
if (!/^(?:0|[1-9]\d*)$/.test(normalized)) throw httpError(400, "data_product_cursor_invalid");
const cursor = BigInt(normalized);
if (cursor > 9_223_372_036_854_775_807n) throw httpError(400, "data_product_cursor_invalid");
return cursor;
}
function writePatchEvent(res, event) {
return writeSseFrame(res, `id: ${event.cursor}\nevent: nodedc.data-product.patch.v1\ndata: ${JSON.stringify(event)}\n\n`);
}
function writeSseFrame(res, frame) {
if (res.writableEnded || res.destroyed) return Promise.reject(httpError(499, "reader_stream_closed"));
if (res.write(frame)) return Promise.resolve();
return new Promise((resolve, reject) => {
const cleanup = () => {
res.off("drain", onDrain);
res.off("close", onClose);
res.off("error", onError);
};
const onDrain = () => { cleanup(); resolve(); };
const onClose = () => { cleanup(); reject(httpError(499, "reader_stream_closed")); };
const onError = (error) => { cleanup(); reject(error); };
res.once("drain", onDrain);
res.once("close", onClose);
res.once("error", onError);
});
}
function sendOneTimeCapability(res, status, payload) {
// Create/rotate is the only boundary where a capability exists in
// plaintext. A trusted provisioner must consume the response in memory and
// place it directly into its opaque destination; intermediaries must never
// cache or persist it.
res.set({
"Cache-Control": "no-store, max-age=0",
Pragma: "no-cache",
Expires: "0",
});
return res.status(status).json(payload);
}
function safeEqual(left, right) {
const leftBuffer = Buffer.from(left);
const rightBuffer = Buffer.from(right);
return leftBuffer.length === rightBuffer.length && timingSafeEqual(leftBuffer, rightBuffer);
}
function boundedLimit(value, fallback, min, max) {
const candidate = Number.parseInt(String(value ?? ""), 10);
if (!Number.isInteger(candidate)) return fallback;
return Math.max(min, Math.min(max, candidate));
}
function hash(value) {
return createHash("sha256").update(JSON.stringify(value)).digest("hex");
}
function isIdentifier(value) {
return /^[a-z][a-z0-9._:-]{2,127}$/i.test(value);
}
function asyncRoute(handler) {
return (req, res, next) => Promise.resolve(handler(req, res, next)).catch(next);
}
function httpError(status, code) {
return Object.assign(new Error(code), { status, code });
}
function safeErrorCode(error) {
return String(error?.code || error?.message || "internal_error").replace(/[^a-z0-9_.:-]/gi, "_").slice(0, 120);
}
async function sweepExpiredRawEnvelopes() {
const result = await pool.query(
`with expired as (
select id from external_data_plane_raw_envelopes
where expires_at < now()
order by expires_at asc
limit $1
)
delete from external_data_plane_raw_envelopes as target
using expired where target.id = expired.id`,
[config.retentionDeleteLimit],
);
return result.rowCount;
}
async function sweepRetention() {
const rawEnvelopeCount = await sweepExpiredRawEnvelopes();
const patchEventCount = await prunePatchOutbox(pool, {
retentionMs: config.patchRetentionMs,
limit: config.retentionDeleteLimit,
});
const historyFactCount = await pruneDataProductHistory(pool, { limit: config.retentionDeleteLimit });
const batchReceiptCount = await pruneBatchReceipts(pool, {
retentionMs: config.receiptRetentionMs,
limit: config.retentionDeleteLimit,
});
lastRetentionSweepAt = new Date().toISOString();
return { rawEnvelopeCount, patchEventCount, historyFactCount, batchReceiptCount };
}
async function shutdown() {
if (shutdownPromise) return shutdownPromise;
shutdownPromise = (async () => {
shuttingDown = true;
if (retentionSweepTimer) clearInterval(retentionSweepTimer);
for (const response of activeStreamResponses) {
if (!response.writableEnded) response.end();
}
const closed = new Promise((resolve) => httpServer.close(resolve));
httpServer.closeIdleConnections?.();
const forceClose = setTimeout(() => httpServer.closeAllConnections?.(), 250);
await Promise.race([closed, new Promise((resolve) => setTimeout(resolve, 2_000))]);
clearTimeout(forceClose);
httpServer.closeAllConnections?.();
await pool.end();
})();
return shutdownPromise;
}
@@ -0,0 +1,182 @@
import { createHash, randomBytes } from "node:crypto";
const IDENTIFIER = /^[a-z][a-z0-9._:-]{2,127}$/;
const TOKEN_PREFIX = "ndc_edpwb_";
const SECRET_LIKE_KEY = /(token|secret|password|access[_-]?token|refresh[_-]?token|api[_-]?key)/i;
const BINDING_REQUEST_KEYS = new Set(["source", "allowedDataProductIds", "expiresAt"]);
const BINDING_SOURCE_KEYS = new Set(["tenantId", "connectionId", "providerId"]);
/**
* Produces an opaque, high-entropy capability. The plaintext is returned only
* to the trusted provisioning caller; persistence uses its SHA-256 digest.
*/
export function createWriterToken() {
return `${TOKEN_PREFIX}${randomBytes(32).toString("base64url")}`;
}
export function hashWriterToken(token) {
return createHash("sha256").update(String(token), "utf8").digest("hex");
}
export function normalizeWriterBindingRequest(value, { now = new Date(), maxTtlDays = 90 } = {}) {
if (!isPlainObject(value) || !isPlainObject(value.source)) {
throw writerBindingError("writer_binding_request_invalid");
}
if (containsSecretLikeKey(value)) {
throw writerBindingError("writer_binding_request_secret_material_forbidden");
}
if (value.endpoint !== undefined || value.url !== undefined || value.host !== undefined) {
throw writerBindingError("writer_binding_request_transport_forbidden");
}
if (!hasOnlyKeys(value, BINDING_REQUEST_KEYS) || !hasOnlyKeys(value.source, BINDING_SOURCE_KEYS)) {
throw writerBindingError("writer_binding_request_fields_invalid");
}
const tenantId = normalizeIdentifier(value.source.tenantId);
const connectionId = normalizeIdentifier(value.source.connectionId);
const providerId = normalizeIdentifier(value.source.providerId);
if (!tenantId || !connectionId || !providerId) {
throw writerBindingError("writer_binding_scope_invalid");
}
const allowedDataProductIds = uniqueIdentifiers(value.allowedDataProductIds);
if (!allowedDataProductIds.length) {
throw writerBindingError("writer_binding_data_products_invalid");
}
const expiresAt = new Date(String(value.expiresAt || ""));
const maxExpiresAt = new Date(now.getTime() + maxTtlDays * 24 * 60 * 60 * 1000);
if (Number.isNaN(expiresAt.getTime()) || expiresAt <= now || expiresAt > maxExpiresAt) {
throw writerBindingError("writer_binding_expiry_invalid");
}
return Object.freeze({
tenantId,
connectionId,
providerId,
allowedDataProductIds,
expiresAt: expiresAt.toISOString(),
});
}
/**
* Converts a caller-provided, deliberately unscoped intake envelope into the
* canonical scoped form. Caller scope is rejected, never trusted or merged.
*/
export function materializeWriterBoundBatch(value, binding, { hasScopeHeaders = false, now = new Date() } = {}) {
if (!isPlainObject(value) || !isPlainObject(value.source) || !isPlainObject(binding)) {
throw writerBindingError("writer_bound_intake_invalid");
}
if (hasScopeHeaders || value.source.tenantId !== undefined || value.source.connectionId !== undefined) {
throw writerBindingError("writer_bound_scope_forbidden");
}
if (binding.active !== true || !bindingIsCurrent(binding, now)) {
throw writerBindingError("writer_binding_inactive");
}
const providerId = normalizeIdentifier(value.source.providerId);
const tenantId = normalizeIdentifier(binding.tenantId);
const connectionId = normalizeIdentifier(binding.connectionId);
if (!providerId || !tenantId || !connectionId || providerId !== binding.providerId) {
throw writerBindingError("writer_binding_provider_forbidden");
}
const dataProductId = normalizeIdentifier(value.contract?.dataProductId);
const allowedDataProductIds = uniqueIdentifiers(binding.allowedDataProductIds);
if (!dataProductId || !allowedDataProductIds.includes(dataProductId)) {
throw writerBindingError("writer_binding_data_product_forbidden");
}
return {
...value,
source: { providerId, tenantId, connectionId },
};
}
/**
* Materializes the provider-neutral Data Product publish wire form. Unlike the
* legacy writer-bound intake, the caller cannot send provider identity,
* contract metadata, receivedAt or any scope at all.
*/
export function materializeDataProductPublish(value, binding, definition, dataProductId, { now = new Date() } = {}) {
if (!isPlainObject(value) || !isPlainObject(binding) || !isPlainObject(definition)) {
throw writerBindingError("data_product_publish_invalid");
}
if (binding.active !== true || !bindingIsCurrent(binding, now)) {
throw writerBindingError("writer_binding_inactive");
}
const normalizedProductId = normalizeIdentifier(dataProductId);
const allowedDataProductIds = uniqueIdentifiers(binding.allowedDataProductIds);
if (!normalizedProductId || normalizedProductId !== definition.id || !allowedDataProductIds.includes(normalizedProductId)) {
throw writerBindingError("writer_binding_data_product_forbidden");
}
const tenantId = normalizeIdentifier(binding.tenantId);
const connectionId = normalizeIdentifier(binding.connectionId);
const providerId = normalizeIdentifier(binding.providerId);
if (!tenantId || !connectionId || !providerId) throw writerBindingError("writer_binding_scope_invalid");
return {
schemaVersion: "nodedc.external-provider-contract/v1",
source: { tenantId, connectionId, providerId },
contract: {
dataProductId: normalizedProductId,
ontologyRevision: definition.ontologyRevision,
version: definition.version,
},
batch: {
runId: value.batch?.runId,
sequence: value.batch?.sequence,
idempotencyKey: value.batch?.idempotencyKey,
receivedAt: now.toISOString(),
},
facts: value.facts,
};
}
export function safeWriterBinding(binding) {
return {
id: binding.id,
tenantId: binding.tenantId,
connectionId: binding.connectionId,
providerId: binding.providerId,
allowedDataProductIds: uniqueIdentifiers(binding.allowedDataProductIds),
active: binding.active === true,
expiresAt: new Date(binding.expiresAt).toISOString(),
createdAt: binding.createdAt ? new Date(binding.createdAt).toISOString() : undefined,
rotatedAt: binding.rotatedAt ? new Date(binding.rotatedAt).toISOString() : undefined,
revokedAt: binding.revokedAt ? new Date(binding.revokedAt).toISOString() : undefined,
};
}
export function writerBindingError(code) {
return Object.assign(new Error(code), { status: 400, code });
}
function bindingIsCurrent(binding, now) {
const expiresAt = new Date(binding.expiresAt);
return !Number.isNaN(expiresAt.getTime()) && expiresAt > now;
}
function normalizeIdentifier(value) {
const normalized = typeof value === "string" ? value.trim() : "";
return IDENTIFIER.test(normalized) ? normalized : "";
}
function uniqueIdentifiers(value) {
if (!Array.isArray(value)) return [];
return [...new Set(value.map(normalizeIdentifier).filter(Boolean))];
}
function isPlainObject(value) {
return Boolean(value) && typeof value === "object" && !Array.isArray(value);
}
function containsSecretLikeKey(value) {
if (Array.isArray(value)) return value.some(containsSecretLikeKey);
if (!isPlainObject(value)) return false;
return Object.entries(value).some(([key, child]) => SECRET_LIKE_KEY.test(key) || containsSecretLikeKey(child));
}
function hasOnlyKeys(value, allowedKeys) {
return Object.keys(value).every((key) => allowedKeys.has(key));
}
@@ -0,0 +1,280 @@
import assert from "node:assert/strict";
import { spawn } from "node:child_process";
import { mkdtemp, rm, writeFile } from "node:fs/promises";
import net from "node:net";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { validateDataProductPatch, validateDataProductSnapshot } from "@nodedc/external-provider-contract";
const databaseUrl = process.env.EXTERNAL_DATA_PLANE_TEST_DATABASE_URL;
if (!databaseUrl) throw new Error("EXTERNAL_DATA_PLANE_TEST_DATABASE_URL_required");
const port = await freePort();
const baseUrl = `http://127.0.0.1:${port}`;
const directory = await mkdtemp(join(tmpdir(), "nodedc-edp-api-test-"));
const provisionerSecret = "nodedc_edp_integration_provisioner_secret_0123456789abcdef";
const secretPath = join(directory, "provisioner-token");
await writeFile(secretPath, `${provisionerSecret}\n`, { mode: 0o600 });
const child = spawn(process.execPath, ["src/server.mjs"], {
cwd: new URL("..", import.meta.url),
env: {
...process.env,
PORT: String(port),
EXTERNAL_DATA_PLANE_DATABASE_URL: databaseUrl,
EXTERNAL_DATA_PLANE_PROVISIONING_ENABLED: "true",
EXTERNAL_DATA_PLANE_PROVISIONER_TOKEN_FILE: secretPath,
EXTERNAL_DATA_PLANE_RETENTION_SWEEP_MS: "60000",
EXTERNAL_DATA_PLANE_STREAM_POLL_MS: "250",
EXTERNAL_DATA_PLANE_STREAM_HEARTBEAT_MS: "5000",
},
stdio: ["ignore", "pipe", "pipe"],
});
let childOutput = "";
child.stdout.on("data", (chunk) => { childOutput += chunk.toString(); });
child.stderr.on("data", (chunk) => { childOutput += chunk.toString(); });
try {
await waitForHealth();
const productId = "api.test.positions.v1";
const product = await jsonRequest(`/internal/data-plane/v1/data-products/${productId}`, {
method: "PUT",
token: provisionerSecret,
body: {
version: "1.0.0",
ontologyRevision: "ontology.api.test.v1",
deliveryMode: "snapshot+patch",
semanticTypes: ["map.moving_object"],
fields: ["source_id", "observed_at", "geometry", "status"],
history: { mode: "sampled", intervalMs: 60_000, strategy: "latest-per-entity-per-bucket", retentionDays: 30 },
},
});
assert.equal(product.dataProduct.id, productId);
const expiry = new Date(Date.now() + 24 * 60 * 60 * 1000).toISOString();
const scope = { tenantId: "api-tenant", connectionId: "api-connection", providerId: "api-provider" };
const writerIssuance = await rawJsonRequest("/internal/data-plane/v1/writer-bindings", {
method: "POST",
token: provisionerSecret,
body: { source: scope, allowedDataProductIds: [productId], expiresAt: expiry },
});
const readerIssuance = await rawJsonRequest("/internal/data-plane/v1/reader-bindings", {
method: "POST",
token: provisionerSecret,
body: { source: scope, allowedDataProductIds: [productId], expiresAt: expiry },
});
assertOneTimeCapabilityResponse(writerIssuance.response);
assertOneTimeCapabilityResponse(readerIssuance.response);
const writer = writerIssuance.value;
const reader = readerIssuance.value;
assert.equal(typeof writer.token, "string");
assert.equal(typeof reader.token, "string");
const catalog = await jsonRequest("/internal/data-plane/v1/writer/data-products", { token: writer.token });
assert.deepEqual(catalog.dataProducts.map((value) => value.id), [productId]);
const first = await jsonRequest(`/internal/data-plane/v1/data-products/${productId}/publish`, {
method: "POST",
token: writer.token,
body: publish("api-run-01", "2026-07-15T12:00:00.000Z", 37.61),
});
assert.equal(first.currentUpdatedCount, 1);
const duplicate = await jsonRequest(`/internal/data-plane/v1/data-products/${productId}/publish`, {
method: "POST",
token: writer.token,
body: publish("api-run-01", "2026-07-15T12:00:00.000Z", 37.61),
});
assert.equal(duplicate.idempotent, true);
const reusedKey = await fetch(`${baseUrl}/internal/data-plane/v1/data-products/${productId}/publish`, {
method: "POST",
headers: { Authorization: `Bearer ${writer.token}`, "Content-Type": "application/json" },
body: JSON.stringify(publish("api-run-01", "2026-07-15T12:00:00.000Z", 99.9)),
});
assert.equal(reusedKey.status, 409);
assert.equal((await reusedKey.json()).error, "idempotency_key_reused");
const snapshot = await jsonRequest(`/internal/data-plane/v1/data-products/${productId}/snapshot`, { token: reader.token });
assert.equal(validateDataProductSnapshot(snapshot).ok, true);
assert.equal(snapshot.facts.length, 1);
const controller = new AbortController();
const response = await fetch(`${baseUrl}/internal/data-plane/v1/data-products/${productId}/stream?after=${snapshot.cursor}`, {
headers: { Authorization: `Bearer ${reader.token}` },
signal: controller.signal,
});
assert.equal(response.status, 200);
const patchPromise = readFirstPatch(response, controller);
await jsonRequest(`/internal/data-plane/v1/data-products/${productId}/publish`, {
method: "POST",
token: writer.token,
body: publish("api-run-02", "2026-07-15T12:00:10.000Z", 37.62),
});
const patch = await patchPromise;
assert.equal(validateDataProductPatch(patch).ok, true);
assert.equal(patch.previousCursor, snapshot.cursor);
const forbidden = await fetch(`${baseUrl}/internal/data-plane/v1/data-products/${productId}/snapshot`, {
headers: {
Authorization: `Bearer ${reader.token}`,
"X-NODEDC-Tenant-Id": scope.tenantId,
"X-NODEDC-Connection-Id": scope.connectionId,
},
});
assert.equal(forbidden.status, 400);
const revocationSnapshot = await jsonRequest(`/internal/data-plane/v1/data-products/${productId}/snapshot`, { token: reader.token });
const revocationStream = await fetch(`${baseUrl}/internal/data-plane/v1/data-products/${productId}/stream?after=${revocationSnapshot.cursor}`, {
headers: { Authorization: `Bearer ${reader.token}` },
});
assert.equal(revocationStream.status, 200);
const revokedEventPromise = readNamedEvent(revocationStream, "error");
const rotatedReaderIssuance = await rawJsonRequest(`/internal/data-plane/v1/reader-bindings/${reader.readerBinding.id}/rotate`, {
method: "POST",
token: provisionerSecret,
});
assertOneTimeCapabilityResponse(rotatedReaderIssuance.response);
const rotatedReader = rotatedReaderIssuance.value;
const revokedEvent = await revokedEventPromise;
assert.equal(revokedEvent.error, "reader_stream_access_revoked");
const rejectedOldReader = await fetch(`${baseUrl}/internal/data-plane/v1/data-products/${productId}/snapshot`, {
headers: { Authorization: `Bearer ${reader.token}` },
});
assert.equal(rejectedOldReader.status, 401);
const rotatedSnapshot = await jsonRequest(`/internal/data-plane/v1/data-products/${productId}/snapshot`, { token: rotatedReader.token });
assert.equal(rotatedSnapshot.dataProduct.id, productId);
const shutdownStream = await fetch(`${baseUrl}/internal/data-plane/v1/data-products/${productId}/stream?after=${rotatedSnapshot.cursor}`, {
headers: { Authorization: `Bearer ${rotatedReader.token}` },
});
assert.equal(shutdownStream.status, 200);
child.kill("SIGTERM");
assert.equal(await waitForChildExit(child, 3_000), true);
console.log("external-data-plane API integration: ok");
} catch (error) {
error.message = `${error.message}\nserver_output=${childOutput.replace(/[A-Za-z0-9_-]{48,}/g, "[redacted]").slice(-4000)}`;
throw error;
} finally {
child.kill("SIGTERM");
await Promise.race([
new Promise((resolve) => child.once("exit", resolve)),
new Promise((resolve) => setTimeout(resolve, 3000)),
]);
await rm(directory, { recursive: true, force: true });
}
function publish(runId, observedAt, longitude) {
return {
schemaVersion: "nodedc.data-product.publish/v1",
batch: { runId, sequence: 0, idempotencyKey: `${runId}.chunk-0` },
facts: [{
sourceId: "unit-01",
semanticType: "map.moving_object",
observedAt,
attributes: { status: "online" },
geometry: { type: "Point", coordinates: [longitude, 55.75] },
}],
};
}
async function jsonRequest(path, { method = "GET", token, body } = {}) {
const { response, value } = await rawJsonRequest(path, { method, token, body });
if (!response.ok) throw new Error(`request_failed:${response.status}:${value.error}`);
return value;
}
async function rawJsonRequest(path, { method = "GET", token, body } = {}) {
const response = await fetch(`${baseUrl}${path}`, {
method,
headers: {
...(token ? { Authorization: `Bearer ${token}` } : {}),
...(body ? { "Content-Type": "application/json" } : {}),
},
body: body ? JSON.stringify(body) : undefined,
});
const value = await response.json();
if (!response.ok) throw new Error(`request_failed:${response.status}:${value.error}`);
return { response, value };
}
function assertOneTimeCapabilityResponse(response) {
assert.equal(response.headers.get("cache-control"), "no-store, max-age=0");
assert.equal(response.headers.get("pragma"), "no-cache");
assert.equal(response.headers.get("expires"), "0");
}
async function waitForHealth() {
const deadline = Date.now() + 15_000;
while (Date.now() < deadline) {
try {
const response = await fetch(`${baseUrl}/healthz`);
if (response.ok) return;
} catch {}
await new Promise((resolve) => setTimeout(resolve, 100));
}
throw new Error("server_health_timeout");
}
async function readFirstPatch(response, controller) {
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
const timeout = setTimeout(() => controller.abort(), 10_000);
try {
while (true) {
const { done, value } = await reader.read();
if (done) throw new Error("stream_closed_before_patch");
buffer += decoder.decode(value, { stream: true });
for (const frame of buffer.split("\n\n")) {
const data = frame.split("\n").find((line) => line.startsWith("data: "));
if (frame.includes("event: nodedc.data-product.patch.v1") && data) {
controller.abort();
return JSON.parse(data.slice(6));
}
}
const boundary = buffer.lastIndexOf("\n\n");
if (boundary >= 0) buffer = buffer.slice(boundary + 2);
}
} finally {
clearTimeout(timeout);
}
}
async function readNamedEvent(response, eventName) {
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
const deadline = Date.now() + 10_000;
while (Date.now() < deadline) {
const next = await Promise.race([
reader.read(),
new Promise((_, reject) => setTimeout(() => reject(new Error("stream_event_timeout")), 10_000)),
]);
if (next.done) throw new Error(`stream_closed_before_${eventName}`);
buffer += decoder.decode(next.value, { stream: true }).replace(/\r\n/g, "\n");
let boundary;
while ((boundary = buffer.indexOf("\n\n")) !== -1) {
const frame = buffer.slice(0, boundary);
buffer = buffer.slice(boundary + 2);
if (!frame.includes(`event: ${eventName}`)) continue;
const data = frame.split("\n").find((line) => line.startsWith("data: "));
if (data) return JSON.parse(data.slice(6));
}
}
throw new Error(`stream_event_timeout:${eventName}`);
}
async function waitForChildExit(process, timeoutMs) {
if (process.exitCode !== null) return true;
return Promise.race([
new Promise((resolve) => process.once("exit", () => resolve(true))),
new Promise((resolve) => setTimeout(() => resolve(false), timeoutMs)),
]);
}
async function freePort() {
const server = net.createServer();
await new Promise((resolve, reject) => server.listen(0, "127.0.0.1", resolve).once("error", reject));
const { port } = server.address();
await new Promise((resolve) => server.close(resolve));
return port;
}
@@ -0,0 +1,51 @@
import assert from "node:assert/strict";
import { mkdtemp, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { readConfig } from "../src/config.mjs";
const base = {
EXTERNAL_DATA_PLANE_DATABASE_URL: "postgresql://user:pass@example.invalid/data_plane",
NODEDC_INTERNAL_ACCESS_TOKEN: "legacy-internal-token",
};
const directory = await mkdtemp(join(tmpdir(), "nodedc-edp-config-"));
const secretPath = join(directory, "provisioner-token");
const secret = "provisioner_secret_is_separate_from_legacy_internal_token_123456";
try {
await writeFile(secretPath, `${secret}\n`, { encoding: "utf8", mode: 0o600 });
const config = readConfig({
...base,
EXTERNAL_DATA_PLANE_PROVISIONING_ENABLED: "true",
EXTERNAL_DATA_PLANE_PROVISIONER_TOKEN_FILE: secretPath,
EXTERNAL_DATA_PLANE_MAX_FUTURE_SKEW_SECONDS: "120",
EXTERNAL_DATA_PLANE_RETENTION_SWEEP_MS: "60000",
});
assert.equal(config.provisionerAccessToken, secret);
assert.equal(config.maxFutureSkewSeconds, 120);
assert.equal(config.retentionSweepMs, 60000);
assert.equal(config.maxPatchBytes, 262144);
assert.equal(config.receiptRetentionMs, 604800000);
assert.equal(config.legacyIntakeEnabled, false);
assert.equal(readConfig({ ...base, EXTERNAL_DATA_PLANE_LEGACY_INTAKE_ENABLED: "true" }).legacyIntakeEnabled, true);
assert.throws(() => readConfig({
...base,
EXTERNAL_DATA_PLANE_PROVISIONING_ENABLED: "true",
EXTERNAL_DATA_PLANE_PROVISIONER_TOKEN_FILE: secretPath,
NODEDC_INTERNAL_ACCESS_TOKEN: secret,
}), /provisioner_token_must_differ_from_internal_token/);
assert.throws(() => readConfig({
...base,
EXTERNAL_DATA_PLANE_PROVISIONING_ENABLED: "true",
EXTERNAL_DATA_PLANE_PROVISIONER_TOKEN_FILE: join(directory, "missing"),
}), /external_data_plane_provisioner_token_file_unreadable/);
assert.equal(readConfig({
...base,
EXTERNAL_DATA_PLANE_PROVISIONING_ENABLED: "false",
EXTERNAL_DATA_PLANE_PROVISIONER_TOKEN_FILE: join(directory, "missing"),
}).provisionerAccessToken, "");
} finally {
await rm(directory, { recursive: true, force: true });
}
console.log("external-data-plane config: ok");
@@ -0,0 +1,184 @@
import assert from "node:assert/strict";
import { Pool } from "pg";
import { validateDataProductPatch, validateDataProductSnapshot } from "@nodedc/external-provider-contract";
import {
loadDataProductDefinition,
persistDataProductDefinition,
persistDataProductPublish,
readDataProductSnapshot,
readPatchEvents,
} from "../src/data-product-delivery.mjs";
import { normalizeDataProductDefinition } from "../src/data-product-policy.mjs";
import { migrate } from "../src/schema.mjs";
const databaseUrl = process.env.EXTERNAL_DATA_PLANE_TEST_DATABASE_URL;
if (!databaseUrl) throw new Error("EXTERNAL_DATA_PLANE_TEST_DATABASE_URL_required");
const pool = new Pool({ connectionString: databaseUrl, max: 4 });
try {
await migrate(pool);
await pool.query(`truncate table
external_data_plane_patch_outbox,
external_data_plane_history,
external_data_plane_current,
external_data_plane_facts,
external_data_plane_raw_envelopes,
external_data_plane_batches,
external_data_plane_delivery_state,
external_data_plane_reader_bindings,
external_data_plane_writer_bindings,
external_data_plane_products
restart identity cascade`);
const definition = normalizeDataProductDefinition({
id: "test.positions.current.v1",
version: "1.0.0",
ontologyRevision: "ontology.test.positions.v1",
deliveryMode: "snapshot+patch",
semanticTypes: ["map.moving_object"],
fields: ["source_id", "observed_at", "geometry", "status"],
history: {
mode: "sampled",
intervalMs: 60_000,
strategy: "latest-per-entity-per-bucket",
retentionDays: 30,
},
});
await persistDataProductDefinition(pool, definition);
const storedDefinition = await loadDataProductDefinition(pool, definition.id);
const binding = {
id: "reader-binding-test",
tenantId: "tenant-test",
connectionId: "connection-test",
providerId: "provider-test",
allowedDataProductIds: [definition.id],
active: true,
expiresAt: "2027-01-01T00:00:00.000Z",
};
const firstBatch = batch("run-01", "2026-07-15T10:00:05.000Z", [
fact("unit-01", 37.61, 55.75, "online"),
fact("unit-02", 37.62, 55.76, "online"),
]);
const first = await persistDataProductPublish(pool, firstBatch, storedDefinition);
assert.equal(first.idempotent, false);
assert.equal(first.currentUpdatedCount, 2);
assert.equal(first.historyInsertedCount, 2);
assert.equal(first.cursor, "1");
const duplicate = await persistDataProductPublish(pool, firstBatch, storedDefinition);
assert.equal(duplicate.idempotent, true);
assert.equal(duplicate.cursor, "1");
await assert.rejects(
persistDataProductPublish(pool, {
...firstBatch,
facts: [fact("unit-01", 99.9, 55.75, "changed")],
}, storedDefinition),
(error) => error?.status === 409 && error?.code === "idempotency_key_reused",
);
const concurrentBatch = batch("run-concurrent", "2026-07-15T10:00:08.000Z", [
fact("unit-03", 37.63, 55.77, "online", "2026-07-15T10:00:08.000Z"),
]);
const concurrent = await Promise.all([
persistDataProductPublish(pool, concurrentBatch, storedDefinition),
persistDataProductPublish(pool, concurrentBatch, storedDefinition),
]);
assert.deepEqual(concurrent.map((value) => value.idempotent).sort(), [false, true]);
const second = await persistDataProductPublish(pool, batch(
"run-02",
"2026-07-15T10:00:15.000Z",
[fact("unit-01", 37.611, 55.751, "moving", "2026-07-15T10:00:15.000Z")],
), storedDefinition);
assert.equal(second.currentUpdatedCount, 1);
assert.equal(second.cursor, "3");
const latest = await persistDataProductPublish(pool, batch(
"run-latest",
"2026-07-15T10:02:05.000Z",
[fact("unit-01", 37.7, 55.8, "latest", "2026-07-15T10:02:00.000Z")],
), storedDefinition);
assert.equal(latest.currentUpdatedCount, 1);
const late = await persistDataProductPublish(pool, batch(
"run-late",
"2026-07-15T10:03:00.000Z",
[fact("unit-01", 37.65, 55.78, "late", "2026-07-15T10:01:30.000Z")],
), storedDefinition);
assert.equal(late.currentUpdatedCount, 0);
assert.equal(late.historyInsertedCount, 1);
assert.equal(late.patchOperationCount, 0);
await assert.rejects(
persistDataProductPublish(pool, batch("run-bad-type", "2026-07-15T10:03:10.000Z", [{
...fact("unit-04", 37.6, 55.7, "online"),
semanticType: "map.forbidden",
}]), storedDefinition),
(error) => error?.status === 422 && error?.code === "data_product_semantic_type_forbidden",
);
await assert.rejects(
persistDataProductPublish(pool, batch("run-bad-field", "2026-07-15T10:03:11.000Z", [{
...fact("unit-04", 37.6, 55.7, "online"),
attributes: { undeclared: true },
}]), storedDefinition),
(error) => error?.status === 422 && error?.code === "data_product_field_forbidden",
);
await assert.rejects(
persistDataProductPublish(pool, batch("run-duplicate-fact", "2026-07-15T10:03:12.000Z", [
fact("unit-04", 37.6, 55.7, "online"),
fact("unit-04", 37.61, 55.71, "moving"),
]), storedDefinition),
(error) => error?.status === 422 && error?.code === "data_product_duplicate_entity_key",
);
const snapshot = await readDataProductSnapshot(pool, binding, storedDefinition);
assert.equal(validateDataProductSnapshot(snapshot).ok, true);
assert.equal(snapshot.cursor, "4");
assert.equal(snapshot.facts.length, 3);
assert.equal(snapshot.facts.find((value) => value.sourceId === "unit-01").attributes.status, "latest");
const patches = await readPatchEvents(pool, binding, storedDefinition, 0n);
assert.equal(patches.length, 4);
assert.equal(patches.every((patch) => validateDataProductPatch(patch).ok), true);
assert.deepEqual(patches.map((patch) => patch.cursor), ["1", "2", "3", "4"]);
await assert.rejects(
readPatchEvents(pool, binding, storedDefinition, 999n),
(error) => error?.status === 409 && error?.code === "resync_required",
);
const history = await pool.query("select source_id, observed_at from external_data_plane_history order by source_id");
assert.equal(history.rowCount, 5);
const unitOneHistory = history.rows.filter((row) => row.source_id === "unit-01");
assert.equal(unitOneHistory.some((row) => new Date(row.observed_at).toISOString() === "2026-07-15T10:01:30.000Z"), true);
assert.equal(unitOneHistory.some((row) => new Date(row.observed_at).toISOString() === "2026-07-15T10:02:00.000Z"), true);
console.log("external-data-plane delivery integration: ok");
} finally {
await pool.end();
}
function batch(runId, receivedAt, facts) {
return {
schemaVersion: "nodedc.external-provider-contract/v1",
source: { tenantId: "tenant-test", connectionId: "connection-test", providerId: "provider-test" },
contract: {
dataProductId: "test.positions.current.v1",
ontologyRevision: "ontology.test.positions.v1",
version: "1.0.0",
},
batch: { runId, sequence: 0, idempotencyKey: `${runId}.chunk-0`, receivedAt },
facts,
};
}
function fact(sourceId, longitude, latitude, status, observedAt = "2026-07-15T10:00:00.000Z") {
return {
sourceId,
semanticType: "map.moving_object",
observedAt,
attributes: { status },
geometry: { type: "Point", coordinates: [longitude, latitude] },
};
}
@@ -0,0 +1,81 @@
import assert from "node:assert/strict";
import { normalizeDataProductDefinition } from "../src/data-product-policy.mjs";
const input = {
id: "fleet.positions.current.v1",
version: "1.0.0",
ontologyRevision: "ontology.map.v1",
deliveryMode: "snapshot+patch",
semanticTypes: ["vehicle.trike", "map.moving_object"],
fields: ["status", "source_id", "observed_at", "geometry"],
history: {
mode: "sampled",
intervalMs: 60_000,
strategy: "latest-per-entity-per-bucket",
retentionDays: 365,
},
};
const definition = normalizeDataProductDefinition(input);
assert.equal(definition.history.mode, "sampled");
assert.equal(definition.history.intervalMs, 60_000);
assert.deepEqual(definition.semanticTypes, ["map.moving_object", "vehicle.trike"]);
assert.deepEqual(definition.fields, ["geometry", "observed_at", "source_id", "status"]);
assert.deepEqual(input.semanticTypes, ["vehicle.trike", "map.moving_object"]);
assert.deepEqual(input.fields, ["status", "source_id", "observed_at", "geometry"]);
assert.equal(Object.isFrozen(definition.semanticTypes), true);
assert.equal(Object.isFrozen(definition.fields), true);
const reordered = normalizeDataProductDefinition({
...input,
semanticTypes: [...input.semanticTypes].reverse(),
fields: [...input.fields].reverse(),
});
assert.deepEqual(reordered.semanticTypes, definition.semanticTypes);
assert.deepEqual(reordered.fields, definition.fields);
assert.throws(() => normalizeDataProductDefinition({ ...definition, providerId: "gelios" }), /data_product_definition_invalid/);
assert.throws(() => normalizeDataProductDefinition({ ...definition, history: { mode: "none", intervalMs: 1000 } }), /history_policy_none_has_sampling_fields/);
for (const semanticTypes of [
undefined,
"map.moving_object",
[],
["map.moving_object", null],
["map.moving_object", "Map.invalid"],
]) {
assert.throws(
() => normalizeDataProductDefinition({ ...input, semanticTypes }),
/data_product_definition_(semantic_types_invalid|shape_invalid)/,
);
}
assert.throws(
() => normalizeDataProductDefinition({ ...input, semanticTypes: ["map.moving_object", "map.moving_object"] }),
/data_product_definition_semantic_types_duplicate/,
);
assert.throws(
() => normalizeDataProductDefinition({ ...input, semanticTypes: ["map.moving_object", " map.moving_object "] }),
/data_product_definition_semantic_types_duplicate/,
);
for (const fields of [
undefined,
"source_id",
[],
["source_id", null],
["source_id", "Invalid"],
]) {
assert.throws(
() => normalizeDataProductDefinition({ ...input, fields }),
/data_product_definition_(fields_invalid|shape_invalid)/,
);
}
assert.throws(
() => normalizeDataProductDefinition({ ...input, fields: ["source_id", "source_id"] }),
/data_product_definition_fields_duplicate/,
);
assert.throws(
() => normalizeDataProductDefinition({ ...input, fields: ["source_id", " source_id "] }),
/data_product_definition_fields_duplicate/,
);
console.log("external-data-plane data product policy: ok");
@@ -0,0 +1,32 @@
import assert from "node:assert/strict";
import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { loadDataProductDefinitions } from "../src/definitions.mjs";
const bundled = await loadDataProductDefinitions();
assert.deepEqual(bundled.map((definition) => definition.id), ["fleet.positions.current.v1"]);
assert.deepEqual(bundled[0].semanticTypes, ["map.moving_object"]);
assert.equal(bundled[0].ontologyRevision, "ontology.map.moving_object.v1");
assert.equal(bundled[0].history.mode, "sampled");
assert.equal(bundled[0].history.intervalMs, 60_000);
assert.equal(bundled[0].history.retentionDays, 90);
assert.equal(bundled[0].fields.includes("geometry"), true);
assert.equal(bundled[0].fields.includes("providerUnitId"), false);
assert.equal(bundled[0].fields.includes("provider_unit_id"), false);
const directory = await mkdtemp(join(tmpdir(), "nodedc-edp-definitions-"));
try {
await writeFile(join(directory, "wrong-name.json"), JSON.stringify({
...bundled[0],
id: "another.product.v1",
}));
await assert.rejects(loadDataProductDefinitions(directory), /data_product_definition_filename_mismatch/);
await mkdir(join(directory, "ignored.json"));
await assert.rejects(loadDataProductDefinitions(directory), /data_product_definition_file_invalid/);
} finally {
await rm(directory, { recursive: true, force: true });
}
console.log("external-data-plane definitions: ok");
@@ -0,0 +1,26 @@
import assert from "node:assert/strict";
import { assertBatchTimeBounds, rawRetentionExpiry } from "../src/intake-policy.mjs";
const now = new Date("2026-07-15T12:00:00.000Z");
const batch = {
batch: { receivedAt: "2026-07-15T12:04:59.000Z" },
facts: [{ observedAt: "2026-07-15T12:05:00.000Z" }],
};
assert.doesNotThrow(() => assertBatchTimeBounds(batch, { now, maxFutureSkewSeconds: 300 }));
assert.throws(() => assertBatchTimeBounds({
...batch,
batch: { receivedAt: "2026-07-15T12:05:01.000Z" },
}, { now, maxFutureSkewSeconds: 300 }), /batch_received_at_too_far_in_future/);
assert.throws(() => assertBatchTimeBounds({
...batch,
facts: [{ observedAt: "2026-07-15T12:05:01.000Z" }],
}, { now, maxFutureSkewSeconds: 300 }), /fact_observed_at_too_far_in_future/);
assert.equal(
rawRetentionExpiry({ now, rawRetentionDays: 14 }).toISOString(),
"2026-07-29T12:00:00.000Z",
);
assert.throws(() => rawRetentionExpiry({ now, rawRetentionDays: 0 }), /raw_retention_days_invalid/);
console.log("external-data-plane intake policy: ok");
@@ -0,0 +1,25 @@
import assert from "node:assert/strict";
import {
assertReaderProduct,
createReaderToken,
hashReaderToken,
normalizeReaderBindingRequest,
safeReaderBinding,
} from "../src/reader-binding.mjs";
const now = new Date("2026-07-15T12:00:00.000Z");
const policy = normalizeReaderBindingRequest({
source: { tenantId: "tenant-01", connectionId: "connection-01", providerId: "example-provider" },
allowedDataProductIds: ["fleet.positions.current.v1"],
expiresAt: "2026-08-01T12:00:00.000Z",
}, { now });
const token = createReaderToken();
assert.match(token, /^ndc_edprb_[A-Za-z0-9_-]{40,}$/);
assert.equal(hashReaderToken(token), hashReaderToken(token));
assert.equal(assertReaderProduct({ ...policy, active: true }, "fleet.positions.current.v1", now), "fleet.positions.current.v1");
assert.throws(() => assertReaderProduct({ ...policy, active: true }, "other.product.v1", now), /reader_binding_data_product_forbidden/);
const safe = safeReaderBinding({ id: "reader-01", ...policy, active: true, token, tokenHash: hashReaderToken(token) });
assert.equal("token" in safe, false);
assert.equal("tokenHash" in safe, false);
console.log("external-data-plane reader bindings: ok");
@@ -0,0 +1,117 @@
import assert from "node:assert/strict";
import { validateIntakeBatch } from "../../../packages/external-provider-contract/src/index.mjs";
import {
createWriterToken,
hashWriterToken,
materializeDataProductPublish,
materializeWriterBoundBatch,
normalizeWriterBindingRequest,
safeWriterBinding,
} from "../src/writer-binding.mjs";
const now = new Date("2026-07-15T12:00:00.000Z");
const request = {
source: {
tenantId: "tenant-01",
connectionId: "connection-01",
providerId: "example-provider",
},
allowedDataProductIds: ["fleet.positions.current.v1"],
expiresAt: "2026-08-01T12:00:00.000Z",
};
const bindingPolicy = normalizeWriterBindingRequest(request, { now, maxTtlDays: 90 });
assert.deepEqual(bindingPolicy, {
tenantId: "tenant-01",
connectionId: "connection-01",
providerId: "example-provider",
allowedDataProductIds: ["fleet.positions.current.v1"],
expiresAt: "2026-08-01T12:00:00.000Z",
});
const token = createWriterToken();
assert.match(token, /^ndc_edpwb_[A-Za-z0-9_-]{40,}$/);
assert.equal(hashWriterToken(token), hashWriterToken(token));
assert.notEqual(hashWriterToken(token), hashWriterToken(`${token}x`));
const unscopedBatch = {
schemaVersion: "nodedc.external-provider-contract/v1",
source: { providerId: "example-provider" },
contract: { dataProductId: "fleet.positions.current.v1", ontologyRevision: "example.v1", version: "1.0.0" },
batch: { runId: "run-01", sequence: 0, idempotencyKey: "run-01.batch-0", receivedAt: "2026-07-15T12:00:00.000Z" },
facts: [{ sourceId: "unit-01", semanticType: "map.moving_object", observedAt: "2026-07-15T12:00:00.000Z" }],
};
const bound = materializeWriterBoundBatch(unscopedBatch, { ...bindingPolicy, active: true }, { now });
assert.deepEqual(bound.source, { providerId: "example-provider", tenantId: "tenant-01", connectionId: "connection-01" });
assert.equal(validateIntakeBatch(bound).ok, true);
assert.throws(() => materializeWriterBoundBatch({
...unscopedBatch,
source: { ...unscopedBatch.source, tenantId: "forged-tenant" },
}, { ...bindingPolicy, active: true }, { now }), /writer_bound_scope_forbidden/);
assert.throws(() => materializeWriterBoundBatch(unscopedBatch, { ...bindingPolicy, active: true }, {
now,
hasScopeHeaders: true,
}), /writer_bound_scope_forbidden/);
assert.throws(() => materializeWriterBoundBatch(unscopedBatch, {
...bindingPolicy,
providerId: "other-provider",
active: true,
}, { now }), /writer_binding_provider_forbidden/);
assert.throws(() => materializeWriterBoundBatch({
...unscopedBatch,
contract: { ...unscopedBatch.contract, dataProductId: "other.product.v1" },
}, { ...bindingPolicy, active: true }, { now }), /writer_binding_data_product_forbidden/);
assert.throws(() => materializeWriterBoundBatch(unscopedBatch, { ...bindingPolicy, active: false }, { now }), /writer_binding_inactive/);
const materializedPublish = materializeDataProductPublish({
schemaVersion: "nodedc.data-product.publish/v1",
batch: { runId: "run-02", sequence: 0, idempotencyKey: "run-02.batch-0" },
facts: unscopedBatch.facts,
}, { ...bindingPolicy, active: true }, {
id: "fleet.positions.current.v1",
version: "1.0.0",
ontologyRevision: "ontology.example-fleet.v1",
}, "fleet.positions.current.v1", { now });
assert.deepEqual(materializedPublish.source, {
tenantId: "tenant-01",
connectionId: "connection-01",
providerId: "example-provider",
});
assert.equal(materializedPublish.batch.receivedAt, now.toISOString());
assert.equal(materializedPublish.contract.version, "1.0.0");
assert.equal(validateIntakeBatch(materializedPublish).ok, true);
assert.throws(() => materializeDataProductPublish({
schemaVersion: "nodedc.data-product.publish/v1",
batch: { runId: "run-02", sequence: 0, idempotencyKey: "run-02.batch-0" },
facts: unscopedBatch.facts,
}, { ...bindingPolicy, active: true }, {
id: "other.product.v1",
version: "1.0.0",
ontologyRevision: "ontology.example-fleet.v1",
}, "other.product.v1", { now }), /writer_binding_data_product_forbidden/);
assert.throws(() => normalizeWriterBindingRequest({
...request,
expiresAt: "2027-01-01T00:00:00.000Z",
}, { now, maxTtlDays: 90 }), /writer_binding_expiry_invalid/);
assert.throws(() => normalizeWriterBindingRequest({
...request,
apiToken: "must-never-be-here",
}, { now, maxTtlDays: 90 }), /writer_binding_request_secret_material_forbidden/);
assert.throws(() => normalizeWriterBindingRequest({
...request,
arbitraryRuntimeSetting: "must-not-be-accepted",
}, { now, maxTtlDays: 90 }), /writer_binding_request_fields_invalid/);
const safeBinding = safeWriterBinding({
id: "binding-01",
...bindingPolicy,
active: true,
createdAt: now,
token: "must-not-leak",
tokenHash: "must-not-leak",
});
assert.equal("token" in safeBinding, false);
assert.equal("tokenHash" in safeBinding, false);
console.log("external-data-plane writer bindings: ok");