import { Kafka, type Consumer, type Producer } from "kafkajs";
import { env } from "../../config/env.js";
import { logger } from "../observability/logger.js";

/**
 * msg 服务 Kafka 客户端。
 *
 * 仲裁依据：
 * - producer 配置 idempotent=true + transactionalId（at-least-once + 有序）
 * - consumer 使用 KAFKA_CONSUMER_GROUP_ID 隔离消费组
 * - Kafka 不可用时服务仍启动（降级模式，outbox 消息暂存待发）
 *
 * 使用 lazy initialization：连接在首次 connect 时建立，
 * 避免模块导入副作用导致测试环境难以 mock。
 */
let kafkaInstance: Kafka | null = null;
let producerInstance: Producer | null = null;
let consumerInstance: Consumer | null = null;
let kafkaHealthy = false;

function getKafka(): Kafka {
  if (!kafkaInstance) {
    kafkaInstance = new Kafka({
      brokers: env.KAFKA_BROKERS.split(","),
      clientId: env.KAFKA_CLIENT_ID,
    });
  }
  return kafkaInstance;
}

export function getProducer(): Producer {
  if (!producerInstance) {
    producerInstance = getKafka().producer({
      idempotent: true,
      transactionalId: "msg-service-tx",
    });
  }
  return producerInstance;
}

export function getConsumer(): Consumer {
  if (!consumerInstance) {
    consumerInstance = getKafka().consumer({
      groupId: env.KAFKA_CONSUMER_GROUP_ID,
      sessionTimeout: 30000,
      rebalanceTimeout: 60000,
    });
  }
  return consumerInstance;
}

export async function connectKafka(): Promise<void> {
  try {
    await getProducer().connect();
    await getConsumer().connect();
    kafkaHealthy = true;
    logger.info(
      { brokers: env.KAFKA_BROKERS, clientId: env.KAFKA_CLIENT_ID },
      "Kafka connected",
    );
  } catch (err) {
    kafkaHealthy = false;
    logger.warn(
      { err },
      "Kafka connect failed, running without Kafka (outbox will buffer)",
    );
  }
}

/** /readyz 健康检查：Kafka producer+consumer 是否已连接 */
export function isKafkaHealthy(): boolean {
  return kafkaHealthy;
}

export async function disconnectKafka(): Promise<void> {
  try {
    if (consumerInstance) {
      await consumerInstance.disconnect();
    }
    if (producerInstance) {
      await producerInstance.disconnect();
    }
    kafkaHealthy = false;
    logger.info("Kafka disconnected");
  } catch (err) {
    logger.warn({ err }, "Kafka disconnect error (ignored on shutdown)");
  }
}