{# Neo4j CDC source connector main.tf
   Path on disk: terraform-infra/<env>/kafka-infra/neo4j_cdc_source/<purpose>/main.tf

   Supports BOTH on-prem Neo4j and Neo4j Aura.
   The connector image is the same for both — only inputs differ:
     - on_prem URI:  neo4j+s://<host>.theorchard.io:7687   (port required)
     - aura URI:     neo4j+s://<instance-id>.databases.neo4j.io  (no port)

   Required template variables:
     purpose                  e.g. "neo4j_v5", "mg_fingerprint", "aura_graph" (snake_case)
     purpose_hyphens          e.g. "neo4j-v5" — used in backend state key
     environment              "dev" | "qa" | "prod"
     neo4j_instance_type      "on_prem" | "aura"
     connector_name           user-provided short identifier (e.g. "fingerprinting", "fp", "mg-v5").
                              Service name = kc-neo-src-<connector_name>; terraform-fargate prepends
                              "<env>-" → final "<env>-kc-neo-src-<connector_name>" (ALB name, <= 32 chars,
                              so len(env) + len(connector_name) <= 20). Set in variables.tf.
     neo4j_server_uri         on_prem: "neo4j+s://qa-neo4j-cluster.theorchard.io:7687"
                              aura:    "neo4j+s://9e17b596.databases.neo4j.io"
     neo4j_database_name      on_prem default "graph.db"; aura default "neo4j"
     topic_mappings           list of { mapping_key, topic, pattern, key_strategy? }
                              e.g. { mapping_key: "LABELSOUNDRECORDING",
                                     topic: "cdc.musicGraphV5.labelSoundRecording",
                                     pattern: "(:LabelSoundRecording)" }
     task_cpu                 default 1024
     task_memory              default 3072
     desired_task_count       default 1
     maximum_capacity         default 2
     msk_cluster_name_short   default "managed-kafka-cdc-destination"
     use_fargate_spot         default true (qa); set false for prod if needed
     team_name                user-provided (no plugin default; ask the user)
     neo4j_key_strategy       optional. If omitted, the image default WHOLE_VALUE applies
                              (set via ENV in kafka-connect/neo4j_cdc_source/Dockerfile).
                              Valid values: WHOLE_VALUE | ELEMENT_ID | ENTITY_KEYS.
                              Per-mapping override available via 3rd pipe segment in
                              TOPIC_MAPPING_<KEY>="<topic>|<pattern>|<key-strategy>".
     neo4j_start_from         default "NOW". Set to "USER_PROVIDED" to resume from a specific
                              change identifier — then also pass `neo4j_start_from_value`.
     neo4j_start_from_value   only used when neo4j_start_from = "USER_PROVIDED"; the change
                              identifier string to resume from.
     neo4j_connection_timeout            optional; recommend "60s" for Aura.
     neo4j_connection_acquisition_timeout optional; recommend "60s" for Aura.

   References:
     - CDC mode docs:   https://neo4j.com/docs/kafka/current/source/cdc/
     - QUERY mode docs: https://neo4j.com/docs/kafka/current/source/query/
     - Connector class: org.neo4j.connectors.kafka.source.Neo4jConnector
     - Requires CDC enabled on the target Neo4j database.

   Module versions (pinned to verified qa neo4j_v5 layout):
     terraform_fargate_ref = "6.3.0"
     terraform_default_tags_ref = "2.0.0"
     terraform_datadog_kafka_connector_ref = "6.15.3"
     terraform_vpc_info_ref = "3.1.0"
     terraform_secrets_manager_ref = "1.5.1"
     terraform_required_version = "1.14.0"

   Pair this file with:
     - variables.tf rendered from neo4j_cdc_source_variables.tf.j2
     - versions.tf rendered from versions.tf.j2 (omit_datadog_provider=true)

   Do NOT emit a top-of-file comment banner — match the existing
   instances which start directly with the `terraform {` block.
#}
terraform {
  backend "s3" {
    bucket  = "orcd-terraform-state"
    key     = "{{ environment }}/kafka-infra/neo4j_cdc_source/{{ purpose_hyphens }}/terraform.tfstate"
    region  = "us-east-1"
    encrypt = "true"
  }
}

module "default_tags" {
  source             = "git@github.com:theorchard/terraform-default-tags.git//?ref={{ terraform_default_tags_ref | default('2.0.0') }}"
  environment        = var.environment
  application_family = var.application_family
  service_name       = local.service_name
  team_name          = "{{ team_name }}"
}

provider "aws" {
  region = var.aws_region

  default_tags {
    tags = module.default_tags.tags
  }
}

provider "aws" {
  region  = var.aws_region
  alias   = "networking"
  profile = "networking"

  default_tags {
    tags = module.default_tags.tags
  }
}

# VPC Info
data "aws_caller_identity" "current" {}

module "vpc_info" {
  source = "git@github.com:theorchard/terraform-vpc-info.git?ref={{ terraform_vpc_info_ref | default('3.1.0') }}"

  environment = var.environment
}

data "aws_acm_certificate" "theorchard_io" {
  domain   = "*.theorchard.io"
  statuses = ["ISSUED"]
}

data "aws_route53_zone" "route53_zone" {
  name = "theorchard.io"
}

# MSK Cluster
data "aws_msk_cluster" "kafka_infra" {
  cluster_name = "${var.environment}-${var.msk_cluster_name}"
}

# Fargate module
module "kafka_{{ purpose }}_fargate_environment" {
  source = "git@github.com:theorchard/terraform-fargate.git//?ref={{ terraform_fargate_ref | default('6.3.0') }}"

  providers = {
    aws.dns = aws.networking
  }

  environment                       = var.environment
  application_family                = var.application_family
  additional_tags                   = var.additional_tags
  service_name                      = local.service_name
  non_ecr_image                     = "086679231553.dkr.ecr.${var.aws_region}.amazonaws.com/${local.ecr_image_name}:latest"
  aws_region                        = var.aws_region
  service_platform_version          = "1.4.0"
  desired_task_count                = {{ desired_task_count | default(1) }}
  minimum_capacity                  = {{ minimum_capacity | default(1) }}
  maximum_capacity                  = {{ maximum_capacity | default(2) }}
  task_cpu                          = {{ task_cpu | default(1024) }}
  task_memory                       = {{ task_memory | default(3072) }}
  container_port                    = 8083
  health_check_grace_period_seconds = 300
  container_start_period_seconds    = 300
  blocking_waf_enabled              = true
  health_check_path                 = "/connectors/neo4j_cdc_source"
  web_service_health_check_command  = "[[ $(curl -s http://localhost:8083/connectors/neo4j_cdc_source/status | jq -r '.tasks[].state' | sort | uniq) == 'RUNNING' ]] || exit 1"
{% if use_fargate_spot | default(true) %}
  # Fargate SPOT Task (typically qa)
  fargate_spot_capacity_provider_weight = 100
  task_placement_failure_alert_enabled  = true
{% endif %}
  propagate_tags                = true
  vpc_id                        = module.vpc_info.vpc_id
  https_listener_certificate_id = split("/", data.aws_acm_certificate.theorchard_io.arn)[1]
  fargate_service_subnets       = module.vpc_info.default_private_subnet_ids
  load_balancer_subnets         = module.vpc_info.default_private_subnet_ids

  # datadog agent crashes due to OutOfMemoryError with JMX. So increase it from 256
  datadog_task_memory  = 512
  datadog_custom_image = "086679231553.dkr.ecr.${var.aws_region}.amazonaws.com/orchard-datadog-agent-jmx:latest"
  datadog_agent_environment_variables = [
    {
      JMX_PORT = "9095"
    },
    {
      DD_ENV = var.environment
    },
    {
      SERVICE_NAME = local.service_name
    },
    {
      CONNECTOR_TYPE = "default_source"
    },
  ]

  environment_variables = [
    {
      Environment = var.environment
    },
    {
      CONNECT_GROUP_UNIQUE_IDENTIFIER = "${var.environment}-${local.service_name}"
    },
    {
      SERVICE_NAME = local.service_name
    },
    {
      CONNECT_BOOTSTRAP_SERVERS = data.aws_msk_cluster.kafka_infra.bootstrap_brokers_tls
    },
    {
      NEO4J_SERVER_URI = "{{ neo4j_server_uri }}"
    },
    {
      NEO4J_DATABASE_NAME = "{{ neo4j_database_name }}"
    },
    {
      MAX_TASKS = "{{ max_tasks | default(1) }}"
    },
    {
      NEO4J_ENCRYPTION_ENABLED = "{{ neo4j_encryption_enabled | default('true') }}"
    },
    {
      NEO4J_ENFORCE_SCHEMA = "{{ neo4j_enforce_schema | default('false') }}"
    },
    {
      NEO4J_BATCH_SIZE = "{{ neo4j_batch_size | default('1000') }}"
    },
    {
      NEO4J_POLL_INTERVAL = "{{ neo4j_poll_interval | default('10s') }}"
    },
    {
      NEO4J_START_FROM = "{{ neo4j_start_from | default('NOW') }}"
    },
{%- if neo4j_start_from is defined and neo4j_start_from == 'USER_PROVIDED' %}
    {
      NEO4J_START_FROM_VALUE = "{{ neo4j_start_from_value }}"
    },
{%- endif %}
{%- if neo4j_key_strategy is defined %}
    {
      NEO4J_KEY_STRATEGY = "{{ neo4j_key_strategy }}"
    },
{%- endif %}
    {
      NEO4J_IGNORE_STORED_OFFSET = "{{ neo4j_ignore_stored_offset | default('false') }}"
    },
{%- if neo4j_connection_timeout is defined %}
    {
      NEO4J_CONNECTION_TIMEOUT = "{{ neo4j_connection_timeout }}"
    },
{%- endif %}
{%- if neo4j_connection_acquisition_timeout is defined %}
    {
      NEO4J_CONNECTION_ACQUISITION_TIMEOUT = "{{ neo4j_connection_acquisition_timeout }}"
    },
{%- endif %}
    {
      NEO4J_KEY_CONVERTER = "org.apache.kafka.connect.storage.StringConverter"
    },
    {
      NEO4J_VALUE_CONVERTER = "org.apache.kafka.connect.json.JsonConverter"
    },
    {
      NEO4J_KEY_CONVERTER_SCHEMAS_ENABLE = "false"
    },
    {
      NEO4J_VALUE_CONVERTER_SCHEMAS_ENABLE = "false"
    },
    {
      AWS_SECRETS_MANAGER = "true"
    },
    {
      DEBUG_MODE = "{{ debug_mode | default('false') }}"
    },
    {
      CONNECT_LOG4J_ROOT_LOGLEVEL = "{{ log_level | default('WARN') }}"
    },
{%- for m in topic_mappings %}
    {
      TOPIC_MAPPING_{{ m.mapping_key }} = "{{ m.topic }}|{{ m.pattern }}{% if m.key_strategy %}|{{ m.key_strategy }}{% endif %}"
    },
{%- endfor %}
  ]
}

# This rule will allow kafka-connect tasks to communicate with each other in distributed mode
resource "aws_security_group_rule" "allow_intra_cluster_communication" {
  from_port         = 8083
  to_port           = 8083
  protocol          = "tcp"
  security_group_id = module.kafka_{{ purpose }}_fargate_environment.fargate_security_group_id
  self              = true
  type              = "ingress"
}

# Secrets for Neo4j credentials (read at connector startup via Lenses AWS secret provider)
module "service_credentials_secret" {
  source = "git@github.com:theorchard/terraform-secrets-manager.git//?ref={{ terraform_secrets_manager_ref | default('1.5.1') }}"

  environment                    = var.environment
  application_family             = var.application_family
  additional_tags                = var.additional_tags
  service_name                   = local.service_name
  secret_name                    = "NEO4J_CREDENTIALS"
  secret_recovery_window_in_days = 7
}

# Datadog Dashboard
module "fargate_service_dashboard_{{ purpose }}" {
  source = "git@github.com:theorchard/terraform-datadog.git//modules/kafka_connector?ref={{ terraform_datadog_kafka_connector_ref | default('6.15.3') }}"

  environment        = var.environment
  application_family = var.application_family
  additional_tags    = var.additional_tags
  service_name       = local.service_name
  connector_type     = "default_source"

  environment_type = "fargate"

  notification_endpoints            = "{{ notification_endpoints }}"
  escalation_notification_endpoints = "{{ escalation_notification_endpoints }}"
  monitor_kafka_topics = [
{%- for m in topic_mappings %}
    "{{ m.topic }}"{% if not loop.last %},{% endif %}
{%- endfor %}
  ]

  healthy_tasks_monitor_enabled   = true
  service_cpu_monitor_enabled     = true
  error_monitor_enabled           = true
  healthy_tasks_evaluation_window = "last_1h"
  service_cpu_time_window         = "last_1h"
}