AWSTemplateFormatVersion: '2010-09-09'
Description: >
  AWS Glue Job / Crawler / Trigger Cost Review Remediator.
  Review-only module that inventories AWS Glue jobs, crawlers, triggers,
  development endpoints, databases, tables, and recent job/crawler activity
  across enabled Regions. This module intentionally does not delete, stop,
  disable, or modify Glue resources.

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

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

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

  MaxJobsPerRegion:
    Type: Number
    Default: 200
    MinValue: 1
    Description: Maximum Glue jobs to inspect per Region.

  MaxCrawlersPerRegion:
    Type: Number
    Default: 200
    MinValue: 1
    Description: Maximum Glue crawlers to inspect per Region.

  MaxJobRunsPerJob:
    Type: Number
    Default: 10
    MinValue: 0
    MaxValue: 200
    Description: Maximum recent Glue job runs to sample per job.

  HighDpuSecondsWarningThreshold:
    Type: Number
    Default: 3600
    MinValue: 1
    Description: Per-job sampled DPU seconds threshold that triggers a 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: GlueCostReviewPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Sid: RegionDiscovery
                Effect: Allow
                Action:
                  - ec2:DescribeRegions
                Resource: '*'
              - Sid: GlueRead
                Effect: Allow
                Action:
                  - glue:GetJobs
                  - glue:GetJob
                  - glue:GetJobRuns
                  - glue:GetCrawlers
                  - glue:GetCrawler
                  - glue:GetTriggers
                  - glue:GetTrigger
                  - glue:GetDevEndpoints
                  - glue:GetDatabases
                  - glue:GetTables
                  - glue:GetTags
                Resource: '*'

  ReviewFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: !Ref ProjectName
      Description: Review-only Lambda for AWS Glue 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_JOBS_PER_REGION: !Ref MaxJobsPerRegion
          MAX_CRAWLERS_PER_REGION: !Ref MaxCrawlersPerRegion
          MAX_JOB_RUNS_PER_JOB: !Ref MaxJobRunsPerJob
          HIGH_DPU_SECONDS_WARNING_THRESHOLD: !Ref HighDpuSecondsWarningThreshold
          EXCLUDED_REGIONS_CSV: !Ref ExcludedRegionsCsv
      Code:
        ZipFile: |
          import json
          import os
          from datetime import datetime, timezone

          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 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 is_not_found_or_unsupported(err):
              if isinstance(err, ClientError):
                  code = err.response.get("Error", {}).get("Code", "").lower()
                  text = (code + " " + err.response.get("Error", {}).get("Message", "")).lower()
                  return (
                      "entitynotfound" in text or
                      "resourcenotfound" in text or
                      "operationtimeout" in text and "crawler" in text or
                      "unsupportedoperation" in text
                  )
              return False

          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_not_found_or_unsupported(e):
                      return default
                  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 get_all_jobs(client, max_items):
              jobs = []
              token = None
              while True:
                  kwargs = {"MaxResults": min(100, max_items)}
                  if token:
                      kwargs["NextToken"] = token
                  response = client.get_jobs(**kwargs)
                  jobs.extend(response.get("Jobs", []) or [])
                  if len(jobs) >= max_items:
                      return jobs[:max_items]
                  token = response.get("NextToken")
                  if not token:
                      return jobs

          def get_all_crawlers(client, max_items):
              crawlers = []
              token = None
              while True:
                  kwargs = {"MaxResults": min(100, max_items)}
                  if token:
                      kwargs["NextToken"] = token
                  response = client.get_crawlers(**kwargs)
                  crawlers.extend(response.get("Crawlers", []) or [])
                  if len(crawlers) >= max_items:
                      return crawlers[:max_items]
                  token = response.get("NextToken")
                  if not token:
                      return crawlers

          def get_all_triggers(client):
              triggers = []
              token = None
              while True:
                  kwargs = {"MaxResults": 100}
                  if token:
                      kwargs["NextToken"] = token
                  response = client.get_triggers(**kwargs)
                  triggers.extend(response.get("Triggers", []) or [])
                  token = response.get("NextToken")
                  if not token:
                      return triggers

          def get_all_dev_endpoints(client):
              endpoints = []
              token = None
              while True:
                  kwargs = {}
                  if token:
                      kwargs["NextToken"] = token
                  response = client.get_dev_endpoints(**kwargs)
                  endpoints.extend(response.get("DevEndpoints", []) or [])
                  token = response.get("NextToken")
                  if not token:
                      return endpoints

          def get_database_count_and_tables(client):
              db_count = 0
              table_count = 0
              token = None
              while True:
                  kwargs = {}
                  if token:
                      kwargs["NextToken"] = token
                  response = client.get_databases(**kwargs)
                  dbs = response.get("DatabaseList", []) or []
                  db_count += len(dbs)
                  for db in dbs:
                      db_name = db.get("Name")
                      table_token = None
                      while True:
                          tkwargs = {"DatabaseName": db_name, "MaxResults": 100}
                          if table_token:
                              tkwargs["NextToken"] = table_token
                          tresp = client.get_tables(**tkwargs)
                          table_count += len(tresp.get("TableList", []) or [])
                          table_token = tresp.get("NextToken")
                          if not table_token:
                              break
                  token = response.get("NextToken")
                  if not token:
                      return db_count, table_count

          def get_job_runs(client, job_name, max_runs):
              if max_runs <= 0:
                  return []
              runs = []
              token = None
              while True:
                  kwargs = {"JobName": job_name, "MaxResults": min(200, max_runs)}
                  if token:
                      kwargs["NextToken"] = token
                  response = client.get_job_runs(**kwargs)
                  runs.extend(response.get("JobRuns", []) or [])
                  if len(runs) >= max_runs:
                      return runs[:max_runs]
                  token = response.get("NextToken")
                  if not token:
                      return runs

          def get_tags_for_arn(client, arn):
              if not arn:
                  return {}
              response = client.get_tags(ResourceArn=arn)
              return response.get("Tags", {}) or {}

          def job_run_stats(runs):
              stats = {
                  "sampled_job_runs": len(runs or []),
                  "sampled_succeeded_job_runs": 0,
                  "sampled_failed_job_runs": 0,
                  "sampled_stopped_job_runs": 0,
                  "sampled_timeout_job_runs": 0,
                  "sampled_running_job_runs": 0,
                  "sampled_dpu_seconds": 0.0,
                  "latest_sampled_job_run_started_on": None
              }
              latest = None
              for run in runs or []:
                  state = run.get("JobRunState")
                  if state == "SUCCEEDED":
                      stats["sampled_succeeded_job_runs"] += 1
                  elif state == "FAILED":
                      stats["sampled_failed_job_runs"] += 1
                  elif state == "STOPPED":
                      stats["sampled_stopped_job_runs"] += 1
                  elif state == "TIMEOUT":
                      stats["sampled_timeout_job_runs"] += 1
                  elif state in ("RUNNING", "STARTING", "STOPPING", "WAITING"):
                      stats["sampled_running_job_runs"] += 1

                  dpu = run.get("DPUSeconds")
                  if dpu is not None:
                      stats["sampled_dpu_seconds"] += float(dpu)

                  started = run.get("StartedOn")
                  if started and (latest is None or started > latest):
                      latest = started

              stats["latest_sampled_job_run_started_on"] = latest
              return stats

          def job_item(job, tags, run_stats, include_tags):
              item = {
                  "name": job.get("Name"),
                  "role_present": bool(job.get("Role")),
                  "command_name": (job.get("Command") or {}).get("Name"),
                  "glue_version": job.get("GlueVersion"),
                  "worker_type": job.get("WorkerType"),
                  "number_of_workers": job.get("NumberOfWorkers"),
                  "max_capacity": job.get("MaxCapacity"),
                  "max_retries": job.get("MaxRetries"),
                  "timeout": job.get("Timeout"),
                  "execution_class": job.get("ExecutionClass"),
                  "max_concurrent_runs": (job.get("ExecutionProperty") or {}).get("MaxConcurrentRuns"),
                  "created_on": job.get("CreatedOn"),
                  "last_modified_on": job.get("LastModifiedOn"),
                  "default_arguments_count": len(job.get("DefaultArguments", {}) or {}),
                  "non_overridable_arguments_count": len(job.get("NonOverridableArguments", {}) or {}),
                  **run_stats
              }
              if include_tags:
                  item["tags"] = tags or {}
              return item

          def crawler_item(crawler, tags, include_tags):
              sched = crawler.get("Schedule", {}) or {}
              item = {
                  "name": crawler.get("Name"),
                  "state": crawler.get("State"),
                  "role_present": bool(crawler.get("Role")),
                  "database_name": crawler.get("DatabaseName"),
                  "targets_count": sum(len((crawler.get("Targets", {}) or {}).get(k, []) or []) for k in ["S3Targets", "JdbcTargets", "MongoDBTargets", "DynamoDBTargets", "CatalogTargets", "DeltaTargets", "IcebergTargets", "HudiTargets"]),
                  "schedule_present": bool(sched),
                  "schedule_state": sched.get("State"),
                  "last_crawl_status": (crawler.get("LastCrawl") or {}).get("Status"),
                  "last_crawl_start_time": (crawler.get("LastCrawl") or {}).get("StartTime"),
                  "last_crawl_error_present": bool((crawler.get("LastCrawl") or {}).get("ErrorMessage")),
                  "recrawl_behavior": ((crawler.get("RecrawlPolicy") or {}).get("RecrawlBehavior")),
                  "schema_change_update_behavior": ((crawler.get("SchemaChangePolicy") or {}).get("UpdateBehavior")),
                  "schema_change_delete_behavior": ((crawler.get("SchemaChangePolicy") or {}).get("DeleteBehavior")),
                  "created_on": crawler.get("CreationTime"),
                  "last_updated": crawler.get("LastUpdated")
              }
              if include_tags:
                  item["tags"] = tags or {}
              return item

          def scan_region(region, settings):
              rr = {
                  "region": region,
                  "jobs": 0,
                  "jobs_with_recent_runs": 0,
                  "jobs_with_failed_or_timeout_runs": 0,
                  "jobs_over_dpu_seconds_threshold": 0,
                  "sampled_job_runs": 0,
                  "sampled_succeeded_job_runs": 0,
                  "sampled_failed_job_runs": 0,
                  "sampled_stopped_job_runs": 0,
                  "sampled_timeout_job_runs": 0,
                  "sampled_running_job_runs": 0,
                  "sampled_dpu_seconds": 0.0,
                  "crawler_count": 0,
                  "crawler_ready": 0,
                  "crawler_running": 0,
                  "crawler_scheduled": 0,
                  "crawler_last_run_failed": 0,
                  "trigger_count": 0,
                  "scheduled_triggers": 0,
                  "conditional_triggers": 0,
                  "on_demand_triggers": 0,
                  "active_triggers": 0,
                  "dev_endpoints": 0,
                  "databases": 0,
                  "tables": 0,
                  "review_flags": [],
                  "job_inventory": [],
                  "crawler_inventory": [],
                  "trigger_inventory": [],
                  "dev_endpoint_inventory": [],
                  "access_denied": 0,
                  "failed": 0,
                  "failures": []
              }

              glue = boto3.client("glue", region_name=region)

              jobs = safe_call(rr, "Glue", "GetJobs", lambda: get_all_jobs(glue, settings["max_jobs_per_region"]), [])
              rr["jobs"] = len(jobs)

              for job in jobs:
                  name = job.get("Name")
                  runs = safe_call(rr, "Glue", "GetJobRuns", lambda n=name: get_job_runs(glue, n, settings["max_job_runs_per_job"]), [])
                  stats = job_run_stats(runs)
                  for key in [
                      "sampled_job_runs", "sampled_succeeded_job_runs", "sampled_failed_job_runs",
                      "sampled_stopped_job_runs", "sampled_timeout_job_runs", "sampled_running_job_runs"
                  ]:
                      rr[key] += int(stats.get(key, 0) or 0)
                  rr["sampled_dpu_seconds"] += float(stats.get("sampled_dpu_seconds", 0.0) or 0.0)

                  if stats["sampled_job_runs"] > 0:
                      rr["jobs_with_recent_runs"] += 1
                  if stats["sampled_failed_job_runs"] > 0 or stats["sampled_timeout_job_runs"] > 0:
                      rr["jobs_with_failed_or_timeout_runs"] += 1
                  if stats["sampled_dpu_seconds"] >= settings["high_dpu_seconds_warning_threshold"]:
                      rr["jobs_over_dpu_seconds_threshold"] += 1

                  tags = {}
                  arn = job.get("JobArn")
                  if settings["include_tags"] and arn:
                      tags = safe_call(rr, "Glue", "GetTags", lambda a=arn: get_tags_for_arn(glue, a), {})

                  if settings["include_details"]:
                      rr["job_inventory"].append(job_item(job, tags, stats, settings["include_tags"]))

              crawlers = safe_call(rr, "Glue", "GetCrawlers", lambda: get_all_crawlers(glue, settings["max_crawlers_per_region"]), [])
              rr["crawler_count"] = len(crawlers)

              for crawler in crawlers:
                  state = crawler.get("State")
                  if state == "READY":
                      rr["crawler_ready"] += 1
                  elif state == "RUNNING":
                      rr["crawler_running"] += 1
                  sched = crawler.get("Schedule") or {}
                  if sched.get("State") == "SCHEDULED":
                      rr["crawler_scheduled"] += 1
                  if (crawler.get("LastCrawl") or {}).get("Status") == "FAILED":
                      rr["crawler_last_run_failed"] += 1

                  tags = {}
                  arn = crawler.get("CrawlerArn")
                  if settings["include_tags"] and arn:
                      tags = safe_call(rr, "Glue", "GetTags", lambda a=arn: get_tags_for_arn(glue, a), {})

                  if settings["include_details"]:
                      rr["crawler_inventory"].append(crawler_item(crawler, tags, settings["include_tags"]))

              triggers = safe_call(rr, "Glue", "GetTriggers", lambda: get_all_triggers(glue), [])
              rr["trigger_count"] = len(triggers)
              for trigger in triggers:
                  ttype = trigger.get("Type")
                  if ttype == "SCHEDULED":
                      rr["scheduled_triggers"] += 1
                  elif ttype == "CONDITIONAL":
                      rr["conditional_triggers"] += 1
                  elif ttype == "ON_DEMAND":
                      rr["on_demand_triggers"] += 1
                  if trigger.get("State") == "ACTIVATED":
                      rr["active_triggers"] += 1
                  if settings["include_details"]:
                      rr["trigger_inventory"].append({
                          "name": trigger.get("Name"),
                          "type": trigger.get("Type"),
                          "state": trigger.get("State"),
                          "schedule_present": bool(trigger.get("Schedule")),
                          "actions_count": len(trigger.get("Actions", []) or {}),
                          "workflow_name": trigger.get("WorkflowName")
                      })

              endpoints = safe_call(rr, "Glue", "GetDevEndpoints", lambda: get_all_dev_endpoints(glue), [])
              rr["dev_endpoints"] = len(endpoints)
              if settings["include_details"]:
                  rr["dev_endpoint_inventory"] = [
                      {
                          "name": ep.get("EndpointName"),
                          "status": ep.get("Status"),
                          "worker_type": ep.get("WorkerType"),
                          "number_of_workers": ep.get("NumberOfWorkers"),
                          "glue_version": ep.get("GlueVersion"),
                          "created_timestamp": ep.get("CreatedTimestamp")
                      }
                      for ep in endpoints
                  ]

              db_tables = safe_call(rr, "Glue", "GetDatabases/GetTables", lambda: get_database_count_and_tables(glue), (0, 0))
              rr["databases"], rr["tables"] = db_tables

              if rr["jobs"] > 0:
                  rr["review_flags"].append("glue_jobs_present")
              if rr["jobs_with_recent_runs"] > 0:
                  rr["review_flags"].append("glue_jobs_with_recent_runs")
              if rr["jobs_with_failed_or_timeout_runs"] > 0:
                  rr["review_flags"].append("failed_or_timeout_job_runs_sampled")
              if rr["jobs_over_dpu_seconds_threshold"] > 0:
                  rr["review_flags"].append("jobs_over_dpu_seconds_threshold")
              if rr["crawler_count"] > 0:
                  rr["review_flags"].append("glue_crawlers_present")
              if rr["crawler_scheduled"] > 0:
                  rr["review_flags"].append("scheduled_crawlers_present")
              if rr["crawler_last_run_failed"] > 0:
                  rr["review_flags"].append("failed_crawler_runs_present")
              if rr["active_triggers"] > 0:
                  rr["review_flags"].append("active_glue_triggers_present")
              if rr["dev_endpoints"] > 0:
                  rr["review_flags"].append("glue_dev_endpoints_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_jobs_per_region": int(event.get("max_jobs_per_region") or os.environ.get("MAX_JOBS_PER_REGION", "200")),
                  "max_crawlers_per_region": int(event.get("max_crawlers_per_region") or os.environ.get("MAX_CRAWLERS_PER_REGION", "200")),
                  "max_job_runs_per_job": int(event.get("max_job_runs_per_job") or os.environ.get("MAX_JOB_RUNS_PER_JOB", "10")),
                  "high_dpu_seconds_warning_threshold": float(event.get("high_dpu_seconds_warning_threshold") or os.environ.get("HIGH_DPU_SECONDS_WARNING_THRESHOLD", "3600"))
              }

              summary = {
                  "started_at": utc_now().isoformat(),
                  "review_only": True,
                  "include_details": settings["include_details"],
                  "include_tags": settings["include_tags"],
                  "excluded_regions": sorted(list(excluded)),
                  "job_run_sampling": {
                      "max_job_runs_per_job": settings["max_job_runs_per_job"],
                      "high_dpu_seconds_warning_threshold": settings["high_dpu_seconds_warning_threshold"]
                  },
                  "totals": {
                      "regions_scanned": 0,
                      "jobs": 0,
                      "jobs_with_recent_runs": 0,
                      "jobs_with_failed_or_timeout_runs": 0,
                      "jobs_over_dpu_seconds_threshold": 0,
                      "sampled_job_runs": 0,
                      "sampled_succeeded_job_runs": 0,
                      "sampled_failed_job_runs": 0,
                      "sampled_stopped_job_runs": 0,
                      "sampled_timeout_job_runs": 0,
                      "sampled_running_job_runs": 0,
                      "sampled_dpu_seconds": 0.0,
                      "crawler_count": 0,
                      "crawler_ready": 0,
                      "crawler_running": 0,
                      "crawler_scheduled": 0,
                      "crawler_last_run_failed": 0,
                      "trigger_count": 0,
                      "scheduled_triggers": 0,
                      "conditional_triggers": 0,
                      "on_demand_triggers": 0,
                      "active_triggers": 0,
                      "dev_endpoints": 0,
                      "databases": 0,
                      "tables": 0,
                      "regions_with_jobs": 0,
                      "regions_with_crawlers": 0,
                      "regions_with_triggers": 0,
                      "regions_with_dev_endpoints": 0,
                      "regions_with_review_flags": 0,
                      "access_denied": 0,
                      "failed": 0
                  },
                  "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_jobs", "regions_with_crawlers",
                      "regions_with_triggers", "regions_with_dev_endpoints", "regions_with_review_flags"
                  )
              ]

              for region in regions:
                  rr = scan_region(region, settings)
                  for key in total_keys:
                      value = rr.get(key, 0) or 0
                      if isinstance(summary["totals"][key], float):
                          summary["totals"][key] += float(value)
                      else:
                          summary["totals"][key] += int(value)

                  if rr.get("jobs", 0) > 0:
                      summary["totals"]["regions_with_jobs"] += 1
                  if rr.get("crawler_count", 0) > 0:
                      summary["totals"]["regions_with_crawlers"] += 1
                  if rr.get("trigger_count", 0) > 0:
                      summary["totals"]["regions_with_triggers"] += 1
                  if rr.get("dev_endpoints", 0) > 0:
                      summary["totals"]["regions_with_dev_endpoints"] += 1
                  if rr.get("review_flags"):
                      summary["totals"]["regions_with_review_flags"] += 1

                  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 Glue Cost Review Remediator.
      ScheduleExpression: !Ref ScheduleExpression
      State: ENABLED
      Targets:
        - Arn: !GetAtt ReviewFunction.Arn
          Id: GlueCostReviewTarget
          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'
