# Heavy Rotation Listeners POC - Implementation Plan

## Table of Contents
1. [Overview](#overview)
2. [Database Schema Design](#database-schema-design)
3. [Python UDF Design](#python-udf-design)
4. [Stored Procedure Design](#stored-procedure-design)
5. [Snowflake Task Configuration](#snowflake-task-configuration)
6. [Streamlit Application Structure](#streamlit-application-structure)
7. [Error Handling & Monitoring](#error-handling--monitoring)
8. [Security Considerations](#security-considerations)
9. [Testing Strategy](#testing-strategy)
10. [Implementation Sequence](#implementation-sequence)

---

## Overview

### Technical Architecture
- **UI Layer**: Streamlit in Snowflake (asynchronous polling model)
- **Orchestration**: Snowflake Serverless Tasks + Stored Procedures
- **API Integration**: Python UDF with External Network Access
- **Storage**: Transient tables (POC, no Time Travel needed)
- **Scope**: 5 artists × 10,000 fans = 50,000 fan tokens

### Key Design Decisions
1. **Transient Tables**: POC data, minimize storage costs
2. **Batch Processing**: Process fans in batches of 100 to manage API rate limits
3. **Idempotency**: Job status tracking prevents duplicate processing
4. **Error Resilience**: Retry logic for API failures, partial success handling
5. **Security**: External Access Integration for secure API calls

---

## Database Schema Design

### 1. Schema and Role Setup

```sql
-- Create dedicated POC database and schema
CREATE DATABASE IF NOT EXISTS POC_HEAVY_ROTATION;
USE DATABASE POC_HEAVY_ROTATION;

CREATE SCHEMA IF NOT EXISTS POC_SCHEMA;
USE SCHEMA POC_SCHEMA;

-- Create role for POC operations
CREATE ROLE IF NOT EXISTS POC_HEAVY_ROTATION_ROLE;
GRANT USAGE ON DATABASE POC_HEAVY_ROTATION TO ROLE POC_HEAVY_ROTATION_ROLE;
GRANT USAGE ON SCHEMA POC_SCHEMA TO ROLE POC_HEAVY_ROTATION_ROLE;
GRANT CREATE TABLE ON SCHEMA POC_SCHEMA TO ROLE POC_HEAVY_ROTATION_ROLE;
GRANT CREATE PROCEDURE ON SCHEMA POC_SCHEMA TO ROLE POC_HEAVY_ROTATION_ROLE;
GRANT CREATE FUNCTION ON SCHEMA POC_SCHEMA TO ROLE POC_HEAVY_ROTATION_ROLE;
GRANT CREATE TASK ON SCHEMA POC_SCHEMA TO ROLE POC_HEAVY_ROTATION_ROLE;
GRANT CREATE STREAMLIT ON SCHEMA POC_SCHEMA TO ROLE POC_HEAVY_ROTATION_ROLE;

-- Grant to your user
GRANT ROLE POC_HEAVY_ROTATION_ROLE TO USER <YOUR_USERNAME>;
```

### 2. POC_ARTISTS Table

```sql
-- Pre-populated list of 5 artists for POC
CREATE OR REPLACE TRANSIENT TABLE POC_ARTISTS (
    artist_id VARCHAR(50) PRIMARY KEY,
    artist_name VARCHAR(255) NOT NULL,
    spotify_artist_id VARCHAR(50) NOT NULL,
    created_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
    updated_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

-- Insert 5 pre-selected artists
INSERT INTO POC_ARTISTS (artist_id, artist_name, spotify_artist_id) VALUES
    ('artist_001', 'Taylor Swift', '06HL4z0CvFAxyc27GXpf02'),
    ('artist_002', 'Drake', '3TVXtAsR1Inumwj472S9r4'),
    ('artist_003', 'Bad Bunny', '4q3ewBCX7sLwd24euuV69X'),
    ('artist_004', 'The Weeknd', '1Xyo4u8uXC1ZmMpatF05PJ'),
    ('artist_005', 'Ed Sheeran', '6eUKZXaKkcviH0Ku9w2n3V');
```

### 3. POC_FAN_TOKENS Table

```sql
-- Stores fan OAuth tokens (encrypted in production)
CREATE OR REPLACE TRANSIENT TABLE POC_FAN_TOKENS (
    fan_token_id VARCHAR(50) PRIMARY KEY,
    artist_id VARCHAR(50) NOT NULL,
    spotify_access_token VARCHAR(500) NOT NULL, -- In production: use encryption
    spotify_refresh_token VARCHAR(500),
    token_expires_at TIMESTAMP_NTZ,
    fan_spotify_user_id VARCHAR(100),
    created_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
    updated_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
    is_active BOOLEAN DEFAULT TRUE,

    CONSTRAINT fk_artist FOREIGN KEY (artist_id) REFERENCES POC_ARTISTS(artist_id)
);

-- Index for efficient artist-based queries
CREATE INDEX idx_fan_tokens_artist ON POC_FAN_TOKENS(artist_id) WHERE is_active = TRUE;

-- Insert mock fan tokens (10,000 per artist)
-- In production, these would come from OAuth flow
-- For POC, generate placeholder tokens
```

### 4. POC_JOB_STATUS Table

```sql
-- Tracks analysis job execution status
CREATE OR REPLACE TRANSIENT TABLE POC_JOB_STATUS (
    job_id VARCHAR(50) PRIMARY KEY,
    artist_id VARCHAR(50) NOT NULL,
    status VARCHAR(20) NOT NULL, -- PENDING, IN_PROGRESS, COMPLETED, FAILED
    total_fans INTEGER NOT NULL,
    processed_fans INTEGER DEFAULT 0,
    successful_fans INTEGER DEFAULT 0,
    failed_fans INTEGER DEFAULT 0,
    heavy_rotation_count INTEGER DEFAULT 0,
    started_at TIMESTAMP_NTZ,
    completed_at TIMESTAMP_NTZ,
    error_message VARCHAR(5000),
    created_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
    updated_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),

    CONSTRAINT fk_job_artist FOREIGN KEY (artist_id) REFERENCES POC_ARTISTS(artist_id),
    CONSTRAINT chk_status CHECK (status IN ('PENDING', 'IN_PROGRESS', 'COMPLETED', 'FAILED', 'CANCELLED'))
);

-- Index for Streamlit polling queries
CREATE INDEX idx_job_status_lookup ON POC_JOB_STATUS(job_id, status);
CREATE INDEX idx_job_artist_recent ON POC_JOB_STATUS(artist_id, created_at DESC);
```

### 5. POC_RAW_LISTENING_DATA Table

```sql
-- Stores raw top artists data from Spotify API
CREATE OR REPLACE TRANSIENT TABLE POC_RAW_LISTENING_DATA (
    listening_id VARCHAR(50) PRIMARY KEY,
    job_id VARCHAR(50) NOT NULL,
    fan_token_id VARCHAR(50) NOT NULL,
    artist_id VARCHAR(50) NOT NULL,
    time_range VARCHAR(20) NOT NULL, -- short_term, medium_term, long_term
    top_artists_response VARIANT NOT NULL, -- Full JSON response from Spotify Top Artists API
    api_call_successful BOOLEAN DEFAULT TRUE,
    api_error_message VARCHAR(500),
    created_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),

    CONSTRAINT fk_raw_job FOREIGN KEY (job_id) REFERENCES POC_JOB_STATUS(job_id),
    CONSTRAINT fk_raw_fan FOREIGN KEY (fan_token_id) REFERENCES POC_FAN_TOKENS(fan_token_id),
    CONSTRAINT chk_time_range CHECK (time_range IN ('short_term', 'medium_term', 'long_term'))
);

-- Clustering for efficient aggregation queries
ALTER TABLE POC_RAW_LISTENING_DATA CLUSTER BY (job_id, fan_token_id);
```

### 6. POC_AGG_HEAVY_LISTENERS Table

```sql
-- Aggregated results: fans who are "heavy rotation" listeners
-- A fan is "heavy rotation" if the target artist appears in their Top Artists list
CREATE OR REPLACE TRANSIENT TABLE POC_AGG_HEAVY_LISTENERS (
    heavy_listener_id VARCHAR(50) PRIMARY KEY,
    job_id VARCHAR(50) NOT NULL,
    fan_token_id VARCHAR(50) NOT NULL,
    artist_id VARCHAR(50) NOT NULL,
    artist_spotify_id VARCHAR(50) NOT NULL,
    time_range VARCHAR(20) NOT NULL, -- short_term, medium_term, long_term
    artist_rank_in_top INTEGER, -- Position in fan's top artists list (1-50)
    artist_popularity INTEGER, -- Spotify popularity score (0-100)
    artist_genres VARIANT, -- Array of genres from Spotify
    created_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),

    CONSTRAINT fk_agg_job FOREIGN KEY (job_id) REFERENCES POC_JOB_STATUS(job_id),
    CONSTRAINT fk_agg_fan FOREIGN KEY (fan_token_id) REFERENCES POC_FAN_TOKENS(fan_token_id),
    CONSTRAINT chk_time_range_agg CHECK (time_range IN ('short_term', 'medium_term', 'long_term'))
);

-- Index for result retrieval by job
CREATE INDEX idx_heavy_listeners_job ON POC_AGG_HEAVY_LISTENERS(job_id, artist_rank_in_top ASC);
```

### 7. POC_ERROR_LOG Table

```sql
-- Detailed error tracking for debugging
CREATE OR REPLACE TRANSIENT TABLE POC_ERROR_LOG (
    error_id VARCHAR(50) PRIMARY KEY,
    job_id VARCHAR(50),
    fan_token_id VARCHAR(50),
    error_type VARCHAR(100) NOT NULL, -- API_ERROR, TIMEOUT, INVALID_TOKEN, etc.
    error_message VARCHAR(5000),
    error_details VARIANT, -- Full error context
    occurred_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

CREATE INDEX idx_error_log_job ON POC_ERROR_LOG(job_id, occurred_at DESC);
```

### 8. Schema Summary View

```sql
-- Convenience view for monitoring
CREATE OR REPLACE VIEW V_POC_JOB_SUMMARY AS
SELECT
    j.job_id,
    a.artist_name,
    j.status,
    j.total_fans,
    j.processed_fans,
    j.successful_fans,
    j.failed_fans,
    j.heavy_rotation_count,
    ROUND((j.processed_fans::FLOAT / j.total_fans) * 100, 2) AS progress_percentage,
    TIMEDIFF(SECOND, j.started_at, COALESCE(j.completed_at, CURRENT_TIMESTAMP())) AS duration_seconds,
    j.started_at,
    j.completed_at,
    j.error_message
FROM POC_JOB_STATUS j
JOIN POC_ARTISTS a ON j.artist_id = a.artist_id
ORDER BY j.created_at DESC;
```

---

## Python UDF Design

### 1. External Access Integration Setup

```sql
-- Create network rule for Spotify API
CREATE OR REPLACE NETWORK RULE spotify_api_network_rule
    MODE = EGRESS
    TYPE = HOST_PORT
    VALUE_LIST = ('api.spotify.com:443');

-- Create external access integration
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION spotify_api_integration
    ALLOWED_NETWORK_RULES = (spotify_api_network_rule)
    ENABLED = TRUE;

-- Grant usage to role
GRANT USAGE ON INTEGRATION spotify_api_integration TO ROLE POC_HEAVY_ROTATION_ROLE;
```

### 2. Python UDF: Fetch Top Artists

```sql
CREATE OR REPLACE FUNCTION UDF_FETCH_TOP_ARTISTS(
    ACCESS_TOKEN VARCHAR,
    TIME_RANGE VARCHAR,
    LIMIT_PARAM INTEGER
)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.10'
PACKAGES = ('requests', 'snowflake-snowpark-python')
HANDLER = 'fetch_top_artists'
EXTERNAL_ACCESS_INTEGRATIONS = (spotify_api_integration)
AS
$$
import requests
import json
from datetime import datetime

def fetch_top_artists(access_token, time_range, limit_param):
    """
    Fetch top artists from Spotify API.

    Args:
        access_token: Spotify OAuth access token
        time_range: 'short_term' (~4 weeks), 'medium_term' (~6 months), or 'long_term' (~1 year)
        limit_param: Number of artists to fetch (1-50)

    Returns:
        {
            "success": bool,
            "data": [...],  # Array of artist objects
            "error": str or None,
            "total_items": int,
            "time_range": str
        }
    """
    url = "https://api.spotify.com/v1/me/top/artists"
    headers = {
        "Authorization": f"Bearer {access_token}"
    }

    # Validate time_range
    valid_ranges = ['short_term', 'medium_term', 'long_term']
    if time_range not in valid_ranges:
        time_range = 'medium_term'  # Default

    params = {
        "time_range": time_range,
        "limit": min(max(limit_param, 1), 50)  # Clamp between 1-50
    }

    try:
        response = requests.get(url, headers=headers, params=params, timeout=10)

        if response.status_code == 200:
            data = response.json()
            return {
                "success": True,
                "data": data.get("items", []),
                "error": None,
                "total_items": len(data.get("items", [])),
                "time_range": time_range,
                "api_response_code": 200
            }
        elif response.status_code == 401:
            return {
                "success": False,
                "data": None,
                "error": "INVALID_TOKEN",
                "time_range": time_range,
                "api_response_code": 401
            }
        elif response.status_code == 429:
            retry_after = response.headers.get('Retry-After', 60)
            return {
                "success": False,
                "data": None,
                "error": f"RATE_LIMITED:retry_after={retry_after}",
                "time_range": time_range,
                "api_response_code": 429
            }
        else:
            return {
                "success": False,
                "data": None,
                "error": f"API_ERROR:{response.status_code}:{response.text[:200]}",
                "time_range": time_range,
                "api_response_code": response.status_code
            }

    except requests.exceptions.Timeout:
        return {
            "success": False,
            "data": None,
            "error": "TIMEOUT",
            "time_range": time_range,
            "api_response_code": None
        }
    except Exception as e:
        return {
            "success": False,
            "data": None,
            "error": f"EXCEPTION:{str(e)[:200]}",
            "time_range": time_range,
            "api_response_code": None
        }
$$;
```

### 3. Python UDF: Refresh Access Token

```sql
CREATE OR REPLACE FUNCTION UDF_REFRESH_SPOTIFY_TOKEN(
    REFRESH_TOKEN VARCHAR,
    CLIENT_ID VARCHAR,
    CLIENT_SECRET VARCHAR
)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.10'
PACKAGES = ('requests')
HANDLER = 'refresh_token'
EXTERNAL_ACCESS_INTEGRATIONS = (spotify_api_integration)
AS
$$
import requests
import base64

def refresh_token(refresh_token, client_id, client_secret):
    """
    Refresh Spotify access token.

    Returns:
        {
            "success": bool,
            "access_token": str or None,
            "expires_in": int or None,
            "error": str or None
        }
    """
    url = "https://accounts.spotify.com/api/token"

    # Base64 encode client credentials
    auth_str = f"{client_id}:{client_secret}"
    auth_bytes = auth_str.encode('utf-8')
    auth_b64 = base64.b64encode(auth_bytes).decode('utf-8')

    headers = {
        "Authorization": f"Basic {auth_b64}",
        "Content-Type": "application/x-www-form-urlencoded"
    }

    data = {
        "grant_type": "refresh_token",
        "refresh_token": refresh_token
    }

    try:
        response = requests.post(url, headers=headers, data=data, timeout=10)

        if response.status_code == 200:
            token_data = response.json()
            return {
                "success": True,
                "access_token": token_data.get("access_token"),
                "expires_in": token_data.get("expires_in"),
                "error": None
            }
        else:
            return {
                "success": False,
                "access_token": None,
                "expires_in": None,
                "error": f"TOKEN_REFRESH_FAILED:{response.status_code}"
            }

    except Exception as e:
        return {
            "success": False,
            "access_token": None,
            "expires_in": None,
            "error": f"EXCEPTION:{str(e)[:200]}"
        }
$$;
```

---

## Stored Procedure Design

### 1. Main Orchestration Procedure

```sql
CREATE OR REPLACE PROCEDURE SP_ORCHESTRATE_HEAVY_ROTATION_ANALYSIS(
    P_ARTIST_ID VARCHAR,
    P_JOB_ID VARCHAR,
    P_TIME_RANGE VARCHAR
)
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS CALLER
AS
$$
DECLARE
    v_total_fans INTEGER;
    v_batch_size INTEGER := 100;
    v_offset INTEGER := 0;
    v_processed INTEGER := 0;
    v_successful INTEGER := 0;
    v_failed INTEGER := 0;
    v_result VARCHAR;
    v_time_range VARCHAR;
BEGIN
    -- Validate and set time_range
    v_time_range := COALESCE(:P_TIME_RANGE, 'medium_term');
    IF v_time_range NOT IN ('short_term', 'medium_term', 'long_term') THEN
        v_time_range := 'medium_term';
    END IF;

    -- Update job status to IN_PROGRESS
    UPDATE POC_JOB_STATUS
    SET status = 'IN_PROGRESS',
        started_at = CURRENT_TIMESTAMP(),
        updated_at = CURRENT_TIMESTAMP()
    WHERE job_id = :P_JOB_ID;

    -- Get total fans for this artist
    SELECT COUNT(*) INTO v_total_fans
    FROM POC_FAN_TOKENS
    WHERE artist_id = :P_ARTIST_ID
        AND is_active = TRUE;

    -- Process fans in batches
    WHILE v_offset < v_total_fans DO
        -- Call batch processing procedure with time_range
        CALL SP_PROCESS_FAN_BATCH(
            :P_JOB_ID,
            :P_ARTIST_ID,
            :v_time_range,
            :v_batch_size,
            :v_offset
        );

        -- Update progress
        v_offset := v_offset + v_batch_size;
        v_processed := LEAST(v_offset, v_total_fans);

        UPDATE POC_JOB_STATUS
        SET processed_fans = :v_processed,
            updated_at = CURRENT_TIMESTAMP()
        WHERE job_id = :P_JOB_ID;

        -- Add small delay to respect rate limits (optional)
        -- CALL SYSTEM$WAIT(2); -- 2 seconds between batches
    END WHILE;

    -- Aggregate heavy rotation listeners
    CALL SP_AGGREGATE_HEAVY_LISTENERS(:P_JOB_ID, :P_ARTIST_ID);

    -- Get final counts
    SELECT
        COUNT(*) AS total_heavy_listeners
    INTO v_successful
    FROM POC_AGG_HEAVY_LISTENERS
    WHERE job_id = :P_JOB_ID;

    -- Update job status to COMPLETED
    UPDATE POC_JOB_STATUS
    SET status = 'COMPLETED',
        successful_fans = :v_processed,
        heavy_rotation_count = :v_successful,
        completed_at = CURRENT_TIMESTAMP(),
        updated_at = CURRENT_TIMESTAMP()
    WHERE job_id = :P_JOB_ID;

    RETURN 'Job completed successfully: ' || :v_successful || ' heavy rotation listeners found';

EXCEPTION
    WHEN OTHER THEN
        -- Log error and update job status
        UPDATE POC_JOB_STATUS
        SET status = 'FAILED',
            error_message = SQLERRM,
            completed_at = CURRENT_TIMESTAMP(),
            updated_at = CURRENT_TIMESTAMP()
        WHERE job_id = :P_JOB_ID;

        INSERT INTO POC_ERROR_LOG (error_id, job_id, error_type, error_message)
        VALUES (UUID_STRING(), :P_JOB_ID, 'ORCHESTRATION_ERROR', SQLERRM);

        RETURN 'Job failed: ' || SQLERRM;
END;
$$;
```

### 2. Batch Processing Procedure

```sql
CREATE OR REPLACE PROCEDURE SP_PROCESS_FAN_BATCH(
    P_JOB_ID VARCHAR,
    P_ARTIST_ID VARCHAR,
    P_TIME_RANGE VARCHAR,
    P_BATCH_SIZE INTEGER,
    P_OFFSET INTEGER
)
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS CALLER
AS
$$
BEGIN
    -- Insert raw top artists data for batch of fans
    INSERT INTO POC_RAW_LISTENING_DATA (
        listening_id,
        job_id,
        fan_token_id,
        artist_id,
        time_range,
        top_artists_response,
        api_call_successful,
        api_error_message,
        created_at
    )
    WITH fan_batch AS (
        SELECT
            fan_token_id,
            spotify_access_token
        FROM POC_FAN_TOKENS
        WHERE artist_id = :P_ARTIST_ID
            AND is_active = TRUE
        ORDER BY fan_token_id
        LIMIT :P_BATCH_SIZE OFFSET :P_OFFSET
    ),
    api_responses AS (
        SELECT
            f.fan_token_id,
            UDF_FETCH_TOP_ARTISTS(f.spotify_access_token, :P_TIME_RANGE, 50) AS api_response
        FROM fan_batch f
    )
    SELECT
        UUID_STRING() AS listening_id,
        :P_JOB_ID AS job_id,
        r.fan_token_id,
        :P_ARTIST_ID AS artist_id,
        :P_TIME_RANGE AS time_range,
        r.api_response AS top_artists_response,
        r.api_response:success::BOOLEAN AS api_call_successful,
        r.api_response:error::STRING AS api_error_message,
        CURRENT_TIMESTAMP() AS created_at
    FROM api_responses r;

    -- Log any failed API calls
    INSERT INTO POC_ERROR_LOG (error_id, job_id, fan_token_id, error_type, error_message)
    SELECT
        UUID_STRING(),
        :P_JOB_ID,
        r.fan_token_id,
        'API_CALL_FAILED',
        r.api_response:error::STRING
    FROM (
        SELECT
            f.fan_token_id,
            UDF_FETCH_TOP_ARTISTS(f.spotify_access_token, :P_TIME_RANGE, 50) AS api_response
        FROM (
            SELECT
                fan_token_id,
                spotify_access_token
            FROM POC_FAN_TOKENS
            WHERE artist_id = :P_ARTIST_ID
                AND is_active = TRUE
            ORDER BY fan_token_id
            LIMIT :P_BATCH_SIZE OFFSET :P_OFFSET
        ) f
    ) r
    WHERE r.api_response:success = FALSE;

    RETURN 'Batch processed';
END;
$$;
```

### 3. Aggregation Procedure

```sql
CREATE OR REPLACE PROCEDURE SP_AGGREGATE_HEAVY_LISTENERS(
    P_JOB_ID VARCHAR,
    P_ARTIST_ID VARCHAR
)
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS CALLER
AS
$$
BEGIN
    -- Get artist's Spotify ID
    DECLARE v_artist_spotify_id VARCHAR;

    SELECT spotify_artist_id INTO v_artist_spotify_id
    FROM POC_ARTISTS
    WHERE artist_id = :P_ARTIST_ID;

    -- Find fans where target artist appears in their top artists list
    INSERT INTO POC_AGG_HEAVY_LISTENERS (
        heavy_listener_id,
        job_id,
        fan_token_id,
        artist_id,
        artist_spotify_id,
        time_range,
        artist_rank_in_top,
        artist_popularity,
        artist_genres,
        created_at
    )
    WITH flattened_top_artists AS (
        SELECT
            raw.fan_token_id,
            raw.time_range,
            artist.value AS artist_data,
            artist.index + 1 AS artist_rank -- Spotify returns 0-indexed, we want 1-indexed
        FROM POC_RAW_LISTENING_DATA raw,
        LATERAL FLATTEN(input => raw.top_artists_response:data) artist
        WHERE raw.job_id = :P_JOB_ID
            AND raw.api_call_successful = TRUE
    )
    SELECT
        UUID_STRING() AS heavy_listener_id,
        :P_JOB_ID AS job_id,
        fta.fan_token_id,
        :P_ARTIST_ID AS artist_id,
        :v_artist_spotify_id AS artist_spotify_id,
        fta.time_range,
        fta.artist_rank AS artist_rank_in_top,
        fta.artist_data:popularity::INTEGER AS artist_popularity,
        fta.artist_data:genres AS artist_genres,
        CURRENT_TIMESTAMP() AS created_at
    FROM flattened_top_artists fta
    WHERE fta.artist_data:id::STRING = :v_artist_spotify_id;

    RETURN 'Aggregation completed';
END;
$$;
```

### 4. Job Initialization Procedure

```sql
CREATE OR REPLACE PROCEDURE SP_INITIALIZE_ANALYSIS_JOB(
    P_ARTIST_ID VARCHAR
)
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS CALLER
AS
$$
DECLARE
    v_job_id VARCHAR;
    v_total_fans INTEGER;
BEGIN
    -- Check if artist exists
    IF NOT EXISTS (SELECT 1 FROM POC_ARTISTS WHERE artist_id = :P_ARTIST_ID) THEN
        RETURN 'ERROR: Invalid artist_id';
    END IF;

    -- Count active fans
    SELECT COUNT(*) INTO v_total_fans
    FROM POC_FAN_TOKENS
    WHERE artist_id = :P_ARTIST_ID
        AND is_active = TRUE;

    IF v_total_fans = 0 THEN
        RETURN 'ERROR: No active fan tokens for this artist';
    END IF;

    -- Generate job ID
    v_job_id := 'JOB_' || :P_ARTIST_ID || '_' || TO_VARCHAR(CURRENT_TIMESTAMP(), 'YYYYMMDD_HH24MISS');

    -- Create job record
    INSERT INTO POC_JOB_STATUS (
        job_id,
        artist_id,
        status,
        total_fans,
        created_at,
        updated_at
    ) VALUES (
        :v_job_id,
        :P_ARTIST_ID,
        'PENDING',
        :v_total_fans,
        CURRENT_TIMESTAMP(),
        CURRENT_TIMESTAMP()
    );

    RETURN :v_job_id;
END;
$$;
```

---

## Snowflake Task Configuration

### 1. Warehouse for Task Execution

```sql
-- Create dedicated warehouse for background tasks
CREATE OR REPLACE WAREHOUSE POC_TASK_WH WITH
    WAREHOUSE_SIZE = 'SMALL'
    AUTO_SUSPEND = 60
    AUTO_RESUME = TRUE
    INITIALLY_SUSPENDED = TRUE
    COMMENT = 'Warehouse for Heavy Rotation POC background tasks';

GRANT USAGE ON WAREHOUSE POC_TASK_WH TO ROLE POC_HEAVY_ROTATION_ROLE;
```

### 2. Task for Job Processing

```sql
-- Create serverless task for processing jobs
-- This task runs every 1 minute and picks up PENDING jobs

CREATE OR REPLACE TASK TSK_PROCESS_PENDING_JOBS
    WAREHOUSE = POC_TASK_WH
    SCHEDULE = '1 MINUTE'
    COMMENT = 'Process pending heavy rotation analysis jobs'
    WHEN SYSTEM$STREAM_HAS_DATA('POC_JOB_STATUS') -- Optimization: only run when there's data
AS
DECLARE
    v_job_id VARCHAR;
    v_artist_id VARCHAR;
    cur_pending_jobs CURSOR FOR
        SELECT job_id, artist_id
        FROM POC_JOB_STATUS
        WHERE status = 'PENDING'
        ORDER BY created_at ASC
        LIMIT 1; -- Process one job at a time
BEGIN
    OPEN cur_pending_jobs;
    FETCH cur_pending_jobs INTO v_job_id, v_artist_id;

    IF (cur_pending_jobs%FOUND) THEN
        -- Process the job
        CALL SP_ORCHESTRATE_HEAVY_ROTATION_ANALYSIS(:v_artist_id, :v_job_id);
    END IF;

    CLOSE cur_pending_jobs;
END;

-- Start the task
ALTER TASK TSK_PROCESS_PENDING_JOBS RESUME;
```

### 3. Alternative: Direct Task Trigger from Streamlit

For more immediate processing, use dynamic task creation:

```sql
-- Procedure to create and execute a one-time task
CREATE OR REPLACE PROCEDURE SP_TRIGGER_ANALYSIS_TASK(
    P_ARTIST_ID VARCHAR,
    P_JOB_ID VARCHAR,
    P_TIME_RANGE VARCHAR
)
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS CALLER
AS
$$
DECLARE
    v_task_name VARCHAR;
    v_sql VARCHAR;
    v_time_range VARCHAR;
BEGIN
    -- Validate time_range
    v_time_range := COALESCE(:P_TIME_RANGE, 'medium_term');
    IF v_time_range NOT IN ('short_term', 'medium_term', 'long_term') THEN
        v_time_range := 'medium_term';
    END IF;

    -- Create unique task name
    v_task_name := 'TSK_' || :P_JOB_ID;

    -- Create one-time task with time_range parameter
    v_sql := 'CREATE OR REPLACE TASK ' || :v_task_name || '
        WAREHOUSE = POC_TASK_WH
        COMMENT = ''One-time task for job ' || :P_JOB_ID || '''
        AS
        CALL SP_ORCHESTRATE_HEAVY_ROTATION_ANALYSIS(''' || :P_ARTIST_ID || ''', ''' || :P_JOB_ID || ''', ''' || :v_time_range || ''');';

    EXECUTE IMMEDIATE :v_sql;

    -- Start the task
    v_sql := 'ALTER TASK ' || :v_task_name || ' RESUME';
    EXECUTE IMMEDIATE :v_sql;

    -- Execute immediately
    v_sql := 'EXECUTE TASK ' || :v_task_name;
    EXECUTE IMMEDIATE :v_sql;

    RETURN 'Task ' || :v_task_name || ' triggered successfully for time range: ' || :v_time_range;
END;
$$;
```

---

## Streamlit Application Structure

### 1. Streamlit App Code

```python
# heavy_rotation_poc_app.py
import streamlit as st
import pandas as pd
from snowflake.snowpark.context import get_active_session
from snowflake.snowpark.functions import col
import time
from datetime import datetime

# Initialize Snowflake session
session = get_active_session()

# Page configuration
st.set_page_config(
    page_title="Heavy Rotation Listeners POC",
    page_icon="🎵",
    layout="wide"
)

# Title
st.title("🎵 Heavy Rotation Listeners POC")
st.markdown("Identify fans who have the artist in their Spotify Top Artists list")

# Sidebar: Configuration
st.sidebar.header("Configuration")

# Fetch available artists
@st.cache_data(ttl=3600)
def get_artists():
    artists_df = session.table("POC_ARTISTS").select("artist_id", "artist_name").to_pandas()
    return artists_df

artists_df = get_artists()
artist_options = dict(zip(artists_df['ARTIST_NAME'], artists_df['ARTIST_ID']))

selected_artist_name = st.sidebar.selectbox(
    "Select Artist",
    options=list(artist_options.keys())
)
selected_artist_id = artist_options[selected_artist_name]

# Time Range Selection
time_range_options = {
    "Last 4 Weeks (short_term)": "short_term",
    "Last 6 Months (medium_term)": "medium_term",
    "Last Year (long_term)": "long_term"
}
selected_time_range_label = st.sidebar.selectbox(
    "Time Range",
    options=list(time_range_options.keys()),
    index=1  # Default to medium_term
)
selected_time_range = time_range_options[selected_time_range_label]

# Display artist info
st.sidebar.info(f"**Artist ID:** {selected_artist_id}")
st.sidebar.info(f"**Time Range:** {selected_time_range}")

# Get fan count
fan_count_query = f"""
    SELECT COUNT(*) as fan_count
    FROM POC_FAN_TOKENS
    WHERE artist_id = '{selected_artist_id}' AND is_active = TRUE
"""
fan_count = session.sql(fan_count_query).collect()[0]['FAN_COUNT']
st.sidebar.metric("Active Fan Tokens", f"{fan_count:,}")

# Main content area
col1, col2 = st.columns([2, 1])

with col1:
    st.header("Start New Analysis")

    if st.button("🚀 Run Heavy Rotation Analysis", type="primary", use_container_width=True):
        # Initialize job
        job_result = session.call("SP_INITIALIZE_ANALYSIS_JOB", selected_artist_id)

        if job_result.startswith("ERROR"):
            st.error(job_result)
        else:
            job_id = job_result
            st.success(f"Analysis job created: {job_id}")

            # Trigger task with time_range parameter
            trigger_result = session.call("SP_TRIGGER_ANALYSIS_TASK", selected_artist_id, job_id, selected_time_range)
            st.info(trigger_result)

            # Store job_id and time_range in session state for monitoring
            st.session_state['current_job_id'] = job_id
            st.session_state['current_time_range'] = selected_time_range
            st.rerun()

with col2:
    st.header("Quick Stats")

    # Get total jobs run
    total_jobs_query = f"""
        SELECT COUNT(*) as total_jobs
        FROM POC_JOB_STATUS
        WHERE artist_id = '{selected_artist_id}'
    """
    total_jobs = session.sql(total_jobs_query).collect()[0]['TOTAL_JOBS']
    st.metric("Total Jobs Run", total_jobs)

# Job Monitoring Section
st.divider()
st.header("📊 Job Monitoring")

# Check if there's a current job being monitored
if 'current_job_id' in st.session_state:
    job_id = st.session_state['current_job_id']

    # Create placeholder for real-time updates
    status_placeholder = st.empty()
    progress_placeholder = st.empty()
    details_placeholder = st.empty()

    # Poll job status
    while True:
        job_status_query = f"""
            SELECT *
            FROM V_POC_JOB_SUMMARY
            WHERE job_id = '{job_id}'
        """
        job_status_df = session.sql(job_status_query).to_pandas()

        if len(job_status_df) == 0:
            status_placeholder.warning("Job not found")
            break

        job_data = job_status_df.iloc[0]
        status = job_data['STATUS']

        # Display status
        if status == 'IN_PROGRESS':
            status_placeholder.info(f"🔄 **Status:** {status}")

            # Progress bar
            progress = job_data['PROGRESS_PERCENTAGE']
            progress_placeholder.progress(progress / 100.0, text=f"Progress: {progress:.1f}%")

            # Details
            with details_placeholder.container():
                col1, col2, col3, col4 = st.columns(4)
                col1.metric("Total Fans", f"{int(job_data['TOTAL_FANS']):,}")
                col2.metric("Processed", f"{int(job_data['PROCESSED_FANS']):,}")
                col3.metric("Successful", f"{int(job_data['SUCCESSFUL_FANS']):,}")
                col4.metric("Duration", f"{int(job_data['DURATION_SECONDS'])}s")

            # Wait before next poll
            time.sleep(2)

        elif status == 'COMPLETED':
            status_placeholder.success(f"✅ **Status:** {status}")
            progress_placeholder.progress(1.0, text="Progress: 100%")

            # Display final results
            with details_placeholder.container():
                st.subheader("📈 Results")

                col1, col2, col3, col4 = st.columns(4)
                col1.metric("Total Fans", f"{int(job_data['TOTAL_FANS']):,}")
                col2.metric("Processed", f"{int(job_data['PROCESSED_FANS']):,}")
                col3.metric("Heavy Rotation Listeners", f"{int(job_data['HEAVY_ROTATION_COUNT']):,}",
                           delta=f"{(job_data['HEAVY_ROTATION_COUNT']/job_data['TOTAL_FANS']*100):.1f}%")
                col4.metric("Duration", f"{int(job_data['DURATION_SECONDS'])}s")

                # Show detailed results
                st.subheader("🎧 Heavy Rotation Listeners Details")

                heavy_listeners_query = f"""
                    SELECT
                        fan_token_id,
                        time_range,
                        artist_rank_in_top AS rank,
                        artist_popularity AS popularity,
                        artist_genres,
                        created_at
                    FROM POC_AGG_HEAVY_LISTENERS
                    WHERE job_id = '{job_id}'
                    ORDER BY artist_rank_in_top ASC
                    LIMIT 100
                """
                heavy_listeners_df = session.sql(heavy_listeners_query).to_pandas()

                # Display summary metrics
                if len(heavy_listeners_df) > 0:
                    col_a, col_b, col_c = st.columns(3)
                    avg_rank = heavy_listeners_df['RANK'].mean()
                    top_10_count = len(heavy_listeners_df[heavy_listeners_df['RANK'] <= 10])
                    top_25_count = len(heavy_listeners_df[heavy_listeners_df['RANK'] <= 25])

                    col_a.metric("Avg. Rank Position", f"#{int(avg_rank)}")
                    col_b.metric("Fans with Artist in Top 10", top_10_count)
                    col_c.metric("Fans with Artist in Top 25", top_25_count)

                st.dataframe(
                    heavy_listeners_df,
                    use_container_width=True,
                    hide_index=True
                )

                # Download button
                csv = heavy_listeners_df.to_csv(index=False)
                st.download_button(
                    label="📥 Download Results as CSV",
                    data=csv,
                    file_name=f"heavy_rotation_{selected_artist_name}_{selected_time_range}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv",
                    mime="text/csv"
                )

            # Clear current job from session state
            del st.session_state['current_job_id']
            break

        elif status == 'FAILED':
            status_placeholder.error(f"❌ **Status:** {status}")
            details_placeholder.error(f"Error: {job_data['ERROR_MESSAGE']}")
            del st.session_state['current_job_id']
            break

        else:  # PENDING
            status_placeholder.warning(f"⏳ **Status:** {status}")
            details_placeholder.info("Waiting for task to start processing...")
            time.sleep(2)

# Recent Jobs History
st.divider()
st.header("📜 Recent Jobs History")

recent_jobs_query = f"""
    SELECT
        job_id,
        artist_name,
        status,
        total_fans,
        processed_fans,
        heavy_rotation_count,
        progress_percentage,
        duration_seconds,
        started_at,
        completed_at
    FROM V_POC_JOB_SUMMARY
    WHERE artist_name = '{selected_artist_name}'
    ORDER BY started_at DESC
    LIMIT 10
"""
recent_jobs_df = session.sql(recent_jobs_query).to_pandas()

if len(recent_jobs_df) > 0:
    st.dataframe(
        recent_jobs_df,
        use_container_width=True,
        hide_index=True
    )
else:
    st.info("No jobs found for this artist yet. Click 'Run Heavy Rotation Analysis' to start!")

# Footer
st.divider()
st.caption("Heavy Rotation POC - Powered by Snowflake")
```

### 2. Deploy Streamlit App

```sql
-- Create Streamlit app in Snowflake
CREATE OR REPLACE STREAMLIT POC_HEAVY_ROTATION.POC_SCHEMA.HEAVY_ROTATION_APP
    ROOT_LOCATION = '@POC_HEAVY_ROTATION.POC_SCHEMA.STREAMLIT_STAGE'
    MAIN_FILE = 'heavy_rotation_poc_app.py'
    QUERY_WAREHOUSE = POC_TASK_WH
    COMMENT = 'Heavy Rotation Listeners POC Application';

-- Grant access to role
GRANT USAGE ON STREAMLIT HEAVY_ROTATION_APP TO ROLE POC_HEAVY_ROTATION_ROLE;
```

---

## Error Handling & Monitoring

### 1. Error Handling Strategy

**Levels of Error Handling:**

1. **UDF Level**: Return structured error responses instead of raising exceptions
2. **Procedure Level**: Use TRY-CATCH blocks, log to POC_ERROR_LOG
3. **Task Level**: Set TASK_ERROR_INTEGRATION for notifications
4. **Application Level**: Display user-friendly error messages in Streamlit

### 2. Monitoring Queries

```sql
-- Monitor task execution history
SELECT
    name,
    state,
    scheduled_time,
    query_start_time,
    completed_time,
    error_code,
    error_message
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
    SCHEDULED_TIME_RANGE_START => DATEADD('hour', -24, CURRENT_TIMESTAMP()),
    TASK_NAME => 'TSK_PROCESS_PENDING_JOBS'
))
ORDER BY scheduled_time DESC;

-- Monitor API call success rates
SELECT
    DATE_TRUNC('hour', occurred_at) AS error_hour,
    error_type,
    COUNT(*) AS error_count
FROM POC_ERROR_LOG
WHERE occurred_at >= DATEADD('day', -7, CURRENT_TIMESTAMP())
GROUP BY 1, 2
ORDER BY 1 DESC, 3 DESC;

-- Monitor warehouse utilization
SELECT
    warehouse_name,
    AVG(avg_running) AS avg_concurrent_queries,
    AVG(avg_queued_load) AS avg_queued_load,
    SUM(credits_used) AS total_credits
FROM SNOWFLAKE.ACCOUNT_USAGE.WAREHOUSE_LOAD_HISTORY
WHERE start_time >= DATEADD('day', -7, CURRENT_TIMESTAMP())
    AND warehouse_name = 'POC_TASK_WH'
GROUP BY 1;

-- Monitor job completion rates
SELECT
    status,
    COUNT(*) AS job_count,
    AVG(DATEDIFF(SECOND, started_at, completed_at)) AS avg_duration_seconds
FROM POC_JOB_STATUS
WHERE created_at >= DATEADD('day', -7, CURRENT_TIMESTAMP())
GROUP BY 1;
```

### 3. Alerting Setup

```sql
-- Create resource monitor for cost control
CREATE OR REPLACE RESOURCE MONITOR POC_COST_MONITOR WITH
    CREDIT_QUOTA = 100 -- Adjust based on POC budget
    FREQUENCY = MONTHLY
    START_TIMESTAMP = IMMEDIATELY
    TRIGGERS
        ON 80 PERCENT DO NOTIFY
        ON 100 PERCENT DO SUSPEND;

ALTER WAREHOUSE POC_TASK_WH SET RESOURCE_MONITOR = POC_COST_MONITOR;

-- Create notification integration (requires ACCOUNTADMIN)
-- CREATE NOTIFICATION INTEGRATION email_integration
--     TYPE = EMAIL
--     ENABLED = TRUE;
```

---

## Security Considerations

### 1. Token Encryption (Production)

For production deployment, encrypt OAuth tokens:

```sql
-- Create encryption key (requires ACCOUNTADMIN)
-- CREATE OR REPLACE SECRET spotify_token_key
--     TYPE = PASSWORD
--     USERNAME = 'encryption_key'
--     PASSWORD = '<strong_random_key>';

-- Modify table to use encryption
-- ALTER TABLE POC_FAN_TOKENS MODIFY COLUMN spotify_access_token
--     SET MASKING POLICY token_masking_policy;
```

### 2. Row-Level Security

For multi-tenant scenarios:

```sql
CREATE OR REPLACE ROW ACCESS POLICY artist_row_policy AS (artist_id VARCHAR) RETURNS BOOLEAN ->
    CASE
        WHEN CURRENT_ROLE() IN ('SYSADMIN', 'ACCOUNTADMIN') THEN TRUE
        WHEN CURRENT_ROLE() = 'ARTIST_001_ROLE' AND artist_id = 'artist_001' THEN TRUE
        ELSE FALSE
    END;

-- Apply policy
ALTER TABLE POC_FAN_TOKENS ADD ROW ACCESS POLICY artist_row_policy ON (artist_id);
```

### 3. API Credentials Management

Store Spotify API credentials in Snowflake secrets:

```sql
-- Create secret for Spotify credentials
CREATE OR REPLACE SECRET spotify_client_credentials
    TYPE = PASSWORD
    USERNAME = '<spotify_client_id>'
    PASSWORD = '<spotify_client_secret>';

-- Grant usage to role
GRANT USAGE ON SECRET spotify_client_credentials TO ROLE POC_HEAVY_ROTATION_ROLE;

-- Reference in UDF:
-- client_id = _sf_secret_get('spotify_client_credentials', 'username')
-- client_secret = _sf_secret_get('spotify_client_credentials', 'password')
```

### 4. Network Security

```sql
-- Restrict external access to only Spotify domains
CREATE OR REPLACE NETWORK RULE spotify_api_network_rule
    MODE = EGRESS
    TYPE = HOST_PORT
    VALUE_LIST = (
        'api.spotify.com:443',
        'accounts.spotify.com:443'
    );

-- Audit external network calls
SELECT
    query_id,
    query_text,
    user_name,
    start_time,
    execution_status
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE query_text ILIKE '%UDF_FETCH_RECENTLY_PLAYED%'
    AND start_time >= DATEADD('day', -7, CURRENT_TIMESTAMP())
ORDER BY start_time DESC;
```

---

## Testing Strategy

### 1. Unit Testing Phase

**Test Individual Components:**

```sql
-- Test 1: Verify UDF with valid token - short term
SELECT UDF_FETCH_TOP_ARTISTS('<valid_test_token>', 'short_term', 50) AS result;

-- Expected: {"success": true, "data": [...], "total_items": 50, "time_range": "short_term"}

-- Test 2: Verify UDF with valid token - medium term
SELECT UDF_FETCH_TOP_ARTISTS('<valid_test_token>', 'medium_term', 50) AS result;

-- Expected: {"success": true, "data": [...], "total_items": 50, "time_range": "medium_term"}

-- Test 3: Verify UDF with invalid token
SELECT UDF_FETCH_TOP_ARTISTS('invalid_token', 'medium_term', 50) AS result;

-- Expected: {"success": false, "error": "INVALID_TOKEN", "time_range": "medium_term"}

-- Test 3: Test job initialization
CALL SP_INITIALIZE_ANALYSIS_JOB('artist_001');

-- Expected: Returns job_id like 'JOB_artist_001_20250103_143022'

-- Test 4: Verify batch processing with 10 fans
-- Create 10 test fan tokens first
INSERT INTO POC_FAN_TOKENS (fan_token_id, artist_id, spotify_access_token, is_active)
SELECT
    'test_fan_' || SEQ4() AS fan_token_id,
    'artist_001' AS artist_id,
    '<valid_test_token>' AS spotify_access_token,
    TRUE AS is_active
FROM TABLE(GENERATOR(ROWCOUNT => 10));

-- Run batch processing with time_range
CALL SP_PROCESS_FAN_BATCH('TEST_JOB_001', 'artist_001', 'medium_term', 10, 0);

-- Verify results
SELECT COUNT(*) FROM POC_RAW_LISTENING_DATA WHERE job_id = 'TEST_JOB_001';

-- Test 5: Verify aggregation identifies heavy listeners correctly
-- Manually check that target artist appears in top artists list
SELECT
    fan_token_id,
    top_artists_response:data[0]:id::STRING AS top_artist_id,
    top_artists_response:data[0]:name::STRING AS top_artist_name
FROM POC_RAW_LISTENING_DATA
WHERE job_id = 'TEST_JOB_001'
LIMIT 5;
```

### 2. Integration Testing Phase

**Test End-to-End Workflow:**

```sql
-- Integration Test 1: Small-scale full workflow (100 fans)
-- 1. Create test job
SET test_job_id = (SELECT SP_INITIALIZE_ANALYSIS_JOB('artist_001'));

-- 2. Run orchestration with medium_term time range
CALL SP_ORCHESTRATE_HEAVY_ROTATION_ANALYSIS('artist_001', $test_job_id, 'medium_term');

-- 3. Verify job completed
SELECT * FROM V_POC_JOB_SUMMARY WHERE job_id = $test_job_id;

-- 4. Verify heavy rotation listeners identified
SELECT COUNT(*) FROM POC_AGG_HEAVY_LISTENERS WHERE job_id = $test_job_id;

-- 5. Verify rank distribution
SELECT
    CASE
        WHEN artist_rank_in_top <= 10 THEN 'Top 10'
        WHEN artist_rank_in_top <= 25 THEN 'Top 25'
        WHEN artist_rank_in_top <= 50 THEN 'Top 50'
    END AS rank_bucket,
    COUNT(*) AS fan_count
FROM POC_AGG_HEAVY_LISTENERS
WHERE job_id = $test_job_id
GROUP BY 1
ORDER BY 1;

-- Integration Test 2: Test task execution
-- Manually execute task
EXECUTE TASK TSK_PROCESS_PENDING_JOBS;

-- Check task history
SELECT * FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
    TASK_NAME => 'TSK_PROCESS_PENDING_JOBS'
))
ORDER BY scheduled_time DESC
LIMIT 5;
```

### 3. Performance Testing Phase

**Test Scalability:**

```sql
-- Performance Test 1: Measure batch processing time
SET start_time = CURRENT_TIMESTAMP();

CALL SP_PROCESS_FAN_BATCH('PERF_TEST_001', 'artist_001', 'medium_term', 100, 0);

SELECT DATEDIFF(SECOND, $start_time, CURRENT_TIMESTAMP()) AS duration_seconds;

-- Expected: <5 seconds for 100 fans (much faster than before - only 100 API calls vs 5000!)

-- Performance Test 2: Full 10,000 fan analysis
-- Monitor with query profile
ALTER SESSION SET USE_CACHED_RESULT = FALSE;

SET perf_job_id = (SELECT SP_INITIALIZE_ANALYSIS_JOB('artist_001'));
CALL SP_ORCHESTRATE_HEAVY_ROTATION_ANALYSIS('artist_001', $perf_job_id, 'medium_term');

-- Analyze performance
SELECT
    query_id,
    total_elapsed_time / 1000 AS seconds,
    bytes_scanned,
    rows_produced
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE query_text ILIKE '%SP_ORCHESTRATE_HEAVY_ROTATION_ANALYSIS%'
    AND start_time >= DATEADD('hour', -1, CURRENT_TIMESTAMP())
ORDER BY start_time DESC
LIMIT 1;
```

### 4. Error Handling Testing

```sql
-- Error Test 1: Invalid artist_id
CALL SP_INITIALIZE_ANALYSIS_JOB('invalid_artist');
-- Expected: 'ERROR: Invalid artist_id'

-- Error Test 2: API rate limiting simulation
-- Temporarily modify UDF to return rate limit error
-- Verify retry logic and error logging

-- Error Test 3: Failed job recovery
-- Simulate failure mid-processing
UPDATE POC_JOB_STATUS SET status = 'FAILED' WHERE job_id = '<test_job_id>';

-- Verify no orphaned data
SELECT COUNT(*) FROM POC_RAW_LISTENING_DATA WHERE job_id = '<test_job_id>';
```

### 5. Streamlit UI Testing

**Manual Testing Checklist:**

- [ ] App loads without errors
- [ ] Artist dropdown populates correctly
- [ ] Fan count displays accurately
- [ ] "Run Analysis" button triggers job
- [ ] Status updates in real-time during processing
- [ ] Progress bar advances correctly
- [ ] Completed jobs show full results
- [ ] CSV download works
- [ ] Recent jobs history displays
- [ ] Error messages display for failed jobs

---

## Implementation Sequence

### Phase 1: Foundation (Day 1)

**Priority: Setup core infrastructure**

```sql
-- Step 1.1: Create database and schema
-- Execute: Database Schema Design > Section 1

-- Step 1.2: Create base tables
-- Execute: Database Schema Design > Sections 2-7

-- Step 1.3: Insert test data
-- Artists: 5 pre-selected artists
-- Fan tokens: 100 test tokens per artist (500 total for initial testing)

-- Step 1.4: Create warehouse
-- Execute: Snowflake Task Configuration > Section 1

-- Step 1.5: Create external access integration
-- Execute: Python UDF Design > Section 1
```

**Validation:**
```sql
SELECT 'Artists' AS table_name, COUNT(*) AS row_count FROM POC_ARTISTS
UNION ALL
SELECT 'Fan Tokens', COUNT(*) FROM POC_FAN_TOKENS
UNION ALL
SELECT 'Job Status', COUNT(*) FROM POC_JOB_STATUS;
```

### Phase 2: API Integration (Day 2)

**Priority: Build and test API connectivity**

```sql
-- Step 2.1: Create Python UDFs
-- Execute: Python UDF Design > Sections 2-3

-- Step 2.2: Test UDFs with real Spotify tokens
-- Manual testing with valid/invalid tokens
-- Verify error handling for rate limits, timeouts

-- Step 2.3: Test batch API calls
-- Process 10 fans, verify data structure
```

**Validation:**
```sql
-- Test successful API call
SELECT UDF_FETCH_RECENTLY_PLAYED('<valid_token>', 50);

-- Test error handling
SELECT UDF_FETCH_RECENTLY_PLAYED('invalid', 50);
```

### Phase 3: Orchestration Logic (Day 3)

**Priority: Build processing workflow**

```sql
-- Step 3.1: Create stored procedures
-- Execute: Stored Procedure Design > Sections 1-4

-- Step 3.2: Test job initialization
CALL SP_INITIALIZE_ANALYSIS_JOB('artist_001');

-- Step 3.3: Test batch processing with 10 fans
-- Manually call SP_PROCESS_FAN_BATCH

-- Step 3.4: Test aggregation logic
-- Manually call SP_AGGREGATE_HEAVY_LISTENERS

-- Step 3.5: Test full orchestration with 100 fans
CALL SP_ORCHESTRATE_HEAVY_ROTATION_ANALYSIS('artist_001', '<job_id>');
```

**Validation:**
```sql
-- Verify data flow through all tables
SELECT
    (SELECT COUNT(*) FROM POC_JOB_STATUS WHERE status = 'COMPLETED') AS completed_jobs,
    (SELECT COUNT(*) FROM POC_RAW_LISTENING_DATA) AS raw_records,
    (SELECT COUNT(*) FROM POC_AGG_HEAVY_LISTENERS) AS heavy_listeners;
```

### Phase 4: Task Automation (Day 4)

**Priority: Enable background processing**

```sql
-- Step 4.1: Create task
-- Execute: Snowflake Task Configuration > Section 2

-- Step 4.2: Test task execution manually
EXECUTE TASK TSK_PROCESS_PENDING_JOBS;

-- Step 4.3: Monitor task history
-- Check for errors, execution time

-- Step 4.4: Enable scheduled task
ALTER TASK TSK_PROCESS_PENDING_JOBS RESUME;

-- Step 4.5: Test end-to-end async workflow
-- Initialize job, verify task picks it up automatically
```

**Validation:**
```sql
SELECT * FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
    TASK_NAME => 'TSK_PROCESS_PENDING_JOBS'
))
ORDER BY scheduled_time DESC;
```

### Phase 5: Streamlit UI (Day 5)

**Priority: Build user interface**

```sql
-- Step 5.1: Create Streamlit stage and upload app
CREATE OR REPLACE STAGE POC_SCHEMA.STREAMLIT_STAGE;

-- Step 5.2: Upload Python file
PUT file:///path/to/heavy_rotation_poc_app.py @STREAMLIT_STAGE AUTO_COMPRESS=FALSE;

-- Step 5.3: Create Streamlit app
-- Execute: Streamlit Application Structure > Section 2

-- Step 5.4: Test UI functionality
-- Open Streamlit app, test all features
```

**Validation:**
- Manual testing of all UI components
- Test job triggering from UI
- Verify real-time status updates
- Test CSV download

### Phase 6: Scale Testing (Day 6)

**Priority: Validate performance at scale**

```sql
-- Step 6.1: Load full dataset (10,000 fans per artist)
-- Generate or import 50,000 fan tokens

-- Step 6.2: Run full-scale analysis for one artist
-- Monitor execution time, resource usage

-- Step 6.3: Run concurrent analyses for multiple artists
-- Test system under load

-- Step 6.4: Optimize based on performance data
-- Adjust batch sizes, warehouse size if needed
```

**Validation:**
```sql
-- Monitor warehouse utilization
SELECT * FROM SNOWFLAKE.ACCOUNT_USAGE.WAREHOUSE_METERING_HISTORY
WHERE warehouse_name = 'POC_TASK_WH'
    AND start_time >= DATEADD('hour', -24, CURRENT_TIMESTAMP());

-- Check query performance
SELECT
    AVG(total_elapsed_time / 1000) AS avg_seconds,
    MAX(total_elapsed_time / 1000) AS max_seconds
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE query_text ILIKE '%SP_ORCHESTRATE%'
    AND start_time >= DATEADD('day', -1, CURRENT_TIMESTAMP());
```

### Phase 7: Error Handling & Monitoring (Day 7)

**Priority: Production readiness**

```sql
-- Step 7.1: Implement monitoring queries
-- Execute: Error Handling & Monitoring > Section 2

-- Step 7.2: Test error scenarios
-- Invalid tokens, API failures, timeout handling

-- Step 7.3: Create resource monitor
-- Execute: Error Handling & Monitoring > Section 3

-- Step 7.4: Document runbook for common issues
-- Create troubleshooting guide
```

**Validation:**
- Test all error scenarios
- Verify error logging
- Confirm alerts trigger correctly

### Phase 8: Security Hardening (Day 8)

**Priority: Production security**

```sql
-- Step 8.1: Implement token encryption (if required)
-- Execute: Security Considerations > Section 1

-- Step 8.2: Apply row-level security (if multi-tenant)
-- Execute: Security Considerations > Section 2

-- Step 8.3: Configure API credential secrets
-- Execute: Security Considerations > Section 3

-- Step 8.4: Audit permissions
-- Review all GRANT statements
```

**Validation:**
```sql
-- Audit role privileges
SHOW GRANTS TO ROLE POC_HEAVY_ROTATION_ROLE;

-- Verify external access is limited
SHOW NETWORK RULES;
```

### Phase 9: Documentation & Handoff (Day 9-10)

**Priority: Knowledge transfer**

1. **User Documentation:**
   - How to use Streamlit app
   - Interpreting results
   - Troubleshooting common issues

2. **Technical Documentation:**
   - Architecture diagram
   - Data flow documentation
   - API integration details
   - Monitoring procedures

3. **Operational Runbook:**
   - Task management procedures
   - Error resolution steps
   - Performance tuning guide
   - Cost optimization tips

4. **Code Repository:**
   - Version control for all SQL scripts
   - Python UDF code
   - Streamlit app code
   - Test scripts

---

## Cost Optimization Checklist

### Pre-Production Optimization

- [ ] Set warehouse AUTO_SUSPEND = 60 for POC_TASK_WH
- [ ] Use TRANSIENT tables (already configured)
- [ ] Monitor and adjust batch size for optimal API usage
- [ ] Implement result caching for Streamlit queries
- [ ] Set resource monitor with appropriate credit quota

### Ongoing Optimization

```sql
-- Weekly cost review query
SELECT
    DATE_TRUNC('day', start_time) AS usage_date,
    warehouse_name,
    SUM(credits_used) AS daily_credits,
    SUM(credits_used) * <cost_per_credit> AS daily_cost
FROM SNOWFLAKE.ACCOUNT_USAGE.WAREHOUSE_METERING_HISTORY
WHERE start_time >= DATEADD('day', -7, CURRENT_TIMESTAMP())
    AND warehouse_name = 'POC_TASK_WH'
GROUP BY 1, 2
ORDER BY 1 DESC;

-- Identify expensive queries
SELECT
    query_id,
    query_text,
    warehouse_size,
    execution_time / 1000 AS seconds,
    credits_used_cloud_services
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE warehouse_name = 'POC_TASK_WH'
    AND start_time >= DATEADD('day', -7, CURRENT_TIMESTAMP())
ORDER BY credits_used_cloud_services DESC
LIMIT 10;
```

---

## Appendix: Sample Test Data Generator

```sql
-- Generate 10,000 mock fan tokens per artist
CREATE OR REPLACE PROCEDURE SP_GENERATE_TEST_FAN_TOKENS(
    P_ARTIST_ID VARCHAR,
    P_COUNT INTEGER
)
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
BEGIN
    INSERT INTO POC_FAN_TOKENS (
        fan_token_id,
        artist_id,
        spotify_access_token,
        spotify_refresh_token,
        token_expires_at,
        fan_spotify_user_id,
        is_active
    )
    SELECT
        'fan_' || :P_ARTIST_ID || '_' || LPAD(SEQ4(), 6, '0') AS fan_token_id,
        :P_ARTIST_ID AS artist_id,
        'mock_access_token_' || UUID_STRING() AS spotify_access_token,
        'mock_refresh_token_' || UUID_STRING() AS spotify_refresh_token,
        DATEADD('hour', 1, CURRENT_TIMESTAMP()) AS token_expires_at,
        'spotify_user_' || UUID_STRING() AS fan_spotify_user_id,
        TRUE AS is_active
    FROM TABLE(GENERATOR(ROWCOUNT => :P_COUNT));

    RETURN 'Generated ' || :P_COUNT || ' test tokens for artist ' || :P_ARTIST_ID;
END;
$$;

-- Generate tokens for all artists
CALL SP_GENERATE_TEST_FAN_TOKENS('artist_001', 10000);
CALL SP_GENERATE_TEST_FAN_TOKENS('artist_002', 10000);
CALL SP_GENERATE_TEST_FAN_TOKENS('artist_003', 10000);
CALL SP_GENERATE_TEST_FAN_TOKENS('artist_004', 10000);
CALL SP_GENERATE_TEST_FAN_TOKENS('artist_005', 10000);
```

---

## Success Criteria

### Technical Success Metrics

1. **Functionality:**
   - [ ] Successfully processes 10,000 fans per artist
   - [ ] Correctly identifies heavy rotation listeners (>10 plays)
   - [ ] Streamlit UI provides real-time status updates
   - [ ] Background tasks execute reliably

2. **Performance:**
   - [ ] Processes 10,000 fans in <15 minutes
   - [ ] Streamlit app loads in <3 seconds
   - [ ] Status polling response time <1 second

3. **Reliability:**
   - [ ] 95%+ API call success rate
   - [ ] Graceful handling of API failures
   - [ ] No data loss on partial failures

4. **Cost:**
   - [ ] Total POC cost under $X (set appropriate budget)
   - [ ] Warehouse auto-suspend prevents idle costs
   - [ ] Efficient query execution (minimal scanning)

### Business Success Metrics

1. **User Experience:**
   - [ ] Intuitive UI requiring no training
   - [ ] Clear status feedback during processing
   - [ ] Actionable results presentation

2. **Data Quality:**
   - [ ] Accurate play count aggregation
   - [ ] No duplicate listener records
   - [ ] Valid timestamp ranges

3. **Scalability:**
   - [ ] Architecture supports 100K+ fans per artist
   - [ ] Can add more artists without code changes
   - [ ] Performance degrades linearly with scale

---

## Next Steps After POC

### If POC is Successful

1. **Production Architecture:**
   - Implement proper OAuth token management
   - Add encryption for sensitive data
   - Set up production monitoring and alerting
   - Scale to full artist roster

2. **Feature Enhancements:**
   - Historical trend analysis
   - Fan segmentation by engagement level
   - Integration with marketing platforms
   - Automated reporting

3. **Operational Excellence:**
   - SLA definition and monitoring
   - Incident response procedures
   - Regular performance reviews
   - Cost optimization continuous improvement

### If POC Needs Iteration

1. **Performance Issues:**
   - Increase warehouse size
   - Optimize batch processing
   - Implement caching strategies
   - Consider Query Acceleration Service

2. **API Rate Limiting:**
   - Adjust batch sizes
   - Implement exponential backoff
   - Add delays between batches
   - Consider multi-warehouse approach

3. **Cost Concerns:**
   - Reduce processing frequency
   - Optimize query patterns
   - Implement incremental processing
   - Use smaller warehouse sizes

---

## Conclusion

This implementation plan provides a comprehensive, step-by-step guide to building the Heavy Rotation Listeners POC entirely within Snowflake. The architecture leverages Snowflake's native capabilities for orchestration, processing, and UI, while following best practices for security, performance, and cost optimization.

**Key Advantages of This Approach:**

1. **Simplicity:** Single-platform solution reduces complexity
2. **Security:** External API calls managed through Snowflake's secure framework
3. **Scalability:** Snowflake's elastic compute supports growth
4. **Cost-Effectiveness:** Pay-per-use model with auto-suspend
5. **Maintainability:** SQL and Python code is version-controlled and testable

**Estimated Timeline:** 10 days from start to production-ready POC

**Estimated Cost:** $100-500 depending on warehouse usage and data volume

Follow this plan sequentially, validate each phase before proceeding, and adjust based on your specific requirements and constraints.