Uber / Lyft Ride-Sharing System Design

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

class TripStateMachine:
    """
    Implements trip state transitions with validation and side effects.
    """
    
    VALID_TRANSITIONS = {
        REQUESTED: [MATCHING],
        MATCHING: [MATCHED, NO_DRIVERS],
        MATCHED: [DRIVER_EN_ROUTE, DRIVER_CANCELED, RIDER_CANCELED],
        DRIVER_EN_ROUTE: [DRIVER_ARRIVED, DRIVER_CANCELED, RIDER_CANCELED],
        DRIVER_ARRIVED: [TRIP_STARTED, NO_SHOW],
        TRIP_STARTED: [TRIP_COMPLETED, TRIP_INTERRUPTED],
        TRIP_COMPLETED: [PAYMENT_PROCESSING],
        PAYMENT_PROCESSING: [COMPLETED, PAYMENT_FAILED],
        PAYMENT_FAILED: [COMPLETED],  // After retry or write-off
    }
    
    async function transition(trip, targetState, context):
        currentState = trip.state
        
        // Validate transition is allowed
        if targetState not in VALID_TRANSITIONS[currentState]:
            raise InvalidTransitionError(currentState, targetState)
        
        // State-specific validation
        await validateTransition(trip, targetState, context)
        
        // Begin transaction
        async with database.transaction():
            // Update trip state
            trip.state = targetState
            trip.stateUpdatedAt = now()
            trip.stateHistory.append({
                fromState: currentState,
                toState: targetState,
                timestamp: now(),
                context: context
            })
            
            await database.save(trip)
            
            // Emit event for downstream consumers
            await eventBus.publish(TripStateChangedEvent(
                tripId: trip.id,
                fromState: currentState,
                toState: targetState,
                timestamp: now()
            ))
        
        // Execute side effects (outside transaction)
        await executeSideEffects(trip, currentState, targetState, context)
        
        return trip
    
    async function validateTransition(trip, targetState, context):
        match targetState:
            case TRIP_STARTED:
                # Can only start if driver is at pickup
                if not trip.driverArrivedAt:
                    raise ValidationError("Driver hasn't arrived")
                    
            case TRIP_COMPLETED:
                # Can only complete if trip was started
                if trip.state != TRIP_STARTED:
                    raise ValidationError("Trip not in progress")
                # Must have valid destination reached
                if not context.get("locationConfirmed"):
                    raise ValidationError("Destination not confirmed")
                    
            case NO_SHOW:
                # Driver must have waited minimum time
                waitTime = now() - trip.driverArrivedAt
                if waitTime < MIN_WAIT_FOR_NOSHOW:  # e.g., 5 minutes
                    raise ValidationError(f"Must wait {MIN_WAIT_FOR_NOSHOW - waitTime} more")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

// Event definitions
abstract class TripEvent:
    tripId: UUID
    eventId: UUID  // For idempotency
    timestamp: DateTime
    version: Int   // Sequence number within trip
 
class TripRequestedEvent(TripEvent):
    riderId: UUID
    pickup: Location
    dropoff: Location
    requestedVehicleType: VehicleType
    fareEstimate: Money
 
class DriverAssignedEvent(TripEvent):
    driverId: UUID
    vehicleInfo: Vehicle
    estimatedPickupTime: Duration
 
class DriverArrivedEvent(TripEvent):
    arrivalLocation: Location
    actualPickupTime: Duration
 
class TripStartedEvent(TripEvent):
    startLocation: Location
 
class TripCompletedEvent(TripEvent):
    endLocation: Location
    actualRoute: Polyline
    distanceMeters: Int
    durationSeconds: Int
 
class FareCalculatedEvent(TripEvent):
    baseFare: Money
    distanceCharge: Money
    timeCharge: Money
    surgeMultiplier: Float
    totalFare: Money
 
class PaymentProcessedEvent(TripEvent):
    paymentMethodId: UUID
    amount: Money
    transactionId: String
    status: PaymentStatus
 
// Event store interface
class TripEventStore:
    
    async function append(tripId, event):
        """Append event to trip's event stream."""
        async with database.transaction():
            # Get current version
            currentVersion = await getLatestVersion(tripId)
            
            # Set event version (optimistic concurrency)
            event.version = currentVersion + 1
            
            # Append to event stream
            await database.insert("trip_events", {
                trip_id: tripId,
                event_id: event.eventId,
                event_type: event.getClass().name,
                version: event.version,
                timestamp: event.timestamp,
                payload: event.toJson()
            })
            
            return event.version
    
    async function getEvents(tripId, fromVersion = 0):
        """Get all events for a trip, optionally from a specific version."""
        return await database.query("""
            SELECT * FROM trip_events 
            WHERE trip_id = $1 AND version > $2
            ORDER BY version ASC
        """, [tripId, fromVersion])
    
    async function rebuildState(tripId):
        """Rebuild current trip state by replaying all events."""
        events = await getEvents(tripId)
        trip = Trip()
        
        for event in events:
            trip.apply(event)  # Each event type knows how to update state
        
        return trip

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

class TripProjection:
    """
    Maintains a queryable view of trips updated from events.
    """
    
    @eventHandler(TripRequestedEvent)
    async function onTripRequested(event):
        await database.insert("trips_current", {
            id: event.tripId,
            rider_id: event.riderId,
            status: "REQUESTED",
            pickup: event.pickup,
            dropoff: event.dropoff,
            created_at: event.timestamp
        })
    
    @eventHandler(DriverAssignedEvent)
    async function onDriverAssigned(event):
        await database.update("trips_current",
            where: { id: event.tripId },
            set: {
                driver_id: event.driverId,
                status: "MATCHED"
            }
        )
    
    @eventHandler(TripCompletedEvent)
    async function onTripCompleted(event):
        await database.update("trips_current",
            where: { id: event.tripId },
            set: {
                status: "COMPLETED",
                actual_distance: event.distanceMeters,
                actual_duration: event.durationSeconds,
                end_location: event.endLocation,
                completed_at: event.timestamp
            }
        )
        
        // Also update analytics projection
        await updateRiderTripsAnalytics(event.tripId)
        await updateDriverTripsAnalytics(event.tripId)
 
// Query the projection, not the event store
async function getActiveTripsForCity(cityId):
    return await database.query("""
        SELECT * FROM trips_current 
        WHERE city_id = $1 AND status IN ('REQUESTED', 'MATCHED', 'IN_PROGRESS')
    """, [cityId])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

class PaymentService:
    
    async function authorizeTrip(tripId, riderId, fareEstimate):
        # Get rider's default payment method
        paymentMethod = await getDefaultPaymentMethod(riderId)
        if not paymentMethod:
            raise NoPaymentMethodError()
        
        # Calculate auth amount with buffer
        # Buffer covers: surge increases, route changes, tips, tolls
        bufferPercent = 0.30  # 30% buffer
        authAmount = fareEstimate * (1 + bufferPercent)
        authAmount = roundUp(authAmount, 5)  # Round up to nearest $5
        
        # Create authorization with idempotency key
        idempotencyKey = f"trip-auth-{tripId}"
        
        try:
            authResult = await paymentGateway.authorize(
                paymentMethodId: paymentMethod.id,
                amount: authAmount,
                currency: paymentMethod.currency,
                idempotencyKey: idempotencyKey,
                metadata: {tripId: tripId}
            )
            
            # Store authorization for later capture
            await database.insert("trip_authorizations", {
                trip_id: tripId,
                auth_id: authResult.id,
                auth_amount: authAmount,
                payment_method_id: paymentMethod.id,
                status: "AUTHORIZED",
                created_at: now(),
                expires_at: now() + 7.days
            })
            
            return {success: true, authId: authResult.id}
            
        except InsufficientFundsError:
            return {success: false, reason: "insufficient_funds"}
        except PaymentMethodDeclinedError:
            return {success: false, reason: "card_declined"}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

class FareCalculator:
    
    async function calculateFare(trip):
        # Get pricing rules for city/vehicle type
        pricing = await getPricingRules(trip.cityId, trip.vehicleType)
        
        # Base calculations
        baseFare = pricing.baseFare
        distanceCharge = trip.actualDistanceKm * pricing.perKmRate
        timeCharge = trip.actualDurationMinutes * pricing.perMinuteRate
        
        # Minimum fare
        calculatedFare = baseFare + distanceCharge + timeCharge
        calculatedFare = max(calculatedFare, pricing.minimumFare)
        
        # Surge multiplier (stored at trip start)
        surgeMultiplier = trip.surgeMultiplierAtRequest
        fareAfterSurge = calculatedFare * surgeMultiplier
        
        # Additional charges
        tolls = await calculateTolls(trip.actualRoute)
        airportFee = await getAirportSurchargeIfApplicable(trip)
        
        # Total before promotions
        subtotal = fareAfterSurge + tolls + airportFee
        
        # Apply promotions (rider credits, promo codes)
        promotionDiscount = await applyPromotions(trip.riderId, subtotal)
        
        # Final rider charge
        riderCharge = subtotal - promotionDiscount
        
        # Calculate driver earnings
        platformFee = riderCharge * pricing.platformFeePercent  # e.g., 25%
        driverEarnings = riderCharge - platformFee + tolls  # Tolls pass-through
        
        return FareBreakdown(
            baseFare: baseFare,
            distanceCharge: distanceCharge,
            timeCharge: timeCharge,
            surgeAmount: fareAfterSurge - calculatedFare,
            tolls: tolls,
            airportFee: airportFee,
            promotionDiscount: promotionDiscount,
            riderCharge: riderCharge,
            platformFee: platformFee,
            driverEarnings: driverEarnings
        )

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

class PaymentService:
    
    async function capturePayment(tripId, fareBreakdown):
        # Get authorization
        auth = await database.get("trip_authorizations", tripId)
        
        if auth.status != "AUTHORIZED":
            raise InvalidAuthStateError(auth.status)
        
        if now() > auth.expires_at:
            raise AuthExpiredError()
        
        # Idempotency key for capture
        idempotencyKey = f"trip-capture-{tripId}"
        
        try:
            # Capture the actual amount (releases excess hold automatically)
            captureResult = await paymentGateway.capture(
                authorizationId: auth.auth_id,
                amount: fareBreakdown.riderCharge,
                idempotencyKey: idempotencyKey
            )
            
            # Record successful capture
            await database.update("trip_authorizations", tripId, {
                status: "CAPTURED",
                captured_amount: fareBreakdown.riderCharge,
                captured_at: now()
            })
            
            # Queue driver payout
            await queueDriverPayout(tripId, fareBreakdown.driverEarnings)
            
            # Send receipt
            await sendReceiptEmail(trip, fareBreakdown)
            
            return {success: true, transactionId: captureResult.id}
            
        except PaymentError as e:
            # Handle failure - will retry
            await database.update("trip_authorizations", tripId, {
                status: "CAPTURE_FAILED",
                failure_reason: e.message
            })
            
            # Queue for retry
            await retryQueue.enqueue("payment_capture_retry", {
                tripId: tripId,
                attempt: 1
            }, delay=5.minutes)
            
            return {success: false, reason: e.message}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

class TripFailureHandler:
    
    async function handleDriverDisconnection(tripId, driverId):
        """Handle when driver's app loses connection during active trip."""
        
        trip = await getTrip(tripId)
        
        if trip.state not in [DRIVER_EN_ROUTE, DRIVER_ARRIVED, TRIP_STARTED]:
            return  # Not in active trip phase
        
        # Record disconnection
        await recordEvent(TripEvent.DRIVER_DISCONNECTED, tripId, {
            timestamp: now(),
            lastKnownLocation: getLastDriverLocation(driverId)
        })
        
        # Start grace period timer
        await scheduler.scheduleAt(
            now() + RECONNECTION_GRACE_PERIOD,  # 5 minutes
            "check_driver_reconnection",
            {tripId, driverId}
        )
        
        # Notify rider
        await notifyRider(trip.riderId, 
            "We're having trouble connecting to your driver. " +
            "Trip will continue once connection is restored."
        )
    
    @scheduledHandler("check_driver_reconnection")
    async function checkDriverReconnection(tripId, driverId):
        trip = await getTrip(tripId)
        
        # Check if driver reconnected
        lastUpdate = await getLastLocationUpdate(driverId)
        if lastUpdate.timestamp > now() - 1.minute:
            return  # Driver is back, all good
        
        # Driver still disconnected after grace period
        if trip.state == TRIP_STARTED:
            # Trip was in progress - need resolution
            await escalateToSupport(tripId, "DRIVER_DISCONNECTION_DURING_TRIP")
            # Potentially auto-complete trip at last known location
            
        elif trip.state in [DRIVER_EN_ROUTE, DRIVER_ARRIVED]:
            # Haven't started yet - reassign
            await transitionTrip(tripId, DRIVER_CANCELED, {
                reason: "DRIVER_DISCONNECTION",
                automated: true
            })
            # Re-enter matching to find new driver
            await triggerRematching(tripId)
 
class PaymentRetryHandler:
    MAX_RETRY_ATTEMPTS = 5
    RETRY_DELAYS = [5.minutes, 15.minutes, 1.hour, 6.hours, 24.hours]
    
    @queueHandler("payment_capture_retry")
    async function retryPaymentCapture(tripId, attempt):
        trip = await getTrip(tripId)
        fareBreakdown = await getFareBreakdown(tripId)
        
        result = await paymentService.capturePayment(tripId, fareBreakdown)
        
        if result.success:
            await transitionTrip(tripId, COMPLETED)
            return
        
        if attempt >= MAX_RETRY_ATTEMPTS:
            # Exhausted retries
            await escalateToFinance(tripId, "PAYMENT_CAPTURE_EXHAUSTED", {
                attempts: attempt,
                lastError: result.reason
            })
            # Mark as requires manual resolution
            await markTripForManualResolution(tripId, "PAYMENT_FAILURE")
            return
        
        # Schedule next retry
        nextDelay = RETRY_DELAYS[min(attempt, len(RETRY_DELAYS) - 1)]
        await retryQueue.enqueue("payment_capture_retry", {
            tripId: tripId,
            attempt: attempt + 1
        }, delay=nextDelay)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

class TripCompletionSaga:
    """
    Orchestrates trip completion across multiple services.
    Each step has a compensating action if subsequent steps fail.
    """
    
    steps = [
        Step("calculate_fare", calculateFare, compensate=None),
        Step("capture_payment", capturePayment, compensate=reversePayment),
        Step("queue_driver_payout", queuePayout, compensate=cancelPayout),
        Step("update_trip_status", markCompleted, compensate=revertStatus),
        Step("send_notifications", sendAllNotifications, compensate=None),
        Step("update_analytics", sendAnalytics, compensate=None),
    ]
    
    async function execute(tripId):
        context = {tripId: tripId, completedSteps: []}
        
        for step in steps:
            try:
                result = await step.action(context)
                context[step.name + "_result"] = result
                context.completedSteps.append(step.name)
                
            except StepError as e:
                # Step failed - compensate completed steps in reverse
                await compensate(context, e)
                raise SagaFailedError(tripId, step.name, e)
        
        return context
    
    async function compensate(context, originalError):
        # Execute compensating actions in reverse order
        for stepName in reversed(context.completedSteps):
            step = getStep(stepName)
            if step.compensate:
                try:
                    await step.compensate(context)
                except CompensationError as e:
                    # Compensation failed - needs manual intervention
                    await alertOps(f"Compensation failed for {stepName}: {e}")
        
        await recordSagaFailure(context, originalError)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

class TripTrackingService:
    """
    Handles real-time trip updates to rider app.
    """
    
    @eventHandler(DriverLocationUpdated)
    async function onDriverLocationUpdate(event):
        # Get active trips for this driver
        activeTrips = await getTripsByDriver(event.driverId, 
            statuses=[DRIVER_EN_ROUTE, DRIVER_ARRIVED, TRIP_STARTED])
        
        for trip in activeTrips:
            # Calculate updated ETA
            if trip.state == DRIVER_EN_ROUTE:
                newETA = await etaService.getPickupETA(
                    event.location, 
                    trip.pickupLocation
                )
            elif trip.state == TRIP_STARTED:
                newETA = await etaService.getArrivalETA(
                    event.location,
                    trip.dropoffLocation
                )
            else:
                newETA = null
            
            # Create update message
            update = TripLocationUpdate(
                tripId: trip.id,
                driverLocation: event.location,
                heading: event.heading,
                eta: newETA,
                timestamp: event.timestamp
            )
            
            # Push to rider
            await pushService.sendToUser(trip.riderId, update, {
                priority: "high",
                channel: "websocket",  # Prefer WebSocket if connected
                ttl: 10.seconds  # Expire quickly if not delivered
            })
 
class PushService:
    websocketGateway: WebSocketGateway
    apnsClient: APNSClient
    fcmClient: FCMClient
    
    async function sendToUser(userId, message, options):
        user = await getUser(userId)
        
        # Try WebSocket first (if app is open)
        if options.channel == "websocket" or options.channel == "any":
            wsConnections = await websocketGateway.getConnections(userId)
            if wsConnections.length > 0:
                for conn in wsConnections:
                    await conn.send(message)
                return {delivered: true, channel: "websocket"}
        
        # Fall back to push notifications
        if user.platform == "ios":
            await apnsClient.send(user.pushToken, message, {
                priority: options.priority,
                expiry: now() + options.ttl
            })
        else:
            await fcmClient.send(user.pushToken, message, {
                priority: options.priority,
                ttl: options.ttl.seconds
            })
        
        return {delivered: true, channel: "push"}