Фоновые задачи и очереди сообщений: BullMQ, Redis и надежные воркеры

[ HTTP API Handler ]
                 |
                 v (queue.add)
     +-----------------------+
     |      BullMQ Queue     |
     +-----------------------+
                 |
                 v (Atomic Lua Script)
  +-------------------------------------+
  |             Redis State             |
  |  - zset: delayed (score: timestamp) |
  |  - list: wait (FIFO / LIFO)         |
  |  - hash: job metadata               |
  +-------------------------------------+
                 |
                 | Worker acquires lock
                 v (LMOVE + SET lock NX PX)
     +-----------------------+
     |    BullMQ Worker      | <----+ Heartbeat Lock Renewal
     | (concurrency: N)      |      | (lockRenewTime: 15s)
     +-----------------------+      |
                 |                  |
        Is Event Loop blocked? -----+
        Worker crashed (OOM)?
                 |
                 v Lock Expired (30s)
  +-------------------------------------+
  |     Stalled Job Recovery Loop       |
  |  Moves job back to 'wait' or 'fail' |
  +-------------------------------------+

+------------+
             |  Delayed   | (Scheduled timestamp reached)
             +------------+
                   |
                   v
             +------------+       Worker fetches
             |    Wait    | ------------------------+
             +------------+                         |
                   ^                                v
                   | Retry / Stalled          +------------+
                   +------------------------- |   Active   |
                                              +------------+
                                               /          \
                             Success Result   /            \   Failure / Max Attempts
                                             v              v
                                      +-----------+    +-----------+
                                      | Completed |    |  Failed   |
                                      +-----------+    +-----------+

import { Queue, Worker, QueueEvents, Job } from "bullmq";
import Redis from "ioredis";

export interface PaymentWebhookJobData {
  transactionId: string;
  amount: number;
  currency: string;
  source: string;
  timestamp: number;
}

export interface PaymentJobResult {
  processed: boolean;
  ledgerRecordId: string;
}

const redisConnection = new Redis(process.env.REDIS_URL || "redis://127.0.0.1:6379", {
  maxRetriesPerRequest: null, // Обязательное требование для работы BullMQ!
  enableReadyCheck: false,
});

export const PAYMENT_QUEUE_NAME = "payment-webhooks";

/**
 * Инициализация очереди (Продюсер)
 */
export const paymentQueue = new Queue<PaymentWebhookJobData, PaymentJobResult>(
  PAYMENT_QUEUE_NAME,
  {
    connection: redisConnection,
    defaultJobOptions: {
      attempts: 5, // Максимум 5 попыток при возникновении ошибок
      backoff: {
        type: "exponential",
        delay: 2000, // Начальная пауза 2 сек, затем 4, 8, 16...
      },
      removeOnComplete: {
        age: 86400, // Хранить завершенные задачи в течение 24 часов
        count: 5000, // Лимит количества завершенных задач
      },
      removeOnFail: {
        age: 604800, // Хранить упавшие задачи 7 дней для аудита
      },
    },
  }
);

/**
 * Инициализация надежного воркера с защитой от Stalled Jobs
 */
export function createPaymentWorker(): Worker<PaymentWebhookJobData, PaymentJobResult> {
  const worker = new Worker<PaymentWebhookJobData, PaymentJobResult>(
    PAYMENT_QUEUE_NAME,
    async (job: Job<PaymentWebhookJobData, PaymentJobResult>): Promise<PaymentJobResult> => {
      // Обновление прогресса выполнения для мониторинга
      await job.updateProgress(10);

      // Имитация интеграции с внутренней финансовой системой
      if (job.data.amount <= 0) {
        throw new Error(`Invalid transaction amount: ${job.data.amount}`);
      }

      await job.updateProgress(50);

      // Бизнес-логика транзакции
      const ledgerRecordId = `led_${job.data.transactionId}`;
      await job.updateProgress(100);

      return {
        processed: true,
        ledgerRecordId,
      };
    },
    {
      connection: redisConnection,
      concurrency: 10, // Одновременная обработка до 10 задач в рамках одного процесса
      lockDuration: 30000, // Время удержания распределенной блокировки (30 сек)
      lockRenewTime: 10000, // Продление блокировки каждые 10 сек (Watchdog)
      maxStalledCount: 2, // Максимум 2 зависания перед окончательным перемещением в Failed
      stalledInterval: 30000, // Интервал проверки зависших задач воркером
    }
  );

  // Регистрация критических событий телеметрии
  worker.on("completed", (job: Job) => {
    console.info(`[Worker] Job ${job.id} completed successfully`);
  });

  worker.on("failed", (job: Job | undefined, err: Error) => {
    console.error(`[Worker] Job ${job?.id} failed on attempt ${job?.attemptsMade}: ${err.message}`);
  });

  worker.on("error", (err: Error) => {
    console.error("[Worker] Global worker error:", err.message);
  });

  worker.on("stalled", (jobId: string) => {
    console.warn(`[Worker] Job ${jobId} was marked as STALLED (lock expired)`);
  });

  return worker;
}

import { FlowProducer, FlowJob } from "bullmq";
import Redis from "ioredis";

const redis = new Redis(process.env.REDIS_URL || "redis://127.0.0.1:6379", {
  maxRetriesPerRequest: null,
});

export const flowProducer = new FlowProducer({ connection: redis });

/**
 * Создание иерархического конвейера:
 * Родительская задача агрегации зависит от параллельного выполнения двух дочерних задач
 */
export async function createReportPipeline(reportId: string, companyId: string) {
  const flow: FlowJob = {
    name: "aggregate-final-report",
    queueName: "report-aggregator",
    data: { reportId, companyId },
    children: [
      {
        name: "extract-sales-data",
        queueName: "data-extraction",
        data: { reportId, source: "sales" },
        opts: { attempts: 3 },
      },
      {
        name: "extract-expenses-data",
        queueName: "data-extraction",
        data: { reportId, source: "expenses" },
        opts: { attempts: 3 },
      },
    ],
  };

  const tree = await flowProducer.add(flow);
  return tree.job.id;
}