import { Construct } from 'constructs';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as firehose from 'aws-cdk-lib/aws-kinesisfirehose';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as iam from 'aws-cdk-lib/aws-iam';
import { Tags, Duration, RemovalPolicy } from 'aws-cdk-lib';
import * as logs from 'aws-cdk-lib/aws-logs';
import { config } from '../../../config';
import * as ec2 from 'aws-cdk-lib/aws-ec2';

export const createRecEventsLoggerFirehose = (scope: Construct, recEventsStreamArn: string, accountId: string, accountStage: string, region: string) => {
  const cdkTagKey = 'CreatedBy';
  const cdkTagValue = 'CDK';

  // Reference the rec-events-logger S3 bucket
  const s3Bucket = s3.Bucket.fromBucketName(scope, 'RecEventsLoggerBucket', `rec-events-logger-${accountStage}`);

  const logGroupID = 'RecEventsLoggerFirehoseLogGroup';

  const lambdaID = 'RecEventsLoggerParserLambda';
  const lambdaNamePrefix = 'rec-events-logger-parser';

  const iamID = 'RecEventsFirehoseServiceRole';
  const iamNamePrefix = 'KinesisFirehose-rec-events-logger';

  const kinesisNamePrefix = 'rec-events';

  const firehoseID = 'RecEventsLoggerFirehose';
  const firehoseNamePrefix = 'rec-events-logger-firehose';

  // Create the CloudWatch log group
  const logGroup = new logs.LogGroup(scope, logGroupID, {
    logGroupName: `/aws/kinesisfirehose/${region}/${firehoseNamePrefix}`,
    removalPolicy: RemovalPolicy.DESTROY,
    retention: logs.RetentionDays.TWO_WEEKS,
  });

  // Create the Lambda function - dedicated processor for rec-events
  const recEventsLoggerParserLambda = new lambda.Function(scope, lambdaID, {
    runtime: lambda.Runtime.PYTHON_3_10,
    handler: 'main.lambda_handler',
    code: lambda.Code.fromAsset(`lambda/rec-events-logger-parser`), 
    timeout: Duration.seconds(60),
    functionName: `${lambdaNamePrefix}-lambda`,
    vpc: ec2.Vpc.fromVpcAttributes(scope, 'RecEventsVpc', {
      vpcId: config[accountStage].vpc,
      availabilityZones: ['us-east-2a', 'us-east-2b', 'us-east-2c'],
      privateSubnetIds: config[accountStage].lambdaSubnets,
    }),
    logRetention: logs.RetentionDays.TWO_WEEKS,
  });

  // Create the IAM Role
  const firehoseRole = new iam.Role(scope, iamID, {
    roleName: `${iamNamePrefix}-${region}`,
    assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
    inlinePolicies: {
      FirehosePolicy: new iam.PolicyDocument({
        statements: [
          new iam.PolicyStatement({
            effect: iam.Effect.ALLOW,
            actions: [
              's3:AbortMultipartUpload',
              's3:GetBucketLocation',
              's3:GetObject',
              's3:ListBucket',
              's3:ListBucketMultipartUploads',
              's3:PutObject',
            ],
            resources: [`${s3Bucket.bucketArn}`, `${s3Bucket.bucketArn}/*`],
          }),
          new iam.PolicyStatement({
            effect: iam.Effect.ALLOW,
            actions: ['lambda:InvokeFunction', 'lambda:GetFunctionConfiguration'],
            resources: [`${recEventsLoggerParserLambda.functionArn}`, `${recEventsLoggerParserLambda.functionArn}:$LATEST`],
          }),
          new iam.PolicyStatement({
            effect: iam.Effect.ALLOW,
            actions: ['logs:PutLogEvents'],
            resources: [
              `arn:aws:logs:${region}:${accountId}:log-group:/aws/kinesisfirehose/${region}/${firehoseNamePrefix}:log-stream:*`,
            ],
          }),
          new iam.PolicyStatement({
            effect: iam.Effect.ALLOW,
            actions: ['kinesis:DescribeStream', 'kinesis:GetShardIterator', 'kinesis:GetRecords', 'kinesis:ListShards'],
            resources: [recEventsStreamArn],
          }),
        ]
      }),
    },
  });

  // Create the Kinesis Firehose delivery stream
  const firehoseStream = new firehose.CfnDeliveryStream(scope, firehoseID, {
    deliveryStreamName: `${firehoseNamePrefix}-${region}`,
    deliveryStreamType: 'KinesisStreamAsSource',
    kinesisStreamSourceConfiguration: {
      kinesisStreamArn: recEventsStreamArn,
      roleArn: firehoseRole.roleArn,
    },
    extendedS3DestinationConfiguration: {
      bucketArn: s3Bucket.bucketArn,
      errorOutputPrefix: 'error/!{firehose:error-output-type}/!{timestamp:yyyy/MM/dd}/',
      customTimeZone: 'UTC',
      bufferingHints: {
        sizeInMBs: 64,
        intervalInSeconds: 60,
      },
      processingConfiguration: {
        enabled: true,
        processors: [
          {
            type: 'Lambda',
            parameters: [
              {
                parameterName: 'LambdaArn',
                parameterValue: recEventsLoggerParserLambda.functionArn,
              },
              {
                parameterName: 'BufferSizeInMBs',
                parameterValue: '1',
              },
              {
                parameterName: 'BufferIntervalInSeconds',
                parameterValue: '60',
              },
            ],
          },
          {
            type: 'AppendDelimiterToRecord',
            parameters: [
              {
                parameterName: 'Delimiter',
                parameterValue: '\\n',
              },
            ],
          },
        ],
      },
      prefix:
        'batched/!{partitionKeyFromLambda:EventVersion}/!{partitionKeyFromLambda:year}/!{partitionKeyFromLambda:month}/!{partitionKeyFromLambda:date}/!{partitionKeyFromLambda:hour}/',
      roleArn: firehoseRole.roleArn,
      cloudWatchLoggingOptions: {
        enabled: true,
        logGroupName: logGroup.logGroupName,
        logStreamName: 'DestinationDelivery',
      },
      dynamicPartitioningConfiguration: {
        enabled: true,
        retryOptions: {
          durationInSeconds: 300,
        },
      },
    },
  });

  Tags.of(recEventsLoggerParserLambda).add(cdkTagKey, cdkTagValue);
  Tags.of(firehoseStream).add(cdkTagKey, cdkTagValue);

  return firehoseStream;
};