import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as kinesis from 'aws-cdk-lib/aws-kinesis';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as kinesisanalytics from 'aws-cdk-lib/aws-kinesisanalyticsv2';
import * as ec2 from 'aws-cdk-lib/aws-ec2';
import * as logs from 'aws-cdk-lib/aws-logs';

import { createRecEventsLoggerFirehose } from './firehose/rec-events-logger-firehose';
import { config } from '../../config';

export interface RecInfraStackProps extends cdk.StackProps {
  accountStage: string;
}

export class RecInfraStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props: RecInfraStackProps) {
    super(scope, id, props);

    const { accountStage } = props;

    // Import existing VPC
    const vpc = ec2.Vpc.fromLookup(this, 'SunoMainVpc', {
      vpcId: config[accountStage].vpc,
    });

    // Create Kinesis Data Stream
    const recEventsStream = new kinesis.Stream(this, 'RecEventsStream', {
      streamName: 'rec-events-stream',
      shardCount: 4,
      retentionPeriod: cdk.Duration.days(14),
    });

    // Create EFO Consumers for each Flink application
    const collaborativeFilteringConsumer = new kinesis.CfnStreamConsumer(this, 'CollaborativeFilteringConsumer', {
      streamArn: recEventsStream.streamArn,
      consumerName: `collaborative-filtering-consumer-${accountStage}`,
    });

    const coExistingLikeConsumer = new kinesis.CfnStreamConsumer(this, 'CoExistingLikeConsumer', {
      streamArn: recEventsStream.streamArn,
      consumerName: `co-existing-like-consumer-${accountStage}`,
    });

    const userLikeHistoryConsumer = new kinesis.CfnStreamConsumer(this, 'UserLikeHistoryConsumer', {
      streamArn: recEventsStream.streamArn,
      consumerName: `user-like-history-consumer-${accountStage}`,
    });

    const hookPositiveSignalConsumer = new kinesis.CfnStreamConsumer(this, 'HookPositiveSignalConsumer', {
      streamArn: recEventsStream.streamArn,
      consumerName: `hooks-positive-signal-consumer-${accountStage}`,
    });

    const hookCommentSignalConsumer = new kinesis.CfnStreamConsumer(this, 'HookCommentSignalConsumer', {
      streamArn: recEventsStream.streamArn,
      consumerName: `hooks-comment-signal-consumer-${accountStage}`,
    });

    const hookProfileViewSignalConsumer = new kinesis.CfnStreamConsumer(this, 'HookProfileViewSignalConsumer', {
      streamArn: recEventsStream.streamArn,
      consumerName: `hooks-profile-view-signal-consumer-${accountStage}`,
    });

    const hookCommentViewSignalConsumer = new kinesis.CfnStreamConsumer(this, 'HookCommentViewSignalConsumer', {
      streamArn: recEventsStream.streamArn,
      consumerName: `hooks-comment-view-signal-consumer-${accountStage}`,
    });

    const hookOmniplaySignalConsumer = new kinesis.CfnStreamConsumer(this, 'HookOmniplaySignalConsumer', {
      streamArn: recEventsStream.streamArn,
      consumerName: `hooks-omniplay-signal-consumer-${accountStage}`,
    });

    const hookRewatchSignalConsumer = new kinesis.CfnStreamConsumer(this, 'HookRewatchSignalConsumer', {
      streamArn: recEventsStream.streamArn,
      consumerName: `hooks-rewatch-signal-consumer-${accountStage}`,
    });

    const hookRemixSignalConsumer = new kinesis.CfnStreamConsumer(this, 'HookRemixSignalConsumer', {
      streamArn: recEventsStream.streamArn,
      consumerName: `hooks-remix-signal-consumer-${accountStage}`,
    });

    // Create Kinesis Firehose for rec-events logging to S3
    const recEventsLoggerFirehose = createRecEventsLoggerFirehose(
      this,
      recEventsStream.streamArn,
      cdk.Stack.of(this).account,
      accountStage,
      this.region
    );

    // Reference existing S3 bucket for Flink app code and checkpoints
    const flinkAppBucket = s3.Bucket.fromBucketName(this, 'FlinkAppBucket', `flink-app-bucket-${accountStage}-${cdk.Stack.of(this).account}`);

    // Create IAM Role for Kinesis Data Analytics
    const kinesisAnalyticsRole = new iam.Role(this, 'KinesisAnalyticsRole', {
      assumedBy: new iam.ServicePrincipal('kinesisanalytics.amazonaws.com'),
      roleName: `RecKinesisAnalyticsRole-${accountStage}`,
    });

    // Grant permissions to read from Kinesis stream
    recEventsStream.grantRead(kinesisAnalyticsRole);

    // Grant permissions to read from Kinesis stream
    kinesisAnalyticsRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: [
          'kinesis:DescribeStream',
          'kinesis:GetShardIterator',
          'kinesis:GetRecords',
          'kinesis:ListShards',
          'kinesis:DescribeStreamSummary',
          'kinesis:DescribeStreamConsumer',
          'kinesis:SubscribeToShard',
          'kinesis:RegisterStreamConsumer',
          'kinesis:DeregisterStreamConsumer',
          'kinesis:ListStreamConsumers',
        ],
        resources: ['*'],
      })
    );

    // Grant permissions to write CloudWatch logs
    kinesisAnalyticsRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: ['logs:CreateLogGroup', 'logs:CreateLogStream', 'logs:PutLogEvents', 'logs:DescribeLogGroups', 'logs:DescribeLogStreams'],
        resources: ['*'],
      })
    );

    // Grant permissions to access S3 bucket
    flinkAppBucket.grantReadWrite(kinesisAnalyticsRole);

    // PyFlink application S3 configuration (manual upload required)
    const pyflinkCodeBucket = flinkAppBucket.bucketName;
    const sharedCodeKey = `pyflink-apps/rec-flink-app.zip`;

    // Grant the Kinesis Analytics role permission to read the PyFlink code
    flinkAppBucket.grantRead(kinesisAnalyticsRole);

    // Add explicit permissions for reading the PyFlink code package
    kinesisAnalyticsRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: ['s3:GetObject', 's3:GetObjectVersion'],
        resources: [`${flinkAppBucket.bucketArn}/${sharedCodeKey}`, `${flinkAppBucket.bucketArn}/pyflink-apps/*`],
      })
    );

    // Grant VPC permissions for Redis access
    kinesisAnalyticsRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: [
          'ec2:CreateNetworkInterface',
          'ec2:DeleteNetworkInterface',
          'ec2:DescribeNetworkInterfaces',
          'ec2:DetachNetworkInterface',
          'ec2:AttachNetworkInterface',
          'ec2:ModifyNetworkInterfaceAttribute',
          'ec2:AssignPrivateIpAddresses',
          'ec2:UnassignPrivateIpAddresses',
          'ec2:DescribeVpcs',
          'ec2:DescribeSubnets',
          'ec2:DescribeSecurityGroups',
          'ec2:DescribeVpcEndpoints',
          'ec2:DescribeRouteTables',
          'ec2:DescribeAvailabilityZones',
          'ec2:CreateTags',
          'ec2:DescribeTags',
          'ec2:DescribeDhcpOptions',
          'ec2:DescribeInstances',
          'ec2:DescribeNetworkAcls',
        ],
        resources: ['*'],
      })
    );

    // Grant additional VPC permissions specifically for Kinesis Analytics service
    kinesisAnalyticsRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: ['ec2:CreateNetworkInterfacePermission', 'ec2:DescribeNetworkInterfacePermissions', 'ec2:DeleteNetworkInterfacePermission'],
        resources: [`arn:aws:ec2:${this.region}:${this.account}:network-interface/*`, `arn:aws:ec2:${this.region}:${this.account}:subnet/*`],
        conditions: {
          StringEquals: {
            'ec2:AuthorizedService': 'kinesisanalytics.amazonaws.com',
          },
        },
      })
    );

    // Grant permission to pass the role to Kinesis Analytics service
    kinesisAnalyticsRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: ['iam:PassRole'],
        resources: [kinesisAnalyticsRole.roleArn],
      })
    );

    // Grant Kinesis Data Analytics application management permissions
    kinesisAnalyticsRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: [
          'kinesisanalytics:DescribeApplication',
          'kinesisanalytics:CreateApplicationSnapshot',
          'kinesisanalytics:DeleteApplicationSnapshot',
          'kinesisanalytics:DescribeApplicationSnapshot',
          'kinesisanalytics:ListApplicationSnapshots',
        ],
        resources: ['*'],
      })
    );

    // Grant CloudWatch metrics permissions for monitoring
    kinesisAnalyticsRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: [
          'cloudwatch:PutMetricData',
          'cloudwatch:DescribeAlarms',
          'cloudwatch:DescribeAlarmsForMetric',
          'cloudwatch:GetMetricStatistics',
          'cloudwatch:ListMetrics',
        ],
        resources: ['*'],
      })
    );

    // Grant S3 permissions for Flink
    kinesisAnalyticsRole.addToPolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: ['s3:ListBucket', 's3:GetBucketLocation', 's3:GetBucketVersioning'],
        resources: [flinkAppBucket.bucketArn],
      })
    );

    // Create security group for Kinesis Analytics Application
    const kinesisAnalyticsSecurityGroup = new ec2.SecurityGroup(this, 'KinesisAnalyticsSG', {
      vpc,
      allowAllOutbound: true,
      description: 'RecInfraStack/KinesisAnalyticsApplication/KinesisAnalyticsSG',
      securityGroupName: `RecKinesisAnalyticsSG-${accountStage}`,
    });

    // Create CloudWatch Log Groups for each Kinesis Analytics Application
    const collaborativeFilteringLogGroup = new logs.LogGroup(this, 'CollaborativeFilteringLogGroup', {
      logGroupName: `/aws/kinesis-analytics/collaborative-filtering-app-${accountStage}`,
      retention: logs.RetentionDays.TWO_WEEKS,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const coExistingLikeLogGroup = new logs.LogGroup(this, 'CoExistingLikeLogGroup', {
      logGroupName: `/aws/kinesis-analytics/co-existing-like-app-${accountStage}`,
      retention: logs.RetentionDays.TWO_WEEKS,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const userLikeHistoryLogGroup = new logs.LogGroup(this, 'UserLikeHistoryLogGroup', {
      logGroupName: `/aws/kinesis-analytics/user-like-history-app-${accountStage}`,
      retention: logs.RetentionDays.TWO_WEEKS,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookPositiveSignalLogGroup = new logs.LogGroup(this, 'HookPositiveSignalLogGroup', {
      logGroupName: `/aws/kinesis-analytics/hooks-positive-signal-app-${accountStage}`,
      retention: logs.RetentionDays.TWO_WEEKS,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookCommentSignalLogGroup = new logs.LogGroup(this, 'HookCommentSignalLogGroup', {
      logGroupName: `/aws/kinesis-analytics/hooks-comment-signal-app-${accountStage}`,
      retention: logs.RetentionDays.TWO_WEEKS,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookProfileViewSignalLogGroup = new logs.LogGroup(this, 'HookProfileViewSignalLogGroup', {
      logGroupName: `/aws/kinesis-analytics/hooks-profile-view-signal-app-${accountStage}`,
      retention: logs.RetentionDays.TWO_WEEKS,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookCommentViewSignalLogGroup = new logs.LogGroup(this, 'HookCommentViewSignalLogGroup', {
      logGroupName: `/aws/kinesis-analytics/hooks-comment-view-signal-app-${accountStage}`,
      retention: logs.RetentionDays.TWO_WEEKS,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookOmniplaySignalLogGroup = new logs.LogGroup(this, 'HookOmniplaySignalLogGroup', {
      logGroupName: `/aws/kinesis-analytics/hooks-omniplay-signal-app-${accountStage}`,
      retention: logs.RetentionDays.TWO_WEEKS,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookRewatchSignalLogGroup = new logs.LogGroup(this, 'HookRewatchSignalLogGroup', {
      logGroupName: `/aws/kinesis-analytics/hooks-rewatch-signal-app-${accountStage}`,
      retention: logs.RetentionDays.TWO_WEEKS,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookRemixSignalLogGroup = new logs.LogGroup(this, 'HookRemixSignalLogGroup', {
      logGroupName: `/aws/kinesis-analytics/hooks-remix-signal-app-${accountStage}`,
      retention: logs.RetentionDays.TWO_WEEKS,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    // Create CloudWatch Log Streams
    const collaborativeFilteringLogStream = new logs.LogStream(this, 'CollaborativeFilteringLogStream', {
      logGroup: collaborativeFilteringLogGroup,
      logStreamName: 'collaborative-filtering-log-stream',
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const coExistingLikeLogStream = new logs.LogStream(this, 'CoExistingLikeLogStream', {
      logGroup: coExistingLikeLogGroup,
      logStreamName: 'co-existing-like-log-stream',
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

      const userLikeHistoryLogStream = new logs.LogStream(this, 'UserLikeHistoryLogStream', {
        logGroup: userLikeHistoryLogGroup,
        logStreamName: 'user-like-history-log-stream',
        removalPolicy: cdk.RemovalPolicy.DESTROY,
      });

    const hookPositiveSignalLogStream = new logs.LogStream(this, 'HookPositiveSignalLogStream', {
      logGroup: hookPositiveSignalLogGroup,
      logStreamName: 'hooks-positive-signal-log-stream',
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookCommentSignalLogStream = new logs.LogStream(this, 'HookCommentSignalLogStream', {
      logGroup: hookCommentSignalLogGroup,
      logStreamName: 'hooks-comment-signal-log-stream',
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookProfileViewSignalLogStream = new logs.LogStream(this, 'HookProfileViewSignalLogStream', {
      logGroup: hookProfileViewSignalLogGroup,
      logStreamName: 'hooks-profile-view-signal-log-stream',
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookCommentViewSignalLogStream = new logs.LogStream(this, 'HookCommentViewSignalLogStream', {
      logGroup: hookCommentViewSignalLogGroup,
      logStreamName: 'hooks-comment-view-signal-log-stream',
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookOmniplaySignalLogStream = new logs.LogStream(this, 'HookOmniplaySignalLogStream', {
      logGroup: hookOmniplaySignalLogGroup,
      logStreamName: 'hooks-omniplay-signal-log-stream',
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookRewatchSignalLogStream = new logs.LogStream(this, 'HookRewatchSignalLogStream', {
      logGroup: hookRewatchSignalLogGroup,
      logStreamName: 'hooks-rewatch-signal-log-stream',
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const hookRemixSignalLogStream = new logs.LogStream(this, 'HookRemixSignalLogStream', {
      logGroup: hookRemixSignalLogGroup,
      logStreamName: 'hooks-remix-signal-log-stream',
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    // Create Kinesis Data Analytics Applications

    // 1. Collaborative Filtering Application
    const collaborativeFilteringApp = new kinesisanalytics.CfnApplication(this, 'CollaborativeFilteringApp', {
      applicationName: `collaborative-filtering-app-${accountStage}`,
      runtimeEnvironment: 'FLINK-1_20',
      serviceExecutionRole: kinesisAnalyticsRole.roleArn,
      applicationConfiguration: {
        applicationCodeConfiguration: {
          codeContent: {
            s3ContentLocation: {
              bucketArn: flinkAppBucket.bucketArn,
              fileKey: sharedCodeKey,
            },
          },
          codeContentType: 'ZIPFILE',
        },
        flinkApplicationConfiguration: {
          parallelismConfiguration: {
            configurationType: 'CUSTOM',
            parallelism: 4,
            parallelismPerKpu: 1,
            autoScalingEnabled: false,
          },
          monitoringConfiguration: {
            configurationType: 'CUSTOM',
            logLevel: 'INFO',
            metricsLevel: 'APPLICATION',
          },
          checkpointConfiguration: {
            configurationType: 'CUSTOM',
            checkpointingEnabled: true,
            checkpointInterval: 600000, // 10 minutes
            minPauseBetweenCheckpoints: 5000,
          },
        },
        environmentProperties: {
          propertyGroups: [
            {
              propertyGroupId: 'kinesis.analytics.flink.run.options',
              propertyMap: {
                'python': 'app-collaborative-filtering/main.py',
                'jarfile': 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'flink.config',
              propertyMap: {
                'env.java.opts.taskmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'env.java.opts.jobmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'taskmanager.debug.memory.log': 'true',
                'taskmanager.debug.memory.log-interval': '10000',
              },
            },
            {
              propertyGroupId: 'rec.config',
              propertyMap: {
                'stream.arn': recEventsStream.streamArn,
                'redis.host':
                  accountStage === 'staging'
                    ? 'valkey-cluster-default.ic34gd.ng.0001.use2.cache.amazonaws.com'
                    : accountStage === 'prod'
                    ? 'valkey-cluster-recommendation.9rcjcr.ng.0001.use2.cache.amazonaws.com'
                    : 'localhost',
                'redis.port': '6379',
                'env.stage': accountStage,
              },
            },
            {
              propertyGroupId: 'pipeline.config',
              propertyMap: {
                'pipeline.jars': 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'logging.config',
              propertyMap: {
                'log4j2.logger.root.level': 'INFO',
                'log4j2.appender.console.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.appender.file.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.rootLogger.appenderRefs': 'console',
                'log4j2.rootLogger.appenderRef.console.ref': 'ConsoleAppender',
              },
            },
          ],
        },
        vpcConfigurations: [
          {
            securityGroupIds: [kinesisAnalyticsSecurityGroup.securityGroupId],
            subnetIds: config[accountStage].appSubnets,
          },
        ],
      },
      applicationMode: 'STREAMING',
    });

    // 2. Co-Existing Like Application
    const coExistingLikeApp = new kinesisanalytics.CfnApplication(this, 'CoExistingLikeApp', {
      applicationName: `co-existing-like-app-${accountStage}`,
      runtimeEnvironment: 'FLINK-1_20',
      serviceExecutionRole: kinesisAnalyticsRole.roleArn,
      applicationConfiguration: {
        applicationCodeConfiguration: {
          codeContent: {
            s3ContentLocation: {
              bucketArn: flinkAppBucket.bucketArn,
              fileKey: sharedCodeKey,
            },
          },
          codeContentType: 'ZIPFILE',
        },
        flinkApplicationConfiguration: {
          parallelismConfiguration: {
            configurationType: 'CUSTOM',
            parallelism: 4,
            parallelismPerKpu: 1,
            autoScalingEnabled: false,
          },
          monitoringConfiguration: {
            configurationType: 'CUSTOM',
            logLevel: 'INFO',
            metricsLevel: 'APPLICATION',
          },
          checkpointConfiguration: {
            configurationType: 'CUSTOM',
            checkpointingEnabled: true,
            checkpointInterval: 60000, // 1 minute
            minPauseBetweenCheckpoints: 5000,
          },
        },
        environmentProperties: {
          propertyGroups: [
            {
              propertyGroupId: 'kinesis.analytics.flink.run.options',
              propertyMap: {
                'python': 'app-co-existing-like/main.py',
                'jarfile': 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'flink.config',
              propertyMap: {
                'env.java.opts.taskmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'env.java.opts.jobmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'taskmanager.debug.memory.log': 'true',
                'taskmanager.debug.memory.log-interval': '10000',
              },
            },
            {
              propertyGroupId: 'rec.config',
              propertyMap: {
                'stream.arn': recEventsStream.streamArn,
                'redis.host':
                  accountStage === 'staging'
                    ? 'valkey-cluster-default.ic34gd.ng.0001.use2.cache.amazonaws.com'
                    : accountStage === 'prod'
                    ? 'valkey-cluster-recommendation.9rcjcr.ng.0001.use2.cache.amazonaws.com'
                    : 'localhost',
                'redis.port': '6379',
                'env.stage': accountStage,
              },
            },
            {
              propertyGroupId: 'pipeline.config',
              propertyMap: {
                'pipeline.jars': 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'logging.config',
              propertyMap: {
                'log4j2.logger.root.level': 'INFO',
                'log4j2.appender.console.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.appender.file.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.rootLogger.appenderRefs': 'console',
                'log4j2.rootLogger.appenderRef.console.ref': 'ConsoleAppender',
              },
            },
          ],
        },
        vpcConfigurations: [
          {
            securityGroupIds: [kinesisAnalyticsSecurityGroup.securityGroupId],
            subnetIds: config[accountStage].appSubnets,
          },
        ],
      },
      applicationMode: 'STREAMING',
    });

    // 3. User Like History Application
    const userLikeHistoryApp = new kinesisanalytics.CfnApplication(this, 'UserLikeHistoryApp', {
      applicationName: `user-like-history-app-${accountStage}`,
      runtimeEnvironment: 'FLINK-1_20',
      serviceExecutionRole: kinesisAnalyticsRole.roleArn,
      applicationConfiguration: {
        applicationCodeConfiguration: {
          codeContent: {
            s3ContentLocation: {
              bucketArn: flinkAppBucket.bucketArn,
              fileKey: sharedCodeKey,
            },
          },
          codeContentType: 'ZIPFILE',
        },
        flinkApplicationConfiguration: {
          parallelismConfiguration: {
            configurationType: 'CUSTOM',
            parallelism: 4,
            parallelismPerKpu: 1,
            autoScalingEnabled: false,
          },
          monitoringConfiguration: {
            configurationType: 'CUSTOM',
            logLevel: 'INFO',
            metricsLevel: 'APPLICATION',
          },
          checkpointConfiguration: {
            configurationType: 'CUSTOM',
            checkpointingEnabled: true,
            checkpointInterval: 60000, // 1 minute
            minPauseBetweenCheckpoints: 5000,
          },
        },
        environmentProperties: {
          propertyGroups: [
            {
              propertyGroupId: 'kinesis.analytics.flink.run.options',
              propertyMap: {
                python: 'app-user-like-history/main.py',
                jarfile: 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'flink.config',
              propertyMap: {
                'env.java.opts.taskmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'env.java.opts.jobmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'taskmanager.debug.memory.log': 'true',
                'taskmanager.debug.memory.log-interval': '10000',
              },
            },
            {
              propertyGroupId: 'rec.config',
              propertyMap: {
                'stream.arn': recEventsStream.streamArn,
                'redis.host':
                  accountStage === 'staging'
                    ? 'valkey-cluster-default.ic34gd.ng.0001.use2.cache.amazonaws.com'
                    : accountStage === 'prod'
                    ? 'valkey-cluster-recommendation.9rcjcr.ng.0001.use2.cache.amazonaws.com'
                    : 'localhost',
                'redis.port': '6379',
                'env.stage': accountStage,
              },
            },
            {
              propertyGroupId: 'pipeline.config',
              propertyMap: {
                'pipeline.jars': 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'logging.config',
              propertyMap: {
                'log4j2.logger.root.level': 'INFO',
                'log4j2.appender.console.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.appender.file.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.rootLogger.appenderRefs': 'console',
                'log4j2.rootLogger.appenderRef.console.ref': 'ConsoleAppender',
              },
            },
          ],
        },
        vpcConfigurations: [
          {
            securityGroupIds: [kinesisAnalyticsSecurityGroup.securityGroupId],
            subnetIds: config[accountStage].appSubnets,
          },
        ],
      },
      applicationMode: 'STREAMING',
    });

    const hookPositiveSignalApp = new kinesisanalytics.CfnApplication(this, 'HookPositiveSignalApp', {
      applicationName: `hooks-positive-signal-app-${accountStage}`,
      runtimeEnvironment: 'FLINK-1_20',
      serviceExecutionRole: kinesisAnalyticsRole.roleArn,
      applicationConfiguration: {
        applicationCodeConfiguration: {
          codeContent: {
            s3ContentLocation: {
              bucketArn: flinkAppBucket.bucketArn,
              fileKey: sharedCodeKey,
            },
          },
          codeContentType: 'ZIPFILE',
        },
        flinkApplicationConfiguration: {
          parallelismConfiguration: {
            configurationType: 'CUSTOM',
            parallelism: 4,
            parallelismPerKpu: 1,
            autoScalingEnabled: false,
          },
          monitoringConfiguration: {
            configurationType: 'CUSTOM',
            logLevel: 'INFO',
            metricsLevel: 'APPLICATION',
          },
          checkpointConfiguration: {
            configurationType: 'CUSTOM',
            checkpointingEnabled: true,
            checkpointInterval: 60000, // 1 minute
            minPauseBetweenCheckpoints: 5000,
          },
        },
        environmentProperties: {
          propertyGroups: [
            {
              propertyGroupId: 'kinesis.analytics.flink.run.options',
              propertyMap: {
                python: 'hooks-positive-signal/main.py',
                jarfile: 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'flink.config',
              propertyMap: {
                'env.java.opts.taskmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'env.java.opts.jobmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'taskmanager.debug.memory.log': 'true',
                'taskmanager.debug.memory.log-interval': '10000',
              },
            },
            {
              propertyGroupId: 'rec.config',
              propertyMap: {
                'stream.arn': recEventsStream.streamArn,
                'redis.host':
                  accountStage === 'staging'
                    ? 'valkey-cluster-default.ic34gd.ng.0001.use2.cache.amazonaws.com'
                    : accountStage === 'prod'
                    ? 'valkey-cluster-recommendation.9rcjcr.ng.0001.use2.cache.amazonaws.com'
                    : 'localhost',
                'redis.port': '6379',
                'env.stage': accountStage,
              },
            },
            {
              propertyGroupId: 'pipeline.config',
              propertyMap: {
                'pipeline.jars': 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'logging.config',
              propertyMap: {
                'log4j2.logger.root.level': 'INFO',
                'log4j2.appender.console.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.appender.file.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.rootLogger.appenderRefs': 'console',
                'log4j2.rootLogger.appenderRef.console.ref': 'ConsoleAppender',
              },
            },
          ],
        },
        vpcConfigurations: [
          {
            securityGroupIds: [kinesisAnalyticsSecurityGroup.securityGroupId],
            subnetIds: config[accountStage].appSubnets,
          },
        ],
      },
      applicationMode: 'STREAMING',
    });

    const hookCommentSignalApp = new kinesisanalytics.CfnApplication(this, 'HookCommentSignalApp', {
      applicationName: `hooks-comment-signal-app-${accountStage}`,
      runtimeEnvironment: 'FLINK-1_20',
      serviceExecutionRole: kinesisAnalyticsRole.roleArn,
      applicationConfiguration: {
        applicationCodeConfiguration: {
          codeContent: {
            s3ContentLocation: {
              bucketArn: flinkAppBucket.bucketArn,
              fileKey: sharedCodeKey,
            },
          },
          codeContentType: 'ZIPFILE',
        },
        flinkApplicationConfiguration: {
          parallelismConfiguration: {
            configurationType: 'CUSTOM',
            parallelism: 4,
            parallelismPerKpu: 1,
            autoScalingEnabled: false,
          },
          monitoringConfiguration: {
            configurationType: 'CUSTOM',
            logLevel: 'INFO',
            metricsLevel: 'APPLICATION',
          },
          checkpointConfiguration: {
            configurationType: 'CUSTOM',
            checkpointingEnabled: true,
            checkpointInterval: 60000, // 1 minute
            minPauseBetweenCheckpoints: 5000,
          },
        },
        environmentProperties: {
          propertyGroups: [
            {
              propertyGroupId: 'kinesis.analytics.flink.run.options',
              propertyMap: {
                python: 'hooks-comment-signal/main.py',
                jarfile: 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'flink.config',
              propertyMap: {
                'env.java.opts.taskmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'env.java.opts.jobmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'taskmanager.debug.memory.log': 'true',
                'taskmanager.debug.memory.log-interval': '10000',
              },
            },
            {
              propertyGroupId: 'rec.config',
              propertyMap: {
                'stream.arn': recEventsStream.streamArn,
                'redis.host':
                  accountStage === 'staging'
                    ? 'valkey-cluster-default.ic34gd.ng.0001.use2.cache.amazonaws.com'
                    : accountStage === 'prod'
                    ? 'valkey-cluster-recommendation.9rcjcr.ng.0001.use2.cache.amazonaws.com'
                    : 'localhost',
                'redis.port': '6379',
                'env.stage': accountStage,
              },
            },
            {
              propertyGroupId: 'pipeline.config',
              propertyMap: {
                'pipeline.jars': 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'logging.config',
              propertyMap: {
                'log4j2.logger.root.level': 'INFO',
                'log4j2.appender.console.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.appender.file.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.rootLogger.appenderRefs': 'console',
                'log4j2.rootLogger.appenderRef.console.ref': 'ConsoleAppender',
              },
            },
          ],
        },
        vpcConfigurations: [
          {
            securityGroupIds: [kinesisAnalyticsSecurityGroup.securityGroupId],
            subnetIds: config[accountStage].appSubnets,
          },
        ],
      },
      applicationMode: 'STREAMING',
    });

    const hookProfileViewSignalApp = new kinesisanalytics.CfnApplication(this, 'HookProfileViewSignalApp', {
      applicationName: `hooks-profile-view-signal-app-${accountStage}`,
      runtimeEnvironment: 'FLINK-1_20',
      serviceExecutionRole: kinesisAnalyticsRole.roleArn,
      applicationConfiguration: {
        applicationCodeConfiguration: {
          codeContent: {
            s3ContentLocation: {
              bucketArn: flinkAppBucket.bucketArn,
              fileKey: sharedCodeKey,
            },
          },
          codeContentType: 'ZIPFILE',
        },
        flinkApplicationConfiguration: {
          parallelismConfiguration: {
            configurationType: 'CUSTOM',
            parallelism: 4,
            parallelismPerKpu: 1,
            autoScalingEnabled: false,
          },
          monitoringConfiguration: {
            configurationType: 'CUSTOM',
            logLevel: 'INFO',
            metricsLevel: 'APPLICATION',
          },
          checkpointConfiguration: {
            configurationType: 'CUSTOM',
            checkpointingEnabled: true,
            checkpointInterval: 60000, // 1 minute
            minPauseBetweenCheckpoints: 5000,
          },
        },
        environmentProperties: {
          propertyGroups: [
            {
              propertyGroupId: 'kinesis.analytics.flink.run.options',
              propertyMap: {
                python: 'hooks-profile-view-signal/main.py',
                jarfile: 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'flink.config',
              propertyMap: {
                'env.java.opts.taskmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'env.java.opts.jobmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'taskmanager.debug.memory.log': 'true',
                'taskmanager.debug.memory.log-interval': '10000',
              },
            },
            {
              propertyGroupId: 'rec.config',
              propertyMap: {
                'stream.arn': recEventsStream.streamArn,
                'redis.host':
                  accountStage === 'staging'
                    ? 'valkey-cluster-default.ic34gd.ng.0001.use2.cache.amazonaws.com'
                    : accountStage === 'prod'
                    ? 'valkey-cluster-recommendation.9rcjcr.ng.0001.use2.cache.amazonaws.com'
                    : 'localhost',
                'redis.port': '6379',
                'env.stage': accountStage,
              },
            },
            {
              propertyGroupId: 'pipeline.config',
              propertyMap: {
                'pipeline.jars': 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'logging.config',
              propertyMap: {
                'log4j2.logger.root.level': 'INFO',
                'log4j2.appender.console.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.appender.file.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.rootLogger.appenderRefs': 'console',
                'log4j2.rootLogger.appenderRef.console.ref': 'ConsoleAppender',
              },
            },
          ],
        },
        vpcConfigurations: [
          {
            securityGroupIds: [kinesisAnalyticsSecurityGroup.securityGroupId],
            subnetIds: config[accountStage].appSubnets,
          },
        ],
      },
      applicationMode: 'STREAMING',
    });

    const hookCommentViewSignalApp = new kinesisanalytics.CfnApplication(this, 'HookCommentViewSignalApp', {
      applicationName: `hooks-comment-view-signal-app-${accountStage}`,
      runtimeEnvironment: 'FLINK-1_20',
      serviceExecutionRole: kinesisAnalyticsRole.roleArn,
      applicationConfiguration: {
        applicationCodeConfiguration: {
          codeContent: {
            s3ContentLocation: {
              bucketArn: flinkAppBucket.bucketArn,
              fileKey: sharedCodeKey,
            },
          },
          codeContentType: 'ZIPFILE',
        },
        flinkApplicationConfiguration: {
          parallelismConfiguration: {
            configurationType: 'CUSTOM',
            parallelism: 4,
            parallelismPerKpu: 1,
            autoScalingEnabled: false,
          },
          monitoringConfiguration: {
            configurationType: 'CUSTOM',
            logLevel: 'INFO',
            metricsLevel: 'APPLICATION',
          },
          checkpointConfiguration: {
            configurationType: 'CUSTOM',
            checkpointingEnabled: true,
            checkpointInterval: 60000, // 1 minute
            minPauseBetweenCheckpoints: 5000,
          },
        },
        environmentProperties: {
          propertyGroups: [
            {
              propertyGroupId: 'kinesis.analytics.flink.run.options',
              propertyMap: {
                python: 'hooks-comment-view-signal/main.py',
                jarfile: 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'flink.config',
              propertyMap: {
                'env.java.opts.taskmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'env.java.opts.jobmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'taskmanager.debug.memory.log': 'true',
                'taskmanager.debug.memory.log-interval': '10000',
              },
            },
            {
              propertyGroupId: 'rec.config',
              propertyMap: {
                'stream.arn': recEventsStream.streamArn,
                'redis.host':
                  accountStage === 'staging'
                    ? 'valkey-cluster-default.ic34gd.ng.0001.use2.cache.amazonaws.com'
                    : accountStage === 'prod'
                    ? 'valkey-cluster-recommendation.9rcjcr.ng.0001.use2.cache.amazonaws.com'
                    : 'localhost',
                'redis.port': '6379',
                'env.stage': accountStage,
              },
            },
            {
              propertyGroupId: 'pipeline.config',
              propertyMap: {
                'pipeline.jars': 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'logging.config',
              propertyMap: {
                'log4j2.logger.root.level': 'INFO',
                'log4j2.appender.console.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.appender.file.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.rootLogger.appenderRefs': 'console',
                'log4j2.rootLogger.appenderRef.console.ref': 'ConsoleAppender',
              },
            },
          ],
        },
        vpcConfigurations: [
          {
            securityGroupIds: [kinesisAnalyticsSecurityGroup.securityGroupId],
            subnetIds: config[accountStage].appSubnets,
          },
        ],
      },
      applicationMode: 'STREAMING',
    });

    const hookOmniplaySignalApp = new kinesisanalytics.CfnApplication(this, 'HookOmniplaySignalApp', {
      applicationName: `hooks-omniplay-signal-app-${accountStage}`,
      runtimeEnvironment: 'FLINK-1_20',
      serviceExecutionRole: kinesisAnalyticsRole.roleArn,
      applicationConfiguration: {
        applicationCodeConfiguration: {
          codeContent: {
            s3ContentLocation: {
              bucketArn: flinkAppBucket.bucketArn,
              fileKey: sharedCodeKey,
            },
          },
          codeContentType: 'ZIPFILE',
        },
        flinkApplicationConfiguration: {
          parallelismConfiguration: {
            configurationType: 'CUSTOM',
            parallelism: 4,
            parallelismPerKpu: 1,
            autoScalingEnabled: false,
          },
          monitoringConfiguration: {
            configurationType: 'CUSTOM',
            logLevel: 'INFO',
            metricsLevel: 'APPLICATION',
          },
          checkpointConfiguration: {
            configurationType: 'CUSTOM',
            checkpointingEnabled: true,
            checkpointInterval: 60000, // 1 minute
            minPauseBetweenCheckpoints: 5000,
          },
        },
        environmentProperties: {
          propertyGroups: [
            {
              propertyGroupId: 'kinesis.analytics.flink.run.options',
              propertyMap: {
                python: 'hooks-omniplay-signal/main.py',
                jarfile: 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'flink.config',
              propertyMap: {
                'env.java.opts.taskmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'env.java.opts.jobmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'taskmanager.debug.memory.log': 'true',
                'taskmanager.debug.memory.log-interval': '10000',
              },
            },
            {
              propertyGroupId: 'rec.config',
              propertyMap: {
                'stream.arn': recEventsStream.streamArn,
                'redis.host':
                  accountStage === 'staging'
                    ? 'valkey-cluster-default.ic34gd.ng.0001.use2.cache.amazonaws.com'
                    : accountStage === 'prod'
                    ? 'valkey-cluster-recommendation.9rcjcr.ng.0001.use2.cache.amazonaws.com'
                    : 'localhost',
                'redis.port': '6379',
                'env.stage': accountStage,
              },
            },
            {
              propertyGroupId: 'pipeline.config',
              propertyMap: {
                'pipeline.jars': 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'logging.config',
              propertyMap: {
                'log4j2.logger.root.level': 'INFO',
                'log4j2.appender.console.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.appender.file.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.rootLogger.appenderRefs': 'console',
                'log4j2.rootLogger.appenderRef.console.ref': 'ConsoleAppender',
              },
            },
          ],
        },
        vpcConfigurations: [
          {
            securityGroupIds: [kinesisAnalyticsSecurityGroup.securityGroupId],
            subnetIds: config[accountStage].appSubnets,
          },
        ],
      },
      applicationMode: 'STREAMING',
    });

    const hookRewatchSignalApp = new kinesisanalytics.CfnApplication(this, 'HookRewatchSignalApp', {
      applicationName: `hooks-rewatch-signal-app-${accountStage}`,
      runtimeEnvironment: 'FLINK-1_20',
      serviceExecutionRole: kinesisAnalyticsRole.roleArn,
      applicationConfiguration: {
        applicationCodeConfiguration: {
          codeContent: {
            s3ContentLocation: {
              bucketArn: flinkAppBucket.bucketArn,
              fileKey: sharedCodeKey,
            },
          },
          codeContentType: 'ZIPFILE',
        },
        flinkApplicationConfiguration: {
          parallelismConfiguration: {
            configurationType: 'CUSTOM',
            parallelism: 4,
            parallelismPerKpu: 1,
            autoScalingEnabled: false,
          },
          monitoringConfiguration: {
            configurationType: 'CUSTOM',
            logLevel: 'INFO',
            metricsLevel: 'APPLICATION',
          },
          checkpointConfiguration: {
            configurationType: 'CUSTOM',
            checkpointingEnabled: true,
            checkpointInterval: 60000, // 1 minute
            minPauseBetweenCheckpoints: 5000,
          },
        },
        environmentProperties: {
          propertyGroups: [
            {
              propertyGroupId: 'kinesis.analytics.flink.run.options',
              propertyMap: {
                python: 'hooks-rewatch-signal/main.py',
                jarfile: 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'flink.config',
              propertyMap: {
                'env.java.opts.taskmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'env.java.opts.jobmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'taskmanager.debug.memory.log': 'true',
                'taskmanager.debug.memory.log-interval': '10000',
              },
            },
            {
              propertyGroupId: 'rec.config',
              propertyMap: {
                'stream.arn': recEventsStream.streamArn,
                'redis.host':
                  accountStage === 'staging'
                    ? 'valkey-cluster-default.ic34gd.ng.0001.use2.cache.amazonaws.com'
                    : accountStage === 'prod'
                    ? 'valkey-cluster-recommendation.9rcjcr.ng.0001.use2.cache.amazonaws.com'
                    : 'localhost',
                'redis.port': '6379',
                'env.stage': accountStage,
              },
            },
            {
              propertyGroupId: 'pipeline.config',
              propertyMap: {
                'pipeline.jars': 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'logging.config',
              propertyMap: {
                'log4j2.logger.root.level': 'INFO',
                'log4j2.appender.console.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.appender.file.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.rootLogger.appenderRefs': 'console',
                'log4j2.rootLogger.appenderRef.console.ref': 'ConsoleAppender',
              },
            },
          ],
        },
        vpcConfigurations: [
          {
            securityGroupIds: [kinesisAnalyticsSecurityGroup.securityGroupId],
            subnetIds: config[accountStage].appSubnets,
          },
        ],
      },
      applicationMode: 'STREAMING',
    });

    const hookRemixSignalApp = new kinesisanalytics.CfnApplication(this, 'HookRemixSignalApp', {
      applicationName: `hooks-remix-signal-app-${accountStage}`,
      runtimeEnvironment: 'FLINK-1_20',
      serviceExecutionRole: kinesisAnalyticsRole.roleArn,
      applicationConfiguration: {
        applicationCodeConfiguration: {
          codeContent: {
            s3ContentLocation: {
              bucketArn: flinkAppBucket.bucketArn,
              fileKey: sharedCodeKey,
            },
          },
          codeContentType: 'ZIPFILE',
        },
        flinkApplicationConfiguration: {
          parallelismConfiguration: {
            configurationType: 'CUSTOM',
            parallelism: 4,
            parallelismPerKpu: 1,
            autoScalingEnabled: false,
          },
          monitoringConfiguration: {
            configurationType: 'CUSTOM',
            logLevel: 'INFO',
            metricsLevel: 'APPLICATION',
          },
          checkpointConfiguration: {
            configurationType: 'CUSTOM',
            checkpointingEnabled: true,
            checkpointInterval: 60000, // 1 minute
            minPauseBetweenCheckpoints: 5000,
          },
        },
        environmentProperties: {
          propertyGroups: [
            {
              propertyGroupId: 'kinesis.analytics.flink.run.options',
              propertyMap: {
                python: 'hooks-remix-signal/main.py',
                jarfile: 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'flink.config',
              propertyMap: {
                'env.java.opts.taskmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'env.java.opts.jobmanager': '-Dlog4j2.configurationFile=log4j2.properties',
                'taskmanager.debug.memory.log': 'true',
                'taskmanager.debug.memory.log-interval': '10000',
              },
            },
            {
              propertyGroupId: 'rec.config',
              propertyMap: {
                'stream.arn': recEventsStream.streamArn,
                'redis.host':
                  accountStage === 'staging'
                    ? 'valkey-cluster-default.ic34gd.ng.0001.use2.cache.amazonaws.com'
                    : accountStage === 'prod'
                    ? 'valkey-cluster-recommendation.9rcjcr.ng.0001.use2.cache.amazonaws.com'
                    : 'localhost',
                'redis.port': '6379',
                'env.stage': accountStage,
              },
            },
            {
              propertyGroupId: 'pipeline.config',
              propertyMap: {
                'pipeline.jars': 'lib/pyflink-dependencies.jar',
              },
            },
            {
              propertyGroupId: 'logging.config',
              propertyMap: {
                'log4j2.logger.root.level': 'INFO',
                'log4j2.appender.console.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.appender.file.layout.pattern': '%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n',
                'log4j2.rootLogger.appenderRefs': 'console',
                'log4j2.rootLogger.appenderRef.console.ref': 'ConsoleAppender',
              },
            },
          ],
        },
        vpcConfigurations: [
          {
            securityGroupIds: [kinesisAnalyticsSecurityGroup.securityGroupId],
            subnetIds: config[accountStage].appSubnets,
          },
        ],
      },
      applicationMode: 'STREAMING',
    });

    // Add CloudWatch Logging Options for each application
    const collaborativeFilteringCloudWatchLogging = new kinesisanalytics.CfnApplicationCloudWatchLoggingOption(
      this,
      'CollaborativeFilteringCloudWatchLogging',
      {
        applicationName: collaborativeFilteringApp.applicationName!,
        cloudWatchLoggingOption: {
          logStreamArn: `arn:aws:logs:${this.region}:${this.account}:log-group:${collaborativeFilteringLogGroup.logGroupName}:log-stream:${collaborativeFilteringLogStream.logStreamName}`,
        },
      }
    );

    const coExistingLikeCloudWatchLogging = new kinesisanalytics.CfnApplicationCloudWatchLoggingOption(this, 'CoExistingLikeCloudWatchLogging', {
      applicationName: coExistingLikeApp.applicationName!,
      cloudWatchLoggingOption: {
        logStreamArn: `arn:aws:logs:${this.region}:${this.account}:log-group:${coExistingLikeLogGroup.logGroupName}:log-stream:${coExistingLikeLogStream.logStreamName}`,
      },
    });

    const userLikeHistoryCloudWatchLogging = new kinesisanalytics.CfnApplicationCloudWatchLoggingOption(this, 'UserLikeHistoryCloudWatchLogging', {
      applicationName: userLikeHistoryApp.applicationName!,
      cloudWatchLoggingOption: {
        logStreamArn: `arn:aws:logs:${this.region}:${this.account}:log-group:${userLikeHistoryLogGroup.logGroupName}:log-stream:${userLikeHistoryLogStream.logStreamName}`,
      },
    });

    const hookPositiveSignalCloudWatchLogging = new kinesisanalytics.CfnApplicationCloudWatchLoggingOption(this, 'HookPositiveSignalCloudWatchLogging', {
      applicationName: hookPositiveSignalApp.applicationName!,
      cloudWatchLoggingOption: {
        logStreamArn: `arn:aws:logs:${this.region}:${this.account}:log-group:${hookPositiveSignalLogGroup.logGroupName}:log-stream:${hookPositiveSignalLogStream.logStreamName}`,
      },
    });

    const hookCommentSignalCloudWatchLogging = new kinesisanalytics.CfnApplicationCloudWatchLoggingOption(this, 'HookCommentSignalCloudWatchLogging', {
      applicationName: hookCommentSignalApp.applicationName!,
      cloudWatchLoggingOption: {
        logStreamArn: `arn:aws:logs:${this.region}:${this.account}:log-group:${hookCommentSignalLogGroup.logGroupName}:log-stream:${hookCommentSignalLogStream.logStreamName}`,
      },
    });

    const hookProfileViewSignalCloudWatchLogging = new kinesisanalytics.CfnApplicationCloudWatchLoggingOption(this, 'HookProfileViewSignalCloudWatchLogging', {
      applicationName: hookProfileViewSignalApp.applicationName!,
      cloudWatchLoggingOption: {
        logStreamArn: `arn:aws:logs:${this.region}:${this.account}:log-group:${hookProfileViewSignalLogGroup.logGroupName}:log-stream:${hookProfileViewSignalLogStream.logStreamName}`,
      },
    });

    const hookCommentViewSignalCloudWatchLogging = new kinesisanalytics.CfnApplicationCloudWatchLoggingOption(this, 'HookCommentViewSignalCloudWatchLogging', {
      applicationName: hookCommentViewSignalApp.applicationName!,
      cloudWatchLoggingOption: {
        logStreamArn: `arn:aws:logs:${this.region}:${this.account}:log-group:${hookCommentViewSignalLogGroup.logGroupName}:log-stream:${hookCommentViewSignalLogStream.logStreamName}`,
      },
    });

    const hookOmniplaySignalCloudWatchLogging = new kinesisanalytics.CfnApplicationCloudWatchLoggingOption(this, 'HookOmniplaySignalCloudWatchLogging', {
      applicationName: hookOmniplaySignalApp.applicationName!,
      cloudWatchLoggingOption: {
        logStreamArn: `arn:aws:logs:${this.region}:${this.account}:log-group:${hookOmniplaySignalLogGroup.logGroupName}:log-stream:${hookOmniplaySignalLogStream.logStreamName}`,
      },
    });

    const hookRewatchSignalCloudWatchLogging = new kinesisanalytics.CfnApplicationCloudWatchLoggingOption(this, 'HookRewatchSignalCloudWatchLogging', {
      applicationName: hookRewatchSignalApp.applicationName!,
      cloudWatchLoggingOption: {
        logStreamArn: `arn:aws:logs:${this.region}:${this.account}:log-group:${hookRewatchSignalLogGroup.logGroupName}:log-stream:${hookRewatchSignalLogStream.logStreamName}`,
      },
    });

    const hookRemixSignalCloudWatchLogging = new kinesisanalytics.CfnApplicationCloudWatchLoggingOption(this, 'HookRemixSignalCloudWatchLogging', {
      applicationName: hookRemixSignalApp.applicationName!,
      cloudWatchLoggingOption: {
        logStreamArn: `arn:aws:logs:${this.region}:${this.account}:log-group:${hookRemixSignalLogGroup.logGroupName}:log-stream:${hookRemixSignalLogStream.logStreamName}`,
      },
    });

    // Ensure applications are created before adding logging options
    collaborativeFilteringCloudWatchLogging.addDependsOn(collaborativeFilteringApp);
    coExistingLikeCloudWatchLogging.addDependsOn(coExistingLikeApp);
    userLikeHistoryCloudWatchLogging.addDependsOn(userLikeHistoryApp);
    hookPositiveSignalCloudWatchLogging.addDependsOn(hookPositiveSignalApp);
    hookCommentSignalCloudWatchLogging.addDependsOn(hookCommentSignalApp);
    hookProfileViewSignalCloudWatchLogging.addDependsOn(hookProfileViewSignalApp);
    hookCommentViewSignalCloudWatchLogging.addDependsOn(hookCommentViewSignalApp);
    hookOmniplaySignalCloudWatchLogging.addDependsOn(hookOmniplaySignalApp);
    hookRewatchSignalCloudWatchLogging.addDependsOn(hookRewatchSignalApp);
    hookRemixSignalCloudWatchLogging.addDependsOn(hookRemixSignalApp);

    // Ensure IAM policies are created before the Kinesis Analytics applications
    const policy = kinesisAnalyticsRole.node.findChild('DefaultPolicy') as iam.Policy;
    const cfnPolicy = policy.node.defaultChild as iam.CfnPolicy;
    collaborativeFilteringApp.addDependsOn(cfnPolicy);
    coExistingLikeApp.addDependsOn(cfnPolicy);
    userLikeHistoryApp.addDependsOn(cfnPolicy);
    hookPositiveSignalApp.addDependsOn(cfnPolicy);
    hookCommentSignalApp.addDependsOn(cfnPolicy);
    hookProfileViewSignalApp.addDependsOn(cfnPolicy);
    hookCommentViewSignalApp.addDependsOn(cfnPolicy);
    hookOmniplaySignalApp.addDependsOn(cfnPolicy);
    hookRewatchSignalApp.addDependsOn(cfnPolicy);
    hookRemixSignalApp.addDependsOn(cfnPolicy);

    // Ensure EFO Consumers are created before the Kinesis Analytics applications
    collaborativeFilteringApp.addDependsOn(collaborativeFilteringConsumer);
    coExistingLikeApp.addDependsOn(coExistingLikeConsumer);
    userLikeHistoryApp.addDependsOn(userLikeHistoryConsumer);
    hookPositiveSignalApp.addDependsOn(hookPositiveSignalConsumer);
    hookCommentSignalApp.addDependsOn(hookCommentSignalConsumer);
    hookProfileViewSignalApp.addDependsOn(hookProfileViewSignalConsumer);
    hookCommentViewSignalApp.addDependsOn(hookCommentViewSignalConsumer);
    hookOmniplaySignalApp.addDependsOn(hookOmniplaySignalConsumer);
    hookRewatchSignalApp.addDependsOn(hookRewatchSignalConsumer);
    hookRemixSignalApp.addDependsOn(hookRemixSignalConsumer);

    // Ensure CloudWatch resources are created before the logging options
    collaborativeFilteringCloudWatchLogging.addDependsOn(collaborativeFilteringLogGroup.node.defaultChild as logs.CfnLogGroup);
    collaborativeFilteringCloudWatchLogging.addDependsOn(collaborativeFilteringLogStream.node.defaultChild as logs.CfnLogStream);

    coExistingLikeCloudWatchLogging.addDependsOn(coExistingLikeLogGroup.node.defaultChild as logs.CfnLogGroup);
    coExistingLikeCloudWatchLogging.addDependsOn(coExistingLikeLogStream.node.defaultChild as logs.CfnLogStream);

    userLikeHistoryCloudWatchLogging.addDependsOn(userLikeHistoryLogGroup.node.defaultChild as logs.CfnLogGroup);
    userLikeHistoryCloudWatchLogging.addDependsOn(userLikeHistoryLogStream.node.defaultChild as logs.CfnLogStream);

    hookPositiveSignalCloudWatchLogging.addDependsOn(hookPositiveSignalLogGroup.node.defaultChild as logs.CfnLogGroup);
    hookPositiveSignalCloudWatchLogging.addDependsOn(hookPositiveSignalLogStream.node.defaultChild as logs.CfnLogStream);

    hookCommentSignalCloudWatchLogging.addDependsOn(hookCommentSignalLogGroup.node.defaultChild as logs.CfnLogGroup);
    hookCommentSignalCloudWatchLogging.addDependsOn(hookCommentSignalLogStream.node.defaultChild as logs.CfnLogStream);

    hookProfileViewSignalCloudWatchLogging.addDependsOn(hookProfileViewSignalLogGroup.node.defaultChild as logs.CfnLogGroup);
    hookProfileViewSignalCloudWatchLogging.addDependsOn(hookProfileViewSignalLogStream.node.defaultChild as logs.CfnLogStream);

    hookCommentViewSignalCloudWatchLogging.addDependsOn(hookCommentViewSignalLogGroup.node.defaultChild as logs.CfnLogGroup);
    hookCommentViewSignalCloudWatchLogging.addDependsOn(hookCommentViewSignalLogStream.node.defaultChild as logs.CfnLogStream);

    hookOmniplaySignalCloudWatchLogging.addDependsOn(hookOmniplaySignalLogGroup.node.defaultChild as logs.CfnLogGroup);
    hookOmniplaySignalCloudWatchLogging.addDependsOn(hookOmniplaySignalLogStream.node.defaultChild as logs.CfnLogStream);

    hookRewatchSignalCloudWatchLogging.addDependsOn(hookRewatchSignalLogGroup.node.defaultChild as logs.CfnLogGroup);
    hookRewatchSignalCloudWatchLogging.addDependsOn(hookRewatchSignalLogStream.node.defaultChild as logs.CfnLogStream);

    hookRemixSignalCloudWatchLogging.addDependsOn(hookRemixSignalLogGroup.node.defaultChild as logs.CfnLogGroup);
    hookRemixSignalCloudWatchLogging.addDependsOn(hookRemixSignalLogStream.node.defaultChild as logs.CfnLogStream);

    // Ensure IAM policies are available before CloudWatch logging options
    collaborativeFilteringCloudWatchLogging.addDependsOn(cfnPolicy);
    coExistingLikeCloudWatchLogging.addDependsOn(cfnPolicy);
    userLikeHistoryCloudWatchLogging.addDependsOn(cfnPolicy);
    hookPositiveSignalCloudWatchLogging.addDependsOn(cfnPolicy);
    hookCommentSignalCloudWatchLogging.addDependsOn(cfnPolicy);
    hookProfileViewSignalCloudWatchLogging.addDependsOn(cfnPolicy);
    hookCommentViewSignalCloudWatchLogging.addDependsOn(cfnPolicy);
    hookOmniplaySignalCloudWatchLogging.addDependsOn(cfnPolicy);
    hookRewatchSignalCloudWatchLogging.addDependsOn(cfnPolicy);
    hookRemixSignalCloudWatchLogging.addDependsOn(cfnPolicy);

    // Note: Redis dependency is managed by RedisCacheStack

    // Output important values
    new cdk.CfnOutput(this, 'RecEventsStreamName', {
      value: recEventsStream.streamName,
      description: 'Name of the Rec Events Kinesis Stream',
    });

    new cdk.CfnOutput(this, 'RecEventsStreamArn', {
      value: recEventsStream.streamArn,
      description: 'ARN of the Rec Events Kinesis Stream',
    });

    new cdk.CfnOutput(this, 'RecEventsLoggerFirehoseName', {
      value: recEventsLoggerFirehose.deliveryStreamName || 'rec-events-logger-firehose',
      description: 'Name of the Rec Events Logger Kinesis Firehose',
    });

    new cdk.CfnOutput(this, 'PyFlinkCodeBucket', {
      value: pyflinkCodeBucket,
      description: 'S3 bucket for PyFlink application code (manual upload required)',
    });

    new cdk.CfnOutput(this, 'SharedFlinkCodeKey', {
      value: sharedCodeKey,
      description: 'S3 object key for shared Flink applications code (manual upload required)',
    });

    new cdk.CfnOutput(this, 'CollaborativeFilteringAppName', {
      value: collaborativeFilteringApp.applicationName!,
      description: 'Name of the Collaborative Filtering Application',
    });

    new cdk.CfnOutput(this, 'CoExistingLikeAppName', {
      value: coExistingLikeApp.applicationName!,
      description: 'Name of the Co-Existing Like Application',
    });

    new cdk.CfnOutput(this, 'UserLikeHistoryAppName', {
      value: userLikeHistoryApp.applicationName!,
      description: 'Name of the User Like History Application',
    });

    new cdk.CfnOutput(this, 'HookPositiveSignalAppName', {
      value: hookPositiveSignalApp.applicationName!,
      description: 'Name of the Hook Positive Signal Application',
    });

    new cdk.CfnOutput(this, 'HookCommentSignalAppName', {
      value: hookCommentSignalApp.applicationName!,
      description: 'Name of the Hook Comment Signal Application',
    });

    new cdk.CfnOutput(this, 'HookProfileViewSignalAppName', {
      value: hookProfileViewSignalApp.applicationName!,
      description: 'Name of the Hook Profile View Signal Application',
    });

    new cdk.CfnOutput(this, 'KinesisAnalyticsSecurityGroupId', {
      value: kinesisAnalyticsSecurityGroup.securityGroupId,
      description: 'Security group ID for Kinesis Analytics Flink application',
    });

    new cdk.CfnOutput(this, 'RedisInfo', {
      value: 'Redis managed by RedisCacheStack',
      description: 'Redis cluster is managed by the RedisCacheStack',
    });

    new cdk.CfnOutput(this, 'CollaborativeFilteringLogGroupOutput', {
      value: collaborativeFilteringLogGroup.logGroupName,
      description: 'CloudWatch Log Group for Collaborative Filtering Application logs',
    });

    new cdk.CfnOutput(this, 'CoExistingLikeLogGroupOutput', {
      value: coExistingLikeLogGroup.logGroupName,
      description: 'CloudWatch Log Group for Co-Existing Like Application logs',
    });

    new cdk.CfnOutput(this, 'UserLikeHistoryLogGroupOutput', {
      value: userLikeHistoryLogGroup.logGroupName,
      description: 'CloudWatch Log Group for User Like History Application logs',
    });

    new cdk.CfnOutput(this, 'HookPositiveSignalLogGroupOutput', {
      value: hookPositiveSignalLogGroup.logGroupName,
      description: 'CloudWatch Log Group for Hook Positive Signal Application logs',
    });

    new cdk.CfnOutput(this, 'HookCommentSignalLogGroupOutput', {
      value: hookCommentSignalLogGroup.logGroupName,
      description: 'CloudWatch Log Group for Hook Comment Signal Application logs',
    });

    new cdk.CfnOutput(this, 'HookProfileViewSignalLogGroupOutput', {
      value: hookProfileViewSignalLogGroup.logGroupName,
      description: 'CloudWatch Log Group for Hook Profile View Signal Application logs',
    });

    new cdk.CfnOutput(this, 'HookCommentViewSignalLogGroupOutput', {
      value: hookCommentViewSignalLogGroup.logGroupName,
      description: 'CloudWatch Log Group for Hook Comment View Signal Application logs',
    });

    new cdk.CfnOutput(this, 'CollaborativeFilteringLogStreamOutput', {
      value: collaborativeFilteringLogStream.logStreamName!,
      description: 'CloudWatch Log Stream for Collaborative Filtering Application logs',
    });

    new cdk.CfnOutput(this, 'CoExistingLikeLogStreamOutput', {
      value: coExistingLikeLogStream.logStreamName!,
      description: 'CloudWatch Log Stream for Co-Existing Like Application logs',
    });

    new cdk.CfnOutput(this, 'UserLikeHistoryLogStreamOutput', {
      value: userLikeHistoryLogStream.logStreamName!,
      description: 'CloudWatch Log Stream for User Like History Application logs',
    });

    new cdk.CfnOutput(this, 'HookPositiveSignalLogStreamOutput', {
      value: hookPositiveSignalLogStream.logStreamName!,
      description: 'CloudWatch Log Stream for Hook Positive Signal Application logs',
    });

    new cdk.CfnOutput(this, 'HookCommentSignalLogStreamOutput', {
      value: hookCommentSignalLogStream.logStreamName!,
      description: 'CloudWatch Log Stream for Hook Comment Signal Application logs',
    });

    new cdk.CfnOutput(this, 'HookProfileViewSignalLogStreamOutput', {
      value: hookProfileViewSignalLogStream.logStreamName!,
      description: 'CloudWatch Log Stream for Hook Profile View Signal Application logs',
    });

    // EFO Consumer Outputs
    new cdk.CfnOutput(this, 'CollaborativeFilteringConsumerArn', {
      value: collaborativeFilteringConsumer.attrConsumerArn,
      description: 'ARN of the Collaborative Filtering EFO Consumer',
    });

    new cdk.CfnOutput(this, 'CoExistingLikeConsumerArn', {
      value: coExistingLikeConsumer.attrConsumerArn,
      description: 'ARN of the Co-Existing Like EFO Consumer',
    });

    new cdk.CfnOutput(this, 'UserLikeHistoryConsumerArn', {
      value: userLikeHistoryConsumer.attrConsumerArn,
      description: 'ARN of the User Like History EFO Consumer',
    });

    new cdk.CfnOutput(this, 'HookPositiveSignalConsumerArn', {
      value: hookPositiveSignalConsumer.attrConsumerArn,
      description: 'ARN of the Hook Positive Signal EFO Consumer',
    });

    new cdk.CfnOutput(this, 'HookCommentSignalConsumerArn', {
      value: hookCommentSignalConsumer.attrConsumerArn,
      description: 'ARN of the Hook Comment Signal EFO Consumer',
    });

    new cdk.CfnOutput(this, 'HookProfileViewSignalConsumerArn', {
      value: hookProfileViewSignalConsumer.attrConsumerArn,
      description: 'ARN of the Hook Profile View Signal EFO Consumer',
    });

    new cdk.CfnOutput(this, 'HookCommentViewSignalConsumerArn', {
      value: hookCommentViewSignalConsumer.attrConsumerArn,
      description: 'ARN of the Hook Comment View Signal EFO Consumer',
    });
  }
}