1. Producer: SendMessage → message enters queue (visible)
2. Consumer: ReceiveMessage → message becomes INVISIBLE for visibility_timeout seconds
3. Consumer: processes the message
   → Success path: DeleteMessage → message permanently removed
   → Failure path: handler throws exception → message remains invisible until timeout
4. After visibility_timeout expires: message becomes VISIBLE again → another consumer picks it up
5. After maxReceiveCount failures: message moves to DLQ

// CDK: SQS queue with DLQ and Lambda ESM
import { Queue } from 'aws-cdk-lib/aws-sqs'
import { SqsEventSource } from 'aws-cdk-lib/aws-lambda-event-sources'

const dlq = new Queue(this, 'OrderProcessorDLQ', {
  retentionPeriod: Duration.days(14), // Keep failed messages for 2 weeks
})

const queue = new Queue(this, 'OrderQueue', {
  // Visibility timeout must be >= 6× the Lambda function timeout
  // If Lambda timeout = 30s, set visibility timeout = 180s minimum
  visibilityTimeout: Duration.seconds(300),
  deadLetterQueue: {
    queue: dlq,
    maxReceiveCount: 3, // After 3 failed attempts, message goes to DLQ
  },
})

orderProcessorFn.addEventSource(new SqsEventSource(queue, {
  batchSize: 10,
  maxBatchingWindow: Duration.seconds(5),
  reportBatchItemFailures: true,
}))

import { SQSClient, ChangeMessageVisibilityCommand } from '@aws-sdk/client-sqs'

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

export const handler = async (event: AWSLambda.SQSEvent, context: AWSLambda.Context) => {
  for (const record of event.Records) {
    // Extend visibility timeout every 30s to prevent re-delivery during long processing
    const extensionInterval = setInterval(async () => {
      const remaining = context.getRemainingTimeInMillis()
      if (remaining > 5000) { // Still have time left
        await sqs.send(new ChangeMessageVisibilityCommand({
          QueueUrl: process.env.QUEUE_URL!,
          ReceiptHandle: record.receiptHandle,
          VisibilityTimeout: 60, // Extend by another 60 seconds
        }))
      }
    }, 30_000)

    try {
      await processLongRunningRecord(record)
    } finally {
      clearInterval(extensionInterval)
    }
  }
}

// FIFO queue: orders within a user's account processed in strict order
const fifoQueue = new Queue(this, 'OrderFIFO', {
  fifo: true,
  contentBasedDeduplication: true, // Hash message body for deduplication ID
})

// Producer: assign Message Group ID per user → user's orders are serialized
await sqs.send(new SendMessageCommand({
  QueueUrl: fifoQueueUrl,
  MessageBody: JSON.stringify(orderPayload),
  MessageGroupId: `USER#${userId}`,    // All orders for this user are ordered
  // Different users (different GroupIds) process concurrently
}))

SNS delivery retry policy (default for Lambda subscriptions):
  - Immediate delivery attempt #1
  - 1s delay → attempt #2
  - 4s delay → attempt #3 (total: 5s from publish)
  - After 3 failures: message DROPPED

Contrast with SQS:
  - Message remains in queue until deleted or maxReceiveCount exceeded
  - DLQ automatically captures exhausted messages
  - No implicit message loss

// ❌ SNS → Lambda as a work queue (message loss risk)
// Work item published to SNS → Lambda fails 3 times → message dropped
// No DLQ on Lambda subscription by default → lost work, no visibility

// ✅ SNS → SQS → Lambda (durable fan-out)
// SNS delivers to SQS queue → SQS holds message until successful deletion
// Lambda ESM on SQS: DLQ captures messages after maxReceiveCount
// Zero message loss path

// CDK: SNS topic → multiple SQS queues with message filtering
import { Topic } from 'aws-cdk-lib/aws-sns'
import { SqsSubscription } from 'aws-cdk-lib/aws-sns-subscriptions'
import { SubscriptionFilter } from 'aws-cdk-lib/aws-sns'

const orderEventsTopic = new Topic(this, 'OrderEvents')

// Fulfillment queue: receives only CONFIRMED and PAID order events
const fulfillmentQueue = new Queue(this, 'FulfillmentQueue', {
  visibilityTimeout: Duration.seconds(300),
  deadLetterQueue: { queue: fulfillmentDlq, maxReceiveCount: 3 },
})

orderEventsTopic.addSubscription(new SqsSubscription(fulfillmentQueue, {
  filterPolicy: {
    eventType: SubscriptionFilter.stringFilter({
      allowlist: ['ORDER_CONFIRMED', 'ORDER_PAID']
    }),
    region: SubscriptionFilter.stringFilter({
      allowlist: ['us-east', 'us-west'] // Only US orders for this queue
    })
  }
}))

// Analytics queue: receives ALL order events for complete audit
const analyticsQueue = new Queue(this, 'AnalyticsQueue', {
  visibilityTimeout: Duration.seconds(60),
  deadLetterQueue: { queue: analyticsDlq, maxReceiveCount: 5 },
})

orderEventsTopic.addSubscription(new SqsSubscription(analyticsQueue))
// No filter — analytics receives everything

// Publisher
await sns.send(new PublishCommand({
  TopicArn: orderEventsTopic.topicArn,
  Message: JSON.stringify(orderPayload),
  MessageAttributes: {
    eventType: { DataType: 'String', StringValue: 'ORDER_CONFIRMED' },
    region: { DataType: 'String', StringValue: 'us-east' },
  }
}))

// ❌ Short polling (default WaitTimeSeconds=0)
// Lambda ESM or custom consumer polls → might return empty (no messages)
// Empty polls billed at the same rate as populated polls
// At 1 poll/second × 86,400s/day = 86,400 API calls/day from idle waiting
// Cost: $0.40/million × 86,400 ≈ $0.034/day per queue — multiplied across many queues

// ✅ Long polling (WaitTimeSeconds=20) — always use this
// SQS waits up to 20 seconds for a message before returning an empty response
// Reduces empty polls by ~20× on low-traffic queues
// Cost at 1/20s poll rate: 4,320 API calls/day — 95% cost reduction

// CloudWatch alarm: DLQ depth > 0 triggers PagerDuty/Slack
new Alarm(this, 'OrderDLQDepthAlarm', {
  metric: dlq.metricApproximateNumberOfMessagesVisible({
    period: Duration.minutes(1),
    statistic: 'Maximum',
  }),
  threshold: 0,
  evaluationPeriods: 1,
  comparisonOperator: ComparisonOperator.GREATER_THAN_THRESHOLD,
  alarmDescription: 'Orders are failing processing — DLQ has messages requiring investigation',
  actionsEnabled: true,
})

// DLQ redrive: after fixing the bug, replay DLQ messages to the source queue
// SQS console: Start message move task
// CLI:
await sqs.send(new StartMessageMoveTaskCommand({
  SourceArn: dlq.queueArn,
  DestinationArn: queue.queueArn,
  MaxNumberOfMessagesPerSecond: 100, // Throttle replay to avoid overloading downstream
}))