# Snowflake Streamlit Heavy Rotation App - Architecture Reference

## Project Status

**Phase 1: API Integration POC** - ✅ **COMPLETE**

Successfully implemented:
- ✅ Spotify OAuth token management and refresh
- ✅ DynamoDB integration for production presave tokens
- ✅ Batch processing of multiple fans (up to 200 tokens)
- ✅ Streamlit UI with progress tracking and error handling
- ✅ External Access Integration for secure API calls
- ✅ Export functionality (CSV/JSON)

**Phase 2: Database Integration & Production Features** - 🚧 **IN PROGRESS**

**For Phase 2 tasks, status, and planning, see the [Notion Board](https://www.notion.so/2fe87274f6d74b4ca587fc45e9be2398).**

This document serves as an architectural reference for Phase 1 implementation patterns.

---

## Phase 1 Overview

Built a Streamlit in Snowflake app that:
1. Directly queries DynamoDB for presave user tokens (using boto3) ✅
2. Calls Spotify API to get Heavy Rotation data for each user ✅
3. Displays results with in-memory storage (Phase 2 will add database persistence)

**Approach:** Follow dynamo-test pattern - query DynamoDB dynamically based on album ID ✅
**Scale:** 200 tokens per album analysis run ✅
**Security:** Use Snowflake Secrets + External Access Integration for secure AWS/Spotify API access ✅

---

## Phase 1 Implementation Details

### Snowflake Security Configuration (✅ Complete)

### Step 1.1: Create AWS Credentials Secret

```sql
-- Store AWS IAM credentials as JSON
CREATE OR REPLACE SECRET aws_dynamodb_credentials
  TYPE = GENERIC_STRING
  SECRET_STRING = '{
    "aws_access_key_id": "YOUR_AWS_ACCESS_KEY",
    "aws_secret_access_key": "YOUR_AWS_SECRET_KEY"
  }';
```

### Step 1.2: Create Spotify Credentials Secret

```sql
-- Store Spotify client credentials as JSON
CREATE OR REPLACE SECRET spotify_api_credentials
  TYPE = GENERIC_STRING
  SECRET_STRING = '{
    "client_id": "YOUR_SPOTIFY_CLIENT_ID",
    "client_secret": "YOUR_SPOTIFY_CLIENT_SECRET"
  }';
```

### Step 1.3: Create Network Rules

```sql
-- Whitelist DynamoDB endpoint (adjust region as needed)
CREATE OR REPLACE NETWORK RULE dynamodb_network_rule
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = ('dynamodb.us-east-1.amazonaws.com:443');

-- Whitelist Spotify API endpoints
CREATE OR REPLACE NETWORK RULE spotify_api_network_rule
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = ('api.spotify.com:443', 'accounts.spotify.com:443');
```

### Step 1.4: Create External Access Integration

```sql
-- Bundle secrets and network rules together
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION heavy_rotation_integration
  ALLOWED_NETWORK_RULES = (dynamodb_network_rule, spotify_api_network_rule)
  ALLOWED_AUTHENTICATION_SECRETS = (aws_dynamodb_credentials, spotify_api_credentials)
  ENABLED = TRUE;
```

### Step 1.5: Grant Privileges

```sql
-- Grant to your role (replace YOUR_ROLE with actual role name)
GRANT READ ON SECRET aws_dynamodb_credentials TO ROLE YOUR_ROLE;
GRANT READ ON SECRET spotify_api_credentials TO ROLE YOUR_ROLE;
GRANT USAGE ON INTEGRATION heavy_rotation_integration TO ROLE YOUR_ROLE;

-- After creating the Streamlit app, also grant to the app:
-- GRANT READ ON SECRET aws_dynamodb_credentials TO APPLICATION IDENTIFIER('HEAVY_ROTATION_APP');
-- GRANT READ ON SECRET spotify_api_credentials TO APPLICATION IDENTIFIER('HEAVY_ROTATION_APP');
-- GRANT USAGE ON INTEGRATION heavy_rotation_integration TO APPLICATION IDENTIFIER('HEAVY_ROTATION_APP');
```

---

## Streamlit Application Architecture (✅ Complete - Phase 1)

### Project Structure

```
streamlit-test/
├── heavy_rotation_app.py      # Main Streamlit application
├── environment.yml            # Python dependencies (boto3, requests)
├── plan.md                    # This file (Phase 1 architecture reference)
└── README.md                  # Setup and usage instructions
```

### Step 3.1: Create environment.yml

```yaml
name: heavy_rotation_env
channels:
  - snowflake
dependencies:
  - python=3.10
  - snowflake-snowpark-python
  - boto3>=1.34.0
  - requests>=2.31.0
```

### Step 3.2: Build streamlit_app.py

The app consists of 6 main sections:

#### 1. Setup & Configuration

```python
import streamlit as st
import boto3
import requests
import json
import base64
from datetime import datetime
import uuid
from snowflake.snowpark.context import get_active_session

st.set_page_config(page_title="Heavy Rotation Analysis", layout="wide")
session = get_active_session()
```

#### 2. Cached Clients (Using @st.cache_resource)

```python
@st.cache_resource
def get_dynamodb_client():
    """Initialize and cache boto3 DynamoDB client using Snowflake secrets"""
    creds_json = st.secrets["aws_dynamodb_credentials"]
    creds = json.loads(creds_json)

    client = boto3.client(
        'dynamodb',
        region_name='us-east-1',
        aws_access_key_id=creds['aws_access_key_id'],
        aws_secret_access_key=creds['aws_secret_access_key']
    )
    return client

@st.cache_resource
def get_spotify_credentials():
    """Load Spotify API credentials from Snowflake secrets"""
    creds_json = st.secrets["spotify_api_credentials"]
    creds = json.loads(creds_json)
    return creds['client_id'], creds['client_secret']
```

**Key Pattern:** `@st.cache_resource` prevents recreating boto3 client on every interaction, improving performance.

#### 3. DynamoDB Query Functions

```python
def query_presave_tokens(album_id: str, limit: int = 200):
    """
    Query DynamoDB for presave user tokens
    Port of dynamo-test/src/dynamodb-client.ts logic

    Query pattern:
    - Partition key: f"group:album{album_id}"
    - Sort key begins with: "task:spotify-presave"
    - Limit: {limit} tokens
    """
    client = get_dynamodb_client()

    response = client.query(
        TableName='songwhip-release-tasks-production',
        KeyConditionExpression='partitionKey = :pk AND begins_with(sortKey, :sk)',
        ExpressionAttributeValues={
            ':pk': {'S': f'group:album{album_id}'},
            ':sk': {'S': 'task:spotify-presave'}
        },
        Limit=limit
    )

    # Parse and return tokens
    tokens = []
    for item in response.get('Items', []):
        token_data = {
            'fan_id': item['sortKey']['S'],
            'refresh_token': item.get('refreshToken', {}).get('S', ''),
            'spotify_user_id': item.get('spotifyUserId', {}).get('S', ''),
            'album_id': album_id
        }
        if token_data['refresh_token']:
            tokens.append(token_data)

    return tokens
```

#### 4. Spotify API Functions

```python
def refresh_spotify_token(refresh_token: str):
    """Port of auth.ts SpotifyAuth.refreshAccessToken()"""
    client_id, client_secret = get_spotify_credentials()

    # Base64 encode credentials
    auth_string = f"{client_id}:{client_secret}"
    auth_base64 = base64.b64encode(auth_string.encode('utf-8')).decode('utf-8')

    response = requests.post(
        'https://accounts.spotify.com/api/token',
        data={
            'grant_type': 'refresh_token',
            'refresh_token': refresh_token
        },
        headers={
            'Authorization': f'Basic {auth_base64}',
            'Content-Type': 'application/x-www-form-urlencoded'
        },
        timeout=10
    )

    if response.status_code == 200:
        return {'success': True, 'access_token': response.json()['access_token']}
    else:
        return {'success': False, 'error': f"HTTP {response.status_code}"}

def fetch_top_artists(access_token: str, time_range: str, limit: int = 50):
    """Port of fetch-heavy-rotation.ts getTopArtistsPaginated()"""
    response = requests.get(
        'https://api.spotify.com/v1/me/top/artists',
        params={'time_range': time_range, 'limit': limit, 'offset': 0},
        headers={'Authorization': f'Bearer {access_token}'},
        timeout=10
    )

    if response.status_code == 200:
        data = response.json()
        return {
            'success': True,
            'data': data['items'],
            'total': data.get('total', 0),
            'api_response_code': 200
        }
    else:
        return {
            'success': False,
            'error': f"HTTP {response.status_code}",
            'api_response_code': response.status_code
        }
```

#### 5. Data Processing Functions

```python
def process_fan_batch(job_id: str, tokens: list, artist_id: str, time_range: str):
    """Process a batch of fans - refresh tokens and fetch top artists"""
    results = []

    for token_data in tokens:
        fan_id = token_data['fan_id']
        refresh_token = token_data['refresh_token']

        # Step 1: Refresh access token
        auth_result = refresh_spotify_token(refresh_token)
        if not auth_result['success']:
            # Log error and continue
            session.sql(f"""
                INSERT INTO POC_ERROR_LOG (job_id, fan_id, error_type, error_message)
                VALUES ('{job_id}', '{fan_id}', 'TOKEN_REFRESH_FAILED',
                        '{auth_result.get("error", "")}')
            """).collect()
            continue

        # Step 2: Fetch top artists
        api_result = fetch_top_artists(auth_result['access_token'], time_range)

        # Step 3: Store raw response as VARIANT
        raw_response_json = json.dumps(api_result)
        session.sql(f"""
            INSERT INTO POC_RAW_LISTENING_DATA
            (job_id, fan_id, spotify_user_id, album_id, time_range,
             raw_response, api_response_code)
            VALUES (
                '{job_id}',
                '{fan_id}',
                '{token_data.get("spotify_user_id", "")}',
                '{token_data["album_id"]}',
                '{time_range}',
                PARSE_JSON('{raw_response_json}'),
                {api_result.get('api_response_code', 0)}
            )
        """).collect()

        results.append({'fan_id': fan_id, 'success': api_result['success']})

    return results

def aggregate_heavy_listeners(job_id: str, artist_id: str):
    """Flatten VARIANT data and identify heavy rotation fans"""
    session.sql(f"""
        INSERT INTO POC_AGG_HEAVY_LISTENERS
        (job_id, fan_id, spotify_user_id, artist_id, album_id, time_range,
         artist_found, artist_rank, artist_name, genres)
        SELECT
            raw.job_id,
            raw.fan_id,
            raw.spotify_user_id,
            '{artist_id}' as artist_id,
            raw.album_id,
            raw.time_range,
            TRUE as artist_found,
            artists.index + 1 as artist_rank,
            artists.value:name::STRING as artist_name,
            artists.value:genres::ARRAY as genres
        FROM POC_RAW_LISTENING_DATA raw,
        LATERAL FLATTEN(input => raw.raw_response:data) artists
        WHERE raw.job_id = '{job_id}'
          AND raw.raw_response:success = TRUE
          AND artists.value:id::STRING = '{artist_id}'

        UNION ALL

        -- Also insert records where artist was NOT found
        SELECT
            raw.job_id,
            raw.fan_id,
            raw.spotify_user_id,
            '{artist_id}' as artist_id,
            raw.album_id,
            raw.time_range,
            FALSE as artist_found,
            NULL as artist_rank,
            NULL as artist_name,
            NULL as genres
        FROM POC_RAW_LISTENING_DATA raw
        WHERE raw.job_id = '{job_id}'
          AND raw.raw_response:success = TRUE
          AND NOT EXISTS (
              SELECT 1
              FROM LATERAL FLATTEN(input => raw.raw_response:data) f
              WHERE f.value:id::STRING = '{artist_id}'
          )
    """).collect()
```

**Key Pattern:** Use `LATERAL FLATTEN` to parse VARIANT column and search for target artist.

#### 6. Streamlit UI

```python
st.title("🎵 Heavy Rotation Listener Analysis")
st.markdown("Analyze which presave fans have an artist in their Spotify Top Artists list")

# Input section
col1, col2, col3 = st.columns(3)
with col1:
    album_id = st.text_input("Album ID", value="6644715")
with col2:
    artist_id = st.text_input("Spotify Artist ID", value="4gzpq5DPGxSnKTe4SA8HAU")
with col3:
    time_range_label = st.selectbox("Time Range",
        ["Short (4 weeks)", "Medium (6 months)", "Long (1 year)"])

    time_range_map = {
        "Short (4 weeks)": "short_term",
        "Medium (6 months)": "medium_term",
        "Long (1 year)": "long_term"
    }
    time_range = time_range_map[time_range_label]

# STEP 1: Fetch tokens
if st.button("Query DynamoDB", type="primary"):
    tokens = query_presave_tokens(album_id, limit=200)
    if tokens:
        st.session_state['tokens'] = tokens
        st.success(f"✅ Found {len(tokens)} presave users")

# STEP 2: Run analysis
if 'tokens' in st.session_state:
    if st.button("Run Analysis", type="primary"):
        tokens = st.session_state['tokens']
        job_id = str(uuid.uuid4())

        # Initialize job
        session.sql(f"""
            INSERT INTO POC_JOB_STATUS
            (job_id, album_id, artist_id, time_range, status, total_fans)
            VALUES ('{job_id}', '{album_id}', '{artist_id}',
                    '{time_range}', 'RUNNING', {len(tokens)})
        """).collect()

        # Process in batches of 100
        batch_size = 100
        progress_bar = st.progress(0)

        for i in range(0, len(tokens), batch_size):
            batch = tokens[i:i+batch_size]
            process_fan_batch(job_id, batch, artist_id, time_range)
            progress_bar.progress(min((i + batch_size) / len(tokens), 1.0))

        # Aggregate results
        aggregate_heavy_listeners(job_id, artist_id)

        # Finalize
        session.sql(f"""
            UPDATE POC_JOB_STATUS
            SET status = 'COMPLETED', completed_at = CURRENT_TIMESTAMP()
            WHERE job_id = '{job_id}'
        """).collect()

        st.success(f"✅ Analysis complete! Job ID: {job_id}")
        st.session_state['job_id'] = job_id
        st.session_state['show_results'] = True

# STEP 3: Display results
if st.session_state.get('show_results'):
    job_id = st.session_state['job_id']

    results_df = session.sql(f"""
        SELECT fan_id, spotify_user_id, artist_found, artist_rank,
               artist_name, genres, processed_at
        FROM POC_AGG_HEAVY_LISTENERS
        WHERE job_id = '{job_id}'
        ORDER BY artist_rank ASC NULLS LAST
    """).to_pandas()

    # Metrics
    total = len(results_df)
    heavy_rotation = len(results_df[results_df['artist_found'] == True])
    percentage = (heavy_rotation / total * 100) if total > 0 else 0

    col1, col2, col3 = st.columns(3)
    col1.metric("Total Fans", total)
    col2.metric("Heavy Rotation", heavy_rotation)
    col3.metric("Match %", f"{percentage:.1f}%")

    st.dataframe(results_df, use_container_width=True)

    # Download CSV
    csv = results_df.to_csv(index=False)
    st.download_button("📥 Download CSV", csv,
                      f"heavy_rotation_{job_id}.csv", "text/csv")
```

---

## Phase 4: Deployment & Testing

### Step 4.1: Prepare SQL Setup Files

Create two files:

**File: `sql/setup_security.sql`**
- Contains all Phase 1 SQL (secrets, network rules, integration, grants)

**File: `sql/setup_tables.sql`**
- Contains all Phase 2 SQL (database, schema, 4 tables)

### Step 4.2: Deploy to Snowflake

1. **Run SQL Setup**
   - Execute `setup_security.sql` in Snowflake worksheet
   - Execute `setup_tables.sql` in Snowflake worksheet

2. **Create Streamlit App**
   - In Snowflake UI: Projects → Streamlit → + Streamlit App
   - Upload `streamlit_app.py` as main file
   - Upload `environment.yml` for dependencies
   - Select warehouse and database

3. **Grant App Privileges**
   ```sql
   GRANT READ ON SECRET aws_dynamodb_credentials
     TO APPLICATION IDENTIFIER('HEAVY_ROTATION_APP');
   GRANT READ ON SECRET spotify_api_credentials
     TO APPLICATION IDENTIFIER('HEAVY_ROTATION_APP');
   GRANT USAGE ON INTEGRATION heavy_rotation_integration
     TO APPLICATION IDENTIFIER('HEAVY_ROTATION_APP');
   ```

### Step 4.3: Testing Checklist

- [ ] App loads without errors
- [ ] DynamoDB query returns tokens for album ID 6644715
- [ ] Token preview displays correctly
- [ ] Token refresh works (test with 1 fan first)
- [ ] Top artists API call succeeds
- [ ] Raw data stored in POC_RAW_LISTENING_DATA with VARIANT
- [ ] Aggregation correctly identifies artist matches
- [ ] Progress bar updates during processing
- [ ] Results display with correct metrics
- [ ] CSV download works
- [ ] Error handling graceful (invalid album ID, API failures)
- [ ] Job history shows past runs

---

## Key Technical Decisions

### 1. Security Pattern
- **Snowflake Secrets** store credentials (AWS + Spotify) as JSON
- **Network Rules** whitelist specific API endpoints only
- **External Access Integration** bundles secrets + rules
- **st.secrets** automatically resolves references in Streamlit

### 2. DynamoDB Query Pattern
Exactly matches dynamo-test:
```python
partition_key = f"group:album{album_id}"
sort_key_prefix = "task:spotify-presave"
limit = 200
```

### 3. Caching Strategy
- `@st.cache_resource` on boto3 client (prevents reconnecting)
- `@st.cache_resource` on Spotify credentials (read once)
- Session state for tokens (persists during app interactions)

### 4. Batch Processing
- Process 100 fans per batch (manageable for rate limits)
- Sequential processing (simpler error handling)
- Progress bar updates between batches
- Spotify rate limit: 180 requests/minute (safe margin)

### 5. Error Handling
- Token refresh failures displayed in UI
- API errors shown with detailed error messages
- Processing continues even if some fans fail
- Graceful degradation (show partial results)

### 6. Data Flow (Phase 1)
```
User Input (Album ID)
  ↓
Query DynamoDB (boto3, cached client)
  ↓
Store tokens in session_state (200 fans)
  ↓
Process Batch 1 (100 fans)
  → Refresh token (Spotify OAuth)
  → Fetch top artists (Spotify API)
  → Store in session_state
  ↓
Process Batch 2 (100 fans)
  → Same process
  ↓
Display results (metrics + table + CSV)
```

**Phase 2 Enhancement:** Database persistence with Snowflake tables (see Notion for tasks)

---

## Migration from TypeScript to Python

| Component | dynamo-test (TypeScript) | streamlit-test (Python) |
|-----------|-------------------------|-------------------------|
| DynamoDB Client | @aws-sdk/client-dynamodb | boto3 |
| AWS Auth | awsume + fromIni() | boto3.client() + Snowflake secrets |
| Token Fetch | getPresaveTokensForAlbum() | query_presave_tokens() |
| Spotify Auth | SpotifyAuth class | refresh_spotify_token() |
| API Calls | axios + interceptors | requests |
| Data Storage | Console output | Session state (Phase 1), Snowflake tables (Phase 2) |
| UI | Terminal | Streamlit |
| Base64 Encoding | Buffer.from().toString() | base64.b64encode() |
| JSON Parsing | JSON.parse() | json.loads() |
| Error Handling | try-catch | try-except |

---

## Phase 1 Success Criteria (✅ Complete)

✅ Snowflake secrets configured correctly with AWS + Spotify credentials
✅ External Access Integration allows DynamoDB + Spotify API calls
✅ App loads and displays clean UI
✅ DynamoDB query returns up to 200 tokens for album
✅ Spotify token refresh succeeds
✅ Top artists API call returns data
✅ Progress bar tracks batch processing accurately
✅ Results display with correct metrics
✅ CSV/JSON export works
✅ Error handling graceful (shows errors, continues processing)

---

## References

- **[Notion Board](https://www.notion.so/2fe87274f6d74b4ca587fc45e9be2398)** - Phase 2 tasks and current status
- **dynamo-test/** - DynamoDB query pattern reference
- **api-test/** - Spotify API integration pattern reference
- **docs/implementation.md** - Original Snowflake architecture design
- **CLAUDE.md** - Project context and workflow guidance