Worker asíncrono con escala a cero y 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 — 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
  }
}