Per-user chat event counts in a sliding window: Frontier Interview's solution

Question: Per-user chat event counts in a sliding window

Companies: OpenAI

Difficulty: Medium

Problem statement

A chat service runs for months without restarting. It sends you a stream of message events. Each event has a user_id, a chat_id and an integer timestamp in seconds. You must answer, for any user, how many of their events fall in the last m seconds.

Implement ChatEventAggregator:

ChatEventAggregator(m: int): m > 0 is the window length in seconds.
record(user_id: str, chat_id: str, timestamp: int) -> None: store one event.
get_count(user_id: str, timestamp: int) -> int: return the number of events by user_id with timestamp in the inclusive window [timestamp - m + 1, timestamp]. Return 0 for a user with no events.

Guarantees

For one user, record timestamps strictly increase.
Time never goes backwards: every call (record or query) has a timestamp greater than or equal to the largest timestamp of any earlier call.

Memory requirement

The process runs indefinitely and sees millions of users. Memory must stay proportional to the events in the most recent m seconds. Discard any data that no future query can use, including all data about users who have gone quiet.

Example

agg = ChatEventAggregator(10)
agg.record("ana", "c1", 1)
agg.record("ana", "c2", 5)
agg.record("bo",  "c1", 6)
agg.record("ana", "c1", 9)
agg.get_count("ana", 10)  # 3, window [1, 10]
agg.get_count("ana", 11)  # 2, window [2, 11]: t=1 is out
agg.get_count("bo", 16)   # 0, window [7, 16]

After get_count("ana", 19) returns 0, the aggregator should hold no data for either user.

Constraints

1 <= m <= 10^6
0 <= timestamp <= 10^9
Up to 10^6 calls in total.

State the time and space cost of each operation.

Approach

This is the hit-counter problem with two changes. Counts are kept per user, and the process runs indefinitely, so memory has to stay proportional to recent activity rather than to all history.

What is dead. Window for a query at t is [t - m + 1, t], so an event at s counts if and only if s > t - m. Agree one assumption with the interviewer: time never goes backwards across calls. Each record and get_count carries a timestamp at least as large as any seen before. Under that assumption, once some call has reached time T, every event with s <= T - m can never be counted again. Call T - m the cutoff. If queries could look into the past, nothing could be discarded, so the assumption is the whole reason the problem is solvable in bounded memory.

Brute force. Keep a list of all timestamps per user and count the ones in the window on each query. That is O(n) per query and memory grows forever. Not acceptable, but say it, then improve it.

Per-user queue. Within a user, timestamps strictly increase, so a deque per user is already sorted. On a record, append on the right. On a record or a query, pop from the left while the head is <= cutoff. The deque then holds exactly the live events, and the count is its length. Each event is appended once and popped at most once, so the work is amortized O(1) per event.

Idle users. The per-user queue only gets trimmed when that user is touched. A user who sends one message and never comes back would keep a deque forever, and with millions of users that is the real leak. Fix it with a global order of users by their latest event. An OrderedDict does this: on each record, move_to_end(user). Because time is globally non-decreasing, the dict is sorted by each user's newest timestamp, so the least recently active users sit at the front. On every call, pop users from the front while their newest event is <= cutoff. Every one of their events is then dead too. Each user leaves at most once per time it was inserted, so this is also amortized O(1).

Invariants after every call at time T:

Every stored user has at least one event with timestamp > T - m.
The dict order matches each user's newest timestamp, oldest first.
The deque of a user touched in this call holds only live events. Other users may still have a few dead entries at the front, but never more than m of them, because their newest event is live and timestamps strictly increase.

So memory is O(number of events in the last m seconds), which is the best possible when you must answer exact counts.

chat_id does not affect the count in the base problem. It is in the API because follow-ups use it (per-chat counts, active chats).

Code

from collections import OrderedDict, deque


class ChatEventAggregator:
    """Per-user count of events in the trailing window [t - m + 1, t].

    Assumes time never goes backwards across calls: every record or query
    carries a timestamp >= the largest one seen so far. That is what makes
    it safe to throw history away.
    """

    def __init__(self, m: int) -> None:
        if m <= 0:
            raise ValueError("m must be positive")
        self.m = m
        self.clock = 0  # largest timestamp seen by any call
        # user -> deque of event timestamps, oldest first.
        # Ordered by each user's latest event, so idle users sit at the front.
        self.users: OrderedDict[str, deque[int]] = OrderedDict()

    def _advance(self, timestamp: int) -> None:
        if timestamp < self.clock:
            raise ValueError("timestamps must not go backwards")
        self.clock = timestamp
        cutoff = timestamp - self.m  # events at or before cutoff are dead
        # Drop users whose newest event is already dead. Amortized O(1):
        # each user is removed at most once per time it was (re)inserted.
        while self.users:
            user, events = next(iter(self.users.items()))
            if events[-1] > cutoff:
                break
            self.users.popitem(last=False)

    @staticmethod
    def _trim(events: deque[int], cutoff: int) -> None:
        while events and events[0] <= cutoff:
            events.popleft()

    def record(self, user_id: str, chat_id: str, timestamp: int) -> None:
        self._advance(timestamp)
        events = self.users.get(user_id)
        if events is None:
            events = deque()
            self.users[user_id] = events
        elif events[-1] >= timestamp:
            raise ValueError("per-user timestamps must strictly increase")
        events.append(timestamp)
        self.users.move_to_end(user_id)
        self._trim(events, timestamp - self.m)

    def get_count(self, user_id: str, timestamp: int) -> int:
        self._advance(timestamp)
        events = self.users.get(user_id)
        if events is None:
            return 0
        self._trim(events, timestamp - self.m)
        return len(events)

    def stored_events(self) -> int:
        """Total timestamps kept in memory (for tests and monitoring)."""
        return sum(len(e) for e in self.users.values())




agg = ChatEventAggregator(10)
agg.record("ana", "c1", 1)
agg.record("ana", "c2", 5)
agg.record("bo", "c1", 6)
agg.record("ana", "c1", 9)
assert agg.get_count("ana", 10) == 3   # window [1, 10]
assert agg.get_count("ana", 11) == 2   # window [2, 11]
assert agg.get_count("bo", 16) == 0    # window [7, 16]
assert agg.get_count("ana", 19) == 0
assert agg.stored_events() == 0        # all history discarded

Complexity

Let k be the number of events from all users in the last m seconds.

record: amortized O(1). Each event is appended once and removed once. Each user entry is evicted at most once per insertion.
get_count: amortized O(1), for the same reason. The worst case for a single call is O(m) (popping one user's dead events), but that cost is paid for by earlier appends.
Space: O(k + live users), and live users <= k. A user with no live event is removed.

If the interviewer wants a hard O(1) worst case per call instead of amortized, use the bucket version (follow-up 1).

Test cases

m = 10. Record (ana, c1, 1), (ana, c2, 5), (bo, c1, 6), (ana, c1, 9). get_count(ana, 10) -> 3. get_count(ana, 11) -> 2, since t = 1 leaves the window [2, 11].
Same data: get_count(bo, 15) -> 1 (window [6, 15]), get_count(bo, 16) -> 0.
Boundary: m = 1, record (u, c, 7). get_count(u, 7) -> 1, get_count(u, 8) -> 0. The window has exactly one second.
Unknown user: get_count(zed, 16) -> 0, and no entry is created for zed.
Memory: after get_count(ana, 19) in the first example, no users and no events are stored.
Many one-time users: m = 5, record (u1, c, 1) through (u10000, c, 10000). Only 5 users stay stored.
Invalid input: a second record for the same user at the same or an earlier timestamp, or any call earlier than the clock, raises ValueError. In an interview, agree with the interviewer whether to raise or ignore.

Follow-ups

Bursts: many events per second. Store (second, count) pairs instead of single timestamps, and add 1 to the last pair when the second repeats. Keep a running total per user so get_count does not need to sum. Memory per user is then at most m pairs, whatever the event rate. For a hard O(1) bound, use the LeetCode 362 ring of m slots per user, indexed by t % m, each storing (second, count). That only pays off when users are dense, because it costs m slots per user.
Count per (user, chat). Key a second structure by (user_id, chat_id) with the same deque and the same idle-eviction order. Alternatively, store (t, chat_id) in the user deque and keep a Counter of live events per chat, decremented on pop. The second option gives both counts from one queue.
Active chats per user (distinct chats with at least one event in the window). Keep the per-chat counter from follow-up 2 and delete a chat when its count reaches 0. The answer is len(counter). If chats can be closed explicitly, a chat is active only when its last event is in the window and after its last close.
Out-of-order events within a lateness bound L. Events can arrive up to L seconds late. Keep cutoff = clock - m - L, so events are not discarded until no late event can still land in a window. Insert late events in sorted order (bisect.insort, or per-second buckets in a dict). A count may change after it is first answered, so define "final" counts as those older than L.
Many processes or machines. Shard by user_id so one user's events and queries reach one node, and the per-user ordering guarantee holds. For approximate counts at very large scale, use sliding-window counters (fixed buckets per second or per minute) or a Count-Min sketch per time bucket. Eviction is then dropping whole buckets.

Interviewer expectations

Mid-level (L4)

Reads the inclusive window correctly: an event at s counts when t - m < s <= t.
Writes the per-user deque with trimming from the left, and gets the boundary right with a test.
States the time and space cost of each operation.

Senior (L5)

Names the assumption that makes discarding safe (time never goes backwards), and asks about it before coding.
Sees the idle-user leak without being told, and fixes it with an order of users by their latest activity, in amortized O(1).
Moves to (second, count) buckets for bursty traffic, and explains amortized versus worst case.

Staff (L6+)

Argues that the memory bound is optimal for exact counts, and describes what changes for approximate counts (bucketed counters, sketches).
Handles late events with a lateness bound, and is clear about which answers can still change.
Discusses sharding by user, recovery after a restart (replay the last m + L seconds from a log), and metrics such as stored events and evictions per second.