-- ============================================================================
-- STORYINTEL: ClickHouse Analytics Schema
-- ============================================================================
-- Purpose: Heavy-read analytics, time-series metrics, trend detection
--
-- DEV MODE vs PRODUCTION MODE
-- ============================================================================
-- This schema is ONLY used when analytics.mode = 'clickhouse' in system_config.
--
-- In DEV MODE (analytics.mode = 'd1'):
-- - All analytics tables exist in D1 (see database-schema.sql)
-- - No ClickHouse connection required
-- - Suitable for development and low-volume deployments
--
-- In PRODUCTION MODE (analytics.mode = 'clickhouse'):
-- - High-volume data streams to ClickHouse via analytics.stream queue
-- - D1 retains recent data for real-time queries (configurable retention)
-- - ClickHouse handles historical queries and dashboards
--
-- TO ENABLE CLICKHOUSE:
-- 1. Set up ClickHouse Cloud or self-hosted instance
-- 2. Run this schema to create tables
-- 3. Add secrets: wrangler secret put CLICKHOUSE_HOST / CLICKHOUSE_PASSWORD
-- 4. Update system_config: analytics.mode = 'clickhouse', clickhouse.enabled = 'true'
-- 5. Deploy the analytics-consumer worker
-- ============================================================================
--
-- WHY CLICKHOUSE?
-- D1 (SQLite) is great for transactional writes but will choke on:
-- - Time-series aggregations (article counts over time)
-- - Customer analytics (usage patterns, engagement trends)
-- - Pipeline telemetry (latency percentiles, error rates)
-- - Trend detection (keyword velocity, entity mentions)
--
-- ARCHITECTURE:
-- D1 (Source of Truth) ──► Cloudflare Queue ──► ClickHouse (Analytics)
--
-- Data flows from D1 to ClickHouse via queue consumers. D1 remains the
-- source of truth; ClickHouse is a read-optimized analytics layer.
-- ============================================================================

-- ============================================================================
-- ARTICLE METRICS
-- ============================================================================
-- Track article-level events over time: views, clicks, saves, shares
-- Enables: engagement trending, popular articles, customer behavior analysis
-- ============================================================================

CREATE TABLE IF NOT EXISTS article_metrics
(
    -- Timestamp (partition key)
    timestamp DateTime DEFAULT now(),

    -- Identifiers
    article_id String,
    source_id String,
    customer_id String,
    profile_id Nullable(String),

    -- Event details
    event_type Enum8(
        'view' = 1,
        'click' = 2,
        'save' = 3,
        'share' = 4,
        'dismiss' = 5,
        'dwell' = 6,
        'export' = 7
    ),

    -- Optional metrics
    dwell_time_ms Nullable(UInt32),
    scroll_depth Nullable(Float32),

    -- Context
    source Enum8('web' = 1, 'mobile' = 2, 'api' = 3, 'email' = 4),
    session_id Nullable(String),

    -- Metadata (JSON for flexibility)
    metadata String DEFAULT '{}'
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (timestamp, article_id, customer_id)
TTL timestamp + INTERVAL 1 YEAR
SETTINGS index_granularity = 8192;

-- Materialized view: Daily article engagement rollup
CREATE MATERIALIZED VIEW IF NOT EXISTS article_engagement_daily
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, article_id)
AS SELECT
    toDate(timestamp) AS date,
    article_id,
    source_id,
    countIf(event_type = 'view') AS views,
    countIf(event_type = 'click') AS clicks,
    countIf(event_type = 'save') AS saves,
    countIf(event_type = 'share') AS shares,
    countIf(event_type = 'dismiss') AS dismisses,
    avgIf(dwell_time_ms, dwell_time_ms IS NOT NULL) AS avg_dwell_ms,
    uniq(customer_id) AS unique_customers
FROM article_metrics
GROUP BY date, article_id, source_id;


-- ============================================================================
-- PIPELINE METRICS
-- ============================================================================
-- Track pipeline stage performance: latency, success rates, costs
-- Enables: SLI/SLO monitoring, bottleneck detection, cost analysis
-- ============================================================================

CREATE TABLE IF NOT EXISTS pipeline_metrics
(
    -- Timestamp
    timestamp DateTime DEFAULT now(),

    -- Pipeline context
    stage Enum8(
        'crawl' = 1,
        'dedup' = 2,
        'extract' = 3,
        'enrich' = 4,
        'embed' = 5,
        'classify' = 6,
        'cluster' = 7,
        'match' = 8,
        'notify' = 9
    ),

    -- Operation details
    operation String,              -- e.g., 'fetch_html', 'extract_entities'
    article_id Nullable(String),
    keyword_id Nullable(String),

    -- Performance
    duration_ms UInt32,
    queue_wait_ms Nullable(UInt32),

    -- Status
    success UInt8,
    error_type Nullable(String),
    error_message Nullable(String),

    -- Cost tracking (in microdollars)
    cost_micros UInt32 DEFAULT 0,

    -- Resource usage
    tokens_in Nullable(UInt32),
    tokens_out Nullable(UInt32),

    -- Context
    worker_id Nullable(String),
    workflow_id Nullable(String),

    -- Metadata
    metadata String DEFAULT '{}'
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (timestamp, stage, operation)
TTL timestamp + INTERVAL 90 DAY
SETTINGS index_granularity = 8192;

-- Materialized view: Hourly pipeline health
CREATE MATERIALIZED VIEW IF NOT EXISTS pipeline_health_hourly
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (hour, stage, operation)
AS SELECT
    toStartOfHour(timestamp) AS hour,
    stage,
    operation,
    count() AS total_ops,
    sum(success) AS success_count,
    count() - sum(success) AS error_count,
    avg(duration_ms) AS avg_duration_ms,
    quantile(0.5)(duration_ms) AS p50_duration_ms,
    quantile(0.95)(duration_ms) AS p95_duration_ms,
    quantile(0.99)(duration_ms) AS p99_duration_ms,
    sum(cost_micros) AS total_cost_micros
FROM pipeline_metrics
GROUP BY hour, stage, operation;


-- ============================================================================
-- KEYWORD VELOCITY
-- ============================================================================
-- Track keyword appearance velocity for trend detection
-- Enables: Breaking news detection, topic emergence, keyword health
-- ============================================================================

CREATE TABLE IF NOT EXISTS keyword_velocity
(
    -- Time bucket (hourly)
    hour DateTime,

    -- Keyword identification
    keyword_hash String,
    keyword String,

    -- Counts
    article_count UInt32,
    source_count UInt32,

    -- Aggregated metrics
    avg_sentiment Float32 DEFAULT 0,
    total_engagement UInt64 DEFAULT 0,

    -- Velocity calculations (computed on insert)
    velocity Float32 DEFAULT 0,           -- Rate of change vs previous hour
    acceleration Float32 DEFAULT 0        -- Rate of change of velocity
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (hour, keyword_hash)
TTL hour + INTERVAL 180 DAY
SETTINGS index_granularity = 8192;

-- Materialized view: Daily keyword summary
CREATE MATERIALIZED VIEW IF NOT EXISTS keyword_daily_summary
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, keyword_hash)
AS SELECT
    toDate(hour) AS date,
    keyword_hash,
    any(keyword) AS keyword,
    sum(article_count) AS total_articles,
    max(source_count) AS max_sources,
    avg(avg_sentiment) AS avg_sentiment,
    sum(total_engagement) AS total_engagement,
    max(velocity) AS peak_velocity
FROM keyword_velocity
GROUP BY date, keyword_hash;


-- ============================================================================
-- CUSTOMER ANALYTICS
-- ============================================================================
-- Track customer usage patterns and engagement
-- Enables: Cohort analysis, churn prediction, feature usage
-- ============================================================================

CREATE TABLE IF NOT EXISTS customer_analytics
(
    -- Timestamp
    timestamp DateTime DEFAULT now(),

    -- Customer identification
    customer_id String,
    tier_id String,

    -- Event type
    event_type Enum8(
        'api_call' = 1,
        'search' = 2,
        'feed_view' = 3,
        'article_view' = 4,
        'export' = 5,
        'profile_create' = 6,
        'keyword_add' = 7,
        'notification_sent' = 8,
        'login' = 9,
        'logout' = 10
    ),

    -- Event details
    endpoint Nullable(String),
    response_time_ms Nullable(UInt32),
    status_code Nullable(UInt16),

    -- Usage metrics
    articles_returned Nullable(UInt32),
    bytes_transferred Nullable(UInt64),

    -- Context
    source Enum8('web' = 1, 'mobile' = 2, 'api' = 3, 'sdk' = 4),

    -- Metadata
    metadata String DEFAULT '{}'
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (timestamp, customer_id, event_type)
TTL timestamp + INTERVAL 1 YEAR
SETTINGS index_granularity = 8192;

-- Materialized view: Daily customer activity
CREATE MATERIALIZED VIEW IF NOT EXISTS customer_activity_daily
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, customer_id)
AS SELECT
    toDate(timestamp) AS date,
    customer_id,
    tier_id,
    count() AS total_events,
    countIf(event_type = 'api_call') AS api_calls,
    countIf(event_type = 'search') AS searches,
    countIf(event_type = 'article_view') AS article_views,
    countIf(event_type = 'export') AS exports,
    avg(response_time_ms) AS avg_response_ms,
    sum(bytes_transferred) AS total_bytes
FROM customer_analytics
GROUP BY date, customer_id, tier_id;


-- ============================================================================
-- COST ANALYTICS
-- ============================================================================
-- Detailed cost tracking for optimization and billing
-- Enables: Cost attribution, budget forecasting, anomaly detection
-- ============================================================================

CREATE TABLE IF NOT EXISTS cost_analytics
(
    -- Timestamp
    timestamp DateTime DEFAULT now(),

    -- Cost attribution
    service Enum8(
        'zenrows' = 1,
        'dataforseo' = 2,
        'sharedcount' = 3,
        'workers_ai' = 4,
        'openai' = 5,
        'vectorize' = 6,
        'r2' = 7,
        'd1' = 8
    ),
    operation String,

    -- Attribution
    customer_id Nullable(String),
    article_id Nullable(String),
    keyword_id Nullable(String),
    project_id Nullable(String),

    -- Cost details (in microdollars)
    cost_micros UInt32,

    -- Usage details
    input_units UInt32 DEFAULT 0,     -- tokens, requests, bytes
    output_units UInt32 DEFAULT 0,

    -- Performance
    latency_ms Nullable(UInt32),
    success UInt8 DEFAULT 1,

    -- Metadata
    metadata String DEFAULT '{}'
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (timestamp, service, operation)
TTL timestamp + INTERVAL 1 YEAR
SETTINGS index_granularity = 8192;

-- Materialized view: Hourly cost rollup
CREATE MATERIALIZED VIEW IF NOT EXISTS cost_rollup_hourly
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (hour, service, operation)
AS SELECT
    toStartOfHour(timestamp) AS hour,
    service,
    operation,
    sum(cost_micros) AS total_micros,
    count() AS operation_count,
    sum(input_units) AS total_input_units,
    sum(output_units) AS total_output_units,
    avg(latency_ms) AS avg_latency_ms,
    sum(success) AS success_count,
    count() - sum(success) AS error_count
FROM cost_analytics
GROUP BY hour, service, operation;

-- Materialized view: Daily cost by customer
CREATE MATERIALIZED VIEW IF NOT EXISTS cost_by_customer_daily
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, customer_id, service)
AS SELECT
    toDate(timestamp) AS date,
    customer_id,
    service,
    sum(cost_micros) AS total_micros,
    count() AS operation_count
FROM cost_analytics
WHERE customer_id IS NOT NULL
GROUP BY date, customer_id, service;


-- ============================================================================
-- ENTITY VELOCITY
-- ============================================================================
-- Track entity (person, org) mention velocity for trend detection
-- Enables: Entity trending, breaking news about entities
-- ============================================================================

CREATE TABLE IF NOT EXISTS entity_velocity
(
    -- Time bucket
    hour DateTime,

    -- Entity identification
    entity_id String,
    entity_name String,
    entity_type Enum8(
        'person' = 1,
        'organization' = 2,
        'location' = 3,
        'product' = 4,
        'event' = 5
    ),

    -- Counts
    mention_count UInt32,
    article_count UInt32,
    source_count UInt32,

    -- Sentiment
    avg_sentiment Float32 DEFAULT 0,
    sentiment_variance Float32 DEFAULT 0,

    -- Velocity
    velocity Float32 DEFAULT 0
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (hour, entity_id)
TTL hour + INTERVAL 180 DAY
SETTINGS index_granularity = 8192;


-- ============================================================================
-- SOURCE HEALTH
-- ============================================================================
-- Track source crawl health over time
-- Enables: Source reliability monitoring, anti-bot detection
-- ============================================================================

CREATE TABLE IF NOT EXISTS source_health
(
    -- Time bucket
    hour DateTime,

    -- Source identification
    source_id String,
    domain String,

    -- Crawl metrics
    crawl_attempts UInt32,
    success_count UInt32,
    blocked_count UInt32,
    timeout_count UInt32,
    error_count UInt32,

    -- Fetch method distribution
    direct_count UInt32 DEFAULT 0,
    zenrows_count UInt32 DEFAULT 0,
    dataforseo_count UInt32 DEFAULT 0,

    -- Performance
    avg_latency_ms Float32,
    p95_latency_ms Float32,

    -- Articles
    articles_found UInt32,
    new_articles UInt32
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (hour, source_id)
TTL hour + INTERVAL 90 DAY
SETTINGS index_granularity = 8192;


-- ============================================================================
-- TREND SIGNALS
-- ============================================================================
-- Detected trends for alerting and analysis
-- Enables: Trend notifications, breaking news detection
-- ============================================================================

CREATE TABLE IF NOT EXISTS trend_signals
(
    -- Timestamp
    detected_at DateTime DEFAULT now(),

    -- Trend identification
    signal_id String,
    signal_type Enum8(
        'keyword_spike' = 1,
        'entity_surge' = 2,
        'topic_emergence' = 3,
        'sentiment_shift' = 4,
        'source_spike' = 5
    ),

    -- Target
    target_type String,           -- 'keyword', 'entity', 'topic'
    target_id String,
    target_label String,

    -- Metrics
    baseline_value Float64,
    current_value Float64,
    delta_percent Float64,
    velocity Float64,

    -- Context
    window_hours UInt8,
    article_count UInt32,
    source_count UInt32,

    -- Confidence
    confidence Float32,
    is_confirmed UInt8 DEFAULT 0,

    -- Sample articles (JSON array of IDs)
    sample_article_ids String DEFAULT '[]'
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(detected_at)
ORDER BY (detected_at, signal_type, target_id)
TTL detected_at + INTERVAL 1 YEAR
SETTINGS index_granularity = 8192;


-- ============================================================================
-- USEFUL QUERIES
-- ============================================================================

-- Top trending keywords (last 24 hours)
-- SELECT
--     keyword,
--     sum(article_count) AS articles,
--     max(velocity) AS peak_velocity,
--     avg(avg_sentiment) AS sentiment
-- FROM keyword_velocity
-- WHERE hour >= now() - INTERVAL 24 HOUR
-- GROUP BY keyword
-- ORDER BY peak_velocity DESC
-- LIMIT 20;

-- Pipeline health dashboard
-- SELECT
--     stage,
--     operation,
--     sum(total_ops) AS ops,
--     sum(success_count) AS success,
--     sum(error_count) AS errors,
--     round(100.0 * sum(success_count) / sum(total_ops), 2) AS success_rate,
--     round(avg(avg_duration_ms), 0) AS avg_ms,
--     round(avg(p95_duration_ms), 0) AS p95_ms
-- FROM pipeline_health_hourly
-- WHERE hour >= now() - INTERVAL 24 HOUR
-- GROUP BY stage, operation
-- ORDER BY stage, operation;

-- Daily cost breakdown
-- SELECT
--     date,
--     service,
--     sum(total_micros) / 1000000.0 AS cost_usd,
--     sum(operation_count) AS ops
-- FROM cost_rollup_hourly
-- WHERE hour >= now() - INTERVAL 7 DAY
-- GROUP BY date, service
-- ORDER BY date DESC, cost_usd DESC;

-- Customer engagement cohorts
-- SELECT
--     customer_id,
--     tier_id,
--     sum(api_calls) AS total_api_calls,
--     sum(article_views) AS total_views,
--     count(DISTINCT date) AS active_days
-- FROM customer_activity_daily
-- WHERE date >= now() - INTERVAL 30 DAY
-- GROUP BY customer_id, tier_id
-- ORDER BY total_api_calls DESC;


-- ============================================================================
-- NOTES
-- ============================================================================
--
-- DATA FLOW:
-- 1. D1 writes happen in Workers (source of truth)
-- 2. Queue consumer batches events and writes to ClickHouse
-- 3. Materialized views auto-aggregate for dashboard queries
--
-- RETENTION:
-- - Detailed events: 90 days to 1 year (configurable per table)
-- - Aggregated views: Indefinite (small data volume)
--
-- RECOMMENDED CLICKHOUSE SETUP:
-- - ClickHouse Cloud (managed) for simplicity
-- - Or self-hosted with replication for cost savings
-- - Connect via HTTPS with API key auth
--
-- COST OPTIMIZATION:
-- - Use TTL to auto-delete old data
-- - Partition by month for efficient pruning
-- - Use SummingMergeTree for aggregates (no re-computation)
--
-- ============================================================================
