Design an in-memory service for a food-delivery platform to compute driver payouts and provide basic live analytics.
0 < (endTime - startTime) ≤ 3 hours.A delivery pays:
[Source: darkinterview.com]
Example: A driver paid $10.00/hour completes a 1 hour 30 minute delivery → payout = $15.00.
Implement the following methods:
class DeliveryBillingSystem:
def add_driver(self, driver_id: int, usd_hourly_rate: float) -> None:
"""
Registers a new driver with their hourly rate.
The driver will not already exist in the system.
"""
pass
def record_delivery(self, driver_id: int, start_time: int, end_time: int) -> None:
"""
Records a completed delivery for an existing driver.
- Times are Unix epoch seconds
- Delivery has already completed (end_time is in the past)
- Driver is guaranteed to exist
- 0 < (end_time - start_time) <= 10800 (3 hours max)
"""
pass
def get_total_cost(self) -> float:
"""
Returns the total accumulated payout across all recorded deliveries.
This is displayed on a live dashboard - should be O(1).
"""
pass
system = DeliveryBillingSystem()
# Add drivers with different rates
system.add_driver(1, 10.00) # Driver 1: $10/hour
system.add_driver(2, 15.00) # Driver 2: $15/hour
# Record deliveries (timestamps in epoch seconds)
system.record_delivery(1, 1000, 4600) # 1 hour → $10
system.record_delivery(1, 5000, 6800) # 30 min → $5
system.record_delivery(2, 2000, 5600) # 1 hour → $15
print(system.get_total_cost()) # Output: 30.00
get_total_cost() should not iterate all deliveries.Key insight: Maintain a running total to make get_total_cost() O(1). [Source: darkinterview.com]
from dataclasses import dataclass, field
@dataclass
class Driver:
driver_id: int
hourly_rate: float
deliveries: list = field(default_factory=list) # For Part 2/3
class DeliveryBillingSystem:
def __init__(self):
self.drivers: dict[int, Driver] = {}
self.total_cost: float = 0.0
def add_driver(self, driver_id: int, usd_hourly_rate: float) -> None:
self.drivers[driver_id] = Driver(driver_id, usd_hourly_rate)
def record_delivery(self, driver_id: int, start_time: int, end_time: int) -> None:
driver = self.drivers[driver_id]
duration_hours = (end_time - start_time) / 3600.0
payout = driver.hourly_rate * duration_hours
# Store for Part 2/3
driver.deliveries.append({
'start_time': start_time,
'end_time': end_time,
'payout': payout,
'paid': False
})
# O(1) total cost tracking
self.total_cost += payout
def get_total_cost(self) -> float:
return .total_cost
Time & Space Complexity:
| Operation | Time | Space |
|---|---|---|
add_driver | O(1) | O(1) |
record_delivery | O(1) | O(1) |
get_total_cost | O(1) | O(1) |
Add support to mark deliveries as paid when they finish before a cutoff time.
def pay_up_to(self, end_time: int) -> None:
"""
Marks as PAID every delivery with delivery.end_time <= end_time.
- A delivery can only be paid once (no double-counting)
- Calls to pay_up_to are guaranteed to have increasing end_time values
"""
pass
def get_unpaid_amount(self) -> float:
"""
Returns total unpaid amount = (total cost) - (total paid so far).
Should be O(1) for the live dashboard.
"""
pass
system = DeliveryBillingSystem()
system.add_driver(1, 10.00) # $10/hour
# Record three 1-hour deliveries ($10 each) with different end times
system.record_delivery(1, 0, 3600) # ends at 3600, payout = $10
system.record_delivery(1, 7200, 10800) # ends at 10800, payout = $10
system.record_delivery(1, 14400, 18000) # ends at 18000, payout = $10
print(system.get_unpaid_amount()) # $30.00 (all unpaid)
system.pay_up_to(5000) # Pay deliveries ending at or before 5000
print(system.get_unpaid_amount()) # $20.00 (first delivery paid)
system.pay_up_to(12000) # Pay deliveries ending at or before 12000
print(system.get_unpaid_amount()) # $10.00 (first two paid)
pay_up_to is called with strictly increasing times.Key insight: Since pay_up_to is called with increasing times, we can use a min-heap ordered by end_time to efficiently find and process unpaid deliveries. [Source: darkinterview.com]
import heapq
from dataclasses import dataclass, field
from typing import Optional
@dataclass(order=True)
class Delivery:
end_time: int
start_time: int = field(compare=False)
driver_id: int = field(compare=False)
payout: float = field(compare=False)
paid: bool = field(default=False, compare=False)
class DeliveryBillingSystem:
def __init__(self):
self.drivers: dict[int, float] = {} # driver_id -> hourly_rate
self.total_cost: float = 0.0
self.total_paid: float = 0.0
self.unpaid_heap: list[Delivery] = [] # min-heap by end_time
self.all_deliveries: list[Delivery] = [] # For Part 3
def add_driver(self, driver_id: int, usd_hourly_rate: float) -> None:
self.drivers[driver_id] = usd_hourly_rate
def record_delivery(self, driver_id: int, start_time: int, end_time: ) -> :
hourly_rate = .drivers[driver_id]
duration_hours = (end_time - start_time) /
payout = hourly_rate * duration_hours
delivery = Delivery(
end_time=end_time,
start_time=start_time,
driver_id=driver_id,
payout=payout
)
heapq.heappush(.unpaid_heap, delivery)
.all_deliveries.append(delivery)
.total_cost += payout
() -> :
.total_cost
() -> :
.unpaid_heap .unpaid_heap[].end_time <= cutoff_time:
delivery = heapq.heappop(.unpaid_heap)
delivery.paid:
delivery.paid =
.total_paid += delivery.payout
() -> :
.total_cost - .total_paid
Time & Space Complexity:
| Operation | Time | Space |
|---|---|---|
record_delivery | O(log D) | O(1) |
pay_up_to | O(K log D) | O(1) |
get_unpaid_amount | O(1) | O(1) |
Where D = total deliveries, K = deliveries processed in this call.
Amortized: Each delivery is added to and removed from the heap exactly once, so across all pay_up_to calls, total work is O(D log D). [Source: darkinterview.com]
Compute how many distinct drivers were simultaneously active in a time window.
def max_simultaneous_drivers_in_past_24_hours(self, now: int) -> int:
"""
Returns the maximum number of distinct drivers who were active
at the same time in the past 24 hours: [now - 86400, now).
Key considerations:
- A driver counts as 1 at any instant, even with multiple overlapping deliveries
- Must handle the window [now - 24h, now)
"""
pass
system = DeliveryBillingSystem()
system.add_driver(1, 10.00)
system.add_driver(2, 12.00)
system.add_driver(3, 15.00)
# Timeline (assuming now = 10000):
# Driver 1: |---delivery 1---| |---delivery 2---|
# 1000 3000 5000 7000
#
# Driver 2: |--------delivery 1--------|
# 2000 6000
#
# Driver 3: |----delivery 1----|
# 4000 8000
# At time 2500: Driver 1 and 2 are active → 2 drivers
# At time 5500: Driver 1, 2, and 3 are active → 3 drivers (max)
# At time 7500: Only Driver 3 is active → 1 driver
system.record_delivery(1, 1000, 3000)
system.record_delivery(1, 5000, 7000)
system.record_delivery(2, 2000, 6000)
system.record_delivery(3, 4000, 8000)
print(system.max_simultaneous_drivers_in_past_24_hours(10000)) # Output: 3
Unlike typical interval overlap problems, we must count unique drivers, not deliveries.
Algorithm: [Source: darkinterview.com]
from collections import defaultdict
from typing import List, Tuple
class DeliveryBillingSystem:
# ... (Part 1 and Part 2 methods) ...
def max_simultaneous_drivers_in_past_24_hours(self, now: int) -> int:
window_start = now - 86400 # 24 hours in seconds
window_end = now
# Step 1: Group deliveries by driver, filtered to window
driver_intervals: dict[int, list[tuple[int, int]]] = defaultdict(list)
for delivery in self.all_deliveries:
# Check overlap with [window_start, window_end)
if delivery.end_time <= window_start or delivery.start_time >= window_end:
continue
# Clip to window bounds
clipped_start = max(delivery.start_time, window_start)
clipped_end = min(delivery.end_time, window_end)
driver_intervals[delivery.driver_id].append((clipped_start, clipped_end))
# Step 2: Merge overlapping intervals for each driver
def merge_intervals(intervals: list[tuple[int, int]]) -> list[tuple[int, int]]:
if not intervals:
return []
intervals.sort()
merged = [intervals[]]
start, end intervals[:]:
start <= merged[-][]:
merged[-] = (merged[-][], (merged[-][], end))
:
merged.append((start, end))
merged
events = []
driver_id, intervals driver_intervals.items():
merged = merge_intervals(intervals)
start, end merged:
events.append((start, , driver_id))
events.append((end, , driver_id))
events.sort()
active_drivers = ()
max_drivers =
time, event_type, driver_id events:
event_type == :
active_drivers.add(driver_id)
max_drivers = (max_drivers, (active_drivers))
:
active_drivers.discard(driver_id)
max_drivers
At the same timestamp, if a driver ends one delivery and starts another, we should:
This handles the half-open interval [start, end) semantics correctly. It also avoids double-counting if one driver's delivery ends exactly when another starts.
Example: If driver 1 has delivery ending at time 100 and driver 2 has delivery starting at time 100: [Source: darkinterview.com]
| Step | Time | Space |
|---|---|---|
| Filter deliveries | O(D) | O(D) |
| Merge per driver | O(D log D) | O(D) |
| Build events | O(D) | O(D) |
| Sort events | O(D log D) | O(1) |
| Sweep line | O(D) | O(N) |
Total: O(D log D) time, O(D) space, where D = deliveries in window, N = unique drivers.
import heapq
from collections import defaultdict
from dataclasses import dataclass, field
from typing import List, Tuple
@dataclass(order=True)
class Delivery:
end_time: int
start_time: int = field(compare=False)
driver_id: int = field(compare=False)
payout: float = field(compare=False)
paid: bool = field(default=False, compare=False)
class DeliveryBillingSystem:
def __init__(self):
self.drivers: dict[int, float] = {}
self.total_cost: float = 0.0
self.total_paid: float = 0.0
self.unpaid_heap: list[Delivery] = []
self.all_deliveries: list[Delivery] = []
# ========== Part 1: Core Billing ==========
def add_driver(self, driver_id: int, usd_hourly_rate: float) -> None:
self.drivers[driver_id] = usd_hourly_rate
def record_delivery(self, driver_id: , start_time: , end_time: ) -> :
hourly_rate = .drivers[driver_id]
duration_hours = (end_time - start_time) /
payout = hourly_rate * duration_hours
delivery = Delivery(
end_time=end_time,
start_time=start_time,
driver_id=driver_id,
payout=payout
)
heapq.heappush(.unpaid_heap, delivery)
.all_deliveries.append(delivery)
.total_cost += payout
() -> :
.total_cost
() -> :
.unpaid_heap .unpaid_heap[].end_time <= cutoff_time:
delivery = heapq.heappop(.unpaid_heap)
delivery.paid:
delivery.paid =
.total_paid += delivery.payout
() -> :
.total_cost - .total_paid
() -> :
window_start = now -
window_end = now
driver_intervals: [, [[, ]]] = defaultdict()
delivery .all_deliveries:
delivery.end_time <= window_start delivery.start_time >= window_end:
clipped_start = (delivery.start_time, window_start)
clipped_end = (delivery.end_time, window_end)
driver_intervals[delivery.driver_id].append((clipped_start, clipped_end))
():
intervals:
[]
intervals.sort()
merged = [intervals[]]
s, e intervals[:]:
s <= merged[-][]:
merged[-] = (merged[-][], (merged[-][], e))
:
merged.append((s, e))
merged
events = []
driver_id, intervals driver_intervals.items():
start, end merge(intervals):
events.append((start, , driver_id))
events.append((end, , driver_id))
events.sort()
active = ()
max_count =
_, event_type, driver_id events:
event_type == :
active.add(driver_id)
max_count = (max_count, (active))
:
active.discard(driver_id)
max_count
Driver-level queries: Add get_driver_cost(driver_id) and get_driver_unpaid(driver_id).
Return peak interval: Instead of just the max count, return the time interval(s) when the peak occurred. [Source: darkinterview.com]
Sliding window optimization: If max_simultaneous_drivers_in_past_24_hours is called frequently, how would you optimize it?
Concurrent delivery limit: How would you add a rule that a driver can't have more than N concurrent deliveries?
| Approach | record_delivery | max_drivers_24h |
|---|---|---|
| Store all, compute on query | O(log D) | O(D log D) |
| Maintain sorted events | O(log D) | O(D) |
| Pre-computed sliding window | O(log D) + update | O(1) |
The right choice depends on read vs. write patterns and how often analytics are queried. [Source: darkinterview.com]