Skip to content

355. Design Twitter (Medium)

Problem

Design a simplified Twitter where each user can post tweets, follow/unfollow users, and view the 10 most recent tweets in their news feed (including from themselves and followed users).

Required methods:

  • postTweet(userId, tweetId)
  • follow(followerId, followeeId)
  • unfollow(followerId, followeeId)
  • getNewsFeed(userId), return the 10 most-recent tweet IDs from the user and those they follow, most recent first.

LeetCode 355 · Link · Medium

Try it yourself

idle

Starter code: this editor begins with intentional TODOs. Fill the function, run the embedded tests, then compare your solution with the worked approaches below.

Click Run Python to execute. First run downloads Python (~10 MB, cached after that).

Approach 1: Brute force, merge all tweets and sort

Per-user tweet list; on getNewsFeed, concatenate relevant lists and sort by timestamp.

from collections import defaultdict
class Twitter:
def __init__(self):
self.time = 0
self.tweets = defaultdict(list) # user -> list[(time, tweetId)]
self.follows = defaultdict(set) # follower -> set(followees)
def postTweet(self, userId, tweetId):
self.tweets[userId].append((self.time, tweetId)) # L1: O(1) append
self.time += 1
def follow(self, followerId, followeeId):
self.follows[followerId].add(followeeId) # L2: O(1) set add
def unfollow(self, followerId, followeeId):
self.follows[followerId].discard(followeeId) # L3: O(1) set discard
def getNewsFeed(self, userId):
users = self.follows[userId] | {userId}
feed = []
for u in users:
feed.extend(self.tweets[u]) # L4: O(T) collect all
feed.sort(reverse=True) # L5: O(T log T) sort
return [t for _, t in feed[:10]]

Where the time goes, line by line

Variables: k = number of users being followed, T = total tweets across those users.

LinePer-call costTimes executedContribution
L1-L3 (postTweet/follow/unfollow)O(1)O(1)1 per callO(1)O(1)
L4 (collect all tweets)O(T)O(T)1 per getNewsFeedO(T)O(T)
L5 (sort feed)O(TlogT)O(T log T)1 per getNewsFeedO(TlogT)O(T log T) ← dominates getNewsFeed

Complexity

  • getNewsFeed: O(TlogT)O(T log T) where T is total tweets across followed users (L5).
  • Others: O(1)O(1).

Approach 2: Per-user sorted list + k-way merge with a heap (optimal)

Store each user’s tweets as a list (append-only, so it’s naturally sorted by time). On getNewsFeed, k-way-merge the latest 10 from each followed user using a max-heap of size at most k + 1.

from collections import defaultdict
import heapq
class Twitter:
def __init__(self):
self.time = 0
self.tweets = defaultdict(list) # user -> list[(time, tweetId)]
self.follows = defaultdict(set)
def postTweet(self, userId, tweetId):
self.tweets[userId].append((self.time, tweetId)) # L1: O(1)
self.time += 1
def follow(self, followerId, followeeId):
self.follows[followerId].add(followeeId) # L2: O(1)
def unfollow(self, followerId, followeeId):
self.follows[followerId].discard(followeeId) # L3: O(1)
def getNewsFeed(self, userId):
users = self.follows[userId] | {userId}
heap = []
# Seed the heap with the latest tweet from each user
for u in users: # L4: O(k) seed
if self.tweets[u]:
i = len(self.tweets[u]) - 1
t, tid = self.tweets[u][i]
heap.append((-t, u, i, tid))
heapq.heapify(heap) # L5: O(k) heapify
feed = []
while heap and len(feed) < 10: # L6: at most 10 iters
neg_t, u, i, tid = heapq.heappop(heap) # L7: O(log k) pop
feed.append(tid)
if i > 0:
i -= 1
t2, tid2 = self.tweets[u][i]
heapq.heappush(heap, (-t2, u, i, tid2)) # L8: O(log k) push
return feed

Where the time goes, line by line

Variables: k = number of users being followed (heap size at most k + 1).

LinePer-call costTimes executedContribution
L1-L3 (postTweet/follow/unfollow)O(1)O(1)1 per callO(1)O(1)
L4-L5 (seed + heapify)O(k)O(k)1 per getNewsFeedO(k)O(k)
L6-L8 (drain 10 from heap)O(logk)O(log k) each10O(10logk)O(10 log k) = O(logk)O(log k) ← dominates getNewsFeed

The heap never exceeds k + 1 entries. Each of the 10 pops and up to 10 pushes costs O(logk)O(log k). The seed phase at L4-L5 is O(k)O(k).

Complexity

  • getNewsFeed: O(k+10logk)O(k + 10 log k) = O(k)O(k) for seeding + O(logk)O(log k) for draining (L4-L8).
  • Others: O(1)O(1) amortized.

Same pattern as problem 23 (Merge k Sorted Lists).

Try this approach:

idle
Click Run Python to execute. First run downloads Python (~10 MB, cached after that).

Approach 3: Cap per-user feed; global merge

If each user only ever needs the last 10 tweets, truncate the per-user list to length 10 (a rolling window). Then merge at most 10 · k tweets on getNewsFeed.

Complexity

  • getNewsFeed: O(10k)O(10 · k).
  • Others: O(1)O(1).

Lower memory if you don’t need long-term history.

private struct TweetRecord { let time: Int; let id: Int }
private struct FeedCursor { let user: Int; let index: Int; let tweet: TweetRecord }
final class Twitter {
private var clock = 0
private var tweets: [Int: [TweetRecord]] = [:]
private var following: [Int: Set<Int>] = [:]
init() {}
func postTweet(_ userId: Int, _ tweetId: Int) { clock += 1; tweets[userId, default: []].append(TweetRecord(time: clock, id: tweetId)); if tweets[userId]!.count > 10 { tweets[userId]!.removeFirst() } }
func getNewsFeed(_ userId: Int) -> [Int] {
var users = following[userId, default: []]; users.insert(userId)
var heap = BinaryHeap<FeedCursor> { $0.tweet.time > $1.tweet.time }
for user in users { if let list = tweets[user], let tweet = list.last { heap.insert(FeedCursor(user: user, index: list.count - 1, tweet: tweet)) } }
var feed: [Int] = []
while feed.count < 10, let cursor = heap.removeRoot() { feed.append(cursor.tweet.id); let next = cursor.index - 1; if next >= 0, let tweet = tweets[cursor.user]?[next] { heap.insert(FeedCursor(user: cursor.user, index: next, tweet: tweet)) } }
return feed
}
func follow(_ followerId: Int, _ followeeId: Int) { following[followerId, default: []].insert(followeeId) }
func unfollow(_ followerId: Int, _ followeeId: Int) { following[followerId]?.remove(followeeId) }
}

Test cases

# Quick smoke tests, paste into a REPL or save as test_355.py and run.
# Uses the heap k-way merge approach (Approach 2).
import heapq
from collections import defaultdict
class Twitter:
def __init__(self):
self.time = 0
self.tweets = defaultdict(list)
self.follows = defaultdict(set)
def postTweet(self, userId, tweetId):
self.tweets[userId].append((self.time, tweetId))
self.time += 1
def follow(self, followerId, followeeId):
self.follows[followerId].add(followeeId)
def unfollow(self, followerId, followeeId):
self.follows[followerId].discard(followeeId)
def getNewsFeed(self, userId):
users = self.follows[userId] | {userId}
heap = []
for u in users:
if self.tweets[u]:
i = len(self.tweets[u]) - 1
t, tid = self.tweets[u][i]
heap.append((-t, u, i, tid))
heapq.heapify(heap)
feed = []
while heap and len(feed) < 10:
neg_t, u, i, tid = heapq.heappop(heap)
feed.append(tid)
if i > 0:
i -= 1
t2, tid2 = self.tweets[u][i]
heapq.heappush(heap, (-t2, u, i, tid2))
return feed
def _run_tests():
t = Twitter()
t.postTweet(1, 5)
assert t.getNewsFeed(1) == [5]
t.follow(1, 2)
t.postTweet(2, 6)
# user 1 follows user 2; most recent is tweet 6, then 5
assert t.getNewsFeed(1) == [6, 5]
t.unfollow(1, 2)
# after unfollow, user 1 only sees their own tweet
assert t.getNewsFeed(1) == [5]
# User sees own tweets even without explicit self-follow
t2 = Twitter()
for i in range(12):
t2.postTweet(1, i)
feed = t2.getNewsFeed(1)
assert len(feed) == 10
assert feed == list(range(11, 1, -1)) # most recent 10: 11,10,...,2
print("all tests pass")
if __name__ == "__main__":
_run_tests()

Summary

ApproachgetNewsFeedpostTweetNotes
Collect all + sortO(TlogT)O(T log T)O(1)O(1)Simplest
Heap k-way mergeO(10logk)O(10 log k)O(1)O(1)Canonical answer
Rolling per-user windowO(10k)O(10 · k)O(1)O(1)If history truncation is OK

The k-way merge pattern is the right generalization: it’s the same code you’d use to implement Kafka partition consumers, merged log readers, or any “top-N across ordered streams” system.

  • Heap and Priority Queue, priority-frontier tactics for repeatedly extracting the smallest, largest, or most urgent item.
  • K-way Merge, multi-stream ordering tactics for combining several sorted sources through one priority queue.
  • Simulation, state-machine tactics for faithfully executing rules while keeping state small and explicit.