Фонові завдання та черги повідомлень: BullMQ, Redis та надійні воркери

[ Обробник HTTP API ]
                 |
                 v (queue.add)
     +-----------------------+
     |      BullMQ Queue     |
     +-----------------------+
                 |
                 v (Атомарний Lua-скрипт)
  +-------------------------------------+
  |             Стан Redis              |
  |  - zset: delayed (score: timestamp) |
  |  - list: wait (FIFO / LIFO)         |
  |  - hash: job metadata               |
  +-------------------------------------+
                 |
                 | Воркер захоплює замок
                 v (LMOVE + SET lock NX PX)
     +-----------------------+
     |    BullMQ Worker      | <----+ Продовження замка (Watchdog)
     | (concurrency: N)      |      | (lockRenewTime: 15s)
     +-----------------------+      |
                 |                  |
        Event Loop заблоковано? ----+
        Воркер аварійно впав (OOM)?
                 |
                 v Замок минув (30s)
  +-------------------------------------+
  |     Цикл відновлення Stalled        |
  |  Повертає задачу у 'wait' чи 'fail' |
  +-------------------------------------+

+------------+
             |  Delayed   | (Настав запланований час)
             +------------+
                   |
                   v
             +------------+       Воркер отримує задачу
             |    Wait    | ------------------------+
             +------------+                         |
                   ^                                v
                   | Ретрай / Stalled         +------------+
                   +------------------------- |   Active   |
                                              +------------+
                                               /          \
                             Успішний фінал   /            \   Збій / Вичерпано спроби
                                             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, // Зберігати виконані завдання у Redis 24 години
        count: 5000, // Ліміт до 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;
}