Designa Rate Limiter och Twitter Feed

import time
from collections import deque

class SlidingWindowLog:
    def __init__(self, limit, window_seconds):
        self.limit = limit
        self.window = window_seconds
        self.logs = {}     # user_id -> deque of timestamps

    def allow(self, user_id):
        now = time.time()
        if user_id not in self.logs:
            self.logs[user_id] = deque()
        log = self.logs[user_id]
        window_start = now - self.window
        # Remove expired timestamps
        while log and log[0] <= window_start:
            log.popleft()
        # Check limit
        if len(log) < self.limit:
            log.append(now)
            return True
        return False

limiter = SlidingWindowLog(limit=3, window_seconds=10)
for i in range(5):
    allowed = limiter.allow('user123')
    print(f'Request {i+1}: {"ALLOWED" if allowed else "REJECTED"}')
    time.sleep(0.5)

import time
import math

class SlidingWindowCounter:
    def __init__(self, limit, window_seconds=60):
        self.limit = limit
        self.window = window_seconds
        self.buckets = {}   # user_id -> {prev_count, curr_count, curr_window_start}

    def allow(self, user_id):
        now = time.time()
        window_start = int(now // self.window) * self.window

        if user_id not in self.buckets or self.buckets[user_id]['window'] < window_start - self.window:
            self.buckets[user_id] = {'prev': 0, 'curr': 0, 'window': window_start}
        elif self.buckets[user_id]['window'] < window_start:
            self.buckets[user_id] = {'prev': self.buckets[user_id]['curr'], 'curr': 0, 'window': window_start}

        b = self.buckets[user_id]
        fraction = (now - window_start) / self.window
        estimated = b['prev'] * (1 - fraction) + b['curr']

        if estimated < self.limit:
            b['curr'] += 1
            return True
        return False

limiter = SlidingWindowCounter(limit=5, window_seconds=10)
for i in range(7):
    print(f'Request {i+1}: {"OK" if limiter.allow("user1") else "RATE LIMITED"}')
    time.sleep(0.3)

# Distributed rate limiting with Redis (pseudocode / simulation)

# Fixed window counter using Redis INCR + EXPIRE
def redis_fixed_window(redis_client, user_id, limit, window_sec):
    key = f'rl:{user_id}:{int(time.time() // window_sec)}'
    count = redis_client.incr(key)          # atomic increment
    if count == 1:
        redis_client.expire(key, window_sec)  # set TTL on first request
    return count <= limit

# Sliding window with sorted set
def redis_sliding_window(redis_client, user_id, limit, window_sec):
    now = time.time()
    key = f'rl:{user_id}'
    # Remove old entries, count recent, add current
    # Atomic with Lua: multi-step operation
    lua_script = '''
    local key = KEYS[1]
    local now = ARGV[1]
    local window = ARGV[2]
    local limit = ARGV[3]
    redis.call('ZREMRANGEBYSCORE', key, '-inf', now - window)
    local count = redis.call('ZCARD', key)
    if count < tonumber(limit) then
        redis.call('ZADD', key, now, now)
        redis.call('EXPIRE', key, window)
        return 1  -- allowed
    end
    return 0  -- rejected
    '''
    print('Redis Lua script ensures atomicity across ZREM + ZCARD + ZADD')

# Twitter feed requirements and estimates
reqs = {
    'Functional': [
        'Post tweet (text, image, video)',
        'Follow/unfollow users',
        'View home feed (tweets from followees, newest first)',
        'View user timeline (all tweets by one user)',
        'Like and retweet',
        'Search tweets (basic keyword)',
    ],
    'Non-functional': [
        '300M DAU, 500M tweets/day => 5800 writes/sec',
        '100:1 read:write => 580K feed reads/sec',
        'Feed load < 2 seconds (p95)',
        '99.99% availability',
        'Tweets retained indefinitely (tweets never deleted by default)',
    ],
    'Estimates': [
        'Storage: 500M tweets * 300B = 150 GB/day, 54 TB/year',
        'Media: separate object store (S3), CDN-served',
        'Feed cache: 300M users * top-100-tweets * 100B = 3 TB (hot feeds in Redis)',
    ],
}
for category, items in reqs.items():
    print(f'{category}:')
    for item in items: print(f'  - {item}')
    print()

# Fan-out on write (push model)
fan_out_steps = [
    '1. User posts tweet => write to tweets table (source of truth)',
    '2. Publish event to message queue (Kafka topic: tweet-created)',
    '3. Fan-out workers consume from queue:',
    '   a. Fetch list of followers from follows table',
    '   b. For each follower: LPUSH feed:{follower_id} tweet_id',
    '   c. Trim feed to last 1000 tweets: LTRIM feed:{follower_id} 0 999',
    '4. Feed read: LRANGE feed:{user_id} 0 99 => hydrate tweet_ids => response',
]
for step in fan_out_steps:
    print(step)

print('\nPros:')
print('  - Feed reads are O(1): just read from Redis list')
print('  - Feed is always sorted by recency automatically')
print('\nCons:')
print('  - Celebrities with 100M followers => 100M Redis writes per tweet')
print('  - Fan-out lag: followers may see tweet 10-30 seconds late at peak')
print('  - Inactive users waste Redis storage for precomputed feeds')

# Hybrid fan-out implementation sketch
CELEBRITY_THRESHOLD = 1_000_000   # followers > 1M => celebrity

def on_post_tweet(user_id, tweet_id, follower_count):
    if follower_count <= CELEBRITY_THRESHOLD:
        # Fan-out to all followers (async via Kafka)
        print(f'User {user_id}: fan-out tweet {tweet_id} to {follower_count} followers')
        # => queue to fan-out workers
    else:
        print(f'Celebrity {user_id}: tweet {tweet_id} stored in timeline only')
        # => only write to tweets table + user timeline
        # => followers get it on demand when reading feed

def get_home_feed(user_id, followees):
    # 1. Get precomputed feed (fan-out on write tweets)
    precomputed = f'LRANGE feed:{user_id} 0 499'   # up to 500 tweets

    # 2. Find celebrity followees
    celebrity_followees = [u for u in followees if is_celebrity(u)]

    # 3. Fetch recent tweets from celebrities (fan-out on read)
    celebrity_tweets = []
    for celeb in celebrity_followees:
        tweets = f'GET tweets WHERE user_id={celeb} ORDER BY created_at DESC LIMIT 20'
        celebrity_tweets.extend(tweets)

    # 4. Merge and sort by recency
    combined = merge_and_sort(precomputed, celebrity_tweets)
    return combined[:100]

print('on_post_tweet for regular user:')
on_post_tweet('user123', 'tweet_abc', 500)
print('on_post_tweet for celebrity:')
on_post_tweet('celebrity456', 'tweet_xyz', 50_000_000)

# Twitter architecture summary
architecture = '''
[User] --> [API Gateway + Load Balancer]
                |
    +-----------+-----------+
    |           |           |
[Tweet Svc] [Feed Svc]  [Follow Svc]
    |           |           |
[Cassandra] [Redis Feeds] [Graph DB]
    |           |
[Kafka] <-- [Fan-out
    |        Workers]
[S3 + CDN]  (tweet_ids
(media)      => follower
             feed lists)

Key design choices:
- Tweets stored in Cassandra (PRIMARY KEY (user_id, created_at))
- Feed stored in Redis as list of tweet_ids per user (LPUSH/LTRIM/LRANGE)
- Fan-out via Kafka + workers (decoupled, retryable, scalable)
- Hybrid: regular users = push; celebrities = pull-on-read
- Hydration: tweet_ids -> full tweet objects via Cassandra read
'''
print(architecture)

# Rate limit response headers
def build_rate_limit_headers(limit, remaining, reset_timestamp, retry_after=None):
    headers = {
        'X-RateLimit-Limit': str(limit),
        'X-RateLimit-Remaining': str(max(0, remaining)),
        'X-RateLimit-Reset': str(int(reset_timestamp)),
    }
    if retry_after is not None:
        headers['Retry-After'] = str(retry_after)
    return headers

import time

# Simulated response for allowed request
headers = build_rate_limit_headers(
    limit=100,
    remaining=73,
    reset_timestamp=time.time() + 45
)
print('Allowed request headers:')
for k, v in headers.items():
    print(f'  {k}: {v}')

# Rate limited response
headers_429 = build_rate_limit_headers(
    limit=100,
    remaining=0,
    reset_timestamp=time.time() + 30,
    retry_after=30
)
print('\n429 Too Many Requests headers:')
for k, v in headers_429.items():
    print(f'  {k}: {v}')

# Algorithm comparison matrix
comparison = [
    ('Token Bucket',          'Allows bursts',     'O(1)',         'Most APIs, default choice'),
    ('Leaky Bucket',          'Smooth output rate', 'O(1)',         'Traffic shaping, message queues'),
    ('Fixed Window Counter',  'Very simple',        'O(1)',         'Low-traffic, approximate OK'),
    ('Sliding Window Log',    'Most accurate',      'O(requests)', 'High-accuracy, low traffic'),
    ('Sliding Window Counter','Approximate+fast',   'O(1)',         'High-traffic, Cloudflare-style'),
]
print(f'{'Algorithm':30s} {'Burst Handling':20s} {'Memory':15s} {'Use Case'}')
print('-'*85)
for name, burst, mem, use in comparison:
    print(f'{name:30s} {burst:20s} {mem:15s} {use}')