import assert from "node:assert/strict";
import { Pool } from "pg";
import { validateDataProductHistory, validateDataProductPatch, validateDataProductSnapshot } from "@nodedc/external-provider-contract/data-plane";
import {
  loadDataProductDefinition,
  persistDataProductDefinition,
  persistDataProductPublish,
  readDataProductHistory,
  readDataProductSnapshot,
  readPatchEvents,
} from "../src/data-product-delivery.mjs";
import { normalizeDataProductDefinition } from "../src/data-product-policy.mjs";
import { migrate } from "../src/schema.mjs";

const databaseUrl = process.env.EXTERNAL_DATA_PLANE_TEST_DATABASE_URL;
if (!databaseUrl) throw new Error("EXTERNAL_DATA_PLANE_TEST_DATABASE_URL_required");
const pool = new Pool({ connectionString: databaseUrl, max: 4 });

try {
  await migrate(pool);
  await pool.query(`truncate table
    external_data_plane_patch_outbox,
    external_data_plane_history,
    external_data_plane_current,
    external_data_plane_facts,
    external_data_plane_raw_envelopes,
    external_data_plane_batches,
    external_data_plane_delivery_state,
    external_data_plane_reader_bindings,
    external_data_plane_writer_bindings,
    external_data_plane_products
    restart identity cascade`);

  const definition = normalizeDataProductDefinition({
    id: "test.positions.current.v1",
    version: "1.0.0",
    ontologyRevision: "ontology.test.positions.v1",
    deliveryMode: "snapshot+patch",
    semanticTypes: ["map.moving_object"],
    fields: ["source_id", "observed_at", "geometry", "status"],
    fieldContracts: {
      source_id: { type: "string", required: true },
      observed_at: { type: "string", required: true },
      geometry: { type: "point", required: true },
      status: { type: "string", required: true },
    },
    history: {
      mode: "sampled",
      intervalMs: 60_000,
      strategy: "latest-per-entity-per-bucket",
      retentionDays: 30,
    },
  });
  await persistDataProductDefinition(pool, definition);
  const storedDefinition = await loadDataProductDefinition(pool, definition.id);
  assert.deepEqual(storedDefinition.fieldContracts, definition.fieldContracts);
  const binding = {
    id: "reader-binding-test",
    bindingKey: "managed-reader-binding-test",
    tenantId: "tenant-test",
    connectionId: "consumer-connection-test",
    sourceConnectionId: "connection-test",
    providerId: "provider-test",
    allowedDataProductIds: [definition.id],
    active: true,
    expiresAt: "2027-01-01T00:00:00.000Z",
  };

  const firstBatch = batch("run-01", "2026-07-15T10:00:05.000Z", [
    fact("unit-01", 37.61, 55.75, "online"),
    fact("unit-02", 37.62, 55.76, "online"),
  ]);
  const first = await persistDataProductPublish(pool, firstBatch, storedDefinition);
  assert.equal(first.idempotent, false);
  assert.equal(first.currentUpdatedCount, 2);
  assert.equal(first.historyInsertedCount, 2);
  assert.equal(first.cursor, "1");

  const duplicate = await persistDataProductPublish(pool, firstBatch, storedDefinition);
  assert.equal(duplicate.idempotent, true);
  assert.equal(duplicate.cursor, "1");

  await assert.rejects(
    persistDataProductPublish(pool, {
      ...firstBatch,
      facts: [fact("unit-01", 99.9, 55.75, "changed")],
    }, storedDefinition),
    (error) => error?.status === 409 && error?.code === "idempotency_key_reused",
  );

  const concurrentBatch = batch("run-concurrent", "2026-07-15T10:00:08.000Z", [
    fact("unit-03", 37.63, 55.77, "online", "2026-07-15T10:00:08.000Z"),
  ]);
  const concurrent = await Promise.all([
    persistDataProductPublish(pool, concurrentBatch, storedDefinition),
    persistDataProductPublish(pool, concurrentBatch, storedDefinition),
  ]);
  assert.deepEqual(concurrent.map((value) => value.idempotent).sort(), [false, true]);

  const second = await persistDataProductPublish(pool, batch(
    "run-02",
    "2026-07-15T10:00:15.000Z",
    [fact("unit-01", 37.611, 55.751, "moving", "2026-07-15T10:00:15.000Z")],
  ), storedDefinition);
  assert.equal(second.currentUpdatedCount, 1);
  assert.equal(second.cursor, "3");

  const latest = await persistDataProductPublish(pool, batch(
    "run-latest",
    "2026-07-15T10:02:05.000Z",
    [fact("unit-01", 37.7, 55.8, "latest", "2026-07-15T10:02:00.000Z")],
  ), storedDefinition);
  assert.equal(latest.currentUpdatedCount, 1);
  const late = await persistDataProductPublish(pool, batch(
    "run-late",
    "2026-07-15T10:03:00.000Z",
    [fact("unit-01", 37.65, 55.78, "late", "2026-07-15T10:01:30.000Z")],
  ), storedDefinition);
  assert.equal(late.currentUpdatedCount, 0);
  assert.equal(late.historyInsertedCount, 1);
  assert.equal(late.patchOperationCount, 0);

  await assert.rejects(
    persistDataProductPublish(pool, batch("run-bad-type", "2026-07-15T10:03:10.000Z", [{
      ...fact("unit-04", 37.6, 55.7, "online"),
      semanticType: "map.forbidden",
    }]), storedDefinition),
    (error) => error?.status === 422 && error?.code === "data_product_semantic_type_forbidden",
  );
  await assert.rejects(
    persistDataProductPublish(pool, batch("run-bad-field", "2026-07-15T10:03:11.000Z", [{
      ...fact("unit-04", 37.6, 55.7, "online"),
      attributes: { undeclared: true },
    }]), storedDefinition),
    (error) => error?.status === 422 && error?.code === "data_product_field_forbidden",
  );
  await assert.rejects(
    persistDataProductPublish(pool, batch("run-duplicate-fact", "2026-07-15T10:03:12.000Z", [
      fact("unit-04", 37.6, 55.7, "online"),
      fact("unit-04", 37.61, 55.71, "moving"),
    ]), storedDefinition),
    (error) => error?.status === 422 && error?.code === "data_product_duplicate_entity_key",
  );

  const snapshot = await readDataProductSnapshot(pool, binding, storedDefinition);
  assert.equal(validateDataProductSnapshot(snapshot).ok, true);
  assert.equal(snapshot.cursor, "4");
  assert.equal(snapshot.facts.length, 3);
  assert.equal(snapshot.facts.find((value) => value.sourceId === "unit-01").attributes.status, "latest");

  const patches = await readPatchEvents(pool, binding, storedDefinition, 0n);
  assert.equal(patches.length, 4);
  assert.equal(patches.every((patch) => validateDataProductPatch(patch).ok), true);
  assert.deepEqual(patches.map((patch) => patch.cursor), ["1", "2", "3", "4"]);

  await assert.rejects(
    readPatchEvents(pool, binding, storedDefinition, 999n),
    (error) => error?.status === 409 && error?.code === "resync_required",
  );

  const history = await pool.query("select source_id, observed_at from external_data_plane_history order by source_id");
  assert.equal(history.rowCount, 5);
  const unitOneHistory = history.rows.filter((row) => row.source_id === "unit-01");
  assert.equal(unitOneHistory.some((row) => new Date(row.observed_at).toISOString() === "2026-07-15T10:01:30.000Z"), true);
  assert.equal(unitOneHistory.some((row) => new Date(row.observed_at).toISOString() === "2026-07-15T10:02:00.000Z"), true);

  const historyPageOne = await readDataProductHistory(pool, binding, storedDefinition, {
    from: "2026-07-15T10:00:00.000Z",
    to: "2026-07-15T10:04:00.000Z",
    resolutionMs: 60_000,
    sourceIds: ["unit-01"],
    limit: 2,
  });
  assert.equal(validateDataProductHistory(historyPageOne).ok, true);
  assert.equal(historyPageOne.facts.length, 2);
  assert.equal(typeof historyPageOne.nextCursor, "string");
  const historyPageTwo = await readDataProductHistory(pool, binding, storedDefinition, {
    from: "2026-07-15T10:00:00.000Z",
    to: "2026-07-15T10:04:00.000Z",
    resolutionMs: 60_000,
    sourceIds: ["unit-01"],
    limit: 2,
    cursor: historyPageOne.nextCursor,
  });
  assert.equal(validateDataProductHistory(historyPageTwo).ok, true);
  assert.equal(historyPageTwo.facts.length, 1);
  assert.equal(historyPageTwo.nextCursor, undefined);
  assert.deepEqual(
    [...historyPageOne.facts, ...historyPageTwo.facts].map((value) => value.bucketStart),
    [
      "2026-07-15T10:00:00.000Z",
      "2026-07-15T10:01:00.000Z",
      "2026-07-15T10:02:00.000Z",
    ],
  );
  await assert.rejects(
    readDataProductHistory(pool, binding, storedDefinition, {
      from: "2026-07-15T10:00:00.000Z",
      to: "2026-07-15T10:04:00.000Z",
      resolutionMs: 60_000,
      sourceIds: ["unit-02"],
      limit: 2,
      cursor: historyPageOne.nextCursor,
    }),
    (error) => error?.status === 400 && error?.code === "data_product_history_cursor_invalid",
  );

  console.log("external-data-plane delivery integration: ok");
} finally {
  await pool.end();
}

function batch(runId, receivedAt, facts) {
  return {
    schemaVersion: "nodedc.external-provider-contract/v1",
    source: { tenantId: "tenant-test", connectionId: "connection-test", providerId: "provider-test" },
    contract: {
      dataProductId: "test.positions.current.v1",
      ontologyRevision: "ontology.test.positions.v1",
      version: "1.0.0",
    },
    batch: { runId, sequence: 0, idempotencyKey: `${runId}.chunk-0`, receivedAt },
    facts,
  };
}

function fact(sourceId, longitude, latitude, status, observedAt = "2026-07-15T10:00:00.000Z") {
  return {
    sourceId,
    semanticType: "map.moving_object",
    observedAt,
    attributes: { status },
    geometry: { type: "Point", coordinates: [longitude, latitude] },
  };
}