> ## 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.

# Radar backend architecture

# Radar Backend Architecture

## Overview

**Radar** is a proactive AI insights system that runs scheduled prompts for users. The system executes financial analysis on a schedule (daily/weekly/monthly/custom) and stores results for users to review later.

**Key Features:**

* Scheduled execution of AI prompts (portfolio analysis, spending alerts, goal tracking)
* Single daily EventBridge trigger with smart filtering
* Scalable architecture supporting millions of users
* Pre-calculated `next_run` timestamps for efficient querying
* Cost-effective (\~\$5-10/month for 10K users)

***

## System Flow

```
EventBridge (6am UTC) → Orchestrator Lambda → SQS FIFO Queue → Processor Lambda → DynamoDB Results
```

**Daily Execution:**

1. EventBridge triggers Orchestrator at 6am UTC
2. Orchestrator queries DynamoDB for schedules with `next_run_date = today`
3. Matching schedules are sent to SQS FIFO queue (grouped by user\_id)
4. Processor Lambda consumes queue, executes AI prompts with user financial data
5. Results stored in DynamoDB, `next_run` updated for next occurrence

***

## Technical Diagram

```
┌─────────────────────────────────────────────────────────────┐
│                     RADAR SYSTEM                             │
└─────────────────────────────────────────────────────────────┘

EventBridge Rule                    DynamoDB Tables
┌──────────────┐                   ┌─────────────────────┐
│ Daily 6am UTC│──────────────────▶│ radar_schedules     │
└──────────────┘                   │ PK: USER#id         │
                                    │ SK: SCHEDULE#type   │
       │                            │ next_run: timestamp │
       │                            │ GSI: next_run_date  │
       ▼                            └─────────────────────┘
┌──────────────────────┐
│ Orchestrator Lambda  │
│ - Query next_run     │
│ - Send to SQS        │
└──────────────────────┘
       │
       ▼
┌──────────────────────┐
│ SQS FIFO Queue       │
│ - Per-user ordering  │
│ - Rate limiting      │
│ - DLQ for failures   │
└──────────────────────┘
       │
       ▼
┌──────────────────────┐           ┌─────────────────────┐
│ Processor Lambda     │──────────▶│ radar_results       │
│ - Fetch user data    │           │ PK: USER#id         │
│ - Call AI with tools │           │ SK: timestamp#type  │
│ - Store results      │           │ status: completed   │
│ - Update next_run    │           └─────────────────────┘
└──────────────────────┘
```

***

## Data Models

### DynamoDB Table: `radar_schedules`

```typescript theme={null}
{
  PK: "USER#clerk_user_123",              // Partition Key
  SK: "SCHEDULE#PORTFOLIO_RISK",          // Sort Key
  schedule_id: "sched_xyz789",
  radar_type: "PORTFOLIO_RISK",
  prompt_template: "Analyze my portfolio risk exposure",
  enabled: true,
  
  // Schedule configuration
  schedule_config: {
    frequency: "WEEKLY",                   // DAILY | WEEKLY | MONTHLY | CUSTOM
    days_of_week: [1, 3],                  // Monday, Wednesday (ISO 8601)
    time_utc: "06:00",
    timezone: "America/New_York",          // For display only
    day_of_month?: 1                       // For MONTHLY schedules
  },
  
  last_run: "2025-01-22T06:00:00Z",
  next_run: "2025-01-24T06:00:00Z",       // Pre-calculated in UTC
  next_run_date: "2025-01-24",            // GSI key for efficient querying
  created_at: "2025-01-01T00:00:00Z"
}
```

**GSI:** `next_run_date-index`

* Partition Key: `next_run_date` (string, format: YYYY-MM-DD)
* Enables efficient query: "Find all schedules due today"

***

### DynamoDB Table: `radar_results`

```typescript theme={null}
{
  PK: "USER#clerk_user_123",
  SK: "#PORTFOLIO_RISK#2025-01-23T06:00:00Z",  // Type first for sorting
  radar_id: "radar_abc123",
  radar_type: "PORTFOLIO_RISK",
  prompt: "Analyze my portfolio risk exposure",
  result: {
    summary: "Your tech exposure increased 5%...",     // <30 words for tiles
    full_analysis: "Detailed analysis...",
    insights: [
      {
        type: "warning",                                // warning | info | success
        title: "High tech concentration",
        description: "65% vs 40% recommended",
        priority: "high"                                // high | medium | low
      }
    ],
    recommendations: [
      {
        action: "Sell 10% of tech holdings",
        reason: "Reduce concentration risk",
        impact: "high"                                  // high | medium | low
      }
    ],
    metrics: {
      "tech_exposure": 0.65,
      "portfolio_volatility": 0.18
    }
  },
  status: "COMPLETED",                                  // PENDING | PROCESSING | COMPLETED | FAILED
  created_at: "2025-01-23T06:00:00Z",
  completed_at: "2025-01-23T06:02:15Z",
  metadata: {
    model_used: "claude-sonnet-4-5",
    tokens_used: 1500,
    processing_time_ms: 2150
  }
}
```

***

## File Structure

```
aws/
├── radar/
│   ├── __init__.py
│   ├── radar_stack.py              # CDK infrastructure definition
│   ├── lambdas/
│   │   ├── orchestrator/
│   │   │   ├── handler.py          # Query schedules, send to SQS
│   │   │   └── requirements.txt    # boto3, etc.
│   │   └── processor/
│   │       ├── handler.py          # Execute AI prompts, store results
│   │       └── requirements.txt    # anthropic, openai, boto3
│   └── utils/
│       ├── schedule_calculator.py  # Calculate next_run timestamps
│       └── ai_client.py            # Shared AI SDK logic
└── app.py                          # Add: app.add(RadarStack(app, "RadarStack"))
```

***

## Lambda Functions

### Orchestrator Lambda

**Trigger:** EventBridge (daily at 6am UTC)\
**Purpose:** Query schedules due today and send to SQS

```python theme={null}
def handler(event, context):
    """
    Queries radar_schedules where next_run_date = today.
    Sends matching schedules to SQS FIFO queue.
    """
    today = datetime.utcnow().date().isoformat()
    
    # Query schedules due today using GSI
    response = dynamodb.query(
        TableName='radar_schedules',
        IndexName='next_run_date-index',
        KeyConditionExpression='next_run_date = :today',
        ExpressionAttributeValues={':today': today}
    )
    
    # Batch send to SQS (10 messages per batch)
    for batch in chunk(response['Items'], 10):
        sqs.send_message_batch(
            QueueUrl=RADAR_QUEUE_URL,
            Entries=[{
                'Id': schedule['schedule_id'],
                'MessageBody': json.dumps(schedule),
                'MessageGroupId': schedule['user_id']  # FIFO grouping per user
            } for schedule in batch]
        )
    
    return {'queued': len(response['Items'])}
```

***

### Processor Lambda

**Trigger:** SQS FIFO Queue\
**Purpose:** Execute AI prompts and store results

```python theme={null}
def handler(event, context):
    """
    Processes radar jobs from SQS.
    Fetches user data, calls AI with tools, stores results.
    """
    for record in event['Records']:
        schedule = json.loads(record['body'])
        
        # Create pending result record
        radar_id = create_radar_result(schedule, status='PENDING')
        
        try:
            # Fetch user financial data from Supabase
            user_data = fetch_user_data(schedule['user_id'])
            
            # Execute AI prompt with tools (similar to chat route)
            result = execute_ai_prompt(
                prompt=schedule['prompt_template'],
                user_data=user_data,
                tools=create_chat_tools(schedule['user_id'])
            )
            
            # Update result to COMPLETED
            update_radar_result(radar_id, {
                'status': 'COMPLETED',
                'result': result,
                'completed_at': datetime.utcnow().isoformat()
            })
            
            # Calculate and update next_run
            next_run = calculate_next_run(schedule['schedule_config'])
            update_schedule_next_run(schedule['schedule_id'], next_run)
            
        except Exception as e:
            # Update result to FAILED
            update_radar_result(radar_id, {
                'status': 'FAILED',
                'error': str(e)
            })
```

***

## CDK Infrastructure

### Key Resources

```python theme={null}
class RadarStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs):
        super().__init__(scope, id, **kwargs)
        
        # DynamoDB Tables
        schedules_table = dynamodb.Table(
            self, "RadarSchedules",
            partition_key=dynamodb.Attribute(
                name="PK",
                type=dynamodb.AttributeType.STRING
            ),
            sort_key=dynamodb.Attribute(
                name="SK",
                type=dynamodb.AttributeType.STRING
            ),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST
        )
        
        # GSI for efficient querying by next_run_date
        schedules_table.add_global_secondary_index(
            index_name="next_run_date-index",
            partition_key=dynamodb.Attribute(
                name="next_run_date",
                type=dynamodb.AttributeType.STRING
            )
        )
        
        results_table = dynamodb.Table(
            self, "RadarResults",
            partition_key=dynamodb.Attribute(
                name="PK",
                type=dynamodb.AttributeType.STRING
            ),
            sort_key=dynamodb.Attribute(
                name="SK",
                type=dynamodb.AttributeType.STRING
            ),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST
        )
        
        # SQS FIFO Queue
        dlq = sqs.Queue(self, "RadarDLQ", fifo=True)
        
        radar_queue = sqs.Queue(
            self, "RadarQueue",
            fifo=True,
            content_based_deduplication=True,
            visibility_timeout=Duration.minutes(15),
            dead_letter_queue=sqs.DeadLetterQueue(
                max_receive_count=3,
                queue=dlq
            )
        )
        
        # Lambda Functions
        orchestrator = lambda_.Function(
            self, "RadarOrchestrator",
            runtime=lambda_.Runtime.PYTHON_3_11,
            handler="handler.handler",
            code=lambda_.Code.from_asset("radar/lambdas/orchestrator"),
            environment={
                "SCHEDULES_TABLE": schedules_table.table_name,
                "QUEUE_URL": radar_queue.queue_url
            },
            timeout=Duration.minutes(5)
        )
        
        processor = lambda_.Function(
            self, "RadarProcessor",
            runtime=lambda_.Runtime.PYTHON_3_11,
            handler="handler.handler",
            code=lambda_.Code.from_asset("radar/lambdas/processor"),
            environment={
                "SCHEDULES_TABLE": schedules_table.table_name,
                "RESULTS_TABLE": results_table.table_name,
                "SUPABASE_URL": os.getenv("SUPABASE_URL"),
                "SUPABASE_KEY": os.getenv("SUPABASE_SERVICE_KEY"),
                "ANTHROPIC_API_KEY": os.getenv("ANTHROPIC_API_KEY")
            },
            timeout=Duration.minutes(15)
        )
        
        # EventBridge Rule (Daily at 6am UTC)
        rule = events.Rule(
            self, "DailyRadarTrigger",
            schedule=events.Schedule.cron(
                minute="0",
                hour="6",
                month="*",
                week_day="*",
                year="*"
            )
        )
        rule.add_target(targets.LambdaFunction(orchestrator))
        
        # SQS Event Source for Processor
        processor.add_event_source(
            lambda_event_sources.SqsEventSource(
                radar_queue,
                batch_size=10,
                max_batching_window=Duration.seconds(5)
            )
        )
        
        # Permissions
        schedules_table.grant_read_data(orchestrator)
        radar_queue.grant_send_messages(orchestrator)
        schedules_table.grant_read_write_data(processor)
        results_table.grant_write_data(processor)
```

***

## Frontend Integration

### API Routes

```typescript theme={null}
// app/api/radar/dashboard/route.ts
// Returns latest result per radar type (for dashboard tiles)

// app/api/radar/history/[radarType]/route.ts
// Returns paginated history for specific radar type (for detail view)

// app/api/radar/trigger/route.ts
// Manually trigger a radar run (sends to SQS immediately)
```

### Components

```typescript theme={null}
// components/radar/radar-dashboard.tsx
// Grid of radar tiles showing latest results

// components/radar/radar-tile.tsx
// Individual tile with summary, timestamp, history link

// app/radar/[radarType]/page.tsx
// Detail view with expanded latest result + history
```

***

## Schedule Configuration Examples

### Daily Schedule

```typescript theme={null}
{
  frequency: "DAILY",
  time_utc: "06:00",
  timezone: "America/New_York"
}
// Runs every day at 6am UTC
```

### Weekly Schedule

```typescript theme={null}
{
  frequency: "WEEKLY",
  days_of_week: [1, 3, 5],  // Monday, Wednesday, Friday
  time_utc: "06:00",
  timezone: "America/New_York"
}
// Runs Mon/Wed/Fri at 6am UTC
```

### Monthly Schedule

```typescript theme={null}
{
  frequency: "MONTHLY",
  day_of_month: 1,
  time_utc: "06:00",
  timezone: "America/New_York"
}
// Runs 1st of each month at 6am UTC
```

### Custom Schedule

```typescript theme={null}
{
  frequency: "CUSTOM",
  custom_pattern: {
    type: "EVERY_N_DAYS",
    interval: 3,
    start_date: "2025-01-01"
  },
  time_utc: "06:00"
}
// Runs every 3 days starting from Jan 1
```

***

## Cost Estimation

**Assumptions:** 10,000 active users, mix of daily/weekly schedules

| Service | Usage | Monthly Cost |
| - | - | - |
| Lambda (Orchestrator) | 30 invocations/month @ 512MB, 1s | \$0.01 |
| Lambda (Processor) | \~3,770 invocations/day @ 512MB, 30s | \$4.50 |
| DynamoDB | 30K writes, 100K reads (on-demand) | \$2.00 |
| SQS | 113K requests/month | \$0.50 |
| EventBridge | 30 events/month | \$0.00 |
| **Total** | | **\~\$7.00/month** |

***

## Performance Metrics

* **Orchestrator execution time:** \~750ms for 10K users
* **Queue processing:** \~3,770 jobs/day = \~2.6 jobs/minute
* **Processor execution time:** 10-30 seconds per job (AI call)
* **End-to-end latency:** 1-5 minutes from trigger to result stored

***

## Monitoring & Alerts

### CloudWatch Metrics

* Orchestrator invocation count & duration
* Processor invocation count & duration
* SQS queue depth & age of oldest message
* DynamoDB read/write capacity utilization

### CloudWatch Alarms

* Processor error rate > 5%
* DLQ message count > 0
* Queue depth > 1000 messages
* Lambda timeout errors

***

## Deployment

```bash theme={null}
# Install dependencies
cd aws
pip install -r requirements.txt

# Deploy CDK stack
cdk deploy RadarStack

# Deploy with environment variables
cdk deploy RadarStack \
  --context supabase_url=$SUPABASE_URL \
  --context supabase_key=$SUPABASE_SERVICE_KEY \
  --context anthropic_key=$ANTHROPIC_API_KEY
```

***

## Future Enhancements

1. **Multiple daily triggers** - Add 12pm, 6pm, 12am UTC for timezone flexibility
2. **Priority queue** - Separate queue for urgent alerts vs routine checks
3. **Real-time notifications** - Push notifications when high-priority radar completes
4. **Radar templates** - Pre-built radar types users can enable with one click
5. **AI-powered scheduling** - "Analyze my portfolio when volatility spikes"


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