// Test script: Kafka → push-gateway → WebSocket full chain (v2)
// Usage: node scripts/test-kafka-ws.mjs
// Prerequisites: push-gateway running on localhost:8081, Kafka on localhost:9092,
// topic edu.notify.notification.sent exists.

import WebSocket from "ws";
import { Kafka } from "kafkajs";

const WS_URL = "ws://localhost:8081/ws?token=dev-token";
const KAFKA_BROKER = "localhost:9092";
const TOPIC = "edu.notify.notification.sent";
const USER_ID = "dev-user"; // DevMode maps dev-token to this user
const EVENT_TYPE = "TestEventV2KafkaChain";

console.log("[test] connecting WebSocket to", WS_URL);
const ws = new WebSocket(WS_URL);

const messages = [];
ws.on("open", () => {
  console.log("[test] WebSocket connected, publishing Kafka message...");
  publishKafka();
});
ws.on("message", (data) => {
  const msg = JSON.parse(data.toString());
  console.log("[test] received WS message:", JSON.stringify(msg));
  messages.push(msg);
});
ws.on("error", (err) => {
  console.error("[test] WebSocket error:", err.message);
  process.exit(1);
});

async function publishKafka() {
  const kafka = new Kafka({ brokers: [KAFKA_BROKER] });
  const producer = kafka.producer();
  await producer.connect();
  const payload = {
    event_id: `evt-test-${Date.now()}`,
    user_id: USER_ID,
    event_type: EVENT_TYPE,
    channel: "ws",
    title: "v2 Kafka chain test",
    content: "Verifying edu.notify.notification.sent consumption",
    data: { message: "hello from v2 kafka chain test" },
    broadcast: false,
    occurred_at: Date.now(),
  };
  await producer.send({
    topic: TOPIC,
    messages: [{ value: JSON.stringify(payload) }],
  });
  console.log("[test] Kafka message published to", TOPIC);
  await producer.disconnect();

  // Wait up to 10 seconds for the WS message.
  const deadline = Date.now() + 10000;
  while (Date.now() < deadline && messages.length === 0) {
    await new Promise((r) => setTimeout(r, 200));
  }
  if (messages.length === 0) {
    console.error("[test] FAIL: no WS message received within 10s");
    process.exit(1);
  }
  const got = messages[0];
  if (got.event === EVENT_TYPE && got.type === "message") {
    console.log("[test] PASS: WS message matches expected event_type");
    process.exit(0);
  } else {
    console.error("[test] FAIL: WS message mismatch:", JSON.stringify(got));
    process.exit(1);
  }
}

// Hard timeout.
setTimeout(() => {
  console.error("[test] FAIL: overall timeout 15s");
  process.exit(1);
}, 15000);