# -----------------------------------------------------------------------------
# Snowflake Sink Connector — Fargate Service (Phase 4b)
#
# Runs Kafka Connect with the Snowflake sink connector on Fargate.
# Consumes from MSK topic "resonance-engine.spotify-data" and writes to
# Snowflake landing table RE_SPOTIFY_DATA_LANDING via Snowpipe Streaming.
#
# Follows org pattern: terraform-infra/prod/kafka-infra/snowflake_sink_fanresponse/
# -----------------------------------------------------------------------------

# -----------------------------------------------------------------------------
# Secrets Manager — Snowflake private key + passphrase
# Creates empty secret shells; populate manually after terraform apply.
# -----------------------------------------------------------------------------

module "snowflake_sink_secrets" {
  source   = "git@github.com:theorchard/terraform-secrets-manager.git//?ref=1.5.1"
  for_each = toset(var.sink_snowflake_secret_names)

  environment        = var.environment
  application_family = var.application_family
  service_name       = var.sink_service_name
  secret_name        = each.value
}

# -----------------------------------------------------------------------------
# Fargate Service — Kafka Connect with Snowflake Sink Connector
# -----------------------------------------------------------------------------

module "snowflake_sink_fargate" {
  source = "git@github.com:theorchard/terraform-fargate.git//?ref=6.4.1"

  providers = {
    aws.dns = aws.networking
  }

  environment        = var.environment
  application_family = var.application_family

  service_name  = var.sink_service_name
  non_ecr_image = var.sink_ecr_image
  aws_region    = var.aws_region

  # Worker task type: this is a pure MSK consumer → Snowflake writer.
  # Setting task_type = "worker" suppresses the ALB, HTTPS listener, LB security
  # group, Route53 record, and WAF association — none of which are needed for an
  # internal background service that never serves HTTP traffic.
  task_type = "worker"

  desired_task_count = var.sink_desired_task_count
  minimum_capacity   = var.sink_desired_task_count == 0 ? 0 : 1
  maximum_capacity   = 2

  task_cpu    = 2048
  task_memory = 4096

  # Port 8083 is the Kafka Connect REST API, used here only for the in-container
  # health check command below — not for external traffic routing.
  container_port                    = 8083
  health_check_grace_period_seconds = 180
  container_start_period_seconds    = 180
  propagate_tags                    = true

  # Disabled in dev: the fargate module reads SSM parameter
  # /events/destination-api/datadog when enabled, and generic-engineer-role
  # lacks ssm:GetParameter. Enable for prod (dedicated account with proper IAM).
  task_placement_failure_alert_enabled = false

  # Container-level health check: polls the connector REST API locally.
  # This runs inside the task via CMD-SHELL — no ALB health check path needed.
  # IMPORTANT: For task_type = "worker", must use health_check_command (not
  # web_service_health_check_command, which only applies to web_service task type).
  health_check_command = "[[ $(curl -s http://localhost:8083/connectors/${var.sink_connector_name}/status | jq -r '.tasks[].state,.connector.state' | sort | uniq) == 'RUNNING' ]] || exit 1"

  vpc_id                  = module.vpc_info.vpc_id
  fargate_service_subnets = module.vpc_info.default_private_subnet_ids

  # Datadog disabled in dev — enable for prod
  datadog_enabled = false

  # Disable unused org integrations
  ows_machine_to_machine_enabled = false
  splitio_enabled                = false

  environment_variables = [
    { Environment = var.environment },
    { CONNECT_GROUP_UNIQUE_IDENTIFIER = "${var.environment}-${var.sink_service_name}" },
    { CONSUMER_PARTITION_ASSIGNMENT_STRATEGY = "org.apache.kafka.clients.consumer.RoundRobinAssignor" },
    { MAX_TASKS = 1 },
    { BUFFER_COUNT_RECORDS = 10000 },
    { BUFFER_SIZE_BYTES = 5000000 },
    { BUFFER_FLUSH_TIME = 180 },
    { CONNECT_BOOTSTRAP_SERVERS = var.sink_bootstrap_servers },
    { SNOWFLAKE_INGESTION_METHOD = "SNOWPIPE_STREAMING" },
    { SNOWFLAKE_HOST = "delphi.us-east-1.snowflakecomputing.com:443" },
    { SNOWFLAKE_USER = var.sink_snowflake_user },
    { SNOWFLAKE_ROLE = var.sink_snowflake_role },
    { SNOWFLAKE_DATABASE = "FANSIFTER_APP_REPORTING" },
    { SNOWFLAKE_SCHEMA = var.sink_snowflake_schema },
    { SERVICE_NAME = var.sink_service_name },
    { KAFKA_TOPICS = var.kafka_topic },
    { SNOWFLAKE_TOPIC_TABLE_MAP = "${var.kafka_topic}:RE_SPOTIFY_DATA_LANDING" },
    { DLQ_TOPIC_NAME = "dlq.resonance-engine" },
  ]

  secrets = [
    {
      SNOWFLAKE_PRIVATE_KEY = "${var.environment}/${var.sink_service_name}/SNOWFLAKE_PRIVATE_KEY"
    },
    {
      SNOWFLAKE_PRIVATE_KEY_PASSPHRASE = "${var.environment}/${var.sink_service_name}/SNOWFLAKE_PRIVATE_KEY_PASSPHRASE"
    },
  ]
}

# -----------------------------------------------------------------------------
# Security Group Rule — Intra-cluster communication (Kafka Connect distributed)
# Required for Kafka Connect distributed mode rebalancing on port 8083.
# -----------------------------------------------------------------------------

resource "aws_security_group_rule" "sink_intra_cluster" {
  from_port         = 8083
  to_port           = 8083
  protocol          = "tcp"
  security_group_id = module.snowflake_sink_fargate.fargate_security_group_id
  self              = true
  type              = "ingress"
}

# -----------------------------------------------------------------------------
# Datadog Monitoring — Disabled in dev
# Uncomment and configure for prod deployment.
# -----------------------------------------------------------------------------

# module "sink_datadog_monitor" {
#   source = "git@github.com:theorchard/terraform-datadog.git//modules/kafka_connector?ref=6.13.4"
#
#   environment        = var.environment
#   application_family = var.application_family
#   environment_type   = "fargate"
#   service_name       = var.sink_service_name
#   connector_type     = "default_sink"
#
#   notification_endpoints            = "@slack-resonance-engine-alerts"
#   escalation_notification_endpoints = "@slack-data"
#   monitor_kafka_topics_cluster_name = "dev-managed-kafka-cdc-destination"
#   monitor_kafka_topics              = [var.kafka_topic, "dlq.resonance-engine"]
#   dlq_topic_name                    = "dlq.resonance-engine"
#
#   healthy_tasks_monitor_enabled = true
#   service_cpu_monitor_enabled   = true
#   error_monitor_enabled         = true
# }