use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
use std::path::{Path, PathBuf};
use std::time::Duration;

use super::{GenericPrinter, PrinterBase, PrinterError};

// region: Certificates

/// The canonical shape for every Bambu MQTT message lives in
/// continuum-schemas, not here - see `send_gcode_line` below, which builds
/// an instance of it directly via `continuum-schema-instance` instead of a
/// hand-written serde struct. `v1`/`v2` underneath this directory are
/// BambuMQTT's two message schemas, not to be confused with the MQTT
/// *transport* version (v3.1.1 vs v5) below - they happen to move
/// together on real Bambu firmware, but they're two different things.
const BAMBU_SCHEMA_DIR: &str = concat!(
    env!("CARGO_MANIFEST_DIR"),
    "/../continuum-schemas/schemas/bambulab"
);

fn get_schema_for_command(version: BambuProtocolVersion, root: String, command: String) -> PathBuf {
    Path::new(BAMBU_SCHEMA_DIR)
        .join(version.schema_dir())
        .join(format!("request/{root}/{command}.schema.json"))
}

/// Bambu's shared LAN-mode root CA ("BBL CA2 RSA") - see certs/README.md
/// for how this was obtained and verified. `include_bytes!` embeds it into
/// the compiled binary at build time, so there's no file to ship alongside
/// the executable.
const BUNDLED_CA: &[u8] = include_bytes!("../../certs/bambu_ca2.pem");

/// Where auto-pinned certificates get saved - same directory
/// `examples/auto_connect_bambu.rs` already used by hand. Relative to
/// wherever the daemon is run from, same convention as `printers.toml`.
const CERT_DIR: &str = "certs/pinned";

/// Which certificate to trust when connecting to a Bambu printer's MQTTS
/// port (8883 - Bambu's LAN mode is TLS-only, there's no unencrypted
/// fallback).
pub enum BambuTls {
    /// Bambu's shared root CA. Correct for current-generation printers.
    BundledCa,
    /// A specific certificate instead - for a printer whose firmware
    /// doesn't chain to the shared CA (P1P, at least).
    Custom(Vec<u8>),
    /// Skip certificate verification entirely. The least safe option, only
    /// reasonable because LAN mode never leaves your local network.
    Insecure,
}

impl BambuTls {
    /// Resolves the trust mode for *one* printer. Call this per printer,
    /// passing that printer's own `ca_cert_path`/`require_valid_cert` from
    /// `printers.toml` (see `fleet.rs::build_bambu`) - there's no single
    /// override for the whole fleet, because not every Bambu model chains
    /// to the same CA, and most printers don't need this at all (see
    /// `BambuPrinter::ensure_trusted` for the automatic path).
    /// `require_valid_cert = false` always means `Insecure`, regardless of
    /// whether that printer also has a `cert_path`.
    pub fn resolve(cert_path: Option<&Path>, require_valid_cert: bool) -> anyhow::Result<Self> {
        if !require_valid_cert {
            return Ok(BambuTls::Insecure);
        }
        match cert_path {
            Some(path) => Ok(BambuTls::Custom(std::fs::read(path)?)),
            None => Ok(BambuTls::BundledCa),
        }
    }

    /// The CA certificate bytes to hand to the TLS layer - empty when
    /// `Insecure`, since there's nothing to verify against.
    pub fn ca_bytes(&self) -> &[u8] {
        match self {
            BambuTls::BundledCa => BUNDLED_CA,
            BambuTls::Custom(bytes) => bytes,
            BambuTls::Insecure => &[],
        }
    }

    /// Builds a `native_tls` connector configured for this trust mode:
    /// trusting only `ca_bytes()`, or accepting anything for `Insecure`.
    /// `pub(crate)`, not private: `bambu_mqtt.rs` (a sibling module, not a
    /// descendant of this one) reuses this exact connector to configure
    /// rumqttc's TLS transport, instead of every consumer's TLS trust
    /// decisions living in two separate places.
    pub(crate) fn build_connector(&self) -> anyhow::Result<native_tls::TlsConnector> {
        let mut builder = native_tls::TlsConnector::builder();

        // Bambu printer certificates don't carry their LAN IP as a Subject
        // Alternative Name - it's DHCP-assigned, so baking it in at
        // manufacture time wouldn't make sense. Hostname verification has
        // to be off regardless of trust mode: what matters is whether the
        // certificate chains to a CA we trust, not whether its SAN matches
        // the specific IP we happened to dial today. Chain verification
        // itself stays fully enforced below for BundledCa - this narrows
        // *what* gets checked, it doesn't turn checking off.
        builder.danger_accept_invalid_hostnames(true);

        match self {
            BambuTls::BundledCa => {
                // A real CA root - chain validation is exactly the right
                // check here.
                builder.add_root_certificate(native_tls::Certificate::from_pem(self.ca_bytes())?);
            }
            BambuTls::Custom(_) => {
                // `ensure_trusted` pins a *leaf* certificate (the printer's
                // own, CN = its serial number, issued by Bambu's own CA -
                // not a root), for byte-exact fingerprint comparison, not
                // chain validation. Handing that same leaf to
                // `add_root_certificate` here would ask OpenSSL to treat it
                // as its own trust anchor - exactly the bug
                // `test_tls_handshake`'s doc comment already describes
                // ("self-signed certificate in certificate chain"), just
                // not fixed at this second call site too until now.
                // native_tls has no hook for a custom per-certificate
                // verification callback (unlike rustls, which would let us
                // do the real fingerprint check here) - so this accepts
                // any certificate at the TLS layer instead, relying on
                // `ensure_trusted` already having fingerprint-verified this
                // exact printer moments before `connect()` opens this
                // transport. Weaker than continuous fingerprint pinning on
                // every reconnect, but not a fresh regression: pinning a
                // leaf as a root here never actually worked in the first
                // place.
                builder.danger_accept_invalid_certs(true);
            }
            BambuTls::Insecure => {
                builder.danger_accept_invalid_certs(true);
            }
        }

        Ok(builder.build()?)
    }
}

fn resolve_addr(host: &str) -> anyhow::Result<SocketAddr> {
    (host, 8883)
        .to_socket_addrs()?
        .next()
        .ok_or_else(|| anyhow::anyhow!("could not resolve {host}"))
}

/// Minimal DER -> PEM encoder (base64, 64-char lines, standard
/// header/footer) - just enough to save a fetched certificate in the same
/// format as the bundled one.
fn der_to_pem(der: &[u8]) -> Vec<u8> {
    use base64::Engine;
    let encoded = base64::engine::general_purpose::STANDARD.encode(der);

    let mut pem = String::from("-----BEGIN CERTIFICATE-----\n");
    for line in encoded.as_bytes().chunks(64) {
        pem.push_str(std::str::from_utf8(line).expect("base64 output is always ASCII"));
        pem.push('\n');
    }
    pem.push_str("-----END CERTIFICATE-----\n");
    pem.into_bytes()
}

/// Re-encodes a PEM certificate to DER, so it can be byte-compared against
/// what a live handshake presents (`Certificate::to_der()`).
fn pem_to_der(pem: &[u8]) -> anyhow::Result<Vec<u8>> {
    Ok(native_tls::Certificate::from_pem(pem)?.to_der()?)
}

// endregion: Certificates

// region: BambuMQTTProtocol
/// Which BambuMQTT message schema (and, on real firmware, which MQTT
/// transport version) a connection ended up using. This is an *outcome*
/// `connect()` settles on, not something you can assume from the printer
/// model - a firmware update can move the same physical X1C from v1 to
/// v2. Every later message this printer sends is built against
/// `schemas/bambulab/{v1,v2}/` depending on which one this is.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BambuProtocolVersion {
    /// MQTT v3.1.1, BambuMQTT v1 - older firmware (P1P/P1S, original X1 launch firmware).
    V1,
    /// MQTT v5, BambuMQTT v2 - current firmware.
    V2,
}

impl BambuProtocolVersion {
    fn schema_dir(self) -> &'static str {
        match self {
            BambuProtocolVersion::V1 => "v1",
            BambuProtocolVersion::V2 => "v2",
        }
    }
}

/// How to pick the protocol generation for a given printer - config
/// *input*, as opposed to `BambuProtocolVersion` (the outcome `connect()`
/// settles on). `Auto` is the default for exactly the reason
/// `BambuProtocolVersion` isn't derivable from config alone: you can't
/// reliably know which one a specific printer speaks today without
/// asking it. `ForceV1`/`ForceV2` stay available as an override - useful
/// if auto-detection ever misbehaves, or to test one path deliberately.
#[derive(Debug, Clone, Copy)]
pub enum BambuProtocolPreference {
    /// Try MQTT v5 first; if the broker's CONNACK rejects that protocol
    /// version (older firmware only understands v3.1.1), fall back and
    /// reconnect with v3.1.1 instead.
    Auto,
    ForceV1,
    ForceV2,
}
// endregion: BambuMQTTProtocol

/// A Bambu printer. Only one struct now, not two: the old
/// `BambuGenericPrinter` (shared fields/behavior) + `BambuV1Printer`/
/// `BambuV2Printer` (thin per-version wrappers) split existed to let two
/// *separate types* share a base - but since which protocol version a
/// printer uses is now something `connect()` discovers at runtime rather
/// than something config picks between two types for, there's only ever
/// one type to begin with, so the extra layer stopped earning its keep.
pub struct BambuPrinter {
    base: PrinterBase,
    pub access_code: String,
    pub sn: String,
    tls: BambuTls,
    preference: BambuProtocolPreference,
    /// `None` until `connect()` succeeds. Carries the live MQTT session
    /// alongside the protocol version it settled on, since the two are
    /// decided together and always used together - see `bambu_mqtt.rs`.
    connection: Option<(super::bambu_mqtt::BambuMqttSession, BambuProtocolVersion)>,
}

/// Which plate of a multi-plate project gets printed.
///
/// Always the first, matching `start_uploaded_file`'s hardcoded
/// `Metadata/plate_1.gcode`. The two have to agree: deriving the filament
/// mapping from plate 2 while printing plate 1 would map confidently onto
/// the wrong thing. When one becomes a parameter, so must the other.
const FIRST_PLATE: u32 = 1;

/// How many slots Bambu numbers each AMS unit as, for turning a
/// (unit, tray) pair into the single slot number `ams_mapping` wants.
///
/// 4 even for an AMS HT, which physically has one: the numbering is by unit
/// index, not by how many slots the unit happens to have.
const SLOTS_PER_AMS: i32 = 4;

/// External spools, which `ams_mapping` has to be padded for even though it
/// cannot address them.
///
/// Still a constant now that `vt_tray` is modelled, because modelling it did
/// not make the count derivable. A v1 P1P reports one `vt_tray` object, which
/// confirms 1 for that printer - but a v2 H2C omits the key entirely (see
/// `PrintReport::vt_tray`), so on v2 an absent `vt_tray` means "not reported"
/// and not "no external spool". Counting what is reported would therefore
/// give 0 on exactly the printer the ams_mapping schema says has *two*.
///
/// 1 is right for every printer this has run against. A dual-nozzle H2C has
/// two, and getting that right needs a v2 capture that actually shows how a
/// v2 printer describes its spools.
const EXTERNAL_SPOOLS: usize = 1;

/// Flattens the AMS report into trays with global slot numbers, plus the
/// total tray count `ams_mapping` must be padded to.
///
/// The count includes *empty* AMS slots and the external spool: the padding
/// rule is over every tray the printer has, not every tray with filament in
/// it. Getting that wrong produces an under-length array, which is one of
/// the quiet ways a mapping fails.
fn available_trays(ams: Option<&super::AmsState>) -> (Vec<super::AvailableTray>, usize) {
    let Some(ams) = ams else {
        return (Vec::new(), EXTERNAL_SPOOLS);
    };

    let mut trays = Vec::new();
    let mut total = EXTERNAL_SPOOLS;

    for unit in &ams.ams {
        let unit_index: i32 = unit.id.parse().unwrap_or(0);
        total += unit.tray.len();

        for tray in &unit.tray {
            // Only a loaded tray can be matched against; an empty slot still
            // counts towards `total` above.
            let super::AmsTray::Loaded(loaded) = tray else {
                continue;
            };
            let Ok(slot_in_unit) = loaded.id.parse::<i32>() else {
                continue;
            };

            trays.push(super::AvailableTray {
                slot: unit_index * SLOTS_PER_AMS + slot_in_unit,
                tray_info_idx: loaded.tray_info_idx.clone(),
                filament_type: loaded.tray_type.clone(),
                color: super::normalise_color(&loaded.tray_color),
            });
        }
    }

    (trays, total)
}

/// Whether the printer's external spool holds something every one of this
/// project's filaments could be printed from.
///
/// Reuses `map_to_trays` with the spool presented as the single candidate
/// tray, rather than reimplementing the match. With exactly one candidate,
/// an `Ok` means every filament matched that one spool - which is the
/// question being asked - and the slot numbers it returns are thrown away,
/// since the external spool is not addressable by slot. Hence the `0`
/// placeholder.
///
/// The point of this is to turn "a project wants the external spool" from an
/// error into a correct answer. Before `vt_tray` was modelled there was no
/// way to see what the spool held, and claiming a match without looking is
/// how the `use_ams: false` bug happened in the first place.
fn external_spool_satisfies(
    filaments: &[super::ProjectFilament],
    spool: &super::VtTray,
) -> bool {
    let (Some(tray_info_idx), Some(filament_type), Some(color)) = (
        spool.tray_info_idx.as_ref(),
        spool.tray_type.as_ref(),
        spool.tray_color.as_ref(),
    ) else {
        // An empty spool reports these as absent or blank. Nothing to match.
        return false;
    };

    if tray_info_idx.is_empty() || filament_type.is_empty() {
        return false;
    }

    let candidate = [super::AvailableTray {
        slot: 0,
        tray_info_idx: tray_info_idx.clone(),
        filament_type: filament_type.clone(),
        color: super::normalise_color(color),
    }];

    super::map_to_trays(filaments, &candidate).is_ok()
}

/// What a printer's light should be doing.
///
/// Only the two steady states. Bambu's `system.ledctrl` also has a
/// `flashing` mode taking `led_on_time`/`led_off_time`/`loop_times`/
/// `interval_time` (see
/// `schemas/bambulab/v1/request/system/ledctrl.schema.json`), and it is
/// deliberately not wired up: the meaning of `loop_times` versus
/// `interval_time` is not confirmed against hardware, so a caller asking for
/// "blink for ten seconds" cannot currently predict what the firmware will
/// do with it. Driving the blink from here out of `On` and `Off` gives an
/// exact duration out of two commands whose behaviour is obvious. Switch to
/// the firmware mode once someone has watched a real printer do it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LightState {
    On,
    Off,
}

impl LightState {
    /// The wire word. The schema accepts upper and lower case; lower is what
    /// Bambu Studio sends.
    fn as_str(self) -> &'static str {
        match self {
            Self::On => "on",
            Self::Off => "off",
        }
    }
}

/// The light every Bambu has. `work_light` and `chamber_light2` are in the
/// schema's enum but not on every model, so they are not assumed here.
pub const CHAMBER_LIGHT: &str = "chamber_light";

/// Where a project print draws its filament from.
///
/// This exists because the alternative - hardcoding `use_ams: false` -
/// silently overrode the slicer's own choice. Confirmed on a P1P
/// (2026-10-11): a project sliced for AMS slot 3 started, heated to
/// 250C, reached 1% and then the printer paused itself with
/// `print_error: 50364420` and `tray_now: 255`, the external spool. The
/// print was never going to work, and nothing in the command's reply
/// said so - `result` was `success`, because the command was accepted.
///
/// So this is a required argument rather than an option with a default.
/// There is no safe default: picking the wrong source draws the wrong
/// filament or none at all.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FilamentSource {
    /// The external spool (`vt_tray`). `use_ams: false`, and
    /// `ams_mapping` must be the empty *string* - the schema's trailing
    /// `oneOf` ties those two together.
    ExternalSpool,
    /// The AMS, with a mapping from the project's filament indices to
    /// AMS slots. Build it with `FilamentSource::ams`.
    Ams(Vec<i32>),
}

impl FilamentSource {
    /// `slots[i]` is the AMS slot the project's filament index `i` should
    /// draw from, or `-1` for an index this project does not use.
    ///
    /// `tray_count` is every tray the printer has, AMS slots *and* external
    /// spools - 5 for a P1P with one AMS. The mapping is padded to that
    /// length with `-1`, because
    /// `schemas/bambulab/v1/common/ams/ams_mapping.schema.json` says to
    /// always pad it and an under-length array is one of the ways this goes
    /// wrong quietly.
    pub fn ams(slots: impl Into<Vec<i32>>, tray_count: usize) -> Result<Self, PrinterError> {
        let mut slots: Vec<i32> = slots.into();

        if slots.is_empty() {
            return Err(PrinterError(
                "an AMS mapping needs at least one filament index; use ExternalSpool instead".into(),
            ));
        }
        if slots.len() > tray_count {
            return Err(PrinterError(format!(
                "an AMS mapping of {} entries cannot fit a printer with {tray_count} trays",
                slots.len()
            )));
        }
        if let Some(bad) = slots.iter().find(|slot| **slot < -1) {
            return Err(PrinterError(format!(
                "{bad} is not a valid AMS slot: use -1 for an unused filament index"
            )));
        }

        slots.resize(tray_count, -1);
        Ok(Self::Ams(slots))
    }

    /// Writes `use_ams` and `ams_mapping` into a `print.project_file` body.
    ///
    /// The two always move together: a real mapping requires
    /// `use_ams: true`, and the no-AMS case requires `ams_mapping` to be the
    /// empty string rather than an empty array.
    fn apply(&self, body: &mut serde_json::Value) {
        match self {
            Self::ExternalSpool => {
                body["use_ams"] = serde_json::Value::Bool(false);
                body["ams_mapping"] = serde_json::Value::String(String::new());
            }
            Self::Ams(slots) => {
                body["use_ams"] = serde_json::Value::Bool(true);
                body["ams_mapping"] = serde_json::json!(slots);
            }
        }
    }
}

impl BambuPrinter {
    pub fn new(
        id: &str,
        name: &str,
        host: &str,
        access_code: &str,
        sn: &str,
        tls: BambuTls,
        preference: BambuProtocolPreference,
    ) -> Self {
        Self {
            base: PrinterBase {
                id: id.into(),
                name: name.into(),
                host: host.into(),
            },
            access_code: access_code.into(),
            sn: sn.into(),
            tls,
            preference,
            connection: None,
        }
    }

    pub fn base(&self) -> &PrinterBase {
        &self.base
    }

    /* A real (blocking, one-shot) TLS handshake to port 8883 - no MQTT
    /// protocol, just "does the certificate verify" against this printer's
    /// configured trust mode.
    ///
    /// `BundledCa` verifies normally (Bambu's shared CA is a proper,
    /// well-formed root - OpenSSL's usual chain validation handles it
    /// fine). `Custom` does something different: *fingerprint pinning*,
    /// not chain validation. A device's own leaf certificate often isn't a
    /// well-formed CA (no `CA:TRUE`), and the device may present it
    /// alongside a separate self-signed root you never captured - asking
    /// OpenSSL's normal path-building to accept an arbitrary leaf as a
    /// trust anchor is unreliable for exactly that reason (this is a real
    /// bug that shipped and failed against a real P1P: `add_root_certificate`
    /// with a captured leaf produced `self-signed certificate in
    /// certificate chain` / `unable to get local issuer certificate` on
    /// the very next connection). So instead: connect with verification
    /// off, then compare the certificate actually presented against the
    /// exact bytes pinned, byte for byte. A mismatch is a hard failure -
    /// no PKI judgment call, just "is this the same certificate as before." */
    pub fn test_tls_handshake(&self) -> anyhow::Result<()> {
        let addr = resolve_addr(&self.base.host)?;
        let stream = TcpStream::connect_timeout(&addr, Duration::from_secs(5))?;

        if let BambuTls::Custom(pinned_pem) = &self.tls {
            let tls_stream = BambuTls::Insecure
                .build_connector()?
                .connect(&self.base.host, stream)?;
            let presented = tls_stream
                .peer_certificate()?
                .ok_or_else(|| anyhow::anyhow!("{} presented no certificate", self.base.name))?
                .to_der()?;

            if presented != pem_to_der(pinned_pem)? {
                anyhow::bail!(
                    "{} presented a certificate different from the one pinned - could be a legitimate \
                     certificate rotation, could be something worse; delete its file under certs/pinned/ \
                     and reconnect to re-pin deliberately if you're sure it's the former",
                    self.base.name
                );
            }
            return Ok(());
        }

        self.tls
            .build_connector()?
            .connect(&self.base.host, stream)?;
        Ok(())
    }

    /* "Trust on first connect": connects once with certificate
    /// verification disabled, captures whatever certificate the printer
    /// presents, and returns it as PEM bytes.
    ///
    /// This is meaningfully weaker than verifying against a CA you already
    /// trust - anyone on the network at the exact moment this runs gets
    /// trusted forever after. `ensure_trusted` below is the only caller -
    /// it only reaches this path when the bundled CA didn't verify, saves
    /// the result under `certs/pinned/`, and every connection after this
    /// one re-verifies against that pinned file instead of trusting the
    /// network again. Not exposed as a manual step for the caller to run
    /// ahead of time - there's no longer a reason to, now that `connect()`
    /// does this automatically the first time a printer needs it. */
    pub fn fetch_certificate(&self) -> anyhow::Result<Vec<u8>> {
        let addr = resolve_addr(&self.base.host)?;
        let stream = TcpStream::connect_timeout(&addr, Duration::from_secs(5))?;
        let tls_stream = BambuTls::Insecure
            .build_connector()?
            .connect(&self.base.host, stream)?;

        let cert = tls_stream
            .peer_certificate()?
            .ok_or_else(|| anyhow::anyhow!("{} presented no certificate", self.base.name))?;

        Ok(der_to_pem(&cert.to_der()?))
    }

    /* Makes sure this printer has a *working* trust configuration,
    /// auto-pinning a certificate via trust-on-first-connect if needed -
    /// no manual step required, built the way SSH handles host keys:
    ///
    /// - A pinned certificate already on disk (`cert_dir/<sn>.pem`) is used
    ///   directly. No new trust decision gets made on every run - that
    ///   already happened once, this just re-verifies against it. Keyed by
    ///   *serial number*, not `id`: `id` is just an arbitrary label you
    ///   picked in printers.toml - rename a printer's `id` and the pin
    ///   should still find it, because a pin certifies "this physical
    ///   printer", and the serial number is the one thing about it that
    ///   can't change.
    /// - No pin yet, and the bundled CA verifies fine (current-generation
    ///   printers): nothing to do.
    /// - No pin yet, and the bundled CA does *not* verify (P1P, etc.):
    ///   fetches the certificate the printer presents, confirms it
    ///   produces a working connection, and saves it to `cert_dir/<sn>.pem`
    ///   for every run after this one. This is the one moment a MITM
    ///   active on your network *right now* could plant a certificate that
    ///   gets trusted from then on - the same tradeoff SSH accepts on a
    ///   first connection.
    /// - A printer with an *explicit* pin (you set `ca_cert_path` yourself
    ///   in printers.toml) never auto-pins over it - a verification
    ///   failure there is a real error, not something to paper over,
    ///   since that's exactly the signal pinning exists to give you (the
    ///   cert rotated, or something worse). */
    pub fn ensure_trusted(&mut self, cert_dir: &Path) -> anyhow::Result<()> {
        let pinned_path = cert_dir.join(format!("{}.pem", self.sn));

        if matches!(self.tls, BambuTls::BundledCa) && pinned_path.exists() {
            self.tls = BambuTls::Custom(std::fs::read(&pinned_path)?);
        }

        match self.test_tls_handshake() {
            Ok(()) => Ok(()),
            Err(err) if matches!(self.tls, BambuTls::BundledCa) => {
                eprintln!(
                    "{}: bundled CA didn't verify ({err}); auto-pinning on first connect",
                    self.base.name
                );
                let pem = self.fetch_certificate()?;
                std::fs::create_dir_all(cert_dir)?;
                std::fs::write(&pinned_path, &pem)?;
                self.tls = BambuTls::Custom(pem);
                self.test_tls_handshake()
            }
            Err(err) => Err(err),
        }
    }

    /// Bambu-only - no `GenericPrinter` equivalent, since every vendor's
    /// way of reporting firmware info (if it has one at all) is shaped
    /// completely differently. Same "vendor-only extension" pattern as
    /// `test_tls_handshake`/`fetch_certificate`/`ensure_trusted` above:
    /// reached by matching `PrinterHandle::Bambu` back out of the enum,
    /// not through the shared trait.
    ///
    /// The first real use of `BambuMqttSession::request` -
    /// `info.get_version` is a genuine one-shot, correlated reply (unlike
    /// `pushing.pushall`, see `request`'s doc comment), so this is exactly
    /// the case it's for.
    pub async fn get_version(&self) -> Result<super::GetVersionResponse, PrinterError> {
        let (session, version) =
            self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;

        let body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command(
            *version,
            "info".into(),
            "get_version".into(),
        ))
        .map_err(|err| PrinterError(format!("failed to build get_version request: {err}")))?;

        let response = session.request("info", body).await?;
        serde_json::from_value(response)
            .map_err(|err| PrinterError(format!("failed to parse get_version response: {err}")))
    }

    /// Bambu-only, same reasoning as `get_version` above. Sends
    /// `print.calibration` and waits for the correlated `print.calibration`
    /// *report* it triggers - confirmed live on both protocol versions,
    /// which respond identically here (an earlier, incorrect assumption
    /// had v1 folding this into `push_status` instead of replying
    /// directly; fixed in continuum-schemas). The printer also keeps
    /// sending ordinary `push_status` updates reflecting progress while
    /// calibration runs, same as any other job - read those via
    /// `print_status()` if you want progress, this call just confirms the
    /// calibration sequence started and got acknowledged.
    ///
    /// `option` is the bitmask this command's own schema defines:
    /// `lidarCalibration = 1`, `bedLevelling = 1 << 1`,
    /// `vibrationCompensation = 1 << 2`, `motorCancellation = 1 << 3` (OR
    /// together whichever steps you want). Deliberately not defaulted to
    /// "everything" here - this starts a real, physical calibration
    /// sequence on the printer, so the caller states exactly what they
    /// mean to trigger rather than this method guessing.
    pub async fn get_calibration(&self, option: i64) -> Result<serde_json::Value, PrinterError> {
        let (session, version) =
            self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;

        let mut body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command(
            *version,
            "print".into(),
            "calibration".into(),
        ))
        .map_err(|err| PrinterError(format!("failed to build calibration request: {err}")))?;
        body["option"] = serde_json::Value::from(option);

        session.request("print", body).await
    }

    /// Bambu-only, same reasoning as `get_version` above. Publishes
    /// `pushing.pushall`, which - unlike `get_version` - doesn't reply
    /// under its own category: it triggers/refreshes the ongoing
    /// `print.push_status` stream instead. Fire-and-forget, same as
    /// `send_gcode_line`: there's nothing to wait for here, since the
    /// forced full update this triggers self-describes as such via its
    /// own `msg` field once it arrives (see `PrintStatusCache::apply`),
    /// whenever that turns out to be - waiting on it added two real bugs
    /// (a race, and matching the wrong "print" message) for no benefit
    /// once that was understood.
    ///
    /// Call this periodically for v1 printers specifically - v2 has no
    /// `pushing` category at all (checked: `schemas/bambulab/v2/
    /// request.schema.json` only wires camera/info/print/system/upgrade/
    /// xcam). A v2 connection makes this call a no-op rather than an error.
    ///
    /// **Resolved 2026-10-03, live against a real H2C mid-print:** v2 needs
    /// no trigger whatsoever. It pushes a *full* `print.push_status` (97
    /// top-level fields, `msg: 0`) roughly once a second, unprompted, from
    /// the moment you subscribe. The earlier reading that it sent nothing
    /// was a local bug, not printer behaviour - every one of those pushes
    /// was arriving and being silently dropped by a `PrintReport`
    /// deserialization failure (see `handle_incoming_report`). So there is
    /// no missing "other trigger" to go looking for.
    /// Turns one of the printer's lights on or off.
    ///
    /// **Unverified against hardware.** Every other live-confirmed call in
    /// this file says so explicitly; this one was written while the only
    /// available printer was powered off. The command shape comes from
    /// `schemas/bambulab/{v1,v2}/request/system/ledctrl.schema.json` and is
    /// built from it rather than by hand, so the field names and types are
    /// as good as the schema is - but whether the printer accepts
    /// `chamber_light` and answers, and whether it replies at all, is
    /// untested. It goes out as a correlated `request` so that a rejection
    /// is visible rather than lost, the same reason `control_print` does.
    ///
    /// If this turns out not to reply, `request` will time out after 10s and
    /// this should become a `publish` - the symptom to look for.
    pub async fn set_light(&self, node: &str, state: LightState) -> Result<(), PrinterError> {
        let (session, version) = self
            .connection
            .as_ref()
            .ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;

        let mut body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command(
            *version,
            "system".into(),
            "ledctrl".into(),
        ))
        .map_err(|err| PrinterError(format!("failed to build ledctrl request: {err}")))?;

        body["led_node"] = serde_json::Value::String(node.to_string());
        body["led_mode"] = serde_json::Value::String(state.as_str().to_string());

        let reply = session.request("system", body).await?;
        Self::command_outcome(&self.base.name, "ledctrl", &reply)
    }

    pub async fn pushall(&self) -> Result<(), PrinterError> {
        let (session, version) =
            self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;

        if matches!(version, BambuProtocolVersion::V2) {
            return Ok(());
        }

        let mut body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command(
            *version,
            "pushing".into(),
            "pushall".into(),
        ))
        .map_err(|err| PrinterError(format!("failed to build pushall request: {err}")))?;
        body["push_target"] = serde_json::Value::from(1);

        session.publish("pushing", &body).await
    }

    /// Just the FTPS transfer: puts `data` in the printer's `cache/`
    /// directory and stops there, without telling it to print anything.
    ///
    /// Split out from `upload_project_file` (which is this plus
    /// `start_uploaded_file`) because the two halves have very different
    /// consequences and testing them together is awkward: moving a file onto
    /// a printer is inert and reversible, while starting a print commits the
    /// machine to hours of physical work. Being able to validate the
    /// transfer against real hardware without that second part is worth the
    /// extra method - and a caller that wants to stage a file now and print
    /// it later wants exactly this too.
    ///
    /// Returns however many bytes the printer's FTPS server accepted, so a
    /// caller can check that against what it sent.
    ///
    /// Lands in the SD card's root, not in `cache/` - it used to be the
    /// latter, which put the file somewhere `start_uploaded_file` could not
    /// then print from. See `bambu_ftps::SD_ROOT` for the evidence.
    pub async fn upload_file(&mut self, file_name: &str, data: Vec<u8>) -> Result<u64, PrinterError> {
        // Trust gets resolved out here, on the async side, so the blocking
        // closure below owns everything it needs and borrows nothing.
        let connector = self
            .tls
            .build_connector()
            .map_err(|err| PrinterError(format!("{}: TLS setup failed: {err}", self.base.name)))?;

        let host = self.base.host.clone();
        let access_code = self.access_code.clone();
        let name = file_name.to_string();

        tokio::task::spawn_blocking(move || {
            super::bambu_ftps::upload(
                &host,
                &access_code,
                connector,
                super::bambu_ftps::SD_ROOT,
                &name,
                &data,
            )
        })
        .await
        // The outer error is `spawn_blocking`'s own: the thread panicked or
        // was cancelled. The inner one is the upload's.
        .map_err(|err| PrinterError(format!("{}: FTPS upload thread failed: {err}", self.base.name)))?
        .map_err(|err| {
            PrinterError(format!("{}: FTPS upload of {file_name} failed: {err}", self.base.name))
        })
    }

    /// What is currently staged in the printer's `cache/` directory.
    ///
    /// Read-only. Worth having on its own because `start_uploaded_file` can
    /// only print what is already in there, so this answers "will that
    /// command find anything?" without sending it.
    pub async fn list_cache(&self) -> Result<Vec<String>, PrinterError> {
        self.list_dir(super::bambu_ftps::CACHE_DIR).await
    }

    /// Works out where a project's filaments should come from, by reading
    /// the project and comparing it with what the printer actually has
    /// loaded.
    ///
    /// This is the thing that `use_ams: false` was standing in for. The
    /// archive's `Metadata/slice_info.config` names each filament's
    /// `tray_info_idx`, type and colour; the AMS report names the same three
    /// things per loaded tray; matching them is the whole job. See
    /// `project_file::map_to_trays` for the tie-break order, and
    /// `FilamentSource` for what getting this wrong does to a real print.
    ///
    /// Requires a `print.push_status` to have arrived, because there is no
    /// way to guess what is loaded. On v1 that means `pushall` must have
    /// been called since connecting; `connect()` does.
    ///
    /// ## What it will not do
    ///
    /// It never falls back to the external spool to make an unmatched
    /// project "work". An unmatched filament is an error, because every
    /// alternative is a guess about what someone meant to print.
    ///
    /// It answers `ExternalSpool` when the spool is what the project asks
    /// for - checked against `vt_tray`, not assumed. A printer with an AMS
    /// whose project wants the spool used to be an error here purely because
    /// `vt_tray` was not modelled; now that it is, the question can be
    /// answered instead of refused.
    ///
    /// The one place it still answers `ExternalSpool` without being able to
    /// verify anything is a v2 printer that reports no AMS trays and no
    /// `vt_tray` either - v2 omits the key entirely (see
    /// `PrintReport::vt_tray`), so there is nothing to check against and
    /// nothing else it could mean. That one is logged as unverified rather
    /// than passed off as a match.
    pub async fn filament_source_for(
        &self,
        archive: &[u8],
        plate: u32,
    ) -> Result<FilamentSource, PrinterError> {
        let filaments = super::plate_filaments(archive, plate)
            .map_err(|err| PrinterError(format!("{}: {err}", self.base.name)))?;

        let report = self.print_status().await.ok_or_else(|| {
            PrinterError(format!(
                "{}: no print.push_status yet, so there is no way to know what filament is loaded",
                self.base.name
            ))
        })?;

        let (trays, tray_count) = available_trays(report.ams.as_ref());
        let spool = report.vt_tray.as_ref();

        // The AMS first, when there is one: a project sliced for the AMS
        // naming a filament that also happens to be on the spool should
        // still use the AMS, because that is what it was sliced for.
        if !trays.is_empty() {
            match super::map_to_trays(&filaments, &trays) {
                Ok(mapping) => {
                    tracing::info!(
                        printer = %self.base.name,
                        ?mapping,
                        tray_count,
                        "mapped the project's filaments onto AMS slots"
                    );
                    return FilamentSource::ams(mapping, tray_count);
                }
                Err(ams_err) => {
                    // Nothing in the AMS fits. The spool is the only other
                    // thing it could have meant, so check it rather than
                    // failing outright - but report the AMS error if the
                    // spool does not fit either, since that one names what
                    // was wanted and what was loaded.
                    if spool.is_some_and(|spool| external_spool_satisfies(&filaments, spool)) {
                        tracing::info!(
                            printer = %self.base.name,
                            "no AMS tray fits this project, but the external spool does"
                        );
                        return Ok(FilamentSource::ExternalSpool);
                    }
                    return Err(PrinterError(format!("{}: {ams_err}", self.base.name)));
                }
            }
        }

        match spool {
            Some(spool) if external_spool_satisfies(&filaments, spool) => {
                tracing::info!(
                    printer = %self.base.name,
                    "no AMS; the external spool matches this project"
                );
                Ok(FilamentSource::ExternalSpool)
            }
            // A spool is reported and does not match. Now that `vt_tray`
            // exists this is a real negative rather than an unknown, so it
            // is worth refusing on.
            Some(spool) => Err(PrinterError(format!(
                "{}: no AMS trays, and the external spool holds {} {} ({}), \
                 which does not match this project",
                self.base.name,
                spool.tray_type.clone().unwrap_or_else(|| "nothing".into()),
                spool.tray_info_idx.clone().unwrap_or_default(),
                spool.tray_color.clone().unwrap_or_default(),
            ))),
            // v2 omits `vt_tray`, so this is "not reported", not "empty".
            None => {
                tracing::warn!(
                    printer = %self.base.name,
                    "no AMS trays and no vt_tray reported; assuming the external spool, unverified"
                );
                Ok(FilamentSource::ExternalSpool)
            }
        }
    }

    /// Reads one file back off the printer's SD card.
    ///
    /// Read-only, and the counterpart to `upload_file`. Needed because a
    /// `.3mf` sitting on the card is the only record of which filament its
    /// plates were sliced for, which is what `ams_mapping` has to agree
    /// with.
    pub async fn download_file(&self, path: &str) -> Result<Vec<u8>, PrinterError> {
        let connector = self
            .tls
            .build_connector()
            .map_err(|err| PrinterError(format!("{}: TLS setup failed: {err}", self.base.name)))?;

        let host = self.base.host.clone();
        let access_code = self.access_code.clone();
        let path = path.to_string();

        tokio::task::spawn_blocking(move || {
            super::bambu_ftps::download(&host, &access_code, connector, &path)
        })
        .await
        .map_err(|err| PrinterError(format!("{}: FTPS download thread failed: {err}", self.base.name)))?
        .map_err(|err| PrinterError(format!("{}: FTPS download failed: {err}", self.base.name)))
    }

    /// Removes one file from the printer's SD card.
    ///
    /// Takes one explicit path, because the convenient version of this is
    /// also the dangerous one - see `bambu_ftps::delete`. Nothing calls it
    /// automatically yet: a gateway that uploads a project per job needs to
    /// clear them eventually, but *which* file is safe to remove is a
    /// question about the job queue, not about FTPS.
    pub async fn delete_file(&self, path: &str) -> Result<(), PrinterError> {
        let connector = self
            .tls
            .build_connector()
            .map_err(|err| PrinterError(format!("{}: TLS setup failed: {err}", self.base.name)))?;

        let host = self.base.host.clone();
        let access_code = self.access_code.clone();
        let path = path.to_string();

        tokio::task::spawn_blocking(move || {
            super::bambu_ftps::delete(&host, &access_code, connector, &path)
        })
        .await
        .map_err(|err| PrinterError(format!("{}: FTPS delete thread failed: {err}", self.base.name)))?
        .map_err(|err| PrinterError(format!("{}: FTPS delete failed: {err}", self.base.name)))
    }

    /// `list_cache` for any directory on the printer's SD card.
    ///
    /// Read-only, and the way to answer "what is the path to this file?"
    /// without sending a print command to find out - which matters because
    /// `print.gcode_file` answers a wrong path and a missing file with the
    /// same `result: "fail"`, `reason: "error string"`.
    pub async fn list_dir(&self, dir: &str) -> Result<Vec<String>, PrinterError> {
        let connector = self
            .tls
            .build_connector()
            .map_err(|err| PrinterError(format!("{}: TLS setup failed: {err}", self.base.name)))?;

        let host = self.base.host.clone();
        let access_code = self.access_code.clone();
        let dir = dir.to_string();

        tokio::task::spawn_blocking(move || {
            super::bambu_ftps::list(&host, &access_code, connector, &dir)
        })
        .await
        .map_err(|err| PrinterError(format!("{}: FTPS list thread failed: {err}", self.base.name)))?
        .map_err(|err| PrinterError(format!("{}: FTPS list failed: {err}", self.base.name)))
    }

    /// Sends `body` in `category` as a correlated request and hands back the
    /// printer's reply verbatim.
    ///
    /// An escape hatch for probing a command shape against real hardware,
    /// which is how every "confirmed live" note in this file was earned. Not
    /// for ordinary use - a command worth sending twice deserves a typed
    /// method next to the others. Kept public because the alternative is
    /// reshaping an API around a guess and then discovering the guess was
    /// wrong, which is exactly what `start_uploaded_file` did.
    pub async fn request_raw(
        &self,
        category: &str,
        body: serde_json::Value,
    ) -> Result<serde_json::Value, PrinterError> {
        let (session, _) = self
            .connection
            .as_ref()
            .ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;
        session.request(category, body).await
    }

    /// Interprets the `result`/`reason` pair a command's reply carries,
    /// turning a rejection into an `Err` instead of losing it.
    ///
    /// Used for `print` and `system` commands alike - both answer in this
    /// shape.
    ///
    /// Confirmed live on a P1P (2026-10-11): `print.gcode_file` naming a
    /// file that doesn't exist replies
    /// `{"command":"gcode_file","param":"/cache/...","sequence_id":"",
    /// "reason":"error string","result":"fail"}`. Before this, that reply
    /// was thrown away - `start_uploaded_file` published fire-and-forget,
    /// so asking the printer to print a file it doesn't have returned
    /// `Ok(())` and the print simply never happened. There is no
    /// distinguishing it from a successful start by watching
    /// `print.push_status` either: both leave the printer IDLE for a while.
    ///
    /// A reply with no `result` field at all is accepted, with a warning.
    /// A correlated reply arriving is itself proof the command was parsed
    /// (it echoed our `sequence_id` back), and not every firmware/command
    /// pair is known to include `result` - so this refuses to invent a
    /// failure, while still not hiding an unrecognised shape.
    fn command_outcome(
        printer: &str,
        command: &str,
        reply: &serde_json::Value,
    ) -> Result<(), PrinterError> {
        match reply.get("result").and_then(serde_json::Value::as_str) {
            Some("success") => Ok(()),
            Some(other) => {
                let reason = reply
                    .get("reason")
                    .and_then(serde_json::Value::as_str)
                    .filter(|reason| !reason.is_empty())
                    .map(|reason| format!(" (reason: {reason})"))
                    .unwrap_or_default();
                Err(PrinterError(format!(
                    "{printer} rejected print.{command}: result '{other}'{reason}"
                )))
            }
            None => {
                tracing::warn!(
                    target: "continuum_proxy::printer::bambu",
                    printer,
                    command,
                    reply = %reply,
                    "a command's reply carried no 'result' field; treating it as accepted"
                );
                Ok(())
            }
        }
    }

    /// Starts printing a file that's already on the printer's SD card.
    /// `upload_project_file` calls this straight after its FTPS transfer;
    /// it's also public on its own, so a file uploaded on some earlier run
    /// can be reprinted without sending the bytes again.
    ///
    /// Bambu has *two* commands here and which one is right depends on what
    /// kind of file it is, which is why this isn't one uniform command:
    ///
    /// - A `.3mf` is a project archive - a container with slicer settings,
    ///   plate layout and one or more gcode files inside it.
    ///   `print.project_file` takes `url` (where the archive is) and `param`
    ///   (which gcode file *inside* the archive to run).
    /// - A bare `.gcode` has no container and nothing to select, so
    ///   `print.gcode_file` just takes `param`: the path on the printer's
    ///   filesystem.
    ///
    /// ## The `.3mf` branch is confirmed live
    ///
    /// On a P1P, 2026-10-11, this exact body started a real print
    /// (`gcode_state: IDLE -> PREPARE -> RUNNING`, then stopped from
    /// `control_print`):
    ///
    /// ```json
    /// {"command":"project_file","param":"Metadata/plate_1.gcode",
    ///  "url":"ftp:///Cube.gcode.3mf","project_id":"0","profile_id":"0",
    ///  "task_id":"0","subtask_id":"0","subtask_name":"",
    ///  "use_ams":false,"ams_mapping":""}
    /// ```
    ///
    /// That settles two of the three things this comment used to flag as
    /// guesses:
    ///
    /// - **`url` spelling.** `ftp:///<name>` - empty host, triple slash,
    ///   resolved against the card's root - is accepted. `file:///` was
    ///   never needed and remains untested.
    /// - **An accepted start is not a successful print, and the failure can
    ///   be physical.** Confirmed end to end on 2026-10-15 via
    ///   `upload_project_file`: the mapping derived correctly
    ///   (`ams_mapping` slot 3, and the printer echoed `tray_tar: 3`), the
    ///   upload matched byte for byte, the printer went
    ///   `IDLE -> PREPARE -> RUNNING`, heated to 255C - and then paused
    ///   itself at stage 54 with `print_error: 0x07008003` and
    ///   `hms: [{attr: 0x07002300, code: 0x00020004}]`, i.e.
    ///   `HMS_0700_2300_0002_0004`, which is a broken-filament fault on the
    ///   slot it was loading. Nothing in this code was wrong; the filament
    ///   was. Which is the point: `command_outcome` can only report what
    ///   the printer said about the *command*, so anything that wants to
    ///   know whether a print is actually running has to watch
    ///   `gcode_state`, `print_error` and `hms` afterwards.
    /// - **`use_ams: false` / `ams_mapping: ""`.** Accepted by the printer -
    ///   and *wrong* on an AMS-equipped machine. Confirmed on a second run
    ///   the same day, against a project the owner had sliced in Bambu
    ///   Studio: the print started, heated to 250C, reached 1% and then the
    ///   printer **paused itself** with `print_error: 50364420`
    ///   (`0x03008004`), `hms: []`, `tray_now: 255`. 255 is the external
    ///   spool, which is what `use_ams: false` selects - so this hardcoded
    ///   pair overrode the slicer's own filament choice and pointed a
    ///   project sliced for the AMS at an external spool instead.
    ///
    ///   So this is not merely "the only case wired up", as this comment
    ///   used to say: it silently changes which filament a print uses.
    ///   Until it is a real parameter, `start_uploaded_file` can only be
    ///   trusted for projects that genuinely print from the external spool.
    ///   Fixing it means reading the plate's filament mapping out of the
    ///   `.3mf` and filling `use_ams` and `ams_mapping` together.
    ///
    /// Still a guess: **`param` is hardcoded to `Metadata/plate_1.gcode`**,
    /// i.e. always the first plate. Correct for the single-plate slice this
    /// was confirmed with, which is the common case, but a multi-plate
    /// project needs the caller to choose - a parameter to add once there's
    /// a caller that cares, not a default to guess harder at.
    ///
    /// ## The `.gcode` branch is not, and the obvious path for it is wrong
    ///
    /// The command *shape* is understood - the printer parses it and echoes
    /// `param` back - but both spellings tried against a file that
    /// demonstrably exists were rejected with `result: "fail"`,
    /// `reason: "error string"`:
    ///
    /// - `param: "/cache/Cube_plate_1.gcode"`, for a file the FTPS listing
    ///   shows is right there.
    ///
    /// The likely reason is in `bambu_ftps::SD_ROOT`: `cache/` holds the
    /// printer's own extracted per-plate gcode, with `.gcode.bbl` sidecars,
    /// and those extracted copies appear not to be independently printable.
    /// What this code now sends is `/<file_name>` against the root, for
    /// consistency with where `upload_file` puts things - **untested**,
    /// because testing it means uploading a bare `.gcode` and starting a
    /// real print. If you are the one who tests it, record the answer here.
    ///
    /// A rejection is at least no longer silent: both branches go out as
    /// correlated requests and `command_outcome` turns a `fail` into
    /// an `Err`. Before that, asking this printer to print a file it could
    /// not find returned `Ok(())`.
    ///
    /// The optional calibration flags (`bed_levelling`, `flow_cali`,
    /// `timelapse`, ...) are deliberately left unset rather than defaulted
    /// here: omitted, the printer applies its own settings, which is a
    /// better default than this method inventing one. They become real
    /// parameters when a caller actually wants to control them. Worth
    /// noting while you're in there: the schema spells it `bed_levelling`
    /// (two l's) and Bambu's firmware has been seen using `bed_leveling` -
    /// check which the printer accepts before relying on that field, and
    /// fix continuum-schemas if it's wrong. The confirmed body above sets
    /// neither, so this is still open.
    pub async fn start_uploaded_file(
        &self,
        file_name: &str,
        filament: &FilamentSource,
    ) -> Result<(), PrinterError> {
        let (session, version) =
            self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;

        let is_project = file_name.to_ascii_lowercase().ends_with(".3mf");
        let command = if is_project { "project_file" } else { "gcode_file" };

        let mut body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command(
            *version,
            "print".into(),
            command.into(),
        ))
        .map_err(|err| PrinterError(format!("failed to build {command} request: {err}")))?;

        if is_project {
            // `ftp:///<name>` against the card's root - confirmed live, this
            // exact spelling, see this method's doc comment. `upload_file`
            // puts the file there, so the two agree by construction.
            body["url"] = serde_json::Value::String(format!("ftp:///{file_name}"));
            body["param"] = serde_json::Value::String("Metadata/plate_1.gcode".into());

            // Local print, no Bambu-cloud job behind it: these are all the
            // literal string "0" rather than absent, per the schema's own
            // field descriptions.
            for key in ["project_id", "profile_id", "task_id", "subtask_id"] {
                body[key] = serde_json::Value::String("0".into());
            }
            body["subtask_name"] = serde_json::Value::String(String::new());

            // These two are a pair, not independent switches - see
            // `FilamentSource::apply`.
            filament.apply(&mut body);
        } else {
            // The root, for the same reason: it is where `upload_file` puts
            // things. Unverified - see this method's doc comment, which
            // records what the hardware said about the `cache/` spelling.
            body["param"] = serde_json::Value::String(format!("/{file_name}"));

            // `filament` has nowhere to go here: `gcode_file` takes only
            // `command` and `param` - no `use_ams`, no `ams_mapping` - so a
            // bare gcode draws whatever its own tool changes select. That is
            // correct, but it means this argument is silently discarded for
            // half the inputs this method accepts, and "a filament choice
            // quietly overridden" is precisely the bug `FilamentSource` was
            // introduced to stop. So say so rather than letting a caller
            // believe its mapping took effect.
            if !matches!(filament, FilamentSource::ExternalSpool) {
                tracing::warn!(
                    printer = %self.base.name,
                    file = file_name,
                    ?filament,
                    "ignoring the filament source: a bare .gcode has no use_ams/ams_mapping \
                     field to put it in, so the gcode's own tool changes decide"
                );
            }
        }

        // A correlated `request`, not a fire-and-forget publish. This was
        // the latter until a P1P was asked to print a file it didn't have
        // and happily returned `Ok(())` - see `command_outcome`.
        let reply = session.request("print", body).await?;
        Self::command_outcome(&self.base.name, command, &reply)
    }

    /// The current in-memory `print.push_status` object - `None` until at
    /// least one has arrived (which, on v1, requires having called
    /// `pushall` at least once since connecting; v2 pushes on its own).
    pub async fn print_status(&self) -> Option<super::PrintReport> {
        match self.connection.as_ref() {
            Some((session, _)) => session.print_status().await,
            None => None,
        }
    }
}

impl GenericPrinter for BambuPrinter {
    /// OVERRIDE: one MQTT publish in the `print` category. Bambu's command
    /// names are exactly `pause`/`resume`/`stop`, and all three schemas take
    /// nothing but the envelope plus an empty `param`, so the schema-driven
    /// builder produces the whole body.
    async fn control_print(&mut self, action: super::PrintControl) -> Result<(), PrinterError> {
        let (session, version) = self
            .connection
            .as_ref()
            .ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;

        let body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command(
            *version,
            "print".into(),
            action.as_str().into(),
        ))
        .map_err(|err| PrinterError(format!("failed to build print {} request: {err}", action.as_str())))?;

        // Correlated, for the same reason as `start_uploaded_file`: a
        // pause that the printer refused must not read as a pause that
        // happened.
        //
        // Confirmed live on a P1P (2026-10-11): all three reply, echoing
        // the `sequence_id` this crate generated, as
        // `{"command":"pause","sequence_id":"cxproxy-1",
        // "reason":"success","result":"success"}`.
        //
        // Note what that reply does *not* mean. All three answered
        // `result: "success"` while the printer was IDLE with nothing to
        // pause, resume or stop. So `Ok(())` here means the printer
        // accepted the command, not that the print state changed - a
        // caller that needs to know the printer actually paused has to
        // read `gcode_state` back via `print_status()`. The value of the
        // reply is the negative case: a rejection is no longer silent.
        let reply = session.request("print", body).await?;
        Self::command_outcome(&self.base.name, action.as_str(), &reply)
    }


    fn base(&self) -> &PrinterBase {
        &self.base
    }

    /// Real, not a stub: opens the MQTT session (see `bambu_mqtt.rs` for
    /// `preference`'s `Auto`-tries-v5-then-v3.1.1 fallback) and keeps it -
    /// same reasoning as `PrusaSerialPrinter::connect` holding its port
    /// open, just an MQTT session instead of a serial port.
    ///
    /// `ensure_trusted` runs first so a newly-added printer with no
    /// `ca_cert_path` configured just works on the first real connect
    /// attempt: bundled CA if it verifies, auto-pin via trust-on-first-
    /// connect if it doesn't (P1P, at least) - the same thing
    /// `examples/auto_connect_bambu.rs` already did by hand, now
    /// automatic. A printer with an explicit `ca_cert_path` (already
    /// resolved to `Custom` in `fleet.rs`) is untouched by this - see
    /// `ensure_trusted`'s own doc comment for why it never auto-pins over
    /// an operator's explicit choice.
    ///
    /// `ensure_trusted` is a blocking call (a raw, synchronous TCP+TLS
    /// handshake) run here inside an `async fn` - tolerable for a couple
    /// of printers connecting concurrently on tokio's multi-thread
    /// runtime, but at real farm scale this should move to
    /// `tokio::task::spawn_blocking` so a slow/unreachable printer's
    /// trust check can't tie up a worker thread other printers' connects
    /// need.
    async fn connect(&mut self) -> Result<(), PrinterError> {
        self.ensure_trusted(Path::new(CERT_DIR))
            .map_err(|err| PrinterError(format!("{} certificate trust failed: {err}", self.base.name)))?;

        let (session, version) = super::bambu_mqtt::BambuMqttSession::open(
            &self.base.host,
            &self.access_code,
            &self.sn,
            &self.tls,
            self.preference,
        )
        .await?;

        println!("[{}] connected over MQTT ({version:?})", self.base.name);
        self.connection = Some((session, version));
        Ok(())
    }

    /// Real: a raw pass-through publish to `device/{sn}/request`, no
    /// category envelope opinion - `send_gcode_line` below builds and
    /// sends its own properly-wrapped request instead of going through
    /// this.
    async fn publish(&mut self, payload: String) -> Result<(), PrinterError> {
        let (session, _) =
            self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;
        session.publish_raw(payload).await
    }

    /// Same MQTT gcode-injection mechanism regardless of protocol
    /// generation - just routed through `send_gcode_line` below instead of
    /// its own separate stub, since that's genuinely the same operation.
    async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> {
        let pwm = super::percent_to_pwm(percent);
        self.send_gcode_line(format!("M106 P1 S{pwm}")).await
    }

    /// Real, both halves: FTPS-`PUT`s `data` into the printer's own
    /// `cache/` directory (storage the printer's firmware owns, same idea
    /// as PrusaLink's SD card), then publishes the MQTT command that starts
    /// printing it. Two genuinely separate protocols - see
    /// `bambu_ftps.rs` for the transfer and `start_uploaded_file` below for
    /// the command.
    ///
    /// The FTPS half is blocking code (suppaftp's sync client, so it can
    /// reuse `BambuTls::build_connector()` - see `bambu_ftps.rs`'s doc
    /// comment), so it runs on `tokio::task::spawn_blocking`. That's
    /// tokio's "this call parks a whole thread, keep it off the async
    /// worker threads" escape hatch; the nearest TS equivalent is moving
    /// work to a worker thread so it can't stall the event loop. Note this
    /// is the *right* fix for the problem `connect()`'s doc comment above
    /// flags about `ensure_trusted` still being blocking inline - a
    /// multi-megabyte upload is far more obviously worth getting off the
    /// runtime than a handshake is.
    async fn upload_project_file(
        &mut self,
        file_name: String,
        data: Vec<u8>,
    ) -> Result<(), PrinterError> {
        let byte_count = data.len();

        // Worked out *before* the upload, from the bytes still in hand. If
        // this project cannot be mapped to what the printer has loaded, the
        // right outcome is to say so and not move a file onto the machine
        // that cannot then be printed.
        let filament = self.filament_source_for(&data, FIRST_PLATE).await?;

        let written = self.upload_file(&file_name, data).await?;

        tracing::info!(
            printer = %self.base.name,
            file = %file_name,
            written,
            byte_count,
            "uploaded a project file to the SD card root"
        );

        // These two were logged next to each other and never compared, which
        // made a short write something you could only notice by reading the
        // logs afterwards - and the next line would have printed the
        // truncated file anyway. A `.3mf` is a zip: cut the end off and the
        // central directory goes with it, so the printer gets an archive it
        // cannot open rather than a partial model. It is also the failure a
        // slow transfer invites, and this one is slow: 4.96 MB took 31.3s on
        // a P1P (2026-10-15), so a 62 MB project is minutes of exposure.
        if written != byte_count as u64 {
            return Err(PrinterError(format!(
                "{}: uploaded {written} bytes of {byte_count} - refusing to print a \
                 truncated project",
                self.base.name
            )));
        }

        self.start_uploaded_file(&file_name, &filament).await
    }

    /// Builds the real JSON payload straight from continuum-schemas'
    /// `schemas/bambulab/{v1,v2}/request/print/gcode_line.schema.json` -
    /// which one depends on `self.version`, settled by `connect()` above -
    /// instead of a hand-written serde struct: `sequence_id` and the
    /// `command` const get filled in by the schema itself, `param` is the
    /// one field we actually set. Real, end to end: publishes over the
    /// live MQTT session `connect()` opened, no waiting for a reply -
    /// `gcode_line` is fire-and-forget, the printer just executes it.
    async fn send_gcode_line(&mut self, gcode_line: String) -> Result<(), PrinterError> {
        let (session, version) =
            self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?;

        let mut body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command(
            *version,
            "print".into(),
            "gcode_line".into(),
        ))
        .map_err(|err| PrinterError(format!("failed to build gcode_line request: {err}")))?;
        body["param"] = serde_json::Value::String(gcode_line);

        session.publish("print", &body).await
    }
}

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

    /// The padding rule from
    /// `schemas/bambulab/v1/common/ams/ams_mapping.schema.json`: always pad
    /// to every tray the printer has. An under-length array is one of the
    /// ways a mapping goes wrong quietly.
    #[test]
    fn an_ams_mapping_is_padded_to_the_tray_count() {
        let FilamentSource::Ams(slots) = FilamentSource::ams(vec![3], 5).expect("valid") else {
            panic!("expected an AMS source");
        };
        assert_eq!(slots, vec![3, -1, -1, -1, -1]);
    }

    #[test]
    fn an_already_full_length_mapping_is_left_alone() {
        let FilamentSource::Ams(slots) =
            FilamentSource::ams(vec![3, -1, 0, -1, -1], 5).expect("valid")
        else {
            panic!("expected an AMS source");
        };
        assert_eq!(slots, vec![3, -1, 0, -1, -1]);
    }

    #[test]
    fn an_empty_mapping_is_refused_rather_than_padded_into_nonsense() {
        // Padding `[]` would produce `[-1,-1,-1,-1,-1]`: `use_ams: true` with
        // every filament unmapped, which is not a thing to send a printer.
        assert!(FilamentSource::ams(Vec::new(), 5).is_err());
    }

    #[test]
    fn a_mapping_longer_than_the_printer_has_trays_is_refused() {
        assert!(FilamentSource::ams(vec![0, 1, 2, 3, 0, 1], 5).is_err());
    }

    #[test]
    fn minus_one_is_the_only_negative_slot_allowed() {
        assert!(FilamentSource::ams(vec![-1], 5).is_ok());
        assert!(FilamentSource::ams(vec![-2], 5).is_err());
    }

    /// The pair that cost a real print: `use_ams` and `ams_mapping` have to
    /// move together, and the no-AMS case needs the empty *string* rather
    /// than an empty array - the schema's trailing `oneOf` says so.
    #[test]
    fn the_external_spool_sends_an_empty_string_not_an_empty_array() {
        let mut body = serde_json::json!({});
        FilamentSource::ExternalSpool.apply(&mut body);
        assert_eq!(body["use_ams"], serde_json::json!(false));
        assert_eq!(body["ams_mapping"], serde_json::json!(""));
    }

    fn real_report() -> super::super::PrintReport {
        let raw = include_str!("../../tests/fixtures/push_status_v1_p1p_idle.json");
        let value: serde_json::Value = serde_json::from_str(raw).expect("fixture is json");
        let body = value.get("print").unwrap_or(&value);
        serde_json::from_value(body.clone()).expect("fixture should parse")
    }

    fn filament(tray_info_idx: &str, filament_type: &str, color: &str) -> super::super::ProjectFilament {
        super::super::ProjectFilament {
            index: 0,
            tray_info_idx: tray_info_idx.into(),
            filament_type: filament_type.into(),
            color: color.into(),
        }
    }

    /// The real P1P's external spool: PETG `P9599a25` in `D3C5A3FF`.
    #[test]
    fn the_real_external_spool_satisfies_a_project_asking_for_it() {
        let report = real_report();
        let spool = report.vt_tray.as_ref().expect("the v1 fixture has a vt_tray");

        assert!(external_spool_satisfies(
            &[filament("P9599a25", "PETG", "D3C5A3")],
            spool
        ));
    }

    /// The case that used to be indistinguishable from the one above: the
    /// project wants a filament the spool does not hold. Before `vt_tray`
    /// was modelled there was no way to tell these apart, so the code
    /// assumed a match - which is the whole `use_ams: false` bug.
    #[test]
    fn the_real_external_spool_does_not_satisfy_a_different_filament() {
        let report = real_report();
        let spool = report.vt_tray.as_ref().expect("the v1 fixture has a vt_tray");

        // Right material, wrong product and colour - the AMS's blue GFG99.
        assert!(!external_spool_satisfies(
            &[filament("GFG99", "PETG", "2850E0")],
            spool
        ));
    }

    #[test]
    fn an_empty_external_spool_satisfies_nothing() {
        let empty = super::super::VtTray {
            id: "254".into(),
            tray_info_idx: Some(String::new()),
            tray_type: Some(String::new()),
            tray_color: Some("00000000".into()),
            bed_temp: None,
            bed_temp_type: None,
            nozzle_temp_max: None,
            nozzle_temp_min: None,
            remain: None,
            tag_uid: None,
            tray_diameter: None,
            tray_id_name: None,
            tray_sub_brands: None,
            tray_uuid: None,
            tray_weight: None,
            xcam_info: None,
            tray_temp: None,
            tray_time: None,
            cali_idx: None,
            k: None,
            n: None,
        };

        assert!(!external_spool_satisfies(
            &[filament("P9599a25", "PETG", "D3C5A3")],
            &empty
        ));
    }

    /// The AMS half of the mapping, against the real P1P report in
    /// `tests/fixtures/push_status_v1_p1p_idle.json`: one unit, slots 0 and
    /// 1 empty, slots 2 and 3 holding PETG in two different colours.
    #[test]
    fn real_ams_report_flattens_to_the_loaded_slots() {
        let raw = include_str!("../../tests/fixtures/push_status_v1_p1p_idle.json");
        let value: serde_json::Value = serde_json::from_str(raw).expect("fixture is json");
        let body = value.get("print").unwrap_or(&value);
        let report: super::super::PrintReport =
            serde_json::from_value(body.clone()).expect("fixture should parse");

        let (trays, total) = available_trays(report.ams.as_ref());

        assert_eq!(trays.len(), 2, "only slots 2 and 3 hold filament");
        assert_eq!(trays[0].slot, 2);
        assert_eq!(trays[1].slot, 3);
        // The printer reports RGBA; the project writes RGB.
        assert_eq!(trays[0].color, "161616");
        assert_eq!(trays[1].color, "2850E0");

        // 4 AMS slots - including the two empty ones - plus one external
        // spool. The padding rule is over every tray, not every loaded tray,
        // and an under-length array is one of the quiet failure modes.
        assert_eq!(total, 5);
    }

    /// The whole thing, end to end, on the two real captures: the project
    /// that cost a print and the AMS report of the printer it was sent to.
    /// This is the case the old `use_ams: false` got wrong.
    #[test]
    fn the_real_project_and_the_real_printer_agree_on_slot_three() {
        let raw = include_str!("../../tests/fixtures/push_status_v1_p1p_idle.json");
        let value: serde_json::Value = serde_json::from_str(raw).expect("fixture is json");
        let body = value.get("print").unwrap_or(&value);
        let report: super::super::PrintReport =
            serde_json::from_value(body.clone()).expect("fixture should parse");
        let (trays, total) = available_trays(report.ams.as_ref());

        let project = include_bytes!("../../tests/fixtures/cube_slice_info.3mf");
        let filaments = super::super::plate_filaments(project, FIRST_PLATE).expect("should parse");

        let mapping = super::super::map_to_trays(&filaments, &trays).expect("should match");
        let source = FilamentSource::ams(mapping, total).expect("valid");

        let mut sent = serde_json::json!({});
        source.apply(&mut sent);
        assert_eq!(sent["use_ams"], serde_json::json!(true));
        assert_eq!(sent["ams_mapping"], serde_json::json!([3, -1, -1, -1, -1]));
    }

    #[test]
    fn an_ams_source_sends_an_array_and_turns_use_ams_on() {
        let mut body = serde_json::json!({});
        FilamentSource::ams(vec![3], 5).expect("valid").apply(&mut body);
        assert_eq!(body["use_ams"], serde_json::json!(true));
        assert_eq!(body["ams_mapping"], serde_json::json!([3, -1, -1, -1, -1]));
        // Never the external-spool spelling while use_ams is on: that
        // combination is what silently printed from the wrong source.
        assert!(!body["ams_mapping"].is_string());
    }
}