Real-Time Analytics for Video Streaming Platform.

# Event type hierarchy

EVENT_TYPES = {
    # Playback events (highest volume)
    "playback.started": "User started playing content",
    "playback.paused": "User paused playback",
    "playback.resumed": "User resumed from pause",
    "playback.seeked": "User jumped to different position",
    "playback.completed": "User finished content (>90% watched)",
    "playback.abandoned": "User stopped before completion",
    "playback.heartbeat": "Periodic signal during playback",
    
    # Quality events
    "quality.buffering_started": "Playback buffering began",
    "quality.buffering_ended": "Playback resumed after buffer",
    "quality.resolution_changed": "Video quality changed",
    "quality.error": "Playback error occurred",
    
    # Browse events
    "browse.search": "User performed search",
    "browse.page_viewed": "User viewed a page/row",
    "browse.content_clicked": "User clicked on content tile",
    "browse.content_added_list": "User added to My List",
    
    # Session events
    "session.started": "App opened / session began",
    "session.ended": "App closed / session ended"
}

# Schema registry for safe evolution

class ViewingEventSchema:
    """
    Schema evolution rules:
    - SAFE: Add optional fields
    - UNSAFE: Remove fields, change types, rename
    """
    
    VERSION = "2.1"
    
    # v1.0: Original schema
    # v2.0: Added experiments field
    # v2.1: Added subtitle_language to payload
    
    REQUIRED_FIELDS = [
        "event_id", "event_type", "event_time",
        "user_id", "device_id", "payload"
    ]
    
    OPTIONAL_FIELDS = [
        "session_id", "geo", "experiments",
        "received_time", "app_version"
    ]

# Ingestion service optimized for throughput

from dataclasses import dataclass
from typing import List, Dict
import asyncio
from datetime import datetime, timezone


@dataclass
class IngestionConfig:
    batch_size: int = 100
    batch_timeout_ms: int = 100
    max_concurrent_batches: int = 50
    kafka_linger_ms: int = 5
    kafka_batch_size: int = 16384


class HighThroughputCollector:
    """
    Event collector sized for ~10K events/second per instance
    (20 instances × 5 regions ≈ 1M events/second at peak).
    
    Key optimizations:
    1. Async everywhere - no blocking
    2. Batch events in the Kafka producer (linger_ms, batch_size)
    3. Schema validation with compiled validators
    4. Connection pooling for Kafka
    """
    
    def __init__(self, config: IngestionConfig, kafka_producer, validator):
        self.config = config
        self.producer = kafka_producer
        self.validator = validator
        
        # Metrics
        self.events_received = 0
        self.events_validated = 0
        self.events_produced = 0
        self.validation_errors = 0
    
    async def ingest(self, events: List[Dict]) -> Dict:
        """
        Ingest a batch of events from client.
        
        Acknowledges only after Kafka has the events (acks=all).
        Acking before the produce completes would lose buffered events
        if this collector crashed, breaking the "no event loss"
        requirement; the client would never retry them.
        """
        
        results = {"accepted": 0, "rejected": 0, "errors": []}
        valid_events = []
        
        for event in events:
            self.events_received += 1
            
            # Fast validation (compiled schema)
            validation_result = self.validator.validate(event)
            
            if not validation_result.is_valid:
                results["rejected"] += 1
                results["errors"].append({
                    "event_id": event.get("event_id"),
                    "error": validation_result.error
                })
                self.validation_errors += 1
                continue
            
            # Enrich event
            valid_events.append(self._enrich(event))
            self.events_validated += 1
        
        # Produce and wait for broker acks before acknowledging the client
        # (the Kafka producer still batches internally via linger_ms)
        await self._produce_batch(valid_events)
        results["accepted"] = len(valid_events)
        
        return results
    
    def _enrich(self, event: Dict) -> Dict:
        """Add server-side enrichments."""
        
        event["received_time"] = datetime.now(timezone.utc).isoformat()
        event["collector_id"] = self.collector_id
        event["processing_region"] = self.region
        
        # Add derived fields
        if "geo" in event and "country" in event["geo"]:
            event["geo"]["continent"] = self._country_to_continent(
                event["geo"]["country"]
            )
        
        return event
    
    async def _produce_batch(self, batch: List[Dict]):
        """Produce batch to Kafka and wait for the broker acks."""
        
        futures = []
        for event in batch:
            topic = self._get_topic(event["event_type"])
            # Partition by user_id for ordering
            partition_key = event.get("user_id", event["event_id"])
            
            # send() enqueues and returns a delivery future
            futures.append(await self.producer.send(
                topic=topic,
                key=partition_key,
                value=event
            ))
        
        # Raises if any send failed, so the client gets an error and retries
        await asyncio.gather(*futures)
        self.events_produced += len(batch)
    
    def _get_topic(self, event_type: str) -> str:
        """Route event to appropriate Kafka topic."""
        
        if event_type.startswith("playback."):
            return "events.playback"
        elif event_type.startswith("quality."):
            return "events.quality"
        elif event_type.startswith("browse."):
            return "events.browse"
        else:
            return "events.other"
    
    def _country_to_continent(self, country: str) -> str:
        """Map country to continent."""
        # Simplified mapping
        mapping = {
            "US": "NA", "CA": "NA", "MX": "NA",
            "GB": "EU", "DE": "EU", "FR": "EU",
            "JP": "AS", "KR": "AS", "IN": "AS",
            # ... more mappings
        }
        return mapping.get(country, "OTHER")

# Streaming trending calculation with Flink

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Dict, Optional
import math


@dataclass
class TrendingScore:
    """Trending score for a piece of content."""
    content_id: str
    score: float
    unique_viewers_1h: int
    unique_viewers_24h: int
    velocity: float  # Rate of change
    percentile_rank: float
    updated_at: datetime


class TrendingCalculator:
    """
    Calculates trending scores using streaming data.
    
    Algorithm:
    1. Count unique viewers in sliding windows
    2. Calculate velocity (change in viewership)
    3. Normalize by content category baseline
    4. Apply time decay for recency
    5. Combine into final score
    """
    
    def __init__(self):
        # Viewer counts per content per window
        self.viewers_1h: Dict[str, set] = {}   # Last 1 hour
        self.viewers_24h: Dict[str, set] = {}  # Last 24 hours
        
        # Historical baselines by category
        self.category_baselines: Dict[str, float] = {}
        
        # Previous scores for velocity calculation
        self.previous_scores: Dict[str, float] = {}
    
    def process_play_event(self, event: Dict):
        """
        Process a playback event and update trending state.
        """
        
        content_id = event["payload"]["content_id"]
        user_id = event["user_id"]
        
        # Add to viewer sets (HyperLogLog in production)
        if content_id not in self.viewers_1h:
            self.viewers_1h[content_id] = set()
            self.viewers_24h[content_id] = set()
        
        self.viewers_1h[content_id].add(user_id)
        self.viewers_24h[content_id].add(user_id)
    
    def calculate_trending_scores(
        self,
        content_metadata: Dict[str, Dict]
    ) -> list[TrendingScore]:
        """
        Calculate trending scores for all content.
        
        Called every 30-60 seconds.
        """
        
        scores = []
        
        for content_id, viewers_1h in self.viewers_1h.items():
            if len(viewers_1h) < 10:  # Minimum threshold
                continue
            
            metadata = content_metadata.get(content_id, {})
            category = metadata.get("category", "unknown")
            
            # Base score: unique viewers in last hour
            base_score = len(viewers_1h)
            
            # Velocity: how fast is viewership growing?
            viewers_24h = len(self.viewers_24h.get(content_id, set()))
            velocity = self._calculate_velocity(
                content_id, base_score, viewers_24h
            )
            
            # Normalize by category baseline
            baseline = self.category_baselines.get(category, 1000)
            normalized_score = base_score / baseline
            
            # Apply velocity boost (trending UP gets bonus)
            velocity_multiplier = 1 + (velocity * 0.5) if velocity > 0 else 1
            
            # Final score
            final_score = normalized_score * velocity_multiplier
            
            # Store for next velocity calculation
            self.previous_scores[content_id] = base_score
            
            scores.append(TrendingScore(
                content_id=content_id,
                score=final_score,
                unique_viewers_1h=base_score,
                unique_viewers_24h=viewers_24h,
                velocity=velocity,
                percentile_rank=0,  # Calculated after sorting
                updated_at=datetime.now(timezone.utc)
            ))
        
        # Sort and assign percentile ranks
        scores.sort(key=lambda x: x.score, reverse=True)
        for i, score in enumerate(scores):
            score.percentile_rank = 1 - (i / len(scores))
        
        return scores
    
    def _calculate_velocity(
        self,
        content_id: str,
        current_1h: int,
        current_24h: int
    ) -> float:
        """
        Calculate velocity (rate of change) of viewership.
        
        Positive = trending up, Negative = trending down
        """
        
        previous = self.previous_scores.get(content_id, current_1h)
        
        if previous == 0:
            return 0
        
        # Percentage change from previous calculation
        change = (current_1h - previous) / previous
        
        # Also factor in 1h vs 24h ratio
        # If 1h viewers >> 24h/24 average, content is spiking
        hourly_avg_24h = current_24h / 24
        spike_factor = current_1h / hourly_avg_24h if hourly_avg_24h > 0 else 1
        
        return change * spike_factor
    
    def expire_old_data(self, window_cutoff: datetime):
        """
        Remove old viewer data from windows.
        
        In production, use Flink's windowing with watermarks.
        """
        # Implementation would use event time windows
        pass


# Flink job definition (pseudo-code)
FLINK_TRENDING_JOB = """
-- Flink SQL for trending calculation

-- Step 1: Deduplicate events by user per content per window
-- (1-hour window that slides every minute, so trending refreshes
--  every minute; a 1-hour TUMBLE would only emit once an hour)
CREATE VIEW unique_viewers AS
SELECT 
    content_id,
    COUNT(DISTINCT user_id) as viewer_count,
    HOP_END(event_time, INTERVAL '1' MINUTE, INTERVAL '1' HOUR) as window_end,
    -- HOP_END is a plain timestamp, not a time attribute; HOP_ROWTIME is
    HOP_ROWTIME(event_time, INTERVAL '1' MINUTE, INTERVAL '1' HOUR) as window_time
FROM playback_events
WHERE event_type = 'playback.started'
GROUP BY 
    content_id,
    HOP(event_time, INTERVAL '1' MINUTE, INTERVAL '1' HOUR);

-- Step 2: Calculate velocity using LAG
CREATE VIEW trending_with_velocity AS
SELECT 
    content_id,
    viewer_count,
    window_end,
    window_time,
    viewer_count - LAG(viewer_count) OVER (
        PARTITION BY content_id 
        ORDER BY window_time
    ) as velocity
FROM unique_viewers;

-- Step 3: Join with metadata and calculate final score
CREATE VIEW trending_scores AS
SELECT 
    t.content_id,
    t.viewer_count,
    t.velocity,
    t.viewer_count / m.category_baseline * 
        (1 + GREATEST(t.velocity, 0) * 0.5) as trending_score,
    t.window_end
FROM trending_with_velocity t
JOIN content_metadata m ON t.content_id = m.content_id;
"""

# Sink trending scores to ClickHouse for dashboards

class TrendingClickHouseSink:
    """
    Writes trending scores to ClickHouse for dashboard queries.
    """
    
    async def write_scores(self, scores: list[TrendingScore]):
        """
        Write batch of trending scores.
        
        Uses ReplacingMergeTree to handle updates.
        """
        
        values = [
            (
                s.content_id,
                s.score,
                s.unique_viewers_1h,
                s.unique_viewers_24h,
                s.velocity,
                s.percentile_rank,
                s.updated_at
            )
            for s in scores
        ]
        
        await self.client.execute(
            """
            INSERT INTO trending_scores 
            (content_id, score, viewers_1h, viewers_24h, 
             velocity, percentile_rank, updated_at)
            VALUES
            """,
            values
        )


CLICKHOUSE_SCHEMA = """
CREATE TABLE trending_scores (
    content_id String,
    score Float64,
    viewers_1h UInt32,
    viewers_24h UInt32,
    velocity Float64,
    percentile_rank Float64,
    updated_at DateTime
)
ENGINE = ReplacingMergeTree(updated_at)
ORDER BY content_id;

-- Top 100 trending: a plain VIEW, evaluated at query time.
-- (A materialized view only sees each INSERT block, so ORDER BY ...
--  LIMIT 100 there would keep a per-batch top 100, not a global one.)
CREATE VIEW top_trending AS
SELECT *
FROM trending_scores FINAL  -- collapse rows not yet merged by ReplacingMergeTree
ORDER BY score DESC
LIMIT 100;
"""

# Late data handling with versioned results

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Optional
from enum import Enum


class DataFreshness(Enum):
    REALTIME = "realtime"      # Streaming result, may change
    PRELIMINARY = "preliminary" # T+1, most late data included
    FINAL = "final"            # T+7, official number


@dataclass
class ViewingMetric:
    """A metric with versioning for late data."""
    content_id: str
    date: str
    metric_name: str
    value: float
    freshness: DataFreshness
    computed_at: datetime
    event_count: int
    late_event_count: int


class LateDataAwareAggregator:
    """
    Aggregates viewing metrics with late data handling.
    
    Maintains multiple versions:
    - Real-time: Updated every minute, provisional
    - Preliminary: Updated T+1, includes most late data
    - Final: Updated T+7, official numbers
    """
    
    def __init__(self, metrics_store):
        self.store = metrics_store
        
        # Late data tracking
        self.lateness_histogram: Dict[str, int] = {}
    
    async def process_event(self, event: Dict):
        """
        Process event and update appropriate metrics version.
        """
        
        event_time = datetime.fromisoformat(event["event_time"])
        received_time = datetime.fromisoformat(event["received_time"])
        
        lateness = received_time - event_time
        lateness_bucket = self._get_lateness_bucket(lateness)
        self.lateness_histogram[lateness_bucket] = \
            self.lateness_histogram.get(lateness_bucket, 0) + 1
        
        event_date = event_time.date().isoformat()
        
        # Determine which version to update
        if lateness < timedelta(minutes=5):
            # On-time: update real-time
            await self._update_realtime(event, event_date)
        elif lateness < timedelta(hours=24):
            # Moderately late: queue for preliminary
            await self._queue_for_preliminary(event, event_date)
        else:
            # Very late: queue for final processing
            await self._queue_for_final(event, event_date)
    
    async def _update_realtime(self, event: Dict, event_date: str):
        """Update real-time metrics (streaming)."""
        
        content_id = event["payload"]["content_id"]
        
        # Increment view count (starts only; heartbeats are not views,
        # matching _compute_daily_metrics)
        if event["event_type"] == "playback.started":
            await self.store.increment(
                content_id=content_id,
                date=event_date,
                metric="views",
                freshness=DataFreshness.REALTIME
            )
        
        # Update completion if applicable
        if event["event_type"] == "playback.completed":
            await self.store.increment(
                content_id=content_id,
                date=event_date,
                metric="completions",
                freshness=DataFreshness.REALTIME
            )
    
    async def _queue_for_preliminary(self, event: Dict, event_date: str):
        """Queue late event for preliminary processing."""
        await self.store.queue_late_event(
            event=event,
            target_freshness=DataFreshness.PRELIMINARY
        )
    
    async def _queue_for_final(self, event: Dict, event_date: str):
        """Queue very late event for final processing."""
        await self.store.queue_late_event(
            event=event,
            target_freshness=DataFreshness.FINAL
        )
    
    async def run_preliminary_job(self, target_date: str):
        """
        Run T+1 job to produce preliminary numbers.
        
        Scheduled to run at 6 AM for previous day.
        """
        
        # Read all events for target_date from data lake
        events = await self.store.read_events_for_date(target_date)
        
        # Also read late events that arrived after streaming cutoff
        late_events = await self.store.read_late_events(
            target_date=target_date,
            max_lateness=timedelta(hours=30)
        )
        
        all_events = events + late_events
        
        # Compute aggregates
        metrics = self._compute_daily_metrics(all_events, target_date)
        
        # Store as preliminary
        for metric in metrics:
            metric.freshness = DataFreshness.PRELIMINARY
            await self.store.save_metric(metric)
    
    async def run_final_job(self, target_date: str):
        """
        Run T+7 job to produce final numbers.
        
        Scheduled to run 7 days after target_date.
        """
        
        # Read ALL events including very late ones
        events = await self.store.read_events_for_date(
            target_date,
            include_very_late=True
        )
        
        # Compute final aggregates
        metrics = self._compute_daily_metrics(events, target_date)
        
        # Store as final (immutable after this)
        for metric in metrics:
            metric.freshness = DataFreshness.FINAL
            await self.store.save_metric(metric)
    
    def _compute_daily_metrics(
        self,
        events: List[Dict],
        date: str
    ) -> List[ViewingMetric]:
        """Compute all metrics for a day."""
        
        by_content: Dict[str, Dict] = {}
        
        for event in events:
            content_id = event["payload"]["content_id"]
            
            if content_id not in by_content:
                by_content[content_id] = {
                    "views": 0,
                    "unique_viewers": set(),
                    "completions": 0,
                    "total_watch_time": 0,
                    "event_count": 0,
                    "late_event_count": 0
                }
            
            stats = by_content[content_id]
            stats["event_count"] += 1
            
            # Check if late
            event_time = datetime.fromisoformat(event["event_time"])
            received_time = datetime.fromisoformat(event["received_time"])
            if (received_time - event_time) > timedelta(minutes=5):
                stats["late_event_count"] += 1
            
            if event["event_type"] == "playback.started":
                stats["views"] += 1
                stats["unique_viewers"].add(event["user_id"])
            
            if event["event_type"] == "playback.completed":
                stats["completions"] += 1
            
            if event["event_type"] == "playback.heartbeat":
                stats["total_watch_time"] += 10  # 10 second heartbeats
        
        # Convert to metrics
        metrics = []
        for content_id, stats in by_content.items():
            metrics.extend([
                ViewingMetric(
                    content_id=content_id,
                    date=date,
                    metric_name="views",
                    value=stats["views"],
                    freshness=DataFreshness.PRELIMINARY,
                    computed_at=datetime.now(timezone.utc),
                    event_count=stats["event_count"],
                    late_event_count=stats["late_event_count"]
                ),
                ViewingMetric(
                    content_id=content_id,
                    date=date,
                    metric_name="unique_viewers",
                    value=len(stats["unique_viewers"]),
                    freshness=DataFreshness.PRELIMINARY,
                    computed_at=datetime.now(timezone.utc),
                    event_count=stats["event_count"],
                    late_event_count=stats["late_event_count"]
                ),
                ViewingMetric(
                    content_id=content_id,
                    date=date,
                    metric_name="completion_rate",
                    value=stats["completions"] / stats["views"] if stats["views"] > 0 else 0,
                    freshness=DataFreshness.PRELIMINARY,
                    computed_at=datetime.now(timezone.utc),
                    event_count=stats["event_count"],
                    late_event_count=stats["late_event_count"]
                )
            ])
        
        return metrics
    
    def _get_lateness_bucket(self, lateness: timedelta) -> str:
        """Bucket lateness for histogram."""
        seconds = lateness.total_seconds()
        
        if seconds < 60:
            return "<1min"
        elif seconds < 300:
            return "1-5min"
        elif seconds < 3600:
            return "5-60min"
        elif seconds < 86400:
            return "1-24hr"
        else:
            return ">24hr"

# Content analytics data model

BIGQUERY_SCHEMA = """
-- =============================================================================
-- Dimension Tables
-- =============================================================================

-- Content dimension (SCD Type 2 for metadata changes)
CREATE TABLE dim_content (
    content_key INT64 NOT NULL,
    content_id STRING NOT NULL,
    title STRING NOT NULL,
    content_type STRING NOT NULL,  -- 'movie', 'series', 'documentary'
    genre STRING,
    maturity_rating STRING,
    release_date DATE,
    duration_minutes INT64,
    studio STRING,
    creator_id STRING,
    
    -- SCD Type 2 fields
    valid_from DATE NOT NULL,
    valid_to DATE,
    is_current BOOL NOT NULL,
    
    -- Metadata
    created_at TIMESTAMP,
    updated_at TIMESTAMP
);

-- Date dimension (generated)
CREATE TABLE dim_date (
    date_key INT64 NOT NULL,  -- YYYYMMDD
    date DATE NOT NULL,
    day_of_week INT64,
    day_name STRING,
    month INT64,
    month_name STRING,
    quarter INT64,
    year INT64,
    is_weekend BOOL,
    is_holiday BOOL
);

-- Viewer dimension (anonymized/bucketed for privacy)
CREATE TABLE dim_viewer (
    viewer_key INT64 NOT NULL,
    age_group STRING,  -- '18-24', '25-34', etc.
    gender STRING,
    country STRING,
    subscription_type STRING,  -- 'basic', 'standard', 'premium'
    tenure_months INT64,
    preferred_device STRING
);

-- Device dimension
CREATE TABLE dim_device (
    device_key INT64 NOT NULL,
    device_type STRING,  -- 'mobile', 'tv', 'web', 'gaming'
    operating_system STRING,
    app_version_major INT64,
    screen_category STRING  -- 'small', 'medium', 'large'
);

-- Geography dimension
CREATE TABLE dim_geo (
    geo_key INT64 NOT NULL,
    country STRING,
    region STRING,
    timezone STRING
);


-- =============================================================================
-- Fact Tables
-- =============================================================================

-- Daily viewing facts (grain: content × date × viewer_segment × device × geo)
CREATE TABLE fact_daily_views (
    date_key INT64 NOT NULL,
    content_key INT64 NOT NULL,
    viewer_key INT64 NOT NULL,
    device_key INT64 NOT NULL,
    geo_key INT64 NOT NULL,
    
    -- Measures
    view_count INT64,
    unique_viewers INT64,  -- Approximate, using HLL
    watch_minutes INT64,
    completion_count INT64,
    
    -- Derived measures
    avg_completion_pct FLOAT64,
    avg_watch_minutes FLOAT64,
    
    -- Data quality
    freshness STRING,  -- 'realtime', 'preliminary', 'final'
    late_event_pct FLOAT64,
    
    -- Partition column: the viewing date (event time), not load time,
    -- so late events land in the right partition
    view_date DATE NOT NULL
)
PARTITION BY view_date
CLUSTER BY content_key, geo_key;


-- =============================================================================
-- Pre-Aggregated Rollup Tables
-- =============================================================================

-- Daily content summary (most common query)
CREATE TABLE agg_daily_content (
    date DATE NOT NULL,
    content_id STRING NOT NULL,
    
    total_views INT64,
    unique_viewers INT64,       -- that day only
    viewers_hll BYTES,          -- HLL sketch: mergeable across days
    watch_hours FLOAT64,
    completions INT64,
    completion_rate FLOAT64,
    
    -- Breakdown by device
    mobile_views INT64,
    tv_views INT64,
    web_views INT64,
    
    -- Breakdown by top geos
    us_views INT64,
    uk_views INT64,
    other_views INT64,
    
    freshness STRING,
    updated_at TIMESTAMP
)
PARTITION BY date
CLUSTER BY content_id;


-- Weekly content summary
CREATE TABLE agg_weekly_content (
    week_start DATE NOT NULL,
    content_id STRING NOT NULL,
    
    total_views INT64,
    unique_viewers INT64,
    watch_hours FLOAT64,
    avg_daily_views FLOAT64,
    peak_day_views INT64,
    
    freshness STRING,
    updated_at TIMESTAMP
);
"""


# Query examples for creator dashboard

CREATOR_DASHBOARD_QUERIES = """
-- Query 1: Content performance summary (last 30 days)
-- Uses agg_daily_content rollup for speed

SELECT 
    c.title,
    c.content_type,
    SUM(a.total_views) as total_views,
    -- Summing daily uniques would count a 5-day viewer 5 times
    HLL_COUNT.MERGE(a.viewers_hll) as unique_viewers,
    ROUND(SUM(a.watch_hours), 1) as watch_hours,
    ROUND(SAFE_DIVIDE(SUM(a.completions), SUM(a.total_views)) * 100, 1) as completion_pct,
    SUM(a.mobile_views) / SUM(a.total_views) * 100 as mobile_pct
FROM agg_daily_content a
JOIN dim_content c ON a.content_id = c.content_id
WHERE a.date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
  AND c.creator_id = @creator_id
  AND c.is_current = TRUE
GROUP BY c.title, c.content_type
ORDER BY total_views DESC;


-- Query 2: Daily trend for specific content
SELECT 
    a.date,
    a.total_views,
    a.unique_viewers,
    a.completion_rate,
    a.freshness
FROM agg_daily_content a
WHERE a.content_id = @content_id
  AND a.date >= DATE_SUB(CURRENT_DATE(), INTERVAL 90 DAY)
ORDER BY a.date;


-- Query 3: Audience breakdown (uses fact table for detail)
SELECT 
    v.age_group,
    v.gender,
    g.country,
    -- Sum of daily uniques: a viewer active on several days counts
    -- once per day (use HLL sketches for true uniques)
    SUM(f.unique_viewers) as viewer_days,
    AVG(f.avg_completion_pct) as avg_completion
FROM fact_daily_views f
JOIN dim_viewer v ON f.viewer_key = v.viewer_key
JOIN dim_geo g ON f.geo_key = g.geo_key
JOIN dim_content c ON f.content_key = c.content_key
WHERE c.content_id = @content_id
  AND f.date_key >= @start_date_key
  AND f.date_key <= @end_date_key
GROUP BY v.age_group, v.gender, g.country
ORDER BY viewer_days DESC;


-- Query 4: Comparison to similar content
WITH my_content AS (
    SELECT 
        SUM(total_views) as views,
        SAFE_DIVIDE(SUM(completions), SUM(total_views)) as completion_rate
    FROM agg_daily_content
    WHERE content_id = @content_id
      AND date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
),
similar_content AS (
    -- Average of per-title 30-day totals (same basis as my_content)
    SELECT 
        AVG(views) as avg_views,
        AVG(completion_rate) as avg_completion_rate
    FROM (
        SELECT 
            a.content_id,
            SUM(a.total_views) as views,
            SAFE_DIVIDE(SUM(a.completions), SUM(a.total_views)) as completion_rate
        FROM agg_daily_content a
        JOIN dim_content c ON a.content_id = c.content_id AND c.is_current
        WHERE c.genre = @genre
          AND c.content_type = @content_type
          AND a.date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
        GROUP BY a.content_id
    )
)
SELECT 
    m.views,
    m.completion_rate,
    s.avg_views as category_avg_views,
    s.avg_completion_rate as category_avg_completion,
    m.views / s.avg_views as views_vs_category,
    m.completion_rate / s.avg_completion_rate as completion_vs_category
FROM my_content m, similar_content s;
"""

# Multi-layer caching for creator dashboard

from datetime import timedelta


class CreatorDashboardCache:
    """
    Caching strategy for creator dashboard queries.
    """
    
    def __init__(self, redis_client, bigquery_client):
        self.redis = redis_client
        self.bq = bigquery_client
        
        # TTL configuration by query type
        self.ttl_config = {
            "summary_30d": timedelta(minutes=5),     # Updates frequently
            "daily_trend": timedelta(minutes=15),    # Less frequent
            "audience_breakdown": timedelta(hours=1), # Slow-changing
            "comparison": timedelta(hours=1),        # Slow-changing
        }
    
    async def get_content_summary(
        self,
        creator_id: str,
        days: int = 30
    ) -> dict:
        """
        Get content performance summary with caching.
        """
        
        cache_key = f"creator:{creator_id}:summary:{days}d"
        
        # Check cache
        cached = await self.redis.get(cache_key)
        if cached:
            return {
                "data": cached,
                "cache_hit": True,
                "freshness": "cached"
            }
        
        # Query BigQuery
        result = await self._query_content_summary(creator_id, days)
        
        # Cache result
        ttl = self.ttl_config["summary_30d"]
        await self.redis.setex(cache_key, int(ttl.total_seconds()), result)
        
        return {
            "data": result,
            "cache_hit": False,
            "freshness": result.get("freshness", "unknown")
        }
    
    async def get_daily_trend(
        self,
        content_id: str,
        days: int = 90
    ) -> dict:
        """
        Get daily viewing trend for content.
        """
        
        cache_key = f"content:{content_id}:trend:{days}d"
        
        cached = await self.redis.get(cache_key)
        if cached:
            return {"data": cached, "cache_hit": True}
        
        result = await self._query_daily_trend(content_id, days)
        
        ttl = self.ttl_config["daily_trend"]
        await self.redis.setex(cache_key, int(ttl.total_seconds()), result)
        
        return {"data": result, "cache_hit": False}
    
    async def invalidate_for_content(self, content_id: str):
        """
        Invalidate all caches for a content item.
        
        Called when new data arrives.
        """
        
        pattern = f"content:{content_id}:*"
        # SCAN, not KEYS: KEYS is O(N) and blocks Redis while it runs
        keys = [k async for k in self.redis.scan_iter(match=pattern)]
        
        if keys:
            await self.redis.unlink(*keys)  # UNLINK frees memory off the main thread
    
    async def warm_cache_for_creator(self, creator_id: str):
        """
        Pre-warm cache for a creator's common queries.
        
        Called when creator logs into dashboard.
        """
        
        # Fire off all common queries in parallel
        await asyncio.gather(
            self.get_content_summary(creator_id, 30),
            self.get_content_summary(creator_id, 7),
            # Don't pre-warm content-specific queries
            # (too many possible content_ids)
        )