AWSTemplateFormatVersion: '2010-09-09'
Description: AWS CodePipeline Cost Review Remediator - review-only inventory across enabled Regions.

Parameters:
  ScheduleEnabled:
    Type: String
    Default: 'false'
    AllowedValues: ['true', 'false']
  ScheduleExpression:
    Type: String
    Default: cron(0 13 ? * MON *)
  ExcludedRegionsCsv:
    Type: String
    Default: ap-northeast-3
  IncludePipelineDetails:
    Type: String
    Default: 'true'
    AllowedValues: ['true', 'false']
  IncludeExecutionHistory:
    Type: String
    Default: 'true'
    AllowedValues: ['true', 'false']
  IncludeTags:
    Type: String
    Default: 'true'
    AllowedValues: ['true', 'false']
  MaxPipelinesPerRegion:
    Type: Number
    Default: 200
    MinValue: 1
    MaxValue: 500
  MaxExecutionsPerPipeline:
    Type: Number
    Default: 20
    MinValue: 1
    MaxValue: 100

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

Resources:
  CodePipelineCostReviewRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: [lambda.amazonaws.com]
            Action: [sts:AssumeRole]
      Policies:
        - PolicyName: codepipeline-cost-review-policy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Sid: LambdaLogging
                Effect: Allow
                Action: [logs:CreateLogGroup, logs:CreateLogStream, logs:PutLogEvents]
                Resource: '*'
              - Sid: RegionDiscovery
                Effect: Allow
                Action: [ec2:DescribeRegions]
                Resource: '*'
              - Sid: CodePipelineReadOnly
                Effect: Allow
                Action:
                  - codepipeline:ListPipelines
                  - codepipeline:GetPipeline
                  - codepipeline:GetPipelineState
                  - codepipeline:ListPipelineExecutions
                  - codepipeline:ListTagsForResource
                Resource: '*'

  CodePipelineCostReviewFunction:
    Type: AWS::Lambda::Function
    Properties:
      Runtime: python3.12
      Handler: index.lambda_handler
      Role: !GetAtt CodePipelineCostReviewRole.Arn
      Timeout: 900
      MemorySize: 256
      Environment:
        Variables:
          EXCLUDED_REGIONS_CSV: !Ref ExcludedRegionsCsv
          INCLUDE_PIPELINE_DETAILS: !Ref IncludePipelineDetails
          INCLUDE_EXECUTION_HISTORY: !Ref IncludeExecutionHistory
          INCLUDE_TAGS: !Ref IncludeTags
          MAX_PIPELINES_PER_REGION: !Ref MaxPipelinesPerRegion
          MAX_EXECUTIONS_PER_PIPELINE: !Ref MaxExecutionsPerPipeline
      Code:
        ZipFile: |
          import boto3
          import os
          import traceback
          from datetime import datetime, timezone
          from botocore.exceptions import ClientError, EndpointConnectionError

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

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

          def as_int(value, default):
              try:
                  return int(value)
              except Exception:
                  return default

          def csv_set(value):
              return set(x.strip() for x in str(value or "").split(",") if x.strip())

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

          def list_pipelines(client, max_pipelines):
              items = []
              token = None
              while True:
                  kwargs = {}
                  if token:
                      kwargs["nextToken"] = token
                  response = client.list_pipelines(**kwargs)
                  items.extend(response.get("pipelines", []) or [])
                  if len(items) >= max_pipelines:
                      return items[:max_pipelines], True
                  token = response.get("nextToken")
                  if not token:
                      return items, False

          def list_executions(client, pipeline_name, max_executions):
              items = []
              token = None
              while True:
                  kwargs = {
                      "pipelineName": pipeline_name,
                      "maxResults": min(max_executions, 100)
                  }
                  if token:
                      kwargs["nextToken"] = token
                  response = client.list_pipeline_executions(**kwargs)
                  items.extend(response.get("pipelineExecutionSummaries", []) or [])
                  if len(items) >= max_executions:
                      return items[:max_executions], True
                  token = response.get("nextToken")
                  if not token:
                      return items, False

          def action_category_counts(stages):
              counts = {}
              providers = {}
              total = 0
              polling_source_actions = 0
              source_actions = 0

              for stage in stages:
                  for action in stage.get("actions", []) or []:
                      total += 1
                      action_type = action.get("actionTypeId") or {}
                      category = action_type.get("category", "UNKNOWN")
                      provider = action_type.get("provider", "UNKNOWN")
                      counts[category] = counts.get(category, 0) + 1
                      providers[provider] = providers.get(provider, 0) + 1

                      if category == "Source":
                          source_actions += 1
                          config = action.get("configuration") or {}
                          # Some source actions use PollForSourceChanges. This can create extra review noise/cost-adjacent polling.
                          if str(config.get("PollForSourceChanges", "")).lower() == "true":
                              polling_source_actions += 1

              return total, counts, providers, source_actions, polling_source_actions

          def summarize_execution(exe):
              return {
                  "pipeline_execution_id": exe.get("pipelineExecutionId"),
                  "status": exe.get("status"),
                  "start_time": str(exe.get("startTime")) if exe.get("startTime") else None,
                  "last_update_time": str(exe.get("lastUpdateTime")) if exe.get("lastUpdateTime") else None,
                  "source_revisions_count": len(exe.get("sourceRevisions") or []),
                  "trigger": exe.get("trigger")
              }

          def get_tags(client, arn):
              if not arn:
                  return {}, None
              try:
                  response = client.list_tags_for_resource(resourceArn=arn)
                  tags = response.get("tags", []) or []
                  return {t.get("key"): t.get("value") for t in tags if t.get("key")}, None
              except ClientError as e:
                  return {}, e.response.get("Error", {}).get("Code", "ClientError")
              except Exception as e:
                  return {}, str(e)

          def pipeline_arn(region, account_id, name):
              return "arn:aws:codepipeline:%s:%s:%s" % (region, account_id, name)

          def summarize_pipeline(region, account_id, client, pipeline_summary, include_execs, include_tags, max_execs):
              name = pipeline_summary.get("name")
              arn = pipeline_arn(region, account_id, name)
              item = {
                  "region": region,
                  "name": name,
                  "version": pipeline_summary.get("version"),
                  "created": str(pipeline_summary.get("created")) if pipeline_summary.get("created") else None,
                  "updated": str(pipeline_summary.get("updated")) if pipeline_summary.get("updated") else None,
                  "pipeline_type": None,
                  "execution_mode": None,
                  "stage_count": 0,
                  "action_count": 0,
                  "source_action_count": 0,
                  "polling_source_action_count": 0,
                  "artifact_store_type": None,
                  "artifact_stores_count": 0,
                  "recent_executions": [],
                  "recent_executions_truncated": False,
                  "recent_execution_count": 0,
                  "recent_succeeded_executions": 0,
                  "recent_failed_executions": 0,
                  "recent_stopped_executions": 0,
                  "recent_in_progress_executions": 0,
                  "review_flags": []
              }

              try:
                  pipe = client.get_pipeline(name=name).get("pipeline", {}) or {}
                  stages = pipe.get("stages") or []
                  total_actions, category_counts, provider_counts, source_count, polling_source_count = action_category_counts(stages)
                  artifact_store = pipe.get("artifactStore") or {}
                  artifact_stores = pipe.get("artifactStores") or {}

                  item.update({
                      "role_arn_present": bool(pipe.get("roleArn")),
                      "pipeline_type": pipe.get("pipelineType"),
                      "execution_mode": pipe.get("executionMode"),
                      "stage_count": len(stages),
                      "action_count": total_actions,
                      "action_category_counts": category_counts,
                      "action_provider_counts": provider_counts,
                      "source_action_count": source_count,
                      "polling_source_action_count": polling_source_count,
                      "artifact_store_type": artifact_store.get("type"),
                      "artifact_store_location_present": bool(artifact_store.get("location")),
                      "artifact_stores_count": len(artifact_stores) if isinstance(artifact_stores, dict) else 0
                  })

              except ClientError as e:
                  item["get_pipeline_error"] = e.response.get("Error", {}).get("Code", "ClientError")
              except Exception as e:
                  item["get_pipeline_error"] = str(e)

              try:
                  state = client.get_pipeline_state(name=name)
                  stage_states = state.get("stageStates", []) or []
                  item["stage_states_count"] = len(stage_states)
                  item["inbound_transition_disabled_count"] = sum(1 for s in stage_states if (s.get("inboundTransitionState") or {}).get("enabled") is False)
              except ClientError as e:
                  item["get_pipeline_state_error"] = e.response.get("Error", {}).get("Code", "ClientError")
              except Exception as e:
                  item["get_pipeline_state_error"] = str(e)

              if include_execs:
                  try:
                      execs, truncated = list_executions(client, name, max_execs)
                      item["recent_executions_truncated"] = truncated
                      item["recent_execution_count"] = len(execs)
                      item["recent_executions"] = [summarize_execution(e) for e in execs]

                      for e in execs:
                          status = e.get("status")
                          if status == "Succeeded":
                              item["recent_succeeded_executions"] += 1
                          elif status == "Failed":
                              item["recent_failed_executions"] += 1
                          elif status == "Stopped":
                              item["recent_stopped_executions"] += 1
                          elif status in ("InProgress", "Stopping"):
                              item["recent_in_progress_executions"] += 1

                  except ClientError as e:
                      item["executions_error"] = e.response.get("Error", {}).get("Code", "ClientError")
                  except Exception as e:
                      item["executions_error"] = str(e)

              if include_tags:
                  tags, error = get_tags(client, arn)
                  item["tags"] = tags
                  item["tag_count"] = len(tags)
                  if error:
                      item["tags_error"] = error

              flags = []
              if item.get("recent_execution_count", 0) > 0:
                  flags.append("recent_pipeline_activity_review")
              if item.get("recent_in_progress_executions", 0) > 0:
                  flags.append("in_progress_execution_review")
              if item.get("recent_failed_executions", 0) > 0:
                  flags.append("failed_execution_review")
              if item.get("recent_stopped_executions", 0) > 0:
                  flags.append("stopped_execution_review")
              if item.get("pipeline_type") == "V2":
                  flags.append("v2_pipeline_review")
              if item.get("polling_source_action_count", 0) > 0:
                  flags.append("polling_source_action_review")
              if item.get("stage_count", 0) >= 5:
                  flags.append("many_stages_review")
              if item.get("action_count", 0) >= 10:
                  flags.append("many_actions_review")
              if item.get("inbound_transition_disabled_count", 0) > 0:
                  flags.append("disabled_transition_review")

              item["review_flags"] = sorted(set(flags))
              return item

          def lambda_handler(event, context):
              event = event if isinstance(event, dict) else {}
              excluded = csv_set(event.get("excluded_regions_csv", os.environ.get("EXCLUDED_REGIONS_CSV", "ap-northeast-3")))
              include_details = as_bool(event.get("include_details", os.environ.get("INCLUDE_PIPELINE_DETAILS", "true")), True)
              include_execs = as_bool(event.get("include_execution_history", os.environ.get("INCLUDE_EXECUTION_HISTORY", "true")), True)
              include_tags = as_bool(event.get("include_tags", os.environ.get("INCLUDE_TAGS", "true")), True)
              max_pipelines = as_int(event.get("max_pipelines_per_region", os.environ.get("MAX_PIPELINES_PER_REGION", "200")), 200)
              max_execs = as_int(event.get("max_executions_per_pipeline", os.environ.get("MAX_EXECUTIONS_PER_PIPELINE", "20")), 20)

              result = {
                  "module": "aws_codepipeline_cost_review",
                  "review_only": True,
                  "started_at": now(),
                  "ended_at": None,
                  "excluded_regions": sorted(excluded),
                  "totals": {
                      "regions_scanned": 0,
                      "regions_skipped": 0,
                      "pipelines": 0,
                      "v1_pipelines": 0,
                      "v2_pipelines": 0,
                      "recent_executions_sampled": 0,
                      "recent_succeeded_executions": 0,
                      "recent_failed_executions": 0,
                      "recent_stopped_executions": 0,
                      "recent_in_progress_executions": 0,
                      "pipelines_with_recent_activity": 0,
                      "pipelines_with_failed_or_stopped_executions": 0,
                      "polling_source_actions": 0,
                      "source_actions": 0,
                      "stages": 0,
                      "actions": 0,
                      "disabled_transitions": 0,
                      "regions_with_pipelines": 0,
                      "regions_with_review_flags": 0,
                      "truncated_regions": 0,
                      "access_denied": 0,
                      "failed": 0
                  },
                  "regions": [],
                  "pipelines": []
              }

              try:
                  account_id = boto3.client("sts").get_caller_identity().get("Account")
                  regions = enabled_regions()
              except Exception as e:
                  result["fatal_error"] = str(e)
                  result["traceback"] = traceback.format_exc()
                  result["ended_at"] = now()
                  return result

              for region in regions:
                  if region in excluded:
                      result["totals"]["regions_skipped"] += 1
                      result["regions"].append({"region": region, "skipped": True, "reason": "excluded"})
                      continue

                  rec = {
                      "region": region,
                      "skipped": False,
                      "pipelines": 0,
                      "v1_pipelines": 0,
                      "v2_pipelines": 0,
                      "recent_executions_sampled": 0,
                      "recent_succeeded_executions": 0,
                      "recent_failed_executions": 0,
                      "recent_stopped_executions": 0,
                      "recent_in_progress_executions": 0,
                      "polling_source_actions": 0,
                      "source_actions": 0,
                      "stages": 0,
                      "actions": 0,
                      "disabled_transitions": 0,
                      "review_flags": [],
                      "truncated": False,
                      "error": None
                  }

                  try:
                      client = boto3.client("codepipeline", region_name=region)
                      pipeline_summaries, truncated = list_pipelines(client, max_pipelines)
                      rec["truncated"] = truncated
                      if truncated:
                          result["totals"]["truncated_regions"] += 1

                      details = []
                      for p in pipeline_summaries:
                          d = summarize_pipeline(region, account_id, client, p, include_execs, include_tags, max_execs)
                          details.append(d)

                          rec["pipelines"] += 1
                          result["totals"]["pipelines"] += 1

                          if d.get("pipeline_type") == "V2":
                              rec["v2_pipelines"] += 1
                              result["totals"]["v2_pipelines"] += 1
                          else:
                              rec["v1_pipelines"] += 1
                              result["totals"]["v1_pipelines"] += 1

                          rec["recent_executions_sampled"] += d.get("recent_execution_count", 0)
                          rec["recent_succeeded_executions"] += d.get("recent_succeeded_executions", 0)
                          rec["recent_failed_executions"] += d.get("recent_failed_executions", 0)
                          rec["recent_stopped_executions"] += d.get("recent_stopped_executions", 0)
                          rec["recent_in_progress_executions"] += d.get("recent_in_progress_executions", 0)
                          rec["polling_source_actions"] += d.get("polling_source_action_count", 0)
                          rec["source_actions"] += d.get("source_action_count", 0)
                          rec["stages"] += d.get("stage_count", 0)
                          rec["actions"] += d.get("action_count", 0)
                          rec["disabled_transitions"] += d.get("inbound_transition_disabled_count", 0)

                          rec["review_flags"].extend(d.get("review_flags", []))

                      for k in [
                          "recent_executions_sampled",
                          "recent_succeeded_executions",
                          "recent_failed_executions",
                          "recent_stopped_executions",
                          "recent_in_progress_executions",
                          "polling_source_actions",
                          "source_actions",
                          "stages",
                          "actions",
                          "disabled_transitions"
                      ]:
                          result["totals"][k] += rec[k]

                      rec["review_flags"] = sorted(set(rec["review_flags"]))

                      if rec["pipelines"] > 0:
                          result["totals"]["regions_with_pipelines"] += 1
                      if rec["review_flags"]:
                          result["totals"]["regions_with_review_flags"] += 1
                      if rec["recent_executions_sampled"] > 0:
                          result["totals"]["pipelines_with_recent_activity"] += sum(1 for d in details if d.get("recent_execution_count", 0) > 0)
                      result["totals"]["pipelines_with_failed_or_stopped_executions"] += sum(
                          1 for d in details if (d.get("recent_failed_executions", 0) + d.get("recent_stopped_executions", 0)) > 0
                      )

                      if include_details:
                          result["pipelines"].extend(details)

                      result["totals"]["regions_scanned"] += 1
                      result["regions"].append(rec)

                  except ClientError as e:
                      code = e.response.get("Error", {}).get("Code", "ClientError")
                      rec["error"] = code
                      if code in ("AccessDenied", "AccessDeniedException", "UnauthorizedOperation"):
                          result["totals"]["access_denied"] += 1
                      else:
                          result["totals"]["failed"] += 1
                      result["regions"].append(rec)

                  except EndpointConnectionError:
                      rec["error"] = "EndpointConnectionError"
                      result["totals"]["failed"] += 1
                      result["regions"].append(rec)

                  except Exception as e:
                      rec["error"] = str(e)
                      rec["traceback"] = traceback.format_exc()
                      result["totals"]["failed"] += 1
                      result["regions"].append(rec)

              result["ended_at"] = now()
              return result

  CodePipelineCostReviewSchedule:
    Type: AWS::Events::Rule
    Condition: CreateSchedule
    Properties:
      Description: Scheduled review-only AWS CodePipeline cost review.
      ScheduleExpression: !Ref ScheduleExpression
      State: ENABLED
      Targets:
        - Arn: !GetAtt CodePipelineCostReviewFunction.Arn
          Id: CodePipelineCostReviewFunctionTarget
          Input: '{"scheduled":true}'

  CodePipelineCostReviewSchedulePermission:
    Type: AWS::Lambda::Permission
    Condition: CreateSchedule
    Properties:
      FunctionName: !Ref CodePipelineCostReviewFunction
      Action: lambda:InvokeFunction
      Principal: events.amazonaws.com
      SourceArn: !GetAtt CodePipelineCostReviewSchedule.Arn

Outputs:
  ReviewFunctionName:
    Description: Lambda function name.
    Value: !Ref CodePipelineCostReviewFunction
  ReviewFunctionArn:
    Description: Lambda function ARN.
    Value: !GetAtt CodePipelineCostReviewFunction.Arn
  ReviewRoleArn:
    Description: IAM role ARN used by the review Lambda.
    Value: !GetAtt CodePipelineCostReviewRole.Arn
