02 · The driver who drives backwards

// MARK: - Service
public protocol RideTracking: Sendable {
    func driverUpdates(rideID: String) -> AsyncStream<DriverUpdate>
    func eta(for update: DriverUpdate) async throws -> Duration
    func routeData(rideID: String) async throws -> Data
    func cancelRide(rideID: String) async throws
}

public struct RideService: RideTracking {
    public init() {}

    private static let logger = Logger(subsystem: "RideService", category: "tracking")

    /// Pushes the driver's position once a second until they reach the pickup.
    /// Stops as soon as the consumer goes away.
    public func driverUpdates(rideID: String) -> AsyncStream<DriverUpdate> {
        AsyncStream { continuation in
            let producer = Task {
                for (index, waypoint) in Self.waypoints.enumerated() {
                    do {
                        try await Task.sleep(for: .seconds(1))
                    } catch {
                        return                          // cancelled: stop producing
                    }
                    let sequence = index + 1            // local to this stream, not shared
                    Self.logger.notice("ride \(rideID, privacy: .public) sent update \(sequence)")
                    continuation.yield(DriverUpdate(sequence: sequence, position: waypoint.position,
                                                    metersAway: waypoint.metersAway))
                }
                continuation.finish()
            }
            continuation.onTermination = { _ in producer.cancel() }
        }
    }

    // eta(for:), routeData(rideID:) and cancelRide(rideID:): unchanged, except `Self.logger`.
}

import Foundation
import Observation
import os
import RideService

@Observable
final class RideTrackingViewModel {   // main actor: the target's default isolation
    enum Status: Equatable {
        case findingDriver
        case enRoute
        case arrived
    }

    let ride: Ride
    private(set) var status: Status = .findingDriver
    private(set) var route: [Coordinate] = []
    private(set) var driverPosition: Coordinate?
    private(set) var metersAway: Int?
    private(set) var eta: Duration?

    init(ride: Ride, service: any RideTracking) {
        self.ride = ride
        self.service = service
    }

    /// Runs until the driver arrives, or until the caller's task is cancelled (the view's `.task`).
    func track() async {
        await withDiscardingTaskGroup { group in
            for await update in service.driverUpdates(rideID: ride.id) {
                status = .enRoute
                driverPosition = update.position
                metersAway = update.metersAway
                group.addTask { [service] in
                    guard let eta = try? await service.eta(for: update) else { return }
                    await self.apply(eta, from: update.sequence)
                }
            }
            // The stream ends in two ways: the driver arrived, or we were cancelled.
            if !Task.isCancelled {
                status = .arrived
            }
            group.cancelAll()                       // late ETAs are meaningless now
        }
    }

    func loadRoute() async {
        do {
            let data = try await service.routeData(rideID: ride.id)
            route = try await Self.decodeRoute(data)
        } catch is CancellationError {
            // left the screen
        } catch {
            Self.logger.error("Route failed: \(error.localizedDescription)")   // the map still works without it
        }
    }

    @ObservationIgnored private let service: any RideTracking
    @ObservationIgnored private var etaSequence = 0     // the newest update an ETA was applied for

    private static let logger = Logger(subsystem: "RideTracking", category: "tracking")

    /// Results arrive in any order. Apply one only if it is newer than what's on screen.
    private func apply(_ eta: Duration, from sequence: Int) {
        guard status == .enRoute, sequence > etaSequence else { return }
        etaSequence = sequence
        self.eta = eta
    }

    /// `@concurrent` runs this on the global executor. A plain `nonisolated async` function runs on
    /// the caller's actor under Approachable Concurrency — here, the main actor.
    @concurrent
    private nonisolated static func decodeRoute(_ data: Data) async throws -> [Coordinate] {
        try RouteDecoder.decode(data)
    }
}

init(ride: Ride, service: any RideTracking = RideService()) {
    _viewModel = State(initialValue: RideTrackingViewModel(ride: ride, service: service))
}

// body: `Text(headline)`, and the distance only `if viewModel.status == .enRoute`.
    .task {
        await viewModel.loadRoute()
    }
    .task {
        await viewModel.track()                 // cancelled when the screen goes away
    }

private var headline: String {
    switch viewModel.status {
    case .findingDriver:
        "Finding your driver…"
    case .enRoute:
        if let eta = viewModel.eta {
            "Arriving in \(eta.formatted(.units(allowed: [.minutes, .seconds], width: .abbreviated)))"
        } else {
            "Your driver is on the way"
        }
    case .arrived:
        "\(viewModel.ride.driverName) is here"
    }
}

import Foundation
import Observation
import os
import RideService

@Observable
final class RideTrackingViewModel {   // main actor: the target's default isolation
    enum Status: Equatable {
        case findingDriver
        case enRoute
        case arrived
        case cancelling
        case cancelled
    }

    let ride: Ride
    private(set) var status: Status = .findingDriver
    private(set) var route: [Coordinate] = []
    private(set) var driverPosition: Coordinate?
    private(set) var metersAway: Int?
    private(set) var eta: Duration?
    private(set) var cancelError: String?

    /// Shown once per ride, the first time a current ETA drops under two minutes.
    /// A later bump (a missed turn) doesn't hide it: it's a heads-up, not a live value.
    var showsArrivingBanner: Bool { hasCrossedArrivalThreshold && status == .enRoute }

    var canCancel: Bool { status == .findingDriver || status == .enRoute }

    init(ride: Ride, service: any RideTracking) {
        self.ride = ride
        self.service = service
    }

    /// Runs until the driver arrives, the ride is cancelled, or the caller's task is cancelled.
    func track() async {
        let tracking = Task { await runTracking() }   // inherits the main actor
        self.tracking = tracking
        await withTaskCancellationHandler {
            await tracking.value
        } onCancel: {
            tracking.cancel()                         // the view's .task went away
        }
    }

    func cancelRide() async {
        guard canCancel else { return }               // one request at a time
        let previous = status
        status = .cancelling
        cancelError = nil
        do {
            try await service.cancelRide(rideID: ride.id)
            tracking?.cancel()                        // stop only once the server agrees
            status = .cancelled
        } catch {
            status = previous                         // the driver is still coming: keep tracking
            cancelError = "Couldn't cancel the ride. Try again."
        }
    }

    func loadRoute() async { ... unchanged ... }

    @ObservationIgnored private let service: any RideTracking
    @ObservationIgnored private var tracking: Task<Void, Never>?
    @ObservationIgnored private var etaSequence = 0     // the newest update an ETA was applied for
    private var hasCrossedArrivalThreshold = false

    private static let arrivalThreshold: Duration = .seconds(120)
    private static let logger = Logger(subsystem: "RideTracking", category: "tracking")

    private func runTracking() async {
        await withDiscardingTaskGroup { group in
            for await update in service.driverUpdates(rideID: ride.id) {
                if status == .findingDriver { status = .enRoute }
                driverPosition = update.position
                metersAway = update.metersAway
                group.addTask { [service] in
                    guard let eta = try? await service.eta(for: update) else { return }
                    await self.apply(eta, from: update.sequence)
                }
            }
            // The stream ends in two ways: the driver arrived, or we were cancelled.
            if !Task.isCancelled {
                status = .arrived
            }
            group.cancelAll()                       // late ETAs are meaningless now
        }
    }

    /// Results arrive in any order. Apply one only if it is newer than what's on screen.
    private func apply(_ eta: Duration, from sequence: Int) {
        guard status == .enRoute, sequence > etaSequence else { return }
        etaSequence = sequence
        self.eta = eta
        if eta < Self.arrivalThreshold {
            hasCrossedArrivalThreshold = true
        }
    }

    // decodeRoute(_:): unchanged, @concurrent.
}

if viewModel.canCancel || viewModel.status == .cancelling {
    Button("Cancel ride", role: .destructive) {
        Task { await viewModel.cancelRide() }
    }
    .buttonStyle(.bordered)
    .disabled(viewModel.status == .cancelling)
    .padding(.top, 8)
}
if let error = viewModel.cancelError {
    Text(error)
        .font(.footnote)
        .foregroundStyle(.red)
}

.overlay(alignment: .top) {
    if viewModel.showsArrivingBanner {
        Label("\(viewModel.ride.driverName) is almost here. Head to the pickup.", systemImage: "car.fill")
            .font(.headline)
            .padding()
            .frame(maxWidth: .infinity)
            .background(.green.gradient, in: .rect(cornerRadius: 16))
            .foregroundStyle(.white)
            .padding()
            .transition(.move(edge: .top).combined(with: .opacity))
    }
}
.animation(.default, value: viewModel.showsArrivingBanner)

import Foundation
import Synchronization
import Testing
import RideService
@testable import RideTracking

/// A stream the test drives by hand, and ETAs scripted per update.
nonisolated final class FakeRideService: RideTracking {
    let updates: AsyncStream<DriverUpdate>
    let driver: AsyncStream<DriverUpdate>.Continuation
    let etas: [Int: (eta: Duration, delay: Duration)]
    let cancelFails: Bool
    let cancelCalls = Mutex(0)
    let streamTerminated = Mutex(false)

    init(etas: [Int: (eta: Duration, delay: Duration)] = [:], cancelFails: Bool = false) {
        (updates, driver) = AsyncStream.makeStream(of: DriverUpdate.self)
        self.etas = etas
        self.cancelFails = cancelFails
        driver.onTermination = { [weak self] _ in self?.streamTerminated.withLock { $0 = true } }
    }

    func driverUpdates(rideID: String) -> AsyncStream<DriverUpdate> { updates }

    func eta(for update: DriverUpdate) async throws -> Duration {
        let scripted = etas[update.sequence] ?? (.seconds(300), .zero)
        try await Task.sleep(for: scripted.delay)
        return scripted.eta
    }

    func routeData(rideID: String) async throws -> Data { Data("[]".utf8) }

    func cancelRide(rideID: String) async throws {
        cancelCalls.withLock { $0 += 1 }
        if cancelFails { throw URLError(.notConnectedToInternet) }
    }

    func send(_ sequence: Int) {
        driver.yield(DriverUpdate(sequence: sequence, position: Coordinate(latitude: 52.52, longitude: 13.40),
                                  metersAway: 1000))
    }
}

/// Polls until the view model reaches the expected state. Fails instead of hanging.
private func waitUntil(_ condition: () -> Bool) async throws {
    for _ in 0..<200 where !condition() {
        try await Task.sleep(for: .milliseconds(10))
    }
    try #require(condition())
}

struct RideTrackingViewModelTests {
    let ride = Ride(id: "R-1", driverName: "Mehmet", car: "Prius",
                    pickup: Coordinate(latitude: 52.51, longitude: 13.37))

    @Test func aSlowOlderETADoesNotOverwriteANewerOne() async throws {
        let service = FakeRideService(etas: [1: (.seconds(200), .milliseconds(300)),   // slow
                                             2: (.seconds(190), .zero)])               // fast
        let viewModel = RideTrackingViewModel(ride: ride, service: service)
        let tracking = Task { await viewModel.track() }
        service.send(1)
        service.send(2)
        try await waitUntil { viewModel.eta == .seconds(190) }
        try await Task.sleep(for: .milliseconds(500))           // the slow answer for update 1 lands now
        #expect(viewModel.eta == .seconds(190))
        tracking.cancel()
    }

    @Test func theStreamEndingMeansTheDriverArrived() async {
        let service = FakeRideService()
        let viewModel = RideTrackingViewModel(ride: ride, service: service)
        service.send(1)
        service.driver.finish()
        await viewModel.track()
        #expect(viewModel.status == .arrived)
    }

    @Test func theBannerShowsOnceAndStaysWhenTheETAGoesBackUp() async throws {
        let service = FakeRideService(etas: [1: (.seconds(130), .zero),
                                             2: (.seconds(110), .zero),
                                             3: (.seconds(125), .zero)])   // a missed turn
        let viewModel = RideTrackingViewModel(ride: ride, service: service)
        let tracking = Task { await viewModel.track() }
        service.send(1)
        try await waitUntil { viewModel.eta == .seconds(130) }
        #expect(!viewModel.showsArrivingBanner)
        service.send(2)
        try await waitUntil { viewModel.eta == .seconds(110) }
        #expect(viewModel.showsArrivingBanner)
        service.send(3)
        try await waitUntil { viewModel.eta == .seconds(125) }
        #expect(viewModel.showsArrivingBanner)
        tracking.cancel()
    }

    @Test func cancellingStopsTrackingOnceTheServerConfirms() async throws {
        let service = FakeRideService()
        let viewModel = RideTrackingViewModel(ride: ride, service: service)
        let tracking = Task { await viewModel.track() }
        service.send(1)
        try await waitUntil { viewModel.status == .enRoute }
        await viewModel.cancelRide()
        await tracking.value                                    // returns: tracking really stopped
        #expect(viewModel.status == .cancelled)
        #expect(service.cancelCalls.withLock { $0 } == 1)
        #expect(service.streamTerminated.withLock { $0 })
    }

    @Test func aFailedCancelKeepsTracking() async throws {
        let service = FakeRideService(cancelFails: true)
        let viewModel = RideTrackingViewModel(ride: ride, service: service)
        let tracking = Task { await viewModel.track() }
        service.send(1)
        try await waitUntil { viewModel.status == .enRoute }
        await viewModel.cancelRide()
        #expect(viewModel.status == .enRoute)
        #expect(viewModel.cancelError != nil)
        #expect(!service.streamTerminated.withLock { $0 })
        tracking.cancel()
    }
}