SQS - Simple Queue Service

import * as aws from "@pulumi/aws";

// Create a standard queue
const queue = new aws.sqs.Queue("tasks", {
    visibilityTimeoutSeconds: 300,
    messageRetentionSeconds: 345600, // 4 days
});

// Create a FIFO queue for ordered processing
const fifoQueue = new aws.sqs.Queue("orders", {
    fifoQueue: true,
    contentBasedDeduplication: true,
});

// Create queue with dead-letter queue
const dlq = new aws.sqs.Queue("dlq");
const mainQueue = new aws.sqs.Queue("main", {
    redrivePolicy: pulumi.interpolate`{
        "deadLetterTargetArn": "${dlq.arn}",
        "maxReceiveCount": 3
    }`,
});

class Queue extends pulumi.CustomResource {
    constructor(name: string, args?: QueueArgs, opts?: pulumi.CustomResourceOptions);

    readonly arn: pulumi.Output<string>;
    readonly url: pulumi.Output<string>;
}

interface QueueArgs {
    name?: pulumi.Input<string>;
    fifoQueue?: pulumi.Input<boolean>;
    contentBasedDeduplication?: pulumi.Input<boolean>;
    delaySeconds?: pulumi.Input<number>;
    maxMessageSize?: pulumi.Input<number>;
    messageRetentionSeconds?: pulumi.Input<number>;
    receiveWaitTimeSeconds?: pulumi.Input<number>;
    visibilityTimeoutSeconds?: pulumi.Input<number>;
    redrivePolicy?: pulumi.Input<string>;
    kmsMasterKeyId?: pulumi.Input<string>;
    tags?: pulumi.Input<{[key: string]: pulumi.Input<string>}>;
}

const processingQueue = new aws.sqs.Queue("processing", {
    name: "task-processing",

    // Message timing
    delaySeconds: 0, // No delivery delay
    visibilityTimeoutSeconds: 300, // 5 minutes to process
    messageRetentionSeconds: 345600, // Keep for 4 days

    // Performance
    receiveWaitTimeSeconds: 20, // Long polling
    maxMessageSize: 262144, // 256 KB max

    // Encryption
    kmsMasterKeyId: queueKey.id,

    tags: {
        Environment: "production",
        Component: "task-processor",
    },
});

// Lambda to process messages
const processor = new aws.lambda.Function("processor", {
    // ... function config
});

new aws.lambda.EventSourceMapping("sqs-trigger", {
    eventSourceArn: processingQueue.arn,
    functionName: processor.name,
    batchSize: 10,
    maximumBatchingWindowInSeconds: 5,
});

const orderQueue = new aws.sqs.Queue("orders", {
    name: "order-processing.fifo",
    fifoQueue: true,
    contentBasedDeduplication: true,

    // FIFO queues support up to 300 messages/second
    // Use message group ID for ordering within groups
});

// Application code example:
// sqs.sendMessage({
//   QueueUrl: orderQueue.url,
//   MessageBody: JSON.stringify(order),
//   MessageGroupId: order.customerId, // Orders per customer are ordered
//   MessageDeduplicationId: order.id, // Or use contentBasedDeduplication
// });

const dlq = new aws.sqs.Queue("dlq", {
    name: "my-dlq",
    messageRetentionSeconds: 1209600, // 14 days - longer retention for debugging
});

const queue = new aws.sqs.Queue("main", {
    name: "my-queue",
    delaySeconds: 0,
    maxMessageSize: 262144,
    messageRetentionSeconds: 345600, // 4 days
    receiveWaitTimeSeconds: 10,
    visibilityTimeoutSeconds: 30,
    redrivePolicy: pulumi.interpolate`{
        "deadLetterTargetArn": "${dlq.arn}",
        "maxReceiveCount": 3
    }`,
});

export const queueUrl = queue.url;
export const queueArn = queue.arn;

class QueuePolicy extends pulumi.CustomResource {
    constructor(name: string, args: QueuePolicyArgs, opts?: pulumi.CustomResourceOptions);
}

interface QueuePolicyArgs {
    queueUrl: pulumi.Input<string>;
    policy: pulumi.Input<string>;
}

const queue = new aws.sqs.Queue("subscriber");
const topic = new aws.sns.Topic("events");

new aws.sqs.QueuePolicy("sns-publish", {
    queueUrl: queue.url,
    policy: pulumi.all([queue.arn, topic.arn]).apply(([queueArn, topicArn]) =>
        JSON.stringify({
            Version: "2012-10-17",
            Statement: [{
                Effect: "Allow",
                Principal: {
                    Service: "sns.amazonaws.com",
                },
                Action: "sqs:SendMessage",
                Resource: queueArn,
                Condition: {
                    ArnEquals: {
                        "aws:SourceArn": topicArn,
                    },
                },
            }],
        })
    ),
});

new aws.sns.TopicSubscription("queue-sub", {
    topic: topic.arn,
    protocol: "sqs",
    endpoint: queue.arn,
});

class RedrivePolicy extends pulumi.CustomResource {
    constructor(name: string, args: RedrivePolicyArgs, opts?: pulumi.CustomResourceOptions);
}

interface RedrivePolicyArgs {
    queueUrl: pulumi.Input<string>;
    redrivePolicy: pulumi.Input<string>;
}

class RedriveAllowPolicy extends pulumi.CustomResource {
    constructor(name: string, args: RedriveAllowPolicyArgs, opts?: pulumi.CustomResourceOptions);
}

interface RedriveAllowPolicyArgs {
    queueUrl: pulumi.Input<string>;
    redriveAllowPolicy: pulumi.Input<string>;
}

const dlq = new aws.sqs.Queue("shared-dlq");

new aws.sqs.RedriveAllowPolicy("restrict-dlq", {
    queueUrl: dlq.url,
    redriveAllowPolicy: pulumi.all([queue1.arn, queue2.arn]).apply(([arn1, arn2]) =>
        JSON.stringify({
            redrivePermission: "byQueue",
            sourceQueueArns: [arn1, arn2],
        })
    ),
});

// Dead-letter queue for failed messages
const dlq = new aws.sqs.Queue("processing-dlq", {
    messageRetentionSeconds: 1209600, // 14 days
});

// Main processing queue
const queue = new aws.sqs.Queue("processing", {
    visibilityTimeoutSeconds: 300,
    receiveWaitTimeSeconds: 20, // Long polling
    redrivePolicy: pulumi.interpolate`{
        "deadLetterTargetArn": "${dlq.arn}",
        "maxReceiveCount": 3
    }`,
});

// Lambda processor
const processor = new aws.lambda.Function("processor", {
    handler: "index.handler",
    runtime: "nodejs20.x",
    timeout: 300,
    reservedConcurrentExecutions: 10,
    code: new pulumi.asset.AssetArchive({
        ".": new pulumi.asset.FileArchive("./processor"),
    }),
});

// Connect queue to Lambda
new aws.lambda.EventSourceMapping("queue-trigger", {
    eventSourceArn: queue.arn,
    functionName: processor.name,
    batchSize: 10,
    maximumBatchingWindowInSeconds: 5,
});

// Allow Lambda to read from queue
const lambdaRole = new aws.iam.Role("processor-role", {
    assumeRolePolicy: JSON.stringify({
        Version: "2012-10-17",
        Statement: [{
            Action: "sts:AssumeRole",
            Principal: { Service: "lambda.amazonaws.com" },
            Effect: "Allow",
        }],
    }),
});

new aws.iam.RolePolicy("queue-access", {
    role: lambdaRole.id,
    policy: pulumi.all([queue.arn, dlq.arn]).apply(([queueArn, dlqArn]) =>
        JSON.stringify({
            Version: "2012-10-17",
            Statement: [{
                Effect: "Allow",
                Action: [
                    "sqs:ReceiveMessage",
                    "sqs:DeleteMessage",
                    "sqs:GetQueueAttributes",
                ],
                Resource: [queueArn, dlqArn],
            }],
        })
    ),
});