DynamoDB Table → DynamoDB Stream → Lambda Function → VPC NAT Gateway → Supabase (scenarios table)
                                      ↓                    ↓
                              Error Handling & Retry   Static IP
                                      ↓
                                Dead Letter Queue

VPC Architecture:
├── Public Subnet (NAT Gateway + Elastic IP)
└── Private Subnet (Lambda Functions)

aws/ai/
├── lambdas/
│   ├── api/
│   │   ├── handler.py          # Formerly consultation_api.py
│   │   ├── requirements.txt
│   │   └── Dockerfile
│   ├── agent_processor/
│   │   ├── handler.py          # Formerly main.py
│   │   ├── prompts.py
│   │   ├── requirements.txt
│   │   ├── Dockerfile
│   │   └── tools/
│   │       ├── chart_handler.py
│   │       └── code_capture.py
│   ├── supabase_stream_processor/
│   │   ├── handler.py          # New function
│   │   ├── requirements.txt
│   │   └── Dockerfile
│   └── common/
│       ├── aws_utils.py        # Shared utilities
│       └── database.py         # Simplified Supabase connector
└── stack.py                    # Updated for modular structure

# Port 6543 for serverless workloads
cnxn_string = "postgresql://postgres.byoneafwuigeyehkoacw:PASSWORD@aws-0-us-east-1.pooler.supabase.com:6543/postgres?sslmode=require"

psycopg2-binary==2.9.9  # Direct PostgreSQL driver
boto3                   # AWS SDK (already available in Lambda runtime)

# app.py
import aws_cdk as cdk
from ai.stack import AgentStack

app = cdk.App()

# Get environment from context
env_name = app.node.try_get_context("env") or "dev"

AgentStack(
    app, 
    f"AgentStack-{env_name}",
    env_name=env_name,
    env=cdk.Environment(
        account=os.getenv('CDK_DEFAULT_ACCOUNT'),
        region=os.getenv('CDK_DEFAULT_REGION')
    )
)

app.synth()

# In stack.py
class AgentStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, env_name: str, **kwargs):
        super().__init__(scope, construct_id, **kwargs)
        
        self.env_name = env_name
        
        # Create VPC with NAT Gateway for static IP
        self.vpc = ec2.Vpc(
            self, f"{env_name}-AgentVPC",
            max_azs=2,
            subnet_configuration=[
                ec2.SubnetConfiguration(
                    name="Public",
                    subnet_type=ec2.SubnetType.PUBLIC,
                    cidr_mask=24
                ),
                ec2.SubnetConfiguration(
                    name="Private",
                    subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS,
                    cidr_mask=24
                )
            ],
            nat_gateways=1  # Single NAT Gateway with Elastic IP
        )
        
        # Output the NAT Gateway's Elastic IP for Supabase whitelisting
        CfnOutput(
            self, f"{env_name}-NATGatewayIP",
            value=self.vpc.public_subnets[0].node.children[1].ref,  # NAT Gateway EIP
            description=f"Elastic IP for {env_name} environment - Add to Supabase whitelist"
        )

# lambdas/common/database.py
import os
import json
import boto3
import psycopg2
import logging

logger = logging.getLogger(__name__)

# Global connection variable for reuse
_connection = None

def get_supabase_connection():
    """
    Establishes or reuses a connection to the Supabase database.
    Credentials are read from Secrets Manager and cached.
    """
    global _connection
    
    # Reuse existing connection if it's healthy
    if _connection and _connection.closed == 0:
        return _connection

    # Load credentials from Secrets Manager
    secrets_client = boto3.client('secretsmanager')
    secret_name = os.environ['SUPABASE_SECRET_NAME']
    response = secrets_client.get_secret_value(SecretId=secret_name)
    creds = json.loads(response['SecretString'])
    
    conn_params = {
        'host': creds['host'],
        'port': creds.get('port', 6543),  # Default to transaction pooler
        'database': creds['database'],
        'user': f"postgres.{creds['project_id']}",
        'password': creds['password'],
        'sslmode': 'require',
        'connect_timeout': 10
    }
    
    try:
        _connection = psycopg2.connect(**conn_params)
        logger.info("Successfully connected to Supabase")
        return _connection
    except Exception as e:
        logger.error(f"Failed to connect to Supabase: {e}")
        raise

def test_connection():
    """Test database connectivity"""
    conn = get_supabase_connection()
    with conn.cursor() as cursor:
        cursor.execute("SELECT 1")
        result = cursor.fetchone()
        return result[0] == 1

# lambdas/supabase_stream_processor/handler.py
import os
import json
import logging
from boto3.dynamodb.types import TypeDeserializer
from psycopg2.extras import execute_batch

# Import from common layer
import sys
sys.path.append('/opt/python')  # Lambda layer path
from common.database import get_supabase_connection

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Initialize the deserializer once
deserializer = TypeDeserializer()

def unmarshall_dynamodb_item(item):
    """Converts a DynamoDB item into a regular Python dictionary."""
    return {k: deserializer.deserialize(v) for k, v in item.items()}

def transform_record_to_scenario(record):
    """Transforms a single DynamoDB stream record into a Supabase scenario dict."""
    event_name = record['eventName']
    sequence_number = record['dynamodb']['SequenceNumber']
    
    if event_name == 'REMOVE':
        # For DELETE, mark as deleted
        return {
            'dynamo_sequence_number': sequence_number,
            'dynamo_event_name': event_name,
            'scenario_data': json.dumps({'deleted': True, 'keys': record['dynamodb']['Keys']}),
            'user_id': unmarshall_dynamodb_item(record['dynamodb']['Keys']).get('userId'),
            'consultation_id': unmarshall_dynamodb_item(record['dynamodb']['Keys']).get('consultationId'),
        }

    # For INSERT and MODIFY
    new_image = record['dynamodb'].get('NewImage', {})
    data = unmarshall_dynamodb_item(new_image)
    
    return {
        'user_id': data.get('userId'),
        'consultation_id': data.get('consultationId'),
        'scenario_data': json.dumps(data),
        'dynamo_sequence_number': sequence_number,
        'dynamo_event_name': event_name,
    }

def lambda_handler(event, context):
    records = event.get('Records', [])
    if not records:
        return {'statusCode': 200, 'body': 'No records to process.'}

    try:
        scenarios = [transform_record_to_scenario(r) for r in records]
        
        upsert_query = """
            INSERT INTO scenarios (user_id, consultation_id, scenario_data, dynamo_sequence_number, dynamo_event_name, updated_at)
            VALUES (%(user_id)s, %(consultation_id)s, %(scenario_data)s::jsonb, %(dynamo_sequence_number)s, %(dynamo_event_name)s, NOW())
            ON CONFLICT (dynamo_sequence_number) DO UPDATE SET
                scenario_data = EXCLUDED.scenario_data,
                dynamo_event_name = EXCLUDED.dynamo_event_name,
                updated_at = NOW();
        """
        
        conn = get_supabase_connection()
        with conn.cursor() as cursor:
            execute_batch(cursor, upsert_query, scenarios, page_size=100)
        conn.commit()
        
        logger.info(f"Successfully synced {len(scenarios)} scenarios to Supabase.")
        return {'statusCode': 200, 'body': json.dumps({'processed': len(scenarios)})}

    except Exception as e:
        logger.error(f"Error processing DynamoDB stream: {e}")
        # The DLQ configured on the event source will handle the retry/failure
        raise e

CREATE TABLE scenarios (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id TEXT NOT NULL,
    consultation_id TEXT,
    scenario_data JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),
    dynamo_sequence_number TEXT UNIQUE NOT NULL, -- For idempotency
    dynamo_event_name TEXT NOT NULL, -- INSERT, MODIFY, REMOVE
    
    -- Indexes for performance
    INDEX idx_scenarios_user_id (user_id),
    INDEX idx_scenarios_consultation_id (consultation_id),
    INDEX idx_scenarios_created_at (created_at DESC)
);

# Updated stack.py
from aws_cdk import (
    Stack, CfnOutput, SecretValue, Duration, RemovalPolicy,
    aws_lambda as _lambda,
    aws_lambda_event_sources as lambda_event_sources,
    aws_ec2 as ec2,
    aws_secretsmanager as secretsmanager,
    aws_dynamodb as dynamodb,
    aws_sqs as sqs,
)
from constructs import Construct
import os
from dotenv import load_dotenv

class AgentStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, env_name: str, **kwargs):
        super().__init__(scope, construct_id, **kwargs)
        
        load_dotenv(os.path.join(os.path.dirname(__file__), '.env'))
        self.env_name = env_name
        
        # Create VPC with NAT Gateway for static IP
        self.vpc = ec2.Vpc(
            self, f"{env_name}-AgentVPC",
            max_azs=2,
            subnet_configuration=[
                ec2.SubnetConfiguration(
                    name="Public",
                    subnet_type=ec2.SubnetType.PUBLIC,
                    cidr_mask=24
                ),
                ec2.SubnetConfiguration(
                    name="Private",
                    subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS,
                    cidr_mask=24
                )
            ],
            nat_gateways=1
        )
        
        # Create environment-specific resources
        self._create_secrets()
        self._create_dynamodb_tables()
        self._create_lambda_functions()
        self._create_monitoring()
        
        # Output NAT Gateway IP for Supabase whitelisting
        CfnOutput(
            self, f"{env_name}-SupabaseWhitelistIP",
            description=f"Add this IP to Supabase network restrictions for {env_name} environment"
        )
    
    def _create_secrets(self):
        """Create environment-specific secrets"""
        self.supabase_secret = secretsmanager.Secret(
            self, f"{self.env_name}-SupabaseCredentials",
            secret_string_value=SecretValue.unsafe_plain_text(
                json.dumps({
                    "project_id": "byoneafwuigeyehkoacw",
                    "password": os.getenv(f"SUPABASE_PASSWORD_{self.env_name.upper()}"),
                    "host": "aws-0-us-east-1.pooler.supabase.com",
                    "port": 6543,
                    "database": "postgres"
                })
            )
        )
    
    def _create_dynamodb_tables(self):
        """Create environment-specific DynamoDB tables"""
        self.consultations_table = dynamodb.Table(
            self, f"{self.env_name}-ConsultationsTable",
            partition_key=dynamodb.Attribute(name="userId", type=dynamodb.AttributeType.STRING),
            sort_key=dynamodb.Attribute(name="consultationId", type=dynamodb.AttributeType.STRING),
            removal_policy=RemovalPolicy.DESTROY,
            stream=dynamodb.StreamViewType.NEW_AND_OLD_IMAGES  # Enable streams
        )
        
        self.task_status_table = dynamodb.Table(
            self, f"{self.env_name}-TaskStatusTable",
            partition_key=dynamodb.Attribute(name="consultationId", type=dynamodb.AttributeType.STRING),
            removal_policy=RemovalPolicy.DESTROY,
            time_to_live_attribute="ttl"
        )
    
    def _create_lambda_functions(self):
        """Create modular Lambda functions"""
        
        # Common layer for shared utilities
        common_layer = _lambda.LayerVersion(
            self, f"{self.env_name}-CommonLayer",
            code=_lambda.Code.from_asset("lambdas/common"),
            compatible_runtimes=[_lambda.Runtime.PYTHON_3_11],
            description="Common utilities for Lambda functions"
        )
        
        # Supabase Stream Processor Lambda
        self.supabase_sync_lambda = _lambda.DockerImageFunction(
            self, f"{self.env_name}-SupabaseSyncLambda",
            code=_lambda.DockerImageCode.from_image_asset(
                directory="lambdas/supabase_stream_processor"
            ),
            timeout=Duration.seconds(60),
            memory_size=512,
            vpc=self.vpc,
            vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS),
            environment={
                'SUPABASE_SECRET_NAME': self.supabase_secret.secret_name,
                'ENV_NAME': self.env_name
            },
            layers=[common_layer]
        )
        
        # Add DynamoDB Stream event source
        self.supabase_sync_lambda.add_event_source(
            lambda_event_sources.DynamoEventSource(
                self.consultations_table,
                starting_position=_lambda.StartingPosition.LATEST,
                batch_size=25,
                max_batching_window_time=Duration.seconds(5),
                retry_attempts=3,
                on_failure=lambda_event_sources.SqsDlq(
                    sqs.Queue(self, f"{self.env_name}-SupabaseSyncDLQ")
                )
            )
        )
        
        # Grant permissions
        self.supabase_secret.grant_read(self.supabase_sync_lambda)
        self.consultations_table.grant_stream_read(self.supabase_sync_lambda)
    
    def _create_monitoring(self):
        """Create monitoring and alerting"""
        # CloudWatch alarms, dashboards, etc.
        pass

# Deploy dev environment
cd aws
cdk deploy AgentStack-dev -c env=dev

# Get the NAT Gateway IP from CDK output
# Add this IP to Supabase network restrictions

# Test the dev deployment
aws lambda invoke \
  --function-name Dev-SupabaseSyncLambda \
  --payload '{"test": true}' \
  response.json

# Deploy prod environment (when ready)
cdk deploy AgentStack-prod -c env=prod

# List all stacks
cdk list

# Destroy dev environment (if needed)
cdk destroy AgentStack-dev -c env=dev

# Monitor logs
aws logs tail /aws/lambda/Dev-SupabaseSyncLambda --follow