# Policy documents defined as data sources; attached as inline role policies
# via aws_iam_role_policy (generic-engineer-role lacks iam:CreatePolicy).

# -----------------------------------------------------------------------------
# S3 Read — DDB export files
# -----------------------------------------------------------------------------

data "aws_iam_policy_document" "s3_read_ddb_export" {
  statement {
    sid    = "ListBucket"
    effect = "Allow"
    actions = [
      "s3:ListBucket",
      "s3:GetBucketLocation",
    ]
    resources = [data.aws_s3_bucket.ddb_export.arn]
    condition {
      test     = "StringLike"
      variable = "s3:prefix"
      values   = ["${var.ddb_export_prefix}*"]
    }
  }

  statement {
    sid       = "GetObject"
    effect    = "Allow"
    actions   = ["s3:GetObject"]
    resources = ["${data.aws_s3_bucket.ddb_export.arn}/${var.ddb_export_prefix}*"]
  }
}

# -----------------------------------------------------------------------------
# SQS Send — Manifest Parser publishes file paths to queue
# -----------------------------------------------------------------------------

data "aws_iam_policy_document" "sqs_send_manifest_files" {
  statement {
    sid    = "SendMessage"
    effect = "Allow"
    actions = [
      "sqs:SendMessage",
      "sqs:GetQueueUrl",
    ]
    resources = [module.manifest_files_queue.queue_arn]
  }
}

# -----------------------------------------------------------------------------
# Inline Role Policies — attached to Lambda execution roles
# Using aws_iam_role_policy (not aws_iam_policy) because generic-engineer-role
# lacks iam:CreatePolicy but can use iam:PutRolePolicy.
# -----------------------------------------------------------------------------

resource "aws_iam_role_policy" "manifest_parser_s3_read" {
  name   = "S3-${var.environment}-${var.service_name}-read-ddb-export"
  role   = module.lambda_manifest_parser.lambda_role_id
  policy = data.aws_iam_policy_document.s3_read_ddb_export.json
}

resource "aws_iam_role_policy" "manifest_parser_sqs_send" {
  name   = "SQS-${var.environment}-${var.service_name}-send-manifest-files"
  role   = module.lambda_manifest_parser.lambda_role_id
  policy = data.aws_iam_policy_document.sqs_send_manifest_files.json
}

# -----------------------------------------------------------------------------
# SQS Consume — Collector Worker reads file paths from queue (Phase 3)
# -----------------------------------------------------------------------------

data "aws_iam_policy_document" "sqs_consume_manifest_files" {
  statement {
    sid    = "ConsumeMessages"
    effect = "Allow"
    actions = [
      "sqs:ReceiveMessage",
      "sqs:DeleteMessage",
      "sqs:GetQueueAttributes",
      "sqs:ChangeMessageVisibility",
    ]
    resources = [module.manifest_files_queue.queue_arn]
  }
}

# -----------------------------------------------------------------------------
# MSK Produce — Collector Worker writes to Kafka topic (Phase 3)
# Only created when var.msk_cluster_arn is set (empty in dev without MSK).
# -----------------------------------------------------------------------------

# Validation: ensure MSK cluster and topic ARNs are set consistently.
resource "null_resource" "msk_arn_validation" {
  lifecycle {
    precondition {
      condition     = (var.msk_cluster_arn == "" && var.msk_topic_arn == "") || (var.msk_cluster_arn != "" && var.msk_topic_arn != "")
      error_message = "msk_cluster_arn and msk_topic_arn must both be set or both empty."
    }
  }
}

data "aws_iam_policy_document" "msk_produce" {
  count = var.msk_cluster_arn != "" ? 1 : 0

  statement {
    sid       = "ConnectToCluster"
    effect    = "Allow"
    actions   = ["kafka-cluster:Connect"]
    resources = [var.msk_cluster_arn]
  }

  # MSK IAM topic resources use the format:
  # arn:aws:kafka:REGION:ACCOUNT:topic/CLUSTER_NAME/CLUSTER_UUID/TOPIC_NAME
  # Passed as a separate variable because topic ARN cannot be derived from cluster ARN.
  statement {
    sid    = "ProduceToTopic"
    effect = "Allow"
    actions = [
      "kafka-cluster:DescribeTopic",
      "kafka-cluster:WriteData",
    ]
    resources = [var.msk_topic_arn]
  }
}

# -----------------------------------------------------------------------------
# Inline Role Policies — Collector Worker Lambda
# -----------------------------------------------------------------------------

resource "aws_iam_role_policy" "collector_worker_s3_read" {
  name   = "S3-${var.environment}-${var.service_name}-collector-worker-read-ddb-export"
  role   = module.lambda_collector_worker.lambda_role_id
  policy = data.aws_iam_policy_document.s3_read_ddb_export.json
}

resource "aws_iam_role_policy" "collector_worker_sqs_consume" {
  name   = "SQS-${var.environment}-${var.service_name}-consume-manifest-files"
  role   = module.lambda_collector_worker.lambda_role_id
  policy = data.aws_iam_policy_document.sqs_consume_manifest_files.json
}

resource "aws_iam_role_policy" "collector_worker_msk_produce" {
  count = var.msk_cluster_arn != "" ? 1 : 0

  name   = "MSK-${var.environment}-${var.service_name}-produce"
  role   = module.lambda_collector_worker.lambda_role_id
  policy = data.aws_iam_policy_document.msk_produce[0].json
}