// Copyright 2020 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

use {
    anyhow::{self, Context, Error},
    emergency_lib::{
        bss_cache::{Bss, BssCache, BssId, RealBssCache, UpdateError},
        bss_resolver::{BssResolver, RealBssResolver, ResolverError},
    },
    emergency_metrics_registry::{
        self as metrics, EmergencyGetCurrentFailureMetricDimensionCause as GetCurrentFailure,
        EmergencyGetCurrentResultMetricDimensionResult as GetCurrentResult,
        WlanSensorReportMetricDimensionResult as WlanSensorReportResult,
        EMERGENCY_GET_CURRENT_ACCURACY_METRIC_ID as GET_CURRENT_ACCURACY_METRIC_ID,
        EMERGENCY_GET_CURRENT_FAILURE_METRIC_ID as GET_CURRENT_FAILURE_METRIC_ID,
        EMERGENCY_GET_CURRENT_LATENCY_METRIC_ID as GET_CURRENT_LATENCY_METRIC_ID,
        EMERGENCY_GET_CURRENT_RESULT_METRIC_ID as GET_CURRENT_RESULT_METRIC_ID,
        WLAN_SENSOR_REPORT_METRIC_ID,
    },
    fidl_fuchsia_location::Error as LocationError,
    fidl_fuchsia_location_position::{
        EmergencyProviderRequest, EmergencyProviderRequestStream, Position,
    },
    fidl_fuchsia_location_sensor::{
        WlanBaseStationWatcherRequest, WlanBaseStationWatcherRequestStream,
    },
    fidl_fuchsia_net_http::LoaderMarker as HttpLoaderMarker,
    fuchsia_async as fasync,
    fuchsia_cobalt::{CobaltConnector, CobaltSender, ConnectionType},
    fuchsia_component::{client::connect_to_service, server::ServiceFs},
    fuchsia_syslog::{self as syslog},
    futures::{lock::Mutex, prelude::*},
    log::info,
    std::{
        convert::TryFrom,
        time::{Duration, Instant},
    },
};

const CONCURRENCY_LIMIT: Option<usize> = None;
const API_KEY_FILE: &str = "/config/data/google_maps_api_key.txt";

/// Wraps all hosted protocols into a single type that can be matched against
/// and dispatched.
enum IncomingRequest {
    EmergencyProviderRequest(EmergencyProviderRequestStream),
    WlanBaseStationWatcherRequest(WlanBaseStationWatcherRequestStream),
}

#[fasync::run_singlethreaded]
async fn main() -> Result<(), anyhow::Error> {
    syslog::init().context("failed to initialize logging")?;

    let (cobalt_api, cobalt_fut) =
        CobaltConnector::default().serve(ConnectionType::project_id(metrics::PROJECT_ID));
    let _cobalt_task = fasync::Task::spawn(cobalt_fut);
    let bss_cache = Mutex::new(RealBssCache::new());
    let bss_resolver = RealBssResolver::new(
        connect_to_service::<HttpLoaderMarker>().context("failed to connect to http loader")?,
        std::fs::read_to_string(API_KEY_FILE)
            .with_context(|| format!("failed to read {}", API_KEY_FILE))?,
    );
    let mut service_fs = ServiceFs::new_local();
    service_fs
        .dir("svc")
        .add_fidl_service(IncomingRequest::EmergencyProviderRequest)
        .add_fidl_service(IncomingRequest::WlanBaseStationWatcherRequest);
    service_fs
        .take_and_serve_directory_handle()
        .context("failed to serve outgoing namespace")?
        .for_each_concurrent(CONCURRENCY_LIMIT, |connection| {
            handle_client_requests(&bss_cache, &bss_resolver, connection, cobalt_api.clone())
                .unwrap_or_else(|e| info!("connection terminated: {:?}", e))
        })
        .await;

    Ok(())
}

async fn handle_client_requests<C: BssCache, R: BssResolver>(
    bss_cache: &Mutex<C>,
    bss_resolver: &R,
    protocol: IncomingRequest,
    cobalt_api: CobaltSender,
) -> Result<(), Error> {
    match protocol {
        IncomingRequest::EmergencyProviderRequest(client) => {
            process_location_queries(&bss_cache, bss_resolver, client, cobalt_api).await
        }
        IncomingRequest::WlanBaseStationWatcherRequest(client) => {
            process_bss_updates(&bss_cache, client, cobalt_api).await
        }
    }
}

async fn process_location_queries<C: BssCache, R: BssResolver>(
    bss_cache: &Mutex<C>,
    bss_resolver: &R,
    mut stream: EmergencyProviderRequestStream,
    mut cobalt_api: CobaltSender,
) -> Result<(), Error> {
    loop {
        match stream.try_next().await.context("failed to read emergency provider request")? {
            Some(EmergencyProviderRequest::GetCurrent { responder }) => {
                let start_time = Instant::now();
                // We don't want to hold the BSS cache lock while resolving the BSSes to a
                // `Position`, so we copy data from the iterator into our own `Vector`.
                let bss_list: Vec<(BssId, Bss)> =
                    bss_cache.lock().await.iter().map(|(&id, &bss)| (id, bss)).collect();
                match bss_resolver.resolve(bss_list).await {
                    Ok(position) => {
                        report_lookup_success_metrics(
                            &position,
                            &start_time.elapsed(),
                            &mut cobalt_api,
                        );
                        responder
                            .send(&mut Ok(position))
                            .context("failed to send position to caller")?;
                    }
                    Err(e) => {
                        info!("lookup failed: {:?}", e);
                        report_lookup_failure_metrics(e, &mut cobalt_api);
                        responder
                            .send(&mut Err(LocationError::GeneralError))
                            .context("failed to send error to client")?
                    }
                }
            }
            None => return Ok(()),
        }
    }
}

async fn process_bss_updates<C: BssCache>(
    bss_cache: &Mutex<C>,
    mut stream: WlanBaseStationWatcherRequestStream,
    mut cobalt_api: CobaltSender,
) -> Result<(), Error> {
    loop {
        match stream.try_next().await.context("failed to read base station watcher request")? {
            Some(WlanBaseStationWatcherRequest::ReportCurrentStations {
                stations,
                control_handle: _,
            }) => {
                let update_result = bss_cache
                    .lock()
                    .await
                    .update(
                        stations
                            .into_proxy()
                            .context("failed to get proxy for scan result iterator")?,
                    )
                    .await;
                report_bss_update_metrics(update_result, &mut cobalt_api);
                update_result.context("failed to apply base station update")?
            }
            None => return Ok(()),
        }
    }
}

fn report_lookup_success_metrics(
    position: &Position,
    latency: &Duration,
    cobalt_api: &mut CobaltSender,
) {
    const METERS_TO_MILLIMETERS: f64 = 1000.0;
    cobalt_api.log_event(GET_CURRENT_RESULT_METRIC_ID, GetCurrentResult::Success);
    cobalt_api.log_elapsed_time(
        GET_CURRENT_LATENCY_METRIC_ID,
        (),
        i64::try_from(latency.as_millis()).unwrap_or(i64::MAX),
    );
    cobalt_api.log_event_count(
        GET_CURRENT_ACCURACY_METRIC_ID,
        (),
        0,
        match position.extras.accuracy_meters {
            Some(accuracy_meters) => (accuracy_meters * METERS_TO_MILLIMETERS) as i64,
            None => i64::MAX,
        },
    )
}

fn report_lookup_failure_metrics(error: ResolverError, cobalt_api: &mut CobaltSender) {
    cobalt_api.log_event(GET_CURRENT_RESULT_METRIC_ID, GetCurrentResult::Failure);
    cobalt_api.log_event(
        GET_CURRENT_FAILURE_METRIC_ID,
        match error {
            ResolverError::NoBsses => GetCurrentFailure::NoBsses,
            ResolverError::Internal => GetCurrentFailure::Internal,
            ResolverError::Lookup => GetCurrentFailure::Lookup,
        },
    );
}

fn report_bss_update_metrics(result: Result<(), UpdateError>, cobalt_api: &mut CobaltSender) {
    cobalt_api.log_event(
        WLAN_SENSOR_REPORT_METRIC_ID,
        match result {
            Ok(()) => WlanSensorReportResult::Success,
            Err(UpdateError::NoBssIds) => WlanSensorReportResult::NoBssIds,
            Err(UpdateError::NoBsses) => WlanSensorReportResult::NoBsses,
            Err(UpdateError::Ipc) => WlanSensorReportResult::IpcError,
            Err(UpdateError::Service) => WlanSensorReportResult::ServiceError,
        },
    );
}

#[cfg(test)]
mod tests {
    use {
        super::*,
        cobalt_client::traits::AsEventCode,
        fidl::endpoints::create_proxy_and_stream,
        fidl_fuchsia_cobalt::CobaltEvent,
        fuchsia_cobalt::cobalt_event_builder::CobaltEventExt,
        fuchsia_cobalt::CobaltSender,
        futures::channel::mpsc,
        futures::{future, pin_mut},
        matches::assert_matches,
        std::task::Poll,
        test_doubles::{FakeBssCache, StubBssResolver},
    };

    mod base_station_watcher {
        use {
            super::*, emergency_lib::bss_cache::UpdateError,
            fidl::endpoints::create_request_stream,
            fidl_fuchsia_location_sensor::WlanBaseStationWatcherMarker,
            fidl_fuchsia_wlan_policy::ScanResultIteratorMarker, test_case::test_case,
        };

        #[fasync::run_until_stalled(test)]
        async fn propagates_stations_downward() {
            let (cobalt_sender, _cobalt_receiver) = make_fake_cobalt_connection();
            let (proxy, stream) = create_proxy_and_stream::<WlanBaseStationWatcherMarker>()
                .expect("internal error: failed to create base station watcher");
            let bss_cache = Mutex::new(FakeBssCache::new(Ok(())));
            let server_fut = handle_client_requests(
                &bss_cache,
                &StubBssResolver { resolve: || panic!("unexpected call to resolver") },
                IncomingRequest::WlanBaseStationWatcherRequest(stream),
                cobalt_sender,
            );
            let (scan_result_reader, _scan_result_generator) =
                create_request_stream::<ScanResultIteratorMarker>()
                    .expect("internal error: failed to create scan result iterator");
            proxy
                .report_current_stations(scan_result_reader)
                .expect("internal error: proxy failed to send request");
            std::mem::drop(proxy); // Close connection so `server_fut` completes.
            assert!(server_fut.await.is_ok());
            assert!(bss_cache.lock().await.was_update_called())
        }

        #[fasync::run_until_stalled(test)]
        async fn update_error_does_not_panic() {
            let (cobalt_sender, _cobalt_receiver) = make_fake_cobalt_connection();
            let (proxy, stream) = create_proxy_and_stream::<WlanBaseStationWatcherMarker>()
                .expect("internal error: failed to create base station watcher");
            let bss_cache = Mutex::new(FakeBssCache::new(Err(UpdateError::NoBssIds)));
            let server_fut = handle_client_requests(
                &bss_cache,
                &StubBssResolver { resolve: || panic!("unexpected call to resolver") },
                IncomingRequest::WlanBaseStationWatcherRequest(stream),
                cobalt_sender,
            );
            let (scan_result_reader, _scan_result_generator) =
                create_request_stream::<ScanResultIteratorMarker>()
                    .expect("internal error: failed to create scan result iterator");
            proxy
                .report_current_stations(scan_result_reader)
                .expect("internal error: proxy failed to send request");
            // Close connection so `server_fut` completes, even if it chooses to
            // ignore the error.
            std::mem::drop(proxy);

            // The best error handling policy isn't exactly clear: is it useful
            // to report an error upwards (which would cause `main()` to close
            // the client connection), or is it better to leave the connection
            // open, and hope that the client will provide more useful results
            // next time?
            //
            // Rather than take a position on that question, we simply validate
            // that the program doesn't crash when that happens. (If the program
            // crashed, the test framework would report a test failure with
            // the panic message.)
            let _ = server_fut.await;
        }

        #[test_case(Ok(()), WlanSensorReportResult::Success; "success")]
        #[test_case(Err(UpdateError::NoBssIds), WlanSensorReportResult::NoBssIds; "no bss ids")]
        #[test_case(Err(UpdateError::NoBsses), WlanSensorReportResult::NoBsses; "no bsses")]
        #[test_case(Err(UpdateError::Ipc), WlanSensorReportResult::IpcError; "ipc error")]
        #[test_case(Err(UpdateError::Service), WlanSensorReportResult::ServiceError;
             "service error")]
        fn reports_update_result_to_cobalt(
            update_result: Result<(), UpdateError>,
            cobalt_event: WlanSensorReportResult,
        ) {
            let test_fut = async {
                let (cobalt_sender, cobalt_receiver) = make_fake_cobalt_connection();
                let (proxy, stream) = create_proxy_and_stream::<WlanBaseStationWatcherMarker>()
                    .expect("internal error: failed to create emergency provider");
                let bss_cache = Mutex::new(FakeBssCache::new(update_result));
                let server_fut = handle_client_requests(
                    &bss_cache,
                    &StubBssResolver { resolve: || panic!("unexpected call to resolver") },
                    IncomingRequest::WlanBaseStationWatcherRequest(stream),
                    cobalt_sender,
                );
                let (scan_result_reader, _scan_result_generator) =
                    create_request_stream::<ScanResultIteratorMarker>()
                        .expect("internal error: failed to create scan result iterator");
                proxy
                    .report_current_stations(scan_result_reader)
                    .expect("internal error: proxy failed to send request");
                std::mem::drop(proxy); // Close connection so `server_fut` completes.

                let _ = server_fut.await;
                assert_eq!(
                    cobalt_receiver
                        .filter(|event| {
                            future::ready(event.metric_id == WLAN_SENSOR_REPORT_METRIC_ID)
                        })
                        .collect::<Vec<CobaltEvent>>()
                        .await,
                    vec![CobaltEvent::builder(WLAN_SENSOR_REPORT_METRIC_ID)
                        .with_event_code(cobalt_event.as_event_code())
                        .as_event()]
                );
            };
            pin_mut!(test_fut);
            assert_matches!(
                fasync::Executor::new()
                    .expect("internal error: failed to create executor")
                    .run_until_stalled(&mut test_fut),
                Poll::Ready(_)
            );
        }

        #[fasync::run_until_stalled(test)]
        async fn does_not_report_extra_metrics() {
            let (cobalt_sender, cobalt_receiver) = make_fake_cobalt_connection();
            let (proxy, stream) = create_proxy_and_stream::<WlanBaseStationWatcherMarker>()
                .expect("internal error: failed to create emergency provider");
            let bss_cache = Mutex::new(FakeBssCache::new(Ok(())));
            let server_fut = handle_client_requests(
                &bss_cache,
                &StubBssResolver { resolve: || panic!("unexpected call to resolver") },
                IncomingRequest::WlanBaseStationWatcherRequest(stream),
                cobalt_sender,
            );
            let (scan_result_reader, _scan_result_generator) =
                create_request_stream::<ScanResultIteratorMarker>()
                    .expect("internal error: failed to create scan result iterator");
            proxy
                .report_current_stations(scan_result_reader)
                .expect("internal error: proxy failed to send request");
            std::mem::drop(proxy); // Close connection so `server_fut` completes.

            let _ = server_fut.await;
            assert_eq!(
                cobalt_receiver
                    .filter(|event| future::ready(event.metric_id != WLAN_SENSOR_REPORT_METRIC_ID))
                    .collect::<Vec<CobaltEvent>>()
                    .await,
                vec![]
            );
        }
    }

    mod emergency_provider {
        use {
            super::*,
            emergency_lib::bss_resolver::ResolverError,
            fidl_fuchsia_cobalt::EventPayload::ElapsedMicros,
            fidl_fuchsia_location_position::{EmergencyProviderMarker, Position, PositionExtras},
            test_case::test_case,
        };

        const POSITION_WITH_UNKNOWN_ACCURACY: Position = Position {
            latitude: 1.0,
            longitude: -1.0,
            extras: PositionExtras {
                accuracy_meters: None,
                altitude_meters: None,
                ..PositionExtras::EMPTY
            },
        };

        const POSITION_WITH_KNOWN_ACCURACY: Position = Position {
            latitude: 1.0,
            longitude: -1.0,
            extras: PositionExtras {
                accuracy_meters: Some(1.0),
                altitude_meters: None,
                ..PositionExtras::EMPTY
            },
        };

        #[fasync::run_until_stalled(test)]
        async fn propagates_success_to_client() {
            let (cobalt_sender, _cobalt_receiver) = make_fake_cobalt_connection();
            let (proxy, stream) = create_proxy_and_stream::<EmergencyProviderMarker>()
                .expect("internal error: failed to create emergency provider");
            let bss_cache = Mutex::new(FakeBssCache::new(Ok(())));
            let server_fut = handle_client_requests(
                &bss_cache,
                &StubBssResolver { resolve: || Ok(POSITION_WITH_UNKNOWN_ACCURACY) },
                IncomingRequest::EmergencyProviderRequest(stream),
                cobalt_sender,
            );
            let client_fut = proxy.get_current();
            std::mem::drop(proxy); // Close connection so `server_fut` completes.

            let (client_res, _server_res) = future::join(client_fut, server_fut).await;
            assert_matches!(client_res, Ok(Ok(Position {..})))
        }

        #[fasync::run_until_stalled(test)]
        async fn propagates_error_to_client() {
            let (cobalt_sender, _cobalt_receiver) = make_fake_cobalt_connection();
            let (proxy, stream) = create_proxy_and_stream::<EmergencyProviderMarker>()
                .expect("internal error: failed to create emergency provider");
            let bss_cache = Mutex::new(FakeBssCache::new(Ok(())));
            let server_fut = handle_client_requests(
                &bss_cache,
                &StubBssResolver { resolve: || Err(ResolverError::NoBsses) },
                IncomingRequest::EmergencyProviderRequest(stream),
                cobalt_sender,
            );
            let client_fut = proxy.get_current();
            std::mem::drop(proxy); // Close connection so `server_fut` completes.

            let (client_res, _server_res) = future::join(client_fut, server_fut).await;
            assert_matches!(client_res, Ok(Err(_))) // The `Ok` is the FIDL-level result.
        }

        #[fasync::run_until_stalled(test)]
        async fn reports_success_to_cobalt() {
            let (cobalt_sender, cobalt_receiver) = make_fake_cobalt_connection();
            let (proxy, stream) = create_proxy_and_stream::<EmergencyProviderMarker>()
                .expect("internal error: failed to create emergency provider");
            let bss_cache = Mutex::new(FakeBssCache::new(Ok(())));
            let server_fut = handle_client_requests(
                &bss_cache,
                &StubBssResolver { resolve: || Ok(POSITION_WITH_UNKNOWN_ACCURACY) },
                IncomingRequest::EmergencyProviderRequest(stream),
                cobalt_sender,
            );
            let _ = proxy.get_current();
            std::mem::drop(proxy); // Close connection so `server_fut` completes.

            let _ = server_fut.await;
            assert_eq!(
                cobalt_receiver
                    .filter(|event| future::ready(event.metric_id == GET_CURRENT_RESULT_METRIC_ID))
                    .collect::<Vec<CobaltEvent>>()
                    .await,
                vec![CobaltEvent::builder(GET_CURRENT_RESULT_METRIC_ID)
                    .with_event_code(GetCurrentResult::Success.as_event_code())
                    .as_event()]
            );
        }

        #[fasync::run_until_stalled(test)]
        async fn reports_failure_to_cobalt() {
            let (cobalt_sender, cobalt_receiver) = make_fake_cobalt_connection();
            let (proxy, stream) = create_proxy_and_stream::<EmergencyProviderMarker>()
                .expect("internal error: failed to create emergency provider");
            let bss_cache = Mutex::new(FakeBssCache::new(Ok(())));
            let server_fut = handle_client_requests(
                &bss_cache,
                &StubBssResolver { resolve: || Err(ResolverError::NoBsses) },
                IncomingRequest::EmergencyProviderRequest(stream),
                cobalt_sender,
            );
            let _ = proxy.get_current();
            std::mem::drop(proxy); // Close connection so `server_fut` completes.

            let _ = server_fut.await;
            assert_eq!(
                cobalt_receiver
                    .filter(|event| future::ready(event.metric_id == GET_CURRENT_RESULT_METRIC_ID))
                    .collect::<Vec<CobaltEvent>>()
                    .await,
                vec![CobaltEvent::builder(GET_CURRENT_RESULT_METRIC_ID)
                    .with_event_code(GetCurrentResult::Failure.as_event_code())
                    .as_event()]
            );
        }

        #[fasync::run_until_stalled(test)]
        async fn propagates_reported_accuracy_to_cobalt() {
            let (cobalt_sender, cobalt_receiver) = make_fake_cobalt_connection();
            let (proxy, stream) = create_proxy_and_stream::<EmergencyProviderMarker>()
                .expect("internal error: failed to create emergency provider");
            let bss_cache = Mutex::new(FakeBssCache::new(Ok(())));
            let server_fut = handle_client_requests(
                &bss_cache,
                &StubBssResolver { resolve: || Ok(POSITION_WITH_KNOWN_ACCURACY) },
                IncomingRequest::EmergencyProviderRequest(stream),
                cobalt_sender,
            );
            let _ = proxy.get_current();
            std::mem::drop(proxy); // Close connection so `server_fut` completes.

            let _ = server_fut.await;
            assert_eq!(
                cobalt_receiver
                    .filter(|event| future::ready(
                        event.metric_id == GET_CURRENT_ACCURACY_METRIC_ID
                    ))
                    .collect::<Vec<CobaltEvent>>()
                    .await,
                vec![CobaltEvent::builder(GET_CURRENT_ACCURACY_METRIC_ID).as_count_event(
                    0,
                    1000 // In millimeters, for higher precision
                )]
            );
        }

        #[fasync::run_until_stalled(test)]
        async fn reports_worst_accuracy_if_accuracy_is_unknown() {
            let (cobalt_sender, cobalt_receiver) = make_fake_cobalt_connection();
            let (proxy, stream) = create_proxy_and_stream::<EmergencyProviderMarker>()
                .expect("internal error: failed to create emergency provider");
            let bss_cache = Mutex::new(FakeBssCache::new(Ok(())));
            let server_fut = handle_client_requests(
                &bss_cache,
                &StubBssResolver { resolve: || Ok(POSITION_WITH_UNKNOWN_ACCURACY) },
                IncomingRequest::EmergencyProviderRequest(stream),
                cobalt_sender,
            );
            let _ = proxy.get_current();
            std::mem::drop(proxy); // Close connection so `server_fut` completes.

            let _ = server_fut.await;
            assert_eq!(
                cobalt_receiver
                    .filter(|event| future::ready(
                        event.metric_id == GET_CURRENT_ACCURACY_METRIC_ID
                    ))
                    .collect::<Vec<CobaltEvent>>()
                    .await,
                vec![CobaltEvent::builder(GET_CURRENT_ACCURACY_METRIC_ID)
                    .as_count_event(0, i64::MAX)]
            );
        }

        #[fasync::run_until_stalled(test)]
        async fn reports_elapsed_time_to_cobalt_on_success() {
            let (cobalt_sender, cobalt_receiver) = make_fake_cobalt_connection();
            let (proxy, stream) = create_proxy_and_stream::<EmergencyProviderMarker>()
                .expect("internal error: failed to create emergency provider");
            let bss_cache = Mutex::new(FakeBssCache::new(Ok(())));
            let server_fut = handle_client_requests(
                &bss_cache,
                &StubBssResolver { resolve: || Ok(POSITION_WITH_UNKNOWN_ACCURACY) },
                IncomingRequest::EmergencyProviderRequest(stream),
                cobalt_sender,
            );
            let _ = proxy.get_current();
            std::mem::drop(proxy); // Close connection so `server_fut` completes.

            let _ = server_fut.await;
            let latency_reports = cobalt_receiver
                .filter(|event| future::ready(event.metric_id == GET_CURRENT_LATENCY_METRIC_ID))
                .collect::<Vec<CobaltEvent>>()
                .await;
            assert_eq!(latency_reports.len(), 1);
            assert_matches!(
                latency_reports[0],
                CobaltEvent {
                    metric_id: GET_CURRENT_LATENCY_METRIC_ID,
                    payload: ElapsedMicros(_),
                    ..
                }
            );
        }

        #[test_case(ResolverError::NoBsses, GetCurrentFailure::NoBsses; "no bsses")]
        #[test_case(ResolverError::Internal, GetCurrentFailure::Internal; "internal")]
        #[test_case(ResolverError::Lookup, GetCurrentFailure::Lookup; "lookup")]
        fn reports_resolver_error_to_cobalt(
            resolver_error: ResolverError,
            cobalt_error: GetCurrentFailure,
        ) {
            let test_fut = async {
                let (cobalt_sender, cobalt_receiver) = make_fake_cobalt_connection();
                let (proxy, stream) = create_proxy_and_stream::<EmergencyProviderMarker>()
                    .expect("internal error: failed to create emergency provider");
                let bss_cache = Mutex::new(FakeBssCache::new(Ok(())));
                let resolver = StubBssResolver { resolve: || Err(resolver_error) };
                let server_fut = handle_client_requests(
                    &bss_cache,
                    &resolver,
                    IncomingRequest::EmergencyProviderRequest(stream),
                    cobalt_sender,
                );
                let _ = proxy.get_current();
                std::mem::drop(proxy); // Close connection so `server_fut` completes.

                let _ = server_fut.await;
                assert_eq!(
                    cobalt_receiver
                        .filter(|event| {
                            future::ready(event.metric_id == GET_CURRENT_FAILURE_METRIC_ID)
                        })
                        .collect::<Vec<CobaltEvent>>()
                        .await,
                    vec![CobaltEvent::builder(GET_CURRENT_FAILURE_METRIC_ID)
                        .with_event_code(cobalt_error.as_event_code())
                        .as_event()]
                );
            };
            pin_mut!(test_fut);
            assert_matches!(
                fasync::Executor::new()
                    .expect("internal error: failed to create executor")
                    .run_until_stalled(&mut test_fut),
                Poll::Ready(_)
            );
        }

        #[test_case(
            Ok(()),
            &[GET_CURRENT_RESULT_METRIC_ID, GET_CURRENT_ACCURACY_METRIC_ID,
                 GET_CURRENT_LATENCY_METRIC_ID];
            "on_success")]
        #[test_case(
            Err(()),
            &[GET_CURRENT_RESULT_METRIC_ID, GET_CURRENT_FAILURE_METRIC_ID];
            "on_failure")]
        fn does_not_report_extra_metrics(result: Result<(), ()>, expected_metric_ids: &[u32]) {
            let test_fut = async {
                let (cobalt_sender, cobalt_receiver) = make_fake_cobalt_connection();
                let (proxy, stream) = create_proxy_and_stream::<EmergencyProviderMarker>()
                    .expect("internal error: failed to create emergency provider");
                let bss_cache = Mutex::new(FakeBssCache::new(Ok(())));
                let resolver = StubBssResolver {
                    resolve: || match &result {
                        Ok(()) => Ok(POSITION_WITH_KNOWN_ACCURACY),
                        Err(()) => Err(ResolverError::NoBsses),
                    },
                };
                let server_fut = handle_client_requests(
                    &bss_cache,
                    &resolver,
                    IncomingRequest::EmergencyProviderRequest(stream),
                    cobalt_sender,
                );
                let _ = proxy.get_current();
                std::mem::drop(proxy); // Close connection so `server_fut` completes.

                let _ = server_fut.await;
                assert_eq!(
                    cobalt_receiver
                        .filter(|event| future::ready(
                            expected_metric_ids.iter().find(|&&i| i == event.metric_id).is_none()
                        ))
                        .collect::<Vec<CobaltEvent>>()
                        .await,
                    vec![]
                );
            };
            pin_mut!(test_fut);
            assert_matches!(
                fasync::Executor::new()
                    .expect("internal error: failed to create executor")
                    .run_until_stalled(&mut test_fut),
                Poll::Ready(_)
            );
        }
    }

    fn make_fake_cobalt_connection() -> (CobaltSender, mpsc::Receiver<CobaltEvent>) {
        const MAX_METRICS_PER_QUERY: usize = 3;
        const MAX_QUERIES: usize = 1;
        let (sender, receiver) = mpsc::channel(MAX_METRICS_PER_QUERY * MAX_QUERIES);
        (CobaltSender::new(sender), receiver)
    }
}

#[cfg(test)]
mod test_doubles {
    use {
        super::*,
        async_trait::async_trait,
        emergency_lib::{bss_cache::UpdateError, bss_resolver::ResolverError},
        fidl_fuchsia_location_position::Position,
        fidl_fuchsia_wlan_policy::ScanResultIteratorProxyInterface,
    };

    pub(super) struct FakeBssCache {
        update_result: Result<(), UpdateError>,
        bsses: Vec<(BssId, Bss)>,
        was_update_called: bool,
    }

    pub(super) struct StubBssResolver<R: Fn() -> Result<Position, ResolverError>> {
        // Note, we can't just store a value here, because `Position` is not Copy.
        pub resolve: R,
    }

    impl FakeBssCache {
        pub fn new(update_result: Result<(), UpdateError>) -> Self {
            Self { update_result, bsses: Vec::new(), was_update_called: false }
        }

        pub fn was_update_called(&self) -> bool {
            self.was_update_called
        }
    }

    #[async_trait(?Send)]
    impl BssCache for FakeBssCache {
        async fn update<I: ScanResultIteratorProxyInterface>(
            &mut self,
            _new_bsses: I,
        ) -> Result<(), UpdateError> {
            self.was_update_called = true;
            self.update_result.clone()
        }

        fn iter(&self) -> Box<dyn Iterator<Item = (&'_ BssId, &'_ Bss)> + '_> {
            Box::new(self.bsses.iter().map(|(id, bss)| (id, bss)))
        }
    }

    #[async_trait(?Send)]
    impl<R> BssResolver for StubBssResolver<R>
    where
        R: Fn() -> Result<Position, ResolverError>,
    {
        async fn resolve<'a, I, T, U>(&self, _bss_list: I) -> Result<Position, ResolverError> {
            (self.resolve)()
        }
    }
}