983 lines
40 KiB
JavaScript
983 lines
40 KiB
JavaScript
import { createHash, randomUUID } from "node:crypto";
|
|
import { mkdir, readFile, readdir, rename, writeFile } from "node:fs/promises";
|
|
import { join } from "node:path";
|
|
|
|
const STATE_SCHEMA_VERSION = "nodedc.module-foundry.data-product-consumer/v1";
|
|
const PLAN_SCHEMA_VERSION = "nodedc.module-foundry.data-product-consumer-plan/v1";
|
|
const ACCEPTANCE_SCHEMA_VERSION = "nodedc.module-foundry.data-product-consumer-acceptance/v1";
|
|
const PRESENTATION_PATCH_SCHEMA_VERSION = "nodedc.foundry.presentation-patch/v1";
|
|
const identifier = /^[A-Za-z0-9._:-]{1,160}$/;
|
|
const cursorPattern = /^(?:0|[1-9]\d*)$/;
|
|
|
|
function consumerError(code, statusCode = 400) {
|
|
return Object.assign(new Error(code), { statusCode });
|
|
}
|
|
|
|
function targetIdentity(target) {
|
|
const applicationId = String(target?.application?.id || target?.applicationId || "");
|
|
const pageId = String(target?.page?.id || target?.pageId || "");
|
|
const binding = target?.binding || {};
|
|
const bindingId = String(binding.id || target?.bindingId || "");
|
|
if (!/^[0-9a-f-]{36}$/i.test(applicationId) || !/^[a-z0-9-]+$/i.test(pageId) || !identifier.test(bindingId)) {
|
|
throw consumerError("data_product_consumer_target_invalid");
|
|
}
|
|
return { applicationId, pageId, bindingId };
|
|
}
|
|
|
|
function targetKey(target) {
|
|
const identity = targetIdentity(target);
|
|
return `${identity.applicationId}/${identity.pageId}/${identity.bindingId}`;
|
|
}
|
|
|
|
function stateFileName(target) {
|
|
return `${createHash("sha256").update(targetKey(target), "utf8").digest("hex")}.json`;
|
|
}
|
|
|
|
function factKey(fact) {
|
|
return `${fact.semanticType}\u0000${fact.sourceId}`;
|
|
}
|
|
|
|
function safeErrorCode(error) {
|
|
const value = String(error?.message || "data_product_consumer_failed");
|
|
return /^[a-z][a-z0-9_]{2,120}$/.test(value) ? value : "data_product_consumer_failed";
|
|
}
|
|
|
|
function safeStatusFromFact(fact, policy, nowMs) {
|
|
const statusContract = policy.statusContract || null;
|
|
if (statusContract) {
|
|
const sourceStatus = fact?.attributes?.[statusContract.attribute];
|
|
if (typeof sourceStatus !== "string" || !statusContract.allowedValues.includes(sourceStatus)) {
|
|
throw consumerError("data_product_consumer_fact_status_invalid", 502);
|
|
}
|
|
if (statusContract.freshness === "none" || policy.terminalStatuses.includes(sourceStatus)) {
|
|
return sourceStatus;
|
|
}
|
|
const observedAt = Date.parse(String(fact?.observedAt || ""));
|
|
if (Number.isFinite(observedAt) && nowMs - observedAt > policy.staleAfterMs) return "stale";
|
|
return sourceStatus;
|
|
}
|
|
const sourceStatus = [fact?.attributes?.operational_status, fact?.attributes?.status]
|
|
.find((value) => typeof value === "string" && value.trim());
|
|
const normalized = String(sourceStatus || "active").trim().toLowerCase().replace(/[^a-z0-9_-]+/g, "-").slice(0, 64) || "active";
|
|
if (policy.terminalStatuses.includes(normalized)) return normalized;
|
|
const observedAt = Date.parse(String(fact?.observedAt || ""));
|
|
if (Number.isFinite(observedAt) && nowMs - observedAt > policy.staleAfterMs) return "stale";
|
|
return normalized;
|
|
}
|
|
|
|
function decorateFact(subject) {
|
|
return { ...subject.fact, presentationStatus: subject.status };
|
|
}
|
|
|
|
function safeTarget(target) {
|
|
const identity = targetIdentity(target);
|
|
const binding = target.binding || {};
|
|
return {
|
|
...identity,
|
|
dataProductId: String(binding.dataProductId || ""),
|
|
slotId: String(binding.slotId || ""),
|
|
delivery: "snapshot+patch",
|
|
semanticTypes: [...(binding.semanticTypes || [])],
|
|
fieldProjection: [...(binding.fieldProjection || [])],
|
|
};
|
|
}
|
|
|
|
function planHash(value) {
|
|
return `fcp1_${createHash("sha256").update(JSON.stringify(value), "utf8").digest("base64url")}`;
|
|
}
|
|
|
|
function emptyMetrics() {
|
|
return {
|
|
snapshotCommits: 0,
|
|
patchCommits: 0,
|
|
patchOperations: 0,
|
|
replayedPatches: 0,
|
|
snapshotRebaseRemovals: 0,
|
|
canonicalRemovals: 0,
|
|
staleTransitions: 0,
|
|
reconnects: 0,
|
|
};
|
|
}
|
|
|
|
function safeStateSummary(record) {
|
|
const state = record.state;
|
|
const subjects = Object.values(state.subjects || {});
|
|
return {
|
|
schemaVersion: STATE_SCHEMA_VERSION,
|
|
consumerId: state.consumerId,
|
|
target: state.target,
|
|
enabled: state.enabled === true,
|
|
runtimeState: state.runtimeState,
|
|
product: state.product,
|
|
policy: state.policy,
|
|
cursor: state.cursor,
|
|
snapshotGeneration: state.snapshotGeneration,
|
|
subjectCount: subjects.length,
|
|
subjects: subjects.slice(0, 20).map((subject) => ({
|
|
subjectId: subject.fact.sourceId,
|
|
semanticType: subject.fact.semanticType,
|
|
observedAt: subject.fact.observedAt,
|
|
status: subject.status,
|
|
})),
|
|
viewerCount: record.listeners.size,
|
|
upstreamStreamCount: record.stream ? 1 : 0,
|
|
metrics: state.metrics,
|
|
timestamps: state.timestamps,
|
|
lastError: state.lastError || null,
|
|
readerGrant: "target-scoped-server-only",
|
|
readerGrantGeneration: Number.isSafeInteger(state.readerGrantGeneration) ? state.readerGrantGeneration : 1,
|
|
};
|
|
}
|
|
|
|
function activeReaderGrantGeneration(state) {
|
|
const generation = Number(state?.readerGrantGeneration ?? 1);
|
|
if (!Number.isSafeInteger(generation) || generation < 1 || generation > 2_147_483_647) {
|
|
throw consumerError("data_product_reader_grant_generation_invalid", 500);
|
|
}
|
|
return generation;
|
|
}
|
|
|
|
function validatePolicy(policy, product) {
|
|
if (!policy || policy.dataProductId !== product.id || policy.productVersion !== product.version) {
|
|
throw consumerError("data_product_consumer_policy_not_found", 409);
|
|
}
|
|
const statusContract = policy.statusContract === undefined
|
|
? null
|
|
: validateStatusContract(policy.statusContract);
|
|
const staleDisabled = statusContract?.freshness === "none";
|
|
if (
|
|
(staleDisabled && policy.staleAfterMs !== null)
|
|
|| (!staleDisabled && (!Number.isInteger(policy.staleAfterMs) || policy.staleAfterMs < 1_000 || policy.staleAfterMs > 7 * 24 * 60 * 60 * 1000))
|
|
) {
|
|
throw consumerError("data_product_consumer_policy_invalid", 500);
|
|
}
|
|
const terminalStatuses = Array.isArray(policy.terminalStatuses) ? policy.terminalStatuses : [];
|
|
if (terminalStatuses.some((value) => typeof value !== "string" || !/^[a-z0-9_-]{1,64}$/.test(value))) {
|
|
throw consumerError("data_product_consumer_policy_invalid", 500);
|
|
}
|
|
if (statusContract && terminalStatuses.some((value) => !statusContract.allowedValues.includes(value))) {
|
|
throw consumerError("data_product_consumer_policy_invalid", 500);
|
|
}
|
|
if (policy.removeMode !== "canonical-tombstone-or-snapshot-rebase") {
|
|
throw consumerError("data_product_consumer_policy_invalid", 500);
|
|
}
|
|
return {
|
|
id: String(policy.id || ""),
|
|
version: String(policy.version || ""),
|
|
dataProductId: product.id,
|
|
productVersion: product.version,
|
|
staleAfterMs: policy.staleAfterMs,
|
|
terminalStatuses: [...terminalStatuses],
|
|
...(statusContract ? { statusContract } : {}),
|
|
removeMode: policy.removeMode,
|
|
};
|
|
}
|
|
|
|
function validateStatusContract(value) {
|
|
const keys = value && typeof value === "object" && !Array.isArray(value)
|
|
? Object.keys(value).sort()
|
|
: [];
|
|
if (JSON.stringify(keys) !== JSON.stringify(["allowedValues", "attribute", "freshness", "missing"])) {
|
|
throw consumerError("data_product_consumer_policy_invalid", 500);
|
|
}
|
|
const attribute = String(value.attribute || "");
|
|
const allowedValues = Array.isArray(value.allowedValues) ? value.allowedValues : [];
|
|
if (
|
|
!/^[A-Za-z][A-Za-z0-9_.-]{0,127}$/.test(attribute)
|
|
|| allowedValues.length === 0
|
|
|| allowedValues.length > 32
|
|
|| new Set(allowedValues).size !== allowedValues.length
|
|
|| allowedValues.some((item) => typeof item !== "string" || !/^[a-z0-9_-]{1,64}$/.test(item))
|
|
|| value.missing !== "reject"
|
|
|| !["none", "observed-at"].includes(value.freshness)
|
|
) {
|
|
throw consumerError("data_product_consumer_policy_invalid", 500);
|
|
}
|
|
return {
|
|
attribute,
|
|
allowedValues: [...allowedValues],
|
|
missing: "reject",
|
|
freshness: value.freshness,
|
|
};
|
|
}
|
|
|
|
function parseSseBlock(block) {
|
|
const fields = { event: "message", id: "", data: [] };
|
|
for (const line of block.split("\n")) {
|
|
if (!line || line.startsWith(":")) continue;
|
|
const separator = line.indexOf(":");
|
|
const name = separator === -1 ? line : line.slice(0, separator);
|
|
const value = separator === -1 ? "" : line.slice(separator + 1).replace(/^ /, "");
|
|
if (name === "event") fields.event = value;
|
|
if (name === "id") fields.id = value;
|
|
if (name === "data") fields.data.push(value);
|
|
}
|
|
return { event: fields.event, id: fields.id, data: fields.data.join("\n") };
|
|
}
|
|
|
|
function waitForAbortableDelay(ms, signal) {
|
|
if (signal.aborted) return Promise.resolve();
|
|
return new Promise((resolve) => {
|
|
const timer = setTimeout(done, ms);
|
|
function done() {
|
|
clearTimeout(timer);
|
|
signal.removeEventListener("abort", done);
|
|
resolve();
|
|
}
|
|
signal.addEventListener("abort", done, { once: true });
|
|
});
|
|
}
|
|
|
|
export function createFoundryDataProductConsumerManager({
|
|
stateDir,
|
|
dataPlaneUrl,
|
|
resolveTarget,
|
|
readReaderToken,
|
|
inspectReaderGrant,
|
|
ensureReaderGrant,
|
|
revokeReaderGrant,
|
|
resolvePolicy,
|
|
sanitizeSnapshot,
|
|
sanitizePatch,
|
|
fetchImpl = fetch,
|
|
openStream,
|
|
now = () => Date.now(),
|
|
idleStopMs = 2_000,
|
|
staleSweepMs = 5_000,
|
|
reconnectMinMs = 500,
|
|
reconnectMaxMs = 15_000,
|
|
}) {
|
|
const records = new Map();
|
|
const ready = mkdir(stateDir, { recursive: true });
|
|
let closed = false;
|
|
|
|
async function writeState(record) {
|
|
await ready;
|
|
const targetPath = join(stateDir, stateFileName(record.state.target));
|
|
const tempPath = `${targetPath}.${randomUUID()}.tmp`;
|
|
await writeFile(tempPath, `${JSON.stringify(record.state, null, 2)}\n`, { encoding: "utf8", mode: 0o600 });
|
|
await rename(tempPath, targetPath);
|
|
}
|
|
|
|
async function readState(target) {
|
|
await ready;
|
|
try {
|
|
const parsed = JSON.parse(await readFile(join(stateDir, stateFileName(target)), "utf8"));
|
|
if (parsed?.schemaVersion !== STATE_SCHEMA_VERSION || targetKey(parsed.target) !== targetKey(target)) {
|
|
throw consumerError("data_product_consumer_state_invalid", 500);
|
|
}
|
|
return parsed;
|
|
} catch (error) {
|
|
if (error?.code === "ENOENT") return null;
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async function recordFor(target) {
|
|
const key = targetKey(target);
|
|
if (records.has(key)) return records.get(key);
|
|
const persisted = await readState(target);
|
|
const record = {
|
|
key,
|
|
target,
|
|
state: persisted,
|
|
listeners: new Map(),
|
|
stream: null,
|
|
idleTimer: null,
|
|
writing: Promise.resolve(),
|
|
};
|
|
records.set(key, record);
|
|
return record;
|
|
}
|
|
|
|
function serialize(record, action) {
|
|
const pending = record.writing.then(action, action);
|
|
record.writing = pending.catch(() => undefined);
|
|
return pending;
|
|
}
|
|
|
|
function emit(record, event) {
|
|
for (const listener of record.listeners.values()) {
|
|
try { listener(event); } catch { /* one browser cannot stop the shared consumer */ }
|
|
}
|
|
}
|
|
|
|
async function catalog(target, { generation = 1, ensureGeneration = generation } = {}) {
|
|
if (!dataPlaneUrl) throw consumerError("data_product_runtime_not_configured", 503);
|
|
let product = null;
|
|
let readerGrantAction = "reuse";
|
|
let readerGrantGeneration = generation;
|
|
if (inspectReaderGrant) {
|
|
const inspection = await inspectReaderGrant(target, { generation });
|
|
product = inspection?.product || null;
|
|
readerGrantAction = inspection?.readerGrantAction === "ensure" ? "ensure" : "reuse";
|
|
readerGrantGeneration = readerGrantAction === "ensure" ? ensureGeneration : generation;
|
|
} else {
|
|
const token = await readReaderToken(target.application.id, target.page.id, target.binding.id, generation);
|
|
let response;
|
|
try {
|
|
response = await fetchImpl(new URL("/internal/data-plane/v1/reader/data-products", `${dataPlaneUrl}/`), {
|
|
headers: { authorization: `Bearer ${token}`, accept: "application/json" },
|
|
signal: AbortSignal.timeout(10_000),
|
|
});
|
|
} catch {
|
|
throw consumerError("data_product_consumer_catalog_unavailable", 503);
|
|
}
|
|
if (response.status === 401 || response.status === 403) throw consumerError("data_product_access_denied", 403);
|
|
if (!response.ok) throw consumerError("data_product_consumer_catalog_unavailable", 503);
|
|
const payload = await response.json().catch(() => null);
|
|
product = Array.isArray(payload?.dataProducts)
|
|
? payload.dataProducts.find((candidate) => candidate?.id === target.binding.dataProductId)
|
|
: null;
|
|
}
|
|
if (!product || product.active === false) throw consumerError("data_product_not_found", 404);
|
|
if (product.deliveryMode !== "snapshot+patch" || typeof product.version !== "string") {
|
|
throw consumerError("data_product_consumer_delivery_not_supported", 409);
|
|
}
|
|
if (target.binding.semanticTypes.some((semanticType) => !product.semanticTypes?.includes(semanticType))) {
|
|
throw consumerError("data_product_consumer_semantic_scope_mismatch", 409);
|
|
}
|
|
const policy = validatePolicy(resolvePolicy(product), product);
|
|
if (policy.statusContract && !target.binding.fieldProjection.includes(policy.statusContract.attribute)) {
|
|
throw consumerError("data_product_consumer_status_field_not_projected", 409);
|
|
}
|
|
return {
|
|
product,
|
|
policy,
|
|
readerGrantAction,
|
|
readerGrantGeneration,
|
|
};
|
|
}
|
|
|
|
async function plan(input) {
|
|
const target = await resolveTarget(input.applicationId, input.pageId, input.bindingId);
|
|
const record = await recordFor(target);
|
|
const currentReaderGrantGeneration = activeReaderGrantGeneration(record.state);
|
|
const ensureReaderGrantGeneration = record.state
|
|
? Math.min(2_147_483_647, currentReaderGrantGeneration + 1)
|
|
: currentReaderGrantGeneration;
|
|
const { product, policy, readerGrantAction, readerGrantGeneration } = await catalog(target, {
|
|
generation: currentReaderGrantGeneration,
|
|
ensureGeneration: ensureReaderGrantGeneration,
|
|
});
|
|
if (readerGrantAction === "ensure" && record.state && currentReaderGrantGeneration === 2_147_483_647) {
|
|
throw consumerError("data_product_reader_grant_generation_exhausted", 409);
|
|
}
|
|
const safe = safeTarget(target);
|
|
const configuration = {
|
|
target: safe,
|
|
product: { id: product.id, version: product.version },
|
|
policy,
|
|
readerGrant: "target-scoped-server-only",
|
|
readerGrantAction,
|
|
readerGrantGeneration,
|
|
};
|
|
const existingMatches = record.state
|
|
&& JSON.stringify(record.state.target) === JSON.stringify(safe)
|
|
&& record.state.product?.id === product.id
|
|
&& record.state.product?.version === product.version
|
|
&& record.state.policy?.id === policy.id
|
|
&& record.state.policy?.version === policy.version;
|
|
const predecessorReaderGrantGeneration = Number.isSafeInteger(record.state?.predecessorReaderGrantGeneration)
|
|
? record.state.predecessorReaderGrantGeneration
|
|
: null;
|
|
const action = !record.state
|
|
? "create"
|
|
: existingMatches && predecessorReaderGrantGeneration
|
|
? "finalize-reader-grant-rotation"
|
|
: existingMatches
|
|
? (record.state.enabled ? "refresh" : "resume")
|
|
: "replace";
|
|
return {
|
|
schemaVersion: PLAN_SCHEMA_VERSION,
|
|
planId: planHash({ configuration, action }),
|
|
action,
|
|
configuration,
|
|
current: record.state ? { enabled: record.state.enabled === true, cursor: record.state.cursor, product: record.state.product } : null,
|
|
effects: [
|
|
...(readerGrantAction === "ensure"
|
|
? [record.state ? "ensure-successor-target-scoped-reader-grant" : "ensure-target-scoped-reader-grant"]
|
|
: []),
|
|
"persist-consumer-state",
|
|
...(action === "finalize-reader-grant-rotation" ? [] : ["bootstrap-scoped-snapshot"]),
|
|
...(predecessorReaderGrantGeneration ? ["revoke-predecessor-reader-grant"] : []),
|
|
"share-one-upstream-stream-per-active-binding",
|
|
],
|
|
destructive: false,
|
|
};
|
|
}
|
|
|
|
function initialState(target, product, policy, readerGrantGeneration) {
|
|
const timestamp = new Date(now()).toISOString();
|
|
return {
|
|
schemaVersion: STATE_SCHEMA_VERSION,
|
|
consumerId: createHash("sha256").update(targetKey(target), "utf8").digest("hex").slice(0, 32),
|
|
target: safeTarget(target),
|
|
enabled: true,
|
|
runtimeState: "bootstrapping",
|
|
product: { id: product.id, version: product.version },
|
|
policy,
|
|
readerGrantGeneration,
|
|
predecessorReaderGrantGeneration: null,
|
|
cursor: "0",
|
|
snapshotGeneration: 0,
|
|
subjects: {},
|
|
metrics: emptyMetrics(),
|
|
timestamps: { createdAt: timestamp, updatedAt: timestamp, lastSnapshotAt: null, lastPatchAt: null, lastConnectedAt: null },
|
|
lastError: null,
|
|
};
|
|
}
|
|
|
|
function upstreamUrl(record, resource, after = "") {
|
|
const url = new URL(`/internal/data-plane/v1/data-products/${encodeURIComponent(record.state.target.dataProductId)}/${resource}`, `${dataPlaneUrl}/`);
|
|
if (after) url.searchParams.set("after", after);
|
|
return url;
|
|
}
|
|
|
|
async function fetchSnapshot(record) {
|
|
const token = await readReaderToken(
|
|
record.state.target.applicationId,
|
|
record.state.target.pageId,
|
|
record.state.target.bindingId,
|
|
activeReaderGrantGeneration(record.state),
|
|
);
|
|
let response;
|
|
try {
|
|
response = await fetchImpl(upstreamUrl(record, "snapshot"), {
|
|
headers: { authorization: `Bearer ${token}`, accept: "application/json" },
|
|
signal: AbortSignal.timeout(10_000),
|
|
});
|
|
} catch {
|
|
throw consumerError("data_product_runtime_unavailable", 503);
|
|
}
|
|
if (response.status === 401 || response.status === 403) throw consumerError("data_product_access_denied", 403);
|
|
if (response.status === 404) throw consumerError("data_product_not_found", 404);
|
|
if (!response.ok) throw consumerError("data_product_runtime_unavailable", 503);
|
|
const payload = await response.json().catch(() => null);
|
|
return sanitizeSnapshot(payload, record.target.binding);
|
|
}
|
|
|
|
async function bootstrap(record, { notify = true } = {}) {
|
|
return serialize(record, async () => {
|
|
record.state.runtimeState = "bootstrapping";
|
|
record.state.timestamps.updatedAt = new Date(now()).toISOString();
|
|
await writeState(record);
|
|
try {
|
|
const snapshot = await fetchSnapshot(record);
|
|
const previousKeys = new Set(Object.keys(record.state.subjects || {}));
|
|
const subjects = {};
|
|
for (const fact of snapshot.facts) {
|
|
const key = factKey(fact);
|
|
subjects[key] = {
|
|
fact,
|
|
status: safeStatusFromFact(fact, record.state.policy, now()),
|
|
lastAppliedCursor: snapshot.cursor,
|
|
};
|
|
previousKeys.delete(key);
|
|
}
|
|
const timestamp = new Date(now()).toISOString();
|
|
record.state = {
|
|
...record.state,
|
|
enabled: true,
|
|
runtimeState: record.listeners.size ? "connecting" : "idle",
|
|
product: snapshot.dataProduct,
|
|
cursor: snapshot.cursor,
|
|
snapshotGeneration: record.state.snapshotGeneration + 1,
|
|
subjects,
|
|
metrics: {
|
|
...record.state.metrics,
|
|
snapshotCommits: record.state.metrics.snapshotCommits + 1,
|
|
snapshotRebaseRemovals: record.state.metrics.snapshotRebaseRemovals + previousKeys.size,
|
|
},
|
|
timestamps: { ...record.state.timestamps, updatedAt: timestamp, lastSnapshotAt: timestamp },
|
|
lastError: null,
|
|
};
|
|
await writeState(record);
|
|
if (notify) emit(record, {
|
|
event: "nodedc.data-product.resync-required.v1",
|
|
data: { schemaVersion: "nodedc.data-product.resync-required/v1", dataProductId: record.state.product.id },
|
|
});
|
|
return snapshot;
|
|
} catch (error) {
|
|
const timestamp = new Date(now()).toISOString();
|
|
record.state.runtimeState = "error";
|
|
record.state.lastError = { code: safeErrorCode(error), at: timestamp };
|
|
record.state.timestamps.updatedAt = timestamp;
|
|
await writeState(record);
|
|
throw error;
|
|
}
|
|
});
|
|
}
|
|
|
|
async function apply(input) {
|
|
const planned = await plan(input);
|
|
if (input.planId !== planned.planId) throw consumerError("data_product_consumer_plan_mismatch", 409);
|
|
const target = await resolveTarget(input.applicationId, input.pageId, input.bindingId);
|
|
const record = await recordFor(target);
|
|
const plannedGeneration = planned.configuration.readerGrantGeneration;
|
|
|
|
if (planned.action === "finalize-reader-grant-rotation") {
|
|
const predecessorGeneration = record.state?.predecessorReaderGrantGeneration;
|
|
if (!revokeReaderGrant || !Number.isSafeInteger(predecessorGeneration)) {
|
|
throw consumerError("data_product_reader_grant_rotation_invalid", 500);
|
|
}
|
|
await revokeReaderGrant(record.target, { generation: predecessorGeneration });
|
|
record.state.predecessorReaderGrantGeneration = null;
|
|
record.state.timestamps.updatedAt = new Date(now()).toISOString();
|
|
record.state.lastError = null;
|
|
await writeState(record);
|
|
return { plan: planned, consumer: safeStateSummary(record) };
|
|
}
|
|
|
|
if (planned.configuration.readerGrantAction === "ensure") {
|
|
if (!ensureReaderGrant) throw consumerError("data_product_reader_grant_not_found", 403);
|
|
await ensureReaderGrant(target, { generation: plannedGeneration });
|
|
}
|
|
const { product, policy, readerGrantAction, readerGrantGeneration } = await catalog(target, {
|
|
generation: plannedGeneration,
|
|
ensureGeneration: plannedGeneration,
|
|
});
|
|
if (readerGrantAction !== "reuse") throw consumerError("data_product_reader_grant_not_ready", 409);
|
|
if (readerGrantGeneration !== plannedGeneration) throw consumerError("data_product_reader_grant_generation_mismatch", 409);
|
|
|
|
const previousTarget = record.target;
|
|
const previousState = record.state ? structuredClone(record.state) : null;
|
|
const previousGeneration = previousState ? activeReaderGrantGeneration(previousState) : null;
|
|
if (record.stream) await stopStreamAndWait(record);
|
|
if (!record.state || planned.action === "replace") {
|
|
record.target = target;
|
|
record.state = initialState(target, product, policy, plannedGeneration);
|
|
if (previousGeneration && previousGeneration !== plannedGeneration) {
|
|
record.state.predecessorReaderGrantGeneration = previousGeneration;
|
|
}
|
|
}
|
|
else {
|
|
record.target = target;
|
|
record.state.target = safeTarget(target);
|
|
record.state.product = { id: product.id, version: product.version };
|
|
record.state.policy = policy;
|
|
record.state.readerGrantGeneration = plannedGeneration;
|
|
record.state.enabled = true;
|
|
record.state.runtimeState = "bootstrapping";
|
|
record.state.lastError = null;
|
|
}
|
|
try {
|
|
await writeState(record);
|
|
await bootstrap(record, { notify: false });
|
|
} catch (error) {
|
|
if (previousState) {
|
|
record.target = previousTarget;
|
|
record.state = previousState;
|
|
record.state.runtimeState = record.listeners.size ? "connecting" : "idle";
|
|
record.state.timestamps.updatedAt = new Date(now()).toISOString();
|
|
await writeState(record);
|
|
if (record.listeners.size) startStream(record);
|
|
}
|
|
throw error;
|
|
}
|
|
|
|
const predecessorGeneration = record.state.predecessorReaderGrantGeneration;
|
|
if (Number.isSafeInteger(predecessorGeneration)) {
|
|
if (!revokeReaderGrant) throw consumerError("data_product_reader_grant_rotation_not_configured", 503);
|
|
await revokeReaderGrant(previousTarget, { generation: predecessorGeneration });
|
|
record.state.predecessorReaderGrantGeneration = null;
|
|
record.state.timestamps.updatedAt = new Date(now()).toISOString();
|
|
await writeState(record);
|
|
}
|
|
if (record.listeners.size) startStream(record);
|
|
return { plan: planned, consumer: safeStateSummary(record) };
|
|
}
|
|
|
|
async function ensureProvisioned(target) {
|
|
const record = await recordFor(target);
|
|
if (record.state) {
|
|
if (!record.state.enabled) throw consumerError("data_product_consumer_stopped", 409);
|
|
record.target = target;
|
|
return record;
|
|
}
|
|
const planned = await plan(targetIdentity(target));
|
|
await apply({ ...targetIdentity(target), planId: planned.planId });
|
|
return recordFor(target);
|
|
}
|
|
|
|
async function refreshPresentation(record) {
|
|
if (!record.state?.enabled) return [];
|
|
return serialize(record, async () => {
|
|
const operations = [];
|
|
for (const subject of Object.values(record.state.subjects || {})) {
|
|
const status = safeStatusFromFact(subject.fact, record.state.policy, now());
|
|
if (status === subject.status) continue;
|
|
subject.status = status;
|
|
operations.push({ op: "upsert", sourceId: subject.fact.sourceId, semanticType: subject.fact.semanticType, status });
|
|
}
|
|
if (!operations.length) return operations;
|
|
const timestamp = new Date(now()).toISOString();
|
|
record.state.metrics.staleTransitions += operations.filter((operation) => operation.status === "stale").length;
|
|
record.state.timestamps.updatedAt = timestamp;
|
|
await writeState(record);
|
|
emit(record, {
|
|
event: "nodedc.foundry.presentation-patch.v1",
|
|
data: { schemaVersion: PRESENTATION_PATCH_SCHEMA_VERSION, generatedAt: timestamp, operations },
|
|
});
|
|
return operations;
|
|
});
|
|
}
|
|
|
|
async function applyPatch(record, patch) {
|
|
return serialize(record, async () => {
|
|
const currentCursor = BigInt(record.state.cursor);
|
|
const patchCursor = BigInt(patch.cursor);
|
|
if (patchCursor <= currentCursor) {
|
|
record.state.metrics.replayedPatches += 1;
|
|
return { replayed: true };
|
|
}
|
|
if (patch.previousCursor !== record.state.cursor) throw consumerError("resync_required", 409);
|
|
const subjects = { ...record.state.subjects };
|
|
const emittedOperations = [];
|
|
for (const operation of patch.operations) {
|
|
if (operation.op === "upsert") {
|
|
const key = factKey(operation.fact);
|
|
const subject = {
|
|
fact: operation.fact,
|
|
status: safeStatusFromFact(operation.fact, record.state.policy, now()),
|
|
lastAppliedCursor: patch.cursor,
|
|
};
|
|
subjects[key] = subject;
|
|
emittedOperations.push({ op: "upsert", fact: decorateFact(subject) });
|
|
} else if (operation.op === "remove") {
|
|
const key = `${operation.semanticType}\u0000${operation.sourceId}`;
|
|
if (subjects[key]) {
|
|
delete subjects[key];
|
|
emittedOperations.push(operation);
|
|
}
|
|
}
|
|
}
|
|
const timestamp = new Date(now()).toISOString();
|
|
record.state = {
|
|
...record.state,
|
|
runtimeState: "connected",
|
|
cursor: patch.cursor,
|
|
subjects,
|
|
metrics: {
|
|
...record.state.metrics,
|
|
patchCommits: record.state.metrics.patchCommits + 1,
|
|
patchOperations: record.state.metrics.patchOperations + emittedOperations.length,
|
|
canonicalRemovals: record.state.metrics.canonicalRemovals + emittedOperations.filter((operation) => operation.op === "remove").length,
|
|
},
|
|
timestamps: { ...record.state.timestamps, updatedAt: timestamp, lastPatchAt: timestamp },
|
|
lastError: null,
|
|
};
|
|
// Persist the new cursor and semantic state before acknowledging it to
|
|
// any browser listener. A crash can replay a patch but cannot skip one.
|
|
await writeState(record);
|
|
if (emittedOperations.length) emit(record, {
|
|
event: "nodedc.data-product.patch.v1",
|
|
id: patch.cursor,
|
|
data: { ...patch, operations: emittedOperations },
|
|
});
|
|
return { replayed: false };
|
|
});
|
|
}
|
|
|
|
async function consumeOnce(record, signal) {
|
|
const token = await readReaderToken(
|
|
record.state.target.applicationId,
|
|
record.state.target.pageId,
|
|
record.state.target.bindingId,
|
|
activeReaderGrantGeneration(record.state),
|
|
);
|
|
const url = upstreamUrl(record, "stream", record.state.cursor);
|
|
const response = openStream
|
|
? await openStream({ url, token, signal })
|
|
: await fetchImpl(url, {
|
|
headers: { authorization: `Bearer ${token}`, accept: "text/event-stream", connection: "close" },
|
|
signal,
|
|
});
|
|
if (response.status === 409) throw consumerError("resync_required", 409);
|
|
if (response.status === 401 || response.status === 403) throw consumerError("data_product_access_denied", 403);
|
|
if (!response.ok || !response.body || !String(response.headers.get("content-type") || "").startsWith("text/event-stream")) {
|
|
throw consumerError("data_product_runtime_unavailable", 503);
|
|
}
|
|
const timestamp = new Date(now()).toISOString();
|
|
record.state.runtimeState = "connected";
|
|
record.state.timestamps.lastConnectedAt = timestamp;
|
|
record.state.timestamps.updatedAt = timestamp;
|
|
record.state.lastError = null;
|
|
await writeState(record);
|
|
const reader = response.body.getReader();
|
|
if (record.stream) record.stream.reader = reader;
|
|
const decoder = new TextDecoder();
|
|
let pending = "";
|
|
let upstreamDone = false;
|
|
try {
|
|
while (!signal.aborted && record.listeners.size) {
|
|
const next = await reader.read();
|
|
if (next.done) {
|
|
upstreamDone = true;
|
|
break;
|
|
}
|
|
pending += decoder.decode(next.value, { stream: true }).replace(/\r\n/g, "\n");
|
|
let separator;
|
|
while ((separator = pending.indexOf("\n\n")) !== -1) {
|
|
const event = parseSseBlock(pending.slice(0, separator));
|
|
pending = pending.slice(separator + 2);
|
|
if (event.event !== "nodedc.data-product.patch.v1") continue;
|
|
let payload = null;
|
|
try { payload = JSON.parse(event.data); } catch { payload = null; }
|
|
const patch = sanitizePatch(payload, record.target.binding);
|
|
if (!patch) throw consumerError("data_product_patch_contract_invalid", 502);
|
|
await applyPatch(record, patch);
|
|
}
|
|
}
|
|
} finally {
|
|
if (!upstreamDone) await reader.cancel().catch(() => undefined);
|
|
reader.releaseLock();
|
|
if (record.stream?.reader === reader) record.stream.reader = null;
|
|
}
|
|
}
|
|
|
|
function startStream(record) {
|
|
if (closed || record.stream || !record.state?.enabled || !record.listeners.size) return;
|
|
if (record.idleTimer) clearTimeout(record.idleTimer);
|
|
record.idleTimer = null;
|
|
const controller = new AbortController();
|
|
record.stream = { controller, promise: null, reader: null };
|
|
record.stream.promise = (async () => {
|
|
let delay = reconnectMinMs;
|
|
while (!closed && !controller.signal.aborted && record.listeners.size && record.state.enabled) {
|
|
try {
|
|
record.state.runtimeState = "connecting";
|
|
await writeState(record);
|
|
await consumeOnce(record, controller.signal);
|
|
delay = reconnectMinMs;
|
|
if (!controller.signal.aborted && record.listeners.size) throw consumerError("data_product_stream_closed", 503);
|
|
} catch (error) {
|
|
if (controller.signal.aborted || closed || !record.listeners.size) break;
|
|
if (error?.message === "resync_required") {
|
|
await bootstrap(record).catch(() => undefined);
|
|
} else {
|
|
const timestamp = new Date(now()).toISOString();
|
|
record.state.runtimeState = "reconnecting";
|
|
record.state.metrics.reconnects += 1;
|
|
record.state.lastError = { code: safeErrorCode(error), at: timestamp };
|
|
record.state.timestamps.updatedAt = timestamp;
|
|
await writeState(record);
|
|
}
|
|
await waitForAbortableDelay(delay, controller.signal);
|
|
delay = Math.min(reconnectMaxMs, delay * 2);
|
|
}
|
|
}
|
|
})().finally(async () => {
|
|
if (record.stream?.controller === controller) record.stream = null;
|
|
if (record.state?.enabled && !record.listeners.size) {
|
|
record.state.runtimeState = "idle";
|
|
record.state.timestamps.updatedAt = new Date(now()).toISOString();
|
|
await writeState(record).catch(() => undefined);
|
|
}
|
|
});
|
|
}
|
|
|
|
function stopStream(record) {
|
|
if (record.idleTimer) clearTimeout(record.idleTimer);
|
|
record.idleTimer = null;
|
|
void record.stream?.reader?.cancel().catch(() => undefined);
|
|
record.stream?.controller.abort();
|
|
}
|
|
|
|
async function stopStreamAndWait(record) {
|
|
const pending = record.stream?.promise;
|
|
stopStream(record);
|
|
if (!pending) return;
|
|
await Promise.race([
|
|
pending.catch(() => undefined),
|
|
new Promise((resolve) => setTimeout(resolve, 750)),
|
|
]);
|
|
}
|
|
|
|
async function snapshot(input) {
|
|
const target = input?.binding ? input : await resolveTarget(input.applicationId, input.pageId, input.bindingId);
|
|
const record = await ensureProvisioned(target);
|
|
// A removed target grant revokes browser reads immediately even though the
|
|
// last safe snapshot remains persisted for rollback/diagnostics.
|
|
await readReaderToken(
|
|
record.state.target.applicationId,
|
|
record.state.target.pageId,
|
|
record.state.target.bindingId,
|
|
activeReaderGrantGeneration(record.state),
|
|
);
|
|
await refreshPresentation(record);
|
|
return {
|
|
schemaVersion: "nodedc.data-product.snapshot/v1",
|
|
dataProduct: record.state.product,
|
|
generatedAt: record.state.timestamps.lastSnapshotAt || record.state.timestamps.updatedAt,
|
|
cursor: record.state.cursor,
|
|
facts: Object.values(record.state.subjects).map(decorateFact),
|
|
};
|
|
}
|
|
|
|
async function subscribe(input, listener) {
|
|
const target = input?.binding ? input : await resolveTarget(input.applicationId, input.pageId, input.bindingId);
|
|
const record = await ensureProvisioned(target);
|
|
const after = String(input.after || "");
|
|
if (after && !cursorPattern.test(after)) throw consumerError("stream_cursor_invalid");
|
|
if (after && after !== record.state.cursor) return { resyncRequired: true, cursor: record.state.cursor, release() {} };
|
|
const leaseId = randomUUID();
|
|
record.listeners.set(leaseId, listener);
|
|
listener({
|
|
event: "nodedc.data-product.ready.v1",
|
|
data: {
|
|
schemaVersion: "nodedc.data-product.ready/v1",
|
|
dataProductId: record.state.product.id,
|
|
cursor: record.state.cursor,
|
|
emittedAt: new Date(now()).toISOString(),
|
|
},
|
|
});
|
|
startStream(record);
|
|
let released = false;
|
|
return {
|
|
resyncRequired: false,
|
|
cursor: record.state.cursor,
|
|
release() {
|
|
if (released) return;
|
|
released = true;
|
|
record.listeners.delete(leaseId);
|
|
if (!record.listeners.size && record.stream && !record.idleTimer) {
|
|
record.idleTimer = setTimeout(() => {
|
|
record.idleTimer = null;
|
|
if (!record.listeners.size) stopStream(record);
|
|
}, idleStopMs);
|
|
record.idleTimer.unref?.();
|
|
}
|
|
},
|
|
};
|
|
}
|
|
|
|
async function status(input) {
|
|
const target = await resolveTarget(input.applicationId, input.pageId, input.bindingId);
|
|
const record = await recordFor(target);
|
|
if (!record.state) return { found: false, target: safeTarget(target), readerGrant: "target-scoped-server-only" };
|
|
await refreshPresentation(record);
|
|
return { found: true, consumer: safeStateSummary(record) };
|
|
}
|
|
|
|
async function accept(input) {
|
|
const target = await resolveTarget(input.applicationId, input.pageId, input.bindingId);
|
|
const record = await ensureProvisioned(target);
|
|
const timeoutMs = Math.min(30_000, Math.max(250, Number(input.timeoutMs ?? 5_000)));
|
|
const minSubjectCount = Math.min(5_000, Math.max(0, Number(input.minSubjectCount ?? 1)));
|
|
const minPatchCount = Math.min(1_000, Math.max(0, Number(input.minPatchCount ?? 0)));
|
|
if (![timeoutMs, minSubjectCount, minPatchCount].every(Number.isInteger)) throw consumerError("data_product_consumer_acceptance_input_invalid");
|
|
const baselinePatchCount = record.state.metrics.patchCommits;
|
|
const lease = await subscribe(target, () => undefined);
|
|
const deadline = now() + timeoutMs;
|
|
try {
|
|
while (now() < deadline) {
|
|
const patchDelta = record.state.metrics.patchCommits - baselinePatchCount;
|
|
if (Object.keys(record.state.subjects).length >= minSubjectCount && patchDelta >= minPatchCount) break;
|
|
await new Promise((resolve) => setTimeout(resolve, 50));
|
|
}
|
|
await refreshPresentation(record);
|
|
const subjectKeys = Object.keys(record.state.subjects);
|
|
const duplicateCount = subjectKeys.length - new Set(subjectKeys).size;
|
|
const patchDelta = record.state.metrics.patchCommits - baselinePatchCount;
|
|
const upstreamStreamCount = record.stream ? 1 : 0;
|
|
const checks = {
|
|
bindingResolved: true,
|
|
targetScopedReader: true,
|
|
snapshotCommitted: record.state.metrics.snapshotCommits > 0,
|
|
cursorPersisted: cursorPattern.test(record.state.cursor),
|
|
minimumSubjects: subjectKeys.length >= minSubjectCount,
|
|
minimumNewPatches: patchDelta >= minPatchCount,
|
|
duplicateSubjects: duplicateCount === 0,
|
|
singleUpstreamStream: upstreamStreamCount <= 1,
|
|
secretAndEndpointExcluded: true,
|
|
};
|
|
return {
|
|
schemaVersion: ACCEPTANCE_SCHEMA_VERSION,
|
|
accepted: Object.values(checks).every(Boolean),
|
|
checks,
|
|
observed: { subjectCount: subjectKeys.length, newPatchCount: patchDelta, cursor: record.state.cursor, runtimeState: record.state.runtimeState },
|
|
consumer: safeStateSummary(record),
|
|
};
|
|
} finally {
|
|
lease.release();
|
|
}
|
|
}
|
|
|
|
async function rollback(input) {
|
|
const target = await resolveTarget(input.applicationId, input.pageId, input.bindingId);
|
|
const record = await recordFor(target);
|
|
if (!record.state) throw consumerError("data_product_consumer_not_found", 404);
|
|
stopStream(record);
|
|
record.listeners.clear();
|
|
record.state.enabled = false;
|
|
record.state.runtimeState = "stopped";
|
|
record.state.timestamps.updatedAt = new Date(now()).toISOString();
|
|
record.state.lastError = null;
|
|
await writeState(record);
|
|
return { rolledBack: true, snapshotPreserved: true, consumer: safeStateSummary(record) };
|
|
}
|
|
|
|
async function resumePersisted() {
|
|
await ready;
|
|
for (const entry of await readdir(stateDir, { withFileTypes: true })) {
|
|
if (!entry.isFile() || !/^[0-9a-f]{64}\.json$/.test(entry.name)) continue;
|
|
try {
|
|
const state = JSON.parse(await readFile(join(stateDir, entry.name), "utf8"));
|
|
if (state?.schemaVersion !== STATE_SCHEMA_VERSION || !state.target) continue;
|
|
const target = await resolveTarget(state.target.applicationId, state.target.pageId, state.target.bindingId);
|
|
const record = await recordFor(target);
|
|
record.target = target;
|
|
if (record.state?.enabled) {
|
|
record.state.runtimeState = "idle";
|
|
record.state.timestamps.updatedAt = new Date(now()).toISOString();
|
|
await writeState(record);
|
|
}
|
|
} catch {
|
|
// One damaged or obsolete target cannot stop Foundry from serving the
|
|
// remaining applications. Its file remains available for diagnostics.
|
|
}
|
|
}
|
|
}
|
|
|
|
const staleTimer = setInterval(() => {
|
|
for (const record of records.values()) void refreshPresentation(record).catch(() => undefined);
|
|
}, staleSweepMs);
|
|
staleTimer.unref?.();
|
|
|
|
async function shutdown() {
|
|
if (closed) return;
|
|
closed = true;
|
|
clearInterval(staleTimer);
|
|
const streams = [];
|
|
for (const record of records.values()) {
|
|
stopStream(record);
|
|
record.listeners.clear();
|
|
if (record.stream?.promise) streams.push(record.stream.promise);
|
|
}
|
|
await Promise.race([
|
|
Promise.allSettled(streams),
|
|
new Promise((resolve) => setTimeout(resolve, 750)),
|
|
]);
|
|
}
|
|
|
|
return {
|
|
plan,
|
|
apply,
|
|
snapshot,
|
|
subscribe,
|
|
status,
|
|
accept,
|
|
rollback,
|
|
resumePersisted,
|
|
shutdown,
|
|
};
|
|
}
|
|
|
|
export const foundryDataProductConsumerSchemas = Object.freeze({
|
|
state: STATE_SCHEMA_VERSION,
|
|
plan: PLAN_SCHEMA_VERSION,
|
|
acceptance: ACCEPTANCE_SCHEMA_VERSION,
|
|
presentationPatch: PRESENTATION_PATCH_SCHEMA_VERSION,
|
|
});
|