import { Construct } from 'constructs';
import * as certManager from 'aws-cdk-lib/aws-certificatemanager';
import * as path from 'path';
import * as kinesis from 'aws-cdk-lib/aws-kinesis';
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 { multiRegionConfig } from '../../../multi-region-config';
import * as ec2 from 'aws-cdk-lib/aws-ec2';
export const createEventLoggerFirehose = async (scope: Construct, streamArn: string, accountId: string, accountStage: string, region: string) => {
  const apiName = 'event-logger';
  const cdkTagKey = 'CreatedBy';
  const cdkTagValue = 'CDK';

  // Reference the existing S3 bucket
  const prodBucketName = 'event-logger-temp';
  const stagingBucketName = 'event-logger-s3-temp';
  const s3Bucket = s3.Bucket.fromBucketName(scope, 'EventLoggerTempBucket', accountStage === 'staging' ? stagingBucketName : prodBucketName);

  // Create the CloudWatch log group
  const logGroup = new logs.LogGroup(scope, 'EventLoggerFirehoseLogGroup', {
    logGroupName: `/aws/kinesisfirehose/${region}/event-logger-firehose`,
    removalPolicy: RemovalPolicy.DESTROY, // Adjust as needed
    retention: logs.RetentionDays.TWO_WEEKS,
  });

  const vpc = ec2.Vpc.fromVpcAttributes(scope, 'Vpc', {
    vpcId: multiRegionConfig[accountStage][region].vpc,
    availabilityZones: multiRegionConfig[accountStage][region].availabilityZones,
    privateSubnetIds: multiRegionConfig[accountStage][region].lambdaSubnets,
  });
  // const lambdaSecurityGroup = new ec2.SecurityGroup(scope, 'EventLoggerParserLambdaSecurityGroup', {
  //   vpc,
  //   allowAllOutbound: true,
  //   description: 'Security group for the event logger parser Lambda function',
  // });

  // Create the Lambda function
  const eventLoggerParserLambda = new lambda.Function(scope, 'EventLoggerParserLambda', {
    runtime: lambda.Runtime.PYTHON_3_10,
    handler: 'main.lambda_handler',
    code: lambda.Code.fromAsset('lambda/event-logger-parser'),
    timeout: Duration.seconds(300),
    functionName: 'event-logger-parser-lambda',
    vpc: vpc,
    // securityGroups: [lambdaSecurityGroup],
    environment: {
      REDIS_HOST:
        accountStage === 'staging'
          ? 'valkey-cluster-default-001.ic34gd.0001.use2.cache.amazonaws.com'
          : 'valkey-cluster-session.9rcjcr.ng.0001.use2.cache.amazonaws.com',
      REDIS_PORT: '6379',
      REDIS_DB: '0',
    },
    logRetention: logs.RetentionDays.TWO_WEEKS,
  });

  // Create the IAM Role
  const firehoseRole = new iam.Role(scope, 'FirehoseServiceRole', {
    roleName: `KinesisFirehoseServiceRole-event-logger-${region}`,
    assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
    inlinePolicies: {
      FirehosePolicy: new iam.PolicyDocument({
        statements: [
          new iam.PolicyStatement({
            effect: iam.Effect.ALLOW,
            actions: ['glue:GetTable', 'glue:GetTableVersion', 'glue:GetTableVersions'],
            resources: [
              `arn:aws:glue:${region}:${accountId}:catalog`,
              `arn:aws:glue:${region}:${accountId}:database/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%`,
              `arn:aws:glue:${region}:${accountId}:table/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%`,
            ],
          }),
          new iam.PolicyStatement({
            effect: iam.Effect.ALLOW,
            actions: ['kafka:GetBootstrapBrokers', 'kafka:DescribeCluster', 'kafka:DescribeClusterV2', 'kafka-cluster:Connect'],
            resources: [`arn:aws:kafka:${region}:${accountId}:cluster/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%`],
          }),
          new iam.PolicyStatement({
            effect: iam.Effect.ALLOW,
            actions: ['kafka-cluster:DescribeTopic', 'kafka-cluster:DescribeTopicDynamicConfiguration', 'kafka-cluster:ReadData'],
            resources: [
              `arn:aws:kafka:${region}:${accountId}:topic/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%`,
            ],
          }),
          new iam.PolicyStatement({
            effect: iam.Effect.ALLOW,
            actions: ['kafka-cluster:DescribeGroup'],
            resources: [`arn:aws:kafka:${region}:${accountId}:group/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%/*`],
          }),
          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: [`${eventLoggerParserLambda.functionArn}`],
          }),
          new iam.PolicyStatement({
            effect: iam.Effect.ALLOW,
            actions: ['kms:GenerateDataKey', 'kms:Decrypt'],
            resources: [`arn:aws:kms:${region}:${accountId}:key/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%`],
            conditions: {
              StringEquals: {
                'kms:ViaService': `s3.${region}.amazonaws.com`,
              },
              StringLike: {
                'kms:EncryptionContext:aws:s3:arn': [`arn:aws:s3:::${accountId}/*`, `arn:aws:s3:::${accountId}`],
              },
            },
          }),
          new iam.PolicyStatement({
            effect: iam.Effect.ALLOW,
            actions: ['logs:PutLogEvents'],
            resources: [
              `arn:aws:logs:${region}:${accountId}:log-group:/aws/kinesisfirehose/event-logger-firehose-${region}:log-stream:*`,
              `arn:aws:logs:${region}:${accountId}:log-group:%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%:log-stream:*`,
            ],
          }),
          new iam.PolicyStatement({
            effect: iam.Effect.ALLOW,
            actions: ['kinesis:DescribeStream', 'kinesis:GetShardIterator', 'kinesis:GetRecords', 'kinesis:ListShards'],
            resources: [streamArn],
          }),
          new iam.PolicyStatement({
            effect: iam.Effect.ALLOW,
            actions: ['kms:Decrypt'],
            resources: [`arn:aws:kms:${region}:${accountId}:key/%FIREHOSE_POLICY_TEMPLATE_PLACEHOLDER%`],
            conditions: {
              StringEquals: {
                'kms:ViaService': `kinesis.${region}.amazonaws.com`,
              },
              StringLike: {
                'kms:EncryptionContext:aws:kinesis:arn': `arn:aws:kinesis:${region}:${accountId}:stream/event-logger-${region}`,
              },
            },
          }),
        ],
      }),
    },
  });

  // Create the Kinesis Firehose delivery stream
  const firehoseStream = new firehose.CfnDeliveryStream(scope, 'EventLoggerFirehose', {
    deliveryStreamName: `event-logger-firehose-${region}`,
    deliveryStreamType: 'KinesisStreamAsSource',
    kinesisStreamSourceConfiguration: {
      kinesisStreamArn: streamArn,
      roleArn: firehoseRole.roleArn,
    },
    extendedS3DestinationConfiguration: {
      bucketArn: s3Bucket.bucketArn,
      errorOutputPrefix: 'eerror/!{firehose:error-output-type}/!{timestamp:yyyy/MM/dd}/',
      customTimeZone: 'UTC',
      bufferingHints: {
        sizeInMBs: 128,
        intervalInSeconds: 300,
      },
      processingConfiguration: {
        enabled: true,
        processors: [
          {
            type: 'Lambda',
            parameters: [
              {
                parameterName: 'LambdaArn',
                parameterValue: eventLoggerParserLambda.functionArn,
              },
              {
                parameterName: 'BufferSizeInMBs',
                parameterValue: '2',
              },
              {
                parameterName: 'BufferIntervalInSeconds',
                parameterValue: '60',
              },
            ],
          },
          {
            type: 'AppendDelimiterToRecord',
            parameters: [
              {
                parameterName: 'Delimiter',
                parameterValue: '\\n',
              },
            ],
          },
        ],
      },
      prefix:
        '!{partitionKeyFromLambda:EventName}/!{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(eventLoggerParserLambda).add(cdkTagKey, cdkTagValue);
  Tags.of(firehoseStream).add(cdkTagKey, cdkTagValue);

  // return kinesisStream;
};