Feed Caching.

# Fan-Out on Write Implementation

class FanOutOnWriteFeedService:
    """
    Push-based feed system.
    
    When a user posts, we immediately push the post
    to all their followers' feed caches.
    """
    
    def __init__(self, redis_client, db_client, queue_client):
        self.redis = redis_client
        self.db = db_client
        self.queue = queue_client
        
        self.feed_size_limit = 800  # Keep last 800 posts per feed
    
    async def create_post(self, user_id: str, content: str) -> dict:
        """Create post and fan out to followers."""
        
        # 1. Save post to database
        post = await self.db.fetch_one(
            """
            INSERT INTO posts (user_id, content, created_at)
            VALUES ($1, $2, NOW())
            RETURNING *
            """,
            user_id, content
        )
        
        # 2. Get follower count to decide strategy
        follower_count = await self._get_follower_count(user_id)
        
        # 3. Fan out to followers
        if follower_count < 10000:
            # Small account: synchronous fan-out
            await self._fan_out_sync(post)
        else:
            # Large account: async fan-out via queue
            await self._fan_out_async(post)
        
        return dict(post)
    
    async def _fan_out_sync(self, post: dict):
        """Synchronous fan-out for small accounts."""
        user_id = post['user_id']
        post_id = post['id']
        created_at = post['created_at'].timestamp()
        
        # Get all followers
        followers = await self.db.fetch(
            "SELECT follower_id FROM follows WHERE followee_id = $1",
            user_id
        )
        
        # Push to each follower's feed
        pipe = self.redis.pipeline()
        for row in followers:
            follower_id = row['follower_id']
            feed_key = f"feed:{follower_id}"
            
            # Add to sorted set (score = timestamp for ordering)
            pipe.zadd(feed_key, {post_id: created_at})
            
            # Trim to keep only recent posts
            pipe.zremrangebyrank(feed_key, 0, -self.feed_size_limit - 1)
        
        await pipe.execute()
    
    async def _fan_out_async(self, post: dict):
        """Async fan-out via message queue for large accounts."""
        await self.queue.send("fan-out-jobs", {
            "post_id": post['id'],
            "user_id": post['user_id'],
            "created_at": post['created_at'].isoformat()
        })
    
    async def get_feed(self, user_id: str, limit: int = 50, offset: int = 0) -> list:
        """Get user's feed - just read from cache!"""
        feed_key = f"feed:{user_id}"
        
        # Get post IDs from sorted set (newest first)
        post_ids = await self.redis.zrevrange(
            feed_key,
            offset,
            offset + limit - 1
        )
        
        if not post_ids:
            return []
        
        # Fetch post details
        posts = await self._get_posts_by_ids(post_ids)
        return posts
    
    async def _get_posts_by_ids(self, post_ids: list) -> list:
        """Fetch post details, with caching."""
        posts = []
        
        # Try cache first
        pipe = self.redis.pipeline()
        for post_id in post_ids:
            pipe.get(f"post:{post_id}")
        cached = await pipe.execute()
        
        # Separate hits and misses
        to_fetch = []
        for i, (post_id, data) in enumerate(zip(post_ids, cached)):
            if data:
                posts.append((i, json.loads(data)))
            else:
                to_fetch.append((i, post_id))
        
        # Fetch missing from database
        if to_fetch:
            ids = [post_id for _, post_id in to_fetch]
            rows = await self.db.fetch(
                "SELECT * FROM posts WHERE id = ANY($1)",
                ids
            )
            
            # Cache and add to results
            row_map = {str(r['id']): r for r in rows}
            pipe = self.redis.pipeline()
            
            for idx, post_id in to_fetch:
                if post_id in row_map:
                    post = dict(row_map[post_id])
                    posts.append((idx, post))
                    pipe.setex(f"post:{post_id}", 3600, json.dumps(post, default=str))
            
            await pipe.execute()
        
        # Sort by original order
        posts.sort(key=lambda x: x[0])
        return [post for _, post in posts]
    
    async def _get_follower_count(self, user_id: str) -> int:
        """Get follower count (cached)."""
        cache_key = f"follower_count:{user_id}"
        
        cached = await self.redis.get(cache_key)
        if cached:
            return int(cached)
        
        count = await self.db.fetch_val(
            "SELECT COUNT(*) FROM follows WHERE followee_id = $1",
            user_id
        )
        
        await self.redis.setex(cache_key, 300, str(count))
        return count

# Fan-Out on Read Implementation

class FanOutOnReadFeedService:
    """
    Pull-based feed system.
    
    When a user requests their feed, we compute it on demand
    by fetching posts from all accounts they follow.
    """
    
    def __init__(self, redis_client, db_client):
        self.redis = redis_client
        self.db = db_client
    
    async def create_post(self, user_id: str, content: str) -> dict:
        """Create post - just save it, no fan-out!"""
        
        # Save post to database
        post = await self.db.fetch_one(
            """
            INSERT INTO posts (user_id, content, created_at)
            VALUES ($1, $2, NOW())
            RETURNING *
            """,
            user_id, content
        )
        
        # Cache the post for fast retrieval
        await self.redis.setex(
            f"post:{post['id']}",
            3600,
            json.dumps(dict(post), default=str)
        )
        
        # Add to user's posts list (for pull queries)
        await self.redis.zadd(
            f"user_posts:{user_id}",
            {str(post['id']): post['created_at'].timestamp()}
        )
        
        # Trim old posts
        await self.redis.zremrangebyrank(f"user_posts:{user_id}", 0, -1001)
        
        return dict(post)
    
    async def get_feed(self, user_id: str, limit: int = 50) -> list:
        """Get feed by pulling from all followees."""
        
        # 1. Get list of accounts user follows
        following = await self._get_following(user_id)
        
        if not following:
            return []
        
        # 2. Fetch recent posts from each followee (parallel)
        tasks = [
            self._get_user_recent_posts(followee_id)
            for followee_id in following
        ]
        
        all_posts_lists = await asyncio.gather(*tasks)
        
        # 3. Merge all posts
        all_posts = []
        for posts in all_posts_lists:
            all_posts.extend(posts)
        
        # 4. Sort by time (newest first)
        all_posts.sort(key=lambda p: p['created_at'], reverse=True)
        
        # 5. Return top N
        return all_posts[:limit]
    
    async def _get_following(self, user_id: str) -> list:
        """Get list of accounts user follows."""
        cache_key = f"following:{user_id}"
        
        # Try cache
        cached = await self.redis.smembers(cache_key)
        if cached:
            return list(cached)
        
        # Fetch from database
        rows = await self.db.fetch(
            "SELECT followee_id FROM follows WHERE follower_id = $1",
            user_id
        )
        
        following = [str(row['followee_id']) for row in rows]
        
        # Cache for 5 minutes
        if following:
            await self.redis.sadd(cache_key, *following)
            await self.redis.expire(cache_key, 300)
        
        return following
    
    async def _get_user_recent_posts(self, user_id: str, limit: int = 100) -> list:
        """Get recent posts from a user (cached)."""
        cache_key = f"user_posts:{user_id}"
        
        # Get post IDs from sorted set
        post_ids = await self.redis.zrevrange(cache_key, 0, limit - 1)
        
        if not post_ids:
            # Cache miss - fetch from database
            rows = await self.db.fetch(
                """
                SELECT * FROM posts 
                WHERE user_id = $1 
                ORDER BY created_at DESC 
                LIMIT $2
                """,
                user_id, limit
            )
            return [dict(row) for row in rows]
        
        # Fetch post details
        return await self._get_posts_by_ids(post_ids)
    
    async def _get_posts_by_ids(self, post_ids: list) -> list:
        """Fetch post details with caching."""
        # Same as fan-out-on-write implementation
        posts = []
        
        pipe = self.redis.pipeline()
        for post_id in post_ids:
            pipe.get(f"post:{post_id}")
        cached = await pipe.execute()
        
        to_fetch = []
        for i, (post_id, data) in enumerate(zip(post_ids, cached)):
            if data:
                posts.append((i, json.loads(data)))
            else:
                to_fetch.append((i, post_id))
        
        if to_fetch:
            ids = [post_id for _, post_id in to_fetch]
            rows = await self.db.fetch(
                "SELECT * FROM posts WHERE id = ANY($1)",
                ids
            )
            
            row_map = {str(r['id']): r for r in rows}
            pipe = self.redis.pipeline()
            
            for idx, post_id in to_fetch:
                if post_id in row_map:
                    post = dict(row_map[post_id])
                    posts.append((idx, post))
                    pipe.setex(f"post:{post_id}", 3600, json.dumps(post, default=str))
            
            await pipe.execute()
        
        posts.sort(key=lambda x: x[0])
        return [post for _, post in posts]

# Hybrid Feed System

from dataclasses import dataclass
from typing import List, Set
import asyncio


@dataclass 
class FeedConfig:
    """Configuration for hybrid feed system."""
    celebrity_threshold: int = 10000  # Followers to be considered celebrity
    feed_cache_size: int = 800        # Posts to keep in feed cache
    celebrity_posts_limit: int = 100  # Celebrity posts to fetch per request
    feed_ttl: int = 86400             # Feed cache TTL


class HybridFeedService:
    """
    Hybrid feed system combining push and pull.
    
    Regular users (< 10K followers): Fan-out on write
    Celebrities (>= 10K followers): Fan-out on read
    
    This solves the celebrity problem while keeping
    fast reads for most users.
    """
    
    def __init__(
        self,
        redis_client,
        db_client,
        queue_client,
        config: FeedConfig = None
    ):
        self.redis = redis_client
        self.db = db_client
        self.queue = queue_client
        self.config = config or FeedConfig()
        
        # Cache of celebrity user IDs
        self._celebrity_cache: Set[str] = set()
    
    async def create_post(self, user_id: str, content: str) -> dict:
        """Create post with appropriate fan-out strategy."""
        
        # Save post
        post = await self._save_post(user_id, content)
        
        # Check if user is a celebrity
        is_celebrity = await self._is_celebrity(user_id)
        
        if is_celebrity:
            # Celebrity: No fan-out, just index the post
            await self._index_celebrity_post(post)
        else:
            # Regular user: Fan out to followers
            await self._fan_out_to_followers(post)
        
        return post
    
    async def get_feed(self, user_id: str, limit: int = 50) -> list:
        """
        Get feed by merging pre-computed feed with celebrity posts.
        """
        # 1. Get pre-computed feed (from regular users)
        precomputed = await self._get_precomputed_feed(user_id, limit * 2)
        
        # 2. Get celebrities this user follows
        followed_celebrities = await self._get_followed_celebrities(user_id)
        
        # 3. Pull recent posts from followed celebrities
        celebrity_posts = []
        if followed_celebrities:
            celebrity_posts = await self._get_celebrity_posts(
                followed_celebrities,
                limit=self.config.celebrity_posts_limit
            )
        
        # 4. Merge and sort
        all_posts = precomputed + celebrity_posts
        all_posts.sort(key=lambda p: p['created_at'], reverse=True)
        
        # 5. Deduplicate and return top N
        seen = set()
        unique_posts = []
        for post in all_posts:
            if post['id'] not in seen:
                seen.add(post['id'])
                unique_posts.append(post)
                if len(unique_posts) >= limit:
                    break
        
        return unique_posts
    
    async def _is_celebrity(self, user_id: str) -> bool:
        """Check if user is a celebrity (cached)."""
        # Check memory cache first
        if user_id in self._celebrity_cache:
            return True
        
        # Check Redis
        is_celeb = await self.redis.sismember("celebrities", user_id)
        if is_celeb:
            self._celebrity_cache.add(user_id)
            return True
        
        # Check database
        count = await self._get_follower_count(user_id)
        if count >= self.config.celebrity_threshold:
            # Add to celebrity set
            await self.redis.sadd("celebrities", user_id)
            self._celebrity_cache.add(user_id)
            return True
        
        return False
    
    async def _save_post(self, user_id: str, content: str) -> dict:
        """Save post to database."""
        post = await self.db.fetch_one(
            """
            INSERT INTO posts (user_id, content, created_at)
            VALUES ($1, $2, NOW())
            RETURNING *
            """,
            user_id, content
        )
        
        post_dict = dict(post)
        
        # Cache the post
        await self.redis.setex(
            f"post:{post['id']}",
            3600,
            json.dumps(post_dict, default=str)
        )
        
        return post_dict
    
    async def _index_celebrity_post(self, post: dict):
        """Index celebrity post for pull-based retrieval."""
        user_id = post['user_id']
        post_id = str(post['id'])
        timestamp = post['created_at'].timestamp()
        
        # Add to celebrity's posts list
        await self.redis.zadd(
            f"celebrity_posts:{user_id}",
            {post_id: timestamp}
        )
        
        # Trim old posts
        await self.redis.zremrangebyrank(
            f"celebrity_posts:{user_id}",
            0, -self.config.celebrity_posts_limit - 1
        )
    
    async def _fan_out_to_followers(self, post: dict):
        """Fan out post to all followers' feeds."""
        user_id = post['user_id']
        post_id = str(post['id'])
        timestamp = post['created_at'].timestamp()
        
        # Get followers
        followers = await self.db.fetch(
            "SELECT follower_id FROM follows WHERE followee_id = $1",
            user_id
        )
        
        # Update each follower's feed cache
        pipe = self.redis.pipeline()
        
        for row in followers:
            follower_id = row['follower_id']
            feed_key = f"feed:{follower_id}"
            
            pipe.zadd(feed_key, {post_id: timestamp})
            pipe.zremrangebyrank(feed_key, 0, -self.config.feed_cache_size - 1)
        
        await pipe.execute()
    
    async def _get_precomputed_feed(self, user_id: str, limit: int) -> list:
        """Get pre-computed feed from cache."""
        feed_key = f"feed:{user_id}"
        
        post_ids = await self.redis.zrevrange(feed_key, 0, limit - 1)
        
        if not post_ids:
            return []
        
        return await self._get_posts_by_ids(post_ids)
    
    async def _get_followed_celebrities(self, user_id: str) -> list:
        """Get list of celebrities this user follows."""
        # Get all following
        following = await self.redis.smembers(f"following:{user_id}")
        
        if not following:
            rows = await self.db.fetch(
                "SELECT followee_id FROM follows WHERE follower_id = $1",
                user_id
            )
            following = {str(row['followee_id']) for row in rows}
            
            if following:
                await self.redis.sadd(f"following:{user_id}", *following)
                await self.redis.expire(f"following:{user_id}", 300)
        
        # Filter to celebrities only
        celebrities = await self.redis.smembers("celebrities")
        return list(following & celebrities)
    
    async def _get_celebrity_posts(
        self,
        celebrity_ids: list,
        limit: int
    ) -> list:
        """Get recent posts from celebrities."""
        all_posts = []
        
        # Fetch from each celebrity in parallel
        tasks = []
        for celeb_id in celebrity_ids:
            tasks.append(self._get_celebrity_recent_posts(celeb_id))
        
        results = await asyncio.gather(*tasks)
        
        for posts in results:
            all_posts.extend(posts)
        
        # Sort and limit
        all_posts.sort(key=lambda p: p['created_at'], reverse=True)
        return all_posts[:limit]
    
    async def _get_celebrity_recent_posts(self, user_id: str) -> list:
        """Get recent posts from a celebrity."""
        post_ids = await self.redis.zrevrange(
            f"celebrity_posts:{user_id}",
            0, 99
        )
        
        if not post_ids:
            return []
        
        return await self._get_posts_by_ids(post_ids)
    
    async def _get_posts_by_ids(self, post_ids: list) -> list:
        """Fetch posts by IDs with caching."""
        posts = []
        
        # Batch fetch from cache
        pipe = self.redis.pipeline()
        for post_id in post_ids:
            pipe.get(f"post:{post_id}")
        cached = await pipe.execute()
        
        to_fetch = []
        for i, (post_id, data) in enumerate(zip(post_ids, cached)):
            if data:
                posts.append((i, json.loads(data)))
            else:
                to_fetch.append((i, post_id))
        
        # Fetch missing from database
        if to_fetch:
            ids = [pid for _, pid in to_fetch]
            rows = await self.db.fetch(
                "SELECT * FROM posts WHERE id = ANY($1)",
                ids
            )
            
            row_map = {str(r['id']): dict(r) for r in rows}
            
            for idx, post_id in to_fetch:
                if post_id in row_map:
                    posts.append((idx, row_map[post_id]))
        
        posts.sort(key=lambda x: x[0])
        return [post for _, post in posts]
    
    async def _get_follower_count(self, user_id: str) -> int:
        """Get follower count."""
        cached = await self.redis.get(f"follower_count:{user_id}")
        if cached:
            return int(cached)
        
        count = await self.db.fetch_val(
            "SELECT COUNT(*) FROM follows WHERE followee_id = $1",
            user_id
        )
        
        await self.redis.setex(f"follower_count:{user_id}", 300, str(count))
        return count

# Ranked Feed Implementation
from datetime import datetime, timezone


@dataclass
class RankingConfig:
    """Configuration for feed ranking."""
    recency_weight: float = 0.3
    engagement_weight: float = 0.3
    affinity_weight: float = 0.3
    content_weight: float = 0.1
    
    recency_half_life_hours: float = 6.0  # Score halves every 6 hours


class RankedFeedService:
    """
    Feed service with relevance ranking.
    
    Instead of pure chronological order, posts are ranked
    by a combination of recency, engagement, and affinity.
    """
    
    def __init__(self, redis_client, db_client, config: RankingConfig = None):
        self.redis = redis_client
        self.db = db_client
        self.config = config or RankingConfig()
    
    async def get_ranked_feed(
        self,
        user_id: str,
        limit: int = 50
    ) -> list:
        """Get feed ranked by relevance."""
        
        # 1. Get candidate posts (more than we need)
        candidates = await self._get_candidate_posts(user_id, limit * 3)
        
        # 2. Get user's affinity scores
        affinity_scores = await self._get_affinity_scores(user_id)
        
        # 3. Score each post
        scored_posts = []
        for post in candidates:
            score = self._calculate_score(post, affinity_scores)
            scored_posts.append((score, post))
        
        # 4. Sort by score and return top N
        scored_posts.sort(key=lambda x: x[0], reverse=True)
        return [post for _, post in scored_posts[:limit]]
    
    def _calculate_score(
        self,
        post: dict,
        affinity_scores: dict
    ) -> float:
        """Calculate relevance score for a post."""
        
        # Recency score (exponential decay)
        age_hours = (datetime.now(timezone.utc) - post['created_at']).total_seconds() / 3600
        recency_score = 0.5 ** (age_hours / self.config.recency_half_life_hours)
        
        # Engagement score (log scale)
        engagement = post.get('likes', 0) + post.get('comments', 0) * 2
        engagement_score = math.log(engagement + 1) / 10  # Normalize
        
        # Affinity score (from user interaction history)
        author_id = post['user_id']
        affinity_score = affinity_scores.get(author_id, 0.1)  # Default low
        
        # Content type score
        content_score = 0.5
        if post.get('has_image'):
            content_score = 0.7
        if post.get('has_video'):
            content_score = 0.9
        
        # Weighted combination
        total_score = (
            self.config.recency_weight * recency_score +
            self.config.engagement_weight * engagement_score +
            self.config.affinity_weight * affinity_score +
            self.config.content_weight * content_score
        )
        
        return total_score
    
    async def _get_affinity_scores(self, user_id: str) -> dict:
        """Get user's affinity scores for accounts they follow."""
        cache_key = f"affinity:{user_id}"
        
        cached = await self.redis.hgetall(cache_key)
        if cached:
            return {k: float(v) for k, v in cached.items()}
        
        # Calculate from interaction history
        interactions = await self.db.fetch(
            """
            SELECT target_user_id, 
                   COUNT(*) as interaction_count
            FROM user_interactions
            WHERE user_id = $1
            AND created_at > NOW() - INTERVAL '30 days'
            GROUP BY target_user_id
            """,
            user_id
        )
        
        # Normalize to 0-1 range
        if not interactions:
            return {}
        
        max_count = max(row['interaction_count'] for row in interactions)
        scores = {
            str(row['target_user_id']): row['interaction_count'] / max_count
            for row in interactions
        }
        
        # Cache for 1 hour
        if scores:
            await self.redis.hset(cache_key, mapping=scores)
            await self.redis.expire(cache_key, 3600)
        
        return scores
    
    async def _get_candidate_posts(self, user_id: str, limit: int) -> list:
        """Get candidate posts for ranking."""
        # Use hybrid approach to get candidates
        # Then we'll re-rank them
        
        # ... (same as hybrid feed implementation)
        pass

# Sharded Feed Storage

import zlib


class ShardedFeedStorage:
    """
    Sharded storage for feeds.
    
    Distributes feeds across multiple Redis instances
    for horizontal scaling.
    """
    
    def __init__(self, redis_clients: List, num_shards: int):
        self.redis_clients = redis_clients
        self.num_shards = num_shards
    
    def _get_shard(self, user_id: str) -> int:
        """Get shard ID for a user."""
        # Use a stable hash: Python's built-in hash() of a str is randomized
        # per process, so different servers would pick different shards
        return zlib.crc32(user_id.encode()) % self.num_shards
    
    def _get_client(self, user_id: str):
        """Get Redis client for a user."""
        shard_id = self._get_shard(user_id)
        return self.redis_clients[shard_id]
    
    async def add_to_feed(self, user_id: str, post_id: str, timestamp: float):
        """Add post to user's feed."""
        client = self._get_client(user_id)
        feed_key = f"feed:{user_id}"
        
        await client.zadd(feed_key, {post_id: timestamp})
        await client.zremrangebyrank(feed_key, 0, -801)
    
    async def get_feed(self, user_id: str, limit: int = 50) -> list:
        """Get user's feed."""
        client = self._get_client(user_id)
        feed_key = f"feed:{user_id}"
        
        return await client.zrevrange(feed_key, 0, limit - 1)
    
    async def fan_out(self, post: dict, follower_ids: List[str]):
        """Fan out post to multiple followers (potentially different shards)."""
        post_id = str(post['id'])
        timestamp = post['created_at'].timestamp()
        
        # Group followers by shard
        shard_followers = defaultdict(list)
        for follower_id in follower_ids:
            shard_id = self._get_shard(follower_id)
            shard_followers[shard_id].append(follower_id)
        
        # Execute fan-out per shard (parallel)
        tasks = []
        for shard_id, followers in shard_followers.items():
            client = self.redis_clients[shard_id]
            tasks.append(
                self._fan_out_to_shard(client, followers, post_id, timestamp)
            )
        
        await asyncio.gather(*tasks)
    
    async def _fan_out_to_shard(
        self,
        client,
        follower_ids: List[str],
        post_id: str,
        timestamp: float
    ):
        """Fan out to followers in a single shard."""
        pipe = client.pipeline()
        
        for follower_id in follower_ids:
            feed_key = f"feed:{follower_id}"
            pipe.zadd(feed_key, {post_id: timestamp})
            pipe.zremrangebyrank(feed_key, 0, -801)
        
        await pipe.execute()

# Feed Warming for Active Users

class FeedWarmer:
    """
    Pre-populate feeds for users likely to be active.
    
    Run periodically (e.g., before peak hours) to ensure
    active users have warm feed caches.
    """
    
    def __init__(self, feed_service, redis_client, db_client):
        self.feed_service = feed_service
        self.redis = redis_client
        self.db = db_client
    
    async def warm_active_users(self, hours_ahead: int = 2):
        """Warm feeds for users predicted to be active soon."""
        
        # Get users active at this time yesterday/last week
        predicted_active = await self._predict_active_users(hours_ahead)
        
        logger.info(f"Warming feeds for {len(predicted_active)} users")
        
        # Warm each user's feed
        for user_id in predicted_active:
            try:
                await self._warm_user_feed(user_id)
            except Exception as e:
                logger.warning(f"Failed to warm feed for {user_id}: {e}")
    
    async def _predict_active_users(self, hours_ahead: int) -> List[str]:
        """Predict users who will be active in the next N hours."""
        
        # Simple heuristic: users active at same time yesterday
        return await self.db.fetch_column(
            """
            SELECT DISTINCT user_id 
            FROM user_sessions
            WHERE created_at BETWEEN 
                NOW() - INTERVAL '25 hours' AND 
                NOW() - INTERVAL '23 hours'
            LIMIT 100000
            """
        )
    
    async def _warm_user_feed(self, user_id: str):
        """Populate a user's feed cache."""
        
        # Check if feed already cached
        feed_key = f"feed:{user_id}"
        exists = await self.redis.exists(feed_key)
        
        if exists:
            return  # Already warm
        
        # Compute and cache feed
        feed = await self.feed_service.get_feed(user_id, limit=100)
        
        if feed:
            pipe = self.redis.pipeline()
            for post in feed:
                pipe.zadd(feed_key, {str(post['id']): post['created_at'].timestamp()})
            pipe.expire(feed_key, 86400)
            await pipe.execute()

# Production Feed System - Complete Implementation

import asyncio
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Set, Optional
from collections import defaultdict
import math

logger = logging.getLogger(__name__)


# =============================================================================
# Configuration
# =============================================================================

@dataclass
class ProductionFeedConfig:
    """Complete configuration for production feed system."""
    
    # Celebrity threshold
    celebrity_threshold: int = 10000
    
    # Cache settings
    feed_cache_size: int = 800
    feed_cache_ttl: int = 86400  # 24 hours
    post_cache_ttl: int = 3600   # 1 hour
    
    # Fan-out settings
    async_fanout_threshold: int = 1000  # Use async for > 1000 followers
    fanout_batch_size: int = 1000
    
    # Ranking settings
    enable_ranking: bool = True
    recency_weight: float = 0.3
    engagement_weight: float = 0.3
    affinity_weight: float = 0.3
    content_weight: float = 0.1
    recency_half_life_hours: float = 6.0
    
    # Performance settings
    max_following_to_pull: int = 50  # Max celebrities to pull per request
    candidate_multiplier: int = 3   # Fetch 3x posts for ranking


# =============================================================================
# Metrics
# =============================================================================

@dataclass
class FeedMetrics:
    """Metrics for feed system."""
    feed_requests: int = 0
    posts_created: int = 0
    fanouts_sync: int = 0
    fanouts_async: int = 0
    celebrity_pulls: int = 0
    cache_hits: int = 0
    cache_misses: int = 0
    avg_feed_latency_ms: float = 0
    
    def to_dict(self) -> dict:
        return {
            'feed_requests': self.feed_requests,
            'posts_created': self.posts_created,
            'fanouts': {'sync': self.fanouts_sync, 'async': self.fanouts_async},
            'celebrity_pulls': self.celebrity_pulls,
            'cache_hit_ratio': self.cache_hits / max(1, self.cache_hits + self.cache_misses),
            'avg_feed_latency_ms': round(self.avg_feed_latency_ms, 2)
        }


# =============================================================================
# Main Feed Service
# =============================================================================

class ProductionFeedService:
    """
    Production-ready feed service.
    
    Features:
    - Hybrid push/pull model
    - Celebrity handling
    - Relevance ranking
    - Sharding support
    - Comprehensive metrics
    """
    
    def __init__(
        self,
        redis_client,
        db_client,
        queue_client,
        config: ProductionFeedConfig = None
    ):
        self.redis = redis_client
        self.db = db_client
        self.queue = queue_client
        self.config = config or ProductionFeedConfig()
        self.metrics = FeedMetrics()
        
        # Celebrity cache (in-memory for speed)
        self._celebrities: Set[str] = set()
        self._celebrities_loaded = False
    
    async def initialize(self):
        """Initialize service (call on startup)."""
        await self._load_celebrities()
    
    # =========================================================================
    # Post Creation
    # =========================================================================
    
    async def create_post(
        self,
        user_id: str,
        content: str,
        media_type: str = None
    ) -> dict:
        """
        Create a post and distribute to followers.
        
        Uses appropriate fan-out strategy based on follower count.
        """
        self.metrics.posts_created += 1
        
        # Save post
        post = await self._save_post(user_id, content, media_type)
        
        # Determine fan-out strategy
        is_celebrity = await self._is_celebrity(user_id)
        
        if is_celebrity:
            # Celebrity: Index for pull, no fan-out
            await self._index_celebrity_post(post)
            logger.debug(f"Celebrity post indexed: {post['id']}")
        else:
            # Regular user: Fan out
            follower_count = await self._get_follower_count(user_id)
            
            if follower_count < self.config.async_fanout_threshold:
                # Small fan-out: synchronous
                await self._fan_out_sync(post)
                self.metrics.fanouts_sync += 1
            else:
                # Large fan-out: asynchronous via queue
                await self._fan_out_async(post)
                self.metrics.fanouts_async += 1
        
        return post
    
    async def _save_post(
        self,
        user_id: str,
        content: str,
        media_type: str
    ) -> dict:
        """Save post to database and cache."""
        post = await self.db.fetch_one(
            """
            INSERT INTO posts (user_id, content, media_type, created_at)
            VALUES ($1, $2, $3, NOW())
            RETURNING id, user_id, content, media_type, created_at,
                      0 as likes, 0 as comments, 0 as shares
            """,
            user_id, content, media_type
        )
        
        post_dict = self._post_to_dict(post)
        
        # Cache post
        await self.redis.setex(
            f"post:{post['id']}",
            self.config.post_cache_ttl,
            json.dumps(post_dict, default=str)
        )
        
        # Add to user's posts
        await self.redis.zadd(
            f"user_posts:{user_id}",
            {str(post['id']): post['created_at'].timestamp()}
        )
        
        return post_dict
    
    async def _index_celebrity_post(self, post: dict):
        """Index celebrity post for pull queries."""
        await self.redis.zadd(
            f"celebrity_posts:{post['user_id']}",
            {str(post['id']): post['created_at'].timestamp()}
        )
        await self.redis.zremrangebyrank(
            f"celebrity_posts:{post['user_id']}",
            0, -101  # Keep last 100
        )
    
    async def _fan_out_sync(self, post: dict):
        """Synchronous fan-out for small follower counts."""
        followers = await self._get_followers(post['user_id'])
        
        if not followers:
            return
        
        post_id = str(post['id'])
        timestamp = post['created_at'].timestamp()
        
        # Batch updates
        pipe = self.redis.pipeline()
        for follower_id in followers:
            feed_key = f"feed:{follower_id}"
            pipe.zadd(feed_key, {post_id: timestamp})
            pipe.zremrangebyrank(feed_key, 0, -self.config.feed_cache_size - 1)
        
        await pipe.execute()
    
    async def _fan_out_async(self, post: dict):
        """Asynchronous fan-out via message queue."""
        await self.queue.send("fanout-jobs", {
            "post_id": str(post['id']),
            "user_id": post['user_id'],
            "timestamp": post['created_at'].timestamp()
        })
    
    # =========================================================================
    # Feed Retrieval
    # =========================================================================
    
    async def get_feed(
        self,
        user_id: str,
        limit: int = 50,
        offset: int = 0,
        ranked: bool = None
    ) -> List[dict]:
        """
        Get user's feed.
        
        Combines pre-computed feed (regular users) with
        pulled celebrity posts, optionally ranked.
        """
        start_time = datetime.now(timezone.utc)
        self.metrics.feed_requests += 1
        
        ranked = ranked if ranked is not None else self.config.enable_ranking
        
        # Get more candidates if ranking
        fetch_limit = limit * self.config.candidate_multiplier if ranked else limit
        
        # 1. Get pre-computed feed
        precomputed = await self._get_precomputed_feed(
            user_id, fetch_limit, offset
        )
        
        # 2. Get celebrity posts
        followed_celebrities = await self._get_followed_celebrities(user_id)
        
        celebrity_posts = []
        if followed_celebrities:
            # Limit celebrities to pull
            celebs_to_pull = followed_celebrities[:self.config.max_following_to_pull]
            celebrity_posts = await self._pull_celebrity_posts(celebs_to_pull)
            self.metrics.celebrity_pulls += len(celebs_to_pull)
        
        # 3. Merge
        all_posts = precomputed + celebrity_posts
        
        # 4. Deduplicate
        seen = set()
        unique_posts = []
        for post in all_posts:
            if post['id'] not in seen:
                seen.add(post['id'])
                unique_posts.append(post)
        
        # 5. Rank or sort chronologically
        if ranked and len(unique_posts) > limit:
            affinity = await self._get_affinity_scores(user_id)
            result = self._rank_posts(unique_posts, affinity)[:limit]
        else:
            unique_posts.sort(key=lambda p: p['created_at'], reverse=True)
            result = unique_posts[:limit]
        
        # Record latency
        latency_ms = (datetime.now(timezone.utc) - start_time).total_seconds() * 1000
        self._update_latency(latency_ms)
        
        return result
    
    async def _get_precomputed_feed(
        self,
        user_id: str,
        limit: int,
        offset: int
    ) -> List[dict]:
        """Get pre-computed feed from cache."""
        feed_key = f"feed:{user_id}"
        
        post_ids = await self.redis.zrevrange(
            feed_key,
            offset,
            offset + limit - 1
        )
        
        if post_ids:
            self.metrics.cache_hits += 1
            return await self._get_posts_by_ids(post_ids)
        
        self.metrics.cache_misses += 1
        return []
    
    async def _pull_celebrity_posts(
        self,
        celebrity_ids: List[str]
    ) -> List[dict]:
        """Pull recent posts from celebrities."""
        all_posts = []
        
        # Parallel fetch
        tasks = [
            self._get_celebrity_recent_posts(celeb_id)
            for celeb_id in celebrity_ids
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for result in results:
            if isinstance(result, list):
                all_posts.extend(result)
        
        return all_posts
    
    async def _get_celebrity_recent_posts(
        self,
        user_id: str,
        limit: int = 20
    ) -> List[dict]:
        """Get recent posts from a celebrity."""
        post_ids = await self.redis.zrevrange(
            f"celebrity_posts:{user_id}",
            0, limit - 1
        )
        
        if not post_ids:
            return []
        
        return await self._get_posts_by_ids(post_ids)
    
    # =========================================================================
    # Ranking
    # =========================================================================
    
    def _rank_posts(
        self,
        posts: List[dict],
        affinity_scores: Dict[str, float]
    ) -> List[dict]:
        """Rank posts by relevance."""
        scored = []
        
        for post in posts:
            score = self._calculate_score(post, affinity_scores)
            scored.append((score, post))
        
        scored.sort(key=lambda x: x[0], reverse=True)
        return [post for _, post in scored]
    
    def _calculate_score(
        self,
        post: dict,
        affinity_scores: Dict[str, float]
    ) -> float:
        """Calculate relevance score for a post."""
        c = self.config
        
        # Recency score
        age_hours = (datetime.now(timezone.utc) - post['created_at']).total_seconds() / 3600
        recency = 0.5 ** (age_hours / c.recency_half_life_hours)
        
        # Engagement score (log scale, normalized)
        engagement = post.get('likes', 0) + post.get('comments', 0) * 2
        engagement_score = min(1.0, math.log(engagement + 1) / 10)
        
        # Affinity score
        affinity = affinity_scores.get(str(post['user_id']), 0.1)
        
        # Content score
        content = 0.5
        if post.get('media_type') == 'image':
            content = 0.7
        elif post.get('media_type') == 'video':
            content = 0.9
        
        return (
            c.recency_weight * recency +
            c.engagement_weight * engagement_score +
            c.affinity_weight * affinity +
            c.content_weight * content
        )
    
    async def _get_affinity_scores(self, user_id: str) -> Dict[str, float]:
        """Get user's affinity scores."""
        cache_key = f"affinity:{user_id}"
        
        cached = await self.redis.hgetall(cache_key)
        if cached:
            return {k: float(v) for k, v in cached.items()}
        
        # Calculate from interactions
        rows = await self.db.fetch(
            """
            SELECT target_user_id, COUNT(*) as cnt
            FROM user_interactions
            WHERE user_id = $1 AND created_at > NOW() - INTERVAL '30 days'
            GROUP BY target_user_id
            """,
            user_id
        )
        
        if not rows:
            return {}
        
        max_cnt = max(r['cnt'] for r in rows)
        scores = {str(r['target_user_id']): r['cnt'] / max_cnt for r in rows}
        
        if scores:
            await self.redis.hset(cache_key, mapping={k: str(v) for k, v in scores.items()})
            await self.redis.expire(cache_key, 3600)
        
        return scores
    
    # =========================================================================
    # Helpers
    # =========================================================================
    
    async def _is_celebrity(self, user_id: str) -> bool:
        """Check if user is a celebrity."""
        if not self._celebrities_loaded:
            await self._load_celebrities()
        
        return user_id in self._celebrities
    
    async def _load_celebrities(self):
        """Load celebrity set from Redis."""
        celebrities = await self.redis.smembers("celebrities")
        self._celebrities = set(celebrities) if celebrities else set()
        self._celebrities_loaded = True
    
    async def _get_follower_count(self, user_id: str) -> int:
        """Get follower count (cached)."""
        cached = await self.redis.get(f"follower_count:{user_id}")
        if cached:
            return int(cached)
        
        count = await self.db.fetch_val(
            "SELECT COUNT(*) FROM follows WHERE followee_id = $1",
            user_id
        )
        
        await self.redis.setex(f"follower_count:{user_id}", 300, str(count))
        
        # Update celebrity status
        if count >= self.config.celebrity_threshold:
            await self.redis.sadd("celebrities", user_id)
            self._celebrities.add(user_id)
        
        return count
    
    async def _get_followers(self, user_id: str) -> List[str]:
        """Get list of follower IDs."""
        rows = await self.db.fetch(
            "SELECT follower_id FROM follows WHERE followee_id = $1",
            user_id
        )
        return [str(row['follower_id']) for row in rows]
    
    async def _get_followed_celebrities(self, user_id: str) -> List[str]:
        """Get celebrities that user follows."""
        following = await self.redis.smembers(f"following:{user_id}")
        
        if not following:
            rows = await self.db.fetch(
                "SELECT followee_id FROM follows WHERE follower_id = $1",
                user_id
            )
            following = {str(row['followee_id']) for row in rows}
            
            if following:
                await self.redis.sadd(f"following:{user_id}", *following)
                await self.redis.expire(f"following:{user_id}", 300)
        
        return [uid for uid in following if uid in self._celebrities]
    
    async def _get_posts_by_ids(self, post_ids: List[str]) -> List[dict]:
        """Fetch posts by IDs with caching."""
        posts = []
        to_fetch = []
        
        # Try cache
        pipe = self.redis.pipeline()
        for post_id in post_ids:
            pipe.get(f"post:{post_id}")
        cached = await pipe.execute()
        
        for i, (post_id, data) in enumerate(zip(post_ids, cached)):
            if data:
                posts.append((i, json.loads(data)))
            else:
                to_fetch.append((i, post_id))
        
        # Fetch missing
        if to_fetch:
            ids = [pid for _, pid in to_fetch]
            rows = await self.db.fetch(
                "SELECT * FROM posts WHERE id = ANY($1)",
                ids
            )
            
            row_map = {str(r['id']): self._post_to_dict(r) for r in rows}
            
            pipe = self.redis.pipeline()
            for idx, post_id in to_fetch:
                if post_id in row_map:
                    post = row_map[post_id]
                    posts.append((idx, post))
                    pipe.setex(
                        f"post:{post_id}",
                        self.config.post_cache_ttl,
                        json.dumps(post, default=str)
                    )
            
            await pipe.execute()
        
        posts.sort(key=lambda x: x[0])
        return [post for _, post in posts]
    
    def _post_to_dict(self, row) -> dict:
        """Convert database row to dict."""
        post = dict(row)
        if isinstance(post.get('created_at'), str):
            post['created_at'] = datetime.fromisoformat(post['created_at'])
        return post
    
    def _update_latency(self, latency_ms: float):
        """Update average latency metric."""
        n = self.metrics.feed_requests
        old_avg = self.metrics.avg_feed_latency_ms
        self.metrics.avg_feed_latency_ms = old_avg + (latency_ms - old_avg) / n
    
    def get_metrics(self) -> dict:
        """Get service metrics."""
        return self.metrics.to_dict()