📬 AWS SQS DLQ & Redrive

# 1. DLQ用のキューを作成
aws sqs create-queue \
  --queue-name my-app-dlq \
  --attributes MessageRetentionPeriod=1209600  # 14日間

# 2. DLQのARNを取得
aws sqs get-queue-attributes \
  --queue-url https://sqs.ap-northeast-1.amazonaws.com/123456789012/my-app-dlq \
  --attribute-names QueueArn

# 3. メインキューにRedrive Policyを設定
aws sqs set-queue-attributes \
  --queue-url https://sqs.ap-northeast-1.amazonaws.com/123456789012/my-main-queue \
  --attributes '{
    "RedrivePolicy": "{
      \"deadLetterTargetArn\": \"arn:aws:sqs:ap-northeast-1:123456789012:my-app-dlq\",
      \"maxReceiveCount\": \"3\"
    }",
    "VisibilityTimeout": "300"
  }'

# DLQ用のキュー
resource "aws_sqs_queue" "dlq" {
  name                      = "my-app-dlq"
  message_retention_seconds = 1209600  # 14日
  
  tags = {
    Name        = "My App DLQ"
    Environment = "production"
  }
}

# メインキュー（Redrive Policy設定済み）
resource "aws_sqs_queue" "main" {
  name                      = "my-main-queue"
  visibility_timeout_seconds = 300      # 5分
  message_retention_seconds = 345600   # 4日
  
  # 🔄 Redrive Policy設定
  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.dlq.arn
    maxReceiveCount     = 3
  })
  
  tags = {
    Name        = "My Main Queue"
    Environment = "production"
  }
}

# 💀 DLQからメインキューへのRedrive設定（オプション）
resource "aws_sqs_queue_redrive_allow_policy" "dlq_redrive" {
  queue_url = aws_sqs_queue.dlq.id
  
  redrive_allow_policy = jsonencode({
    redrivePermission = "byQueue"
    sourceQueueArns   = [aws_sqs_queue.main.arn]
  })
}

import boto3
import json
import logging
from time import sleep

logger = logging.getLogger()
logger.setLevel(logging.INFO)

sqs = boto3.client('sqs', region_name='ap-northeast-1')
QUEUE_URL = 'https://sqs.ap-northeast-1.amazonaws.com/123456789012/my-main-queue'

def process_message(message_body):
    """ビジネスロジック：ここで実際の処理を実行"""
    # 例：外部APIを呼び出す、DBに書き込むなど
    data = json.loads(message_body)
    
    # 何らかの処理...
    if data.get('invalid'):
        raise ValueError("Invalid data format")
    
    return True

def consume_messages():
    """メッセージを受信して処理する"""
    while True:
        # メッセージを受信（Long Polling: 20秒）
        response = sqs.receive_message(
            QueueUrl=QUEUE_URL,
            MaxNumberOfMessages=10,  # 一度に10件まで
            WaitTimeSeconds=20,       # Long Polling
            AttributeNames=['ApproximateReceiveCount']
        )
        
        messages = response.get('Messages', [])
        
        if not messages:
            logger.info("No messages, waiting...")
            continue
        
        for message in messages:
            receipt_handle = message['ReceiptHandle']
            message_id = message['MessageId']
            body = message['Body']
            receive_count = int(message['Attributes'].get('ApproximateReceiveCount', 0))
            
            logger.info(f"Processing message {message_id} (ReceiveCount: {receive_count})")
            
            try:
                # ✅ ビジネスロジック実行
                process_message(body)
                
                # ✅ 成功したらメッセージを削除
                sqs.delete_message(
                    QueueUrl=QUEUE_URL,
                    ReceiptHandle=receipt_handle
                )
                logger.info(f"Successfully processed message {message_id}")
                
            except Exception as e:
                # ❌ エラー発生時はDeleteしない → 自動的にキューに戻る
                logger.error(f"Failed to process message {message_id}: {str(e)}")
                logger.error(f"Receive count: {receive_count}. Will retry or move to DLQ.")
                
                # ⚠️ DeleteMessageを呼ばないことで：
                # 1. Visibility Timeout後に再度処理可能になる
                # 2. ReceiveCountが+1される
                # 3. maxReceiveCountに達したら自動的にDLQへ
                continue

if __name__ == "__main__":
    logger.info("Starting SQS Consumer...")
    consume_messages()