import { getProducer, KAFKA_TOPICS } from '../config/kafka';

export interface JobMessage {
  name: string;
  data: Record<string, unknown>;
  enqueuedAt: string;
}

// Publishing is best-effort from the API side: never block a request for more
// than a few seconds when the broker is unreachable.
const PUBLISH_TIMEOUT_MS = 8000;

async function publish(topic: string, name: string, data: Record<string, unknown>): Promise<boolean> {
  const message: JobMessage = { name, data, enqueuedAt: new Date().toISOString() };
  try {
    await Promise.race([
      (async () => {
        const producer = await getProducer();
        await producer.send({
          topic,
          messages: [{ key: name, value: JSON.stringify(message) }],
        });
      })(),
      new Promise((_, reject) =>
        setTimeout(() => reject(new Error('publish timed out')), PUBLISH_TIMEOUT_MS),
      ),
    ]);
    return true;
  } catch (err) {
    console.warn(
      `[Kafka] failed to publish "${name}" to ${topic}:`,
      err instanceof Error ? err.message : err,
    );
    return false;
  }
}

export function enqueueNotification(name: string, data: Record<string, unknown>): Promise<boolean> {
  return publish(KAFKA_TOPICS.notifications, name, data);
}

export function enqueueCalendarSync(name: string, data: Record<string, unknown>): Promise<boolean> {
  return publish(KAFKA_TOPICS.calendarSync, name, data);
}