################
# Runner Service — SQS, S3, Secrets, ALB, Security Groups, IAM, Datadog
################
# The runner service executes scheduled data pipelines (Snowflake queries,
# sandbox transforms) triggered via SQS. It runs on Fargate behind an
# internal ALB exposing a ConnectRPC API on port 8082.
#
# All resources are gated by is_ephemeral — only deployed in the main environment.
# local.runner_service_name and local.runner_port are defined in main.tf.

################
# KMS — CMK for SQS + SNS encryption (Prisma Cloud requires non-default keys)
################

resource "aws_kms_key" "runner_messaging" {
  count                   = local.is_ephemeral ? 0 : 1
  description             = "CMK for ${local.runner_service_name} SQS queues and SNS topics"
  deletion_window_in_days = 14
  enable_key_rotation     = true

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Sid       = "AllowRootAccount"
        Effect    = "Allow"
        Principal = { AWS = "arn:aws:iam::${data.aws_caller_identity.current.account_id}:root" }
        Action    = "kms:*"
        Resource  = "*"
      },
      {
        Sid       = "AllowSQSAndSNS"
        Effect    = "Allow"
        Principal = { Service = ["sqs.amazonaws.com", "sns.amazonaws.com"] }
        Action    = ["kms:GenerateDataKey", "kms:Decrypt"]
        Resource  = "*"
      },
      {
        Sid       = "AllowCloudWatchAlarms"
        Effect    = "Allow"
        Principal = { Service = "cloudwatch.amazonaws.com" }
        Action    = ["kms:GenerateDataKey", "kms:Decrypt"]
        Resource  = "*"
      },
    ]
  })

  tags = {
    environment        = var.environment
    application_family = var.application_family
    service_name       = local.runner_service_name
  }
}

resource "aws_kms_alias" "runner_messaging" {
  count         = local.is_ephemeral ? 0 : 1
  name          = "alias/${var.environment}-${local.runner_service_name}-messaging"
  target_key_id = aws_kms_key.runner_messaging[0].key_id
}

################
# SQS — Task queue + Dead Letter Queue
################

resource "aws_sqs_queue" "runner_tasks_dlq" {
  count                             = local.is_ephemeral ? 0 : 1
  name                              = "${var.environment}-${local.runner_service_name}-tasks-dlq"
  message_retention_seconds         = 1209600 # 14 days
  kms_master_key_id                 = aws_kms_key.runner_messaging[0].arn
  kms_data_key_reuse_period_seconds = 300

  tags = {
    environment        = var.environment
    application_family = var.application_family
    service_name       = local.runner_service_name
  }
}

resource "aws_sqs_queue" "runner_tasks" {
  count                             = local.is_ephemeral ? 0 : 1
  name                              = "${var.environment}-${local.runner_service_name}-tasks"
  visibility_timeout_seconds        = 600
  message_retention_seconds         = 345600 # 4 days
  kms_master_key_id                 = aws_kms_key.runner_messaging[0].arn
  kms_data_key_reuse_period_seconds = 300

  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.runner_tasks_dlq[0].arn
    maxReceiveCount     = 3
  })

  tags = {
    environment        = var.environment
    application_family = var.application_family
    service_name       = local.runner_service_name
  }
}

################
# S3 — Runner cache (pipeline inputs/outputs)
################

module "s3_ows_coda_runner_cache" {
  count              = local.is_ephemeral ? 0 : 1
  source             = "git@github.com:theorchard/terraform-s3.git//modules/s3_bucket?ref=3.15.7"
  env                = var.environment
  bucket_name        = "${local.runner_service_name}-cache"
  application_family = var.application_family

  apply_server_side_encryption_by_default = {
    sse_algorithm     = "aws:kms"
    kms_master_key_id = data.aws_kms_alias.s3.target_key_arn
  }

  block_public_access = true

  lifecycle_rules_options_current_version_expiration = [
    {
      prefix  = "inputs/"
      days    = 1
      enabled = true
    }
  ]

  lifecycle_rules_options_noncurrent_version_expiration = [
    {
      prefix  = "inputs/"
      days    = 3
      enabled = true
    }
  ]

  lifecycle_rules_abort_incomplete_multipart_upload_days = [
    {
      prefix  = ""
      days    = 7
      enabled = true
    }
  ]
}

################
# Secrets — Snowflake + Sentry for runner service
################

module "ows_coda_runner_secrets" {
  source = "git@github.com:theorchard/terraform-secrets-manager.git//?ref=1.6.1"
  for_each = local.is_ephemeral ? toset([]) : toset([
    "SNOWFLAKE_PRIVATE_KEY",
    "SNOWFLAKE_PRIVATE_KEY_PASS",
    "CODA_DB_PASS",
  ])

  application_family = var.application_family
  environment        = var.environment
  service_name       = local.runner_service_name
  secret_name        = each.value
}

################
# Sentry — Runner service project
################

module "sentry_ows_coda_runner" {
  count  = local.is_ephemeral ? 0 : 1
  source = "git@github.com:theorchard/terraform-sentry.git//?ref=5.1.0"

  environment        = var.environment
  platform           = "node"
  service_name       = local.runner_service_name
  teams              = ["coda"]
  application_family = var.application_family
}

################
# Datadog — Runner service dashboard
################

module "ows_coda_runner_service_dashboard" {
  count                             = local.is_ephemeral ? 0 : 1
  source                            = "git@github.com:theorchard/terraform-datadog.git//modules/service?ref=6.19.0"
  environment                       = var.environment
  environment_type                  = "fargate"
  service_name                      = local.runner_service_name
  application_family                = var.application_family
  teams                             = ["coda"]
  notification_endpoints            = local.datadog_notification_endpoints
  escalation_notification_endpoints = local.datadog_escalation_notification_endpoints
}

locals {
  # Extract arn_suffix from the full target group ARN for CloudWatch dimensions.
  # "arn:aws:elasticloadbalancing:REGION:ACCOUNT:targetgroup/NAME/HASH" → "targetgroup/NAME/HASH"
  runner_target_group_arn_suffix = local.is_ephemeral ? "" : element(split(":", module.runner_fargate[0].fargate_target_group_arn), 5)
}

################
# CloudWatch Alarms — Service health + DLQ monitoring
################

resource "aws_sns_topic" "runner_alerts" {
  count             = local.is_ephemeral ? 0 : 1
  name              = "${var.environment}-${local.runner_service_name}-alerts"
  kms_master_key_id = aws_kms_key.runner_messaging[0].arn
}

resource "aws_cloudwatch_metric_alarm" "runner_no_running_tasks" {
  count               = local.is_ephemeral ? 0 : 1
  alarm_name          = "${var.environment}-${local.runner_service_name}-no-running-tasks"
  comparison_operator = "LessThanThreshold"
  evaluation_periods  = 2
  metric_name         = "RunningTaskCount"
  namespace           = "ECS/ContainerInsights"
  period              = 60
  statistic           = "Minimum"
  threshold           = 1
  alarm_description   = "Runner ECS service has 0 running tasks — service is down"
  alarm_actions       = [aws_sns_topic.runner_alerts[0].arn]
  ok_actions          = [aws_sns_topic.runner_alerts[0].arn]

  dimensions = {
    ClusterName = module.runner_fargate[0].fargate_cluster_name
    ServiceName = module.runner_fargate[0].fargate_web_service_name
  }
}

resource "aws_cloudwatch_metric_alarm" "runner_no_healthy_targets" {
  count               = local.is_ephemeral ? 0 : 1
  alarm_name          = "${var.environment}-${local.runner_service_name}-no-healthy-targets"
  comparison_operator = "LessThanThreshold"
  evaluation_periods  = 2
  metric_name         = "HealthyHostCount"
  namespace           = "AWS/ApplicationELB"
  period              = 60
  statistic           = "Minimum"
  threshold           = 1
  alarm_description   = "Runner ALB has 0 healthy targets — health checks failing"
  alarm_actions       = [aws_sns_topic.runner_alerts[0].arn]
  ok_actions          = [aws_sns_topic.runner_alerts[0].arn]

  dimensions = {
    TargetGroup  = local.runner_target_group_arn_suffix
    LoadBalancer = module.runner_fargate[0].fargate_load_balancer_arn_suffix
  }
}

resource "aws_cloudwatch_metric_alarm" "runner_dlq_messages" {
  count               = local.is_ephemeral ? 0 : 1
  alarm_name          = "${var.environment}-${local.runner_service_name}-dlq-messages"
  comparison_operator = "GreaterThanThreshold"
  # Two consecutive periods reduces noise from transient Snowflake rate limits
  # or one-off sandbox timeouts that self-resolve on retry.
  evaluation_periods = 2
  metric_name        = "ApproximateNumberOfMessagesVisible"
  namespace          = "AWS/SQS"
  period             = 60
  statistic          = "Maximum"
  threshold          = 0
  alarm_description  = "Runner DLQ has messages for 2+ minutes — tasks are failing after 3 retries"
  alarm_actions      = [aws_sns_topic.runner_alerts[0].arn]
  ok_actions         = [aws_sns_topic.runner_alerts[0].arn]

  dimensions = {
    QueueName = aws_sqs_queue.runner_tasks_dlq[0].name
  }
}

################
# IAM Policies — SQS access for runner task
################

data "aws_iam_policy_document" "runner_sqs" {
  count = local.is_ephemeral ? 0 : 1

  statement {
    effect = "Allow"
    actions = [
      "sqs:ReceiveMessage",
      "sqs:DeleteMessage",
      "sqs:SendMessage",
      "sqs:GetQueueAttributes",
      "sqs:ChangeMessageVisibility",
    ]
    resources = [
      aws_sqs_queue.runner_tasks[0].arn,
      aws_sqs_queue.runner_tasks_dlq[0].arn,
    ]
  }

  # SQS queues use a CMK — task role needs key access to encrypt/decrypt messages.
  statement {
    effect = "Allow"
    actions = [
      "kms:GenerateDataKey",
      "kms:Decrypt",
    ]
    resources = [aws_kms_key.runner_messaging[0].arn]
  }
}

resource "aws_iam_policy" "runner_sqs" {
  count = local.is_ephemeral ? 0 : 1

  name        = "${var.environment}-${local.runner_service_name}-sqs-policy"
  description = "Allow ${local.runner_service_name} ECS task to consume and produce SQS messages"
  policy      = data.aws_iam_policy_document.runner_sqs[0].json
}

################
# IAM Policies — S3 cache access for runner task
################

data "aws_iam_policy_document" "runner_cache" {
  count = local.is_ephemeral ? 0 : 1

  statement {
    effect    = "Allow"
    actions   = ["s3:GetObject", "s3:PutObject", "s3:DeleteObject"]
    resources = ["${module.s3_ows_coda_runner_cache[0].s3_bucket_arn_output}/*"]
  }

  statement {
    effect    = "Allow"
    actions   = ["s3:GetBucketLocation", "s3:ListBucket"]
    resources = [module.s3_ows_coda_runner_cache[0].s3_bucket_arn_output]
  }

  # SSE-KMS: task role needs key access to encrypt/decrypt cache objects.
  statement {
    effect    = "Allow"
    actions   = ["kms:GenerateDataKey", "kms:Decrypt"]
    resources = [data.aws_kms_alias.s3.target_key_arn]
  }
}

resource "aws_iam_policy" "runner_cache" {
  count = local.is_ephemeral ? 0 : 1

  name        = "${var.environment}-${local.runner_service_name}-cache-policy"
  description = "Allow ${local.runner_service_name} ECS task to read/write S3 cache objects"
  policy      = data.aws_iam_policy_document.runner_cache[0].json
}

################
# Fargate — ECS cluster, task definition, service, ALB, security groups, DNS
################

module "runner_fargate" {
  count  = local.is_ephemeral ? 0 : 1
  source = "git@github.com:theorchard/terraform-fargate.git//?ref=6.5.0"

  providers = {
    aws.dns = aws.networking
  }

  environment        = var.environment
  service_name       = local.runner_service_name
  aws_region         = var.region
  application_family = var.application_family
  commit_sha         = "latest"
  container_port     = tostring(local.runner_port)
  task_type          = "web_service"

  web_service_health_check_command = "node -e \"require('http2').connect('http://localhost:8082').on('connect',()=>process.exit(0)).on('error',()=>process.exit(1))\""

  container_protocol_version = "HTTP2"
  health_check_matcher       = "200"

  desired_task_count = 1 # QA: single task is acceptable (main app falls back gracefully).
  task_cpu           = 1024
  task_memory        = 2048
  maximum_capacity   = 2
  minimum_capacity   = 1

  load_balancer_access_logs_s3_bucket_name = local.elb_logs_bucket
  blocking_waf_enabled                     = true

  vpc_id                           = module.vpc_info.vpc_id
  https_listener_certificate_id    = split("/", data.aws_acm_certificate.theorchard_io.arn)[1]
  health_check_path                = "/health"
  health_check_healthy_threshold   = 2
  health_check_unhealthy_threshold = 5

  secrets_manager_service_name = local.runner_service_name

  fargate_service_subnets = module.vpc_info.default_private_subnet_ids
  load_balancer_subnets   = module.vpc_info.default_public_subnet_ids

  secrets = [
    {
      SNOWFLAKE_PRIVATE_KEY = "${var.environment}/${local.runner_service_name}/SNOWFLAKE_PRIVATE_KEY"
    },
    {
      SNOWFLAKE_PRIVATE_KEY_PASS = "${var.environment}/${local.runner_service_name}/SNOWFLAKE_PRIVATE_KEY_PASS"
    },
    {
      CODA_DB_PASS = "${var.environment}/${local.runner_service_name}/CODA_DB_PASS"
    },
  ]

  environment_variables = [
    {
      ENVIRONMENT = var.environment
    },
    {
      NODE_ENV = var.environment
    },
    {
      S3_CACHE_BUCKET = module.s3_ows_coda_runner_cache[0].s3_bucket_name_output
    },
    {
      SQS_QUEUE_URL = aws_sqs_queue.runner_tasks[0].url
    },
    {
      SNOWFLAKE_ACCOUNT = var.runner_snowflake_account
    },
    {
      SNOWFLAKE_DATABASE = var.runner_snowflake_database
    },
    {
      SNOWFLAKE_ROLE = var.runner_snowflake_role
    },
    {
      SNOWFLAKE_USER = var.runner_snowflake_username
    },
    {
      SNOWFLAKE_WAREHOUSE = var.runner_snowflake_warehouse
    },
    {
      # Shared Aurora MySQL — same DB as main app and platform service
      CODA_DB_HOST = local.rds_host
    },
    {
      CODA_DB_PORT = "3306"
    },
    {
      CODA_DB_USER = "coda_svc"
    },
    {
      CODA_DB_DATABASE = local.db_name
    },
    {
      SENTRY_DSN = try(module.sentry_ows_coda_runner[0].sentry_key_dsn_public_output, "")
    },
  ]
}

################
# IAM — Attach SQS and S3 cache policies to Fargate task role
################

resource "aws_iam_role_policy_attachment" "runner_sqs" {
  count = local.is_ephemeral ? 0 : 1

  role       = module.runner_fargate[0].fargate_task_iam_role_name
  policy_arn = aws_iam_policy.runner_sqs[0].arn
}

resource "aws_iam_role_policy_attachment" "runner_cache" {
  count = local.is_ephemeral ? 0 : 1

  role       = module.runner_fargate[0].fargate_task_iam_role_name
  policy_arn = aws_iam_policy.runner_cache[0].arn
}