Design Twitter/X

async def fanout_tweet(tweet_id, author_id, db, cache, queue):
    # Author-এর follower list নাও
    followers = await db.get_followers(author_id)

    # Hybrid Fanout Logic
    if len(followers) < 10_000:
        # Normal user: push to all followers' timeline cache
        for follower_id in followers:
            cache.lpush(f"timeline:{follower_id}", tweet_id)
            cache.ltrim(f"timeline:{follower_id}", 0, 799)  # last 800 tweets
    else:
        # Celebrity: push to Kafka, workers will process async
        # At read time, celebrity tweets are fetched separately
        await queue.publish("celebrity-fanout", {
            "tweet_id": tweet_id,
            "author_id": author_id
        })

    # Tweet always saved to Tweet DB
    await db.save_tweet(tweet_id, author_id)

async def get_home_timeline(user_id, db, cache):
    # Step 1: User-এর pre-built timeline cache থেকে নাও
    cached = cache.lrange(f"timeline:{user_id}", 0, 19)

    # Step 2: Celebrity following list check
    celebrities = cache.smembers(f"following_celebrities:{user_id}")

    # Step 3: Celebrity tweets আলাদাভাবে fetch করুন
    celebrity_tweets = []
    for celeb_id in celebrities:
        tweets = await db.get_recent_tweets(celeb_id, limit=5)
        celebrity_tweets.extend(tweets)

    # Step 4: Merge করুন, sort by timestamp
    all_tweet_ids = cached + [t.id for t in celebrity_tweets]
    all_tweet_ids.sort(key=lambda x: x.created_at, reverse=True)

    return all_tweet_ids[:20]  # Top 20 tweets

-- Tweet Table — Cassandra Schema
CREATE TABLE tweets (
    tweet_id    TIMEUUID,
    user_id     UUID,
    content     TEXT,
    media_url   TEXT,
    like_count  COUNTER,
    created_at  TIMESTAMP,
    PRIMARY KEY ((user_id), created_at, tweet_id)
) WITH CLUSTERING ORDER BY (created_at DESC);

-- Follow Graph — Redis Sorted Set
-- Key: "followers:{user_id}" → Value: follower_id, Score: follow_timestamp
ZADD followers:user123 1700000000 "follower_id_456"

-- Timeline Cache — Redis List
-- Key: "timeline:{user_id}" → Values: tweet_ids (most recent first)
LPUSH timeline:user123 "tweet_id_789"
LTRIM timeline:user123 0 799   -- Keep last 800 tweets only

import redis
import time

r = redis.Redis()

def record_hashtag(hashtag: str):
    """Tweet post হলে hashtag count increment করুন"""
    current_hour = int(time.time() // 3600)  # sliding window key
    key = f"trending:{current_hour}"

    # Atomic increment
    r.zincrby(key, 1, hashtag)

    # Expire after 2 hours (sliding window)
    r.expire(key, 7200)

def get_trending(top_n: int = 10) -> list:
    """Last 1 hour-এর top trending hashtags"""
    current_hour = int(time.time() // 3600)
    prev_hour = current_hour - 1

    # Merge last 2 windows for sliding effect
    r.zunionstore(
        "trending:current",
        [f"trending:{current_hour}", f"trending:{prev_hour}"]
    )

    # Top N descending by score
    return r.zrevrange("trending:current", 0, top_n - 1, withscores=True)

def like_tweet_atomic(tweet_id: str):
    """Race condition ছাড়া like count increment"""
    # Redis INCR is atomic — no race condition!
    r.incr(f"likes:{tweet_id}")
    return r.get(f"likes:{tweet_id}")