Max Concurrent Calls Per Customer Per Day

Role: Software Engineer

Problem Overview

You are building a billing system for a calling platform. Sales reps use the platform throughout the day to make phone calls to prospects, and your company wants to bill customers based on their peak calling load — the maximum number of concurrent calls.

You are given a list of phone call records. Each record contains:

customerId — unique identifier for a customer (one customer may have many reps making concurrent calls)
callId — unique identifier for a single phone call
startTimestamp — when the call started (UNIX timestamp in milliseconds, inclusive)
endTimestamp — when the call ended (UNIX timestamp in milliseconds, exclusive)

For each customer, for each UTC date, find the maximum number of concurrent calls that occurred at any point during that date.

Notes

The startTimestamp is inclusive and the endTimestamp is exclusive — if call A ends at timestamp 123 and call B starts at 123, they did not overlap.
A single call may span multiple UTC dates.
If no calls occurred for a customer on a given date, omit that date from results.

Input Format

python

call_records = [
    {
        "customerId": 123,
        "callId": "abc-001",
        "startTimestamp": 1707314726000,  # milliseconds
        "endTimestamp":   1707317769000
    },
    ...
]

Output Format

For each (customerId, date) pair, return:

python

results = [
    {
        "customerId": 123,
        "date": "2024-02-07",          # UTC date, YYYY-MM-DD
        "maxConcurrentCalls": 3,
        "timestamp": 1707315000000,    # any timestamp during the peak window (< endTimestamp of all calls in callIds)
        "callIds": ["abc-001", ...]    # the calls overlapping at timestamp, len == maxConcurrentCalls
    }
]

Example

Given these calls for one customer:

text

Call A: [t=1, t=5)
Call B: [t=2, t=6)
Call C: [t=3, t=4)
Call D: [t=7, t=9)

The peak is 3 (calls A, B, C overlap at t=3). maxConcurrentCalls = 3, timestamp can be any value in [3, 4).

Approach

Step 1: Split Calls by UTC Date Boundary

A call spanning midnight needs to be counted for each day it covers. Split each call record at UTC midnight boundaries into per-day segments (keeping the original callId).

Step 2: Group by (customerId, date)

After splitting, group all call segments by (customerId, date).

Step 3: Sweep Line for Max Overlap

For each group, use a sweep line approach:

Create events: +1 at each startTimestamp, -1 at each endTimestamp
Sort events by timestamp (break ties: ends before starts, since end is exclusive)
Track the running count and record the maximum

python

from datetime import datetime, timezone, timedelta
from collections import defaultdict

def get_utc_date(ts_ms: int) -> str:
    return datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc).strftime("%Y-%m-%d")

def midnight_utc_after(ts_ms: int) -> int:
    """Returns the next UTC midnight timestamp (ms) after ts_ms."""
    dt = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
    next_day = (dt + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
    return int(next_day.timestamp() * 1000)

def split_by_date(record: dict) -> list[dict]:
    """Split a call record into per-UTC-day segments."""
    segments = []
    start = record["startTimestamp"]
    end = record["endTimestamp"]
    call_id = record["callId"]
    customer_id = record["customerId"]

    while True:
        midnight = midnight_utc_after(start)
        if midnight >= end:
            # Call ends within the same day
            segments.append({
                "customerId": customer_id,
                "callId": call_id,
                "startTimestamp": start,
                "endTimestamp": end,
                "date": get_utc_date(start)
            })
            break
        else:
            # Call crosses midnight — split at midnight
            segments.append({
                "customerId": customer_id,
                "callId": call_id,
                "startTimestamp": start,
                "endTimestamp": midnight,
                "date": get_utc_date(start)
            })
            start = midnight

    return segments

def max_concurrent_calls(call_records: list[dict]) -> list[dict]:
    # Step 1: Split calls by UTC date boundary
    all_segments = []
    for record in call_records:
        all_segments.extend(split_by_date(record))

    # Step 2: Group by (customerId, date)
    groups = defaultdict(list)
    for seg in all_segments:
        groups[(seg["customerId"], seg["date"])].append(seg)

    results = []

    # Step 3: Sweep line per group
    for (customer_id, date), segments in groups.items():
        # Build events: (timestamp, type, call_id)
        # type: 0 = end (exclusive), 1 = start
        # Sort: same timestamp → ends before starts (since end is exclusive)
        events = []
        for seg in segments:
            events.append((seg["startTimestamp"], 1, seg["callId"]))
            events.append((seg["endTimestamp"], 0, seg["callId"]))

        events.sort(key=lambda e: (e[0], e[1]))

        active = set()
        max_count = 0
        max_ts = None
        max_call_ids = []

        for ts, event_type, call_id in events:
            if event_type == 1:
                active.add(call_id)
                if len(active) > max_count:
                    max_count = len(active)
                    max_ts = ts
                    max_call_ids = list(active)
            else:
                active.discard(call_id)

        results.append({
            "customerId": customer_id,
            "date": date,
            "maxConcurrentCalls": max_count,
            "timestamp": max_ts,
            "callIds": max_call_ids
        })

    return results

Complexity Analysis

	Complexity
Time	O(N log N) per (customerId, date) group — dominated by sorting events
Space	O(N) for segments and event lists

Where N = total number of call records (after date splitting).

Key Edge Cases

Call ends exactly when another starts — not concurrent (end is exclusive, end events sort before start events at the same timestamp)
Multi-day calls — must be split at UTC midnight boundaries and counted separately for each date
Multiple peak windows — any valid timestamp within any peak window is acceptable
Empty groups — if no calls exist for a (customerId, date), omit from output

Follow-Up Questions

What if calls can be arbitrarily long (months)? The splitting approach still works but generates more segments.
How would you scale this to billions of records? Use distributed aggregation (e.g., MapReduce/Spark grouping by customerId, then date, then sweep line per partition).
What if you need the result in real time as calls start/end? Use a priority queue keyed by endTimestamp; when a new call starts, pop all calls that have ended and recompute the count.