Scale-to-zero async worker with dead-letter

# Worker asíncrono con escala a cero — AWS
#
# Lambda disparada por SQS con respuestas parciales de lote, DLQ con retención
# mayor que la cola de origen y resultados en S3. Sin claves: el execution role
# de la función es la identidad.

data "aws_partition" "current" {}

locals {
  # AWS recomienda visibility timeout >= 6× el timeout de la función.
  visibility_timeout_seconds = var.function_timeout_seconds * 6
  results_prefix             = "results/"
}

# --- Colas ----------------------------------------------------------------------

resource "aws_sqs_queue" "dlq" {
  name                    = "${var.name}-jobs-dlq"
  sqs_managed_sse_enabled = true
  # Máximo (14 días): en colas Standard el vencimiento cuenta desde que el mensaje
  # entró a la cola original (gotcha: dlq-retention-original-timestamp).
  message_retention_seconds = 1209600
}

resource "aws_sqs_queue" "jobs" {
  name                       = "${var.name}-jobs"
  sqs_managed_sse_enabled    = true
  visibility_timeout_seconds = local.visibility_timeout_seconds
  message_retention_seconds  = 345600
  receive_wait_time_seconds  = 20

  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.dlq.arn
    maxReceiveCount     = var.max_receive_count
  })
}

# Solo la cola de trabajos puede usar esta DLQ.
resource "aws_sqs_queue_redrive_allow_policy" "dlq" {
  queue_url = aws_sqs_queue.dlq.id
  redrive_allow_policy = jsonencode({
    redrivePermission = "byQueue"
    sourceQueueArns   = [aws_sqs_queue.jobs.arn]
  })
}

# --- Bucket de resultados -------------------------------------------------------

resource "aws_s3_bucket" "results" {
  #checkov:skip=CKV_AWS_21:Resultados regenerables con TTL; versioning duplicaría costo (ver patrón object-storage-ttl).
  #checkov:skip=CKV_AWS_144:Replicación cross-region no aplica a resultados regenerables.
  #checkov:skip=CKV_AWS_18:Access logging fuera de alcance; usar CloudTrail data events si se requiere.
  #checkov:skip=CKV2_AWS_62:Notificaciones de eventos fuera de alcance del patrón.
  bucket = var.results_bucket_name
}

resource "aws_s3_bucket_public_access_block" "results" {
  bucket                  = aws_s3_bucket.results.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

resource "aws_s3_bucket_ownership_controls" "results" {
  bucket = aws_s3_bucket.results.id
  rule {
    object_ownership = "BucketOwnerEnforced"
  }
}

resource "aws_s3_bucket_server_side_encryption_configuration" "results" {
  #checkov:skip=CKV2_AWS_67:Clave administrada por AWS (aws/s3).
  bucket = aws_s3_bucket.results.id
  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm = "aws:kms"
    }
    bucket_key_enabled = true
  }
}

resource "aws_s3_bucket_lifecycle_configuration" "results" {
  bucket = aws_s3_bucket.results.id

  rule {
    id     = "expire-results"
    status = "Enabled"
    filter {
      prefix = local.results_prefix
    }
    expiration {
      days = var.results_ttl_days
    }
  }

  rule {
    id     = "abort-incomplete-multipart"
    status = "Enabled"
    filter {}
    abort_incomplete_multipart_upload {
      days_after_initiation = 1
    }
  }
}

# --- Identidad de la función ----------------------------------------------------

resource "aws_iam_role" "worker" {
  name = "${var.name}-worker"
  assume_role_policy = jsonencode({
    Version   = "2012-10-17"
    Statement = [{ Effect = "Allow", Principal = { Service = "lambda.amazonaws.com" }, Action = "sts:AssumeRole" }]
  })
}

resource "aws_iam_role_policy_attachment" "worker_managed" {
  for_each = toset([
    "service-role/AWSLambdaBasicExecutionRole",
    "AWSXRayDaemonWriteAccess",
  ])
  role       = aws_iam_role.worker.name
  policy_arn = "arn:${data.aws_partition.current.partition}:iam::aws:policy/${each.value}"
}

resource "aws_iam_role_policy" "worker" {
  name = "consume-jobs-write-results"
  role = aws_iam_role.worker.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect   = "Allow"
        Action   = ["sqs:ReceiveMessage", "sqs:DeleteMessage", "sqs:GetQueueAttributes", "sqs:ChangeMessageVisibility"]
        Resource = aws_sqs_queue.jobs.arn
      },
      {
        Effect   = "Allow"
        Action   = ["s3:PutObject"]
        Resource = "${aws_s3_bucket.results.arn}/${local.results_prefix}*"
      },
    ]
  })
}

# --- Lambda + event source mapping ----------------------------------------------

resource "aws_cloudwatch_log_group" "worker" {
  #checkov:skip=CKV_AWS_158:Cifrado con la clave administrada por CloudWatch; CMK opcional.
  name              = "/aws/lambda/${var.name}-worker"
  retention_in_days = 365
}

resource "aws_lambda_function" "worker" {
  #checkov:skip=CKV_AWS_115:La concurrencia se limita en el event source mapping (scaling_config), no con reserved concurrency (gotcha: max-concurrency-protects-downstream).
  #checkov:skip=CKV_AWS_116:Invocación síncrona desde SQS: la DLQ vive en la cola, no en la función.
  #checkov:skip=CKV_AWS_117:Sin VPC: SQS y S3 se acceden por sus endpoints de AWS; ver private-web-api para red privada.
  #checkov:skip=CKV_AWS_173:Variables de entorno sin secretos; cifradas con la clave administrada por Lambda.
  #checkov:skip=CKV_AWS_272:Code signing no aplica a funciones empaquetadas como imagen.
  function_name = "${var.name}-worker"
  role          = aws_iam_role.worker.arn
  package_type  = "Image"
  image_uri     = var.image_uri
  timeout       = var.function_timeout_seconds
  memory_size   = 512

  tracing_config {
    mode = "Active"
  }

  environment {
    variables = {
      RESULTS_BUCKET = aws_s3_bucket.results.bucket
      RESULTS_PREFIX = local.results_prefix
    }
  }

  depends_on = [aws_cloudwatch_log_group.worker, aws_iam_role_policy_attachment.worker_managed]
}

resource "aws_lambda_event_source_mapping" "jobs" {
  event_source_arn                   = aws_sqs_queue.jobs.arn
  function_name                      = aws_lambda_function.worker.arn
  batch_size                         = var.batch_size
  maximum_batching_window_in_seconds = 5

  # Solo los mensajes que fallaron vuelven a la cola (gotcha: partial-batch-failures).
  function_response_types = ["ReportBatchItemFailures"]

  scaling_config {
    maximum_concurrency = var.max_concurrency
  }
}

variable "name" {
  description = "Prefijo para nombrar los recursos (ej. \"worker-demo\")."
  type        = string

  validation {
    condition     = can(regex("^[a-z0-9-]{3,20}$", var.name))
    error_message = "name: 3-20 caracteres, minúsculas, números o guiones."
  }
}

variable "region" {
  description = "Región de AWS."
  type        = string
  default     = "us-east-1"
}

variable "results_bucket_name" {
  description = "Nombre global único del bucket de resultados."
  type        = string

  validation {
    condition     = can(regex("^[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]$", var.results_bucket_name))
    error_message = "results_bucket_name: 3-63 caracteres, minúsculas, números, puntos o guiones."
  }
}

variable "image_uri" {
  description = "Imagen de la función en ECR."
  type        = string
}

variable "function_timeout_seconds" {
  description = "Timeout de la Lambda. El visibility timeout de la cola se fija en 6× este valor (gotcha: visibility-timeout-6x)."
  type        = number
  default     = 60

  validation {
    condition     = var.function_timeout_seconds >= 1 && var.function_timeout_seconds <= 900
    error_message = "function_timeout_seconds: entre 1 y 900 (límite de Lambda)."
  }
}

variable "max_receive_count" {
  description = "Entregas antes de mandar el mensaje a la DLQ."
  type        = number
  default     = 5
}

variable "batch_size" {
  description = "Mensajes por invocación."
  type        = number
  default     = 10
}

variable "max_concurrency" {
  description = "Invocaciones concurrentes máximas disparadas por la cola (gotcha: max-concurrency-protects-downstream)."
  type        = number
  default     = 10

  validation {
    condition     = var.max_concurrency >= 2
    error_message = "maximum_concurrency de un event source mapping de SQS es como mínimo 2."
  }
}

variable "results_ttl_days" {
  description = "Días que se conservan los resultados."
  type        = number
  default     = 30
}

variable "tags" {
  description = "Tags aplicados a todos los recursos."
  type        = map(string)
  default     = {}
}

output "queue_url" {
  description = "URL de la cola de trabajos (los productores necesitan sqs:SendMessage)."
  value       = aws_sqs_queue.jobs.id
}

output "dlq_url" {
  description = "URL de la dead-letter queue."
  value       = aws_sqs_queue.dlq.id
}

output "results_bucket" {
  description = "Bucket de resultados."
  value       = aws_s3_bucket.results.bucket
}

terraform {
  required_version = ">= 1.7.0"

  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 6.67"
    }
  }
}

provider "aws" {
  region = var.region

  default_tags {
    tags = var.tags
  }
}

# Worker asíncrono con escala a cero — Azure
#
# Container Apps Job disparado por KEDA según el largo de una cola de
# Service Bus Standard, con identidad administrada para la cola, el scaler y
# el storage. Sin connection strings: el namespace no acepta SAS.

resource "azurerm_resource_group" "this" {
  name     = "rg-${var.name}"
  location = var.location
  tags     = var.tags
}

# --- Service Bus --------------------------------------------------------------

resource "azurerm_servicebus_namespace" "this" {
  #checkov:skip=CKV_AZURE_204:Standard no soporta Private Endpoint ni reglas de VNet; requiere Premium (gotcha: sb-private-endpoint-premium).
  #checkov:skip=CKV_AZURE_199:Doble cifrado solo disponible en Premium.
  #checkov:skip=CKV_AZURE_201:CMK solo disponible en Premium.
  #checkov:skip=CKV_AZURE_202:La identidad del namespace solo sirve para CMK (Premium); el acceso usa la identidad del worker.
  name                = var.servicebus_namespace_name
  resource_group_name = azurerm_resource_group.this.name
  location            = azurerm_resource_group.this.location
  sku                 = "Standard"
  local_auth_enabled  = false
  minimum_tls_version = "1.2"
  tags                = var.tags
}

resource "azurerm_servicebus_queue" "jobs" {
  name                                 = "jobs"
  namespace_id                         = azurerm_servicebus_namespace.this.id
  lock_duration                        = "PT1M"
  max_delivery_count                   = var.max_delivery_count
  default_message_ttl                  = "P14D"
  dead_lettering_on_message_expiration = true
}

# --- Storage de resultados ----------------------------------------------------

resource "azurerm_storage_account" "results" {
  #checkov:skip=CKV_AZURE_59:Acceso público de red con auth solo Entra ID: el job de Container Apps sin VNet no tiene IP de salida fija.
  #checkov:skip=CKV_AZURE_35:Ídem: sin VNet no hay subnet ni IP estable para reglas de red.
  #checkov:skip=CKV2_AZURE_33:Private Endpoint fuera de alcance de este patrón (ver private-web-api).
  #checkov:skip=CKV2_AZURE_1:CMK fuera de alcance del patrón.
  #checkov:skip=CKV_AZURE_206:ZRS suficiente para resultados regenerables.
  #checkov:skip=CKV_AZURE_33:No se usa Queue Storage.
  name                              = var.storage_account_name
  resource_group_name               = azurerm_resource_group.this.name
  location                          = azurerm_resource_group.this.location
  account_kind                      = "StorageV2"
  account_tier                      = "Standard"
  account_replication_type          = "ZRS"
  shared_access_key_enabled         = false
  default_to_oauth_authentication   = true
  local_user_enabled                = false
  allow_nested_items_to_be_public   = false
  cross_tenant_replication_enabled  = false
  https_traffic_only_enabled        = true
  min_tls_version                   = "TLS1_2"
  infrastructure_encryption_enabled = true

  blob_properties {
    delete_retention_policy {
      days = 7
    }
    container_delete_retention_policy {
      days = 7
    }
  }

  tags = var.tags
}

resource "azurerm_storage_container" "results" {
  #checkov:skip=CKV2_AZURE_21:Logging de lectura vía diagnostic settings, fuera de alcance del patrón.
  name                  = "results"
  storage_account_id    = azurerm_storage_account.results.id
  container_access_type = "private"
}

# --- Identidad del worker (contenedor + scaler) -------------------------------

resource "azurerm_user_assigned_identity" "worker" {
  name                = "id-${var.name}-worker"
  resource_group_name = azurerm_resource_group.this.name
  location            = azurerm_resource_group.this.location
  tags                = var.tags
}

# Receiver alcanza para leer, completar y para que KEDA consulte el largo de la cola.
resource "azurerm_role_assignment" "worker_queue" {
  scope                = azurerm_servicebus_queue.jobs.id
  role_definition_name = "Azure Service Bus Data Receiver"
  principal_id         = azurerm_user_assigned_identity.worker.principal_id
  principal_type       = "ServicePrincipal"
}

resource "azurerm_role_assignment" "worker_results" {
  scope                = azurerm_storage_container.results.id
  role_definition_name = "Storage Blob Data Contributor"
  principal_id         = azurerm_user_assigned_identity.worker.principal_id
  principal_type       = "ServicePrincipal"
}

# --- Container Apps Job (event-driven) ----------------------------------------

resource "azurerm_container_app_environment" "this" {
  name                = "cae-${var.name}"
  resource_group_name = azurerm_resource_group.this.name
  location            = azurerm_resource_group.this.location

  workload_profile {
    name                  = "Consumption"
    workload_profile_type = "Consumption"
  }

  tags = var.tags
}

resource "azurerm_container_app_job" "worker" {
  name                         = "caj-${var.name}-worker"
  resource_group_name          = azurerm_resource_group.this.name
  location                     = azurerm_resource_group.this.location
  container_app_environment_id = azurerm_container_app_environment.this.id
  workload_profile_name        = "Consumption"
  replica_timeout_in_seconds   = var.replica_timeout_seconds
  replica_retry_limit          = 1

  identity {
    type         = "UserAssigned"
    identity_ids = [azurerm_user_assigned_identity.worker.id]
  }

  event_trigger_config {
    parallelism              = 1
    replica_completion_count = 1

    scale {
      min_executions              = 0
      max_executions              = var.max_executions
      polling_interval_in_seconds = 30

      rules {
        name             = "servicebus-jobs"
        custom_rule_type = "azure-servicebus"
        # El scaler se autentica con la identidad, no con un connection string
        # (gotcha: keda-scaler-identity).
        identity_id = azurerm_user_assigned_identity.worker.id
        metadata = {
          namespace    = azurerm_servicebus_namespace.this.name
          queueName    = azurerm_servicebus_queue.jobs.name
          messageCount = tostring(var.messages_per_execution)
        }
      }
    }
  }

  template {
    container {
      name   = "worker"
      image  = var.container_image
      cpu    = 0.25
      memory = "0.5Gi"

      env {
        name  = "AZURE_CLIENT_ID"
        value = azurerm_user_assigned_identity.worker.client_id
      }
      env {
        name  = "SERVICEBUS_FQDN"
        value = "${azurerm_servicebus_namespace.this.name}.servicebus.windows.net"
      }
      env {
        name  = "SERVICEBUS_QUEUE"
        value = azurerm_servicebus_queue.jobs.name
      }
      env {
        name  = "RESULTS_CONTAINER_URL"
        value = "${azurerm_storage_account.results.primary_blob_endpoint}${azurerm_storage_container.results.name}"
      }
    }
  }

  tags = var.tags

  depends_on = [azurerm_role_assignment.worker_queue, azurerm_role_assignment.worker_results]
}

variable "name" {
  description = "Prefijo para nombrar los recursos (ej. \"worker-demo\")."
  type        = string

  validation {
    condition     = can(regex("^[a-z0-9-]{3,20}$", var.name))
    error_message = "name: 3-20 caracteres, minúsculas, números o guiones."
  }
}

variable "location" {
  description = "Región de Azure."
  type        = string
  default     = "eastus2"
}

variable "servicebus_namespace_name" {
  description = "Nombre global único del namespace de Service Bus."
  type        = string

  validation {
    condition     = can(regex("^[a-zA-Z][a-zA-Z0-9-]{4,48}[a-zA-Z0-9]$", var.servicebus_namespace_name))
    error_message = "servicebus_namespace_name: 6-50 caracteres, empieza con letra."
  }
}

variable "storage_account_name" {
  description = "Nombre global único de la Storage Account de resultados."
  type        = string

  validation {
    condition     = can(regex("^[a-z0-9]{3,24}$", var.storage_account_name))
    error_message = "storage_account_name: 3-24 caracteres, solo minúsculas y números."
  }
}

variable "max_delivery_count" {
  description = "Entregas antes de mandar el mensaje a $DeadLetterQueue."
  type        = number
  default     = 10

  validation {
    condition     = var.max_delivery_count >= 1
    error_message = "max_delivery_count debe ser >= 1."
  }
}

variable "messages_per_execution" {
  description = "Mensajes en cola por ejecución del job (messageCount del scaler KEDA)."
  type        = number
  default     = 5
}

variable "max_executions" {
  description = "Máximo de ejecuciones concurrentes del job."
  type        = number
  default     = 10
}

variable "replica_timeout_seconds" {
  description = "Tiempo máximo de cada ejecución."
  type        = number
  default     = 300
}

variable "container_image" {
  description = "Imagen del worker: procesa un lote de mensajes y termina."
  type        = string
  default     = "mcr.microsoft.com/k8se/quickstart-jobs:latest"
}

variable "tags" {
  description = "Tags aplicados a todos los recursos."
  type        = map(string)
  default     = {}
}

output "servicebus_fqdn" {
  description = "FQDN del namespace (los productores se autentican con Entra ID)."
  value       = "${azurerm_servicebus_namespace.this.name}.servicebus.windows.net"
}

output "queue_name" {
  description = "Cola de trabajos. La dead-letter es <cola>/$DeadLetterQueue."
  value       = azurerm_servicebus_queue.jobs.name
}

output "worker_identity_client_id" {
  description = "Client ID de la identidad del worker."
  value       = azurerm_user_assigned_identity.worker.client_id
}

terraform {
  required_version = ">= 1.7.0"

  required_providers {
    azurerm = {
      source  = "hashicorp/azurerm"
      version = "~> 5.8"
    }
  }
}

provider "azurerm" {
  features {}

  # Con shared_access_key_enabled = false el provider tiene que usar Entra ID
  # para las llamadas de data plane (gotcha: shared-key-needs-azuread-provider).
  storage_use_azuread = true

  # subscription_id se toma de ARM_SUBSCRIPTION_ID.
}