> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cfosilvia.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Lambda supabase integration proposal

# Lambda-Supabase Integration Proposal (Updated)

## Executive Summary

This proposal outlines how to integrate Supabase PostgreSQL database with your existing AWS Lambda functions for streaming data from DynamoDB to Supabase. The solution leverages the IPv4-compatible connection patterns discovered during EC2 testing and adapts them for Lambda's serverless environment with a focus on simplicity, modularity, and production safety.

## Architecture Overview

### Current State

* **DynamoDB Tables**: ConsultationsTable, TaskStatusTable
* **Lambda Functions**: Agent processing with SQS triggers
* **Missing**: Real-time data synchronization to Supabase

### Proposed Architecture

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

## Key Technical Decisions

### 1. VPC and Static IP Requirements

**Critical Networking Component:**

* Lambda functions require a **static IP address** to connect to Supabase
* Supabase requires whitelisting specific IP addresses for database access
* Solution: Deploy Lambdas in a **VPC with NAT Gateway** and **Elastic IP**

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

### 2. Environment Isolation Strategy

**Multi-Environment Support:**

* Parameterized CDK stack supporting `dev` and `prod` environments
* All resources prefixed with environment name (e.g., `Dev-ConsultationsTable`)
* Separate CloudFormation stacks (e.g., `AgentStack-dev`, `AgentStack-prod`)
* Isolated VPCs and IP addresses per environment

### 3. Modular Lambda Architecture

**Restructured Directory Layout:**

```
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
```

### 4. Connection Strategy for Lambda

**✅ Transaction Pooler (Recommended for Lambda)**

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

**Why Transaction Pooler over Session Pooler?**

* Lambda functions are ephemeral (short-lived)
* Transaction pooler designed for serverless workloads
* Better connection management for Lambda's lifecycle
* Automatic connection cleanup

### 5. Simplified Dependencies Strategy

**Lightweight Dependencies for Lambda:**

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

**Key Simplifications:**

* Use `boto3.dynamodb.types.TypeDeserializer` instead of manual unmarshalling
* Single-file Lambda handlers for better maintainability
* Cached connection pattern leveraging Lambda execution context reuse

## Implementation Plan

### Phase 1: Environment Configuration

#### 1.1 CDK Environment Parameterization

```python theme={null}
# 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()
```

#### 1.2 VPC with Static IP Configuration

```python theme={null}
# 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"
        )
```

### Phase 2: Simplified Lambda Implementation

#### 2.1 Common Database Connector

```python theme={null}
# 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
```

#### 2.2 Simplified Supabase Stream Processor

```python theme={null}
# 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
```

#### 2.3 Supabase Scenarios Table Schema

```sql theme={null}
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)
);
```

### Phase 3: Updated CDK Stack Configuration

#### 3.1 Environment-Aware Resource Creation

```python theme={null}
# 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
```

## Deployment Strategy

### 1. Pre-deployment Checklist

* [ ] Create Supabase scenarios table in dev environment
* [ ] Set environment variables in `.env` file:
  * `SUPABASE_PASSWORD_DEV`
  * `SUPABASE_PASSWORD_PROD`
* [ ] Test Lambda functions locally
* [ ] Verify VPC and NAT Gateway configuration

### 2. Deployment Commands

```bash theme={null}
# 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
```

### 3. Environment Management

```bash theme={null}
# 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
```

## Cost Analysis

### Per Environment Costs

**VPC Components:**

* **NAT Gateway**: \~\$45/month per environment
* **Elastic IP**: \~\$3.65/month per environment

**Lambda Costs:**

* **Invocations**: \~\$0.20 per 1M requests
* **Duration**: \~\$0.0000166667 per GB-second
* **Estimated monthly cost**: \$5-20 per environment

**DynamoDB Streams:**

* **Read requests**: \$0.02 per 100,000 read request units
* **Estimated monthly cost**: \$1-5 per environment

**Total estimated cost per environment: \$55-75/month**

## Success Metrics

### Technical Metrics

* ✅ Connection success rate > 99%
* ✅ Processing latency \< 5 seconds
* ✅ Error rate \< 1%
* ✅ Zero data loss (idempotency working)
* ✅ Static IP connectivity to Supabase

### Business Metrics

* ✅ Real-time data availability in Supabase
* ✅ Environment isolation (dev/prod)
* ✅ Reduced deployment complexity
* ✅ Improved data consistency across systems

## Risk Mitigation

### 1. Network Connectivity

* **Risk**: Lambda cannot reach Supabase
* **Mitigation**: VPC with NAT Gateway + IP whitelisting

### 2. Environment Conflicts

* **Risk**: Dev changes affecting prod
* **Mitigation**: Complete resource isolation with environment prefixes

### 3. Connection Failures

* **Risk**: Supabase unavailable
* **Mitigation**: DLQ + retry logic + connection caching

### 4. Data Loss

* **Risk**: Lambda failures
* **Mitigation**: DynamoDB streams retain data for 24 hours + idempotency

## Next Steps

1. **Review and approve** this updated proposal
2. **Set up dev Supabase project** and create scenarios table
3. **Configure environment variables** in `.env` file
4. **Deploy dev stack** using `cdk deploy AgentStack-dev -c env=dev`
5. **Add NAT Gateway IP** to Supabase network restrictions
6. **Test end-to-end flow** with sample data
7. **Monitor and optimize** based on real usage patterns
8. **Plan production migration** once dev environment is validated

This updated proposal provides a production-ready, environment-isolated solution for streaming data from DynamoDB to Supabase with proper networking, simplified architecture, and safe deployment practices.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.