Background job đáng tin cậy trong NestJS với BullMQ: retry, idempotency và transactional outbox

npm install @nestjs/bullmq bullmq ioredis

// infrastructure/queue/queues.ts — tên queue là hằng số, không gõ tay chuỗi ở nhiều nơi
export const PAYMENTS_QUEUE = 'payments';
export const NOTIFICATIONS_QUEUE = 'notifications';
export const SHIPPING_QUEUE = 'shipping';
export const LOYALTY_QUEUE = 'loyalty';
export const MAINTENANCE_QUEUE = 'maintenance';

/** Mọi job sinh ra từ outbox đều mang cùng một "phong bì". */
export interface JobEnvelope<T = Record<string, unknown>> {
  eventId: string; // id của dòng outbox: khoá chống trùng cho consumer
  traceId?: string; // trace ID của request gốc, để log của job nối được với log HTTP
  payload: T; // ID và vài giá trị nhỏ, không phải snapshot của cả đơn hàng
}

// infrastructure/queue/queue.module.ts
import { BullModule } from '@nestjs/bullmq';
import { Module } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { BullMQOtel } from 'bullmq-otel';

const DAY = 24 * 3600; // removeOnComplete/removeOnFail tính age bằng GIÂY

@Module({
  imports: [
    BullModule.forRootAsync({
      inject: [ConfigService], // ConfigModule là global
      useFactory: (config: ConfigService<Env, true>) => {
        const url = new URL(config.get('REDIS_URL', { infer: true }));
        return {
          // Truyền options thay vì một instance ioredis: BullMQ tự mở kết nối cho từng
          // Queue/Worker và tự đặt maxRetriesPerRequest: null cho kết nối blocking của worker
          connection: {
            host: url.hostname,
            port: Number(url.port || 6379),
            username: url.username || undefined,
            password: url.password ? decodeURIComponent(url.password) : undefined,
            tls: url.protocol === 'rediss:' ? {} : undefined,
          },
          prefix: 'shop', // mọi key trong Redis bắt đầu bằng "shop:"
          telemetry: new BullMQOtel({ tracerName: 'shop', meterName: 'shop', enableMetrics: true }),
          defaultJobOptions: {
            attempts: 8,
            backoff: { type: 'exponential', delay: 2_000, jitter: 0.5 },
            removeOnComplete: { age: DAY, count: 10_000 },
            removeOnFail: { age: 14 * DAY }, // giữ job hỏng 14 ngày để điều tra
          },
        };
      },
    }),
    BullModule.registerQueue(
      {
        name: PAYMENTS_QUEUE,
        // Thay thế TOÀN BỘ defaultJobOptions chung (trộn nông), nên phải viết lại đủ
        defaultJobOptions: {
          attempts: 12,
          backoff: { type: 'capped' }, // chiến lược tự viết, khai báo ở worker
          removeOnComplete: { age: 7 * DAY },
          removeOnFail: false, // job thanh toán hỏng: hiếm và đắt, giữ lại mãi
        },
      },
      { name: NOTIFICATIONS_QUEUE },
      { name: SHIPPING_QUEUE },
      { name: LOYALTY_QUEUE },
      { name: MAINTENANCE_QUEUE },
    ),
  ],
  exports: [BullModule], // module nào import QueueModule cũng @InjectQueue() được
})
export class QueueModule {}

// ĐỪNG: hai lần ghi vào hai hệ thống, không có gì nối chúng lại
await this.orders.save(order); // Postgres
await this.paymentsQueue.add('capture-payment', { orderId: order.id }); // Redis

CREATE TABLE outbox_events (
  id            uuid PRIMARY KEY,
  topic         text        NOT NULL,          -- 'order.placed', 'payment.captured'...
  aggregate_id  uuid        NOT NULL,
  payload       jsonb       NOT NULL,
  trace_id      text,                          -- nối log của job với request gốc
  created_at    timestamptz NOT NULL DEFAULT now(),
  published_at  timestamptz                    -- NULL = chưa đẩy sang queue
);

-- Relay chỉ quan tâm dòng chưa publish: partial index giữ index nhỏ dù bảng lớn
CREATE INDEX outbox_unpublished ON outbox_events (created_at) WHERE published_at IS NULL;

// contracts/domain-event.ts — contract trung lập, không phụ thuộc Nest
export type OutboxTopic = 'order.placed' | 'order.cancelled' | 'payment.captured';
export type DomainEvent = { topic: OutboxTopic; aggregateId: string; payload: Record<string, unknown> };

// modules/orders/domain/order.ts (trích)
export class Order {
  private readonly pending: DomainEvent[] = [];

  static place(props: PlaceOrderProps): Order {
    const order = new Order({ ...props, status: 'placed' });
    order.pending.push({
      topic: 'order.placed',
      aggregateId: order.id,
      payload: { orderId: order.id, customerId: order.customerId, total: order.total },
    });
    return order;
  }

  /** Lấy và xoá các event chưa lưu. Chỉ repository gọi hàm này, bên trong save(). */
  pullEvents(): DomainEvent[] {
    return this.pending.splice(0);
  }
}

// modules/orders/application/orders.service.ts — bước 3 và 4 của placeOrder()
const order = Order.place({ id: orderId, ...input, total, reservationId, authorizationId, at: this.clock.now() });
await this.orders.save(order); // đơn hàng + order.placed: một commit
return order.toView(); // không còn this.events.emit(...)

// modules/orders/infrastructure/prisma-order.repository.ts
@Injectable()
export class PrismaOrderRepository extends OrderRepository {
  constructor(
    private readonly prisma: PrismaService,
    private readonly cls: ClsService,
  ) {
    super();
  }

  async save(order: Order): Promise<void> {
    const events = order.pullEvents();
    const traceId = this.cls.isActive() ? this.cls.getId() : undefined;

    await this.prisma.$transaction(async (tx) => {
      const row = toOrderRow(order);
      await tx.order.upsert({ where: { id: row.id }, create: row, update: row });
      if (events.length > 0) {
        await tx.outboxEvent.createMany({
          data: events.map((event) => ({
            id: randomUUID(),
            topic: event.topic,
            aggregateId: event.aggregateId,
            payload: event.payload,
            traceId,
          })),
        });
      }
    });
  }
}

import { InjectQueue } from '@nestjs/bullmq';
import { BeforeApplicationShutdown, Injectable, Logger } from '@nestjs/common';
import type { Queue } from 'bullmq';

interface OutboxRow {
  id: string;
  topic: OutboxTopic;
  payload: Record<string, unknown>;
  trace_id: string | null;
}

interface Route {
  queue: Queue;
  jobName: string;
}

type BulkJob = Parameters<Queue['addBulk']>[0][number];

const BATCH_SIZE = 100;

@Injectable()
export class OutboxRelay implements BeforeApplicationShutdown {
  private readonly logger = new Logger(OutboxRelay.name);
  private readonly routes: Record<OutboxTopic, Route[]>;
  private timer?: NodeJS.Timeout;
  private running?: Promise<void>;
  private stopping = false;

  constructor(
    private readonly prisma: PrismaService,
    @InjectQueue(PAYMENTS_QUEUE) payments: Queue,
    @InjectQueue(NOTIFICATIONS_QUEUE) notifications: Queue,
    @InjectQueue(SHIPPING_QUEUE) shipping: Queue,
    @InjectQueue(LOYALTY_QUEUE) loyalty: Queue,
  ) {
    // Một event, nhiều phản ứng. Record<OutboxTopic, ...>: quên một topic là lỗi compile.
    this.routes = {
      'order.placed': [
        { queue: payments, jobName: 'capture-payment' },
        { queue: notifications, jobName: 'order-confirmation' },
      ],
      'payment.captured': [
        { queue: shipping, jobName: 'create-shipment' },
        { queue: loyalty, jobName: 'award-points' },
      ],
      'order.cancelled': [{ queue: payments, jobName: 'void-authorization' }],
    };
  }

  /** Gọi từ worker.ts. API process không chạy relay. */
  start() {
    this.timer = setInterval(() => {
      this.running ??= this.drain()
        .catch((error) => this.logger.error('Outbox relay failed', error))
        .finally(() => (this.running = undefined));
    }, 500);
  }

  async beforeApplicationShutdown() {
    this.stopping = true;
    clearInterval(this.timer); // ngừng nhận việc mới
    await this.running; // làm nốt mẻ đang chạy
  }

  private async drain(): Promise<void> {
    // Mẻ đầy nghĩa là có thể còn nữa: chạy tiếp ngay, không đợi tick sau
    while (!this.stopping && (await this.publishBatch()) === BATCH_SIZE) {}
  }

  private publishBatch(): Promise<number> {
    return this.prisma.$transaction(
      async (tx) => {
        // Khoá các dòng chưa publish. Relay ở pod khác bỏ qua chúng thay vì đứng chờ.
        const rows = await tx.$queryRaw<OutboxRow[]>`
          SELECT id, topic, payload, trace_id
          FROM outbox_events
          WHERE published_at IS NULL
          ORDER BY created_at
          LIMIT ${BATCH_SIZE}
          FOR UPDATE SKIP LOCKED`;
        if (rows.length === 0) return 0;

        // Topic chưa biết (API bản mới deploy trước worker): để nguyên, relay bản mới sẽ đẩy.
        // Đánh dấu published cho nó ở đây là âm thầm làm mất event.
        const known = rows.filter((row) => row.topic in this.routes);
        if (known.length < rows.length) this.logger.warn(`${rows.length - known.length} outbox row(s) with unknown topic`);

        // Gom theo queue: mỗi queue một lệnh addBulk (nguyên tử trong phạm vi queue đó)
        const byQueue = new Map<Queue, BulkJob[]>();
        for (const row of known) {
          for (const { queue, jobName } of this.routes[row.topic]) {
            const data: JobEnvelope = { eventId: row.id, traceId: row.trace_id ?? undefined, payload: row.payload };
            const jobs = byQueue.get(queue) ?? [];
            jobs.push({ name: jobName, data, opts: { jobId: row.id } }); // cùng event -> cùng jobId
            byQueue.set(queue, jobs);
          }
        }
        await Promise.all([...byQueue].map(([queue, jobs]) => queue.addBulk(jobs)));

        const ids = known.map((row) => row.id);
        await tx.$executeRaw`UPDATE outbox_events SET published_at = now() WHERE id = ANY(${ids}::uuid[])`;
        return rows.length;
      },
      { timeout: 15_000 }, // mặc định của Prisma là 5 giây; mẻ này có gọi Redis bên trong
    );
  }
}

// Gọi lúc worker khởi động; là upsert nên mỗi lần deploy gọi lại cũng không tạo bản trùng
await maintenanceQueue.upsertJobScheduler(
  'outbox-cleanup',
  { pattern: '0 3 * * *', tz: 'Asia/Ho_Chi_Minh' }, // 3 giờ sáng mỗi ngày
  { name: 'outbox-cleanup', data: { olderThanDays: 7 } },
);

CREATE TABLE processed_messages (
  consumer     text        NOT NULL,   -- 'loyalty.award-points'
  message_id   uuid        NOT NULL,   -- id của dòng outbox
  processed_at timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (consumer, message_id)
);

async awardPoints(eventId: string, customerId: string, amount: number): Promise<void> {
  const points = Math.floor(amount / 10_000); // 1 điểm cho mỗi 10.000đ

  await this.prisma.$transaction(async (tx) => {
    // Đánh dấu trước, trong cùng transaction với hiệu ứng
    const inserted = await tx.$executeRaw`
      INSERT INTO processed_messages (consumer, message_id)
      VALUES ('loyalty.award-points', ${eventId}::uuid)
      ON CONFLICT DO NOTHING`;
    if (inserted === 0) return; // event này đã được xử lý: không làm gì

    await tx.$executeRaw`
      UPDATE loyalty_accounts SET points = points + ${points}
      WHERE customer_id = ${customerId}::uuid`;
  });
}

// modules/payments/gateway/payment-gateway.ts — port, Stripe hay cổng nào khác nằm sau nó
export interface PaymentGateway {
  capture(input: { authorizationId: string; amount: number; idempotencyKey: string }): Promise<{ captureId: string }>;
  getAuthorization(
    authorizationId: string,
  ): Promise<{ status: 'captured'; captureId: string } | { status: 'authorized' | 'voided' | 'expired' }>;
}

// modules/payments/application/payments.service.ts (trích)
@Injectable()
export class PaymentsService implements PaymentsApi {
  constructor(
    private readonly payments: PaymentRepository,
    @Inject(PAYMENT_GATEWAY) private readonly gateway: PaymentGateway,
  ) {}

  async captureForOrder(orderId: string): Promise<{ captureId: string }> {
    const payment = await this.payments.findByOrderId(orderId);
    if (!payment) throw new PermanentError(`No payment for order ${orderId}`);

    // 1. Lần chạy trước đã làm xong và ghi lại? Trả về kết quả cũ.
    if (payment.status === 'captured' && payment.captureId) return { captureId: payment.captureId };
    if (payment.status !== 'authorized') {
      throw new PermanentError(`Cannot capture a payment in status "${payment.status}"`);
    }

    // 2. Hỏi cổng thanh toán: lần trước có thể đã capture rồi chết trước bước 4
    const remote = await this.gateway.getAuthorization(payment.authorizationId);
    if (remote.status === 'voided' || remote.status === 'expired') {
      throw new PermanentError(`Authorization is ${remote.status}`); // cần người quyết định
    }

    // 3. Key suy ra từ ý định nghiệp vụ: không phải job id, không phải số lần thử
    const { captureId } =
      remote.status === 'captured'
        ? remote
        : await this.gateway.capture({
            authorizationId: payment.authorizationId,
            amount: payment.amount,
            idempotencyKey: `capture-${orderId}`,
          });

    // 4. Ghi kết quả và event tiếp theo trong một transaction — lại là outbox
    await this.payments.markCaptured(payment, captureId);
    return { captureId };
  }
}

// modules/payments/infrastructure/prisma-payment.repository.ts (trích)
async markCaptured(payment: Payment, captureId: string): Promise<void> {
  await this.prisma.$transaction(async (tx) => {
    // Compare-and-set: chỉ một lần chạy chuyển được authorized -> captured
    const changed = await tx.$executeRaw`
      UPDATE payments SET status = 'captured', capture_id = ${captureId}, captured_at = now()
      WHERE order_id = ${payment.orderId}::uuid AND status = 'authorized'`;
    if (changed === 0) return; // lần chạy khác đã ghi trước: không phát event lần hai

    await tx.outboxEvent.create({
      data: {
        id: randomUUID(),
        topic: 'payment.captured',
        aggregateId: payment.orderId,
        payload: { orderId: payment.orderId, customerId: payment.customerId, amount: payment.amount },
        traceId: this.cls.getId(),
      },
    });
  });
}

// common/errors/job-errors.ts — lỗi theo ý nghĩa, không import gì từ BullMQ
export class PermanentError extends Error {
  override name = 'PermanentError'; // thử lại bao nhiêu lần cũng vậy
}

export class RetryLaterError extends Error {
  override name = 'RetryLaterError'; // "chưa phải lúc": đợi delayMs, không tính là thất bại
  constructor(readonly delayMs: number, reason: string) {
    super(reason);
  }
}

export class ThrottledError extends Error {
  // Hạn mức của cả tài khoản đã cạn: cả queue phải chậm lại, không chỉ job này
  override name = 'ThrottledError';
  constructor(readonly retryAfterMs: number) {
    super(`Throttled for ${retryAfterMs}ms`);
  }
}

/** Phân loại response lỗi từ API bên thứ ba. */
export function errorFromResponse(res: Response, context: string): Error {
  const retryAfterMs = parseRetryAfter(res.headers.get('retry-after'));
  if (res.status === 429) return new ThrottledError(retryAfterMs ?? 10_000);
  if (res.status === 408 || res.status >= 500) {
    return retryAfterMs !== undefined
      ? new RetryLaterError(retryAfterMs, `${context}: HTTP ${res.status}`)
      : new Error(`${context}: HTTP ${res.status}`); // tạm thời: để backoff lo
  }
  // 4xx còn lại: request của mình sai, gửi lại y hệt vẫn sai
  return new PermanentError(`${context}: HTTP ${res.status}`);
}

/** Retry-After là số giây hoặc một HTTP-date (RFC 9110). */
export function parseRetryAfter(value: string | null): number | undefined {
  if (!value) return undefined;
  const seconds = Number(value);
  if (Number.isFinite(seconds)) return Math.max(0, seconds * 1000);
  const at = Date.parse(value);
  return Number.isNaN(at) ? undefined : Math.max(0, at - Date.now());
}

import { OnWorkerEvent, WorkerHost } from '@nestjs/bullmq';
import { BeforeApplicationShutdown, Injectable, Logger } from '@nestjs/common';
import { DelayedError, Job, Queue, UnrecoverableError, Worker } from 'bullmq';
import { CLS_ID, ClsService } from 'nestjs-cls';

/** Dependency dùng chung: processor con chỉ truyền một thứ vào super(). */
@Injectable()
export class JobRuntime {
  constructor(
    readonly cls: ClsService,
    readonly deadLetters: DeadLetterService,
  ) {}
}

export abstract class AppWorkerHost extends WorkerHost implements BeforeApplicationShutdown {
  protected readonly logger = new Logger(this.constructor.name);

  constructor(
    protected readonly runtime: JobRuntime,
    private readonly rateLimitQueue?: Queue, // có thì ThrottledError hãm cả queue
  ) {
    super();
  }

  /** Logic của job: đọc job, gọi use case. Lỗi ném ra phải là lỗi đã phân loại. */
  protected abstract handle(job: Job<JobEnvelope>, token?: string): Promise<unknown>;

  async process(job: Job<JobEnvelope>, token?: string): Promise<unknown> {
    const { cls } = this.runtime;
    // Mỗi job một context CLS riêng, mang trace ID của request đã sinh ra nó
    return cls.run(async () => {
      cls.set(CLS_ID, job.data.traceId ?? `job-${job.queueName}-${job.id}`);
      const meta = `${job.name} id=${job.id} attempt=${job.attemptsMade + 1} trace=${cls.getId()}`;
      const started = performance.now();
      try {
        const result = await this.handle(job, token);
        const ageMs = Date.now() - job.timestamp; // từ lúc được add tới lúc xong
        this.logger.log(`${meta} done in ${(performance.now() - started).toFixed(0)}ms, age ${ageMs}ms`);
        return result;
      } catch (error) {
        this.logger.warn(`${meta} error: ${error instanceof Error ? error.message : String(error)}`);
        throw await this.toBullError(error, job, token);
      }
    });
  }

  private async toBullError(error: unknown, job: Job, token?: string): Promise<Error> {
    if (error instanceof PermanentError) {
      return new UnrecoverableError(error.message); // vào failed ngay, không retry
    }
    if (error instanceof ThrottledError && this.rateLimitQueue) {
      // Cả tài khoản bị giới hạn: dừng lấy job mới từ queue, job này quay về waiting
      await this.rateLimitQueue.rateLimit(error.retryAfterMs);
      return Worker.RateLimitError();
    }
    if (error instanceof RetryLaterError || error instanceof ThrottledError) {
      const delayMs = error instanceof RetryLaterError ? error.delayMs : error.retryAfterMs;
      await job.moveToDelayed(Date.now() + delayMs, token); // cần token của lần xử lý này
      return new DelayedError();
    }
    return error instanceof Error ? error : new Error(String(error)); // tạm thời: retry
  }

  // Event 'failed' bắn cho MỌI lần thất bại, kể cả lần sẽ còn được retry
  @OnWorkerEvent('failed')
  onFailed(job: Job | undefined, error: Error) {
    void this.runtime.deadLetters.recordIfFinal(job, error);
  }

  @OnWorkerEvent('stalled')
  onStalled(jobId: string) {
    this.logger.error(`Job ${jobId} stalled: lock expired, it will run again`);
  }

  async beforeApplicationShutdown() {
    await this.worker.close(); // đợi job đang chạy xong — xem phần graceful shutdown
  }
}

import { Processor } from '@nestjs/bullmq';
import { Module } from '@nestjs/common';
import type { Job } from 'bullmq';

@Processor(PAYMENTS_QUEUE, {
  concurrency: 10,
  settings: { backoffStrategy: cappedExponential }, // cho backoff: { type: 'capped' }
  maxStartedAttempts: 50, // lưới an toàn cho vòng lặp DelayedError
})
export class PaymentsProcessor extends AppWorkerHost {
  constructor(
    runtime: JobRuntime,
    private readonly payments: PaymentsApi,
  ) {
    super(runtime);
  }

  protected handle(job: Job<JobEnvelope<{ orderId: string }>>) {
    const { orderId } = job.data.payload;
    switch (job.name) {
      case 'capture-payment':
        return this.payments.captureForOrder(orderId);
      case 'void-authorization':
        return this.payments.voidForOrder(orderId);
      default:
        throw new PermanentError(`Unknown job name "${job.name}"`);
    }
  }
}

// Module riêng cho phía worker: chỉ WorkerModule import nó, AppModule (HTTP) thì không
@Module({
  imports: [PaymentsModule, JobsCoreModule], // PaymentsApi được PaymentsModule export
  providers: [PaymentsProcessor],
})
export class PaymentsJobsModule {}

// jobs/jobs-core.module.ts — dùng chung cho mọi *JobsModule
@Module({
  providers: [JobRuntime, DeadLetterService],
  exports: [JobRuntime],
})
export class JobsCoreModule {}

import type { BackoffStrategy } from 'bullmq';

const BASE_MS = 2_000;
const CAP_MS = 15 * 60_000;

/** Exponential có trần, jitter trong nửa trên của khoảng. attemptsMade = 1 ở lần retry đầu. */
export const cappedExponential: BackoffStrategy = (attemptsMade) => {
  const ceiling = Math.min(CAP_MS, BASE_MS * 2 ** (attemptsMade - 1));
  return Math.round(ceiling / 2 + Math.random() * (ceiling / 2));
};

CREATE TABLE dead_letters (
  id            uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  queue         text        NOT NULL,
  job_name      text        NOT NULL,
  job_id        text        NOT NULL,
  data          jsonb       NOT NULL,
  error         text        NOT NULL,
  attempts_made int         NOT NULL,
  failed_at     timestamptz NOT NULL DEFAULT now(),
  redriven_at   timestamptz,
  UNIQUE (queue, job_id)
);

import { getQueueToken } from '@nestjs/bullmq';
import { Injectable, Logger } from '@nestjs/common';
import { ModuleRef } from '@nestjs/core';
import { Job, Queue, UnrecoverableError } from 'bullmq';

/** Lần thất bại này có phải lần cuối không? */
export function isFinalFailure(job: Job, error: Error): boolean {
  return (
    error instanceof UnrecoverableError ||
    error.name === 'UnrecoverableError' || // gồm cả stalled quá giới hạn và maxStartedAttempts
    job.attemptsMade >= (job.opts.attempts ?? 1) // attemptsMade đã được tăng khi event bắn
  );
}

@Injectable()
export class DeadLetterService {
  private readonly logger = new Logger(DeadLetterService.name);

  constructor(
    private readonly prisma: PrismaService,
    private readonly moduleRef: ModuleRef,
  ) {}

  async recordIfFinal(job: Job | undefined, error: Error): Promise<void> {
    // job có thể là undefined khi job stalled quá giới hạn và bị removeOnFail xoá
    if (!job || !isFinalFailure(job, error)) return;
    try {
      await this.prisma.$executeRaw`
        INSERT INTO dead_letters (queue, job_name, job_id, data, error, attempts_made)
        VALUES (${job.queueName}, ${job.name}, ${job.id}, ${JSON.stringify(job.data)}::jsonb,
                ${error.message}, ${job.attemptsMade})
        ON CONFLICT (queue, job_id) DO NOTHING`;
      this.logger.error(`Dead-lettered ${job.queueName}/${job.name} id=${job.id}: ${error.message}`);
    } catch (e) {
      // Không ném ra ngoài: đây là event listener. Job vẫn nằm trong failed set của BullMQ.
      this.logger.error(`Could not record dead letter for ${job.queueName}/${job.id}`, e);
    }
  }

  /** Gọi từ endpoint admin, sau khi đã sửa nguyên nhân. */
  async redrive(row: DeadLetterRow): Promise<void> {
    const queue = this.moduleRef.get<Queue>(getQueueToken(row.queue), { strict: false });
    // jobId mới: id cũ có thể vẫn nằm trong failed set, add trùng id sẽ bị bỏ qua
    await queue.add(row.jobName, row.data, { jobId: `${row.jobId}-redrive-${Date.now()}` });
    await this.prisma.$executeRaw`UPDATE dead_letters SET redriven_at = now() WHERE id = ${row.id}::uuid`;
  }
}

async createForOrder(orderId: string): Promise<void> {
  if (await this.shipments.existsForOrder(orderId)) return; // đã xong ở lần trước

  // Hãng không nhận idempotency key, nhưng cho tra theo mã tham chiếu của mình.
  // Tra trước khi tạo: lần chạy trước có thể đã tạo vận đơn rồi chết trước khi ghi DB.
  const parcel = await this.orders.getParcel(orderId); // port do orders export
  const shipment =
    (await this.carrier.findByReference(orderId)) ??
    (await this.carrier.createShipment({ reference: orderId, ...parcel }));

  await this.shipments.record(orderId, shipment.trackingNumber); // UNIQUE (order_id)
}

// modules/shipping/jobs/shipping.processor.ts
@Processor(SHIPPING_QUEUE, {
  concurrency: 4, // tối đa 4 job đang gọi hãng vận chuyển, mỗi worker
  limiter: { max: 4, duration: 1_000 }, // chung cả queue: 4 job/giây, mỗi job tối đa 2 request
})
export class ShippingProcessor extends AppWorkerHost {
  constructor(
    runtime: JobRuntime,
    @InjectQueue(SHIPPING_QUEUE) queue: Queue,
    private readonly shipping: ShippingApi,
  ) {
    super(runtime, queue); // có queue: 429 sẽ hãm cả queue, không chỉ job này
  }

  protected handle(job: Job<JobEnvelope<{ orderId: string }>>) {
    return this.shipping.createForOrder(job.data.payload.orderId);
  }
}

import { NestFactory } from '@nestjs/core';
import { WorkerModule } from './worker.module';

async function bootstrap() {
  const app = await NestFactory.createApplicationContext(WorkerModule, { bufferLogs: true });
  app.enableShutdownHooks(); // SIGTERM -> beforeApplicationShutdown -> worker.close()
  app.get(OutboxRelay).start();
}
void bootstrap();

// worker.module.ts
@Module({
  imports: [
    ConfigModule.forRoot({ isGlobal: true, cache: true, validationSchema: envSchema }),
    ClsModule.forRoot({ global: true }), // không mount middleware: mỗi job tự mở context
    PrismaModule,
    QueueModule,
    OutboxModule,
    PaymentsJobsModule,
    NotificationsJobsModule,
    ShippingJobsModule,
    LoyaltyJobsModule,
  ],
})
export class WorkerModule {}

// Gauge quan trọng nhất không đến từ BullMQ: tuổi của dòng outbox chưa publish lâu nhất
const [lag] = await this.prisma.$queryRaw<{ seconds: number | null }[]>`
  SELECT EXTRACT(EPOCH FROM now() - min(created_at))::float8 AS seconds
  FROM outbox_events WHERE published_at IS NULL`;
outboxLagGauge.set(lag?.seconds ?? 0); // prom-client hoặc OpenTelemetry, tuỳ stack của bạn

// Số job theo trạng thái, cho từng queue
const counts = await queue.getJobCounts('waiting', 'active', 'delayed', 'failed');

// payments.service.spec.ts — Postgres thật (Testcontainers), cổng thanh toán giả
it('capture chạy hai lần chỉ thu tiền một lần, chỉ phát một event', async () => {
  await seedAuthorizedPayment(prisma, { orderId, amount: 250_000 });

  await service.captureForOrder(orderId);
  await service.captureForOrder(orderId); // job bị giao lại

  expect(gateway.captureCalls).toHaveLength(1);
  expect(await prisma.outboxEvent.count({ where: { topic: 'payment.captured', aggregateId: orderId } })).toBe(1);
});

it('chết sau khi capture, trước khi ghi DB: lần chạy lại không capture lần nữa', async () => {
  await seedAuthorizedPayment(prisma, { orderId, amount: 250_000 });
  gateway.failAfterCapture = true; // capture thành công ở cổng, rồi ném lỗi mạng

  await expect(service.captureForOrder(orderId)).rejects.toThrow();
  gateway.failAfterCapture = false;
  await service.captureForOrder(orderId); // retry

  expect(gateway.captureCalls).toHaveLength(1); // bước 2 thấy "captured" ở phía cổng
  expect((await payments.findByOrderId(orderId))?.status).toBe('captured');
});