dragonflight/services/worker/src/index.js

import 'dotenv/config';
import { Worker } from 'bullmq';
import { proxyWorker } from './workers/proxy.js';
import { thumbnailWorker } from './workers/thumbnail.js';
import { conformWorker } from './workers/conform.js';
import { youtubeImportWorker } from './workers/youtube-import.js';
import { trimWorker } from './workers/trimWorker.js';
import { startPromotionWorker } from './workers/promotion.js';

const parseRedisUrl = (url) => {
  const parsed = new URL(url);
  return {
    host: parsed.hostname,
    port: parseInt(parsed.port, 10),
    password: parsed.password || undefined,
  };
};

const redisOptions = parseRedisUrl(process.env.REDIS_URL || 'redis://localhost:6379');

const createWorker = (queueName, handler, overrides = {}) => {
  const worker = new Worker(queueName, handler, {
    connection: redisOptions,
    // Stall detection: if a worker dies mid-job, BullMQ moves it back to wait
    // after stalledInterval. Without this a crashed run sits in active forever.
    stalledInterval: 30000,
    maxStalledCount: 1,
    lockDuration: 60000,
    lockRenewTime: 15000,
    ...overrides,
  });

  worker.on('completed', (job) => {
    console.log(`[${queueName}] Job ${job.id} completed`);
  });

  worker.on('failed', (job, err) => {
    console.error(`[${queueName}] Job ${job.id} failed:`, err.message);
  });

  worker.on('stalled', (jobId) => {
    console.warn(`[${queueName}] Job ${jobId} stalled — reclaimed`);
  });

  worker.on('progress', (job, progress) => {
    console.log(`[${queueName}] Job ${job.id} progress:`, progress);
  });

  return worker;
};

// Per-queue concurrency. Defaults to 1, which serialises every job in a
// queue — meaning a single stalled job blocks every other one. We want
// thumbnails (cheap, parallel-safe) to run several at a time so a slow
// outlier doesn't back the rest of the catalog up. Proxy + conform are
// heavier (ffmpeg transcode) so we keep them lower to avoid trashing
// the box; tune via env if a node has more headroom.
const PROXY_CONCURRENCY     = parseInt(process.env.PROXY_CONCURRENCY     || '2', 10);
const THUMBNAIL_CONCURRENCY = parseInt(process.env.THUMBNAIL_CONCURRENCY || '4', 10);
const CONFORM_CONCURRENCY   = parseInt(process.env.CONFORM_CONCURRENCY   || '1', 10);
const TRIM_CONCURRENCY      = parseInt(process.env.TRIM_CONCURRENCY      || '4', 10);

const workers = [
  createWorker('proxy',     proxyWorker,     { concurrency: PROXY_CONCURRENCY }),
  createWorker('thumbnail', thumbnailWorker, { concurrency: THUMBNAIL_CONCURRENCY }),
  createWorker('conform',   conformWorker,   { concurrency: CONFORM_CONCURRENCY }),
  createWorker('trim',      trimWorker,      { concurrency: TRIM_CONCURRENCY }),
  // YouTube imports: keep concurrency at 1 so we don't burn through rate
  // limits when several jobs land back-to-back. Lock window is longer than
  // the default because a long video download can run for minutes.
  createWorker('import', youtubeImportWorker, {
    concurrency: 1,
    lockDuration: 10 * 60 * 1000,
    lockRenewTime: 60000,
  }),
];

console.log(`Concurrency: proxy=${PROXY_CONCURRENCY} thumbnail=${THUMBNAIL_CONCURRENCY} conform=${CONFORM_CONCURRENCY} trim=${TRIM_CONCURRENCY} import=1`);

startPromotionWorker();

console.log('Wild Dragon Worker Service started');
console.log(`Redis: ${redisOptions.host}:${redisOptions.port}`);
console.log('Active queues: proxy, thumbnail, conform, trim, import');
console.log('Background scans:  promotion (growing-files → S3)');

process.on('SIGTERM', async () => {
  console.log('SIGTERM received, shutting down...');
  await Promise.all(workers.map(w => w.close()));
  process.exit(0);
});