Managed Queues

// src/orders/order-publisher.service.ts
import { Inject, Injectable } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { SendMessageCommand, SQSClient } from '@aws-sdk/client-sqs';
import { SQS_CLIENT } from '../aws/aws.module';

interface OrderCreated {
  orderId: string;
  customerId: string;
  totalCents: number;
}

@Injectable()
export class OrderPublisher {
  private readonly queueUrl: string;

  constructor(
    @Inject(SQS_CLIENT) private readonly sqs: SQSClient,
    config: ConfigService,
  ) {
    this.queueUrl = config.getOrThrow<string>('SQS_ORDERS_URL');
  }

  async orderCreated(event: OrderCreated): Promise<void> {
    await this.sqs.send(
      new SendMessageCommand({
        QueueUrl: this.queueUrl,
        // Carry a pattern name yourself: SQS has no concept of one.
        MessageBody: JSON.stringify({ pattern: 'order.created', data: event }),
        MessageAttributes: {
          schemaVersion: { DataType: 'String', StringValue: '1' },
        },
        // FIFO queues only (the URL ends in .fifo). One group is processed in
        // order; different groups run in parallel. SQS rejects these two
        // fields on a standard queue.
        MessageGroupId: event.customerId,
        MessageDeduplicationId: event.orderId,
      }),
    );
  }
}

// src/sqs/sqs.server.ts
import { BaseRpcContext, CustomTransportStrategy, Server } from '@nestjs/microservices';
import {
  DeleteMessageCommand,
  ReceiveMessageCommand,
  SQSClient,
} from '@aws-sdk/client-sqs';

type SqsEvents = { error: (err: unknown) => void };

export class SqsServer extends Server<SqsEvents> implements CustomTransportStrategy {
  // Identifies this transport in logs and in processing hooks.
  transportId = Symbol('SQS');

  private readonly sqs: SQSClient;
  private running = false;

  constructor(private readonly queueUrl: string, region: string) {
    super();
    this.sqs = new SQSClient({ region });
  }

  // Called by app.listen(). Invoke the callback once you are ready.
  listen(callback: (...optionalParams: unknown[]) => void): void {
    this.running = true;
    void this.poll();
    callback();
  }

  // Called on application shutdown.
  close(): void {
    this.running = false;
    this.sqs.destroy();
  }

  // Required by Server. Throwing is legitimate when there is nothing to emit.
  on<K extends keyof SqsEvents>(event: K, callback: SqsEvents[K]): never {
    throw new Error('SqsServer does not expose broker events');
  }

  // Required by Server: hands back the underlying client.
  unwrap<T>(): T {
    return this.sqs as T;
  }

  private async poll(): Promise<void> {
    while (this.running) {
      const { Messages = [] } = await this.sqs.send(
        new ReceiveMessageCommand({
          QueueUrl: this.queueUrl,
          MaxNumberOfMessages: 10,
          WaitTimeSeconds: 20,
        }),
      );

      for (const message of Messages) {
        const { pattern, data } = JSON.parse(message.Body ?? '{}');

        // messageHandlers was filled from the @EventPattern decorators.
        if (!this.getHandlerByPattern(pattern)) continue;

        // handleEvent runs the handler through its guards, interceptors,
        // pipes and exception filters. The context is whatever you pass.
        await this.handleEvent(pattern, { pattern, data }, new BaseRpcContext([message]));

        await this.sqs.send(
          new DeleteMessageCommand({
            QueueUrl: this.queueUrl,
            ReceiptHandle: message.ReceiptHandle!,
          }),
        );
      }
    }
  }
}

// src/main.ts
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  // Longhand: you build the strategy. The only form for SQS.
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
    strategy: new SqsServer(process.env.SQS_ORDERS_URL!, process.env.AWS_REGION!),
  });

  // Shorthand, for the seven brokers Nest ships:
  //   { transport: Transport.RMQ, options: { urls: ['amqp://localhost'] } }
  // Nest's ServerFactory turns that into new ServerRMQ(options) -- the same
  // object you would pass as `strategy` yourself.
  void Transport.RMQ;

  await app.listen();
}
void bootstrap();

// src/events/pubsub.consumer.ts
import {
  Injectable,
  Logger,
  OnApplicationBootstrap,
  OnApplicationShutdown,
} from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { Message, PubSub, Subscription } from '@google-cloud/pubsub';
import { OrdersService } from '../orders/orders.service';

@Injectable()
export class PubSubConsumer implements OnApplicationBootstrap, OnApplicationShutdown {
  private readonly logger = new Logger(PubSubConsumer.name);
  private readonly pubsub: PubSub;
  private subscription?: Subscription;

  constructor(
    private readonly orders: OrdersService,
    private readonly config: ConfigService,
  ) {
    this.pubsub = new PubSub({ projectId: this.config.getOrThrow<string>('GCP_PROJECT_ID') });
  }

  onApplicationBootstrap(): void {
    // The client streams messages to you; there is no polling loop here.
    this.subscription = this.pubsub.subscription('orders-sub', {
      flowControl: { maxMessages: 20 }, // in-flight cap, your backpressure knob
    });

    this.subscription.on('message', async (message: Message) => {
      try {
        // message.data is a Buffer, not a string.
        await this.orders.confirm(JSON.parse(message.data.toString('utf8')));
        message.ack();
      } catch (error) {
        this.logger.error(`Message ${message.id} failed`, error);
        // deliveryAttempt is populated only when the subscription has a
        // dead-letter policy attached.
        message.nack(); // redelivered immediately rather than after the ack deadline
      }
    });

    this.subscription.on('error', (error) => this.logger.error('Subscription error', error));
  }

  async onApplicationShutdown(): Promise<void> {
    await this.subscription?.close(); // stops delivery, waits for callbacks in flight
    await this.pubsub.close();
  }
}