Consumidores de AWS SQS

import {
  SQSClient,
  ReceiveMessageCommand,
  DeleteMessageCommand,
} from "@aws-sdk/client-sqs";
 
const sqs = new SQSClient({});
const queueUrl = process.env.SQS_QUEUE_URL!;
 
const res = await sqs.send(
  new ReceiveMessageCommand({
    QueueUrl: queueUrl,
    MaxNumberOfMessages: 10,
    WaitTimeSeconds: 20,
    VisibilityTimeout: 60,
  })
);
 
for (const msg of res.Messages ?? []) {
  await processBody(msg.Body!);
  await sqs.send(
    new DeleteMessageCommand({ QueueUrl: queueUrl, ReceiptHandle: msg.ReceiptHandle! })
  );
}

// src/sqs/poll.ts
import {
  SQSClient,
  ReceiveMessageCommand,
  DeleteMessageCommand,
  ChangeMessageVisibilityCommand,
} from "@aws-sdk/client-sqs";
 
const sqs = new SQSClient({ region: process.env.AWS_REGION });
const queueUrl = process.env.SQS_QUEUE_URL!;
 
async function handleMessage(body: string) {
  const payload = JSON.parse(body) as { type: string; orderId: string };
  if (payload.type === "fulfill") {
    await fulfillOrder(payload.orderId);
  }
}
 
export async function pollForever(signal: AbortSignal) {
  while (!signal.aborted) {
    const res = await sqs.send(
      new ReceiveMessageCommand({
        QueueUrl: queueUrl,
        MaxNumberOfMessages: 10,
        WaitTimeSeconds: 20,
        VisibilityTimeout: 120,
        MessageAttributeNames: ["All"],
      })
    );
 
    for (const msg of res.Messages ?? []) {
      try {
        await handleMessage(msg.Body ?? "{}");
        await sqs.send(
          new DeleteMessageCommand({
            QueueUrl: queueUrl,
            ReceiptHandle: msg.ReceiptHandle!,
          })
        );
      } catch (err) {
        console.error({ msg: "sqs_handler_error", messageId: msg.MessageId, err });
        // El mensaje vuelve a la cola después del tiempo de espera de visibilidad
      }
    }
  }
}
 
// src/worker-main.ts
const ac = new AbortController();
process.on("SIGTERM", () => ac.abort());
await pollForever(ac.signal);

{
  "RedrivePolicy": {
    "deadLetterTargetArn": "arn:aws:sqs:...:dlq",
    "maxReceiveCount": 5
  }
}

// El cuerpo del mensaje puede ser JSON envuelto de SNS - desenvolver el sobre
const outer = JSON.parse(body);
const inner = outer.Message ? JSON.parse(outer.Message) : outer;