analytics-construct.ts

import * as cdk from 'aws-cdk-lib';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as s3Notifications from 'aws-cdk-lib/aws-s3-notifications';
import * as firehose from 'aws-cdk-lib/aws-kinesisfirehose';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as glue from 'aws-cdk-lib/aws-glue';
import * as athena from 'aws-cdk-lib/aws-athena';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as logs from 'aws-cdk-lib/aws-logs';
import * as ses from 'aws-cdk-lib/aws-ses';
import * as cloudwatch from 'aws-cdk-lib/aws-cloudwatch';
import * as sns from 'aws-cdk-lib/aws-sns';
import { Construct } from 'constructs';
import { AnalyticsConstructProps, AnalyticsResources } from './types';
import * as path from 'path';

/**
 * AnalyticsConstruct manages comprehensive SES analytics pipeline infrastructure
 *
 * This construct creates and configures:
 * - S3 buckets for raw events, processed data, and Athena results
 * - Kinesis Data Firehose for SES event streaming with Parquet conversion
 * - AWS Glue database and table definitions for SES events
 * - Athena workgroup with cost controls for analytics queries
 * - Lambda functions for partition management
 * - SES event destination configuration
 * - Lifecycle policies for cost optimization
 *
 * Requirements: 1.1, 1.2, 1.5 - Analytics infrastructure deployment
 */
export class AnalyticsConstruct extends Construct implements AnalyticsResources {
  public readonly rawDataBucket: s3.Bucket;
  public readonly athenaResultsBucket: s3.Bucket;
  public readonly glueDatabase: glue.CfnDatabase;
  public readonly workGroup: athena.CfnWorkGroup;
  public readonly deliveryStream: firehose.CfnDeliveryStream;
  public readonly sesEventDestination: ses.CfnConfigurationSetEventDestination;

  private readonly stage: string;

  constructor(scope: Construct, id: string, props: AnalyticsConstructProps) {
    super(scope, id);

    this.stage = props.stage;

    // Create SES analytics pipeline infrastructure
    this.rawDataBucket = this.createRawDataBucket(props.stage);
    this.athenaResultsBucket = this.createAthenaResultsBucket(props.stage);
    this.glueDatabase = this.createGlueDatabase();
    this.workGroup = this.createAthenaWorkGroup();
    
    // Create table and partition management
    const { tableCreationTrigger, rawEventsTableName, viewCreationTrigger } = this.createSesEventsTable();
    
    // Create Firehose delivery stream
    this.deliveryStream = this.createFirehoseDeliveryStream(rawEventsTableName, tableCreationTrigger);
    
    // Create partition management Lambda
    this.createPartitionManagementLambda(rawEventsTableName);
    
    // Create SES event destination
    this.sesEventDestination = this.createSesEventDestination(props.sesConfigurationSetName);
  }

  private createRawDataBucket(stage: string): s3.Bucket {
    return new s3.Bucket(this, 'SesRawDataBucket', {
      encryption: s3.BucketEncryption.S3_MANAGED,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      removalPolicy: cdk.RemovalPolicy.RETAIN,
      enforceSSL: true,
      lifecycleRules: [
        {
          id: 'DeleteOldRawData',
          enabled: true,
          expiration: cdk.Duration.days(365), // 1 year retention
        },
        {
          id: 'TransitionToGlacier',
          enabled: true,
          transitions: [
            {
              storageClass: s3.StorageClass.GLACIER,
              transitionAfter: cdk.Duration.days(90),
            },
          ],
        },
      ],
    });
  }

  private createAthenaResultsBucket(stage: string): s3.Bucket {
    return new s3.Bucket(this, 'AthenaResultsBucket', {
      encryption: s3.BucketEncryption.S3_MANAGED,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      removalPolicy: cdk.RemovalPolicy.RETAIN,
      enforceSSL: true,
      lifecycleRules: [
        {
          id: 'DeleteOldQueryResults',
          enabled: true,
          expiration: cdk.Duration.days(30), // 30 days retention for query results
        },
      ],
    });
  }

  private createGlueDatabase(): glue.CfnDatabase {
    return new glue.CfnDatabase(this, 'SesEventDatabase', {
      catalogId: cdk.Stack.of(this).account,
      databaseInput: {
        name: `ses_analytics_db_${this.stage}`,
        description: 'Database for SES campaign analytics events',
      },
    });
  }

  private createAthenaWorkGroup(): athena.CfnWorkGroup {
    return new athena.CfnWorkGroup(this, 'SesAnalyticsWorkGroup', {
      name: `ses-analytics-wg-${this.stage}`,
      description: 'Workgroup for SES campaign analytics queries',
      workGroupConfiguration: {
        resultConfiguration: {
          outputLocation: `s3://${this.athenaResultsBucket.bucketName}/query-results/`,
          encryptionConfiguration: {
            encryptionOption: 'SSE_S3',
          },
        },
        enforceWorkGroupConfiguration: true,
        publishCloudWatchMetricsEnabled: true,
        bytesScannedCutoffPerQuery: 10 * 1024 * 1024 * 1024, // 10 GB scan limit
        engineVersion: {
          selectedEngineVersion: 'Athena engine version 3',
        },
      },
    });
  }

  private createSesEventsTable(): { tableCreationTrigger: cdk.CustomResource; rawEventsTableName: string; viewCreationTrigger: cdk.CustomResource } {
    const rawEventsTableName = 'ses_events_raw';

    // Create Athena Named Query for table creation
    const createTableQuery = new athena.CfnNamedQuery(this, 'CreateSesEventsTable', {
      database: this.glueDatabase.ref,
      description: 'Create table for SES events',
      name: 'create_ses_events_table',
      queryString: `CREATE EXTERNAL TABLE IF NOT EXISTS ${rawEventsTableName} (
  eventType string,
  mail struct<
    timestamp: string,
    source: string,
    sourceArn: string,
    sendingAccountId: string,
    messageId: string,
    destination: array<string>,
    headersTruncated: boolean,
    headers: array<struct<name: string, value: string>>,
    commonHeaders: struct<\`from\`: array<string>, \`to\`: array<string>, messageId: string, subject: string>,
    tags: struct<
      campaign_id: array<string>,
      campaign_name: array<string>
    >
  >,
  send map<string,string>,
  delivery struct<
    timestamp: string,
    processingTimeMillis: bigint,
    recipients: array<string>,
    smtpResponse: string,
    reportingMTA: string
  >,
  open struct<
    ipAddress: string,
    timestamp: string,
    userAgent: string
  >,
  click struct<
    ipAddress: string,
    link: string,
    linkTags: map<string,array<string>>,
    timestamp: string,
    userAgent: string
  >,
  bounce struct<
    bounceType: string,
    bounceSubType: string,
    bouncedRecipients: array<struct<
      emailAddress: string,
      action: string,
      status: string,
      diagnosticCode: string
    >>,
    timestamp: string,
    feedbackId: string,
    reportingMTA: string
  >,
  complaint struct<
    complainedRecipients: array<struct<
      emailAddress: string
    >>,
    timestamp: string,
    feedbackId: string,
    userAgent: string,
    complaintFeedbackType: string,
    arrivalDate: string
  >,
  reject struct<
    reason: string
  >,
  renderingFailure struct<
    errorMessage: string,
    templateName: string
  >
)
PARTITIONED BY (
  year string,
  month string,
  day string,
  hour string
)
STORED AS parquet
LOCATION "s3://${this.rawDataBucket.bucketName}/events/"
TBLPROPERTIES (
  "parquet.compression"="SNAPPY"
)`,
    });

    // Lambda to execute the named query and create the table
    const createTableLambda = new lambda.Function(this, 'CreateTableLambda', {
      runtime: lambda.Runtime.PYTHON_3_12,
      handler: 'createTable.handler',
      code: lambda.Code.fromAsset(path.join(__dirname, '../../lambda/analytics')),
      timeout: cdk.Duration.minutes(5),
      environment: {
        NAMED_QUERY_ID: createTableQuery.attrNamedQueryId,
        DATABASE_NAME: this.glueDatabase.ref,
        OUTPUT_LOCATION: `s3://${this.athenaResultsBucket.bucketName}/table-creation/`,
        ATHENA_WORKGROUP: this.workGroup.name!,
      },
    });

    this.athenaResultsBucket.grantReadWrite(createTableLambda);
    
    createTableLambda.addToRolePolicy(
      new iam.PolicyStatement({
        actions: [
          'athena:GetNamedQuery',
          'athena:StartQueryExecution',
          'athena:GetQueryExecution',
          'athena:GetQueryResults',
        ],
        resources: ['*'],
      }),
    );

    createTableLambda.addToRolePolicy(
      new iam.PolicyStatement({
        actions: [
          'glue:GetDatabase',
          'glue:GetTable',
          'glue:CreateTable',
        ],
        resources: [
          `arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:catalog`,
          `arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:database/${this.glueDatabase.ref}`,
          `arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:table/${this.glueDatabase.ref}/*`,
        ],
      }),
    );

    // Custom resource to trigger table creation
    const tableCreationTrigger = new cdk.CustomResource(this, 'TableCreationTrigger', {
      serviceToken: createTableLambda.functionArn,
    });

    tableCreationTrigger.node.addDependency(this.glueDatabase);
    tableCreationTrigger.node.addDependency(createTableQuery);

    // Create analytics views after table creation
    const viewCreationTrigger = this.createAnalyticsViews(tableCreationTrigger);

    return { tableCreationTrigger, rawEventsTableName, viewCreationTrigger };
  }

  private createAnalyticsViews(tableCreationTrigger: cdk.CustomResource): cdk.CustomResource {
    // Create campaign analytics view
    const campaignAnalyticsViewQuery = new athena.CfnNamedQuery(this, 'CreateCampaignAnalyticsView', {
      database: this.glueDatabase.ref,
      description: 'Create campaign analytics view with campaign_id and campaign_name',
      name: 'create_campaign_analytics_view',
      queryString: `CREATE OR REPLACE VIEW campaign_analytics AS
SELECT
  COALESCE(element_at(mail.tags.campaign_id, 1), 'no-campaign-id') as campaign_id,
  COALESCE(element_at(mail.tags.campaign_name, 1), 'no-campaign-name') as campaign_name,
  DATE(from_iso8601_timestamp(mail.timestamp)) as date,
  COUNT(DISTINCT CASE WHEN eventtype = 'Send' THEN mail.messageId END) as emails_sent,
  COUNT(DISTINCT CASE WHEN eventtype = 'Delivery' THEN mail.messageId END) as emails_delivered,
  COUNT(DISTINCT CASE WHEN eventtype = 'Open' THEN mail.messageId END) as emails_opened,
  COUNT(DISTINCT CASE WHEN eventtype = 'Click' THEN mail.messageId END) as emails_clicked,
  COUNT(DISTINCT CASE WHEN eventtype = 'Bounce' AND bounce.bounceType = 'Permanent' THEN mail.messageId END) as hard_bounces,
  COUNT(DISTINCT CASE WHEN eventtype = 'Bounce' AND bounce.bounceType = 'Transient' THEN mail.messageId END) as soft_bounces,
  COUNT(DISTINCT CASE WHEN eventtype = 'Bounce' THEN mail.messageId END) as total_bounces,
  COUNT(DISTINCT CASE WHEN eventtype = 'Complaint' THEN mail.messageId END) as complaints,
  COUNT(DISTINCT CASE WHEN eventtype = 'Reject' THEN mail.messageId END) as rejects,
  COUNT(DISTINCT CASE WHEN eventtype = 'Rendering Failure' THEN mail.messageId END) as rendering_failures,
  COUNT(DISTINCT element_at(mail.destination, 1)) as unique_recipients,
  CAST(COUNT(DISTINCT CASE WHEN eventtype = 'Delivery' THEN mail.messageId END) AS DOUBLE) / 
    NULLIF(COUNT(DISTINCT CASE WHEN eventtype = 'Send' THEN mail.messageId END), 0) * 100 as delivery_rate,
  CAST(COUNT(DISTINCT CASE WHEN eventtype = 'Open' THEN mail.messageId END) AS DOUBLE) / 
    NULLIF(COUNT(DISTINCT CASE WHEN eventtype = 'Delivery' THEN mail.messageId END), 0) * 100 as open_rate,
  CAST(COUNT(DISTINCT CASE WHEN eventtype = 'Click' THEN mail.messageId END) AS DOUBLE) / 
    NULLIF(COUNT(DISTINCT CASE WHEN eventtype = 'Delivery' THEN mail.messageId END), 0) * 100 as click_rate,
  CAST(COUNT(DISTINCT CASE WHEN eventtype = 'Bounce' THEN mail.messageId END) AS DOUBLE) / 
    NULLIF(COUNT(DISTINCT CASE WHEN eventtype = 'Send' THEN mail.messageId END), 0) * 100 as bounce_rate,
  CAST(COUNT(DISTINCT CASE WHEN eventtype = 'Complaint' THEN mail.messageId END) AS DOUBLE) / 
    NULLIF(COUNT(DISTINCT CASE WHEN eventtype = 'Delivery' THEN mail.messageId END), 0) * 100 as complaint_rate,
  CAST(MAX(from_iso8601_timestamp(mail.timestamp)) AS TIMESTAMP) as last_activity
FROM ses_events_raw
WHERE mail.tags IS NOT NULL
  AND mail.tags.campaign_id IS NOT NULL
  AND element_at(mail.tags.campaign_id, 1) IS NOT NULL
GROUP BY 
  COALESCE(element_at(mail.tags.campaign_id, 1), 'no-campaign-id'),
  COALESCE(element_at(mail.tags.campaign_name, 1), 'no-campaign-name'),
  DATE(from_iso8601_timestamp(mail.timestamp))
ORDER BY date DESC, emails_sent DESC`,
    });

    // Create campaign summary view
    const campaignSummaryViewQuery = new athena.CfnNamedQuery(this, 'CreateCampaignSummaryView', {
      database: this.glueDatabase.ref,
      description: 'Create campaign summary view for all-time metrics',
      name: 'create_campaign_summary_view',
      queryString: `CREATE OR REPLACE VIEW campaign_summary AS
SELECT
  campaign_id,
  campaign_name,
  MIN(date) as first_send_date,
  MAX(date) as last_send_date,
  COUNT(DISTINCT date) as days_active,
  SUM(emails_sent) as total_sent,
  SUM(emails_delivered) as total_delivered,
  SUM(emails_opened) as total_opened,
  SUM(emails_clicked) as total_clicked,
  SUM(hard_bounces) as total_hard_bounces,
  SUM(soft_bounces) as total_soft_bounces,
  SUM(total_bounces) as total_bounces,
  SUM(complaints) as total_complaints,
  SUM(rejects) as total_rejects,
  SUM(rendering_failures) as total_rendering_failures,
  SUM(unique_recipients) as total_unique_recipients,
  CAST(SUM(emails_delivered) AS DOUBLE) / NULLIF(SUM(emails_sent), 0) * 100 as overall_delivery_rate,
  CAST(SUM(emails_opened) AS DOUBLE) / NULLIF(SUM(emails_delivered), 0) * 100 as overall_open_rate,
  CAST(SUM(emails_clicked) AS DOUBLE) / NULLIF(SUM(emails_delivered), 0) * 100 as overall_click_rate,
  CAST(SUM(total_bounces) AS DOUBLE) / NULLIF(SUM(emails_sent), 0) * 100 as overall_bounce_rate,
  CAST(SUM(complaints) AS DOUBLE) / NULLIF(SUM(emails_delivered), 0) * 100 as overall_complaint_rate,
  MAX(last_activity) as last_activity
FROM campaign_analytics
GROUP BY campaign_id, campaign_name
ORDER BY total_sent DESC`,
    });

    // daily_summary: send/delivery/bounce/open/click counts per day
    // across all campaigns. Backs the "dashboard / overview" prompt
    // ("daily volume last 14 days") without scanning ses_events_raw.
    const dailySummaryViewQuery = new athena.CfnNamedQuery(this, 'CreateDailySummaryView', {
      database: this.glueDatabase.ref,
      description: 'Create daily_summary view: per-day SES event counts across all campaigns',
      name: 'create_daily_summary_view',
      queryString: `CREATE OR REPLACE VIEW daily_summary AS
SELECT
  date,
  SUM(emails_sent) as emails_sent,
  SUM(emails_delivered) as emails_delivered,
  SUM(emails_opened) as emails_opened,
  SUM(emails_clicked) as emails_clicked,
  SUM(hard_bounces) as hard_bounces,
  SUM(soft_bounces) as soft_bounces,
  SUM(total_bounces) as total_bounces,
  SUM(complaints) as complaints,
  SUM(rejects) as rejects,
  SUM(unique_recipients) as unique_recipients,
  CAST(SUM(emails_delivered) AS DOUBLE) / NULLIF(SUM(emails_sent), 0) * 100 as delivery_rate,
  CAST(SUM(emails_opened) AS DOUBLE) / NULLIF(SUM(emails_delivered), 0) * 100 as open_rate,
  CAST(SUM(emails_clicked) AS DOUBLE) / NULLIF(SUM(emails_delivered), 0) * 100 as click_rate,
  CAST(SUM(total_bounces) AS DOUBLE) / NULLIF(SUM(emails_sent), 0) * 100 as bounce_rate
FROM campaign_analytics
GROUP BY date
ORDER BY date DESC`,
    });

    // bounce_analysis: bounce composition by type/subtype with top
    // bouncing recipient domains. Backs the "bounce type composition"
    // half of the overview prompt and the agentic segment-EDA flow's
    // breakdown questions.
    const bounceAnalysisViewQuery = new athena.CfnNamedQuery(this, 'CreateBounceAnalysisView', {
      database: this.glueDatabase.ref,
      description: 'Create bounce_analysis view: bounce composition + top bouncing domains',
      name: 'create_bounce_analysis_view',
      queryString: `CREATE OR REPLACE VIEW bounce_analysis AS
SELECT
  DATE(from_iso8601_timestamp(mail.timestamp)) as date,
  COALESCE(bounce.bounceType, 'Unknown') as bounce_type,
  COALESCE(bounce.bounceSubType, 'Unknown') as bounce_subtype,
  CASE
    WHEN element_at(mail.destination, 1) IS NULL THEN 'unknown'
    WHEN strpos(element_at(mail.destination, 1), '@') = 0 THEN 'unknown'
    ELSE lower(substr(element_at(mail.destination, 1), strpos(element_at(mail.destination, 1), '@') + 1))
  END as recipient_domain,
  COUNT(DISTINCT mail.messageId) as bounce_count
FROM ses_events_raw
WHERE eventtype = 'Bounce'
GROUP BY
  DATE(from_iso8601_timestamp(mail.timestamp)),
  COALESCE(bounce.bounceType, 'Unknown'),
  COALESCE(bounce.bounceSubType, 'Unknown'),
  CASE
    WHEN element_at(mail.destination, 1) IS NULL THEN 'unknown'
    WHEN strpos(element_at(mail.destination, 1), '@') = 0 THEN 'unknown'
    ELSE lower(substr(element_at(mail.destination, 1), strpos(element_at(mail.destination, 1), '@') + 1))
  END
ORDER BY date DESC, bounce_count DESC`,
    });

    // Lambda to execute view creation queries
    const createViewsLambda = new lambda.Function(this, 'CreateViewsLambda', {
      runtime: lambda.Runtime.PYTHON_3_12,
      handler: 'createViews.handler',
      code: lambda.Code.fromAsset(path.join(__dirname, '../../lambda/analytics')),
      timeout: cdk.Duration.minutes(5),
      environment: {
        ANALYTICS_VIEW_QUERY_ID: campaignAnalyticsViewQuery.attrNamedQueryId,
        SUMMARY_VIEW_QUERY_ID: campaignSummaryViewQuery.attrNamedQueryId,
        DAILY_SUMMARY_VIEW_QUERY_ID: dailySummaryViewQuery.attrNamedQueryId,
        BOUNCE_ANALYSIS_VIEW_QUERY_ID: bounceAnalysisViewQuery.attrNamedQueryId,
        DATABASE_NAME: this.glueDatabase.ref,
        OUTPUT_LOCATION: `s3://${this.athenaResultsBucket.bucketName}/view-creation/`,
        ATHENA_WORKGROUP: this.workGroup.name!,
      },
    });

    this.athenaResultsBucket.grantReadWrite(createViewsLambda);
    
    createViewsLambda.addToRolePolicy(
      new iam.PolicyStatement({
        actions: [
          'athena:GetNamedQuery',
          'athena:StartQueryExecution',
          'athena:GetQueryExecution',
          'athena:GetQueryResults',
        ],
        resources: ['*'],
      }),
    );

    createViewsLambda.addToRolePolicy(
      new iam.PolicyStatement({
        actions: [
          'glue:GetDatabase',
          'glue:GetTable',
          'glue:CreateTable',
          'glue:UpdateTable',
        ],
        resources: [
          `arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:catalog`,
          `arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:database/${this.glueDatabase.ref}`,
          `arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:table/${this.glueDatabase.ref}/*`,
        ],
      }),
    );

    // Custom resource to trigger view creation
    const viewCreationTrigger = new cdk.CustomResource(this, 'ViewCreationTrigger', {
      serviceToken: createViewsLambda.functionArn,
      properties: {
        // Bumped to 2.1.0 to recreate views with the two new ones
        // (daily_summary, bounce_analysis) added for step 6.
        ViewVersion: '2.1.0',
      },
    });

    viewCreationTrigger.node.addDependency(tableCreationTrigger);
    viewCreationTrigger.node.addDependency(campaignAnalyticsViewQuery);
    viewCreationTrigger.node.addDependency(campaignSummaryViewQuery);
    viewCreationTrigger.node.addDependency(dailySummaryViewQuery);
    viewCreationTrigger.node.addDependency(bounceAnalysisViewQuery);

    return viewCreationTrigger;
  }

  private createFirehoseDeliveryStream(rawEventsTableName: string, tableCreationTrigger: cdk.CustomResource): firehose.CfnDeliveryStream {
    // Role for Firehose service
    const firehoseRole = new iam.Role(this, 'FirehoseRole', {
      assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
      description: 'Role for Kinesis Firehose to write SES events to S3 and convert to Parquet',
    });

    this.rawDataBucket.grantWrite(firehoseRole);

    // CloudWatch Logs for Firehose with enhanced error logging
    const firehoseLogGroup = new logs.LogGroup(this, 'FirehoseLogGroup', {
      logGroupName: `/aws/kinesisfirehose/ses-analytics-${this.stage}`,
      retention: logs.RetentionDays.TWO_WEEKS, // Extended retention for error analysis
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const firehoseLogStream = new logs.LogStream(this, 'FirehoseLogStream', {
      logGroup: firehoseLogGroup,
      logStreamName: 'S3Delivery',
    });

    // Additional log stream for error tracking
    const firehoseErrorLogStream = new logs.LogStream(this, 'FirehoseErrorLogStream', {
      logGroup: firehoseLogGroup,
      logStreamName: 'ErrorDelivery',
    });

    firehoseLogGroup.grantWrite(firehoseRole);

    // Grant Glue permissions to Firehose role (CRITICAL for Parquet conversion)
    const gluePolicy = new iam.Policy(this, 'FirehoseGluePolicy', {
      statements: [
        new iam.PolicyStatement({
          actions: [
            'glue:GetTable',
            'glue:GetTableVersion',
            'glue:GetTableVersions',
          ],
          resources: [
            `arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:catalog`,
            `arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:database/${this.glueDatabase.ref}`,
            `arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:table/${this.glueDatabase.ref}/*`,
          ],
        }),
      ],
    });
    
    gluePolicy.attachToRole(firehoseRole);

    // Create delivery stream with enhanced error handling
    const deliveryStream = new firehose.CfnDeliveryStream(this, 'SesEventsDeliveryStream', {
      deliveryStreamName: `ses-analytics-stream-${this.stage}`,
      deliveryStreamType: 'DirectPut',
      extendedS3DestinationConfiguration: {
        bucketArn: this.rawDataBucket.bucketArn,
        roleArn: firehoseRole.roleArn,
        // Use Hive-style partition prefixes for automatic partition discovery
        prefix: 'events/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/',
        errorOutputPrefix: 'errors/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/!{firehose:error-output-type}/',
        bufferingHints: {
          sizeInMBs: 64, // Minimum for data format conversion
          intervalInSeconds: 0, // 0 for demo purposes
        },
        compressionFormat: 'UNCOMPRESSED', // Required for Parquet conversion
        // Processing configuration with error handling
        processingConfiguration: {
          enabled: false, // Disable processing to avoid transformation errors
        },
        // Data format conversion enabled with proper dependencies
        dataFormatConversionConfiguration: {
          enabled: true,
          schemaConfiguration: {
            databaseName: this.glueDatabase.ref,
            tableName: rawEventsTableName,
            region: cdk.Stack.of(this).region,
            roleArn: firehoseRole.roleArn,
          },
          inputFormatConfiguration: {
            deserializer: {
              openXJsonSerDe: {},
            },
          },
          outputFormatConfiguration: {
            serializer: {
              parquetSerDe: {
                compression: 'SNAPPY',
              },
            },
          },
        },
        // Enhanced CloudWatch logging for error tracking
        cloudWatchLoggingOptions: {
          enabled: true,
          logGroupName: firehoseLogGroup.logGroupName,
          logStreamName: firehoseLogStream.logStreamName,
        },
        // S3 backup configuration for failed records
        s3BackupMode: 'Enabled',
        s3BackupConfiguration: {
          bucketArn: this.rawDataBucket.bucketArn,
          roleArn: firehoseRole.roleArn,
          prefix: 'backup/failed-records/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/',
          errorOutputPrefix: 'backup/failed-records/errors/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/!{firehose:error-output-type}/',
          bufferingHints: {
            sizeInMBs: 5,
            intervalInSeconds: 300,
          },
          compressionFormat: 'GZIP',
          cloudWatchLoggingOptions: {
            enabled: true,
            logGroupName: firehoseLogGroup.logGroupName,
            logStreamName: firehoseErrorLogStream.logStreamName,
          },
        },
      },
    });

    // Ensure delivery stream is created after dependencies
    deliveryStream.node.addDependency(gluePolicy);
    deliveryStream.node.addDependency(tableCreationTrigger);

    // Create CloudWatch alarms for monitoring Firehose delivery failures
    this.createFirehoseMonitoring(deliveryStream);

    // Note: Data format conversion disabled to prevent Parquet corruption
    // JSON format with GZIP compression works reliably with Athena

    return deliveryStream;
  }

  private createPartitionManagementLambda(rawEventsTableName: string): void {
    // Lambda to automatically add partitions when new data arrives in S3
    const athenaPartitionLambda = new lambda.Function(this, 'AthenaPartitionLambda', {
      runtime: lambda.Runtime.PYTHON_3_12,
      handler: 'partitionManager.handler',
      code: lambda.Code.fromAsset(path.join(__dirname, '../../lambda/analytics')),
      timeout: cdk.Duration.minutes(5),
      memorySize: 256,
      environment: {
        DATABASE_NAME: this.glueDatabase.ref,
        TABLE_NAME: rawEventsTableName,
        OUTPUT_LOCATION: `s3://${this.athenaResultsBucket.bucketName}/partition-mgmt/`,
        S3_BUCKET_NAME: this.rawDataBucket.bucketName,
        MAX_PARTITION_RETRIES: '3',
        RETRY_DELAY_SECONDS: '5',
        PARTITION_QUERY_TIMEOUT: '300',
        ATHENA_WORKGROUP: this.workGroup.name!,
      },
    });

    // Grant permissions to partition Lambda
    this.rawDataBucket.grantRead(athenaPartitionLambda);
    this.athenaResultsBucket.grantReadWrite(athenaPartitionLambda);

    athenaPartitionLambda.addToRolePolicy(
      new iam.PolicyStatement({
        actions: [
          'athena:StartQueryExecution',
          'athena:GetQueryExecution',
          'athena:GetQueryResults',
          'athena:StopQueryExecution'
        ],
        resources: [
          `arn:aws:athena:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:workgroup/${this.workGroup.name!}`,
        ],
      }),
    );

    athenaPartitionLambda.addToRolePolicy(
      new iam.PolicyStatement({
        actions: [
          'glue:GetDatabase',
          'glue:GetTable',
          'glue:CreatePartition',
          'glue:BatchCreatePartition',
        ],
        resources: [
          `arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:catalog`,
          `arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:database/${this.glueDatabase.ref}`,
          `arn:aws:glue:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:table/${this.glueDatabase.ref}/*`,
        ],
      }),
    );

    // Add CloudWatch permissions for metrics
    athenaPartitionLambda.addToRolePolicy(
      new iam.PolicyStatement({
        actions: [
          'cloudwatch:PutMetricData',
        ],
        resources: ['*'],
      }),
    );

    // Grant S3 permission to invoke Lambda
    athenaPartitionLambda.addPermission('S3InvokeLambda', {
      principal: new iam.ServicePrincipal('s3.amazonaws.com'),
      action: 'lambda:InvokeFunction',
      sourceAccount: cdk.Stack.of(this).account,
      sourceArn: this.rawDataBucket.bucketArn,
    });

    // Add S3 notification to trigger Lambda when new objects are created
    this.rawDataBucket.addEventNotification(
      s3.EventType.OBJECT_CREATED,
      new s3Notifications.LambdaDestination(athenaPartitionLambda),
      { prefix: 'events/' }  // Trigger on all files in events/ prefix
    );

    // Create CloudWatch alarms for partition management monitoring
    this.createPartitionManagementMonitoring(athenaPartitionLambda);
  }

  private createSesEventDestination(sesConfigurationSetName: string): ses.CfnConfigurationSetEventDestination {
    // Auto-create the SES Configuration Set so users don't have to create it manually.
    // CDK/CloudFormation handles idempotency: if the config set already exists and is
    // managed by this stack, it will simply be updated (no conflict).
    const configurationSet = new ses.CfnConfigurationSet(this, 'SesConfigurationSet', {
      name: sesConfigurationSetName,
    });

    // Role for SES to put records into Firehose
    const sesFirehoseRole = new iam.Role(this, 'SesFirehoseRole', {
      assumedBy: new iam.ServicePrincipal('ses.amazonaws.com'),
      description: 'Role for SES to put records into Kinesis Firehose',
    });

    // Grant SES role permission to put records to Firehose
    sesFirehoseRole.addToPolicy(
      new iam.PolicyStatement({
        actions: ['firehose:PutRecord', 'firehose:PutRecordBatch', 'firehose:DescribeDeliveryStream'],
        resources: [this.deliveryStream.attrArn],
      }),
    );

    // Create SES event destination
    const sesEventDestination = new ses.CfnConfigurationSetEventDestination(this, 'SesEventDestination', {
      configurationSetName: sesConfigurationSetName,
      eventDestination: {
        name: `firehose-destination-${this.stage}`,
        enabled: true,
        matchingEventTypes: ['send', 'reject', 'bounce', 'complaint', 'delivery', 'open', 'click', 'renderingFailure'],
        kinesisFirehoseDestination: {
          deliveryStreamArn: this.deliveryStream.attrArn,
          iamRoleArn: sesFirehoseRole.roleArn,
        },
      },
    });

    // Ensure SES event destination is created after dependencies
    sesEventDestination.node.addDependency(this.deliveryStream);
    sesEventDestination.node.addDependency(sesFirehoseRole);
    sesEventDestination.node.addDependency(configurationSet);

    return sesEventDestination;
  }



  private createFirehoseMonitoring(deliveryStream: firehose.CfnDeliveryStream): void {
    // Create SNS topic for error notifications (optional - can be configured later)
    const errorTopic = new sns.Topic(this, 'FirehoseErrorTopic', {
      displayName: 'SES Analytics Firehose Errors',
      topicName: `ses-analytics-firehose-errors-${this.stage}`,
    });

    // CloudWatch alarm for delivery failures
    const deliveryFailureAlarm = new cloudwatch.Alarm(this, 'FirehoseDeliveryFailureAlarm', {
      alarmName: `SES-Analytics-Firehose-DeliveryFailures-${this.stage}`,
      alarmDescription: 'Alarm when Kinesis Firehose delivery failures exceed threshold',
      metric: new cloudwatch.Metric({
        namespace: 'AWS/Kinesis/Firehose',
        metricName: 'DeliveryToS3.Records',
        dimensionsMap: {
          DeliveryStreamName: deliveryStream.deliveryStreamName!,
        },
        statistic: 'Sum',
        period: cdk.Duration.minutes(5),
      }),
      threshold: 0,
      comparisonOperator: cloudwatch.ComparisonOperator.LESS_THAN_THRESHOLD,
      evaluationPeriods: 2,
      treatMissingData: cloudwatch.TreatMissingData.BREACHING,
    });

    // CloudWatch alarm for data format conversion errors
    const formatConversionErrorAlarm = new cloudwatch.Alarm(this, 'FirehoseFormatConversionErrorAlarm', {
      alarmName: `SES-Analytics-Firehose-FormatErrors-${this.stage}`,
      alarmDescription: 'Alarm when Kinesis Firehose data format conversion errors occur',
      metric: new cloudwatch.Metric({
        namespace: 'AWS/Kinesis/Firehose',
        metricName: 'DeliveryToS3.DataFreshness',
        dimensionsMap: {
          DeliveryStreamName: deliveryStream.deliveryStreamName!,
        },
        statistic: 'Maximum',
        period: cdk.Duration.minutes(15),
      }),
      threshold: 900, // 15 minutes - indicates delivery issues
      comparisonOperator: cloudwatch.ComparisonOperator.GREATER_THAN_THRESHOLD,
      evaluationPeriods: 2,
      treatMissingData: cloudwatch.TreatMissingData.NOT_BREACHING,
    });

    // CloudWatch alarm for high error rate
    const errorRateAlarm = new cloudwatch.Alarm(this, 'FirehoseErrorRateAlarm', {
      alarmName: `SES-Analytics-Firehose-ErrorRate-${this.stage}`,
      alarmDescription: 'Alarm when Kinesis Firehose error rate is high',
      metric: new cloudwatch.Metric({
        namespace: 'AWS/Kinesis/Firehose',
        metricName: 'DeliveryToS3.Success',
        dimensionsMap: {
          DeliveryStreamName: deliveryStream.deliveryStreamName!,
        },
        statistic: 'Average',
        period: cdk.Duration.minutes(5),
      }),
      threshold: 0.95, // 95% success rate threshold
      comparisonOperator: cloudwatch.ComparisonOperator.LESS_THAN_THRESHOLD,
      evaluationPeriods: 3,
      treatMissingData: cloudwatch.TreatMissingData.NOT_BREACHING,
    });

    // Create CloudWatch dashboard for monitoring
    const dashboard = new cloudwatch.Dashboard(this, 'FirehoseDashboard', {
      dashboardName: `SES-Analytics-Firehose-${this.stage}`,
      widgets: [
        [
          new cloudwatch.GraphWidget({
            title: 'Firehose Delivery Metrics',
            left: [
              new cloudwatch.Metric({
                namespace: 'AWS/Kinesis/Firehose',
                metricName: 'DeliveryToS3.Records',
                dimensionsMap: {
                  DeliveryStreamName: deliveryStream.deliveryStreamName!,
                },
                statistic: 'Sum',
              }),
            ],
            right: [
              new cloudwatch.Metric({
                namespace: 'AWS/Kinesis/Firehose',
                metricName: 'DeliveryToS3.Success',
                dimensionsMap: {
                  DeliveryStreamName: deliveryStream.deliveryStreamName!,
                },
                statistic: 'Average',
              }),
            ],
            width: 12,
            height: 6,
          }),
        ],
        [
          new cloudwatch.GraphWidget({
            title: 'Firehose Error Metrics',
            left: [
              new cloudwatch.Metric({
                namespace: 'AWS/Kinesis/Firehose',
                metricName: 'DeliveryToS3.DataFreshness',
                dimensionsMap: {
                  DeliveryStreamName: deliveryStream.deliveryStreamName!,
                },
                statistic: 'Maximum',
              }),
            ],
            width: 12,
            height: 6,
          }),
        ],
      ],
    });

  }

  private createPartitionManagementMonitoring(partitionLambda: lambda.Function): void {
    // CloudWatch alarm for partition management failures
    const partitionFailureAlarm = new cloudwatch.Alarm(this, 'PartitionManagementFailureAlarm', {
      alarmName: `SES-Analytics-Partition-Failures-${this.stage}`,
      alarmDescription: 'Alarm when partition management failures exceed threshold',
      metric: new cloudwatch.Metric({
        namespace: 'SES/Analytics/PartitionManager',
        metricName: 'PartitionsFailed',
        statistic: 'Sum',
        period: cdk.Duration.minutes(5),
      }),
      threshold: 5, // Alert if more than 5 failures in 5 minutes
      comparisonOperator: cloudwatch.ComparisonOperator.GREATER_THAN_THRESHOLD,
      evaluationPeriods: 1,
      treatMissingData: cloudwatch.TreatMissingData.NOT_BREACHING,
    });

    // CloudWatch alarm for sustained partition failures
    const sustainedFailureAlarm = new cloudwatch.Alarm(this, 'PartitionSustainedFailureAlarm', {
      alarmName: `SES-Analytics-Partition-SustainedFailures-${this.stage}`,
      alarmDescription: 'Alarm when partition success rate is consistently low',
      metric: new cloudwatch.Metric({
        namespace: 'SES/Analytics/PartitionManager',
        metricName: 'PartitionSuccessRate',
        statistic: 'Average',
        period: cdk.Duration.minutes(15),
      }),
      threshold: 80, // Alert if success rate below 80%
      comparisonOperator: cloudwatch.ComparisonOperator.LESS_THAN_THRESHOLD,
      evaluationPeriods: 2, // 2 consecutive periods of low success rate
      treatMissingData: cloudwatch.TreatMissingData.NOT_BREACHING,
    });

    // CloudWatch alarm for Lambda function errors
    const lambdaErrorAlarm = new cloudwatch.Alarm(this, 'PartitionLambdaErrorAlarm', {
      alarmName: `SES-Analytics-Partition-Lambda-Errors-${this.stage}`,
      alarmDescription: 'Alarm when partition Lambda function has errors',
      metric: partitionLambda.metricErrors({
        period: cdk.Duration.minutes(5),
      }),
      threshold: 1,
      comparisonOperator: cloudwatch.ComparisonOperator.GREATER_THAN_OR_EQUAL_TO_THRESHOLD,
      evaluationPeriods: 1,
      treatMissingData: cloudwatch.TreatMissingData.NOT_BREACHING,
    });

    // CloudWatch alarm for critical failures
    const criticalFailureAlarm = new cloudwatch.Alarm(this, 'PartitionCriticalFailureAlarm', {
      alarmName: `SES-Analytics-Partition-CriticalFailures-${this.stage}`,
      alarmDescription: 'Alarm when partition manager has critical failures',
      metric: new cloudwatch.Metric({
        namespace: 'SES/Analytics/PartitionManager',
        metricName: 'CriticalFailures',
        statistic: 'Sum',
        period: cdk.Duration.minutes(5),
      }),
      threshold: 0,
      comparisonOperator: cloudwatch.ComparisonOperator.GREATER_THAN_THRESHOLD,
      evaluationPeriods: 1,
      treatMissingData: cloudwatch.TreatMissingData.NOT_BREACHING,
    });

  }
}