//! The trainer supervisor: the app's single owner of an [`FtmsClient`].
//!
//! Everything BLE-shaped happens in one background task. The Tauri commands and
//! the ride loop are synchronous and hold a `Mutex`, so they may never `.await`
//! a radio; they talk to this task over a channel instead, and read its results
//! from two `watch` channels:
//!
//! ```text
//!   commands ──connect/target/release──► [supervisor task] ──► FtmsClient
//!   ride loop ◄──watch<Telemetry>────────      │
//!   device loop ◄──watch<TrainerStatus>────────┘
//! ```
//!
//! Three properties are load-bearing:
//!
//! * **SAF-2** — [`TrainerHandle::shutdown_blocking`] is callable from Tauri's
//!   synchronous `RunEvent` handler, and waits for `FtmsClient::shutdown` to
//!   actually finish. A fire-and-forget send would race the process exit and
//!   leave the trainer loaded.
//! * **FR-9.3** — `control_acquired` is tracked separately from the connection
//!   state. Connected is not controllable.
//! * **Never freeze on stale data** — if the trainer goes quiet while
//!   nominally connected, the published telemetry is *zeroed* rather than held,
//!   so the ride engine coasts to a stop and the UI visibly reacts instead of
//!   showing a plausible-looking lie (FR-1.8).

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{sync_channel, SyncSender};
use std::sync::Arc;
use std::time::{Duration, Instant};

use bikecontrol_ble::{
    Backoff, ControlOutcome, FtmsClient, FtmsConfig, FtmsError, FtmsEvent, TrainerSelector,
};
use bikecontrol_core::types::{ConnectionState, ControlTarget, SafetyLimits, Telemetry};
use serde::Serialize;
use tokio::sync::{broadcast, mpsc, watch};

/// How long the trainer may stay silent before its telemetry is treated as
/// stale. The D100 notifies at 4 Hz, so three seconds is ~12 missed frames.
const STALE_AFTER: Duration = Duration::from_secs(3);
/// Housekeeping tick — staleness only, so it can be lazy.
const HOUSEKEEPING: Duration = Duration::from_millis(500);
/// Upper bound on the SAF-2 sequence at app exit (NFR-9). Longer than the BLE
/// layer's own per-write timeouts, short enough not to hang a window close.
const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(8);
/// Upper bound on the client's own reset sequence, kept under
/// [`SHUTDOWN_TIMEOUT`] so the supervisor always answers before the caller
/// stops listening. A caller that times out learns nothing and leaves.
const SAFETY_SEQUENCE_TIMEOUT: Duration = Duration::from_secs(7);
/// How long auto-reconnect keeps trying before it gives up and says so.
///
/// FR-1.11. Retrying forever sounds kinder than giving up, but it is not: the
/// link sits in `Reconnecting` indefinitely, the screen keeps implying the
/// trainer is on its way back, and the rider is never told to go and look at
/// it. Twenty attempts against the default backoff is about seven minutes.
pub(crate) const RECONNECT_ATTEMPTS: u32 = 20;

/// What the UI needs to know about the trainer link (FR-1.7, FR-9.3).
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct TrainerStatus {
    pub state: ConnectionState,
    /// FTMS control point acquired. Deliberately *not* folded into `state`:
    /// connected is not controllable (FR-9.3).
    pub control_acquired: bool,
    pub address: Option<String>,
    pub name: Option<String>,
    /// Human-readable failure, rendered verbatim (FR-9.2).
    pub error: Option<String>,
    /// Connected, but no Indoor Bike Data for [`STALE_AFTER`]. The rider needs
    /// to know the numbers stopped being real (FR-1.8).
    pub stale: bool,
}

impl Default for TrainerStatus {
    fn default() -> Self {
        Self {
            state: ConnectionState::Idle,
            control_acquired: false,
            address: None,
            name: None,
            error: None,
            stale: false,
        }
    }
}

impl TrainerStatus {
    /// True once a ride would actually reach the trainer (FR-2.1).
    pub fn controllable(&self) -> bool {
        self.control_acquired && self.state == ConnectionState::Controlling
    }

    pub fn is_attached(&self) -> bool {
        !matches!(
            self.state,
            ConnectionState::Idle | ConnectionState::Lost { .. }
        )
    }
}

enum Cmd {
    Connect(TrainerSelector),
    Disconnect,
    Target(ControlTarget),
    /// SAF-2 without dropping the link: used at the end of a ride, so the next
    /// ride does not have to reconnect.
    Release {
        limits: SafetyLimits,
    },
    /// Full SAF-2 sequence plus disconnect. Used on app exit.
    Shutdown {
        reply: SyncSender<()>,
    },
    /// A spawned control write finished. Only reported when it failed.
    WriteFailed(String),
}

/// Cheap, cloneable handle to the supervisor.
#[derive(Clone)]
pub struct TrainerHandle {
    cmd_tx: mpsc::Sender<Cmd>,
    status_rx: watch::Receiver<TrainerStatus>,
    telemetry_rx: watch::Receiver<Telemetry>,
    /// Shared, so every clone of the handle sees that SAF-2 has already run.
    shut_down: Arc<AtomicBool>,
}

impl TrainerHandle {
    /// Start the supervisor task. Must be called with a Tokio runtime available
    /// — `tauri::async_runtime` provides one before the app is built.
    pub fn spawn(config: FtmsConfig) -> Self {
        let (cmd_tx, cmd_rx) = mpsc::channel(32);
        let (status_tx, status_rx) = watch::channel(TrainerStatus::default());
        let (telemetry_tx, telemetry_rx) = watch::channel(Telemetry::default());

        let handle = Self {
            cmd_tx: cmd_tx.clone(),
            status_rx,
            telemetry_rx,
            shut_down: Arc::new(AtomicBool::new(false)),
        };
        tauri::async_runtime::spawn(run(cmd_rx, cmd_tx, config, status_tx, telemetry_tx));
        handle
    }

    pub fn status(&self) -> TrainerStatus {
        self.status_rx.borrow().clone()
    }

    /// Latest telemetry, for the ride engine. Zeroed while disconnected or
    /// stale, never a held-over sample.
    pub fn telemetry(&self) -> watch::Receiver<Telemetry> {
        self.telemetry_rx.clone()
    }

    pub fn connect(&self, selector: TrainerSelector) {
        self.send(Cmd::Connect(selector));
    }

    pub fn disconnect(&self) {
        self.send(Cmd::Disconnect);
    }

    /// Push a control target. Fire-and-forget by design: the ride loop must not
    /// block on the radio, and [`FtmsClient`] already coalesces so the newest
    /// target wins (FR-2.8).
    pub fn set_target(&self, target: ControlTarget) {
        self.send(Cmd::Target(target));
    }

    /// SAF-2 at the end of a ride: zero gradient / minimum resistance, link kept.
    pub fn release(&self, limits: SafetyLimits) {
        self.send(Cmd::Release { limits });
    }

    /// SAF-2 at app exit: full reset sequence, then disconnect. Blocks the
    /// calling (non-async) thread until it is done or [`SHUTDOWN_TIMEOUT`]
    /// elapses.
    /// Idempotent: Tauri delivers `ExitRequested`, `Exit` and window
    /// `Destroyed` on one quit, and the second and third calls must be quiet
    /// no-ops rather than warnings about a supervisor that has already done its
    /// job.
    pub fn shutdown_blocking(&self) {
        if self.shut_down.swap(true, Ordering::SeqCst) {
            return;
        }
        let deadline = Instant::now() + SHUTDOWN_TIMEOUT;
        let (reply, done) = sync_channel(1);

        // Not `try_send` and give up: the one command that must not be dropped
        // is this one, and a queue that is briefly full — the ride loop pushes a
        // target every 250 ms — is not a reason to skip SAF-2. This is already a
        // blocking call, so spending a little of its budget on getting the
        // command in is free.
        let mut cmd = Cmd::Shutdown { reply };
        loop {
            match self.cmd_tx.try_send(cmd) {
                Ok(()) => break,
                Err(mpsc::error::TrySendError::Closed(_)) => {
                    tracing::warn!("trainer supervisor already stopped; nothing to release");
                    return;
                }
                Err(mpsc::error::TrySendError::Full(returned)) => {
                    if Instant::now() >= deadline {
                        tracing::error!("could not reach the trainer supervisor to run SAF-2");
                        return;
                    }
                    cmd = returned;
                    std::thread::sleep(Duration::from_millis(20));
                }
            }
        }

        let remaining = deadline.saturating_duration_since(Instant::now());
        match done.recv_timeout(remaining) {
            Ok(()) => tracing::info!("trainer released (SAF-2)"),
            Err(e) => tracing::error!(error = %e, "SAF-2 shutdown did not complete in time"),
        }
    }

    fn send(&self, cmd: Cmd) {
        if self.cmd_tx.try_send(cmd).is_err() {
            // Capacity 32 against a 4 Hz producer: a full queue means the
            // supervisor is wedged, which the status channel already reports.
            tracing::warn!("trainer command dropped — supervisor queue full or closed");
        }
    }
}

// ---------------------------------------------------------------------------
// Supervisor
// ---------------------------------------------------------------------------

/// The client's three output streams are kept as separate locals rather than
/// one struct, so each `select!` arm borrows a distinct binding.
async fn run(
    mut cmd_rx: mpsc::Receiver<Cmd>,
    cmd_tx: mpsc::Sender<Cmd>,
    config: FtmsConfig,
    status_tx: watch::Sender<TrainerStatus>,
    telemetry_tx: watch::Sender<Telemetry>,
) {
    let mut client: Option<Arc<FtmsClient>> = None;
    let mut telemetry_rx: Option<broadcast::Receiver<Telemetry>> = None;
    let mut events_rx: Option<broadcast::Receiver<FtmsEvent>> = None;
    let mut state_rx: Option<watch::Receiver<ConnectionState>> = None;
    let mut last_sample: Option<Instant> = None;
    let mut status = TrainerStatus::default();

    loop {
        tokio::select! {
            // Telemetry first (NFR-2): control writes are rate-limited anyway.
            biased;

            sample = next_telemetry(&mut telemetry_rx) => match sample {
                Some(sample) => {
                    last_sample = Some(Instant::now());
                    if status.stale {
                        status.stale = false;
                        publish(&status_tx, &status);
                    }
                    let _ = telemetry_tx.send(sample);
                }
                None => {
                    // The client actor stopped; the state watcher reports why.
                    telemetry_rx = None;
                }
            },

            event = next_event(&mut events_rx) => match event {
                Some(FtmsEvent::ControlFault { reason }) => {
                    tracing::error!(reason, "trainer control fault (SAF-4)");
                    status.control_acquired = false;
                    status.error = Some(reason);
                    publish(&status_tx, &status);
                }
                Some(_) => {}
                // NFR-10. A closed stream is *ready* forever, and this select is
                // biased: left in place it wins every poll and the command arm
                // below is never reached again — including for the shutdown
                // command. Drop it and let the state arm explain what happened.
                None => events_rx = None,
            },

            state = next_state(&mut state_rx) => match state {
                Some(state) => {
                    status.control_acquired = state == ConnectionState::Controlling;
                    if let ConnectionState::Lost { reason } = &state {
                        status.error = Some(reason.clone());
                        // A lost link must not leave the last sample standing.
                        let _ = telemetry_tx.send(Telemetry::default());
                        last_sample = None;
                    }
                    status.state = state;
                    publish(&status_tx, &status);
                }
                // The client actor stopped without being asked to — it spent its
                // reconnect budget (FR-1.11). Everything we hold is now a zombie:
                // the handle would fail one write at a time with nothing to
                // explain it, and the closed streams would spin the loop.
                None => {
                    telemetry_rx = None;
                    events_rx = None;
                    state_rx = None;
                    client = None;
                    last_sample = None;
                    let _ = telemetry_tx.send(Telemetry::default());
                    let reason = status.error.clone().unwrap_or_else(|| {
                        "the trainer link ended and could not be re-established".to_string()
                    });
                    tracing::warn!(reason, "trainer supervisor lost its client");
                    status.control_acquired = false;
                    status.stale = false;
                    // Address and name stay: the rider needs a row to click on
                    // to try again (FR-1.12).
                    status.state = ConnectionState::Lost { reason };
                    publish(&status_tx, &status);
                }
            },

            cmd = cmd_rx.recv() => match cmd {
                None => break,
                Some(Cmd::Shutdown { reply }) => {
                    let subs = (&mut telemetry_rx, &mut events_rx, &mut state_rx);
                    shutdown(client.take(), subs, &telemetry_tx).await;
                    status = TrainerStatus::default();
                    publish(&status_tx, &status);
                    let _ = reply.send(());
                    break;
                }
                Some(Cmd::Disconnect) => {
                    let subs = (&mut telemetry_rx, &mut events_rx, &mut state_rx);
                    shutdown(client.take(), subs, &telemetry_tx).await;
                    last_sample = None;
                    // Keep who it was. A trainer stops advertising the moment it
                    // is in use, so a scan will not necessarily put it back —
                    // wiping the address here drops it out of the device list
                    // altogether and leaves the rider nothing to click on to
                    // reconnect (FR-1.12).
                    let (address, name) = (status.address.take(), status.name.take());
                    status = TrainerStatus { address, name, ..TrainerStatus::default() };
                    publish(&status_tx, &status);
                }
                Some(Cmd::Connect(selector)) => {
                    // Authoritative guard against a duplicate click racing the
                    // status watch: reconnecting a live session would drop the
                    // rider's trainer mid-ride for no reason.
                    if client.is_some() && status.is_attached() && already_selected(&status, &selector) {
                        tracing::debug!(selector = selector.describe(), "already connected; ignoring");
                        continue;
                    }
                    let subs = (&mut telemetry_rx, &mut events_rx, &mut state_rx);
                    shutdown(client.take(), subs, &telemetry_tx).await;
                    last_sample = None;
                    status = TrainerStatus {
                        state: ConnectionState::Connecting,
                        ..TrainerStatus::default()
                    };
                    publish(&status_tx, &status);

                    // FR-1.10 / SAF-8. A connect runs for up to `scan_timeout`
                    // plus connect, discovery and handshake — far longer than
                    // the shutdown budget. Awaiting it here without a way out is
                    // what makes a quit-while-connecting skip SAF-2 entirely and
                    // leave the rider on a loaded trainer, so the attempt is
                    // abandoned instead of finished.
                    let mut abort: Option<Abort> = None;
                    let outcome = {
                        let cancel = async {
                            abort = Some(abort_signal(&mut cmd_rx, &selector).await);
                        };
                        FtmsClient::connect_cancellable(
                            selector.clone(),
                            config.clone(),
                            cancel,
                        )
                        .await
                    };

                    match outcome {
                        Ok(Some(c)) => {
                            tracing::info!(
                                address = c.address(),
                                name = c.name().unwrap_or("(no name)"),
                                caps = ?c.capabilities(),
                                "trainer connected and controllable"
                            );
                            telemetry_rx = Some(c.telemetry());
                            events_rx = Some(c.events());
                            state_rx = Some(c.state_stream());
                            status = TrainerStatus {
                                state: c.state(),
                                control_acquired: c.state() == ConnectionState::Controlling,
                                address: Some(c.address().to_string()),
                                name: c.name().map(str::to_string),
                                error: None,
                                stale: false,
                            };
                            client = Some(Arc::new(c));
                            last_sample = Some(Instant::now());
                        }
                        Ok(None) => {
                            tracing::info!(
                                selector = selector.describe(),
                                "connect abandoned; the link it had opened is closed"
                            );
                            status = TrainerStatus::default();
                        }
                        Err(e) => {
                            tracing::warn!(error = %e, selector = selector.describe(), "trainer connect failed");
                            status = TrainerStatus {
                                state: ConnectionState::Lost { reason: e.to_string() },
                                error: Some(connect_hint(&e)),
                                ..TrainerStatus::default()
                            };
                        }
                    }
                    publish(&status_tx, &status);

                    // Whatever interrupted the connect still has to be honoured.
                    match abort {
                        None => {}
                        Some(Abort::Disconnect) => {
                            let subs = (&mut telemetry_rx, &mut events_rx, &mut state_rx);
                            shutdown(client.take(), subs, &telemetry_tx).await;
                            last_sample = None;
                            let (address, name) = (status.address.take(), status.name.take());
                            status = TrainerStatus { address, name, ..TrainerStatus::default() };
                            publish(&status_tx, &status);
                        }
                        Some(Abort::Connect(next)) => {
                            // Back of the queue rather than a recursive call:
                            // the loop picks it up on its next turn, once this
                            // attempt has been fully unwound.
                            let subs = (&mut telemetry_rx, &mut events_rx, &mut state_rx);
                            shutdown(client.take(), subs, &telemetry_tx).await;
                            last_sample = None;
                            if cmd_tx.try_send(Cmd::Connect(next)).is_err() {
                                tracing::warn!("could not re-queue the trainer the rider picked");
                            }
                        }
                        Some(Abort::Shutdown(reply)) => {
                            let subs = (&mut telemetry_rx, &mut events_rx, &mut state_rx);
                            shutdown(client.take(), subs, &telemetry_tx).await;
                            status = TrainerStatus::default();
                            publish(&status_tx, &status);
                            let _ = reply.send(());
                            break;
                        }
                        Some(Abort::Closed) => break,
                    }
                }
                Some(Cmd::Target(target)) => {
                    if let Some(c) = client.clone() {
                        let back = cmd_tx.clone();
                        // Spawned, not awaited: an unacknowledged write takes up
                        // to `ack_timeout`, and the telemetry pump must keep
                        // running through it.
                        tauri::async_runtime::spawn(async move {
                            match c.set_target(target).await {
                                Ok(ControlOutcome::Acknowledged { sent }) => {
                                    tracing::debug!(?sent, "trainer accepted target")
                                }
                                Ok(ControlOutcome::Superseded) => {}
                                Err(e) => {
                                    tracing::warn!(?target, error = %e, "control write failed");
                                    let _ = back.try_send(Cmd::WriteFailed(e.to_string()));
                                }
                            }
                        });
                    }
                }
                Some(Cmd::Release { limits }) => {
                    if let Some(c) = client.clone() {
                        let safe = limits.clamp(ControlTarget::Gradient { percent: 0.0 });
                        tauri::async_runtime::spawn(async move {
                            if let Err(e) = c.set_target(safe).await {
                                tracing::warn!(error = %e, "SAF-2 release write failed");
                            }
                        });
                    }
                }
                Some(Cmd::WriteFailed(reason)) => {
                    if status.error.as_deref() != Some(reason.as_str()) {
                        status.error = Some(reason);
                        publish(&status_tx, &status);
                    }
                }
            },

            _ = tokio::time::sleep(HOUSEKEEPING) => {
                let quiet = last_sample.is_some_and(|t| t.elapsed() >= STALE_AFTER);
                if quiet && !status.stale && status.is_attached() {
                    tracing::warn!("no Indoor Bike Data for {STALE_AFTER:?} — zeroing telemetry");
                    status.stale = true;
                    publish(&status_tx, &status);
                    // Zero rather than hold: a frozen readout is worse than an
                    // obviously dead one.
                    let _ = telemetry_tx.send(Telemetry::default());
                }
            },
        }
    }

    tracing::info!("trainer supervisor stopped");
}

/// A command that arrived while a connect was in flight and means "stop"
/// (FR-1.10). It is carried out of the attempt rather than acted on there,
/// because the attempt has to be unwound first.
enum Abort {
    Disconnect,
    Shutdown(SyncSender<()>),
    /// The rider picked a different trainer while this one was still connecting.
    Connect(TrainerSelector),
    /// Every handle dropped.
    Closed,
}

/// Watch for a reason to abandon a connect that is currently running.
///
/// Resolves only on a command that must interrupt the attempt. Targets and
/// releases arriving mid-connect are consumed and dropped rather than left to
/// pile up: there is no link to write them to, and the ride loop re-sends its
/// target every tick. Draining is half the point — a command channel nobody
/// reads for fifteen seconds is a command channel that fills.
async fn abort_signal(cmd_rx: &mut mpsc::Receiver<Cmd>, connecting_to: &TrainerSelector) -> Abort {
    loop {
        match cmd_rx.recv().await {
            None => return Abort::Closed,
            Some(Cmd::Shutdown { reply }) => return Abort::Shutdown(reply),
            Some(Cmd::Disconnect) => return Abort::Disconnect,
            // Picking a *different* trainer means "not that one, this one".
            // Finishing the first attempt before starting the second would make
            // the rider wait out a scan timeout for a choice they have already
            // changed. The same trainer clicked twice is only impatience.
            Some(Cmd::Connect(selector)) if selector != *connecting_to => {
                return Abort::Connect(selector)
            }
            Some(_) => tracing::trace!("command dropped: nothing to send it to yet"),
        }
    }
}

/// Run SAF-2 and drop the client.
#[allow(clippy::type_complexity)]
async fn shutdown(
    client: Option<Arc<FtmsClient>>,
    subs: (
        &mut Option<broadcast::Receiver<Telemetry>>,
        &mut Option<broadcast::Receiver<FtmsEvent>>,
        &mut Option<watch::Receiver<ConnectionState>>,
    ),
    telemetry_tx: &watch::Sender<Telemetry>,
) {
    *subs.0 = None;
    *subs.1 = None;
    *subs.2 = None;
    let _ = telemetry_tx.send(Telemetry::default());
    let Some(client) = client else { return };
    // Bounded, so the supervisor always answers `shutdown_blocking` before that
    // caller stops listening. A caller that times out reports a failure it
    // cannot do anything about and then lets the process exit anyway.
    match tokio::time::timeout(SAFETY_SEQUENCE_TIMEOUT, client.shutdown_ref()).await {
        Ok(Ok(())) => {}
        Ok(Err(e)) => tracing::warn!(error = %e, "SAF-2 shutdown reported an error"),
        Err(_) => tracing::error!("SAF-2 shutdown did not finish inside its budget"),
    }
}

/// Does `selector` name the trainer we are already attached to?
fn already_selected(status: &TrainerStatus, selector: &TrainerSelector) -> bool {
    match selector {
        TrainerSelector::Any => true,
        TrainerSelector::Address(a) => status
            .address
            .as_deref()
            .is_some_and(|current| current.eq_ignore_ascii_case(a)),
        TrainerSelector::NameContains(n) => status
            .name
            .as_deref()
            .is_some_and(|current| current.to_lowercase().contains(&n.to_lowercase())),
    }
}

fn publish(tx: &watch::Sender<TrainerStatus>, status: &TrainerStatus) {
    let _ = tx.send(status.clone());
}

/// FR-1.8: an empty scan means "nothing was advertising", and the fix is almost
/// always to pedal. Say so rather than reporting a bare error.
fn connect_hint(e: &FtmsError) -> String {
    match e {
        FtmsError::NotFound(_) => {
            "Trainer not found. It only advertises once awake — turn the cranks for a few \
             seconds and try again."
                .to_string()
        }
        FtmsError::NoAdapter => {
            "No Bluetooth adapter. Check the radio is on and BlueZ is running.".to_string()
        }
        // A-3: the D100 accepts one BLE host. A second one gets the link torn
        // down mid-handshake, which surfaces from BlueZ as a bare "Not
        // connected" — the least informative possible description of the most
        // common real-world failure.
        FtmsError::Bluetooth(_) | FtmsError::MissingCharacteristic(_) => format!(
            "Trainer is busy: {e}. It accepts one connection at a time — close any other app \
             (or `probe`) that is holding it, then try again."
        ),
        FtmsError::Rejected { .. } | FtmsError::Unacknowledged { .. } => format!(
            "{e}. Another app may already hold control of the trainer — only one may at a time."
        ),
        other => other.to_string(),
    }
}

// -- select! helpers ---------------------------------------------------------
//
// Each parks forever when there is no client, so the other arms drive the loop.
// The one thing none of them may do is return `Ready` on every poll: the select
// they feed is `biased`, so a permanently-ready arm starves every arm below it
// (NFR-10). That is why a closed stream is reported once and the caller then
// clears the receiver.

async fn next_telemetry(rx: &mut Option<broadcast::Receiver<Telemetry>>) -> Option<Telemetry> {
    match rx {
        None => std::future::pending().await,
        Some(rx) => loop {
            match rx.recv().await {
                Ok(t) => return Some(t),
                // NFR-2: a lagged consumer skips ahead, it does not stall.
                Err(broadcast::error::RecvError::Lagged(n)) => {
                    tracing::debug!(skipped = n, "telemetry consumer lagged")
                }
                Err(broadcast::error::RecvError::Closed) => return None,
            }
        },
    }
}

async fn next_event(rx: &mut Option<broadcast::Receiver<FtmsEvent>>) -> Option<FtmsEvent> {
    match rx {
        None => std::future::pending().await,
        Some(rx) => loop {
            match rx.recv().await {
                Ok(e) => return Some(e),
                Err(broadcast::error::RecvError::Lagged(_)) => {}
                Err(broadcast::error::RecvError::Closed) => return None,
            }
        },
    }
}

async fn next_state(rx: &mut Option<watch::Receiver<ConnectionState>>) -> Option<ConnectionState> {
    match rx {
        None => std::future::pending().await,
        Some(rx) => match rx.changed().await {
            Ok(()) => Some(rx.borrow_and_update().clone()),
            Err(_) => None,
        },
    }
}

/// The FTMS configuration this app rides with.
///
/// `use_simulation_mode` is **true**, unlike the crate default: the D100
/// advertises `SetIndoorBikeSimulationParameters` (`0x11`, target feature bit
/// 13) and accepts it, while it does *not* advertise `SetTargetInclination`
/// (`0x03`) and its inclination range reports 0–6 % with no negatives — useless
/// for descents. Measured against firmware 0.108; see README.md.
pub fn app_config(limits: SafetyLimits) -> FtmsConfig {
    FtmsConfig {
        limits,
        use_simulation_mode: true,
        backoff: Backoff {
            max_attempts: Some(RECONNECT_ATTEMPTS),
            ..Backoff::default()
        },
        ..FtmsConfig::default()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn control_is_tracked_separately_from_connection() {
        // FR-9.3: connected is not controllable.
        let connected = TrainerStatus {
            state: ConnectionState::Connected,
            control_acquired: false,
            ..TrainerStatus::default()
        };
        assert!(connected.is_attached());
        assert!(!connected.controllable());

        let controlling = TrainerStatus {
            state: ConnectionState::Controlling,
            control_acquired: true,
            ..TrainerStatus::default()
        };
        assert!(controlling.controllable());
    }

    #[test]
    fn a_lost_link_is_not_attached() {
        let lost = TrainerStatus {
            state: ConnectionState::Lost {
                reason: "gone".into(),
            },
            ..TrainerStatus::default()
        };
        assert!(!lost.is_attached());
        assert!(!lost.controllable());
        assert!(!TrainerStatus::default().is_attached());
    }

    #[test]
    fn the_app_drives_gradient_through_simulation_mode() {
        // The D100 rejects 0x03 and accepts 0x11 — measured, see README.
        let cfg = app_config(SafetyLimits::default());
        assert!(cfg.use_simulation_mode);
        assert!(
            !cfg.ignore_advertised_features,
            "FR-2.6 is not for the app to bypass"
        );
        assert!(cfg.start_on_connect);
        assert!(
            cfg.min_write_interval >= Duration::from_millis(250),
            "FR-2.8"
        );
    }

    #[test]
    fn safety_limits_reach_the_ble_layer() {
        let limits = SafetyLimits {
            max_gradient_pct: 8.0,
            ..SafetyLimits::default()
        };
        assert_eq!(app_config(limits).limits.max_gradient_pct, 8.0);
    }

    #[test]
    fn a_missing_trainer_is_explained_not_just_reported() {
        let hint = connect_hint(&FtmsError::NotFound("any FTMS trainer".into()));
        assert!(hint.to_lowercase().contains("crank"), "FR-1.8: {hint}");
        let hint = connect_hint(&FtmsError::NoAdapter);
        assert!(hint.to_lowercase().contains("bluetooth"));
    }

    #[test]
    fn a_second_connect_to_the_same_trainer_is_recognised() {
        // A duplicate click must not tear down a live session.
        let status = TrainerStatus {
            state: ConnectionState::Controlling,
            control_acquired: true,
            address: Some("94:05:bb:04:76:28".into()),
            name: Some("VANRYSEL-HT-2876".into()),
            ..TrainerStatus::default()
        };
        assert!(already_selected(
            &status,
            &TrainerSelector::Address("94:05:BB:04:76:28".into())
        ));
        assert!(already_selected(
            &status,
            &TrainerSelector::NameContains("vanrysel".into())
        ));
        assert!(already_selected(&status, &TrainerSelector::Any));
        // A different trainer is a genuine reconnect.
        assert!(!already_selected(
            &status,
            &TrainerSelector::Address("aa:bb:cc:dd:ee:ff".into())
        ));
        assert!(!already_selected(
            &TrainerStatus::default(),
            &TrainerSelector::Address("94:05:bb:04:76:28".into())
        ));
    }

    #[tokio::test]
    async fn a_shutdown_interrupts_a_connect_instead_of_queueing_behind_it() {
        // FR-1.10 / SAF-8. A connect runs for up to `scan_timeout` plus the
        // handshake; reading the shutdown command only after it finished is what
        // made a quit-while-connecting blow its budget and skip SAF-2 entirely.
        let (tx, mut rx) = mpsc::channel(8);
        let (reply, done) = sync_channel(1);

        // Commands that are not a reason to stop are drained, not left to fill
        // the queue while nobody is reading it.
        tx.send(Cmd::Target(ControlTarget::Gradient { percent: 3.0 }))
            .await
            .unwrap();
        tx.send(Cmd::Release {
            limits: SafetyLimits::default(),
        })
        .await
        .unwrap();
        tx.send(Cmd::Shutdown { reply }).await.unwrap();

        match abort_signal(&mut rx, &TrainerSelector::Any).await {
            Abort::Shutdown(reply) => {
                // The caller is blocking on this; it has to be answerable.
                let _ = reply.send(());
                assert!(done.recv().is_ok());
            }
            _ => panic!("the shutdown did not interrupt the connect"),
        }
    }

    #[tokio::test]
    async fn a_disconnect_interrupts_a_connect_too() {
        // Otherwise the rider's Disconnect click does nothing for fifteen
        // seconds and then tears down the link it just finished building.
        let (tx, mut rx) = mpsc::channel(8);
        tx.send(Cmd::Disconnect).await.unwrap();
        assert!(matches!(
            abort_signal(&mut rx, &TrainerSelector::Any).await,
            Abort::Disconnect
        ));
    }

    #[tokio::test]
    async fn picking_a_different_trainer_mid_connect_takes_over() {
        // Otherwise changing your mind costs a whole scan timeout, and the
        // trainer you clicked second is connected to only after the one you
        // gave up on has finished failing.
        let connecting_to = TrainerSelector::Address("94:05:bb:04:76:28".into());
        let (tx, mut rx) = mpsc::channel(8);

        // The same trainer clicked again is impatience, not a new intent — it
        // must not restart the attempt already running.
        tx.send(Cmd::Connect(connecting_to.clone())).await.unwrap();
        tx.send(Cmd::Connect(TrainerSelector::Address(
            "aa:bb:cc:dd:ee:ff".into(),
        )))
        .await
        .unwrap();

        match abort_signal(&mut rx, &connecting_to).await {
            Abort::Connect(TrainerSelector::Address(a)) => assert_eq!(a, "aa:bb:cc:dd:ee:ff"),
            _ => panic!("the second trainer did not take over"),
        }
    }

    #[tokio::test]
    async fn dropping_every_handle_ends_a_connect() {
        let (tx, mut rx) = mpsc::channel::<Cmd>(1);
        drop(tx);
        assert!(matches!(
            abort_signal(&mut rx, &TrainerSelector::Any).await,
            Abort::Closed
        ));
    }

    #[test]
    fn reconnect_is_bounded_so_the_rider_is_eventually_told() {
        // FR-1.11. Retrying forever sounds kinder than giving up, but it leaves
        // the link in `Reconnecting` indefinitely and never says the trainer is
        // gone — so the rider is never prompted to go and look at it.
        let cfg = app_config(SafetyLimits::default());
        assert_eq!(cfg.backoff.max_attempts, Some(RECONNECT_ATTEMPTS));
        assert!(cfg.backoff.exhausted(RECONNECT_ATTEMPTS));
        assert!(!cfg.backoff.exhausted(RECONNECT_ATTEMPTS - 1));
    }

    #[test]
    fn the_supervisor_answers_before_its_caller_stops_listening() {
        // NFR-9. If the inner budget outlasted the outer one, every quit would
        // report a timeout for a sequence that was about to succeed — and then
        // let the process exit on top of it anyway.
        assert!(SAFETY_SEQUENCE_TIMEOUT < SHUTDOWN_TIMEOUT);
    }

    #[test]
    fn a_trainer_held_by_another_app_says_so() {
        // A-3: one BLE host. BlueZ reports the second one's torn-down link as a
        // bare "Not connected", which explains nothing on its own.
        let hint = connect_hint(&FtmsError::MissingCharacteristic(
            "Indoor Bike Data (0x2AD2)",
        ));
        assert!(hint.contains("one connection at a time"), "{hint}");
        assert!(hint.to_lowercase().contains("busy"), "{hint}");
    }
}