Data Warehouse Integration

-- Events table
CREATE TABLE RAW.BEAM_EVENTS (
    id INTEGER,
    event_type VARCHAR(100),
    event_data VARIANT,
    tandem_tracking VARIANT,
    attribution_sources ARRAY,
    -- Attribution columns (NEW)
    attribution_source VARCHAR(50),
    attribution_type VARCHAR(30),
    attribution_priority INTEGER,
    -- UTM columns (NEW)
    utm_source VARCHAR(100),
    utm_medium VARCHAR(100),
    utm_campaign VARCHAR(255),
    created_at TIMESTAMP_NTZ,
    updated_at TIMESTAMP_NTZ,
    _loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

-- Sessions table
CREATE TABLE RAW.BEAM_SESSIONS (
    id VARCHAR(50),
    identity_key_hash VARCHAR(64),
    first_seen TIMESTAMP_NTZ,
    last_seen TIMESTAMP_NTZ,
    event_count INTEGER,
    page_view_count INTEGER,
    conversion_count INTEGER,
    total_value DECIMAL(12,2),
    _loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

-- Platform logs table
CREATE TABLE RAW.BEAM_PLATFORM_LOGS (
    id VARCHAR(50),
    platform_id VARCHAR(50),
    platform_slug VARCHAR(50),
    advertiser_platform_id VARCHAR(50),
    tracking_event_id INTEGER,
    event_name VARCHAR(100),
    vendor_id_used VARCHAR(255),
    attribution_source VARCHAR(50),
    status_code INTEGER,
    sent_at TIMESTAMP_NTZ,
    is_success BOOLEAN,
    _loaded_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

-- Create Python UDF for API calls (Snowpark)
CREATE OR REPLACE FUNCTION RAW.FETCH_BEAM_EXPORT(
    endpoint VARCHAR,
    from_date DATE,
    to_date DATE,
    cursor_val VARCHAR
)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('requests')
HANDLER = 'fetch_data'
AS
$$
import requests
import json

def fetch_data(endpoint, from_date, to_date, cursor_val):
    api_token = 'YOUR_API_TOKEN'  -- Store in Snowflake secrets
    advertiser_id = 'YOUR_ADVERTISER_UUID'

    base_url = 'https://api.tandembeam.com/api/export'
    url = f"{base_url}/{endpoint}"

    params = {
        'from': str(from_date),
        'to': str(to_date),
        'limit': 10000,
        'format': 'json'
    }

    if cursor_val:
        params['cursor'] = cursor_val

    headers = {
        'Authorization': f'Bearer {api_token}',
        'X-Advertiser-ID': advertiser_id
    }

    response = requests.get(url, params=params, headers=headers)
    return response.json()
$$;

-- Stored procedure for incremental sync
CREATE OR REPLACE PROCEDURE RAW.SYNC_BEAM_EVENTS(from_date DATE, to_date DATE)
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
    cursor_val VARCHAR DEFAULT NULL;
    has_more BOOLEAN DEFAULT TRUE;
    batch_count INTEGER DEFAULT 0;
BEGIN
    -- Loop through all pages
    WHILE (has_more) DO
        -- Fetch batch from API
        LET result VARIANT := (SELECT RAW.FETCH_BEAM_EXPORT('events', :from_date, :to_date, :cursor_val));

        -- Insert data into table
        INSERT INTO RAW.BEAM_EVENTS (id, event_type, event_data, tandem_tracking, attribution_sources, created_at, updated_at)
        SELECT
            value:id::INTEGER,
            value:event_type::VARCHAR,
            value:event_data::VARIANT,
            value:tandem_tracking::VARIANT,
            value:attribution_sources::ARRAY,
            value:created_at::TIMESTAMP_NTZ,
            value:updated_at::TIMESTAMP_NTZ
        FROM TABLE(FLATTEN(input => :result:data));

        -- Update cursor and has_more
        cursor_val := result:meta:cursor::VARCHAR;
        has_more := result:meta:has_more::BOOLEAN;
        batch_count := batch_count + 1;
    END WHILE;

    RETURN 'Synced ' || batch_count || ' batches successfully';
END;
$$;

-- Create hourly sync task
CREATE OR REPLACE TASK RAW.BEAM_HOURLY_SYNC
    WAREHOUSE = COMPUTE_WH
    SCHEDULE = 'USING CRON 0 * * * * UTC'
AS
    CALL RAW.SYNC_BEAM_EVENTS(DATEADD('day', -1, CURRENT_DATE()), CURRENT_DATE());

-- Enable the task
ALTER TASK RAW.BEAM_HOURLY_SYNC RESUME;

-- Verify task is running
SHOW TASKS LIKE 'BEAM%';