import { getKafka, KAFKA_TOPICS } from './config/kafka';
import { JobMessage } from './jobs/queues';
import { NotificationService } from './services/NotificationService';
import { CalendarSyncService } from './services/CalendarSyncService';
import { initializeDatabase } from './models/index';

const notificationService = new NotificationService();
const calendarSyncService = new CalendarSyncService();

async function handleNotification(job: JobMessage): Promise<void> {
  switch (job.name) {
    case 'sendConfirmationEmail':
      await notificationService.sendConfirmationEmail(
        job.data.bookingId as string,
        job.data.cancelToken as string,
      );
      break;
    case 'sendCancellationEmail':
      await notificationService.sendCancellationEmail(
        job.data.bookingId as string,
        job.data.cancelledBy as string,
      );
      break;
    case 'sendRescheduleEmail':
      await notificationService.sendRescheduleEmail(job.data.bookingId as string);
      break;
    case 'sendReminder':
      await notificationService.sendReminderEmail(
        job.data.bookingId as string,
        job.data.hoursBefore === 1 ? 1 : 24,
      );
      break;
    case 'sendPostMeetingSurvey':
      await notificationService.sendPostMeetingSurveyEmail(job.data.bookingId as string);
      break;
    case 'sendFollowUp':
      await notificationService.sendFollowUpEmail(job.data.bookingId as string);
      break;
    default:
      console.warn(`[Worker] unknown notification job: ${job.name}`);
  }
}

async function handleCalendarSync(job: JobMessage): Promise<void> {
  switch (job.name) {
    case 'pushCalendarEvent':
    case 'updateCalendarEvent':
      await calendarSyncService.pushCalendarEvent(job.data.bookingId as string);
      break;
    case 'deleteCalendarEvent':
      await calendarSyncService.deleteCalendarEvent(job.data.bookingId as string);
      break;
    case 'syncFreeBusy':
      await calendarSyncService.syncFreeBusyForConnection(job.data.connectionId as string);
      break;
    case 'refreshTokens':
      await calendarSyncService.refreshExpiringTokens();
      break;
    default:
      console.warn(`[Worker] unknown calendar-sync job: ${job.name}`);
  }
}

async function ensureTopics(): Promise<void> {
  const admin = getKafka().admin();
  try {
    await admin.connect();
    await admin.createTopics({
      topics: Object.values(KAFKA_TOPICS).map((topic) => ({
        topic,
        numPartitions: 1,
        replicationFactor: 1,
      })),
      waitForLeaders: true,
    });
  } catch (err) {
    // Topics may already exist or auto-creation may be enabled on the broker.
    console.warn(
      '[Worker] topic creation skipped:',
      err instanceof Error ? err.message : err,
    );
  } finally {
    await admin.disconnect().catch(() => undefined);
  }
}

async function startWorkers() {
  await initializeDatabase();
  await ensureTopics();

  const consumer = getKafka().consumer({ groupId: 'calendar-workers' });
  await consumer.connect();
  await consumer.subscribe({ topics: Object.values(KAFKA_TOPICS) });

  await consumer.run({
    eachMessage: async ({ topic, message }) => {
      if (!message.value) return;

      let job: JobMessage;
      try {
        job = JSON.parse(message.value.toString()) as JobMessage;
      } catch {
        console.warn(`[Worker] skipping malformed message on ${topic}`);
        return;
      }

      try {
        if (topic === KAFKA_TOPICS.notifications) {
          await handleNotification(job);
        } else if (topic === KAFKA_TOPICS.calendarSync) {
          await handleCalendarSync(job);
        }
      } catch (err) {
        // Log and move on: failed jobs must not block the partition.
        console.error(
          `[Worker] job "${job.name}" on ${topic} failed:`,
          err instanceof Error ? err.message : err,
        );
      }
    },
  });

  console.log(`Workers started (Kafka: ${Object.values(KAFKA_TOPICS).join(', ')})`);

  const shutdown = async () => {
    await consumer.disconnect().catch(() => undefined);
    process.exit(0);
  };
  process.on('SIGINT', shutdown);
  process.on('SIGTERM', shutdown);
}

startWorkers().catch((err) => {
  console.error('Worker failed to start:', err);
  process.exit(1);
});