continuum-proxy/src/uplink/mod.rs

//! The cloud connection: a WebSocket to `continuum-backend`'s `/ws/edge/v1`.
//!
//! This is where the proxy earns its name. Until now it connected, said
//! hello, and sent heartbeats into the void while the printer fleet lived in
//! a completely separate part of the program with nothing routing between
//! them. Both directions are wired up now:
//!
//! - **Up:** every `TELEMETRY_INTERVAL`, one snapshot per printer, batched
//!   into a single `Telemetry` message.
//! - **Down:** a `Command` from the cloud is looked up by printer id,
//!   executed, and answered with a `CommandResult`.
//!
//! - **Out of band:** anything that needs to reach the cloud from outside
//!   this loop goes through an mpsc channel (`outbound`), which is how a
//!   command that takes minutes reports its outcome after the loop has
//!   already answered for it.
//!
//! Still deliberately the simplest version of a reconnecting client: a fixed
//! retry delay rather than a backoff curve. Telemetry produced while the
//! uplink is down is not buffered, and that is now a considered decision
//! rather than a gap - the cloud stores only each printer's latest state and
//! has no table for a history, and a Bambu's state is level-triggered
//! (`FINISH` persists until the next print starts), so the first snapshot
//! after a reconnect carries the outcome. Replaying a backlog of five-second
//! snapshots into a row that holds one value would churn it with stale
//! readings for no consumer. See PLAN.md for what would actually be worth
//! buffering.

mod auth;
mod download;
mod webrtc;
pub mod protocol;

pub use protocol::{ClientMessage, DiscoveredPrinter, PrinterCommand, PrinterTelemetry, ServerMessage};

use std::sync::Arc;
use std::time::Duration;

use futures_util::{SinkExt, StreamExt};
use tokio::sync::Mutex;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::handshake::client::{Request, generate_key};
use tracing::{info, warn};

use crate::config::Config;
use crate::provision;
use crate::store::Store;
use protocol::ProvisionPrinter;
use crate::printer::{GenericPrinter, PrinterHandle};

/// The websocket, once connected. Named because `connect` has to hand it to
/// `serve` and the full type is three layers of generics.
type WsStream = tokio_tungstenite::WebSocketStream<
    tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
>;

/// First retry delay after a failed connection, doubling up to
/// `RETRY_DELAY_MAX`.
const RETRY_DELAY: Duration = Duration::from_secs(5);
/// Ceiling for the backoff.
///
/// A gateway cannot tell a cloud that is briefly restarting from a token the
/// cloud will never accept - both are a failed connection here - so a flat
/// 5s retry meant a permanently misconfigured gateway knocking twelve times a
/// minute, forever, which is both pointless and the shape of something worth
/// rate-limiting at the other end. A minute is short enough that an outage
/// resolves without anyone touching the gateway, which matters when the
/// gateway is in someone else's workshop.
const RETRY_DELAY_MAX: Duration = Duration::from_secs(60);
/// Depth of the out-of-band outbound queue.
///
/// Bounded, because an unbounded queue facing a cloud that may be down for
/// hours is just a slow memory leak. Only out-of-band messages go through it,
/// one per long-running command, so 64 is far more than a farm generates
/// between reconnects. A sender waits when it is full rather than dropping,
/// which is the right trade for a message whose whole purpose is to explain a
/// failure.
const OUTBOUND_CAPACITY: usize = 64;
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);
/// Faster than the heartbeat, slower than the printers push. A Bambu on v2
/// emits a full status roughly every second; forwarding at that rate would be
/// a lot of traffic to describe a machine whose state changes on the scale of
/// a layer, so this samples the cached report instead of chasing every push.
const TELEMETRY_INTERVAL: Duration = Duration::from_secs(5);

/// The printer fleet, shared between `main` (which owns it and keeps every
/// connection alive) and this module (which reads status and dispatches
/// commands).
///
/// `Arc` is shared ownership - several places hold the same value and the
/// last one to let go drops it, roughly how every TS object reference
/// already behaves. `Mutex` is what makes mutation across tasks safe: Rust
/// will not let two tasks hold `&mut` to the same data, so a lock is how you
/// take turns. `tokio::sync::Mutex` rather than `std::sync::Mutex`
/// specifically because the code below holds the lock across an `.await`,
/// which the std one is not designed for.
///
/// One lock for the whole `Vec` is coarse: a slow command to one printer
/// stalls telemetry for all of them. Fine at the handful-of-printers scale
/// this runs at now, and the shape to revisit (a lock per printer, or an
/// actor task per printer owning its own handle) before a farm gets large.
pub type SharedFleet = Arc<Mutex<Vec<PrinterHandle>>>;

/// The `Host` header for a websocket URL: authority without the scheme.
///
/// Needed because building the upgrade request by hand (to attach the edge
/// token) means tokio-tungstenite no longer derives the handshake headers
/// from a URL for us. An error rather than a guess if the URL has no
/// authority: a `Host` of the wrong thing fails the handshake with something
/// far less obvious than this does.
fn host_header(url: &str) -> anyhow::Result<String> {
    let authority = url
        .split_once("://")
        .map(|(_, rest)| rest)
        .unwrap_or(url)
        .split(['/', '?', '#'])
        .next()
        .unwrap_or("");

    if authority.is_empty() {
        anyhow::bail!("the uplink URL has no host to connect to: {url}");
    }
    Ok(authority.to_string())
}

/// The queue a task elsewhere puts a message on to have the uplink send it.
///
/// Created by the caller rather than inside [`run`] so that a task which is
/// not the uplink - printer discovery, today - can hold the sender without
/// the uplink having to know what those tasks are. The property that matters
/// is unchanged either way: it outlives any one connection, so a message
/// produced while the socket is down waits for the next socket instead of
/// being dropped, and the sender living for the whole program is what stops
/// `recv()` from ever returning `None`.
pub fn channel() -> (
    tokio::sync::mpsc::Sender<ClientMessage>,
    tokio::sync::mpsc::Receiver<ClientMessage>,
) {
    tokio::sync::mpsc::channel(OUTBOUND_CAPACITY)
}

/// Runs forever: connect, serve the connection until it drops, wait, retry.
pub async fn run(
    config: Config,
    fleet: SharedFleet,
    store: Option<Arc<Store>>,
    outbound: tokio::sync::mpsc::Sender<ClientMessage>,
    mut inbox: tokio::sync::mpsc::Receiver<ClientMessage>,
) {
    let mut delay = RETRY_DELAY;

    loop {
        info!(url = %config.uplink_url, "connecting to cloud uplink");

        match connect(&config).await {
            Ok(ws_stream) => {
                // Reset on a completed handshake, which is the only
                // unambiguous evidence that the cloud is reachable and this
                // gateway's token is accepted - and the reason `connect` is
                // split out at all. Resetting after `serve` returns instead
                // would not distinguish a connection that lived for an hour
                // from one that failed to open.
                delay = RETRY_DELAY;
                if let Err(err) = serve(ws_stream, &config, &fleet, store.as_ref(), &outbound, &mut inbox).await {
                    warn!(?err, "uplink connection dropped, reconnecting in {delay:?}");
                }
            }
            Err(err) => warn!(?err, "uplink connection failed, retrying in {delay:?}"),
        }

        tokio::time::sleep(delay).await;
        delay = (delay * 2).min(RETRY_DELAY_MAX);
    }
}

/// Opens the websocket. Separate from `serve` so `run` can tell a failed
/// handshake from a connection that opened and later dropped; they want
/// different retry behaviour.
async fn connect(config: &Config) -> anyhow::Result<WsStream> {
    // A fresh edge JWT per attempt: it lives an hour, so one fetched at
    // startup would lock a long-running gateway out on its first reconnect.
    let edge_token = auth::fetch_edge_token(config).await?;

    // The edge token goes on the upgrade request, not in the first message.
    // The backend verifies it before the socket exists, so a bad token is an
    // HTTP 401 here rather than a connection that opens and is then closed -
    // and the credential stays out of the URL, where it would land in every
    // access and proxy log along the way. `Authorization` is the header
    // those tools already know to redact.
    let request = Request::builder()
        .uri(&config.uplink_url)
        .header("Authorization", format!("Bearer {edge_token}"))
        // tokio-tungstenite fills these in for a plain URL, but not once we
        // hand it a request we built ourselves - and a handshake missing any
        // of them is rejected by the server rather than ignored.
        .header("Host", host_header(&config.uplink_url)?)
        .header("Connection", "Upgrade")
        .header("Upgrade", "websocket")
        .header("Sec-WebSocket-Version", "13")
        .header("Sec-WebSocket-Key", generate_key())
        .body(())?;

    let (ws_stream, _response) = tokio_tungstenite::connect_async(request).await?;
    Ok(ws_stream)
}

/// Says hello, then pumps the connection until it drops.
async fn serve(
    ws_stream: WsStream,
    config: &Config,
    fleet: &SharedFleet,
    store: Option<&Arc<Store>>,
    outbound: &tokio::sync::mpsc::Sender<ClientMessage>,
    inbox: &mut tokio::sync::mpsc::Receiver<ClientMessage>,
) -> anyhow::Result<()> {
    let (mut write, mut read) = ws_stream.split();

    let hello = ClientMessage::Hello {
        gateway_id: config.gateway_id.clone(),
        farm_id: config.farm_id.clone(),
        stream_base_url: config.stream_base_url.clone(),
    };
    write.send(Message::Text(serde_json::to_string(&hello)?)).await?;

    let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL);
    let mut telemetry = tokio::time::interval(TELEMETRY_INTERVAL);
    let mut seq: u64 = 0;

    loop {
        tokio::select! {
            _ = heartbeat.tick() => {
                seq += 1;
                let msg = ClientMessage::Heartbeat { seq };
                write.send(Message::Text(serde_json::to_string(&msg)?)).await?;
            }

            _ = telemetry.tick() => {
                // Read first, then ask for the next one. A v1 Bambu answers
                // `refresh_status` over MQTT some time after this returns, so
                // collecting first means every tick reports the previous
                // tick's refresh - one interval of lag, rather than a
                // guaranteed miss from reading before the answer could
                // possibly have arrived.
                let printers = collect_telemetry(fleet).await;
                refresh_fleet(fleet).await;
                // Nothing to say is not worth a frame. Happens whenever no
                // printer has reported yet - every vendor but Bambu is still
                // a stub, so a Prusa-only fleet is silent here by design.
                if !printers.is_empty() {
                    let msg = ClientMessage::Telemetry { printers };
                    write.send(Message::Text(serde_json::to_string(&msg)?)).await?;
                }
            }

            // Whatever a task elsewhere wants said. Taken from the queue
            // before it is written, so a send that fails here loses the
            // message - the connection is going down anyway, and the
            // alternative (peek, write, then pop) buys little for a message
            // that is already best-effort diagnostics.
            Some(msg) = inbox.recv() => {
                write.send(Message::Text(serde_json::to_string(&msg)?)).await?;
            }

            frame = read.next() => {
                match frame {
                    Some(Ok(Message::Text(text))) => {
                        match serde_json::from_str::<ServerMessage>(&text) {
                            Ok(ServerMessage::Command { command_id, printer_id, command }) => {
                                // `start_print` is the one command that cannot
                                // be answered inline. See `spawn_long_command`.
                                let msg = if command.runs_for_minutes() {
                                    spawn_long_command(
                                        fleet.clone(),
                                        printer_id.clone(),
                                        command,
                                        command_id.clone(),
                                        outbound.clone(),
                                    );
                                    ClientMessage::CommandResult {
                                        command_id, ok: true, error: None,
                                    }
                                } else {
                                    match dispatch(fleet, &printer_id, command).await {
                                        Ok(()) => ClientMessage::CommandResult {
                                            command_id, ok: true, error: None,
                                        },
                                        Err(err) => {
                                            warn!(%printer_id, %err, "command failed");
                                            ClientMessage::CommandResult {
                                                command_id, ok: false, error: Some(err),
                                            }
                                        }
                                    }
                                };
                                write.send(Message::Text(serde_json::to_string(&msg)?)).await?;
                            }
                            // Retrying will not help: the token is wrong, not
                            // the network. Logged at error level and returned
                            // so it is obvious which of the two it is.
                            Ok(ServerMessage::HelloError { message }) => {
                                tracing::error!(%message, "cloud rejected this gateway's handshake - check CONTINUUM_GATEWAY_TOKEN");
                                return Ok(());
                            }
                            // Spawned, never awaited inline: go2rtc gathers
                            // ICE candidates before answering, which
                            // includes a STUN round trip, so this takes
                            // seconds. Awaiting it here would stall
                            // telemetry and heartbeats for the whole farm
                            // while one person opened one camera.
                            Ok(ServerMessage::WebrtcOffer { session_id, camera_stream, sdp }) => {
                                spawn_webrtc_negotiation(
                                    config.go2rtc_api_listen.clone(),
                                    session_id,
                                    camera_stream,
                                    sdp,
                                    outbound.clone(),
                                );
                            }
                            // Spawned: a connect, a TLS handshake and a frame
                            // from the printer take seconds, and awaiting that
                            // here would stall heartbeats and telemetry.
                            Ok(ServerMessage::CameraSnapshotRequest { request_id, printer_id }) => {
                                spawn_snapshot(fleet.clone(), request_id, printer_id, outbound.clone());
                            }
                            // Spawned: connecting to the new printer can take
                            // many seconds, and awaiting it here would stall
                            // heartbeats and telemetry for the whole farm.
                            Ok(ServerMessage::ProvisionPrinter { request_id, printer }) => {
                                spawn_provision(fleet.clone(), store.cloned(), request_id, Some(printer), outbound.clone());
                            }
                            Ok(ServerMessage::RemovePrinter { request_id, printer_id }) => {
                                spawn_remove(fleet.clone(), store.cloned(), request_id, printer_id, outbound.clone());
                            }
                            Ok(other) => info!(?other, "received from cloud"),
                            // Loudly. This used to be an `if let Ok(..)` with
                            // no else, which is the same silent-drop bug that
                            // hid a real `print.push_status` failure for
                            // weeks - and here it would mean commands
                            // vanishing with the cloud never being told.
                            // The payload is NOT logged. A `provision_printer` message
                            // carries an access code, and one that failed to parse is
                            // exactly the message that would end up here.
                            Err(err) => warn!(%err, bytes = text.len(), "unparseable message from cloud, ignoring"),
                        }
                    }
                    Some(Ok(Message::Close(_))) | None => return Ok(()),
                    Some(Err(err)) => return Err(err.into()),
                    _ => {}
                }
            }
        }
    }
}

/// Asks every printer to report in full, for the next tick to pick up. Failure
/// is logged and otherwise ignored: a printer that cannot be asked right now
/// is a printer whose telemetry goes stale, not a reason to tear down the
/// cloud connection for the whole farm.
async fn refresh_fleet(fleet: &SharedFleet) {
    let fleet = fleet.lock().await;
    for printer in fleet.iter() {
        if let Err(err) = printer.refresh_status().await {
            warn!(printer = %printer.base().id, %err, "failed to request a status refresh");
        }
    }
}

/// One snapshot per printer that has something to report. Translates Bambu's
/// `print.push_status` into the vendor-neutral shape the cloud consumes - see
/// `PrinterTelemetry` for why this isn't just forwarded raw.
async fn collect_telemetry(fleet: &SharedFleet) -> Vec<PrinterTelemetry> {
    let fleet = fleet.lock().await;
    let mut out = Vec::new();

    for printer in fleet.iter() {
        let base = printer.base();
        let Some(report) = printer.print_status().await else {
            continue;
        };

        out.push(PrinterTelemetry {
            printer_id: base.id.clone(),
            name: base.name.clone(),
            state: report.gcode_state.clone(),
            percent: report.mc_percent,
            stage: report.stg_cur,
            layer: report.layer_num,
            total_layers: report.total_layer_num,
            remaining_minutes: report.mc_remaining_time,
            // An idle printer reports `""` rather than omitting the field;
            // empty-string-as-absent is worth normalising here so the cloud
            // doesn't have to know that.
            job_name: report.subtask_name.clone().filter(|s| !s.is_empty()),
            nozzle_temp: report.nozzle_temper,
            nozzle_target: report.nozzle_target_temper,
            bed_temp: report.bed_temper,
            bed_target: report.bed_target_temper,
            print_error: report.print_error,
            active_alerts: report.hms.as_ref().map(|h| h.len()),
            // Derived from the same reported URL `camera::sources` builds
            // the go2rtc config from, so a printer is advertised as
            // streamable here only if it is also actually being served
            // there. Asking `camera` rather than keeping a second copy of
            // the rule means the two cannot drift apart.
            camera_stream: crate::camera::source_for(
                printer,
                report.ipcam.as_ref().and_then(|c| c.rtsp_url.as_deref()),
            )
            .map(|source| source.stream_name),
            camera_snapshot: crate::camera::wants_snapshot(
                printer,
                report.ipcam.is_some(),
                report.ipcam.as_ref().and_then(|c| c.rtsp_url.as_deref()),
            )
            .then_some(true),
        });
    }

    out
}

/// Runs a command that takes minutes, off this connection's task, and answers
/// the cloud immediately.
///
/// Everything else here is an MQTT publish that returns in milliseconds, so
/// awaiting it inline is fine. `start_print` is not: it downloads tens of
/// megabytes and then transfers them to a printer over FTPS at a measured
/// 0.2 MiB/s, which is about four minutes for a 50 MB project. Awaited
/// inline, that blocks this whole `select!` loop - no heartbeats and, far
/// worse, **no telemetry** for the duration.
///
/// No telemetry is the one that makes this mandatory rather than tidy. The
/// cloud derives a job's state from the telemetry that follows a start, so a
/// gateway that goes silent for four minutes during every job start is
/// withholding exactly the information the job depends on, while the backend
/// is also timing the command out after ten seconds and concluding the start
/// failed.
///
/// So the `ok` this produces means "accepted, and working on it" - which is
/// all the cloud ever trusted a start reply to mean anyway. *That* a failure
/// after this point happened is still noticed by the cloud on its own: the
/// job never reaches `printing` and the backend reaps it. What used to be
/// missing was *why*, which lived only in this gateway's log - on a machine
/// in someone else's workshop.
///
/// So a late failure now goes back over the outbound queue as a second
/// `CommandResult` for the same `command_id`. Be clear about what the backend
/// does with it: that command is long since resolved, so it is not awaiting
/// an answer and the result does not change the job's fate. It is logged with
/// its reason, next to the gateway and farm that produced it, which is the
/// difference between "job failed" and "job failed: filament mapping for
/// plate 1 names a slot the AMS does not have". Attaching it to the job row
/// needs a column `print_jobs` does not have yet.
///
/// Nothing is sent on success - the telemetry that follows says so better
/// than a message could, and a result for an already-resolved command is
/// noise unless it carries news.
fn spawn_long_command(
    fleet: SharedFleet,
    printer_id: String,
    command: PrinterCommand,
    command_id: String,
    outbound: tokio::sync::mpsc::Sender<ClientMessage>,
) {
    tokio::spawn(async move {
        match dispatch(&fleet, &printer_id, command).await {
            Ok(()) => info!(%printer_id, "long-running command finished"),
            Err(err) => {
                // Still logged locally, and still at error level: the queue
                // only reaches a cloud that is up, and this is the copy that
                // survives a gateway with no internet.
                tracing::error!(
                    %printer_id, %err,
                    "long-running command failed after the cloud was already told it was accepted"
                );
                let late = ClientMessage::CommandResult {
                    command_id,
                    ok: false,
                    error: Some(err),
                };
                // `send` waits for room rather than dropping, and only fails
                // if the receiver is gone - which means `run` has returned,
                // i.e. the process is shutting down and there is nobody left
                // to tell.
                if outbound.send(late).await.is_err() {
                    warn!(%printer_id, "uplink is shutting down; the failure reason was not reported");
                }
            }
        }
    });
}

/// Adds a printer off the connection's task and always reports how it went:
/// the console is holding a form open waiting for exactly this.
fn spawn_provision(
    fleet: SharedFleet,
    store: Option<Arc<Store>>,
    request_id: String,
    request: Option<ProvisionPrinter>,
    outbound: tokio::sync::mpsc::Sender<ClientMessage>,
) {
    tokio::spawn(async move {
        let result = match (store, request) {
            (Some(store), Some(request)) => provision::add(&fleet, &store, request).await,
            (None, _) => Err("this gateway has no local store, so it cannot add printers".to_string()),
            (_, None) => Err("no printer in the request".to_string()),
        };
        report(request_id, result, &outbound).await;
    });
}

fn spawn_remove(
    fleet: SharedFleet,
    store: Option<Arc<Store>>,
    request_id: String,
    printer_id: String,
    outbound: tokio::sync::mpsc::Sender<ClientMessage>,
) {
    tokio::spawn(async move {
        let result = match store {
            Some(store) => provision::remove(&fleet, &store, &printer_id).await,
            None => Err("this gateway has no local store".to_string()),
        };
        report(request_id, result, &outbound).await;
    });
}

async fn report(request_id: String, result: Result<(), String>, outbound: &tokio::sync::mpsc::Sender<ClientMessage>) {
    let msg = match result {
        Ok(()) => ClientMessage::CommandResult { command_id: request_id, ok: true, error: None },
        Err(error) => {
            warn!(%error, "printer request failed");
            ClientMessage::CommandResult { command_id: request_id, ok: false, error: Some(error) }
        }
    };
    if outbound.send(msg).await.is_err() {
        warn!("uplink is shutting down; the result was not reported");
    }
}

/// Answers one WebRTC offer, off this connection's task.
///
/// Always answers, success or failure: the cloud is holding a browser's
/// request open waiting for this, and a silent failure there is a viewer
/// watching a video element that never loads until something times out.
fn spawn_webrtc_negotiation(
    api_listen: String,
    session_id: String,
    camera_stream: String,
    offer_sdp: String,
    outbound: tokio::sync::mpsc::Sender<ClientMessage>,
) {
    tokio::spawn(async move {
        let reply = match webrtc::negotiate(&api_listen, &camera_stream, &offer_sdp).await {
            Ok(sdp) => ClientMessage::WebrtcAnswer { session_id, sdp },
            Err(err) => {
                // Logged as well as returned. The cloud gets the reason, but
                // this is the copy that survives a gateway whose uplink is
                // about to drop.
                warn!(%camera_stream, %err, "could not negotiate a camera stream");
                ClientMessage::WebrtcError { session_id, message: err.to_string() }
            }
        };
        if outbound.send(reply).await.is_err() {
            warn!(%camera_stream, "uplink is shutting down; the viewer gets no answer");
        }
    });
}

/// Reads one camera frame off this connection's task and reports it.
///
/// The fleet lock is held only to copy out what the read needs (the access
/// code never leaves this process in the message), never across the network.
fn spawn_snapshot(
    fleet: SharedFleet,
    request_id: String,
    printer_id: String,
    outbound: tokio::sync::mpsc::Sender<ClientMessage>,
) {
    tokio::spawn(async move {
        let job = {
            let fleet = fleet.lock().await;
            match fleet.iter().find(|p| p.base().id == printer_id) {
                Some(printer) => printer.snapshot_job().map_err(|err| err.to_string()),
                None => Err(format!("no printer '{printer_id}' in this gateway's fleet")),
            }
        };
        let result = match job {
            Ok(job) => job.run().await.map_err(|err| err.to_string()),
            Err(err) => Err(err),
        };
        let reply = match result {
            Ok(jpeg) => {
                use base64::Engine;
                ClientMessage::CameraSnapshot {
                    request_id,
                    jpeg_base64: Some(base64::engine::general_purpose::STANDARD.encode(jpeg)),
                    error: None,
                }
            }
            Err(err) => {
                warn!(%printer_id, %err, "could not read a camera frame");
                ClientMessage::CameraSnapshot { request_id, jpeg_base64: None, error: Some(err) }
            }
        };
        if outbound.send(reply).await.is_err() {
            warn!(%printer_id, "uplink is shutting down; the viewer gets no picture");
        }
    });
}

/// Finds the printer by our own id and runs the command. The error is a
/// `String` rather than a typed error because its only destination is a
/// `CommandResult` going back to the cloud as text.
async fn dispatch(fleet: &SharedFleet, printer_id: &str, command: PrinterCommand) -> Result<(), String> {
    // Fetched before the fleet lock is taken: a download is bounded by
    // minutes of network time, and holding the lock across it would stall
    // telemetry for every other printer. The transfer to the printer that
    // follows has to hold it (it needs `&mut` access), but that is no reason
    // to hold it through this part too. A `start_print` for a printer that
    // does not exist is therefore found out after the download rather than
    // before; the lookup below is cheap, so it is done first for that case.
    let downloaded = match &command {
        PrinterCommand::StartPrint { url, .. } => {
            if !fleet.lock().await.iter().any(|p| p.base().id == printer_id) {
                return Err(format!("no printer with id '{printer_id}' in this gateway's fleet"));
            }
            Some(
                download::fetch_job_file(url)
                    .await
                    .map_err(|err| format!("download failed (nothing was sent to the printer): {err}"))?,
            )
        }
        _ => None,
    };

    let mut fleet = fleet.lock().await;
    let printer = fleet
        .iter_mut()
        .find(|p| p.base().id == printer_id)
        .ok_or_else(|| format!("no printer with id '{printer_id}' in this gateway's fleet"))?;

    info!(%printer_id, ?command, "dispatching cloud command");

    match command {
        PrinterCommand::SetFanSpeed { percent } => {
            printer.set_fan_speed(percent).await.map_err(|err| err.to_string())
        }
        PrinterCommand::SendGcodeLine { line } => {
            printer.send_gcode_line(line).await.map_err(|err| err.to_string())
        }
        PrinterCommand::ControlPrint { control } => {
            printer.control_print(control).await.map_err(|err| err.to_string())
        }
        PrinterCommand::StartPrint { job_id, file_name, ams_slot, .. } => {
            let bytes = downloaded.ok_or_else(|| "internal error: start_print reached dispatch with no downloaded file".to_string())?;
            let byte_count = bytes.len();
            info!(%printer_id, %job_id, %file_name, byte_count, "job file downloaded, sending to printer");
            // One call covers deriving the filament mapping, the FTPS
            // transfer, the truncation check and the start, and reports them
            // all as one error - so the stage here is "printer", not finer.
            // The printer's own message says which part (mapping, upload of
            // the file, or start) it was.
            if let PrinterHandle::Bambu(bambu) = &mut *printer {
                bambu.set_ams_slot_override(ams_slot);
            } else if ams_slot.is_some() {
                return Err("this printer cannot be told which AMS slot to use".into());
            }
            let result = printer
                .upload_project_file(file_name, bytes)
                .await
                .map_err(|err| format!("printer upload/start failed (download of {byte_count} bytes succeeded): {err}"));
            if let PrinterHandle::Bambu(bambu) = &mut *printer {
                bambu.set_ams_slot_override(None);
            }
            result
        }
    }
}