feat(platform): add managed data product history plane

2026-07-18 14:38:06 +03:00
parent 02816c4352
commit 3c5d8f6cef
34 changed files with 2091 additions and 163 deletions
@@ -1,9 +1,14 @@
 import { createHash, randomUUID } from "node:crypto";
 import {
+  DATA_PRODUCT_HISTORY_SCHEMA_VERSION,
  DATA_PRODUCT_PATCH_SCHEMA_VERSION,
  DATA_PRODUCT_SNAPSHOT_SCHEMA_VERSION,
 } from "@nodedc/external-provider-contract/data-plane";

+const IDENTIFIER = /^[a-z][a-z0-9._:-]{2,127}$/;
+const MAX_HISTORY_RESOLUTION_MS = 24 * 60 * 60 * 1000;
+const MAX_HISTORY_SOURCE_IDS = 1000;
+
 export async function loadDataProductDefinition(db, dataProductId, { activeOnly = true } = {}) {
  const result = await db.query(
    `select id, version, ontology_revision as "ontologyRevision",
@@ -216,6 +221,121 @@ export async function readDataProductSnapshot(pool, binding, definition, { limit
  }
 }

+export function normalizeHistoryReadOptions(value, definition) {
+  const policy = definition?.historyPolicy || {};
+  if (policy.mode === "none") throw deliveryError("data_product_history_not_supported", 409);
+  const from = new Date(String(value?.from || ""));
+  const to = new Date(String(value?.to || ""));
+  if (Number.isNaN(from.getTime()) || Number.isNaN(to.getTime()) || from >= to) {
+    throw deliveryError("data_product_history_range_invalid", 400);
+  }
+  const retentionDays = Math.max(1, Number(policy.retentionDays) || 1);
+  if (to.getTime() - from.getTime() > retentionDays * 24 * 60 * 60 * 1000) {
+    throw deliveryError("data_product_history_range_exceeds_retention", 400);
+  }
+  const nativeResolutionMs = policy.mode === "sampled" ? Number(policy.intervalMs) : 1000;
+  const resolutionMs = value?.resolutionMs === undefined || value?.resolutionMs === ""
+    ? nativeResolutionMs
+    : Number(value.resolutionMs);
+  if (
+    !Number.isInteger(resolutionMs)
+    || resolutionMs < nativeResolutionMs
+    || resolutionMs > MAX_HISTORY_RESOLUTION_MS
+    || resolutionMs % nativeResolutionMs !== 0
+  ) throw deliveryError("data_product_history_resolution_invalid", 400);
+  const sourceIds = [...new Set((Array.isArray(value?.sourceIds) ? value.sourceIds : [])
+    .map((item) => String(item || "").trim())
+    .filter(Boolean))];
+  if (sourceIds.length > MAX_HISTORY_SOURCE_IDS || sourceIds.some((item) => !IDENTIFIER.test(item))) {
+    throw deliveryError("data_product_history_source_ids_invalid", 400);
+  }
+  const limit = Number(value?.limit ?? 1000);
+  if (!Number.isInteger(limit) || limit < 1 || limit > 5000) {
+    throw deliveryError("data_product_history_limit_invalid", 400);
+  }
+  const query = {
+    from: from.toISOString(),
+    to: to.toISOString(),
+    resolutionMs,
+    sourceIds: sourceIds.sort(),
+    order: "asc",
+  };
+  const queryHash = createHash("sha256").update(JSON.stringify(query)).digest("hex");
+  const cursor = decodeHistoryCursor(value?.cursor, queryHash);
+  return Object.freeze({ ...query, limit, queryHash, cursor });
+}
+
+export async function readDataProductHistory(pool, binding, definition, options) {
+  const query = normalizeHistoryReadOptions(options, definition);
+  const cursor = query.cursor || {};
+  const rows = await pool.query(
+    `with sampled as (
+       select
+         to_timestamp(
+           floor(extract(epoch from bucket_start) * 1000 / $8::double precision)
+           * $8::double precision / 1000
+         ) as query_bucket,
+         source_id, semantic_type, observed_at, received_at, attributes, geometry, fingerprint
+       from external_data_plane_history
+       where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4
+         and bucket_start >= $5 and bucket_start < $6
+         and (cardinality($7::text[]) = 0 or source_id = any($7::text[]))
+     ), ranked as (
+       select *, row_number() over (
+         partition by query_bucket, source_id, semantic_type
+         order by observed_at desc, received_at desc, fingerprint desc
+       ) as sample_rank
+       from sampled
+     )
+     select query_bucket as "bucketStart", 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 ranked
+     where sample_rank = 1
+       and (
+         $9::boolean = false
+         or (query_bucket, source_id, semantic_type) > ($10::timestamptz, $11::text, $12::text)
+       )
+     order by query_bucket asc, source_id asc, semantic_type asc
+     limit $13`,
+    [
+      binding.tenantId, binding.connectionId, binding.providerId, definition.id,
+      query.from, query.to, query.sourceIds, query.resolutionMs,
+      Boolean(query.cursor), cursor.bucketStart || query.from, cursor.sourceId || "", cursor.semanticType || "",
+      query.limit + 1,
+    ],
+  );
+  const page = rows.rows.slice(0, query.limit);
+  const last = page.at(-1);
+  return {
+    schemaVersion: DATA_PRODUCT_HISTORY_SCHEMA_VERSION,
+    dataProduct: { id: definition.id, version: definition.version },
+    generatedAt: new Date().toISOString(),
+    query: {
+      from: query.from,
+      to: query.to,
+      resolutionMs: query.resolutionMs,
+      sourceIds: query.sourceIds,
+      order: query.order,
+    },
+    facts: page.map((fact) => ({
+      ...publicFact(fact),
+      bucketStart: new Date(fact.bucketStart).toISOString(),
+    })),
+    ...(rows.rowCount > query.limit && last ? {
+      nextCursor: encodeHistoryCursor({
+        queryHash: query.queryHash,
+        bucketStart: new Date(last.bucketStart).toISOString(),
+        sourceId: last.sourceId,
+        semanticType: last.semanticType,
+      }),
+    } : {}),
+  };
+}
+
 export async function readPatchEvents(pool, binding, definition, after, { limit = 100 } = {}) {
  const client = await pool.connect();
  try {
@@ -450,6 +570,36 @@ function publicFact(fact) {
  return value;
 }

+function encodeHistoryCursor(value) {
+  return Buffer.from(JSON.stringify(value), "utf8").toString("base64url");
+}
+
+function decodeHistoryCursor(value, queryHash) {
+  const raw = String(value || "").trim();
+  if (!raw) return null;
+  if (raw.length > 1024 || !/^[A-Za-z0-9_-]+$/.test(raw)) {
+    throw deliveryError("data_product_history_cursor_invalid", 400);
+  }
+  let cursor;
+  try {
+    cursor = JSON.parse(Buffer.from(raw, "base64url").toString("utf8"));
+  } catch {
+    throw deliveryError("data_product_history_cursor_invalid", 400);
+  }
+  if (
+    !cursor || typeof cursor !== "object" || Array.isArray(cursor)
+    || cursor.queryHash !== queryHash
+    || Number.isNaN(Date.parse(cursor.bucketStart))
+    || !IDENTIFIER.test(String(cursor.sourceId || ""))
+    || !IDENTIFIER.test(String(cursor.semanticType || ""))
+  ) throw deliveryError("data_product_history_cursor_invalid", 400);
+  return Object.freeze({
+    bucketStart: new Date(cursor.bucketStart).toISOString(),
+    sourceId: cursor.sourceId,
+    semanticType: cursor.semanticType,
+  });
+}
+
 function chunkOperations(values, maxOperations, maxBytes) {
  const result = [];
  let current = [];