-- Materialized view for daily sales dashboard
CREATE MATERIALIZED VIEW mv_daily_sales AS
SELECT 
    d.date,
    d.year,
    d.month,
    p.category,
    p.brand,
    c.segment,
    c.country,
    SUM(f.total_amount) as revenue,
    SUM(f.quantity) as units,
    COUNT(DISTINCT f.customer_key) as unique_customers,
    AVG(f.total_amount) as avg_order_value
FROM fact_sales f
JOIN dim_date d ON f.date_key = d.date_key
JOIN dim_product p ON f.product_key = p.product_key
JOIN dim_customer c ON f.customer_key = c.customer_key
GROUP BY 1, 2, 3, 4, 5, 6, 7;

-- Refresh strategy: nightly
REFRESH MATERIALIZED VIEW mv_daily_sales;

# FastAPI data serving endpoint
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from datetime import date
import psycopg2

app = FastAPI(title="Data Serving API")

class SalesSummary(BaseModel):
    date: date
    category: str
    revenue: float
    units: int

@app.get("/api/v1/sales/daily", response_model=list[SalesSummary])
async def get_daily_sales(start_date: date, end_date: date, category: str = None):
    conn = psycopg2.connect("dbname=sales_mart host=warehouse-cluster")
    cur = conn.cursor()
    
    query = """
        SELECT date, category, revenue, units
        FROM mv_daily_sales
        WHERE date BETWEEN %s AND %s
    """
    params = [start_date, end_date]
    
    if category:
        query += " AND category = %s"
        params.append(category)
    
    cur.execute(query, params)
    results = [SalesSummary(*row) for row in cur.fetchall()]
    conn.close()
    return results

@app.get("/api/v1/sales/top-products")
async def get_top_products(limit: int = 10):
    conn = psycopg2.connect("dbname=sales_mart host=warehouse-cluster")
    cur = conn.cursor()
    cur.execute("""
        SELECT p.product_name, SUM(f.total_amount) as revenue
        FROM fact_sales f
        JOIN dim_product p ON f.product_key = p.product_key
        GROUP BY 1
        ORDER BY 2 DESC
        LIMIT %s
    """, (limit,))
    results = [{"product": row[0], "revenue": row[1]} for row in cur.fetchall()]
    conn.close()
    return results

# Redis caching pattern for data serving
import redis
import json
from datetime import timedelta

redis_client = redis.Redis(host='cache-cluster', port=6379, db=0)

def get_sales_summary(date: str, cache_ttl: int = 3600):
    cache_key = f"sales_summary:{date}"
    
    # Check cache first
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # Cache miss: query warehouse
    conn = psycopg2.connect("dbname=sales_mart host=warehouse-cluster")
    cur = conn.cursor()
    cur.execute("""
        SELECT category, SUM(revenue), SUM(units)
        FROM mv_daily_sales
        WHERE date = %s
        GROUP BY 1
    """, (date,))
    
    results = [{"category": row[0], "revenue": row[1], "units": row[2]} 
               for row in cur.fetchall()]
    conn.close()
    
    # Cache the result
    redis_client.setex(cache_key, timedelta(seconds=cache_ttl), json.dumps(results))
    
    return results

# Feature store pattern using AWS SageMaker Feature Store
import boto3

featurestore_client = boto3.client('sagemaker-featurestore-runtime')

def get_customer_features(customer_id: str):
    """Retrieve pre-computed features for ML inference"""
    response = featurestore_client.get_record(
        FeatureGroupName='customer_features',
        RecordIdentifierValueAsString=customer_id
    )
    
    features = {}
    for feature in response['Record']:
        features[feature['FeatureName']] = feature['ValueAsString']
    
    return features

# Features served: customer_lifetime_value, avg_order_value, 
# purchase_frequency, days_since_last_purchase, segment_score

# Real-time aggregation with Kinesis Data Analytics
CREATE OR REPLACE STREAM sales_aggregated_stream (
    category VARCHAR(100),
    window_start TIMESTAMP,
    window_end TIMESTAMP,
    total_revenue DECIMAL(12,2),
    order_count INTEGER
);

CREATE OR REPLACE PUMP sales_aggregated_pump AS
INSERT INTO sales_aggregated_stream
SELECT 
    category,
    TUMBLE_START(proctime, INTERVAL '5' MINUTE) as window_start,
    TUMBLE_END(proctime, INTERVAL '5' MINUTE) as window_end,
    SUM(total_amount) as total_revenue,
    COUNT(*) as order_count
FROM sales_stream
GROUP BY category, TUMBLE(proctime, INTERVAL '5' MINUTE);