Повторні спроби та обробка помилок

затримка = базоваЗатримка × 2 ^ номерСпроби

спроба 1: 5 секунд
спроба 2: 10 секунд
спроба 3: 20 секунд

function getBackoffDelay(
  attempt: number,
  baseDelayMs = 5_000,
  maxDelayMs = 60_000,
): number {
  const exponentialDelay = baseDelayMs * 2 ** attempt;
  const jitter = Math.floor(Math.random() * 1_000);

  return Math.min(exponentialDelay + jitter, maxDelayMs);
}

orders
  ├── orders.retry.1 ── 5 секунд ──┐
  ├── orders.retry.2 ── 10 секунд ─┤──> orders
  └── orders.dlq

npm install @nestjs/common @nestjs/core amqplib reflect-metadata rxjs
npm install --save-dev @types/amqplib typescript ts-node

import {
  Injectable,
  Logger,
  Module,
  OnModuleDestroy,
  OnModuleInit,
} from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import {
  Channel,
  ConfirmChannel,
  Connection,
  ConsumeMessage,
  Options,
} from 'amqplib';
import * as amqp from 'amqplib';

const MAIN_QUEUE = 'orders';
const RETRY_QUEUES = ['orders.retry.1', 'orders.retry.2'];
const DLQ = 'orders.dlq';
const MAX_RETRIES = RETRY_QUEUES.length;

type OrderMessage = {
  orderId: string;
  status?: string;
};

@Injectable()
class OrdersConsumer implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(OrdersConsumer.name);

  private connection!: Connection;
  private channel!: ConfirmChannel;

  async onModuleInit(): Promise<void> {
    const url = process.env.RABBITMQ_URL ?? 'amqp://localhost:5672';

    this.connection = await amqp.connect(url);
    this.channel = await this.connection.createConfirmChannel();

    await this.setupQueues();

    await this.channel.prefetch(10);

    await this.channel.consume(
      MAIN_QUEUE,
      (message) => {
        if (message) {
          void this.handleMessage(message);
        }
      },
      { noAck: false },
    );

    this.logger.log(`Очікування повідомлень у черзі "${MAIN_QUEUE}"`);
  }

  private async setupQueues(): Promise<void> {
    await this.channel.assertQueue(MAIN_QUEUE, {
      durable: true,
    });

    await this.channel.assertQueue(RETRY_QUEUES[0], {
      durable: true,
      arguments: {
        'x-message-ttl': 5_000,
        'x-dead-letter-exchange': '',
        'x-dead-letter-routing-key': MAIN_QUEUE,
      },
    });

    await this.channel.assertQueue(RETRY_QUEUES[1], {
      durable: true,
      arguments: {
        'x-message-ttl': 10_000,
        'x-dead-letter-exchange': '',
        'x-dead-letter-routing-key': MAIN_QUEUE,
      },
    });

    await this.channel.assertQueue(DLQ, {
      durable: true,
    });
  }

  private async handleMessage(message: ConsumeMessage): Promise<void> {
    const headers = message.properties.headers ?? {};
    const retryCount = Number(headers['x-retry-count'] ?? 0);

    try {
      const payload = this.parseMessage(message);

      await this.processOrder(payload);

      this.channel.ack(message);

      this.logger.log(`Замовлення ${payload.orderId} успішно оброблено`);
    } catch (error) {
      const normalizedError = this.normalizeError(error);

      this.logger.error(
        `Помилка обробки повідомлення. Спроба: ${retryCount}`,
        normalizedError.stack,
      );

      try {
        if (normalizedError.permanent || retryCount >= MAX_RETRIES) {
          await this.publishToDeadLetterQueue(message, normalizedError);
        } else {
          await this.publishToRetryQueue(message, retryCount, normalizedError);
        }

        // Підтверджуємо оригінал лише після успішної публікації копії.
        this.channel.ack(message);
      } catch (publishError) {
        this.logger.error(
          'Не вдалося перенаправити повідомлення',
          publishError instanceof Error ? publishError.stack : String(publishError),
        );

        // Якщо нове повідомлення не опубліковане, дозволяємо повторну доставку.
        this.channel.nack(message, false, true);
      }
    }
  }

  private parseMessage(message: ConsumeMessage): OrderMessage {
    try {
      const value = JSON.parse(message.content.toString()) as OrderMessage;

      if (!value.orderId || typeof value.orderId !== 'string') {
        throw new Error('Поле orderId є обов’язковим');
      }

      return value;
    } catch (error) {
      const parseError = new Error(
        `Некоректний формат повідомлення: ${
          error instanceof Error ? error.message : String(error)
        }`,
      );

      Object.assign(parseError, { permanent: true });

      throw parseError;
    }
  }

  private async processOrder(order: OrderMessage): Promise<void> {
    // Імітація тимчасової помилки зовнішнього сервісу.
    if (order.status === 'temporary-failure') {
      const error = new Error('Зовнішній сервіс тимчасово недоступний');
      Object.assign(error, {
        code: 'EXTERNAL_SERVICE_UNAVAILABLE',
        permanent: false,
      });

      throw error;
    }

    // Імітація постійної бізнес-помилки.
    if (order.status === 'invalid') {
      const error = new Error('Замовлення має недійсний статус');
      Object.assign(error, {
        code: 'INVALID_ORDER_STATUS',
        permanent: true,
      });

      throw error;
    }

    // Тут могла б бути транзакція в базі даних або виклик іншого сервісу.
    await Promise.resolve();
  }

  private async publishToRetryQueue(
    message: ConsumeMessage,
    retryCount: number,
    error: NormalizedError,
  ): Promise<void> {
    const nextRetryCount = retryCount + 1;
    const retryQueue = RETRY_QUEUES[nextRetryCount - 1];

    const headers = {
      ...(message.properties.headers ?? {}),
      'x-retry-count': nextRetryCount,
      'x-last-error': error.message.slice(0, 500),
    };

    this.channel.sendToQueue(retryQueue, message.content, {
      persistent: true,
      contentType: message.properties.contentType,
      correlationId: message.properties.correlationId,
      headers,
    });

    await this.channel.waitForConfirms();

    this.logger.warn(
      `Повідомлення перенаправлено в ${retryQueue}. ` +
        `Наступна спроба: ${nextRetryCount}`,
    );
  }

  private async publishToDeadLetterQueue(
    message: ConsumeMessage,
    error: NormalizedError,
  ): Promise<void> {
    const headers = {
      ...(message.properties.headers ?? {}),
      'x-final-error': error.message.slice(0, 500),
      'x-error-code': error.code ?? 'UNKNOWN_ERROR',
    };

    this.channel.sendToQueue(DLQ, message.content, {
      persistent: true,
      contentType: message.properties.contentType,
      correlationId: message.properties.correlationId,
      headers,
    });

    await this.channel.waitForConfirms();

    this.logger.error(`Повідомлення перенаправлено в ${DLQ}`);
  }

  private normalizeError(error: unknown): NormalizedError {
    if (error instanceof Error) {
      const typedError = error as Error & {
        code?: string;
        permanent?: boolean;
      };

      return {
        message: typedError.message,
        stack: typedError.stack,
        code: typedError.code,
        permanent: typedError.permanent === true,
      };
    }

    return {
      message: String(error),
      permanent: false,
    };
  }

  async onModuleDestroy(): Promise<void> {
    await this.channel?.close();
    await this.connection?.close();
  }
}

type NormalizedError = {
  message: string;
  stack?: string;
  code?: string;
  permanent: boolean;
};

@Module({
  providers: [OrdersConsumer],
})
class AppModule {}

async function bootstrap(): Promise<void> {
  const app = await NestFactory.createApplicationContext(AppModule);

  const shutdown = async (): Promise<void> => {
    await app.close();
    process.exit(0);
  };

  process.once('SIGINT', () => void shutdown());
  process.once('SIGTERM', () => void shutdown());
}

void bootstrap();

docker run --rm \
  --name lesson-rabbitmq \
  -p 5672:5672 \
  rabbitmq:3

class PermanentProcessingError extends Error {
  readonly permanent = true;
  readonly code: string;

  constructor(message: string, code: string) {
    super(message);
    this.name = 'PermanentProcessingError';
    this.code = code;
  }
}

class TemporaryProcessingError extends Error {
  readonly permanent = false;
  readonly code: string;

  constructor(message: string, code: string) {
    super(message);
    this.name = 'TemporaryProcessingError';
    this.code = code;
  }
}

channel.nack(message, false, true);

channel.ack(message);
await processMessage(message);

await processMessage(message);
channel.ack(message);

channel.nack(message, false, true);