const path = require('path');
require('dotenv').config({ path: path.join(__dirname, '../.env') });
require('dotenv').config({ path: path.join(__dirname, '../../backend/.env') });

const { Worker } = require('bullmq');
const { connectDB } = require('../src/config/database');
const { getRedisConnection, verifyRedisConnection, QUEUE_PREFIX, INGEST_QUEUE } = require('../src/config/queue');

function attachWorkerErrors(worker, label) {
  worker.on('error', (err) => {
    console.error(`[${label}] Redis/BullMQ error:`, err.message);
    if (err.code === 'ECONNRESET') {
      console.error(
        `[${label}] Connection reset — if using DigitalOcean Redis, ensure REDIS_URL uses rediss:// or REDIS_TLS=true`,
      );
    }
  });
}

async function main() {
  await connectDB();
  await verifyRedisConnection();

  const { runConciergeIngest } = require('../../backend/src/utils/conciergeIngest');

  const worker = new Worker(
    INGEST_QUEUE,
    async (job) => {
      const { sourceId } = job.data;
      console.log('[ingest] processing', sourceId);
      const result = await runConciergeIngest(sourceId);
      console.log('[ingest] done', sourceId, result);
      return result;
    },
    {
      connection: getRedisConnection(),
      prefix: QUEUE_PREFIX,
      concurrency: 2,
      lockDuration: 900000,
      stalledInterval: 120000,
    },
  );

  worker.on('failed', (job, err) => {
    console.error('[ingest] failed', job?.id, err.message);
  });
  attachWorkerErrors(worker, 'ingest');

  console.log('[ingest] worker started');
}

main().catch((e) => {
  console.error(e);
  process.exit(1);
});