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

import { buildWorkbenchProjectionEventFacts } from "../cloud/workbench-projection-writer.ts";
import { projectionOutboxRealtimeEvents } from "../cloud/server-workbench-realtime-http.ts";
import {
  claimHwlabKafkaOutbox,
  commitAgentRunKafkaProjection,
  completeHwlabKafkaOutbox,
  projectionOutboxEventId,
  recordFailedAgentRunKafkaMessage,
  recordIgnoredAgentRunKafkaMessage,
  retryHwlabKafkaOutbox
} from "./runtime-store-postgres-kafka.ts";

test("malformed duplicate source event at a new offset is durably DLQed without mutating projected inbox", async () => {
  const queries = [];
  const client = {
    async query(sql, params) {
      queries.push({ sql, params });
      if (sql.startsWith("SELECT pg_advisory_xact_lock")) return { rows: [] };
      if (sql.startsWith("SELECT source_topic")) return { rows: [{ source_topic: "agentrun.event.v1", source_partition: 0, source_offset: "10", input_sha256: "prior-hash", status: "projected", projected_seq: 7 }] };
      if (sql.startsWith("INSERT INTO workbench_kafka_dlq")) return { rows: [] };
      throw new Error(`unexpected SQL: ${sql}`);
    }
  };
  const store = transactionStore(client);
  const result = await recordFailedAgentRunKafkaMessage(store, {
    transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 0, sourceOffset: "11", sourceKey: "run-1", inputSha256: "new-hash" },
    sourceEventId: "evt-duplicate",
    envelope: { eventId: "evt-duplicate", valuesPrinted: false },
    error: { code: "workbench_kafka_schema_invalid", message: "malformed" }
  });

  assert.equal(result.recorded, true);
  assert.equal(result.conflict, true);
  assert.equal(result.priorStatus, "projected");
  assert.equal(queries[0].params[0], "workbench-kafka-inbox:agentrun.event.v1:transport:0:11");
  assert.equal(queries[1].params[0], "workbench-kafka-inbox:agentrun.event.v1:event:evt-duplicate");
  assert.equal(queries.some(({ sql }) => sql.startsWith("UPDATE workbench_kafka_inbox")), false);
  assert.equal(queries.some(({ sql }) => sql.startsWith("INSERT INTO workbench_kafka_dlq")), true);
});

test("ignored Kafka messages acquire transport then event identity locks", async () => {
  const queries = [];
  const client = {
    async query(sql, params) {
      queries.push({ sql, params });
      if (sql.startsWith("SELECT pg_advisory_xact_lock")) return { rows: [] };
      if (sql.startsWith("SELECT input_sha256")) return { rows: [] };
      if (sql.startsWith("INSERT INTO workbench_kafka_inbox")) return { rows: [] };
      throw new Error(`unexpected SQL: ${sql}`);
    }
  };
  const result = await recordIgnoredAgentRunKafkaMessage(transactionStore(client), {
    transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 2, sourceOffset: "31", sourceKey: "run-ignored", inputSha256: "hash-ignored" },
    envelope: { eventId: "evt-ignored", sourceSeq: 1, runId: "run-ignored" },
    reason: "hwlab-correlation-absent"
  });

  assert.equal(result.ignored, true);
  assert.equal(result.duplicate, false);
  assert.equal(queries[0].params[0], "workbench-kafka-inbox:agentrun.event.v1:transport:2:31");
  assert.equal(queries[1].params[0], "workbench-kafka-inbox:agentrun.event.v1:event:evt-ignored");
});

test("projection outbox derives a stable non-null identity when a fact has no source event", () => {
  const input = {
    fact: { traceId: "trc-derived", projectedSeq: 3, sourceSeq: 2 },
    event: { aggregateId: "trc-derived", aggregateSeq: 3, projectionRevision: 3 },
    family: "traceEvents",
    entityId: "wte-derived",
    commitType: "event"
  };
  const first = projectionOutboxEventId(input);
  assert.match(first, /^projection:[a-f0-9]{64}$/u);
  assert.equal(projectionOutboxEventId(input), first);
  assert.notEqual(projectionOutboxEventId({ ...input, event: { ...input.event, projectionRevision: 4 } }), first);
});

test("Kafka projection preserves admission owner and scope fields", async () => {
  let persistedSession = null;
  const client = {
    async query(sql) {
      if (sql.startsWith("SELECT pg_advisory_xact_lock")) return { rows: [] };
      if (sql.startsWith("SELECT source_topic")) return { rows: [] };
      if (sql.startsWith("INSERT INTO workbench_kafka_inbox")) return { rows: [] };
      if (sql.startsWith("SELECT checkpoint_json")) return { rows: [] };
      if (sql.startsWith("SELECT owner_user_id")) return { rows: [{ owner_user_id: "usr-owner", project_id: "prj-owner", conversation_id: "cnv-owner", thread_id: "thread-owner", session_json: { sessionId: "ses-owner", ownerUserId: "usr-owner", ownerRole: "user", launchContext: { source: "workbench" } } }] };
      if (sql.startsWith("SELECT GREATEST")) return { rows: [{ max_projected_seq: 0 }] };
      if (sql.startsWith("INSERT INTO hwlab_kafka_outbox")) return { rows: [] };
      if (sql.startsWith("UPDATE workbench_kafka_inbox")) return { rows: [] };
      throw new Error(`unexpected SQL: ${sql}`);
    }
  };
  const store = {
    ...transactionStore(client),
    memory: { writeWorkbenchFacts() {} },
    async persistWorkbenchAggregateEvent(event) { return { ...event, eventSeq: 1, aggregateSeq: 1, projectionRevision: 1 }; },
    async persistWorkbenchSessionInputFact() {},
    async persistWorkbenchSessionFact(fact) { persistedSession = fact; },
    async persistWorkbenchMessageFact() {},
    async persistWorkbenchPartFact() {},
    async persistWorkbenchTurnFact() {},
    async persistWorkbenchTraceEventFact() {},
    async persistWorkbenchProjectionCheckpoint() {},
    async persistWorkbenchProjectionOutbox() {}
  };
  await commitAgentRunKafkaProjection(store, {
    transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 0, sourceOffset: "12", sourceKey: "run-owner", inputSha256: "hash-owner" },
    canonicalEvent: { eventId: "evt-owner", sourceSeq: 1, traceId: "trc-owner", hwlabSessionId: "ses-owner", runId: "run-owner", commandId: "cmd-owner" },
    projectedEvent: { traceId: "trc-owner", sessionId: "ses-owner" },
    factsFactory: ({ projectedSeq, projectedAt }) => ({
      written: true,
      facts: {
        sessions: [{ sessionId: "ses-owner", status: "running", lastTraceId: "trc-owner", projectedSeq, sourceSeq: 1, sourceEventId: "evt-owner", terminal: false, sealed: false, updatedAt: projectedAt }],
        messages: [], parts: [], turns: [], traceEvents: [], checkpoints: []
      }
    }),
    hwlabEvent: { eventId: "hwlab:evt-owner" },
    hwlabTopic: "hwlab.event.v1",
    partitionKey: "trc-owner",
    headers: {}
  });

  assert.equal(persistedSession.ownerUserId, "usr-owner");
  assert.equal(persistedSession.ownerRole, "user");
  assert.equal(persistedSession.projectId, "prj-owner");
  assert.equal(persistedSession.conversationId, "cnv-owner");
  assert.equal(persistedSession.threadId, "thread-owner");
  assert.deepEqual(persistedSession.launchContext, { source: "workbench" });
});

test("terminal Kafka transaction persists sealed turn and immutable SSE snapshot", async () => {
  let persistedTurn = null;
  const outbox = [];
  const client = {
    async query(sql) {
      if (sql.startsWith("SELECT pg_advisory_xact_lock")) return { rows: [] };
      if (sql.startsWith("SELECT source_topic")) return { rows: [] };
      if (sql.startsWith("INSERT INTO workbench_kafka_inbox")) return { rows: [] };
      if (sql.startsWith("SELECT checkpoint_json")) return { rows: [{ checkpoint_json: { traceId: "trc_terminal", sessionId: "ses_terminal", assistantText: "final body", finalResponse: { text: "final body", status: "running" }, projectedSeq: 2, sourceSeq: 2, terminal: false, sealed: false } }] };
      if (sql.startsWith("SELECT owner_user_id")) return { rows: [{ owner_user_id: "usr-owner", session_json: { sessionId: "ses_terminal", ownerUserId: "usr-owner" } }] };
      if (sql.startsWith("SELECT GREATEST")) return { rows: [{ max_projected_seq: 2 }] };
      if (sql.startsWith("INSERT INTO hwlab_kafka_outbox") || sql.startsWith("UPDATE workbench_kafka_inbox")) return { rows: [] };
      throw new Error(`unexpected SQL: ${sql}`);
    }
  };
  const store = {
    ...transactionStore(client),
    memory: { writeWorkbenchFacts() {} },
    async persistWorkbenchAggregateEvent(event) { return { ...event, eventSeq: 3, aggregateSeq: 3, projectionRevision: 3 }; },
    async persistWorkbenchSessionInputFact() {}, async persistWorkbenchSessionFact() {}, async persistWorkbenchMessageFact() {}, async persistWorkbenchPartFact() {},
    async persistWorkbenchTurnFact(fact) { persistedTurn = fact; }, async persistWorkbenchTraceEventFact() {}, async persistWorkbenchProjectionCheckpoint() {},
    async persistWorkbenchProjectionOutbox(record) { outbox.push(record); }
  };
  await commitAgentRunKafkaProjection(store, {
    transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 0, sourceOffset: "20", inputSha256: "hash-terminal" },
    canonicalEvent: { eventId: "evt-terminal", sourceSeq: 3, traceId: "trc_terminal", hwlabSessionId: "ses_terminal", runId: "run-terminal", commandId: "cmd-terminal" },
    projectedEvent: { traceId: "trc_terminal", sessionId: "ses_terminal", sourceEventId: "evt-terminal", sourceSeq: 3, type: "result", eventType: "terminal", status: "completed", terminal: true, createdAt: "2026-07-10T10:00:03.000Z" },
    factsFactory: ({ projectedSeq, previousCheckpoint, projectedAt }) => buildWorkbenchProjectionEventFacts({ projectedSeq, previousCheckpoint, projectedAt, event: { traceId: "trc_terminal", sessionId: "ses_terminal", sourceEventId: "evt-terminal", sourceSeq: 3, type: "result", eventType: "terminal", status: "completed", terminal: true, createdAt: "2026-07-10T10:00:03.000Z" } }),
    hwlabEvent: { eventId: "hwlab:evt-terminal" }, hwlabTopic: "hwlab.event.v1", partitionKey: "trc_terminal", headers: {}
  });

  assert.equal(persistedTurn.terminal, true);
  assert.equal(persistedTurn.sealed, true);
  assert.equal(persistedTurn.finalResponse.text, "final body");
  const turnRow = outbox.find((row) => row.entityFamily === "turns");
  assert.equal(turnRow.payload.fact.assistantText, "final body");
  const [sse] = projectionOutboxRealtimeEvents({ events: [{ ...turnRow, outboxSeq: 9 }], facts: {} });
  assert.equal(sse.name, "workbench.turn.snapshot");
  assert.equal(sse.payload.turn.finalResponse.text, "final body");
});

test("late second terminal is acknowledged without facts or either projection outbox", async () => {
  const queries = [];
  const client = {
    async query(sql, params) {
      queries.push({ sql, params });
      if (sql.startsWith("SELECT pg_advisory_xact_lock")) return { rows: [] };
      if (sql.startsWith("SELECT source_topic")) return { rows: [] };
      if (sql.startsWith("INSERT INTO workbench_kafka_inbox")) return { rows: [] };
      if (sql.startsWith("SELECT checkpoint_json")) return { rows: [{ checkpoint_json: { traceId: "trc_sealed", projectedSeq: 9, terminal: true, sealed: true } }] };
      if (sql.startsWith("SELECT owner_user_id")) return { rows: [] };
      if (sql.startsWith("SELECT GREATEST")) return { rows: [{ max_projected_seq: 9 }] };
      if (sql.startsWith("INSERT INTO hwlab_kafka_outbox") || sql.startsWith("UPDATE workbench_kafka_inbox")) return { rows: [] };
      throw new Error(`unexpected SQL: ${sql}`);
    }
  };
  const store = { ...transactionStore(client), memory: { writeWorkbenchFacts() { throw new Error("suppressed facts must not reach memory"); } } };
  const result = await commitAgentRunKafkaProjection(store, {
    transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 0, sourceOffset: "21", inputSha256: "hash-late" },
    canonicalEvent: { eventId: "evt-late", sourceSeq: 10, traceId: "trc_sealed", hwlabSessionId: "ses_sealed" },
    projectedEvent: { traceId: "trc_sealed", sessionId: "ses_sealed" },
    factsFactory: ({ projectedSeq, previousCheckpoint, projectedAt }) => buildWorkbenchProjectionEventFacts({
      projectedSeq,
      previousCheckpoint,
      projectedAt,
      event: { traceId: "trc_sealed", sessionId: "ses_sealed", sourceEventId: "evt-late", sourceSeq: 10, type: "result", eventType: "terminal", status: "failed", terminal: true }
    }),
    hwlabEvent: { eventId: "hwlab:evt-late" },
    hwlabTopic: "hwlab.event.v1",
    partitionKey: "trc_sealed",
    headers: {}
  });

  assert.equal(result.suppressedAfterSeal, true);
  assert.equal(result.projectionWritten, false);
  assert.equal(result.projectedSeq, 9);
  assert.equal(queries.some(({ sql }) => sql.startsWith("INSERT INTO hwlab_kafka_outbox")), false);
  const inboxUpdate = queries.find(({ sql }) => sql.startsWith("UPDATE workbench_kafka_inbox"));
  assert.equal(inboxUpdate.params[0], 9);
});

test("HWLAB outbox claim and completion use attempt and lease fencing", async () => {
  const calls = [];
  const row = { outbox_seq: 3, event_id: "evt-3", topic: "hwlab.event.v1", partition_key: "trc-3", payload_json: {}, headers_json: {}, attempt_count: 4, lease_owner: "relay-a", lease_expires_at: "2026-07-10T10:05:00.000Z", next_attempt_at: "2026-07-10T10:00:00.000Z", created_at: "2026-07-10T10:00:00.000Z" };
  const store = {
    now: () => "2026-07-10T10:04:00.000Z",
    async withDurableTransaction(_label, operation) { return operation({ query: async (sql, params) => { calls.push({ sql, params }); return { rows: [row] }; } }); },
    async query(sql, params) { calls.push({ sql, params }); return { rows: [{ outbox_seq: 3 }] }; }
  };
  const [item] = await claimHwlabKafkaOutbox(store, { owner: "relay-a", leaseMs: 30000, limit: 1 });
  await completeHwlabKafkaOutbox(store, item);
  await retryHwlabKafkaOutbox(store, item, new Error("retry"), "2026-07-10T10:04:30.000Z");

  assert.match(calls[0].sql, /attempt_count = o\.attempt_count \+ 1/u);
  assert.match(calls[0].sql, /NOT EXISTS/u);
  assert.match(calls[1].sql, /lease_owner = \$3 AND attempt_count = \$4 AND lease_expires_at = \$5 AND lease_expires_at > \$1/u);
  assert.deepEqual(calls[1].params.slice(2), ["relay-a", 4, "2026-07-10T10:05:00.000Z"]);
  assert.match(calls[2].sql, /lease_owner = \$5 AND attempt_count = \$6 AND lease_expires_at = \$7 AND lease_expires_at > \$3/u);
});

test("stale HWLAB outbox fencing token cannot complete a claim", async () => {
  const store = { now: () => "2026-07-10T10:06:00.000Z", async query() { return { rows: [] }; } };
  await assert.rejects(completeHwlabKafkaOutbox(store, { outboxSeq: 3, leaseOwner: "relay-old", attemptCount: 1, leaseExpiresAt: "2026-07-10T10:05:00.000Z" }), (error) => error?.code === "hwlab_kafka_outbox_lease_conflict");
});

function transactionStore(client) {
  return {
    now: () => "2026-07-10T10:00:00.000Z",
    async withDurableTransaction(_label, operation) { return operation(client); }
  };
}