AWS SQS Message Processing

# AWS SQS Message Processing

Master message queue processing with AWS SQS using Google Antigravity IDE. This comprehensive guide covers queue configuration, batch processing, and error handling patterns.

## Why SQS?

SQS provides reliable message queuing for decoupled architectures. Google Antigravity IDE's Gemini 3 engine suggests optimal queue configurations and processing patterns.

## Queue Configuration

```typescript
// lib/sqs.ts
import {
  SQSClient,
  SendMessageCommand,
  ReceiveMessageCommand,
  DeleteMessageCommand,
  SendMessageBatchCommand,
} from "@aws-sdk/client-sqs";

const client = new SQSClient({ region: process.env.AWS_REGION });

interface QueueMessage<T = unknown> {
  body: T;
  messageGroupId?: string;
  deduplicationId?: string;
  delaySeconds?: number;
  attributes?: Record<string, string>;
}

export async function sendMessage<T>(
  queueUrl: string,
  message: QueueMessage<T>
): Promise<string> {
  const command = new SendMessageCommand({
    QueueUrl: queueUrl,
    MessageBody: JSON.stringify(message.body),
    MessageGroupId: message.messageGroupId,
    MessageDeduplicationId: message.deduplicationId,
    DelaySeconds: message.delaySeconds,
    MessageAttributes: message.attributes
      ? Object.fromEntries(
          Object.entries(message.attributes).map(([key, value]) => [
            key,
            { DataType: "String", StringValue: value },
          ])
        )
      : undefined,
  });
  
  const response = await client.send(command);
  return response.MessageId || "";
}

export async function sendMessageBatch<T>(
  queueUrl: string,
  messages: Array<QueueMessage<T> & { id: string }>
): Promise<{ successful: string[]; failed: string[] }> {
  const command = new SendMessageBatchCommand({
    QueueUrl: queueUrl,
    Entries: messages.map((msg) => ({
      Id: msg.id,
      MessageBody: JSON.stringify(msg.body),
      MessageGroupId: msg.messageGroupId,
      MessageDeduplicationId: msg.deduplicationId,
      DelaySeconds: msg.delaySeconds,
    })),
  });
  
  const response = await client.send(command);
  
  return {
    successful: response.Successful?.map((s) => s.Id || "") || [],
    failed: response.Failed?.map((f) => f.Id || "") || [],
  };
}
```

## Lambda Handler with Batch Processing

```typescript
// functions/process-messages/index.ts
import { SQSEvent, SQSBatchResponse, SQSRecord } from "aws-lambda";

interface OrderMessage {
  orderId: string;
  userId: string;
  items: Array<{ productId: string; quantity: number }>;
}

export async function handler(event: SQSEvent): Promise<SQSBatchResponse> {
  const batchItemFailures: SQSBatchResponse["batchItemFailures"] = [];
  
  // Process messages concurrently
  const results = await Promise.allSettled(
    event.Records.map((record) => processRecord(record))
  );
  
  // Collect failures for retry
  results.forEach((result, index) => {
    if (result.status === "rejected") {
      console.error(`Failed to process message: ${result.reason}`);
      batchItemFailures.push({
        itemIdentifier: event.Records[index].messageId,
      });
    }
  });
  
  return { batchItemFailures };
}

async function processRecord(record: SQSRecord): Promise<void> {
  const message: OrderMessage = JSON.parse(record.body);
  
  console.log(`Processing order ${message.orderId}`);
  
  // Process order
  await fulfillOrder(message);
  
  // No need to delete - Lambda handles it for successful items
}

async function fulfillOrder(order: OrderMessage): Promise<void> {
  // Order fulfillment logic
  for (const item of order.items) {
    await reserveInventory(item.productId, item.quantity);
  }
}
```

## Dead Letter Queue Handling

```typescript
// functions/process-dlq/index.ts
import { SQSEvent, SQSRecord } from "aws-lambda";
import { sendMessage } from "@/lib/sqs";

const MAIN_QUEUE_URL = process.env.MAIN_QUEUE_URL!;
const MAX_RETRIES = 3;

export async function handler(event: SQSEvent): Promise<void> {
  for (const record of event.Records) {
    await processDeadLetter(record);
  }
}

async function processDeadLetter(record: SQSRecord): Promise<void> {
  const retryCount = parseInt(
    record.messageAttributes?.RetryCount?.stringValue || "0"
  );
  
  if (retryCount < MAX_RETRIES) {
    // Retry the message
    await sendMessage(MAIN_QUEUE_URL, {
      body: JSON.parse(record.body),
      attributes: {
        RetryCount: String(retryCount + 1),
        OriginalMessageId: record.messageId,
      },
      delaySeconds: Math.pow(2, retryCount) * 60, // Exponential backoff
    });
    
    console.log(`Retrying message ${record.messageId}, attempt ${retryCount + 1}`);
  } else {
    // Max retries exceeded, alert and store for manual review
    await alertOperations(record);
    await storeFailedMessage(record);
  }
}
```

## FIFO Queue Processing

```typescript
// lib/fifo-queue.ts
export async function sendOrderMessage(
  queueUrl: string,
  orderId: string,
  action: string,
  data: unknown
): Promise<string> {
  return sendMessage(queueUrl, {
    body: { orderId, action, data, timestamp: Date.now() },
    messageGroupId: orderId, // Ensures order-level FIFO
    deduplicationId: `${orderId}-${action}-${Date.now()}`,
  });
}

// Process in order per customer
export async function sendCustomerEvent(
  queueUrl: string,
  customerId: string,
  event: unknown
): Promise<string> {
  return sendMessage(queueUrl, {
    body: event,
    messageGroupId: customerId,
    deduplicationId: `${customerId}-${Date.now()}-${Math.random()}`,
  });
}
```

## Infrastructure as Code

```typescript
// infrastructure/sqs-stack.ts
import * as cdk from "aws-cdk-lib";
import * as sqs from "aws-cdk-lib/aws-sqs";
import * as lambda from "aws-cdk-lib/aws-lambda";
import * as lambdaEvents from "aws-cdk-lib/aws-lambda-event-sources";

export class SqsStack extends cdk.Stack {
  constructor(scope: cdk.App, id: string) {
    super(scope, id);
    
    // Dead letter queue
    const dlq = new sqs.Queue(this, "OrdersDLQ", {
      queueName: "orders-dlq",
      retentionPeriod: cdk.Duration.days(14),
    });
    
    // Main queue
    const ordersQueue = new sqs.Queue(this, "OrdersQueue", {
      queueName: "orders-queue",
      visibilityTimeout: cdk.Duration.seconds(30),
      deadLetterQueue: {
        queue: dlq,
        maxReceiveCount: 3,
      },
    });
    
    // Lambda processor
    const processor = new lambda.Function(this, "OrderProcessor", {
      runtime: lambda.Runtime.NODEJS_20_X,
      handler: "index.handler",
      code: lambda.Code.fromAsset("functions/process-messages"),
      timeout: cdk.Duration.seconds(25),
    });
    
    processor.addEventSource(
      new lambdaEvents.SqsEventSource(ordersQueue, {
        batchSize: 10,
        reportBatchItemFailures: true,
      })
    );
  }
}
```

## Best Practices

- Enable batch item failure reporting
- Use FIFO queues for ordered processing
- Implement exponential backoff for retries
- Configure appropriate visibility timeout
- Monitor queue depth with CloudWatch
- Use DLQ for unprocessable messages

Google Antigravity IDE provides SQS patterns and automatically suggests optimal queue configurations for your message processing needs.