back/src/modules/queue/queue.service.ts - master - guselnikov/inst

back/src/modules/queue/queue.service.ts
66 строк2 KB
Viktor Guselnikov
first_commit
07 июл 2026, 11:54
07 июл 2026, 11:54098126a
import { Injectable, Logger, OnApplicationBootstrap, OnModuleDestroy } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import amqp, { Channel, ChannelModel, ConsumeMessage } from 'amqplib';
import { QueueName } from '../../common/enums';

export type QueueHandler = (payload: string) => Promise<void>;

@Injectable()
export class QueueService implements OnApplicationBootstrap, OnModuleDestroy {
  private readonly logger = new Logger(QueueService.name);
  private connection: ChannelModel | null = null;
  private channel: Channel | null = null;
  private readonly handlers = new Map<string, QueueHandler>();

  constructor(private readonly config: ConfigService) {}

  registerHandler(queue: QueueName, handler: QueueHandler): void {
    this.handlers.set(queue, handler);
  }

  async onApplicationBootstrap(): Promise<void> {
    const url = this.config.get<string>('rabbitmq.url');
    this.connection = await amqp.connect(url!);
    this.channel = await this.connection.createChannel();

    for (const queue of Object.values(QueueName)) {
      await this.channel.assertQueue(queue, { durable: true });
      const handler = this.handlers.get(queue);
      if (!handler) continue;

      await this.channel.consume(queue, (message) => {
        void this.handleMessage(queue, message, handler);
      });
      this.logger.log(`Consuming queue: ${queue}`);
    }
  }

  async publish(queue: QueueName, payload: Record<string, unknown>): Promise<void> {
    if (!this.channel) {
      throw new Error('RabbitMQ channel is not ready');
    }
    this.channel.sendToQueue(queue, Buffer.from(JSON.stringify(payload)), {
      persistent: true,
    });
  }

  private async handleMessage(
    queue: string,
    message: ConsumeMessage | null,
    handler: QueueHandler,
  ): Promise<void> {
    if (!message || !this.channel) return;
    try {
      await handler(message.content.toString('utf8'));
      this.channel.ack(message);
    } catch (error) {
      this.logger.error(`Queue ${queue} handler failed`, error as Error);
      this.channel.nack(message, false, false);
    }
  }

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