Фоновые воркфлоу и очереди задач: BullMQ, Inngest и идемпотентные Cron-пайплайны

+-----------------------------------------------------------------------------------+
|               АРХИТЕКТУРА ASYNC QUEUES & DURABLE WORKFLOWS В NEXT.JS 16           |
+-----------------------------------------------------------------------------------+
| NEXT.JS APP ROUTER              БРОКЕР ОЧЕРЕДИ / ШЕДУЛЕР        ВОРКЕРЫ / RUNTIME |
|                                                                                   |
| 1. Server Action               [ REDIS / BULLMQ ]               [ DOCKER WORKER ] |
|    queue.add("video_transcode",|  - State: 'waiting'            |  Worker loop    |
|              { videoId })      |  - Delay: 0                    |  concurrency: 5 |
|    -> HTTP 202 Accepted        |  - Attempts: 5 (Backoff)       |  FFmpeg процесс |
|                                +------------------------------> |  Job 'completed'|
|                                                                                   |
| 2. Serverless Durable Flow     [ INNGEST EVENT BUS ]            [ SERVERLESS STEP]|
|    inngest.send({              |  - Шаг 1: step.run()           |  POST /api/inngest
|      name: "order.created"     |  - Пауза: step.sleep("2d")     |  Только активный|
|    })                          |  - Шаг 2: step.run()           |  короткий шаг!  |
|                                +------------------------------> |  Нет зависаний  |
|                                                                                   |
| 3. Защищенный Cron             [ VERCEL CRON / GITHUB ACTIONS ]                   |
|    GET /api/cron/cleanup       |  Authorization: Bearer <SECRET>|                 |
|    -> Distributed Lock Redis   +------------------------------> |  БД Cleanup     |
|    -> Выполнение транзакции                                     |  Релиз Lock     |
+-----------------------------------------------------------------------------------+

import { Queue } from 'bullmq';
import IORedis from 'ioredis';

// Синглтон подключения к Redis
const redisConnection = new IORedis(process.env.REDIS_URL || 'redis://localhost:6379', {
  maxRetriesPerRequest: null, // Обязательное требование BullMQ v5
  enableReadyCheck: false,
});

export interface EmailJobData {
  to: string;
  templateId: string;
  payload: Record<string, unknown>;
  idempotencyKey: string;
}

export const emailQueue = new Queue<EmailJobData>('email-delivery', {
  connection: redisConnection,
  defaultJobOptions: {
    attempts: 5, // 5 попыток при сбоях
    backoff: {
      type: 'exponential',
      delay: 2000, // Повторы через 2с, 4с, 8с, 16с, 32с
    },
    removeOnComplete: {
      age: 24 * 3600, // Хранить успешные логи 24 часа
      count: 1000,
    },
    removeOnFail: {
      age: 7 * 24 * 3600, // Хранить ошибки 7 дней для разбора в DLQ
    },
  },
});

import { Worker, Job } from 'bullmq';
import IORedis from 'ioredis';
import { EmailJobData } from '../lib/queues/emailQueue';

const redisConnection = new IORedis(process.env.REDIS_URL || 'redis://localhost:6379', {
  maxRetriesPerRequest: null,
});

export const emailWorker = new Worker<EmailJobData>(
  'email-delivery',
  async (job: Job<EmailJobData>) => {
    console.log(`[Worker] Обработка задачи ${job.id}, попытка ${job.attemptsMade + 1}`);

    const { to, templateId, payload, idempotencyKey } = job.data;

    // ПРОВЕРКА ИДЕМПОТЕНТНОСТИ В REDIS
    const lockKey = `idempotency:email:${idempotencyKey}`;
    const alreadyProcessed = await redisConnection.get(lockKey);

    if (alreadyProcessed) {
      console.warn(`[Worker] Задача ${idempotencyKey} уже была успешно выполнена. Пропуск.`);
      return { status: 'skipped_duplicate' };
    }

    // Имитация отправки через почтовый шлюз (Resend, SendGrid)
    await job.updateProgress(50);
    // await sendEmailViaGateway({ to, templateId, payload });

    // Фиксируем успешное завершение на 48 часов
    await redisConnection.set(lockKey, 'PROCESSED', 'EX', 48 * 3600);
    await job.updateProgress(100);

    return { status: 'sent', sentAt: new Date().toISOString() };
  },
  {
    connection: redisConnection,
    concurrency: 10, // До 10 одновременных задач на один процесс воркера
  }
);

emailWorker.on('failed', (job, err) => {
  console.error(`[Worker Alert] Задача ${job?.id} завершилась ошибкой:`, err.message);
});

import { Inngest } from 'inngest';

// Инициализация типизированного клиента Inngest
export const inngest = new Inngest({
  id: 'wpskills-enterprise-app',
  eventKey: process.env.INNGEST_EVENT_KEY,
});

import { inngest } from '../client';

export const processOrderWorkflow = inngest.createFunction(
  {
    id: 'process-order-durable-workflow',
    name: 'Order Lifecycle & Fraud Check',
    retries: 3, // Автоповторы каждого отдельного шага
  },
  { event: 'shop/order.created' },
  async ({ event, step }) => {
    const { orderId, customerEmail, amountTotal } = event.data;

    // ШАГ 1: Антифрод-проверка (изолированный шаг)
    const fraudResult = await step.run('verify-fraud-risk', async () => {
      // Имитация вызова скоринга
      return { isSuspicious: amountTotal > 500000 };
    });

    if (fraudResult.isSuspicious) {
      await step.run('flag-order-for-review', async () => {
        // await db.orders.update({ where: { id: orderId }, data: { status: 'MANUAL_REVIEW' } });
      });
      return { status: 'flagged_for_review' };
    }

    // ШАГ 2: Списание средств
    const payment = await step.run('charge-payment', async () => {
      // await stripe.paymentIntents.capture(...)
      return { paymentId: `pi_${orderId}`, status: 'succeeded' };
    });

    // ШАГ 3: Засыпание воркфлоу на 24 часа без потребления ресурсов серверов!
    // Serverless-функция полностью завершает работу, Inngest разбудит её через 24 часа
    await step.sleep('wait-before-feedback-email', '24h');

    // ШАГ 4: Отправка письма с просьбой оставить отзыв
    await step.run('send-customer-review-request', async () => {
      // await sendEmail({ to: customerEmail, template: 'review-invitation' });
      return { sent: true };
    });

    return { success: true, orderId, paymentId: payment.paymentId };
  }
);

import { serve } from 'inngest/next';
import { inngest } from '@/inngest/client';
import { processOrderWorkflow } from '@/inngest/functions/orderWorkflow';

export const runtime = 'nodejs';
export const dynamic = 'force-dynamic';

// Next.js Route Handler автоматически маршрутизирует шаги выполнения от Inngest Cloud
export const { GET, POST, PUT } = serve({
  client: inngest,
  functions: [
    processOrderWorkflow,
  ],
});

import { NextRequest, NextResponse } from 'next/server';

export const dynamic = 'force-dynamic';
export const runtime = 'nodejs';

export async function GET(request: NextRequest) {
  // 1. Авторизация вызова планировщика
  const authHeader = request.headers.get('authorization');
  const expectedSecret = process.env.CRON_SECRET;

  if (!expectedSecret || authHeader !== `Bearer ${expectedSecret}`) {
    return NextResponse.json(
      { error: 'Unauthorized: Недействительный или отсутствующий CRON_SECRET' },
      { status: 401 }
    );
  }

  const startTime = Date.now();
  console.log('[CRON] Запуск плановой синхронизации каталога...');

  try {
    // 2. Идемпотентная операция синхронизации
    // const result = await executeCatalogSync();

    const durationMs = Date.now() - startTime;
    return NextResponse.json({
      success: true,
      durationMs,
      timestamp: new Date().toISOString(),
    });
  } catch (error: unknown) {
    console.error('[CRON ERROR]:', error);
    return NextResponse.json(
      { error: 'Internal Error', message: (error as Error).message },
      { status: 500 }
    );
  }
}

{
  "crons": [
    {
      "path": "/api/cron/sync-catalog",
      "schedule": "0 3 * * *"
    }
  ]
}