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

import { handleWorkbenchKafkaSseDebugHttp } from "./workbench-kafka-sse-debug.ts";

test("workbench Kafka SSE debug endpoint streams filtered raw HWLAB Kafka events", async () => {
  const fakeKafka = createFakeKafkaFactory();
  const env = {
    HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092",
    HWLAB_KAFKA_CLIENT_ID: "test-hwlab-cloud-api",
    HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1"
  };
  const server = createServer((request, response) => {
    const url = new URL(request.url ?? "/", "http://127.0.0.1");
    void handleWorkbenchKafkaSseDebugHttp(request, response, url, { env, kafkaFactory: fakeKafka.factory, logger: null });
  });
  await listen(server);
  const abort = new AbortController();
  try {
    const response = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab&sessionId=ses_kafka_sse_debug`, { signal: abort.signal });
    assert.equal(response.status, 200);
    assert.match(response.headers.get("content-type") ?? "", /text\/event-stream/u);
    const reader = response.body?.getReader();
    assert.ok(reader, "SSE response must expose a readable stream");
    const connected = await readUntil(reader, "hwlab.kafka.connected");
    assert.match(connected, /hwlab\.event\.v1/u);

    await fakeKafka.emit({ eventType: "hwlab.trace.event.projected", sessionId: "ses_other", event: { label: "ignored" } });
    await fakeKafka.emit({ eventType: "hwlab.trace.event.projected", sessionId: "ses_kafka_sse_debug", traceId: "trc_kafka_sse_debug", context: { runId: "run_kafka_sse_debug", commandId: "cmd_kafka_sse_debug" }, event: { label: "agentrun:event:debug", status: "running" } });
    const streamed = await readUntil(reader, "trc_kafka_sse_debug");
    assert.match(streamed, /hwlab\.kafka\.event/u);
    assert.match(streamed, /ses_kafka_sse_debug/u);
    assert.doesNotMatch(streamed, /ses_other/u);
  } finally {
    abort.abort();
    await close(server);
  }
});

function createFakeKafkaFactory() {
  let eachMessage: ((input: any) => Promise<void>) | null = null;
  const consumer = {
    connect: async () => undefined,
    subscribe: async () => undefined,
    run: async (input: any) => { eachMessage = input.eachMessage; },
    stop: async () => undefined,
    disconnect: async () => undefined
  };
  return {
    factory: () => ({ consumer: () => consumer }),
    emit: async (value: Record<string, unknown>) => {
      assert.ok(eachMessage, "consumer.run must be called before emitting fake Kafka events");
      await eachMessage({
        topic: "hwlab.event.v1",
        partition: 0,
        message: {
          offset: String(value.sessionId === "ses_other" ? 1 : 2),
          key: Buffer.from(String(value.sessionId ?? "unknown")),
          timestamp: "2026-07-09T18:30:00.000Z",
          value: Buffer.from(JSON.stringify(value))
        }
      });
    }
  };
}

async function readUntil(reader: ReadableStreamDefaultReader<Uint8Array>, pattern: string): Promise<string> {
  const decoder = new TextDecoder();
  let text = "";
  const deadline = Date.now() + 3000;
  while (!text.includes(pattern)) {
    if (Date.now() > deadline) throw new Error(`SSE stream did not include ${pattern}: ${text}`);
    const next = await reader.read();
    if (next.done) break;
    text += decoder.decode(next.value, { stream: true });
  }
  return text;
}

async function listen(server: ReturnType<typeof createServer>) {
  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
}

async function close(server: ReturnType<typeof createServer>) {
  await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}

function serverUrl(server: ReturnType<typeof createServer>): string {
  const address = server.address();
  assert.equal(typeof address, "object");
  assert.ok(address && typeof address.port === "number");
  return `http://127.0.0.1:${address.port}`;
}