import { createHash, randomUUID } from "node:crypto";
import {
  DATA_PRODUCT_HISTORY_SCHEMA_VERSION,
  DATA_PRODUCT_PATCH_SCHEMA_VERSION,
  DATA_PRODUCT_SNAPSHOT_SCHEMA_VERSION,
  isBoundedGeoJsonGeometry,
} 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",
       delivery_mode as "deliveryMode", semantic_types as "semanticTypes",
       fields, field_contracts as "fieldContracts", 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, field_contracts, history_policy
     ) values ($1, $2, $3, $4, $5::jsonb, $6::jsonb, $7::jsonb, $8::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.field_contracts = excluded.field_contracts
       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, field_contracts as "fieldContracts", 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.fieldContracts || {}),
      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",
           current_removed_count as "currentRemovedCount",
           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 };
    }

    if (batch.batch.mode === "replace") await lockReplacementGeneration(client, batch);

    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_GeomFromGeoJSON(item.geometry), 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 ST_AsGeoJSON(geometry::geometry)::jsonb end as geometry,
        fingerprint`,
      [
        batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
        batch.contract.dataProductId, JSON.stringify(records),
      ],
    );

    const removed = batch.batch.mode === "replace"
      ? await removeMissingCurrent(client, batch, records)
      : [];
    const historyInsertedCount = await persistHistory(client, batch, definition.historyPolicy, records);
    const operations = [
      ...updated.rows.map((fact) => ({ op: "upsert", fact: publicFact(fact) })),
      ...removed.map((fact) => ({ op: "remove", sourceId: fact.sourceId, semanticType: fact.semanticType })),
    ];
    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);
    if (batch.batch.mode === "replace") await persistReplacementGeneration(client, batch);

    await client.query(
      `update external_data_plane_batches set
         current_updated_count = $2,
         current_removed_count = $3,
         history_inserted_count = $4,
         patch_operation_count = $5,
         delivery_cursor = $6
       where id = $1`,
      [batchId, updated.rowCount, removed.length, historyInsertedCount, operations.length, cursor],
    );
    await client.query("commit");
    return normalizeReceipt({
      batchId,
      idempotent: false,
      publishedFactCount: batch.facts.length,
      currentUpdatedCount: updated.rowCount,
      currentRemovedCount: removed.length,
      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 sourceConnectionId = readerSourceConnectionId(binding);
  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 ST_AsGeoJSON(geometry::geometry)::jsonb 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, sourceConnectionId, 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 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 sourceConnectionId = readerSourceConnectionId(binding);
  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 ST_AsGeoJSON(geometry::geometry)::jsonb 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, sourceConnectionId, 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 sourceConnectionId = readerSourceConnectionId(binding);
  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, sourceConnectionId, 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 lockReplacementGeneration(client, batch) {
  const scope = [batch.source.tenantId, batch.source.connectionId, batch.source.providerId, batch.contract.dataProductId];
  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, 0)
     on conflict (tenant_id, connection_id, provider_id, data_product_id) do nothing`,
    scope,
  );
  const state = await client.query(
    `select current_generation_at as "currentGenerationAt"
     from external_data_plane_delivery_state
     where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4
     for update`,
    scope,
  );
  const generationAt = new Date(batch.batch.generationAt);
  const currentGenerationAt = state.rows[0]?.currentGenerationAt
    ? new Date(state.rows[0].currentGenerationAt)
    : null;
  if (currentGenerationAt && generationAt <= currentGenerationAt) {
    throw deliveryError("data_product_generation_not_newer", 409);
  }
}

async function removeMissingCurrent(client, batch, records) {
  const result = await client.query(
    `delete from external_data_plane_current as target
     where target.tenant_id = $1 and target.connection_id = $2 and target.provider_id = $3
       and target.data_product_id = $4
       and not exists (
         select 1
         from jsonb_to_recordset($5::jsonb) as incoming(source_id text, semantic_type text)
         where incoming.source_id = target.source_id and incoming.semantic_type = target.semantic_type
       )
     returning source_id as "sourceId", semantic_type as "semanticType"`,
    [
      batch.source.tenantId,
      batch.source.connectionId,
      batch.source.providerId,
      batch.contract.dataProductId,
      JSON.stringify(records),
    ],
  );
  return result.rows;
}

async function persistReplacementGeneration(client, batch) {
  await client.query(
    `update external_data_plane_delivery_state
     set current_generation_at = $5, current_generation_id = $6, updated_at = now()
     where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4`,
    [
      batch.source.tenantId,
      batch.source.connectionId,
      batch.source.providerId,
      batch.contract.dataProductId,
      batch.batch.generationAt,
      batch.batch.runId,
    ],
  );
}

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_GeomFromGeoJSON(item.geometry), 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 connectionId = scope.source
    ? scope.source.connectionId
    : readerSourceConnectionId(scope);
  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, connectionId,
      scope.providerId || scope.source.providerId, dataProductId],
  );
  return result.rowCount ? BigInt(result.rows[0].current_cursor) : 0n;
}

async function patchFloor(db, binding, dataProductId) {
  const sourceConnectionId = readerSourceConnectionId(binding);
  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, sourceConnectionId, 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 readerSourceConnectionId(binding) {
  const sourceConnectionId = String(binding?.sourceConnectionId || "").trim();
  if (sourceConnectionId) return sourceConnectionId;
  if (!binding?.bindingKey) {
    const legacyConnectionId = String(binding?.connectionId || "").trim();
    if (legacyConnectionId) return legacyConnectionId;
  }
  throw deliveryError("reader_source_scope_unresolved", 409);
}

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),
    currentRemovedCount: Number(value.currentRemovedCount || 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 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 = [];
  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,
      mode: batch.batch.mode || "upsert",
      ...(batch.batch.generationAt ? { generationAt: batch.batch.generationAt } : {}),
    },
    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 fieldContracts = definition?.fieldContracts && typeof definition.fieldContracts === "object"
    && !Array.isArray(definition.fieldContracts) ? definition.fieldContracts : {};
  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);
    }
    for (const [field, contract] of Object.entries(fieldContracts)) {
      const resolved = resolveContractField(fact, field);
      if (!resolved.present) {
        if (contract.required === true) throw deliveryError("data_product_field_required", 422);
        continue;
      }
      assertFieldContractValue(resolved.value, contract);
    }
  }
}

function resolveContractField(fact, field) {
  if (field === "geometry") return { present: fact.geometry !== undefined, value: fact.geometry };
  if (field === "source_id") return { present: fact.sourceId !== undefined, value: fact.sourceId };
  if (field === "semantic_type") return { present: fact.semanticType !== undefined, value: fact.semanticType };
  if (field === "observed_at") return { present: fact.observedAt !== undefined, value: fact.observedAt };
  const attribute = field.startsWith("attributes.") ? field.slice("attributes.".length) : field;
  const attributes = fact.attributes || {};
  return { present: Object.hasOwn(attributes, attribute), value: attributes[attribute] };
}

function assertFieldContractValue(value, contract) {
  if (!fieldContractValueMatchesType(value, contract.type)) {
    throw deliveryError("data_product_field_type_invalid", 422);
  }
  if (Array.isArray(contract.enum) && !contract.enum.some((allowed) => Object.is(allowed, value))) {
    throw deliveryError("data_product_field_value_forbidden", 422);
  }
  if (typeof value === "number" && (
    (Number.isFinite(contract.minimum) && value < contract.minimum)
    || (Number.isFinite(contract.maximum) && value > contract.maximum)
  )) throw deliveryError("data_product_field_value_out_of_range", 422);
}

function fieldContractValueMatchesType(value, type) {
  if (type === "string_array") return Array.isArray(value) && value.every((item) => typeof item === "string");
  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 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 });
}