AWSTemplateFormatVersion: '2010-09-09'
Description: >
  Kinesis Data Streams and Firehose Cost Review Remediator.
  Review-only module that inventories Kinesis Data Streams and Firehose delivery
  streams across enabled Regions. It does not delete or modify resources.

Parameters:
  ProjectName:
    Type: String
    Default: kinesis-firehose-cost-review
  IncludeDetails:
    Type: String
    Default: 'true'
    AllowedValues: ['true', 'false']
  IncludeTags:
    Type: String
    Default: 'true'
    AllowedValues: ['true', 'false']
  MaxStreamsPerRegion:
    Type: Number
    Default: 200
    MinValue: 1
  MaxFirehoseStreamsPerRegion:
    Type: Number
    Default: 200
    MinValue: 1
  ShardWarningThreshold:
    Type: Number
    Default: 5
    MinValue: 1
  RetentionWarningHours:
    Type: Number
    Default: 24
    MinValue: 24
  ExcludedRegionsCsv:
    Type: String
    Default: ap-northeast-3
  ScheduleEnabled:
    Type: String
    Default: 'false'
    AllowedValues: ['true', 'false']
  ScheduleExpression:
    Type: String
    Default: rate(30 days)

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: KinesisFirehoseCostReviewPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - ec2:DescribeRegions
                Resource: '*'
              - Effect: Allow
                Action:
                  - kinesis:ListStreams
                  - kinesis:DescribeStreamSummary
                  - kinesis:ListShards
                  - kinesis:ListStreamConsumers
                  - kinesis:ListTagsForStream
                Resource: '*'
              - Effect: Allow
                Action:
                  - firehose:ListDeliveryStreams
                  - firehose:DescribeDeliveryStream
                  - firehose:ListTagsForDeliveryStream
                Resource: '*'

  ReviewFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: !Ref ProjectName
      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_STREAMS_PER_REGION: !Ref MaxStreamsPerRegion
          MAX_FIREHOSE_STREAMS_PER_REGION: !Ref MaxFirehoseStreamsPerRegion
          SHARD_WARNING_THRESHOLD: !Ref ShardWarningThreshold
          RETENTION_WARNING_HOURS: !Ref RetentionWarningHours
          EXCLUDED_REGIONS_CSV: !Ref ExcludedRegionsCsv
      Code:
        ZipFile: |
          import json, os
          from datetime import datetime, timezone
          import boto3
          from botocore.exceptions import ClientError, EndpointConnectionError

          def now(): return datetime.now(timezone.utc).isoformat()
          def b(v, d=False): return str(v if v is not None else d).lower() in ('1','true','yes','y')
          def csv(v): return [x.strip() for x in (v or '').split(',') if x.strip()]
          def err(svc, op, e):
              if isinstance(e, ClientError):
                  er=e.response.get('Error',{})
                  return {'service':svc,'operation':op,'error_code':er.get('Code','Unknown'),'error_message':er.get('Message',str(e))}
              return {'service':svc,'operation':op,'error_code':type(e).__name__,'error_message':str(e)}
          def denied(e):
              t=str(e).lower()
              if isinstance(e, ClientError):
                  er=e.response.get('Error',{}); t=(er.get('Code','')+' '+er.get('Message','')).lower()
              return any(x in t for x in ['accessdenied','not authorized','explicit deny','service control policy'])
          def benign(e):
              t=str(e).lower()
              if isinstance(e, ClientError):
                  er=e.response.get('Error',{}); t=(er.get('Code','')+' '+er.get('Message','')).lower()
              return any(x in t for x in ['resourcenotfound','streamnotfound','not found','unsupportedoperation'])
          def call(rr, svc, op, fn, default):
              try: return fn()
              except EndpointConnectionError as e:
                  rr['failed']+=1; rr['failures'].append(err(svc,op,e)); return default
              except ClientError as e:
                  if benign(e): return default
                  if denied(e): rr['access_denied']+=1
                  else: rr['failed']+=1
                  rr['failures'].append(err(svc,op,e)); return default
              except Exception as e:
                  rr['failed']+=1; rr['failures'].append(err(svc,op,e)); return default
          def regions(excluded):
              ec2=boto3.client('ec2',region_name='us-east-1')
              out=[]
              for r in ec2.describe_regions(AllRegions=True).get('Regions',[]):
                  if r.get('OptInStatus') in (None,'opt-in-not-required','opted-in') and r['RegionName'] not in excluded:
                      out.append(r['RegionName'])
              return sorted(out)
          def tags_dict(tags):
              return {t.get('Key') or t.get('key'): t.get('Value') or t.get('value') for t in tags or [] if t.get('Key') or t.get('key')}
          def list_kinesis(k, maxn):
              names=[]; token=None; start=None
              while len(names)<maxn:
                  kw={}
                  if token: kw['NextToken']=token
                  elif start: kw['ExclusiveStartStreamName']=start
                  r=k.list_streams(**kw); names += r.get('StreamNames',[])
                  token=r.get('NextToken')
                  if token: continue
                  if r.get('HasMoreStreams') and names: start=names[-1]; continue
                  break
              return names[:maxn]
          def list_shards(k, name):
              out=[]; token=None
              while True:
                  kw={'StreamName':name} if not token else {'NextToken':token}
                  r=k.list_shards(**kw); out += r.get('Shards',[]); token=r.get('NextToken')
                  if not token: return out
          def open_shards(shards):
              return sum(1 for s in shards or [] if not (s.get('SequenceNumberRange') or {}).get('EndingSequenceNumber'))
          def consumers(k, arn):
              out=[]; token=None
              while True:
                  kw={'StreamARN':arn}
                  if token: kw['NextToken']=token
                  r=k.list_stream_consumers(**kw); out += r.get('Consumers',[]); token=r.get('NextToken')
                  if not token: return out
          def firehose_names(fh,maxn):
              out=[]; start=None
              while len(out)<maxn:
                  kw={'Limit':min(100,maxn)}
                  if start: kw['ExclusiveStartDeliveryStreamName']=start
                  r=fh.list_delivery_streams(**kw); out += r.get('DeliveryStreamNames',[])
                  if not r.get('HasMoreDeliveryStreams') or not r.get('DeliveryStreamNames'): break
                  start=r['DeliveryStreamNames'][-1]
              return out[:maxn]
          def dests(desc):
              out=[]
              for d in desc.get('Destinations',[]) or []:
                  typ='Unknown'
                  for k in d.keys():
                      if k.endswith('DestinationDescription'):
                          typ=k.replace('DestinationDescription',''); break
                  out.append({'destination_id':d.get('DestinationId'),'destination_type':typ})
              return out
          def scan(region, settings):
              rr={k:0 for k in ['kinesis_streams','active_kinesis_streams','on_demand_streams','provisioned_streams','total_open_shards','streams_over_shard_threshold','streams_with_retention_over_threshold','streams_with_enhanced_monitoring','enhanced_monitoring_metric_count','enhanced_fanout_consumers','streams_with_consumers','streams_encrypted_with_kms','firehose_streams','active_firehose_streams','firehose_direct_put','firehose_kinesis_source','firehose_with_encryption_enabled','firehose_with_failure_description','firehose_destination_s3','firehose_destination_extended_s3','firehose_destination_redshift','firehose_destination_opensearch','firehose_destination_elasticsearch','firehose_destination_splunk','firehose_destination_http_endpoint','firehose_destination_other','access_denied','failed']}
              rr.update({'region':region,'review_flags':[],'kinesis_stream_inventory':[],'firehose_inventory':[],'failures':[]})
              k=boto3.client('kinesis',region_name=region); fh=boto3.client('firehose',region_name=region)
              names=call(rr,'Kinesis','ListStreams',lambda:list_kinesis(k,settings['max_streams']),[])
              rr['kinesis_streams']=len(names)
              for name in names:
                  s=call(rr,'Kinesis','DescribeStreamSummary',lambda n=name:k.describe_stream_summary(StreamName=n).get('StreamDescriptionSummary',{}),{})
                  if not s: continue
                  if s.get('StreamStatus')=='ACTIVE': rr['active_kinesis_streams']+=1
                  mode=(s.get('StreamModeDetails') or {}).get('StreamMode')
                  if mode=='ON_DEMAND': rr['on_demand_streams']+=1
                  elif mode=='PROVISIONED': rr['provisioned_streams']+=1
                  shards=call(rr,'Kinesis','ListShards',lambda n=name:list_shards(k,n),[])
                  osn=open_shards(shards); rr['total_open_shards']+=osn
                  if osn>=settings['shard_threshold']: rr['streams_over_shard_threshold']+=1
                  if int(s.get('RetentionPeriodHours') or 24)>settings['retention_hours']: rr['streams_with_retention_over_threshold']+=1
                  metrics=sorted({m for e in s.get('EnhancedMonitoring',[]) or [] for m in e.get('ShardLevelMetrics',[]) or []})
                  if metrics: rr['streams_with_enhanced_monitoring']+=1; rr['enhanced_monitoring_metric_count']+=len(metrics)
                  cons=call(rr,'Kinesis','ListStreamConsumers',lambda a=s.get('StreamARN'):consumers(k,a) if a else [],[])
                  rr['enhanced_fanout_consumers']+=len(cons)
                  if cons: rr['streams_with_consumers']+=1
                  if s.get('EncryptionType')=='KMS': rr['streams_encrypted_with_kms']+=1
                  tags={}
                  if settings['include_tags']:
                      tags=tags_dict(call(rr,'Kinesis','ListTagsForStream',lambda n=name:k.list_tags_for_stream(StreamName=n).get('Tags',[]),[]))
                  if settings['include_details']:
                      item={'stream_name':name,'stream_arn':s.get('StreamARN'),'stream_status':s.get('StreamStatus'),'stream_mode':mode,'retention_period_hours':s.get('RetentionPeriodHours'),'open_shard_count':osn,'reported_open_shard_count':s.get('OpenShardCount'),'enhanced_monitoring_metrics':metrics,'consumers':len(cons),'consumer_names':[c.get('ConsumerName') for c in cons],'encryption_type':s.get('EncryptionType'),'kms_key_id_present':bool(s.get('KeyId'))}
                      if settings['include_tags']: item['tags']=tags
                      rr['kinesis_stream_inventory'].append(item)
              fnames=call(rr,'Firehose','ListDeliveryStreams',lambda:firehose_names(fh,settings['max_firehose']),[])
              rr['firehose_streams']=len(fnames)
              for name in fnames:
                  d=call(rr,'Firehose','DescribeDeliveryStream',lambda n=name:fh.describe_delivery_stream(DeliveryStreamName=n),{})
                  ds=d.get('DeliveryStreamDescription',{}) if d else {}
                  if not ds: continue
                  if ds.get('DeliveryStreamStatus')=='ACTIVE': rr['active_firehose_streams']+=1
                  if ds.get('DeliveryStreamType')=='DirectPut': rr['firehose_direct_put']+=1
                  elif ds.get('DeliveryStreamType')=='KinesisStreamAsSource': rr['firehose_kinesis_source']+=1
                  enc=(ds.get('DeliveryStreamEncryptionConfiguration') or {}).get('Status')
                  if enc=='ENABLED': rr['firehose_with_encryption_enabled']+=1
                  if ds.get('FailureDescription'): rr['firehose_with_failure_description']+=1
                  dd=dests(ds)
                  for de in dd:
                      t=(de.get('destination_type') or '').lower()
                      if t=='s3': rr['firehose_destination_s3']+=1
                      elif t=='extendeds3': rr['firehose_destination_extended_s3']+=1
                      elif t=='redshift': rr['firehose_destination_redshift']+=1
                      elif 'opensearch' in t: rr['firehose_destination_opensearch']+=1
                      elif 'elastic' in t: rr['firehose_destination_elasticsearch']+=1
                      elif t=='splunk': rr['firehose_destination_splunk']+=1
                      elif 'http' in t: rr['firehose_destination_http_endpoint']+=1
                      else: rr['firehose_destination_other']+=1
                  tags={}
                  if settings['include_tags']:
                      tags=tags_dict(call(rr,'Firehose','ListTagsForDeliveryStream',lambda n=name:fh.list_tags_for_delivery_stream(DeliveryStreamName=n).get('Tags',[]),[]))
                  if settings['include_details']:
                      item={'delivery_stream_name':name,'delivery_stream_arn':ds.get('DeliveryStreamARN'),'delivery_stream_status':ds.get('DeliveryStreamStatus'),'delivery_stream_type':ds.get('DeliveryStreamType'),'delivery_destinations_count':len(dd),'delivery_destinations':dd,'source_kinesis_stream_arn_present':bool(((ds.get('Source') or {}).get('KinesisStreamSourceDescription') or {}).get('KinesisStreamARN')),'delivery_stream_encryption_status':enc,'failure_description':ds.get('FailureDescription')}
                      if settings['include_tags']: item['tags']=tags
                      rr['firehose_inventory'].append(item)
              if rr['kinesis_streams']>0: rr['review_flags'].append('kinesis_data_streams_present')
              if rr['provisioned_streams']>0: rr['review_flags'].append('provisioned_kinesis_streams_present')
              if rr['streams_over_shard_threshold']>0: rr['review_flags'].append('streams_over_shard_threshold')
              if rr['streams_with_retention_over_threshold']>0: rr['review_flags'].append('streams_with_extended_retention')
              if rr['streams_with_enhanced_monitoring']>0: rr['review_flags'].append('enhanced_monitoring_enabled')
              if rr['enhanced_fanout_consumers']>0: rr['review_flags'].append('enhanced_fanout_consumers_present')
              if rr['firehose_streams']>0: rr['review_flags'].append('firehose_delivery_streams_present')
              if rr['firehose_with_failure_description']>0: rr['review_flags'].append('firehose_streams_with_failure_description')
              return rr
          def handler(event, context):
              event=event or {}; excluded=set(csv(event.get('excluded_regions_csv') or os.environ.get('EXCLUDED_REGIONS_CSV','ap-northeast-3')))
              settings={'include_details':b(event.get('include_details'), os.environ.get('INCLUDE_DETAILS','true')), 'include_tags':b(event.get('include_tags'), os.environ.get('INCLUDE_TAGS','true')), 'max_streams':int(event.get('max_streams_per_region') or os.environ.get('MAX_STREAMS_PER_REGION','200')), 'max_firehose':int(event.get('max_firehose_streams_per_region') or os.environ.get('MAX_FIREHOSE_STREAMS_PER_REGION','200')), 'shard_threshold':int(event.get('shard_warning_threshold') or os.environ.get('SHARD_WARNING_THRESHOLD','5')), 'retention_hours':int(event.get('retention_warning_hours') or os.environ.get('RETENTION_WARNING_HOURS','24'))}
              totals={k:0 for k in ['regions_scanned','kinesis_streams','active_kinesis_streams','on_demand_streams','provisioned_streams','total_open_shards','streams_over_shard_threshold','streams_with_retention_over_threshold','streams_with_enhanced_monitoring','enhanced_monitoring_metric_count','enhanced_fanout_consumers','streams_with_consumers','streams_encrypted_with_kms','firehose_streams','active_firehose_streams','firehose_direct_put','firehose_kinesis_source','firehose_with_encryption_enabled','firehose_with_failure_description','firehose_destination_s3','firehose_destination_extended_s3','firehose_destination_redshift','firehose_destination_opensearch','firehose_destination_elasticsearch','firehose_destination_splunk','firehose_destination_http_endpoint','firehose_destination_other','regions_with_kinesis_streams','regions_with_firehose_streams','regions_with_review_flags','access_denied','failed']}
              out={'started_at':now(),'review_only':True,'include_details':settings['include_details'],'include_tags':settings['include_tags'],'excluded_regions':sorted(excluded),'thresholds':{'shard_warning_threshold':settings['shard_threshold'],'retention_warning_hours':settings['retention_hours']},'totals':totals,'region_results':[],'failures':[]}
              try: regs=regions(excluded)
              except Exception as e:
                  totals['failed']+=1; out['failures'].append(err('EC2','DescribeRegions',e)); out['ended_at']=now(); print(json.dumps(out,indent=2,default=str)); return json.loads(json.dumps(out,default=str))
              for r in regs:
                  rr=scan(r,settings)
                  for k in totals:
                      if k.startswith('regions_') or k=='regions_scanned': continue
                      totals[k]+=int(rr.get(k,0) or 0)
                  totals['regions_scanned']+=1
                  if rr['kinesis_streams']>0: totals['regions_with_kinesis_streams']+=1
                  if rr['firehose_streams']>0: totals['regions_with_firehose_streams']+=1
                  if rr['review_flags']: totals['regions_with_review_flags']+=1
                  out['failures'].extend([dict(f,region=r) for f in rr.get('failures',[])])
                  out['region_results'].append(rr)
              out['ended_at']=now(); print(json.dumps(out,indent=2,default=str)); return json.loads(json.dumps(out,default=str))

  ReviewScheduleRule:
    Type: AWS::Events::Rule
    Condition: IsScheduleEnabled
    Properties:
      Name: !Sub '${ProjectName}-schedule'
      Description: Scheduled run for Kinesis and Firehose Cost Review Remediator.
      ScheduleExpression: !Ref ScheduleExpression
      State: ENABLED
      Targets:
        - Arn: !GetAtt ReviewFunction.Arn
          Id: KinesisFirehoseCostReviewTarget
          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'
