Design a News Feed

import math
from datetime import datetime, timedelta
from dataclasses import dataclass

@dataclass
class Post:
    post_id: str
    author_id: str
    created_at: datetime
    likes: int
    comments: int
    shares: int
    post_type: str  # 'text', 'image', 'video'


class FeedRanker:
    """
    Multi-signal feed ranking algorithm.

    Score = affinity * edge_weight * time_decay

    Affinity:    How close is the viewer to the author?
    Edge Weight: How engaging is this type of content?
    Time Decay:  How fresh is the post?
    """

    # Content type weights (based on engagement data)
    TYPE_WEIGHTS = {
        "video": 1.5,
        "image": 1.2,
        "text": 1.0,
        "link": 0.8,
    }

    def calculate_score(self, post: Post, viewer_id: str) -> float:
        """Calculate ranking score for a post relative to a viewer."""
        affinity = self._get_affinity(viewer_id, post.author_id)
        engagement = self._engagement_score(post)
        freshness = self._time_decay(post.created_at)
        type_boost = self.TYPE_WEIGHTS.get(post.post_type, 1.0)

        return affinity * engagement * freshness * type_boost

    def _get_affinity(self, viewer_id: str, author_id: str) -> float:
        """
        Relationship strength between viewer and author.
        Based on: messages sent, profile views, likes, comments.
        Range: 0.0 to 1.0
        """
        # In production: ML model trained on interaction data
        interactions = get_interaction_count(viewer_id, author_id)
        return min(1.0, interactions / 100)

    def _engagement_score(self, post: Post) -> float:
        """Weighted sum of engagement signals."""
        return (
            post.likes * 1.0 +
            post.comments * 2.0 +  # comments are more valuable
            post.shares * 3.0     # shares indicate high quality
        )

    def _time_decay(self, created_at: datetime) -> float:
        """
        Exponential time decay.
        Half-life of ~6 hours: a post loses half its
        freshness score every 6 hours.
        """
        hours_old = (datetime.utcnow() - created_at).total_seconds() / 3600
        half_life = 6.0
        return math.pow(0.5, hours_old / half_life)

    def rank_feed(self, posts: list, viewer_id: str, limit: int = 50) -> list:
        """Rank and return top posts for the viewer's feed."""
        scored = [
            (self.calculate_score(post, viewer_id), post)
            for post in posts
        ]
        scored.sort(key=lambda x: x[0], reverse=True)
        return [post for _, post in scored[:limit]]

import redis
import json

class FeedCache:
    """
    Multi-tier caching strategy for the news feed.

    Tier 1: Pre-computed feed (Redis sorted set per user)
    Tier 2: Post content cache (Redis hash)
    Tier 3: Social graph cache (who follows whom)
    """

    def __init__(self):
        self.redis = redis.Redis(host='redis-cluster')
        self.FEED_SIZE = 500  # Cache last 500 post IDs per user
        self.FEED_TTL = 86400  # 24 hours

    def add_to_feed(self, user_id: str, post_id: str, score: float):
        """Push a new post into a user's feed cache (fan-out on write)."""
        key = f"feed:{user_id}"
        # Sorted set: score determines feed ordering
        self.redis.zadd(key, {post_id: score})
        # Trim to keep only the most recent N posts
        self.redis.zremrangebyrank(key, 0, -(self.FEED_SIZE + 1))
        self.redis.expire(key, self.FEED_TTL)

    def get_feed(self, user_id: str, page: int = 0,
                  size: int = 20) -> list:
        """Fetch a page of feed items for the user."""
        key = f"feed:{user_id}"
        start = page * size
        end = start + size - 1

        # Get post IDs from sorted set (highest score first)
        post_ids = self.redis.zrevrange(key, start, end)

        if not post_ids:
            # Cache miss: regenerate from database
            return self._rebuild_feed(user_id, page, size)

        # Fetch full post content (Tier 2 cache)
        return self._hydrate_posts(post_ids)

    def _hydrate_posts(self, post_ids: list) -> list:
        """Fetch full post data using pipeline for efficiency."""
        pipe = self.redis.pipeline()
        for pid in post_ids:
            pipe.hgetall(f"post:{pid}")
        return pipe.execute()