import { Kafka } from "kafkajs";
import { env } from "./env.js";
import { logger } from "../shared/observability/logger.js";

export const kafka = new Kafka({
  brokers: env.KAFKA_BROKERS.split(","),
  clientId: "content-service",
});

export const producer = kafka.producer({
  idempotent: true,
  transactionalId: "content-tx",
});

export const neo4jSyncConsumer = kafka.consumer({
  groupId: "content-neo4j-sync",
});

let producerIsConnected = false;

producer.on("producer.connect", () => {
  producerIsConnected = true;
});
producer.on("producer.disconnect", () => {
  producerIsConnected = false;
});

export function isKafkaProducerConnected(): boolean {
  return producerIsConnected;
}

export async function connectKafka(): Promise<void> {
  try {
    await producer.connect();
    logger.info("Kafka producer connected");
  } catch (err) {
    logger.warn(
      {
        err: err instanceof Error ? err.message : String(err),
      },
      "Kafka connect failed, running without Kafka (outbox will retry)",
    );
  }
}

export async function disconnectKafka(): Promise<void> {
  try {
    await producer.disconnect();
  } catch {
    // ignore
  }
  try {
    await neo4jSyncConsumer.disconnect();
  } catch {
    // ignore
  }
}