module "lambda-nr-ownership-ingest-ar-rights-filter" {
  source                      = "git@github.com:theorchard/terraform-lambda.git//?ref=4.0.1"
  environment                 = var.environment
  lambda_name                 = var.ar_rights_filter_lambda_name
  lambda_description          = "Consumes cdc art relation rights data."
  lambda_function_timeout     = "900"
  use_container_image         = true
  vpc_subnet_ids              = module.vpc_info.default_private_subnet_ids
  vpc_enabled                 = true
  vpc_id                      = module.vpc_info.vpc_id
  datadog_advanced_enabled    = true
  application_family          = var.application_family
  lambda_function_memory_size = var.lambda_memory_size
  splitio_enabled             = true

  msk_event_enabled                      = var.msk_event_enabled
  event_source_mapping_msk_cluster_name  = var.trigger_msk_cluster_name
  event_source_mapping_msk_cluster_uuid  = var.trigger_msk_cluster_uuid
  event_source_mapping_batch_size        = 20
  kafka_topics                           = [var.ar_track_master_rights_topic, var.ar_vendor_contract_topic, var.ar_release_territory_restriction_topic, var.ar_subaccount_royalty_collection_topic, var.ar_subaccount_royalty_collection_territories_topic, var.project_used_to_belong_to_msk_topic_v5]
  event_source_mapping_starting_position = "LATEST"

  # MSK ESM filters can only match on the message value, not the topic/key, so a
  # single $or pattern is applied across all topics on this mapping. The 5 AR CDC
  # topics carry no `event.type` field, so they always match the first clause and
  # pass through unfiltered. Only the projectUsedToBelongTo v5 topic is actually
  # filtered: USED_TO_BELONG_TO events are dropped unless the operation is CREATE.
  #   - no event.type            -> AR CDC topics, allow everything
  #   - event.type != USED_TO... -> any other Neo4j CDC event type, allow
  #   - USED_TO_BELONG_TO        -> allow only CREATE (drop UPDATE/DELETE)
  event_source_mapping_filter_criteria_pattern = jsonencode({
    value = {
      "$or" = [
        { "event.type" = [{ "exists" = false }] },
        { "event.type" = [{ "anything-but" = ["USED_TO_BELONG_TO"] }] },
        { "event.type" = ["USED_TO_BELONG_TO"], "event.operation" = ["CREATE"] },
      ]
    }
  })

  lambda_function_environment_variables = {
    Environment        = var.environment
    DD_LAMBDA_HANDLER  = "app.handler"
    SENTRY_DSN         = module.lambda_nr_ownership_ingest_ar_rights_filter_sentry_project.sentry_key_dsn_public_output
    SPLIT_API_KEY_PATH = "${var.environment}/split/API_KEY"
    BUCKET_NAME        = data.aws_s3_bucket.s3_bucket.bucket
    KAFKAJS_LOG_LEVEL  = "nothing"
    KAFKA_BROKERS      = var.kafka_brokers
    SQS_QUEUE_URL      = module.sqs_sync_rights_queue.queue_url
    SQS_RELEASES_URL   = module.sqs_filter_products_queue.queue_url
    SQS_PRODUCT_TRANSFER_URL = module.sqs_product_transfer_queue.queue_url
    LOGGER_LEVEL       = "info"
    SEND_TO_SQS        = var.send_to_sqs
  }

  lambda_function_reserved_concurrent_executions = 1

  iam_managed_policy_attachments = [
    data.aws_iam_policy.ownership_ingestion_bucket_rw_policy.arn,
    module.sqs_sync_rights_queue.sqs_minimal_policy_arn_output,
    module.sqs_filter_products_queue.sqs_minimal_policy_arn_output,
    module.sqs_product_transfer_queue.sqs_minimal_policy_arn_output
  ]
}

module "lambda_nr_ownership_ingest_ar_rights_filter_sentry_project" {
  source                       = "git@github.com:theorchard/terraform-sentry//?ref=4.1.2"
  service_name                 = var.ar_rights_filter_lambda_name
  secrets_manager_service_name = var.ar_rights_filter_lambda_name
  environment                  = var.environment
  platform                     = "node"
  application_family           = var.application_family
}

module "datadog_lambda_nr_ownership_ingest_ar_rights_filter_dashboards" {
  source                 = "git@github.com:theorchard/terraform-datadog.git//modules/lambda?ref=6.13.4"
  service_name           = var.ar_rights_filter_lambda_name
  environment            = var.environment
  notification_endpoints = var.notification_endpoints
  teams                  = [var.team_name]
}

locals {
  ar_rights_filter_consumer_group_ids = var.msk_event_enabled ? join(" OR ", [for c in module.lambda-nr-ownership-ingest-ar-rights-filter.msk_event_source_mapping_uuid : "consumer_group:${c}"]) : ""
  ar_rights_filter_quiet_topics = var.msk_event_enabled ? join(" OR ", [for topic in [var.ar_release_territory_restriction_topic,  var.ar_track_master_rights_topic, var.ar_subaccount_royalty_collection_topic, var.ar_vendor_contract_topic] : "topic:${lower(topic)}"]) : ""
}


#### Offset lag monitoring is split into two monitors:
#### 1. High-volume topic ( ar_subaccount_royalty_collection_territories_topic) with 2k threshold
#### 2. Lower-volume for the other topics with 600 threshold
#### This separation prevents alert fatigue from the busier track artist topic
resource "datadog_monitor" "lambda_nr_ownership_ingest_ar_rights_filter_offset_lag_monitor" {
  count              = var.msk_event_enabled ? 1 : 0
  name               = "${var.environment}-${var.ar_rights_filter_lambda_name}-offset-lag"
  type               = "metric alert"
  message            = "{{#is_alert}} The Lambda offset lag is too high {{/is_alert}} {{#is_alert_recovery}} Offset lag is recovering {{/is_alert_recovery}} in {{[aws.kafka.sum_offset_lag].topic}} \n Notify: ${var.notification_endpoints}"
  query              = "avg(last_2h):aws.kafka.sum_offset_lag{environment:${var.environment} AND ${local.ar_rights_filter_quiet_topics} AND (${local.ar_rights_filter_consumer_group_ids})} by {topic} > 600"

  monitor_thresholds {
    critical = 600
    warning  = 500
  }

  notify_no_data = true
  no_data_timeframe = 60
  notify_audit = false
  renotify_interval = 30
  timeout_h = 0
  require_full_window = false

  tags = ["team:${var.team_name}"]
}

#### Separate monitor for ar_subaccount_royalty_collection_territories_topic due to significantly higher 
#### traffic volume compared to other topics consumed by this lambda.
#### Higher thresholds reflect normal operating conditions.
resource "datadog_monitor" "lambda_nr_ownership_ingest_ar_rights_filter_sarct_offset_lag_monitor" {
  count              = var.msk_event_enabled ? 1 : 0
  name               = "${var.environment}-${var.ar_rights_filter_lambda_name}-sarct-offset-lag"
  type               = "metric alert"
  message            = "{{#is_alert}} The Lambda offset on cdc.artRelations.subAccountRoyaltyCollectionTerritories lag is too high {{/is_alert}} {{#is_alert_recovery}} Offset lag is recovering {{/is_alert_recovery}} in {{[aws.kafka.sum_offset_lag].topic}} \n Notify: ${var.notification_endpoints}"
  query              = "avg(last_2h):aws.kafka.sum_offset_lag{environment:${var.environment} AND topic:${lower(var.ar_subaccount_royalty_collection_territories_topic)} AND (${local.ar_rights_filter_consumer_group_ids})} by {topic} > 2000"

  monitor_thresholds {
    critical = 2000
    warning  = 1000
  }

  notify_no_data = true
  no_data_timeframe = 60
  notify_audit = false
  renotify_interval = 30
  timeout_h = 0
  require_full_window = false

  tags = ["team:${var.team_name}"]
}