navcore_signalk_client/

onboard.rs

//! Getting let in, and staying that way -- the one sequence every Signal K
//! client goes through before it has anything else to do: find a server,
//! connect if there is already a token, ask for one otherwise, wait for a
//! human, and start over under a fresh token if the server ever takes an
//! old one back.
//!
//! [`Onboard`] is that sequence, factored out of a first client's own
//! delta-feed module, where it was worked out and tested first. Two
//! applications built on Signal K do not differ in how any of this goes
//! -- only in what they do
//! with a delta once one arrives, which stays each application's own
//! [`Onboard::poll`] never sees a [`crate::stream::Event::Connected`],
//! [`crate::stream::Event::Disconnected`] or
//! [`crate::stream::Event::Unauthorized`] leak out to a caller: this module
//! is exactly the place that already knows what each one means, and a
//! caller with its own copy of that knowledge is the duplication this
//! exists to end. Only [`signalk::Delta`] comes out the other side.
//!
//! # What stays with the application
//!
//! Where a device token lives on disk, what GSettings a mariner's own
//! discovery preference is in, and how a raw address is resolved when
//! discovery is not used -- none of that is here, the same "not this
//! crate's job" boundary the rest of `signalk-client` already draws (see
//! its own module doc). [`Credentials`] is the seam: an application
//! implements four small file operations, and hands [`Onboard::seek`] a
//! closure for finding a server however it prefers to. Everything about
//! *sequencing* those into "connected, or why not" is what this module
//! does instead of every application working it out for itself again.

use std::sync::Arc;
use std::sync::mpsc::{Receiver, TryRecvError, channel};
use std::thread;
use std::time::Duration;

use crate::access::{Access, Permissions, PendingRequest, Requested, check_access, request_access};
use crate::discovery::Server;
use crate::stream::{Connection, Event};
use crate::token::Token;
use crate::trust::Trust;
use signalk::Delta;

/// How often to ask the server again whether a human has decided yet.
const APPROVAL_POLL: Duration = Duration::from_secs(2);

/// How long to keep asking before giving up on this run. Approval
/// requires a human to act on the server's own admin interface.
const APPROVAL_PATIENCE: Duration = Duration::from_secs(600);

/// What an application keeps on this desktop so [`Onboard`] does not have
/// to: a device token, the identity it asks under, and a request it is
/// still waiting to hear back about. Implemented once per application --
/// see the module's own doc for why storage itself stays out of this
/// crate.
///
/// `Send + Sync + 'static`: [`Onboard::seek`] moves the implementor onto a
/// background thread, alongside the network call it accompanies.
pub trait Credentials: Send + Sync {
    /// The saved device token, if there is one.
    fn token(&self) -> Option<String>;
    /// Saves a freshly issued token. Errors are the caller's to log: a
    /// token that fails to save still works for the rest of this run, it
    /// just will not survive a restart.
    fn save_token(&self, token: &str) -> std::io::Result<std::path::PathBuf>;
    /// Throws the saved token away, so the next start asks again.
    fn forget_token(&self);
    /// What this installation calls itself to the server -- made once and
    /// kept, so the same device keeps its place in the administrator's
    /// list across restarts.
    fn client_id(&self) -> String;
    /// The request `identity` is still waiting on, if one was saved.
    fn pending_request(&self, identity: &str) -> Option<PendingRequest>;
    /// Remembers a request just sent.
    fn save_pending_request(&self, identity: &str, request: &PendingRequest);
    /// Drops the record of a request once it has been decided.
    fn clear_pending_request(&self);
}

/// How the socket itself is doing, apart from whether a human has
/// approved this device yet -- see [`Onboarding`] for that half.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Link {
    /// Listening for a server, or connecting to one already found.
    Reaching,
    /// Connected.
    Up,
    /// Not connected, and why.
    Down(String),
}

/// Where this device stands on its way to a first delta.
///
/// Worth showing to a caller directly: approval can take minutes, and
/// an interface that stays empty in the meantime is indistinguishable
/// from a broken one.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Onboarding {
    /// Listening for the server, or opening the connection to one.
    Connecting,
    /// Waiting for a human to approve this device, under the name it
    /// asked as.
    Asking {
        /// What the administrator will see in the approval list.
        device: String,
    },
    /// The administrator looked and said no.
    Denied {
        /// The name that was refused.
        device: String,
    },
    /// The server already holds a request under this name that this run
    /// has no way to poll -- typically a stray one left over from an
    /// earlier run whose own record of it was lost. Unlike [`Self::Asking`],
    /// this run will *not* notice an approval on its own; deciding it on
    /// the server clears the way, but this run itself needs a restart to
    /// try again after that.
    AlreadyPending {
        /// The name whose earlier request is being waited on.
        device: String,
    },
    /// Gave up before ever connecting, in log words rather than
    /// translated -- there is no small set of reasons to enumerate here,
    /// only whatever the network or the server said.
    GaveUp(String),
    /// The token this installation held was refused with `401
    /// Unauthorized`, discarded, and a fresh request just went out under
    /// the same name.
    Reauthorizing {
        /// What the fresh request is going out under.
        device: String,
        /// Whether the discarded token's own `exp` claim said it had
        /// already expired, read before it was thrown away. `false` also
        /// covers a token with no expiry claim at all, refused for some
        /// other reason -- revoked by the administrator, most likely --
        /// that only the server's own log would show; a `401` alone
        /// cannot tell the two apart.
        was_expired: bool,
    },
}

/// Whether this run has ever gotten in.
///
/// Once [`Link::Up`] has happened at least once, [`Onboard::stage`] stops
/// reporting the two [`Onboarding`] states that resolve on their own --
/// [`Onboarding::Connecting`] and [`Onboarding::GaveUp`] -- and leaves an
/// ordinary dropped link to [`Link`] (and a caller's own "data is going
/// stale") to show instead, so the two are never reconciled on screen at
/// once. The states that need a human to act again -- [`Onboarding::Asking`],
/// [`Onboarding::Denied`], [`Onboarding::AlreadyPending`],
/// [`Onboarding::Reauthorizing`] -- are reported regardless of whether
/// this is the first time or the fifth: needing a fresh approval reads
/// the same either way, and a caller that only ever asks for approval
/// once would leave a mariner staring at "reconnecting" with nothing
/// telling them Signal K is waiting on a click.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Stage {
    /// Nothing has arrived yet, or is expected any moment: what to say
    /// instead, when there is nothing else worth showing.
    Onboarding(Onboarding),
    /// Past all of that. [`Link`] owns the story from here.
    Underway,
}

/// What became of asking for access.
///
/// Apart from [`Access`]: this also has to say "gave up after ten
/// minutes" and "could not even ask", neither of which the server itself
/// reports.
enum Sought {
    Approved(String),
    /// A decision, worth showing differently from an outage.
    Denied,
    /// The server already holds a request under this identity that this
    /// run has no id for -- see [`Onboarding::AlreadyPending`]. Nothing
    /// went wrong; there is simply nothing left for *this run* to poll.
    StillPending,
    /// Could not ask, could not find out, or nobody decided in time.
    Failed(String),
}

/// Finding a server, connecting to it, and staying let in.
pub struct Onboard {
    connection: Option<Connection>,
    finding: Option<Receiver<Option<Server>>>,
    asking: Option<Receiver<Sought>>,
    server: Option<Server>,
    token: Option<String>,
    trust: Trust,
    link: Link,
    identity: Option<String>,
    denied: bool,
    already_pending: bool,
    ever_connected: bool,
    reauthorizing: Option<bool>,
    credentials: Arc<dyn Credentials>,
    human_name: &'static str,
    permissions: Permissions,
    log_prefix: &'static str,
    wake: Arc<dyn Fn() + Send + Sync>,
}

impl Onboard {
    /// Starts looking for the server. Returns at once: `find` runs on a
    /// thread of its own, because discovery can take several seconds and
    /// a name lookup can hang.
    ///
    /// `credentials` is where a token, this device's own name and a
    /// pending request are kept -- see [`Credentials`]'s own doc.
    /// `human_name` is what an administrator sees in the device list's
    /// own description column, e.g. `"AIS target list"`. `log_prefix`
    /// tags this application's own log lines, e.g. `"ais-list"`. `wake`
    /// is passed to every [`Connection`] this opens -- see
    /// [`Connection::open`] for what it is for.
    #[must_use]
    pub fn seek(
        find: impl FnOnce() -> Option<Server> + Send + 'static,
        credentials: Arc<dyn Credentials>,
        trust: Trust,
        human_name: &'static str,
        permissions: Permissions,
        log_prefix: &'static str,
        wake: impl Fn() + Send + Sync + 'static,
    ) -> Self {
        let (sender, finding) = channel();
        thread::spawn(move || {
            let _ = sender.send(find());
        });

        Self {
            connection: None,
            finding: Some(finding),
            asking: None,
            server: None,
            token: credentials.token(),
            trust,
            link: Link::Reaching,
            identity: None,
            denied: false,
            already_pending: false,
            ever_connected: false,
            reauthorizing: None,
            credentials,
            human_name,
            permissions,
            log_prefix,
            wake: Arc::new(wake),
        }
    }

    /// How the link is doing.
    #[must_use]
    pub fn link(&self) -> &Link {
        &self.link
    }

    /// The server this run found, once discovery has answered.
    #[must_use]
    pub fn server(&self) -> Option<&Server> {
        self.server.as_ref()
    }

    /// The authority this run trusts -- for a caller making its own REST
    /// calls (fetching display units, say) against the same server this
    /// opened its stream to, rather than reloading it from disk again.
    #[must_use]
    pub fn trust(&self) -> &Trust {
        &self.trust
    }

    /// Where this device stands. See [`Stage`]'s own doc for why this
    /// stops answering [`Stage::Onboarding`] once the link has ever come
    /// up.
    #[must_use]
    pub fn stage(&self) -> Stage {
        let onboarding = if self.denied {
            Some(Onboarding::Denied { device: self.identity.clone().unwrap_or_default() })
        } else if self.already_pending {
            Some(Onboarding::AlreadyPending { device: self.identity.clone().unwrap_or_default() })
        } else if let Some(was_expired) = self.reauthorizing {
            Some(Onboarding::Reauthorizing { device: self.identity.clone().unwrap_or_default(), was_expired })
        } else if self.asking.is_some() {
            Some(Onboarding::Asking { device: self.identity.clone().unwrap_or_default() })
        } else if self.ever_connected {
            // An ordinary drop -- the socket alone, nothing waiting on a
            // human -- is staleness's story to tell from here on, not
            // this one's; see `Stage`'s own doc for why the two must
            // never be shown at once. A state that *does* need a human
            // (denied, already pending, reauthorizing, asking) is
            // reported above regardless of `ever_connected`: needing a
            // fresh approval is exactly the same situation the first
            // time or the fifth, and hiding it behind "reconnecting"
            // would leave a mariner waiting on a click nobody told them
            // to make.
            None
        } else if let Link::Down(reason) = &self.link {
            Some(Onboarding::GaveUp(reason.clone()))
        } else {
            Some(Onboarding::Connecting)
        };

        onboarding.map_or(Stage::Underway, Stage::Onboarding)
    }

    /// Picks up whatever has arrived since last time, and returns every
    /// delta read off the stream -- own ship's or another vessel's, this
    /// module does not look. Everything else the stream can say
    /// ([`Event::Connected`], [`Event::Disconnected`],
    /// [`Event::Unauthorized`]) is acted on here and never handed out;
    /// see the module's own doc for why.
    pub fn poll(&mut self) -> Vec<Delta> {
        self.take_server();
        self.take_approval();

        let events: Vec<Event> = match &self.connection {
            // Drained into a batch first: reading the channel borrows the
            // connection, and handling what comes out of it changes self.
            Some(connection) => connection.drain().collect(),
            None => Vec::new(),
        };

        let mut deltas = Vec::new();
        for event in events {
            match event {
                Event::Connected { .. } => self.note(Link::Up),
                Event::Disconnected { reason } => self.note(Link::Down(reason)),
                Event::Unauthorized => self.reauthorize(),
                Event::Delta(delta) => deltas.push(delta),
            }
        }
        deltas
    }

    /// Queues `delta_json` to be sent on the wire, as-is -- for an
    /// application that also publishes, not only reads.
    ///
    /// # Errors
    ///
    /// If there is no open connection to send it on, or the send itself
    /// fails.
    pub fn publish(&self, delta_json: String) -> Result<(), crate::ClientError> {
        let connection = self
            .connection
            .as_ref()
            .ok_or_else(|| crate::ClientError::Http("no connection is open".to_owned()))?;
        connection.publish(delta_json)
    }

    /// Records a change of link state, and says so once.
    ///
    /// Once, not every tick: a link that is down stays down for as long
    /// as the weather lasts, and a line a second would bury everything
    /// else.
    fn note(&mut self, link: Link) {
        if self.link == link {
            return;
        }
        let prefix = self.log_prefix;
        match &link {
            Link::Reaching => eprintln!("[{prefix}] looking for a Signal K server"),
            Link::Up => {
                eprintln!("[{prefix}] Signal K link up");
                self.ever_connected = true;
            }
            Link::Down(reason) => eprintln!("[{prefix}] Signal K link down: {reason}"),
        }
        self.link = link;
    }

    /// Acts on what discovery answered: connects, or asks to be let in.
    fn take_server(&mut self) {
        let Some(finding) = &self.finding else {
            return;
        };
        match finding.try_recv() {
            Err(TryRecvError::Empty) => return,
            Ok(Some(server)) => {
                eprintln!(
                    "[{}] found {} at {}",
                    self.log_prefix,
                    server.vessel_name.as_deref().unwrap_or("a vessel"),
                    server.http_base()
                );
                if self.token.is_some() {
                    self.connect(&server);
                } else {
                    self.ask(&server);
                }
                self.server = Some(server);
            }
            Ok(None) => self.note(Link::Down("no server answered".to_owned())),
            Err(TryRecvError::Disconnected) => self.note(Link::Down("discovery gave up".to_owned())),
        }
        self.finding = None;
    }

    /// Opens the stream, with whatever token this installation holds.
    fn connect(&mut self, server: &Server) {
        let wake = Arc::clone(&self.wake);
        self.connection = Some(Connection::open(&server.stream_url(), self.token.clone(), &self.trust, move || wake()));
    }

    /// Asks the server to let this installation in, on a thread of its
    /// own -- off the main loop because the wait is for a human, and a
    /// user interface that stopped meanwhile would look broken at
    /// exactly the moment it is working.
    fn ask(&mut self, server: &Server) {
        // Cleared unconditionally, not only on the paths that mean to
        // set it: this is the one place every kind of ask -- first-ever,
        // a manual forget, a reauthorize after a 401 -- passes through,
        // so it is the one place that can promise a *previous* cycle's
        // reason never bleeds into this one. `Self::reauthorize` sets it
        // again immediately afterwards, for its own call only.
        self.reauthorizing = None;

        let identity = self.credentials.client_id();
        self.identity = Some(identity.clone());

        if self.credentials.pending_request(&identity).is_some() {
            eprintln!(
                "[{}] still waiting to be let in as \"{identity}\" -- resuming the earlier request",
                self.log_prefix
            );
        } else {
            eprintln!(
                "[{0}] no token yet -- asking to be let in as \"{identity}\".\n\
                 [{0}] approve it in the Signal K server's Security > Devices list.",
                self.log_prefix
            );
        }

        let (sender, asking) = channel();
        let http_base = server.http_base();
        let trust = self.trust.clone();
        let credentials = Arc::clone(&self.credentials);
        let human_name = self.human_name;
        let permissions = self.permissions;
        thread::spawn(move || {
            let _ = sender.send(seek_approval(&http_base, &identity, human_name, permissions, &trust, &*credentials));
        });

        self.asking = Some(asking);
        self.note(Link::Reaching);
    }

    /// Reacts to [`Event::Unauthorized`]: discards the token this
    /// connection was opened with and immediately asks for a new one,
    /// under the same name as before.
    ///
    /// A `401` during the handshake means exactly one thing either way --
    /// this token is not getting back in -- so there is nothing to wait
    /// for the way [`Link::Down`] is: the dead connection is dropped here
    /// rather than left to retry itself into the same wall.
    fn reauthorize(&mut self) {
        // Read before the token is thrown away -- once it is gone, there
        // is nothing left to read the claim from. See
        // [`Onboarding::Reauthorizing`]'s own doc for why this is the
        // honest half of the story and not the whole of it.
        let was_expired = self.token.as_deref().and_then(Token::read).is_some_and(|token| {
            let now = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map_or(0, |elapsed| elapsed.as_secs() as i64);
            token.expired(now)
        });

        self.credentials.forget_token();
        self.token = None;
        // Dropping asks its own background thread to stop, but that
        // thread has already stopped itself on this exact event -- see
        // `crate::stream`'s own doc on why a `401` is not retried.
        self.connection = None;
        self.note(Link::Down(if was_expired {
            "the token had expired".to_owned()
        } else {
            "the token was no longer accepted".to_owned()
        }));

        if let Some(server) = self.server.clone() {
            self.ask(&server);
            self.reauthorizing = Some(was_expired);
        }
        // If there is no server yet, there is nothing to ask -- but
        // `Event::Unauthorized` only ever arrives from a connection this
        // opened against a known server, so in practice this branch is
        // never reached.
    }

    /// Picks up an approval once it has been granted, and connects.
    fn take_approval(&mut self) {
        let Some(asking) = &self.asking else {
            return;
        };
        let device = self.identity.clone().unwrap_or_default();
        match asking.try_recv() {
            Err(TryRecvError::Empty) => return,
            Ok(Sought::Approved(token)) => {
                // Written down before anything is done with it: the
                // server issues a token exactly once.
                match self.credentials.save_token(&token) {
                    Ok(path) => eprintln!("[{}] approved -- token saved in {}", self.log_prefix, path.display()),
                    Err(error) => eprintln!(
                        "[{}] approved, but the token could NOT be saved ({error}). \
                         It will not survive a restart, and Signal K issues it only once.",
                        self.log_prefix
                    ),
                }
                self.token = Some(token);
                if let Some(server) = self.server.clone() {
                    self.connect(&server);
                }
            }
            Ok(Sought::Denied) => {
                self.denied = true;
                self.note(Link::Down("the access request was refused".to_owned()));
            }
            Ok(Sought::StillPending) => {
                // Not a link failure -- nothing was wrong with the
                // network or the server, only a stray request from an
                // earlier run that this one has no id to poll. Link is
                // left alone on purpose: see `Onboarding::AlreadyPending`.
                self.already_pending = true;
                eprintln!(
                    "[{}] a request for \"{device}\" is already waiting on the server; \
                     approve or deny it there, then restart",
                    self.log_prefix
                );
            }
            Ok(Sought::Failed(reason)) => self.note(Link::Down(reason)),
            Err(TryRecvError::Disconnected) => self.note(Link::Down("the access request gave up".to_owned())),
        }
        self.asking = None;
    }
}

/// Asks to be let in, then waits for a human to decide.
///
/// Resumes a request already sent rather than sending a second one: a
/// restart taken mid-wait is the ordinary way to reach this, and the
/// server rightly refuses a second request under a name that already has
/// one outstanding.
fn seek_approval(
    http_base: &str,
    identity: &str,
    human_name: &str,
    permissions: Permissions,
    trust: &Trust,
    credentials: &dyn Credentials,
) -> Sought {
    let request = match credentials.pending_request(identity) {
        Some(request) => request,
        None => {
            let description = format!("{human_name} ({identity})");
            match request_access(http_base, identity, &description, permissions, trust) {
                Ok(Requested::New(request)) => {
                    credentials.save_pending_request(identity, &request);
                    request
                }
                Ok(Requested::AlreadyPending) => return Sought::StillPending,
                Err(error) => return Sought::Failed(format!("could not ask for access: {error}")),
            }
        }
    };

    let deadline = std::time::Instant::now() + APPROVAL_PATIENCE;
    while std::time::Instant::now() < deadline {
        match check_access(http_base, &request, trust) {
            Ok(Access::Approved { token }) => {
                credentials.clear_pending_request();
                return Sought::Approved(token);
            }
            Ok(Access::Denied) => {
                credentials.clear_pending_request();
                return Sought::Denied;
            }
            Ok(Access::Pending) => thread::sleep(APPROVAL_POLL),
            // Not resumed next run: a request the server can no longer
            // answer about (found live: an HTTP 500 for one its own
            // admin page no longer listed) would fail the same way every
            // start, forever. Asking fresh is the better bet.
            Err(error) => {
                credentials.clear_pending_request();
                return Sought::Failed(format!("asking about the request: {error}"));
            }
        }
    }

    credentials.clear_pending_request();
    Sought::Failed("nobody approved the access request in time".to_owned())
}

#[cfg(test)]
mod tests {
    use std::sync::mpsc::channel;

    use super::*;

    /// A [`Credentials`] that keeps nothing -- these tests exercise
    /// [`Onboard::stage`] and the pure half of [`Onboard::reauthorize`],
    /// neither of which reads one back.
    struct NullCredentials;

    impl Credentials for NullCredentials {
        fn token(&self) -> Option<String> {
            None
        }

        fn save_token(&self, _token: &str) -> std::io::Result<std::path::PathBuf> {
            Ok(std::path::PathBuf::new())
        }

        fn forget_token(&self) {}

        fn client_id(&self) -> String {
            "test".to_owned()
        }

        fn pending_request(&self, _identity: &str) -> Option<PendingRequest> {
            None
        }

        fn save_pending_request(&self, _identity: &str, _request: &PendingRequest) {}

        fn clear_pending_request(&self) {}
    }

    /// An [`Onboard`] in whatever state a test wants, without the thread
    /// [`Onboard::seek`] would spawn.
    fn onboard(link: Link) -> Onboard {
        Onboard {
            connection: None,
            finding: None,
            asking: None,
            server: None,
            token: None,
            trust: Trust::plaintext(),
            link,
            identity: None,
            denied: false,
            already_pending: false,
            ever_connected: false,
            reauthorizing: None,
            credentials: Arc::new(NullCredentials),
            human_name: "Test",
            permissions: Permissions::ReadOnly,
            log_prefix: "test",
            wake: Arc::new(|| {}),
        }
    }

    #[test]
    fn a_fresh_onboard_is_onboarding_and_connecting() {
        assert_eq!(onboard(Link::Reaching).stage(), Stage::Onboarding(Onboarding::Connecting));
    }

    #[test]
    fn asking_names_the_device_being_asked_for() {
        let mut state = onboard(Link::Reaching);
        state.identity = Some("test-device".to_owned());
        let (_sender, asking) = channel();
        state.asking = Some(asking);

        assert_eq!(state.stage(), Stage::Onboarding(Onboarding::Asking { device: "test-device".to_owned() }));
    }

    #[test]
    fn a_denial_is_told_apart_from_an_ordinary_outage() {
        // Both leave the link Down with a sentence; only `denied` tells a
        // decision from a dropped connection, and take_approval is what
        // sets it -- checked here on the derived state rather than by
        // reaching for the private flag.
        let mut state = onboard(Link::Down("the access request was refused".to_owned()));
        state.identity = Some("test-device".to_owned());
        state.denied = true;

        assert_eq!(state.stage(), Stage::Onboarding(Onboarding::Denied { device: "test-device".to_owned() }));
    }

    #[test]
    fn reauthorizing_is_told_apart_from_a_first_ever_ask() {
        // Both have `asking` set at the same time -- reauthorizing sets it
        // via the same `ask` call a first-ever request uses. Only the
        // dedicated flag tells the two apart, checked here on the derived
        // state a caller actually sees rather than the flag itself.
        let mut state = onboard(Link::Reaching);
        state.identity = Some("test-device".to_owned());
        let (_sender, asking) = channel();
        state.asking = Some(asking);
        state.reauthorizing = Some(true);

        assert_eq!(
            state.stage(),
            Stage::Onboarding(Onboarding::Reauthorizing { device: "test-device".to_owned(), was_expired: true })
        );
    }

    #[test]
    fn a_token_refused_for_no_stated_reason_is_not_called_expired() {
        // A 401 on a token whose own claim did not say it had expired --
        // revoked by the administrator, most likely. Reauthorizing still
        // happens, but `was_expired` must stay false: this cannot honestly
        // claim to know why, only that it was refused.
        let mut state = onboard(Link::Reaching);
        state.identity = Some("test-device".to_owned());
        let (_sender, asking) = channel();
        state.asking = Some(asking);
        state.reauthorizing = Some(false);

        assert_eq!(
            state.stage(),
            Stage::Onboarding(Onboarding::Reauthorizing { device: "test-device".to_owned(), was_expired: false })
        );
    }

    /// A JWT carrying `payload`, encoded the same way a real Signal K
    /// device token is -- see [`Token::read`]'s own doc. No signature:
    /// nothing here checks one, only reads the claims.
    fn jwt_with(payload: &str) -> String {
        use base64::Engine;
        let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(payload.as_bytes());
        format!("header.{encoded}.signature")
    }

    #[test]
    fn reauthorize_reads_expiry_from_the_token_it_is_about_to_discard() {
        // No server, so `ask` (a real thread, a real HTTP call) never
        // runs -- this exercises only the read-then-discard half of
        // `reauthorize`.
        let mut state = onboard(Link::Reaching);
        state.token = Some(jwt_with(r#"{"device":"test","exp":1000}"#));

        state.reauthorize();

        assert_eq!(state.token, None);
        assert_eq!(state.link, Link::Down("the token had expired".to_owned()));
        assert_eq!(state.reauthorizing, None, "no server to ask again against");
    }

    #[test]
    fn reauthorize_does_not_call_an_unexpired_token_expired() {
        let mut state = onboard(Link::Reaching);
        state.token = Some(jwt_with(r#"{"device":"test","exp":99999999999}"#));

        state.reauthorize();

        assert_eq!(state.link, Link::Down("the token was no longer accepted".to_owned()));
    }

    #[test]
    fn a_denial_after_reauthorizing_is_reported_as_a_denial() {
        // The more specific, final outcome wins over the transitional
        // "asking again" state that led to it -- the same priority Denied
        // and AlreadyPending already take over plain Asking.
        let mut state = onboard(Link::Down("the access request was refused".to_owned()));
        state.identity = Some("test-device".to_owned());
        state.denied = true;
        state.reauthorizing = Some(true);

        assert_eq!(state.stage(), Stage::Onboarding(Onboarding::Denied { device: "test-device".to_owned() }));
    }

    #[test]
    fn a_stray_pending_request_is_told_apart_from_a_link_failure() {
        // Nothing here is Link::Down -- there is nothing wrong with the
        // network or the server, only a request this run has no id to
        // poll. GaveUp would have implied a failure worth retrying the
        // same way; AlreadyPending is a different kind of stuck, and reads
        // that way to a caller.
        let mut state = onboard(Link::Reaching);
        state.identity = Some("test-device".to_owned());
        state.already_pending = true;

        assert_eq!(state.stage(), Stage::Onboarding(Onboarding::AlreadyPending { device: "test-device".to_owned() }));
    }

    #[test]
    fn giving_up_before_ever_connecting_is_still_onboarding() {
        let state = onboard(Link::Down("no server answered".to_owned()));
        assert_eq!(state.stage(), Stage::Onboarding(Onboarding::GaveUp("no server answered".to_owned())));
    }

    #[test]
    fn once_the_link_has_ever_been_up_an_ordinary_drop_is_underway_not_onboarding() {
        // An outage that resolves on its own: a second telling of the
        // same loss, this time in onboarding's words instead of
        // staleness's, would only leave the two disagreeing about the
        // same fact.
        let mut state = onboard(Link::Reaching);
        state.note(Link::Up);
        state.note(Link::Down("Signal K link down".to_owned()));

        assert_eq!(state.stage(), Stage::Underway);
    }

    #[test]
    fn reauthorizing_still_shows_after_a_previous_successful_connection() {
        // Guards against a token revoked mid-run reading as plain
        // "Underway" with a dropped link, indistinguishable from an
        // ordinary outage that clears itself: a caller needs to know
        // Signal K is waiting on a fresh approval, the same as on the
        // first connection.
        let mut state = onboard(Link::Reaching);
        state.note(Link::Up);
        state.identity = Some("test-device".to_owned());
        state.note(Link::Down("the token was no longer accepted".to_owned()));
        let (_sender, asking) = channel();
        state.asking = Some(asking);
        state.reauthorizing = Some(false);

        assert_eq!(
            state.stage(),
            Stage::Onboarding(Onboarding::Reauthorizing { device: "test-device".to_owned(), was_expired: false })
        );
    }

    #[test]
    fn a_denial_still_shows_after_a_previous_successful_connection() {
        let mut state = onboard(Link::Reaching);
        state.note(Link::Up);
        state.identity = Some("test-device".to_owned());
        state.denied = true;

        assert_eq!(state.stage(), Stage::Onboarding(Onboarding::Denied { device: "test-device".to_owned() }));
    }

    #[test]
    fn an_ordinary_reconnect_after_a_previous_connection_stays_underway() {
        // The other half of the fix: only the states that need a human
        // are exempted from `ever_connected` -- plain `Reaching` (no
        // token trouble, just dialling back in) must still collapse to
        // `Underway`, or every routine wifi hiccup would wrongly read as
        // needing an approval nobody is waiting to give.
        let mut state = onboard(Link::Reaching);
        state.note(Link::Up);
        state.note(Link::Down("Signal K link down".to_owned()));
        state.note(Link::Reaching);

        assert_eq!(state.stage(), Stage::Underway);
    }
}