Function bootWorkers

apps/worker/src/boot-workers.ts:107–408

()

}

export function bootWorkers() {
  const enabledQueues = getEnabledQueues();

  const workers: (Worker | GroupWorker<any>)[] = [];
  const extraStops: Array<() => Promise<unknown>> = [];

  // Start event workers based on enabled queues.
  // When Kafka is configured the producer routes every event to Kafka, so the
  // GroupMQ event shards would only poll an empty queue — skip them entirely
  // and let the Kafka consumer handle ingestion.
  const eventQueuesToStart: number[] = [];

  if (isKafkaConfigured()) {
    logger.info('Kafka is configured, skipping GroupMQ event workers');
  } else if (enabledQueues.includes('events')) {
    // Start all event shards
    for (let i = 0; i < EVENTS_GROUP_QUEUES_SHARDS; i++) {
      eventQueuesToStart.push(i);
    }
  } else {
    // Start specific event shards (events_0, events_1, etc.)
    for (let i = 0; i < EVENTS_GROUP_QUEUES_SHARDS; i++) {
      if (enabledQueues.includes(`events_${i}`)) {
        eventQueuesToStart.push(i);
      }
    }
  }

  if (eventQueuesToStart.length > 0) {
    enableEventsHeartbeat();
  }

  for (const index of eventQueuesToStart) {
    const queue = eventsGroupQueues[index];
    if (!queue) {
      continue;
    }

    const queueName = `events_${index}`;
    const concurrency = getConcurrencyFor(
      queueName,
      Number.parseInt(process.env.EVENT_JOB_CONCURRENCY || '10', 10)
    );

    const worker = new GroupWorker<EventsQueuePayloadIncomingEvent['payload']>({
      queue,
      concurrency,
      logger: process.env.NODE_ENV === 'production' ? queueLogger : undefined,
      blockingTimeoutSec: Number.parseFloat(
        process.env.EVENT_BLOCKING_TIMEOUT_SEC || '1'
      ),
      handler: async (job) => {
        return await incomingEvent(job.data);
      },
    });

    // Consumer-loop heartbeat for the readiness probe. `completed` fires after
    // each processed job; `drained` fires on each poll cycle that finds the