Low-Level Design: Ride-Sharing Driver App
The driver-side of a ride-sharing app manages driver state, trip offers, earnings tracking, and location reporting. It is a stateful LLD covering state machines, observer pattern, and earnings aggregation.
Core Entities
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime, date
from typing import Optional, Callable
import uuid
class DriverStatus(Enum):
OFFLINE = "offline"
AVAILABLE = "available"
OFFER_PENDING = "offer_pending" # received trip offer, deciding
EN_ROUTE_PICKUP = "en_route_pickup"
WAITING_FOR_RIDER = "waiting_for_rider"
ON_TRIP = "on_trip"
class TripOfferDecision(Enum):
ACCEPTED = "accepted"
DECLINED = "declined"
EXPIRED = "expired"
@dataclass
class Location:
lat: float
lng: float
recorded_at: datetime = field(default_factory=datetime.utcnow)
@dataclass
class TripOffer:
offer_id: str
trip_id: str
pickup_location: Location
destination_location: Location
estimated_distance_km: float
estimated_fare_cents: int
expires_at: datetime
@property
def is_expired(self) -> bool:
return datetime.utcnow() > self.expires_at
@dataclass
class CompletedTrip:
trip_id: str
rider_id: str
pickup: Location
destination: Location
start_time: datetime
end_time: datetime
distance_km: float
fare_cents: int
tip_cents: int = 0
rating: Optional[int] = None # 1-5
@property
def earnings_cents(self) -> int:
# Driver gets 80% of fare + 100% of tip
return int(self.fare_cents * 0.80) + self.tip_cents
@dataclass
class Driver:
driver_id: str
name: str
license_plate: str
vehicle_model: str
rating: float = 5.0
total_trips: int = 0
acceptance_rate: float = 1.0
status: DriverStatus = DriverStatus.OFFLINE
current_location: Optional[Location] = None
Driver State Machine
class DriverApp:
VALID_TRANSITIONS = {
DriverStatus.OFFLINE: {DriverStatus.AVAILABLE},
DriverStatus.AVAILABLE: {DriverStatus.OFFER_PENDING, DriverStatus.OFFLINE},
DriverStatus.OFFER_PENDING: {DriverStatus.EN_ROUTE_PICKUP, DriverStatus.AVAILABLE},
DriverStatus.EN_ROUTE_PICKUP: {DriverStatus.WAITING_FOR_RIDER},
DriverStatus.WAITING_FOR_RIDER:{DriverStatus.ON_TRIP},
DriverStatus.ON_TRIP: {DriverStatus.AVAILABLE},
}
def __init__(self, driver: Driver):
self.driver = driver
self.current_offer: Optional[TripOffer] = None
self.current_trip_id: Optional[str] = None
self._observers: list[Callable] = []
def add_observer(self, callback: Callable) -> None:
self._observers.append(callback)
def _notify(self, event: str, data: dict) -> None:
for obs in self._observers:
obs(event, data)
def _transition(self, new_status: DriverStatus) -> None:
allowed = self.VALID_TRANSITIONS.get(self.driver.status, set())
if new_status not in allowed:
raise ValueError(
f"Cannot transition {self.driver.status} -> {new_status}"
)
old = self.driver.status
self.driver.status = new_status
self._notify("status_changed", {"from": old, "to": new_status,
"driver_id": self.driver.driver_id})
def go_online(self) -> None:
self._transition(DriverStatus.AVAILABLE)
def go_offline(self) -> None:
self._transition(DriverStatus.OFFLINE)
def receive_offer(self, offer: TripOffer) -> None:
self._transition(DriverStatus.OFFER_PENDING)
self.current_offer = offer
self._notify("offer_received", {"offer_id": offer.offer_id,
"fare_cents": offer.estimated_fare_cents})
def respond_to_offer(self, decision: TripOfferDecision) -> None:
if not self.current_offer:
raise ValueError("No pending offer")
if self.current_offer.is_expired:
decision = TripOfferDecision.EXPIRED
if decision == TripOfferDecision.ACCEPTED:
self._transition(DriverStatus.EN_ROUTE_PICKUP)
self.current_trip_id = self.current_offer.trip_id
self._notify("offer_accepted", {"trip_id": self.current_trip_id})
else:
self._transition(DriverStatus.AVAILABLE)
# Update acceptance rate
total = self.driver.total_trips + 1
self.driver.acceptance_rate = (
(self.driver.acceptance_rate * (total - 1) + 0) / total
)
self._notify("offer_declined", {"reason": decision.value})
self.current_offer = None
def arrived_at_pickup(self) -> None:
self._transition(DriverStatus.WAITING_FOR_RIDER)
def start_trip(self) -> None:
self._transition(DriverStatus.ON_TRIP)
self._notify("trip_started", {"trip_id": self.current_trip_id})
def complete_trip(self, fare_cents: int, distance_km: float) -> CompletedTrip:
trip = CompletedTrip(
trip_id=self.current_trip_id,
rider_id="", # filled by system
pickup=self.driver.current_location,
destination=self.driver.current_location,
start_time=datetime.utcnow(),
end_time=datetime.utcnow(),
distance_km=distance_km,
fare_cents=fare_cents,
)
self.driver.total_trips += 1
self.driver.acceptance_rate = (
(self.driver.acceptance_rate * (self.driver.total_trips - 1) + 1)
/ self.driver.total_trips
)
self.current_trip_id = None
self._transition(DriverStatus.AVAILABLE)
self._notify("trip_completed", {"trip_id": trip.trip_id,
"earnings": trip.earnings_cents})
return trip
Earnings Tracker
from collections import defaultdict
class EarningsTracker:
def __init__(self, driver_id: str):
self.driver_id = driver_id
self._daily: dict[date, int] = defaultdict(int) # cents
self._weekly: dict[int, int] = defaultdict(int) # week_number -> cents
self._trips: list[CompletedTrip] = []
def record_trip(self, trip: CompletedTrip) -> None:
self._trips.append(trip)
trip_date = trip.end_time.date()
week = trip_date.isocalendar()[1]
self._daily[trip_date] += trip.earnings_cents
self._weekly[week] += trip.earnings_cents
def daily_earnings(self, day: date = None) -> int:
return self._daily[day or date.today()]
def weekly_earnings(self, week: int = None) -> int:
if week is None:
week = date.today().isocalendar()[1]
return self._weekly[week]
def average_earnings_per_trip(self) -> float:
if not self._trips:
return 0.0
total = sum(t.earnings_cents for t in self._trips)
return total / len(self._trips)
def trips_today(self) -> int:
today = date.today()
return sum(1 for t in self._trips if t.end_time.date() == today)
Location Reporting
import threading
import time
class LocationReporter:
"""Background thread updating driver location every 4 seconds when online."""
def __init__(self, driver_app: DriverApp, location_service):
self.app = driver_app
self.location_service = location_service
self._running = False
self._thread: Optional[threading.Thread] = None
def start(self) -> None:
self._running = True
self._thread = threading.Thread(target=self._report_loop, daemon=True)
self._thread.start()
def stop(self) -> None:
self._running = False
def _report_loop(self) -> None:
while self._running:
if self.app.driver.status != DriverStatus.OFFLINE:
loc = self._get_current_gps()
self.app.driver.current_location = loc
self.location_service.update(
self.app.driver.driver_id,
loc.lat, loc.lng,
self.app.driver.status.value,
)
time.sleep(4)
def _get_current_gps(self) -> Location:
# In production: read from device GPS sensor
return Location(lat=37.7749, lng=-122.4194) # placeholder
Asked at: Uber Interview Guide
Asked at: Lyft Interview Guide
Asked at: DoorDash Interview Guide
Asked at: Snap Interview Guide