import { fileIngestionWorker } from './workers/file-ingestion.ts';
import { FILE_INGESTION_CANCEL_CHANNEL } from '@supai/queue-contracts/files';
import { getRedisSubscriber } from '@supai/redis';

let activeJobCount = 0;

fileIngestionWorker.on('active', () => {
  activeJobCount++;
});

fileIngestionWorker.on('completed', () => {
  activeJobCount--;
});

fileIngestionWorker.on('failed', () => {
  activeJobCount--;
});

let isExiting = false;

const exit = async (signal: 'SIGTERM' | 'SIGINT') => {
  if (isExiting) {
    console.log(`Received ${signal}, but already shutting down. Ignoring.`);
    return;
  }

  isExiting = true;

  console.log(
    `Received ${signal}. Starting graceful shutdown... ${activeJobCount.toLocaleString()} job${activeJobCount === 1 ? '' : 's'} processing`,
  );

  const interval = setInterval(() => {
    console.log(
      `Waiting for shutdown... ${activeJobCount.toLocaleString()} job${activeJobCount === 1 ? '' : 's'} still processing`,
    );
  }, 5_000);

  try {
    await fileIngestionWorker.close();
  } catch (error) {
    console.error('Error during exit:', error);
    process.exitCode = 1;
  } finally {
    clearInterval(interval);
  }

  console.log(`Shutdown complete. Exiting with code ${process.exitCode ?? 0}.`);
  process.exit();
};

process.on('SIGTERM', () => exit('SIGTERM'));
process.on('SIGINT', () => exit('SIGINT'));

await Promise.all([
  (async () => {
    const subscribe = await getRedisSubscriber();

    await subscribe(FILE_INGESTION_CANCEL_CHANNEL, fileId => {
      fileIngestionWorker.cancelJob(fileId);
    });

    console.log('File ingestion cancel channel subscribed');
  })(),
  (async () => {
    await new Promise((resolve, reject) => {
      fileIngestionWorker.once('ready', () => {
        resolve();
      });

      fileIngestionWorker.once('error', error => {
        reject(error);
      });
    });

    console.log('File ingestion worker ready');
  })(),
]);