import { Kafka, Producer, logLevel } from 'kafkajs';
import { env } from './env';

export const KAFKA_TOPICS = {
  notifications: 'calendar.notifications',
  calendarSync: 'calendar.calendar-sync',
} as const;

let kafka: Kafka | null = null;
let producer: Producer | null = null;
let producerConnecting: Promise<void> | null = null;
let producerReady = false;

export function getKafka(): Kafka {
  if (!kafka) {
    kafka = new Kafka({
      clientId: env.KAFKA_CLIENT_ID,
      brokers: [env.KAFKA_BROKERS],
      connectionTimeout: 10000,
      requestTimeout: 30000,
      retry: { initialRetryTime: 300, retries: 5 },
      logLevel: logLevel.ERROR,
    });
  }
  return kafka;
}

export async function getProducer(): Promise<Producer> {
  if (!producer) {
    producer = getKafka().producer({ allowAutoTopicCreation: true });
    producer.on('producer.disconnect', () => {
      producerReady = false;
    });
  }
  if (!producerReady) {
    if (!producerConnecting) {
      producerConnecting = producer
        .connect()
        .then(() => {
          producerReady = true;
        })
        .finally(() => {
          producerConnecting = null;
        });
    }
    await producerConnecting;
  }
  return producer;
}

/** Connect the producer in the background so the first publish is fast. */
export function warmUpProducer(): void {
  getProducer()
    .then(() => console.log('[Kafka] producer connected'))
    .catch((err) =>
      console.warn('[Kafka] producer warm-up failed:', err instanceof Error ? err.message : err),
    );
}

export async function closeKafkaProducer(): Promise<void> {
  if (producer) {
    await producer.disconnect().catch(() => undefined);
    producer = null;
    producerReady = false;
  }
}