import assert from "node:assert/strict";
import { randomUUID } from "node:crypto";
import { test } from "bun:test";
import pg from "pg";

import { buildWorkbenchProjectionEventFacts } from "../cloud/workbench-projection-writer.ts";
import { PostgresCloudRuntimeStore } from "./runtime-store-postgres.ts";
import { commitAgentRunKafkaProjection as commitProjection } from "./runtime-store-postgres-kafka.ts";

const dbUrl = String(process.env.HWLAB_KAFKA_PROJECTOR_INTEGRATION_DB_URL ?? "").trim();
const enabled = Boolean(dbUrl) && process.env.HWLAB_KAFKA_PROJECTOR_INTEGRATION_CONFIRM_NON_PRODUCTION === "1";

if (!enabled) {
  test.skip("Kafka projector real PostgreSQL transaction and fault-injection contract", () => {});
} else {
  test("Kafka projector real PostgreSQL transaction and fault-injection contract", async () => {
    const { Pool } = pg;
    const pool = new Pool({ connectionString: dbUrl, ssl: false, max: 4 });
    const suffix = randomUUID().replaceAll("-", "");
    const traceId = `trc_it_${suffix}`;
    const sessionId = `ses_it_${suffix}`;
    const ownerUserId = `usr_it_${suffix}`;
    const now = "2026-07-10T12:00:00.000Z";
    const store = new PostgresCloudRuntimeStore({
      dbUrl,
      sslMode: "disable",
      env: { ...process.env, HWLAB_CLOUD_DB_SSL_MODE: "disable" },
      queryClient: pool,
      now: () => now,
      logger: { error() {}, warn() {}, info() {} }
    });

    try {
      const readiness = await store.readiness();
      assert.equal(readiness.ready, true);
      assert.equal(readiness.durable, true);
      assert.equal(readiness.migration?.ready, true);

      await pool.query(
        "INSERT INTO users (id, username, display_name, role, status, created_at, updated_at) VALUES ($1,$2,$2,'user','active',$3,$3)",
        [ownerUserId, `integration-${suffix}`, now]
      );

      await store.writeWorkbenchSessionAdmissionFact({
        fact: {
          sessionId,
          ownerUserId,
          projectId: `prj_it_${suffix}`,
          conversationId: `cnv_it_${suffix}`,
          threadId: `thread_it_${suffix}`,
          status: "queued",
          lastTraceId: traceId,
          projectedSeq: 0,
          sourceSeq: 0,
          terminal: false,
          sealed: false,
          sessionJson: { sessionId, launchContext: { source: "integration-test" }, valuesRedacted: true },
          createdAt: now,
          updatedAt: now
        }
      });

      const running = await project(store, {
        traceId,
        sessionId,
        sourceEventId: `evt_running_${suffix}`,
        sourceSeq: 1,
        sourceOffset: "100",
        inputSha256: "1".repeat(64),
        event: { eventType: "assistant_message", status: "running", assistantText: "partial", createdAt: now }
      });
      assert.equal(running.duplicate, false);
      assert.equal(running.projectedSeq, 1);
      assert.equal(await count(pool, "workbench_kafka_inbox", "trace_id = $1", [traceId]), 1);
      assert.equal(await count(pool, "workbench_trace_events", "trace_id = $1", [traceId]), 1);
      assert.equal(await count(pool, "workbench_projection_outbox", "trace_id = $1", [traceId]), 2);
      assert.equal(await count(pool, "hwlab_kafka_outbox", "partition_key = $1", [traceId]), 1);

      const sameTransport = await project(store, {
        traceId,
        sessionId,
        sourceEventId: `evt_running_${suffix}`,
        sourceSeq: 1,
        sourceOffset: "100",
        inputSha256: "1".repeat(64),
        event: { eventType: "assistant_message", status: "running", assistantText: "partial", createdAt: now }
      });
      const newOffsetSameEvent = await project(store, {
        traceId,
        sessionId,
        sourceEventId: `evt_running_${suffix}`,
        sourceSeq: 1,
        sourceOffset: "101",
        inputSha256: "1".repeat(64),
        event: { eventType: "assistant_message", status: "running", assistantText: "partial", createdAt: now }
      });
      assert.equal(sameTransport.duplicate, true);
      assert.equal(newOffsetSameEvent.duplicate, true);
      assert.equal(await count(pool, "workbench_kafka_inbox", "trace_id = $1", [traceId]), 1);
      assert.equal(await count(pool, "workbench_projection_outbox", "trace_id = $1", [traceId]), 2);

      const conflict = await project(store, {
        traceId,
        sessionId,
        sourceEventId: `evt_running_${suffix}`,
        sourceSeq: 1,
        sourceOffset: "102",
        inputSha256: "2".repeat(64),
        event: { eventType: "assistant_message", status: "running", assistantText: "mutated", createdAt: now }
      });
      assert.equal(conflict.conflict, true);
      assert.equal(await count(pool, "workbench_kafka_dlq", "source_event_id = $1", [`evt_running_${suffix}`]), 1);
      const projectedInbox = await pool.query(
        "SELECT status, projected_seq FROM workbench_kafka_inbox WHERE source_topic = $1 AND source_event_id = $2",
        ["agentrun.event.v1", `evt_running_${suffix}`]
      );
      assert.deepEqual(projectedInbox.rows[0], { status: "projected", projected_seq: 1 });

      await pool.query(`
        CREATE OR REPLACE FUNCTION issue_2464_delay_inbox_insert() RETURNS trigger AS $$
        BEGIN
          IF NEW.source_topic = 'agentrun.event.v1' AND NEW.source_partition = 0 AND NEW.source_offset = 300 THEN
            PERFORM pg_sleep(0.25);
          END IF;
          RETURN NEW;
        END;
        $$ LANGUAGE plpgsql;
        DROP TRIGGER IF EXISTS issue_2464_delay_inbox_insert ON workbench_kafka_inbox;
        CREATE TRIGGER issue_2464_delay_inbox_insert BEFORE INSERT ON workbench_kafka_inbox
          FOR EACH ROW EXECUTE FUNCTION issue_2464_delay_inbox_insert();
      `);
      try {
        const concurrent = await Promise.allSettled([
          project(store, {
            traceId: `trc_transport_a_${suffix}`,
            sessionId: `ses_transport_a_${suffix}`,
            sourceEventId: `evt_transport_a_${suffix}`,
            sourceSeq: 1,
            sourceOffset: "300",
            inputSha256: "6".repeat(64),
            event: { eventType: "backend_status", status: "running", createdAt: now }
          }),
          project(store, {
            traceId: `trc_transport_b_${suffix}`,
            sessionId: `ses_transport_b_${suffix}`,
            sourceEventId: `evt_transport_b_${suffix}`,
            sourceSeq: 1,
            sourceOffset: "300",
            inputSha256: "7".repeat(64),
            event: { eventType: "backend_status", status: "running", createdAt: now }
          })
        ]);
        assert.deepEqual(concurrent.map((result) => result.status), ["fulfilled", "fulfilled"]);
        const values = concurrent.map((result) => {
          if (result.status !== "fulfilled") throw result.reason;
          return result.value;
        });
        assert.equal(values.filter((result) => result.conflict === true).length, 1);
        assert.equal(values.filter((result) => result.conflict !== true && result.duplicate !== true).length, 1);
        assert.equal(await count(pool, "workbench_kafka_inbox", "source_topic = $1 AND source_partition = 0 AND source_offset = 300", ["agentrun.event.v1"]), 1);
        assert.equal(await count(pool, "workbench_kafka_dlq", "source_topic = $1 AND source_partition = 0 AND source_offset = 300", ["agentrun.event.v1"]), 1);
      } finally {
        await pool.query("DROP TRIGGER IF EXISTS issue_2464_delay_inbox_insert ON workbench_kafka_inbox; DROP FUNCTION IF EXISTS issue_2464_delay_inbox_insert()");
      }

      const faultTraceId = `trc_fault_${suffix}`;
      const faultSessionId = `ses_fault_${suffix}`;
      const faultStore = new PostgresCloudRuntimeStore({
        dbUrl,
        sslMode: "disable",
        env: { ...process.env, HWLAB_CLOUD_DB_SSL_MODE: "disable" },
        queryClient: faultInjectingPool(pool, /^INSERT INTO workbench_projection_outbox/u),
        now: () => now,
        logger: { error() {}, warn() {}, info() {} }
      });
      await assert.rejects(project(faultStore, {
        traceId: faultTraceId,
        sessionId: faultSessionId,
        sourceEventId: `evt_fault_${suffix}`,
        sourceSeq: 1,
        sourceOffset: "200",
        inputSha256: "3".repeat(64),
        event: { eventType: "assistant_message", status: "running", assistantText: "must roll back", createdAt: now }
      }), /injected projection outbox failure/u);
      assert.equal(await count(pool, "workbench_kafka_inbox", "trace_id = $1", [faultTraceId]), 0);
      assert.equal(await count(pool, "workbench_trace_events", "trace_id = $1", [faultTraceId]), 0);
      assert.equal(await count(pool, "workbench_projection_outbox", "trace_id = $1", [faultTraceId]), 0);
      assert.equal(await count(pool, "hwlab_kafka_outbox", "partition_key = $1", [faultTraceId]), 0);

      const terminal = await project(store, {
        traceId,
        sessionId,
        sourceEventId: `evt_terminal_${suffix}`,
        sourceSeq: 2,
        sourceOffset: "103",
        inputSha256: "4".repeat(64),
        event: {
          eventType: "terminal_status",
          status: "completed",
          terminal: true,
          assistantText: "authoritative final",
          finalResponse: { text: "authoritative final", status: "completed" },
          startedAt: now,
          finishedAt: "2026-07-10T12:00:01.000Z",
          durationMs: 1000,
          createdAt: "2026-07-10T12:00:01.000Z"
        }
      });
      assert.equal(terminal.projectedSeq, 2);
      const sealed = await pool.query(
        "SELECT projected_seq, terminal, sealed, checkpoint_json FROM workbench_projection_checkpoints WHERE trace_id = $1",
        [traceId]
      );
      assert.equal(sealed.rows[0].projected_seq, 2);
      assert.equal(sealed.rows[0].terminal, true);
      assert.equal(sealed.rows[0].sealed, true);
      assert.equal(JSON.parse(sealed.rows[0].checkpoint_json).finalResponse.text, "authoritative final");
      assert.equal(await count(pool, "workbench_projection_outbox", "trace_id = $1", [traceId]), 5);

      const late = await project(store, {
        traceId,
        sessionId,
        sourceEventId: `evt_late_${suffix}`,
        sourceSeq: 3,
        sourceOffset: "104",
        inputSha256: "5".repeat(64),
        event: { eventType: "backend_status", status: "running", terminal: false, createdAt: "2026-07-10T12:00:02.000Z" }
      });
      assert.equal(late.suppressedAfterSeal, true);
      assert.equal(late.projectedSeq, 2);
      assert.equal(await count(pool, "workbench_trace_events", "trace_id = $1", [traceId]), 2);
      assert.equal(await count(pool, "workbench_projection_outbox", "trace_id = $1", [traceId]), 5);
      const lateInbox = await pool.query(
        "SELECT status, projected_seq FROM workbench_kafka_inbox WHERE source_topic = $1 AND source_event_id = $2",
        ["agentrun.event.v1", `evt_late_${suffix}`]
      );
      assert.deepEqual(lateInbox.rows[0], { status: "projected", projected_seq: 2 });

      const firstSync = await store.readAtomicWorkbenchProjectionSync({ traceId, afterOutboxSeq: 0, limit: 100 });
      assert.equal(firstSync.events.length, 5);
      assert.equal(firstSync.hasMore, false);
      assert.equal(firstSync.cursorOutboxSeq, firstSync.cutoffOutboxSeq);
      assert.equal(new Set(firstSync.events.map((event) => event.outboxSeq)).size, firstSync.events.length);
      assert.equal(firstSync.facts.checkpoints[0].projectedSeq, 2);
      assert.equal(firstSync.facts.turns[0].finalResponse.text, "authoritative final");
      const caughtUp = await store.readAtomicWorkbenchProjectionSync({ traceId, afterOutboxSeq: firstSync.cursorOutboxSeq, limit: 100, deltaOnly: true });
      assert.equal(caughtUp.events.length, 0);
      assert.equal(caughtUp.cursorOutboxSeq, firstSync.cursorOutboxSeq);
    } finally {
      await pool.end();
    }
  }, 30_000);
}

async function project(store, { traceId, sessionId, sourceEventId, sourceSeq, sourceOffset, inputSha256, event }) {
  const canonicalEvent = {
    schema: "agentrun.event.v1",
    eventId: sourceEventId,
    sourceSeq,
    traceId,
    hwlabSessionId: sessionId,
    runId: `run_${traceId}`,
    commandId: `cmd_${traceId}`,
    event
  };
  const projectedEvent = { ...event, traceId, sessionId, sourceEventId, sourceSeq, runId: canonicalEvent.runId, commandId: canonicalEvent.commandId };
  return commitProjection(store, {
    transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 0, sourceOffset, sourceKey: canonicalEvent.runId, inputSha256 },
    canonicalEvent,
    projectedEvent,
    requestMeta: { traceId, sessionId, valuesPrinted: false },
    factsFactory: ({ projectedSeq, previousCheckpoint, projectedAt }) => buildWorkbenchProjectionEventFacts({ projectedSeq, previousCheckpoint, projectedAt, event: projectedEvent }),
    hwlabEvent: { schema: "hwlab.event.v1", eventId: `hwlab:${sourceEventId}`, traceId, sessionId, sourceEventId, sourceSeq, valuesPrinted: false },
    hwlabTopic: "hwlab.event.v1",
    partitionKey: traceId,
    headers: { sourceEventId }
  });
}

function faultInjectingPool(pool, pattern) {
  return {
    async connect() {
      const client = await pool.connect();
      let injected = false;
      return {
        async query(sql, params) {
          if (!injected && pattern.test(String(sql))) {
            injected = true;
            const error = new Error("injected projection outbox failure");
            error.code = "XX2464";
            throw error;
          }
          return client.query(sql, params);
        },
        release() {
          client.release();
        }
      };
    }
  };
}

async function count(pool, table, where, params) {
  if (!/^[a-z_]+$/u.test(table)) throw new Error("unsafe test table");
  const result = await pool.query(`SELECT COUNT(*)::int AS count FROM ${table} WHERE ${where}`, params);
  return Number(result.rows[0].count);
}