Social Media Platform

# post_service.py from cassandra.cluster import Cluster from datetime import datetime import uuid class PostService: def __init__(self, contact_points): cluster = Cluster(contact_points) self.session = cluster.connect('socialhub') def create_post(self, user_id, username, content, media_urls=[], hashtags=[]): """Create a new post""" post_id = uuid.uuid1() # TIMEUUID created_at = datetime.now() # 1. Insert into posts_by_id (canonical) self.session.execute(""" INSERT INTO posts_by_id (post_id, user_id, username, content, media_urls, hashtags, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) """, (post_id, user_id, username, content, media_urls, set(hashtags), created_at)) # 2. Insert into posts_by_user (profile) self.session.execute(""" INSERT INTO posts_by_user (user_id, created_at, post_id, content, media_urls, like_count, comment_count) VALUES (?, ?, ?, ?, ?, 0, 0) """, (user_id, created_at, post_id, content, media_urls)) # 3. Insert into posts_by_hashtag (for each hashtag) for hashtag in hashtags: self.session.execute(""" INSERT INTO posts_by_hashtag (hashtag, created_at, post_id, user_id, username, content, media_urls, like_count) VALUES (?, ?, ?, ?, ?, ?, ?, 0) """, (hashtag.lower(), created_at, post_id, user_id, username, content, media_urls)) # 4. Fan-out to followers (async/background job) # self.fan_out_to_followers(user_id, post_id, created_at, username, content, media_urls) return { 'post_id': post_id, 'created_at': created_at } def get_post(self, post_id): """Get post details""" row = self.session.execute( "SELECT * FROM posts_by_id WHERE post_id = ?", (post_id,) ).one() if row: return { 'post_id': row.post_id, 'user_id': row.user_id, 'username': row.username, 'content': row.content, 'media_urls': row.media_urls, 'hashtags': list(row.hashtags), 'created_at': row.created_at, 'like_count': row.like_count, 'comment_count': row.comment_count } return None def get_user_posts(self, user_id, limit=20): """Get user's timeline (their posts)""" rows = self.session.execute( "SELECT * FROM posts_by_user WHERE user_id = ? LIMIT ?", (user_id, limit) ) return [ { 'post_id': row.post_id, 'content': row.content, 'media_urls': row.media_urls, 'created_at': row.created_at, 'like_count': row.like_count, 'comment_count': row.comment_count } for row in rows ] def get_posts_by_hashtag(self, hashtag, limit=50): """Discover posts by hashtag""" rows = self.session.execute( "SELECT * FROM posts_by_hashtag WHERE hashtag = ? LIMIT ?", (hashtag.lower(), limit) ) return [ { 'post_id': row.post_id, 'user_id': row.user_id, 'username': row.username, 'content': row.content, 'created_at': row.created_at, 'like_count': row.like_count } for row in rows ] # Usage post_service = PostService(['localhost']) # Create post post = post_service.create_post( user_id=user_id, username='alice', content='Learning Cassandra for social media! #cassandra #database', media_urls=['https://...'], hashtags=['cassandra', 'database'] )

# social_graph_service.py class SocialGraphService: def __init__(self, session): self.session = session def follow_user(self, follower_id, follower_username, following_id, following_username): """Follow a user""" followed_at = datetime.now() # 1. Add to followers_by_user (following_id's followers) self.session.execute(""" INSERT INTO followers_by_user (user_id, follower_id, follower_username, followed_at) VALUES (?, ?, ?, ?) """, (following_id, follower_id, follower_username, followed_at)) # 2. Add to following_by_user (follower_id's following) self.session.execute(""" INSERT INTO following_by_user (user_id, following_id, following_username, followed_at) VALUES (?, ?, ?, ?) """, (follower_id, following_id, following_username, followed_at)) # 3. Increment counters self.session.execute( "UPDATE users SET follower_count = follower_count + 1 WHERE user_id = ?", (following_id,) ) self.session.execute( "UPDATE users SET following_count = following_count + 1 WHERE user_id = ?", (follower_id,) ) def unfollow_user(self, follower_id, following_id): """Unfollow a user""" # Delete from both tables self.session.execute( "DELETE FROM followers_by_user WHERE user_id = ? AND follower_id = ?", (following_id, follower_id) ) self.session.execute( "DELETE FROM following_by_user WHERE user_id = ? AND following_id = ?", (follower_id, following_id) ) # Decrement counters self.session.execute( "UPDATE users SET follower_count = follower_count - 1 WHERE user_id = ?", (following_id,) ) self.session.execute( "UPDATE users SET following_count = following_count - 1 WHERE user_id = ?", (follower_id,) ) def get_followers(self, user_id, limit=100): """Get user's followers""" rows = self.session.execute( "SELECT * FROM followers_by_user WHERE user_id = ? LIMIT ?", (user_id, limit) ) return [ { 'follower_id': row.follower_id, 'follower_username': row.follower_username, 'followed_at': row.followed_at } for row in rows ] def get_following(self, user_id, limit=100): """Get users this user follows""" rows = self.session.execute( "SELECT * FROM following_by_user WHERE user_id = ? LIMIT ?", (user_id, limit) ) return [ { 'following_id': row.following_id, 'following_username': row.following_username, 'followed_at': row.followed_at } for row in rows ]

# timeline_service.py class TimelineService: def __init__(self, session): self.session = session def fan_out_to_followers(self, author_id, post_id, created_at, author_username, content, media_urls): """ Fan-out post to all followers' timelines This is the MAGIC of social media at scale! """ # 1. Get all followers followers = self.get_all_followers(author_id) # 2. Insert into each follower's timeline for follower in followers: self.session.execute(""" INSERT INTO timeline_by_user (user_id, created_at, post_id, author_id, author_username, content, media_urls, like_count) VALUES (?, ?, ?, ?, ?, ?, ?, 0) """, ( follower['follower_id'], created_at, post_id, author_id, author_username, content, media_urls )) print(f"Fanned out to {len(followers)} followers") def get_home_timeline(self, user_id, limit=50): """ Get user's home timeline This is FAST - single partition query! """ rows = self.session.execute( "SELECT * FROM timeline_by_user WHERE user_id = ? LIMIT ?", (user_id, limit) ) return [ { 'post_id': row.post_id, 'author_id': row.author_id, 'author_username': row.author_username, 'content': row.content, 'media_urls': row.media_urls, 'created_at': row.created_at, 'like_count': row.like_count } for row in rows ] def get_all_followers(self, user_id): """Get ALL followers for fan-out""" # In production, use pagination or streaming rows = self.session.execute( "SELECT follower_id FROM followers_by_user WHERE user_id = ?", (user_id,) ) return [{'follower_id': r.follower_id} for r in rows] # IMPORTANT: Fan-out considerations # - For celebrities with millions of followers, fan-out is expensive # - Solution 1: Hybrid (fan-out for normal users, on-demand for celebrities) # - Solution 2: Async job queue (Kafka/RabbitMQ) # - Solution 3: Rate limit fan-out, prioritize active followers