Amazon SQS #

[dependencies]
aws-config = { version = "1", features = ["behavior-version-latest"] }
aws-sdk-sqs = "1"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

use aws_sdk_sqs::Client;

async fn buat_client() -> Client {
    // Reads configuration from the environment:
    // AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
    // or from ~/.aws/credentials when local
    let config = aws_config::load_from_env().await;
    Client::new(&config)
}

// For local development with LocalStack
async fn buat_client_lokal() -> Client {
    use aws_sdk_sqs::config::Builder;
    use aws_types::region::Region;

    let config = Builder::new()
        .endpoint_url("http://localhost:4566")  // LocalStack endpoint
        .region(Region::new("us-east-1"))
        .credentials_provider(aws_credential_types::provider::SharedCredentialsProvider::new(
            aws_credential_types::Credentials::new(
                "test", "test", None, None, "test"
            )
        ))
        .build();

    Client::from_conf(config)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = buat_client().await;

    // List existing queues
    let hasil = client.list_queues().send().await?;
    println!("Existing queues:");
    for url in hasil.queue_urls() {
        println!("  - {}", url);
    }

    Ok(())
}

use aws_sdk_sqs::{Client, types::QueueAttributeName};
use std::collections::HashMap;

async fn buat_standard_queue(
    client: &Client,
    nama: &str,
) -> Result<String, aws_sdk_sqs::Error> {
    let mut atribut = HashMap::new();
    // Visibility timeout: a message is invisible for N seconds after being received
    atribut.insert(QueueAttributeName::VisibilityTimeout, "30".to_string());
    // Message retention: how long messages are stored (default 4 days)
    atribut.insert(QueueAttributeName::MessageRetentionPeriod, "86400".to_string()); // 1 day
    // Long polling: wait up to N seconds for messages (cost saving)
    atribut.insert(QueueAttributeName::ReceiveMessageWaitTimeSeconds, "20".to_string());

    let hasil = client
        .create_queue()
        .queue_name(nama)
        .set_attributes(Some(atribut))
        .send()
        .await?;

    let url = hasil.queue_url().unwrap_or("").to_string();
    println!("Standard queue created: {}", url);
    Ok(url)
}

async fn buat_fifo_queue(
    client: &Client,
    nama: &str,  // must end in .fifo
) -> Result<String, aws_sdk_sqs::Error> {
    let mut atribut = HashMap::new();
    atribut.insert(QueueAttributeName::FifoQueue, "true".to_string());
    // Content-based deduplication: hash the message body as the dedup ID
    atribut.insert(QueueAttributeName::ContentBasedDeduplication, "true".to_string());
    atribut.insert(QueueAttributeName::VisibilityTimeout, "60".to_string());

    let hasil = client
        .create_queue()
        .queue_name(nama)  // e.g. "pesanan-prioritas.fifo"
        .set_attributes(Some(atribut))
        .send()
        .await?;

    let url = hasil.queue_url().unwrap_or("").to_string();
    println!("FIFO queue created: {}", url);
    Ok(url)
}

async fn buat_queue_dengan_dlq(
    client: &Client,
    nama_utama: &str,
    maks_receive: i32,
) -> Result<String, Box<dyn std::error::Error>> {
    // 1. Create the DLQ first
    let dlq = client
        .create_queue()
        .queue_name(&format!("{}-dlq", nama_utama))
        .send()
        .await?;
    let dlq_url = dlq.queue_url().unwrap_or("").to_string();

    // 2. Get the DLQ ARN
    let dlq_attrs = client
        .get_queue_attributes()
        .queue_url(&dlq_url)
        .attribute_names(QueueAttributeName::QueueArn)
        .send()
        .await?;
    let dlq_arn = dlq_attrs
        .attributes()
        .and_then(|a| a.get(&QueueAttributeName::QueueArn))
        .cloned()
        .unwrap_or_default();

    // 3. Create the main queue with a Redrive Policy to the DLQ
    let redrive_policy = serde_json::json!({
        "deadLetterTargetArn": dlq_arn,
        "maxReceiveCount": maks_receive  // move to the DLQ after N failures
    })
    .to_string();

    let mut atribut = HashMap::new();
    atribut.insert(
        QueueAttributeName::RedrivePolicy,
        redrive_policy,
    );
    atribut.insert(QueueAttributeName::VisibilityTimeout, "30".to_string());

    let hasil = client
        .create_queue()
        .queue_name(nama_utama)
        .set_attributes(Some(atribut))
        .send()
        .await?;

    let url = hasil.queue_url().unwrap_or("").to_string();
    println!("Queue '{}' with DLQ created successfully", nama_utama);
    Ok(url)
}

use aws_sdk_sqs::{Client, types::MessageAttributeValue};
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize, Clone)]
struct EventPemrosesan {
    pub id: u64,
    pub tipe: String,
    pub data: serde_json::Value,
    pub dibuat_pada: String,
}

async fn kirim_pesan(
    client: &Client,
    queue_url: &str,
    event: &EventPemrosesan,
) -> Result<String, aws_sdk_sqs::Error> {
    let body = serde_json::to_string(event).unwrap();

    let hasil = client
        .send_message()
        .queue_url(queue_url)
        .message_body(&body)
        // Delivery delay: the message is invisible for N seconds
        .delay_seconds(0)
        // Message attributes: extra metadata that can be filtered
        .message_attributes(
            "EventType",
            MessageAttributeValue::builder()
                .data_type("String")
                .string_value(&event.tipe)
                .build()
                .unwrap(),
        )
        .message_attributes(
            "Source",
            MessageAttributeValue::builder()
                .data_type("String")
                .string_value("api-server")
                .build()
                .unwrap(),
        )
        .send()
        .await?;

    let msg_id = hasil.message_id().unwrap_or("").to_string();
    println!("Message sent: {}", msg_id);
    Ok(msg_id)
}

// Send to a FIFO queue — needs MessageGroupId and MessageDeduplicationId
async fn kirim_ke_fifo(
    client: &Client,
    queue_url: &str,
    event: &EventPemrosesan,
    group_id: &str,  // all messages in the same group are processed in order
) -> Result<String, aws_sdk_sqs::Error> {
    let body = serde_json::to_string(event).unwrap();

    let hasil = client
        .send_message()
        .queue_url(queue_url)
        .message_body(&body)
        .message_group_id(group_id)
        // Deduplication ID: messages with the same ID within 5 minutes are ignored
        .message_deduplication_id(&format!("{}-{}", event.tipe, event.id))
        .send()
        .await?;

    Ok(hasil.message_id().unwrap_or("").to_string())
}

// Batch send — send up to 10 messages at once (cost saving)
async fn kirim_batch(
    client: &Client,
    queue_url: &str,
    events: &[EventPemrosesan],
) -> Result<usize, Box<dyn std::error::Error>> {
    use aws_sdk_sqs::types::SendMessageBatchRequestEntry;

    let entries: Vec<SendMessageBatchRequestEntry> = events
        .iter()
        .enumerate()
        .map(|(i, event)| {
            SendMessageBatchRequestEntry::builder()
                .id(i.to_string())
                .message_body(serde_json::to_string(event).unwrap())
                .build()
                .unwrap()
        })
        .collect();

    let hasil = client
        .send_message_batch()
        .queue_url(queue_url)
        .set_entries(Some(entries))
        .send()
        .await?;

    let berhasil = hasil.successful().len();
    let gagal = hasil.failed().len();

    if gagal > 0 {
        for f in hasil.failed() {
            eprintln!("Failed to send entry {}: {}", f.id(), f.message().unwrap_or("?"));
        }
    }

    println!("Batch: {}/{} succeeded", berhasil, events.len());
    Ok(berhasil)
}

use aws_sdk_sqs::{Client, types::Message};

async fn terima_dan_proses(
    client: &Client,
    queue_url: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    loop {
        // Long polling: wait up to 20 seconds if the queue is empty
        // More cost-effective than constant short polling
        let hasil = client
            .receive_message()
            .queue_url(queue_url)
            .max_number_of_messages(10)    // max 10 messages per request
            .wait_time_seconds(20)          // long polling
            .visibility_timeout(30)         // invisible to other consumers for 30s
            .message_attribute_names("All") // get all message attributes
            .send()
            .await?;

        let pesan_list = hasil.messages();

        if pesan_list.is_empty() {
            println!("Queue empty, waiting...");
            continue;
        }

        for pesan in pesan_list {
            match proses_satu_pesan(client, queue_url, pesan).await {
                Ok(_) => println!("Message processed successfully"),
                Err(e) => eprintln!("Error processing message: {}", e),
            }
        }
    }
}

async fn proses_satu_pesan(
    client: &Client,
    queue_url: &str,
    pesan: &Message,
) -> Result<(), Box<dyn std::error::Error>> {
    let receipt_handle = pesan.receipt_handle().unwrap_or("");
    let body = pesan.body().unwrap_or("{}");

    println!("Message ID: {}", pesan.message_id().unwrap_or("?"));

    // Read message attributes
    if let Some(attrs) = pesan.message_attributes() {
        if let Some(event_type) = attrs.get("EventType") {
            println!("EventType: {}", event_type.string_value().unwrap_or("?"));
        }
    }

    // Parse the body
    let event: EventPemrosesan = serde_json::from_str(body)?;
    println!("Processing event: {} #{}", event.tipe, event.id);

    // Process the event...
    jalankan_logika_bisnis(&event).await?;

    // Delete the message after successful processing
    // IMPORTANT: if not deleted, the message reappears after the visibility timeout
    client
        .delete_message()
        .queue_url(queue_url)
        .receipt_handle(receipt_handle)
        .send()
        .await?;

    println!("Message deleted from the queue");
    Ok(())
}

async fn jalankan_logika_bisnis(event: &EventPemrosesan) -> Result<(), String> {
    println!("Running logic for: {}", event.tipe);
    Ok(())
}

// Batch delete — saves API calls when successfully processing many messages
async fn hapus_batch(
    client: &Client,
    queue_url: &str,
    pesan_list: &[Message],
) -> Result<(), aws_sdk_sqs::Error> {
    use aws_sdk_sqs::types::DeleteMessageBatchRequestEntry;

    let entries: Vec<DeleteMessageBatchRequestEntry> = pesan_list
        .iter()
        .enumerate()
        .filter_map(|(i, m)| {
            m.receipt_handle().map(|rh| {
                DeleteMessageBatchRequestEntry::builder()
                    .id(i.to_string())
                    .receipt_handle(rh)
                    .build()
                    .unwrap()
            })
        })
        .collect();

    client
        .delete_message_batch()
        .queue_url(queue_url)
        .set_entries(Some(entries))
        .send()
        .await?;

    println!("{} messages deleted", pesan_list.len());
    Ok(())
}

use aws_sdk_sqs::Client;

// Extend the visibility timeout for messages that need longer processing
async fn perpanjang_visibility(
    client: &Client,
    queue_url: &str,
    receipt_handle: &str,
    tambahan_detik: i32,
) -> Result<(), aws_sdk_sqs::Error> {
    client
        .change_message_visibility()
        .queue_url(queue_url)
        .receipt_handle(receipt_handle)
        .visibility_timeout(tambahan_detik)
        .send()
        .await?;

    println!("Visibility timeout extended by {} seconds", tambahan_detik);
    Ok(())
}

// A worker that processes long messages and extends visibility periodically
async fn proses_pesan_panjang(
    client: std::sync::Arc<Client>,
    queue_url: String,
    pesan: Message,
) {
    let receipt_handle = pesan.receipt_handle().unwrap_or("").to_string();

    // Spawn a task to extend visibility every 25 seconds (before the 30s timeout)
    let client_clone = std::sync::Arc::clone(&client);
    let queue_clone = queue_url.clone();
    let handle_clone = receipt_handle.clone();

    let keepalive = tokio::spawn(async move {
        loop {
            tokio::time::sleep(std::time::Duration::from_secs(25)).await;
            let _ = perpanjang_visibility(&client_clone, &queue_clone, &handle_clone, 30).await;
        }
    });

    // Process the long-running message
    tokio::time::sleep(std::time::Duration::from_secs(60)).await; // simulate 60 seconds
    println!("Message processed");

    // Stop the keepalive
    keepalive.abort();

    // Delete the message
    let _ = client
        .delete_message()
        .queue_url(&queue_url)
        .receipt_handle(&receipt_handle)
        .send()
        .await;
}

# Terraform: SNS topic + SQS queue + subscription

resource "aws_sns_topic" "pesanan_events" {
  name = "pesanan-events"
}

resource "aws_sqs_queue" "pemrosesan" {
  name                       = "pemrosesan-pesanan"
  visibility_timeout_seconds = 30
  message_retention_seconds  = 86400
  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.pemrosesan_dlq.arn
    maxReceiveCount     = 3
  })
}

resource "aws_sns_topic_subscription" "pemrosesan_sub" {
  topic_arn = aws_sns_topic.pesanan_events.arn
  protocol  = "sqs"
  endpoint  = aws_sqs_queue.pemrosesan.arn
}

#[derive(Debug, serde::Deserialize)]
struct SnsSqsWrapper {
    #[serde(rename = "Type")]
    tipe: String,
    #[serde(rename = "Message")]
    pesan: String,  // JSON string from SNS
    #[serde(rename = "MessageId")]
    message_id: String,
    #[serde(rename = "TopicArn")]
    topic_arn: String,
}

async fn proses_pesan_dari_sns(body: &str) -> Result<(), Box<dyn std::error::Error>> {
    // Messages from SNS have a wrapper
    if let Ok(wrapper) = serde_json::from_str::<SnsSqsWrapper>(body) {
        if wrapper.tipe == "Notification" {
            // Unwrap: the real message is in the "Message" field
            let event: EventPemrosesan = serde_json::from_str(&wrapper.pesan)?;
            println!("Event from SNS via SQS: {} #{}", event.tipe, event.id);
        }
    } else {
        // Direct message (not from SNS)
        let event: EventPemrosesan = serde_json::from_str(body)?;
        println!("Direct event: {} #{}", event.tipe, event.id);
    }
    Ok(())
}