Delivery Billing System

Problem Overview

Design an in-memory service for a food-delivery platform to compute driver payouts and provide basic live analytics.

Context

• There are tens of thousands of delivery drivers, each submitting hundreds of deliveries per week.
• Delivery details are sent to the system immediately after completion.
• Each driver has a unique hourly pay rate (can differ by driver).
• Drivers may have multiple concurrent deliveries. Each delivery pays independently—overlapping deliveries earn the sum of payout for each.
• Use efficient data structures. Ignore persistence and thread-safety.

Time Format

• Use Unix epoch seconds (long / 64-bit integer).
• Minimum precision: 1 second.
• Each delivery satisfies: 0 < (endTime - startTime) ≤ 3 hours.

Payment Calculation

A delivery pays:

payout=driverHourlyRate×endTime−startTime3600\text{payout} = \text{driverHourlyRate} \times \frac{\text{endTime} - \text{startTime}}{3600}payout=driverHourlyRate×3600endTime−startTime​

Example: A driver paid $10.00/hour completes a 1 hour 30 minute delivery → payout = $15.00.

---

Part 1: Core Billing API

Problem Statement

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

Example

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

Design Considerations

• Floating-point precision: Consider storing money as integer cents internally.
• Dashboard performance: get_total_cost() should not iterate all deliveries.
• Data structures: What do you need to store per driver?

---

Part 1 Solution

Key insight: Maintain a running total to make get_total_cost() O(1).

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 self.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)

---

Part 2: Payment Tracking

Problem Statement

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

Example

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)

Key Design Points

1. Monotonic cutoff: pay_up_to is called with strictly increasing times.
2. Efficient lookup: Use a data structure that allows finding deliveries by end time.
3. Avoid double-payment: Track which deliveries have been paid.

---

Part 2 Solution

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.

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: int) -> None:
        hourly_rate = self.drivers[driver_id]
        duration_hours = (end_time - start_time) / 3600.0
        payout = hourly_rate * duration_hours
        
        delivery = Delivery(
            end_time=end_time,
            start_time=start_time,
            driver_id=driver_id,
            payout=payout
        )
        
        heapq.heappush(self.unpaid_heap, delivery)
        self.all_deliveries.append(delivery)
        self.total_cost += payout
    
    def get_total_cost(self) -> float:
        return self.total_cost
    
    def pay_up_to(self, cutoff_time: int) -> None:
        """Pop deliveries from heap while end_time <= cutoff_time."""
        while self.unpaid_heap and self.unpaid_heap[0].end_time <= cutoff_time:
            delivery = heapq.heappop(self.unpaid_heap)
            if not delivery.paid:
                delivery.paid = True
                self.total_paid += delivery.payout
    
    def get_unpaid_amount(self) -> float:
        return self.total_cost - self.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).

---

Part 3: Concurrency Analytics

Problem Statement

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

Clarifications

• Active means the driver is performing a delivery (between start_time and end_time).
• A driver with 3 overlapping deliveries still counts as 1 active driver.
• The challenge is distinguishing driver count from delivery count.

Example

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

Key Insight: Merge Intervals Per Driver

Unlike typical interval overlap problems, we must count unique drivers, not deliveries.

Algorithm:

1. Filter deliveries to the 24-hour window
2. For each driver: Merge their overlapping deliveries into disjoint intervals
3. Run a sweep line over all merged intervals
4. Track distinct driver count at each event

---

Part 3 Solution

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[0]]
            for start, end in intervals[1:]:
                if start <= merged[-1][1]:  # Overlapping or adjacent
                    merged[-1] = (merged[-1][0], max(merged[-1][1], end))
                else:
                    merged.append((start, end))
            return merged
        
        # Step 3: Create sweep line events
        # Event: (time, type, driver_id)
        # type: 0 = end, 1 = start (process ends before starts at same time)
        events = []
        
        for driver_id, intervals in driver_intervals.items():
            merged = merge_intervals(intervals)
            for start, end in merged:
                events.append((start, 1, driver_id))   # start event
                events.append((end, 0, driver_id))     # end event
        
        # Sort: by time, then by type (ends before starts at same time)
        events.sort()
        
        # Step 4: Sweep line
        active_drivers = set()
        max_drivers = 0
        
        for time, event_type, driver_id in events:
            if event_type == 1:  # start
                active_drivers.add(driver_id)
                max_drivers = max(max_drivers, len(active_drivers))
            else:  # end
                active_drivers.discard(driver_id)
        
        return max_drivers

Why Process Ends Before Starts?

At the same timestamp, if a driver ends one delivery and starts another, we should:

1. Remove them from active set (end event)
2. Re-add them (start event)

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:

• Process end first → remove driver 1
• Process start → add driver 2
• The count reflects the correct state at time 100

Time & Space Complexity

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.

---

Complete Implementation

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: int, start_time: int, end_time: int) -> None:
        hourly_rate = self.drivers[driver_id]
        duration_hours = (end_time - start_time) / 3600.0
        payout = hourly_rate * duration_hours
        
        delivery = Delivery(
            end_time=end_time,
            start_time=start_time,
            driver_id=driver_id,
            payout=payout
        )
        
        heapq.heappush(self.unpaid_heap, delivery)
        self.all_deliveries.append(delivery)
        self.total_cost += payout
    
    def get_total_cost(self) -> float:
        return self.total_cost
    
    # ========== Part 2: Payment Tracking ==========
    
    def pay_up_to(self, cutoff_time: int) -> None:
        while self.unpaid_heap and self.unpaid_heap[0].end_time <= cutoff_time:
            delivery = heapq.heappop(self.unpaid_heap)
            if not delivery.paid:
                delivery.paid = True
                self.total_paid += delivery.payout
    
    def get_unpaid_amount(self) -> float:
        return self.total_cost - self.total_paid
    
    # ========== Part 3: Concurrency Analytics ==========
    
    def max_simultaneous_drivers_in_past_24_hours(self, now: int) -> int:
        window_start = now - 86400
        window_end = now
        
        # Group by driver and filter to window
        driver_intervals: dict[int, list[tuple[int, int]]] = defaultdict(list)
        for delivery in self.all_deliveries:
            if delivery.end_time <= window_start or delivery.start_time >= window_end:
                continue
            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))
        
        # Merge intervals per driver
        def merge(intervals):
            if not intervals:
                return []
            intervals.sort()
            merged = [intervals[0]]
            for s, e in intervals[1:]:
                if s <= merged[-1][1]:
                    merged[-1] = (merged[-1][0], max(merged[-1][1], e))
                else:
                    merged.append((s, e))
            return merged
        
        # Build sweep line events
        events = []
        for driver_id, intervals in driver_intervals.items():
            for start, end in merge(intervals):
                events.append((start, 1, driver_id))
                events.append((end, 0, driver_id))
        
        events.sort()
        
        # Sweep
        active = set()
        max_count = 0
        for _, event_type, driver_id in events:
            if event_type == 1:
                active.add(driver_id)
                max_count = max(max_count, len(active))
            else:
                active.discard(driver_id)
        
        return max_count

---

Follow-Up Discussion Topics

Extensions

1. Driver-level queries: Add get_driver_cost(driver_id) and get_driver_unpaid(driver_id).

2. Return peak interval: Instead of just the max count, return the time interval(s) when the peak occurred.

3. Sliding window optimization: If max_simultaneous_drivers_in_past_24_hours is called frequently, how would you optimize it?

   • Maintain a sorted structure of events
   • Only process new deliveries since last call
   • Cache intermediate results

4. Concurrent delivery limit: How would you add a rule that a driver can't have more than N concurrent deliveries?

Performance Trade-offs

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.