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 });
}