Real-time Streaming
Architecture

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.window import EventTimeSessionWindows
from pyflink.common.time import Time
from pyflink.common.watermark_strategy import WatermarkStrategy
import json, math

env = StreamExecutionEnvironment.get_execution_environment()
env.enable_checkpointing(60_000)  # exactly-once, 60s interval

# Consume user activity events from Kafka
kafka_source = KafkaSource.builder() \
    .set_bootstrap_servers("kafka:9092") \
    .set_topics("user.watch-events", "user.trailer-events") \
    .set_group_id("profile-aggregator-cg") \
    .set_value_only_deserializer(JsonRowDeserializationSchema()) \
    .build()

events = env.from_source(
    kafka_source,
    WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(5)),
    "kafka-events"
)

def compute_genre_profile(key, events, ctx):
    """Sliding 30-day window with exponential recency decay."""
    now = ctx.timestamp()
    HALF_LIFE_DAYS = 7
    genre_weights = {}

    for event in events:
        age_days = (now - event["timestamp"]) / 86_400_000
        decay = math.exp(-0.693 * age_days / HALF_LIFE_DAYS)
        weight = event["watch_pct"] * decay
        genre = event["genre"]
        genre_weights[genre] = genre_weights.get(genre, 0.0) + weight

    # Normalize to unit vector for cosine similarity downstream
    magnitude = math.sqrt(sum(v ** 2 for v in genre_weights.values()))
    if magnitude > 0:
        genre_weights = {k: v / magnitude for k, v in genre_weights.items()}

    return {
        "user_id": key,
        "genre_vector": genre_weights,
        "event_count": len(list(events)),
        "updated_at": now
    }

# Session window groups bursts of activity (gap = 30 min)
profiles = events \
    .key_by(lambda e: e["user_id"]) \
    .window(EventTimeSessionWindows.with_gap(Time.minutes(30))) \
    .apply(compute_genre_profile)

# Sink updated profiles to the feature store (Redis)
profiles.add_sink(
    RedisSink(redis_config, RedisMapper(
        key_fn=lambda p: f"user:{p['user_id']}:profile",
        value_fn=lambda p: json.dumps(p),
        ttl=86_400  # 24h TTL, refreshed on each update
    ))
)

env.execute("user-profile-aggregator")

import numpy as np
from scipy.sparse import csr_matrix
from implicit.als import AlternatingLeastSquares

def build_interaction_matrix(events, user_idx, item_idx):
    """Build user-item matrix: weight = completion% x recency decay."""
    rows, cols, vals = [], [], []
    now = max(e["timestamp"] for e in events)

    for e in events:
        uid = user_idx[e["user_id"]]
        iid = item_idx[e["content_id"]]
        age_days = (now - e["timestamp"]) / 86_400_000
        recency = np.exp(-0.693 * age_days / 7)  # half-life = 7 days
        weight = e["watch_pct"] * recency
        rows.append(uid)
        cols.append(iid)
        vals.append(weight)

    return csr_matrix(
        (vals, (rows, cols)),
        shape=(len(user_idx), len(item_idx))
    )

# Train ALS for implicit feedback (no explicit ratings)
model = AlternatingLeastSquares(
    factors=128,           # latent dimensions
    regularization=0.01,   # L2 penalty
    iterations=20,         # alternating steps
    use_gpu=False
)

interaction_matrix = build_interaction_matrix(events, user_idx, item_idx)
model.fit(interaction_matrix)

def recommend(user_id, n=20, already_watched=None):
    """Generate top-K recs via dot product of user x item factors."""
    uid = user_idx[user_id]
    user_factors = model.user_factors[uid]       # (128,)
    scores = model.item_factors.dot(user_factors) # (n_items,)

    # Zero out already-watched items
    if already_watched:
        for item_id in already_watched:
            scores[item_idx[item_id]] = -np.inf

    top_k = np.argsort(scores)[::-1][:n]
    return [(idx_to_item[i], float(scores[i])) for i in top_k]

from fastapi import FastAPI
from redis.asyncio import Redis
import numpy as np, json

app = FastAPI()
redis = Redis(host="redis", port=6379, decode_responses=True)

# Pre-loaded item factors (n_items x 128) from last ALS training
ITEM_FACTORS = np.load("/models/item_factors.npy")
ITEM_META = json.load(open("/models/item_meta.json"))

@app.get("/recommend/{user_id}")
async def recommend(user_id: str, n: int = 10):
    # 1. Load pre-computed user factors from Redis
    raw = await redis.get(f"user:{user_id}:factors")
    if not raw:
        return {"recs": [], "reason": "cold-start: no user factors"}
    user_factors = np.frombuffer(raw, dtype=np.float32)  # (128,)

    # 2. Score all candidate items via dot product
    scores = ITEM_FACTORS @ user_factors  # (n_items,)

    # 3. Filter already-watched content
    watched = await redis.smembers(f"user:{user_id}:watched")
    for item_id in watched:
        idx = ITEM_META["id_to_idx"].get(item_id)
        if idx is not None:
            scores[idx] = -np.inf

    # 4. Apply diversity constraint: max 3 per genre in top results
    ranked = np.argsort(scores)[::-1]
    results, genre_counts = [], {}

    for idx in ranked:
        if len(results) >= n:
            break
        meta = ITEM_META["items"][idx]
        genre = meta["genre"]
        if genre_counts.get(genre, 0) >= 3:
            continue
        # Boost new releases (< 30 days old)
        boost = 1.15 if meta.get("is_new_release") else 1.0
        genre_counts[genre] = genre_counts.get(genre, 0) + 1
        results.append({
            "item_id": meta["id"],
            "title": meta["title"],
            "genre": genre,
            "score": round(float(scores[idx]) * boost, 4),
            "reason": f"Similar to your {genre} preferences"
        })

    return {"user_id": user_id, "recs": results}