AWSTemplateFormatVersion: '2010-09-09'
Description: >
  SNS and SQS Cost Review Remediator.
  Review-only module that inventories SNS topics/subscriptions and SQS queues
  across enabled Regions. Captures queue attributes, message counts, encryption,
  DLQ redrive policies, retention settings, long polling settings, and topic
  subscription counts. This module intentionally does not delete, purge, or modify
  SNS or SQS resources.

Parameters:
  ProjectName:
    Type: String
    Default: sns-sqs-cost-review
    Description: Short base name used for created resources.

  IncludeDetails:
    Type: String
    Default: 'true'
    AllowedValues: ['true', 'false']
    Description: Include detailed SNS/SQS inventory in Lambda response.

  IncludeTags:
    Type: String
    Default: 'true'
    AllowedValues: ['true', 'false']
    Description: Include tags where supported.

  MaxTopicsPerRegion:
    Type: Number
    Default: 200
    MinValue: 1
    Description: Maximum SNS topics to inspect per Region.

  MaxQueuesPerRegion:
    Type: Number
    Default: 500
    MinValue: 1
    Description: Maximum SQS queues to inspect per Region.

  QueueBacklogWarningThreshold:
    Type: Number
    Default: 1000
    MinValue: 0
    Description: Approximate visible message threshold for backlog review flag.

  ExcludedRegionsCsv:
    Type: String
    Default: ap-northeast-3
    Description: Comma-separated Regions to skip.

  ScheduleEnabled:
    Type: String
    Default: 'false'
    AllowedValues: ['true', 'false']
    Description: Whether to create an enabled EventBridge schedule.

  ScheduleExpression:
    Type: String
    Default: rate(30 days)
    Description: EventBridge schedule expression used when ScheduleEnabled is true.

Conditions:
  IsScheduleEnabled: !Equals [!Ref ScheduleEnabled, 'true']

Resources:
  ReviewRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: SnsSqsCostReviewPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Sid: RegionDiscovery
                Effect: Allow
                Action:
                  - ec2:DescribeRegions
                Resource: '*'
              - Sid: SnsRead
                Effect: Allow
                Action:
                  - sns:ListTopics
                  - sns:GetTopicAttributes
                  - sns:ListSubscriptionsByTopic
                  - sns:ListTagsForResource
                Resource: '*'
              - Sid: SqsRead
                Effect: Allow
                Action:
                  - sqs:ListQueues
                  - sqs:GetQueueAttributes
                  - sqs:ListQueueTags
                Resource: '*'

  ReviewFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: !Ref ProjectName
      Description: Review-only Lambda for SNS and SQS cost visibility.
      Runtime: python3.12
      Handler: index.handler
      Timeout: 900
      MemorySize: 512
      Role: !GetAtt ReviewRole.Arn
      Environment:
        Variables:
          INCLUDE_DETAILS: !Ref IncludeDetails
          INCLUDE_TAGS: !Ref IncludeTags
          MAX_TOPICS_PER_REGION: !Ref MaxTopicsPerRegion
          MAX_QUEUES_PER_REGION: !Ref MaxQueuesPerRegion
          QUEUE_BACKLOG_WARNING_THRESHOLD: !Ref QueueBacklogWarningThreshold
          EXCLUDED_REGIONS_CSV: !Ref ExcludedRegionsCsv
      Code:
        ZipFile: |
          import json
          import os
          from datetime import datetime, timezone
          from urllib.parse import urlparse

          import boto3
          from botocore.exceptions import ClientError, EndpointConnectionError

          def utc_now():
              return datetime.now(timezone.utc)

          def env_bool(name, default=False):
              return os.environ.get(name, str(default)).strip().lower() in ("1", "true", "yes", "y")

          def parse_bool(value, default=False):
              if value is None:
                  return default
              return str(value).strip().lower() in ("1", "true", "yes", "y")

          def csv_list(value):
              raw = value or ""
              return [x.strip() for x in raw.split(",") if x.strip()]

          def to_int(value, default=0):
              try:
                  if value is None or value == "":
                      return default
                  return int(value)
              except Exception:
                  return default

          def error_record(service, operation, err):
              if isinstance(err, ClientError):
                  return {
                      "service": service,
                      "operation": operation,
                      "error_code": err.response.get("Error", {}).get("Code", "Unknown"),
                      "error_message": err.response.get("Error", {}).get("Message", str(err))
                  }
              return {
                  "service": service,
                  "operation": operation,
                  "error_code": type(err).__name__,
                  "error_message": str(err)
              }

          def is_access_denied(err):
              text = str(err).lower()
              if isinstance(err, ClientError):
                  code = err.response.get("Error", {}).get("Code", "").lower()
                  msg = err.response.get("Error", {}).get("Message", "").lower()
                  text = f"{code} {msg}"
              return "accessdenied" in text or "not authorized" in text or "explicit deny" in text or "service control policy" in text

          def safe_call(region_result, service_name, operation_name, func, default):
              try:
                  return func()
              except EndpointConnectionError as e:
                  region_result["failed"] += 1
                  region_result["failures"].append(error_record(service_name, operation_name, e))
                  return default
              except ClientError as e:
                  if is_access_denied(e):
                      region_result["access_denied"] += 1
                  else:
                      region_result["failed"] += 1
                  region_result["failures"].append(error_record(service_name, operation_name, e))
                  return default
              except Exception as e:
                  region_result["failed"] += 1
                  region_result["failures"].append(error_record(service_name, operation_name, e))
                  return default

          def get_enabled_regions(excluded):
              ec2 = boto3.client("ec2", region_name="us-east-1")
              response = ec2.describe_regions(AllRegions=True)
              regions = []
              for r in response.get("Regions", []):
                  status = r.get("OptInStatus")
                  name = r.get("RegionName")
                  if status in (None, "opt-in-not-required", "opted-in") and name not in excluded:
                      regions.append(name)
              return sorted(regions)

          def paginate_sns_topics(client):
              items = []
              token = None
              while True:
                  kwargs = {}
                  if token:
                      kwargs["NextToken"] = token
                  resp = client.list_topics(**kwargs)
                  items.extend(resp.get("Topics", []) or [])
                  token = resp.get("NextToken")
                  if not token:
                      break
              return items

          def paginate_sns_subscriptions_by_topic(client, topic_arn):
              items = []
              token = None
              while True:
                  kwargs = {"TopicArn": topic_arn}
                  if token:
                      kwargs["NextToken"] = token
                  resp = client.list_subscriptions_by_topic(**kwargs)
                  items.extend(resp.get("Subscriptions", []) or [])
                  token = resp.get("NextToken")
                  if not token:
                      break
              return items

          def paginate_sqs_queues(client):
              urls = []
              token = None
              while True:
                  kwargs = {}
                  if token:
                      kwargs["NextToken"] = token
                  resp = client.list_queues(**kwargs)
                  urls.extend(resp.get("QueueUrls", []) or [])
                  token = resp.get("NextToken")
                  if not token:
                      break
              return urls

          def safe_json_loads(value):
              if not value:
                  return None
              try:
                  return json.loads(value)
              except Exception:
                  return None

          def queue_name_from_url(url):
              try:
                  return urlparse(url).path.strip("/").split("/")[-1]
              except Exception:
                  return url

          def sns_topic_item(topic_arn, attrs, subscriptions, tags, include_tags):
              protocol_counts = {}
              pending_subscriptions = 0
              confirmed_subscriptions = 0
              for sub in subscriptions:
                  proto = sub.get("Protocol", "unknown")
                  protocol_counts[proto] = protocol_counts.get(proto, 0) + 1
                  if sub.get("SubscriptionArn") == "PendingConfirmation":
                      pending_subscriptions += 1
                  else:
                      confirmed_subscriptions += 1

              item = {
                  "topic_arn": topic_arn,
                  "display_name": attrs.get("DisplayName"),
                  "owner": attrs.get("Owner"),
                  "subscriptions_confirmed": to_int(attrs.get("SubscriptionsConfirmed"), confirmed_subscriptions),
                  "subscriptions_pending": to_int(attrs.get("SubscriptionsPending"), pending_subscriptions),
                  "subscriptions_deleted": to_int(attrs.get("SubscriptionsDeleted"), 0),
                  "subscription_count_seen": len(subscriptions),
                  "protocol_counts": protocol_counts,
                  "kms_master_key_id": attrs.get("KmsMasterKeyId"),
                  "fifo_topic": attrs.get("FifoTopic") == "true",
                  "content_based_deduplication": attrs.get("ContentBasedDeduplication") == "true",
                  "delivery_policy_present": bool(attrs.get("DeliveryPolicy")),
                  "policy_present": bool(attrs.get("Policy")),
                  "tracing_config": attrs.get("TracingConfig")
              }
              if include_tags:
                  item["tags"] = tags or {}
              return item

          def sqs_queue_item(queue_url, attrs, tags, include_tags):
              redrive_policy = safe_json_loads(attrs.get("RedrivePolicy"))
              redrive_allow_policy = safe_json_loads(attrs.get("RedriveAllowPolicy"))
              visible = to_int(attrs.get("ApproximateNumberOfMessages"))
              not_visible = to_int(attrs.get("ApproximateNumberOfMessagesNotVisible"))
              delayed = to_int(attrs.get("ApproximateNumberOfMessagesDelayed"))

              item = {
                  "queue_url": queue_url,
                  "queue_name": queue_name_from_url(queue_url),
                  "queue_arn": attrs.get("QueueArn"),
                  "fifo_queue": attrs.get("FifoQueue") == "true",
                  "content_based_deduplication": attrs.get("ContentBasedDeduplication") == "true",
                  "sqs_managed_sse_enabled": attrs.get("SqsManagedSseEnabled") == "true",
                  "kms_master_key_id": attrs.get("KmsMasterKeyId"),
                  "kms_data_key_reuse_period_seconds": to_int(attrs.get("KmsDataKeyReusePeriodSeconds"), None),
                  "approximate_visible_messages": visible,
                  "approximate_not_visible_messages": not_visible,
                  "approximate_delayed_messages": delayed,
                  "approximate_total_messages": visible + not_visible + delayed,
                  "message_retention_period_seconds": to_int(attrs.get("MessageRetentionPeriod")),
                  "visibility_timeout_seconds": to_int(attrs.get("VisibilityTimeout")),
                  "receive_message_wait_time_seconds": to_int(attrs.get("ReceiveMessageWaitTimeSeconds")),
                  "delay_seconds": to_int(attrs.get("DelaySeconds")),
                  "maximum_message_size": to_int(attrs.get("MaximumMessageSize")),
                  "redrive_policy_present": redrive_policy is not None,
                  "redrive_policy": redrive_policy,
                  "redrive_allow_policy_present": redrive_allow_policy is not None,
                  "redrive_allow_policy": redrive_allow_policy,
                  "created_timestamp": attrs.get("CreatedTimestamp"),
                  "last_modified_timestamp": attrs.get("LastModifiedTimestamp"),
                  "policy_present": bool(attrs.get("Policy"))
              }
              if include_tags:
                  item["tags"] = tags or {}
              return item

          def scan_region(region, settings):
              rr = {
                  "region": region,
                  "sns_topics": 0,
                  "sns_fifo_topics": 0,
                  "sns_encrypted_topics": 0,
                  "sns_topics_with_delivery_policy": 0,
                  "sns_topics_with_subscriptions": 0,
                  "sns_subscriptions": 0,
                  "sns_pending_subscriptions": 0,
                  "sns_confirmed_subscriptions": 0,
                  "sns_protocol_counts": {},
                  "sqs_queues": 0,
                  "sqs_fifo_queues": 0,
                  "sqs_encrypted_queues": 0,
                  "sqs_sse_sqs_queues": 0,
                  "sqs_kms_encrypted_queues": 0,
                  "sqs_queues_with_dlq_redrive_policy": 0,
                  "sqs_queues_with_redrive_allow_policy": 0,
                  "sqs_queues_with_messages": 0,
                  "sqs_queues_over_backlog_threshold": 0,
                  "sqs_queues_with_long_polling": 0,
                  "sqs_queues_with_default_short_polling": 0,
                  "sqs_queues_with_max_retention": 0,
                  "sqs_approx_visible_messages": 0,
                  "sqs_approx_not_visible_messages": 0,
                  "sqs_approx_delayed_messages": 0,
                  "sqs_approx_total_messages": 0,
                  "review_flags": [],
                  "sns_topic_inventory": [],
                  "sqs_queue_inventory": [],
                  "access_denied": 0,
                  "failed": 0,
                  "failures": []
              }

              sns = boto3.client("sns", region_name=region)
              sqs = boto3.client("sqs", region_name=region)

              topics = safe_call(
                  rr,
                  "SNS",
                  "ListTopics",
                  lambda: paginate_sns_topics(sns),
                  []
              )[:settings["max_topics_per_region"]]

              rr["sns_topics"] = len(topics)

              for topic in topics:
                  topic_arn = topic.get("TopicArn")
                  if not topic_arn:
                      continue

                  attrs = safe_call(
                      rr,
                      "SNS",
                      "GetTopicAttributes",
                      lambda arn=topic_arn: sns.get_topic_attributes(TopicArn=arn).get("Attributes", {}),
                      {}
                  )

                  subscriptions = safe_call(
                      rr,
                      "SNS",
                      "ListSubscriptionsByTopic",
                      lambda arn=topic_arn: paginate_sns_subscriptions_by_topic(sns, arn),
                      []
                  )

                  tags = {}
                  if settings["include_tags"]:
                      tags = safe_call(
                          rr,
                          "SNS",
                          "ListTagsForResource",
                          lambda arn=topic_arn: {t.get("Key"): t.get("Value") for t in sns.list_tags_for_resource(ResourceArn=arn).get("Tags", []) if t.get("Key")},
                          {}
                      )

                  if attrs.get("FifoTopic") == "true":
                      rr["sns_fifo_topics"] += 1
                  if attrs.get("KmsMasterKeyId"):
                      rr["sns_encrypted_topics"] += 1
                  if attrs.get("DeliveryPolicy"):
                      rr["sns_topics_with_delivery_policy"] += 1

                  confirmed = to_int(attrs.get("SubscriptionsConfirmed"), 0)
                  pending = to_int(attrs.get("SubscriptionsPending"), 0)
                  deleted = to_int(attrs.get("SubscriptionsDeleted"), 0)
                  seen = len(subscriptions)

                  rr["sns_subscriptions"] += seen
                  rr["sns_pending_subscriptions"] += pending
                  rr["sns_confirmed_subscriptions"] += confirmed

                  if confirmed or pending or seen:
                      rr["sns_topics_with_subscriptions"] += 1

                  for sub in subscriptions:
                      proto = sub.get("Protocol", "unknown")
                      rr["sns_protocol_counts"][proto] = rr["sns_protocol_counts"].get(proto, 0) + 1

                  if settings["include_details"]:
                      rr["sns_topic_inventory"].append(
                          sns_topic_item(topic_arn, attrs, subscriptions, tags, settings["include_tags"])
                      )

              queue_urls = safe_call(
                  rr,
                  "SQS",
                  "ListQueues",
                  lambda: paginate_sqs_queues(sqs),
                  []
              )[:settings["max_queues_per_region"]]

              rr["sqs_queues"] = len(queue_urls)

              attribute_names = [
                  "All"
              ]

              for queue_url in queue_urls:
                  attrs = safe_call(
                      rr,
                      "SQS",
                      "GetQueueAttributes",
                      lambda url=queue_url: sqs.get_queue_attributes(QueueUrl=url, AttributeNames=attribute_names).get("Attributes", {}),
                      {}
                  )

                  tags = {}
                  if settings["include_tags"]:
                      tags = safe_call(
                          rr,
                          "SQS",
                          "ListQueueTags",
                          lambda url=queue_url: sqs.list_queue_tags(QueueUrl=url).get("Tags", {}),
                          {}
                      )

                  visible = to_int(attrs.get("ApproximateNumberOfMessages"))
                  not_visible = to_int(attrs.get("ApproximateNumberOfMessagesNotVisible"))
                  delayed = to_int(attrs.get("ApproximateNumberOfMessagesDelayed"))
                  total = visible + not_visible + delayed

                  rr["sqs_approx_visible_messages"] += visible
                  rr["sqs_approx_not_visible_messages"] += not_visible
                  rr["sqs_approx_delayed_messages"] += delayed
                  rr["sqs_approx_total_messages"] += total

                  if attrs.get("FifoQueue") == "true":
                      rr["sqs_fifo_queues"] += 1

                  if attrs.get("SqsManagedSseEnabled") == "true" or attrs.get("KmsMasterKeyId"):
                      rr["sqs_encrypted_queues"] += 1
                  if attrs.get("SqsManagedSseEnabled") == "true":
                      rr["sqs_sse_sqs_queues"] += 1
                  if attrs.get("KmsMasterKeyId"):
                      rr["sqs_kms_encrypted_queues"] += 1

                  if attrs.get("RedrivePolicy"):
                      rr["sqs_queues_with_dlq_redrive_policy"] += 1
                  if attrs.get("RedriveAllowPolicy"):
                      rr["sqs_queues_with_redrive_allow_policy"] += 1
                  if total > 0:
                      rr["sqs_queues_with_messages"] += 1
                  if visible >= settings["queue_backlog_warning_threshold"] and settings["queue_backlog_warning_threshold"] > 0:
                      rr["sqs_queues_over_backlog_threshold"] += 1

                  if to_int(attrs.get("ReceiveMessageWaitTimeSeconds")) > 0:
                      rr["sqs_queues_with_long_polling"] += 1
                  else:
                      rr["sqs_queues_with_default_short_polling"] += 1

                  if to_int(attrs.get("MessageRetentionPeriod")) >= 1209600:
                      rr["sqs_queues_with_max_retention"] += 1

                  if settings["include_details"]:
                      rr["sqs_queue_inventory"].append(
                          sqs_queue_item(queue_url, attrs, tags, settings["include_tags"])
                      )

              if rr["sns_topics"] > 0:
                  rr["review_flags"].append("sns_topics_present")
              if rr["sns_topics_with_subscriptions"] > 0:
                  rr["review_flags"].append("sns_topics_with_subscriptions_present")
              if rr["sns_pending_subscriptions"] > 0:
                  rr["review_flags"].append("sns_pending_subscriptions_present")
              if rr["sqs_queues"] > 0:
                  rr["review_flags"].append("sqs_queues_present")
              if rr["sqs_queues_with_messages"] > 0:
                  rr["review_flags"].append("sqs_queues_with_messages_present")
              if rr["sqs_queues_over_backlog_threshold"] > 0:
                  rr["review_flags"].append("sqs_queue_backlog_threshold_exceeded")
              if rr["sqs_queues_with_max_retention"] > 0:
                  rr["review_flags"].append("sqs_queues_with_max_retention_present")

              return rr

          def handler(event, context):
              event = event or {}
              excluded = set(csv_list(event.get("excluded_regions_csv") or os.environ.get("EXCLUDED_REGIONS_CSV", "ap-northeast-3")))
              settings = {
                  "include_details": parse_bool(event.get("include_details"), env_bool("INCLUDE_DETAILS", True)),
                  "include_tags": parse_bool(event.get("include_tags"), env_bool("INCLUDE_TAGS", True)),
                  "max_topics_per_region": int(event.get("max_topics_per_region") or os.environ.get("MAX_TOPICS_PER_REGION", "200")),
                  "max_queues_per_region": int(event.get("max_queues_per_region") or os.environ.get("MAX_QUEUES_PER_REGION", "500")),
                  "queue_backlog_warning_threshold": int(event.get("queue_backlog_warning_threshold") or os.environ.get("QUEUE_BACKLOG_WARNING_THRESHOLD", "1000"))
              }

              summary = {
                  "started_at": utc_now().isoformat(),
                  "review_only": True,
                  "include_details": settings["include_details"],
                  "include_tags": settings["include_tags"],
                  "excluded_regions": sorted(list(excluded)),
                  "queue_backlog_warning_threshold": settings["queue_backlog_warning_threshold"],
                  "totals": {
                      "regions_scanned": 0,
                      "sns_topics": 0,
                      "sns_fifo_topics": 0,
                      "sns_encrypted_topics": 0,
                      "sns_topics_with_delivery_policy": 0,
                      "sns_topics_with_subscriptions": 0,
                      "sns_subscriptions": 0,
                      "sns_pending_subscriptions": 0,
                      "sns_confirmed_subscriptions": 0,
                      "sqs_queues": 0,
                      "sqs_fifo_queues": 0,
                      "sqs_encrypted_queues": 0,
                      "sqs_sse_sqs_queues": 0,
                      "sqs_kms_encrypted_queues": 0,
                      "sqs_queues_with_dlq_redrive_policy": 0,
                      "sqs_queues_with_redrive_allow_policy": 0,
                      "sqs_queues_with_messages": 0,
                      "sqs_queues_over_backlog_threshold": 0,
                      "sqs_queues_with_long_polling": 0,
                      "sqs_queues_with_default_short_polling": 0,
                      "sqs_queues_with_max_retention": 0,
                      "sqs_approx_visible_messages": 0,
                      "sqs_approx_not_visible_messages": 0,
                      "sqs_approx_delayed_messages": 0,
                      "sqs_approx_total_messages": 0,
                      "regions_with_sns": 0,
                      "regions_with_sqs": 0,
                      "regions_with_review_flags": 0,
                      "access_denied": 0,
                      "failed": 0
                  },
                  "sns_protocol_counts": {},
                  "region_results": [],
                  "failures": []
              }

              try:
                  regions = get_enabled_regions(excluded)
              except Exception as e:
                  summary["totals"]["failed"] += 1
                  summary["failures"].append(error_record("EC2", "DescribeRegions", e))
                  summary["ended_at"] = utc_now().isoformat()
                  print(json.dumps(summary, indent=2, default=str))
                  return json.loads(json.dumps(summary, default=str))

              total_keys = [
                  k for k in summary["totals"].keys()
                  if k not in ("regions_scanned", "regions_with_sns", "regions_with_sqs", "regions_with_review_flags")
              ]

              for region in regions:
                  rr = scan_region(region, settings)

                  for key in total_keys:
                      summary["totals"][key] += int(rr.get(key, 0) or 0)

                  if rr.get("sns_topics", 0) > 0:
                      summary["totals"]["regions_with_sns"] += 1
                  if rr.get("sqs_queues", 0) > 0:
                      summary["totals"]["regions_with_sqs"] += 1
                  if rr.get("review_flags"):
                      summary["totals"]["regions_with_review_flags"] += 1

                  for proto, count in (rr.get("sns_protocol_counts") or {}).items():
                      summary["sns_protocol_counts"][proto] = summary["sns_protocol_counts"].get(proto, 0) + count

                  summary["totals"]["regions_scanned"] += 1
                  summary["failures"].extend([dict(f, region=region) for f in rr.get("failures", [])])
                  summary["region_results"].append(rr)

              summary["ended_at"] = utc_now().isoformat()
              print(json.dumps(summary, indent=2, default=str))
              return json.loads(json.dumps(summary, default=str))

  ReviewScheduleRule:
    Type: AWS::Events::Rule
    Condition: IsScheduleEnabled
    Properties:
      Name: !Sub '${ProjectName}-schedule'
      Description: Scheduled run for SNS/SQS Cost Review Remediator.
      ScheduleExpression: !Ref ScheduleExpression
      State: ENABLED
      Targets:
        - Arn: !GetAtt ReviewFunction.Arn
          Id: SnsSqsCostReviewTarget
          Input: '{"source":"eventbridge-schedule"}'

  ReviewSchedulePermission:
    Type: AWS::Lambda::Permission
    Condition: IsScheduleEnabled
    Properties:
      FunctionName: !Ref ReviewFunction
      Action: lambda:InvokeFunction
      Principal: events.amazonaws.com
      SourceArn: !GetAtt ReviewScheduleRule.Arn

Outputs:
  ReviewFunctionName:
    Description: Lambda function name.
    Value: !Ref ReviewFunction

  RemediatorFunctionName:
    Description: Lambda function name alias.
    Value: !Ref ReviewFunction

  ReviewFunctionArn:
    Description: Lambda function ARN.
    Value: !GetAtt ReviewFunction.Arn

  ReviewOnly:
    Description: This module is intentionally review-only.
    Value: 'true'
