Scale-to-zero async worker with dead-letter

# Worker asíncrono con escala a cero — GCP
#
# Pub/Sub push autenticado (OIDC) hacia Cloud Run con ingress interno,
# dead-letter topic con su suscripción y permisos del service agent, y
# resultados en Cloud Storage.

data "google_project" "this" {
  project_id = var.project_id
}

locals {
  pubsub_service_agent = "serviceAccount:service-${data.google_project.this.number}@gcp-sa-pubsub.iam.gserviceaccount.com"
}

# --- Identidades -----------------------------------------------------------------

resource "google_service_account" "worker" {
  account_id   = "${var.name}-worker"
  display_name = "Worker (${var.name})"
}

# Identidad con la que Pub/Sub firma el OIDC token del push.
resource "google_service_account" "push" {
  account_id   = "${var.name}-push"
  display_name = "Pub/Sub push invoker (${var.name})"
}

# --- Resultados ------------------------------------------------------------------

resource "google_storage_bucket" "results" {
  #checkov:skip=CKV_GCP_62:Access logs fuera de alcance; usar Cloud Audit Logs (Data Access) si se requiere.
  #checkov:skip=CKV_GCP_78:Resultados regenerables con TTL; versioning duplicaría costo.
  name                        = var.results_bucket_name
  location                    = upper(var.region)
  storage_class               = "STANDARD"
  uniform_bucket_level_access = true
  public_access_prevention    = "enforced"
  labels                      = var.labels

  soft_delete_policy {
    retention_duration_seconds = 604800
  }

  lifecycle_rule {
    condition {
      age = var.results_ttl_days
    }
    action {
      type = "Delete"
    }
  }

  lifecycle_rule {
    condition {
      age = 1
    }
    action {
      type = "AbortIncompleteMultipartUpload"
    }
  }
}

resource "google_storage_bucket_iam_member" "worker_results" {
  bucket = google_storage_bucket.results.name
  role   = "roles/storage.objectCreator"
  member = google_service_account.worker.member
}

# --- Worker en Cloud Run -----------------------------------------------------------

resource "google_cloud_run_v2_service" "worker" {
  name     = "${var.name}-worker"
  location = var.region
  # Pub/Sub cuenta como origen interno: no hace falta exponer el servicio.
  ingress             = "INGRESS_TRAFFIC_INTERNAL_ONLY"
  deletion_protection = false
  labels              = var.labels

  template {
    service_account = google_service_account.worker.email
    timeout         = "${var.ack_deadline_seconds}s"

    scaling {
      min_instance_count = 0
      max_instance_count = var.max_instances
    }

    containers {
      image = var.container_image

      env {
        name  = "RESULTS_BUCKET"
        value = google_storage_bucket.results.name
      }
    }
  }
}

resource "google_cloud_run_v2_service_iam_member" "push_invoker" {
  name     = google_cloud_run_v2_service.worker.name
  location = google_cloud_run_v2_service.worker.location
  role     = "roles/run.invoker"
  member   = google_service_account.push.member
}

# --- Pub/Sub -----------------------------------------------------------------------

resource "google_pubsub_topic" "jobs" {
  #checkov:skip=CKV_GCP_83:Cifrado con claves administradas por Google; CMEK opcional.
  name   = "${var.name}-jobs"
  labels = var.labels
}

resource "google_pubsub_topic" "dead_letter" {
  #checkov:skip=CKV_GCP_83:Cifrado con claves administradas por Google; CMEK opcional.
  name   = "${var.name}-jobs-dead-letter"
  labels = var.labels
}

# Sin suscripción, lo que llega al dead-letter topic se pierde
# (gotcha: dead-letter-topic-needs-subscription).
resource "google_pubsub_subscription" "dead_letter" {
  name                       = "${var.name}-jobs-dead-letter"
  topic                      = google_pubsub_topic.dead_letter.id
  message_retention_duration = "604800s"
  ack_deadline_seconds       = 60
  labels                     = var.labels

  expiration_policy {
    ttl = ""
  }
}

resource "google_pubsub_subscription" "jobs" {
  name                 = "${var.name}-jobs-push"
  topic                = google_pubsub_topic.jobs.id
  ack_deadline_seconds = var.ack_deadline_seconds
  labels               = var.labels

  push_config {
    push_endpoint = google_cloud_run_v2_service.worker.uri

    oidc_token {
      service_account_email = google_service_account.push.email
    }
  }

  dead_letter_policy {
    dead_letter_topic     = google_pubsub_topic.dead_letter.id
    max_delivery_attempts = var.max_delivery_attempts
  }

  retry_policy {
    minimum_backoff = "10s"
    maximum_backoff = "600s"
  }

  expiration_policy {
    ttl = ""
  }

  depends_on = [google_pubsub_subscription.dead_letter]
}

# El service agent de Pub/Sub es quien mueve mensajes al dead-letter
# (gotcha: dlq-needs-service-agent-roles).
resource "google_pubsub_topic_iam_member" "dead_letter_publisher" {
  topic  = google_pubsub_topic.dead_letter.id
  role   = "roles/pubsub.publisher"
  member = local.pubsub_service_agent
}

resource "google_pubsub_subscription_iam_member" "source_subscriber" {
  subscription = google_pubsub_subscription.jobs.id
  role         = "roles/pubsub.subscriber"
  member       = local.pubsub_service_agent
}

variable "project_id" {
  description = "ID del proyecto de GCP."
  type        = string
}

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

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

variable "region" {
  description = "Región de GCP."
  type        = string
  default     = "us-central1"
}

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

variable "container_image" {
  description = "Imagen del worker (responde 2xx cuando procesó el mensaje)."
  type        = string
}

variable "ack_deadline_seconds" {
  description = "Ack deadline de la suscripción push; también es el timeout del servicio (gotcha: push-ack-deadline)."
  type        = number
  default     = 60

  validation {
    condition     = var.ack_deadline_seconds >= 10 && var.ack_deadline_seconds <= 600
    error_message = "ack_deadline_seconds: entre 10 y 600."
  }
}

variable "max_delivery_attempts" {
  description = "Intentos antes de reenviar al dead-letter topic (Pub/Sub: 5 a 100)."
  type        = number
  default     = 5

  validation {
    condition     = var.max_delivery_attempts >= 5 && var.max_delivery_attempts <= 100
    error_message = "max_delivery_attempts: entre 5 y 100."
  }
}

variable "max_instances" {
  description = "Máximo de instancias del worker."
  type        = number
  default     = 10
}

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

variable "labels" {
  description = "Labels aplicados a los recursos que los soportan."
  type        = map(string)
  default     = {}
}

output "jobs_topic" {
  description = "Topic al que publican los productores (necesitan roles/pubsub.publisher)."
  value       = google_pubsub_topic.jobs.id
}

output "dead_letter_subscription" {
  description = "Suscripción pull para inspeccionar y reprocesar mensajes fallidos."
  value       = google_pubsub_subscription.dead_letter.id
}

output "results_bucket" {
  description = "Bucket de resultados."
  value       = google_storage_bucket.results.url
}

terraform {
  required_version = ">= 1.7.0"

  required_providers {
    google = {
      source  = "hashicorp/google"
      version = "~> 8.6"
    }
  }
}

provider "google" {
  project = var.project_id
  region  = var.region
}

# 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.
}