# Low-Level Design: Ride-Sharing Driver App (State Machine, Earnings, Location) Source: https://www.techinterview.org/post/3233464508/lld-ride-sharing-driver-app/ Updated: 2026-04-17 ยท techinterview.org ## 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](https://techinterview.org/uber-interview/) Asked at: [Lyft Interview Guide](https://techinterview.org/lyft-interview/) Asked at: [DoorDash Interview Guide](https://techinterview.org/doordash-interview/) Asked at: [Snap Interview Guide](https://techinterview.org/snap-interview/)