import { isBoundedGeoJsonGeometry, isBoundedTelemetryReadings } from "@nodedc/external-provider-contract/data-plane"; 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 FIELD_CONTRACT_TYPES = new Set(["string", "number", "boolean", "string_array", "telemetry_readings", "point", "geometry"]); const DEFINITION_KEYS = new Set([ "id", "version", "ontologyRevision", "deliveryMode", "semanticTypes", "fields", "fieldContracts", "history", ]); const HISTORY_KEYS = new Set(["mode", "intervalMs", "retentionDays", "strategy"]); const FIELD_CONTRACT_KEYS = new Set(["type", "required", "enum", "minimum", "maximum"]); 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 fieldContracts = normalizeFieldContracts(value.fieldContracts, fields); const history = normalizeHistoryPolicy(value.history); return Object.freeze({ id, version, ontologyRevision, deliveryMode, semanticTypes, fields, fieldContracts, history }); } export function normalizeFieldContracts(value, fields) { if (value === undefined) return Object.freeze({}); if (!isPlainObject(value)) throw policyError("data_product_field_contracts_invalid"); const names = Object.keys(value); if (names.length === 0) return Object.freeze({}); if (names.length !== fields.length || fields.some((field) => !Object.hasOwn(value, field))) { throw policyError("data_product_field_contracts_must_match_fields"); } const normalized = {}; for (const field of fields) normalized[field] = normalizeFieldContract(value[field]); return Object.freeze(normalized); } function normalizeFieldContract(value) { if (!isPlainObject(value) || !hasOnlyKeys(value, FIELD_CONTRACT_KEYS)) { throw policyError("data_product_field_contract_invalid"); } const type = string(value.type); if (!FIELD_CONTRACT_TYPES.has(type) || typeof value.required !== "boolean") { throw policyError("data_product_field_contract_shape_invalid"); } const contract = { type, required: value.required }; if (value.enum !== undefined) { if (!Array.isArray(value.enum) || value.enum.length === 0 || new Set(value.enum.map(stableLiteral)).size !== value.enum.length || new Set(["point", "geometry", "string_array", "telemetry_readings"]).has(type) || value.enum.some((item) => !fieldContractValueMatchesType(item, type))) { throw policyError("data_product_field_contract_enum_invalid"); } contract.enum = Object.freeze([...value.enum].sort((left, right) => stableLiteral(left).localeCompare(stableLiteral(right)))); } if (value.minimum !== undefined || value.maximum !== undefined) { if (type !== "number" || (value.minimum !== undefined && !Number.isFinite(value.minimum)) || (value.maximum !== undefined && !Number.isFinite(value.maximum)) || (Number.isFinite(value.minimum) && Number.isFinite(value.maximum) && value.minimum > value.maximum)) { throw policyError("data_product_field_contract_range_invalid"); } if (value.minimum !== undefined) contract.minimum = value.minimum; if (value.maximum !== undefined) contract.maximum = value.maximum; if (contract.enum?.some((item) => ( (contract.minimum !== undefined && item < contract.minimum) || (contract.maximum !== undefined && item > contract.maximum) ))) throw policyError("data_product_field_contract_enum_out_of_range"); } return Object.freeze(contract); } 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), fieldContracts: isPlainObject(row.fieldContracts) ? row.fieldContracts : {}, 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)); } function fieldContractValueMatchesType(value, type) { if (type === "string_array") return Array.isArray(value) && value.every((item) => typeof item === "string"); if (type === "telemetry_readings") return isBoundedTelemetryReadings(value); if (type === "point") return isBoundedGeoJsonGeometry(value, new Set(["Point"])); if (type === "geometry") return isBoundedGeoJsonGeometry(value); return typeof value === type && (type !== "number" || Number.isFinite(value)); } function stableLiteral(value) { return `${typeof value}:${JSON.stringify(value)}`; }