Add a read-only dry run of a print's filament mapping

bambu_mapping_check downloads one project from the card and prints the AMS
mapping the production path would send, starting nothing.

.env.example

@@ -1,10 +1,12 @@
 # Identity
 CONTINUUM_GATEWAY_ID=gw-farm01-rpi01
 CONTINUUM_FARM_ID=replace-with-farm-uuid
-# The edge JWT from POST /auth/edge-token. Sent as an Authorization: Bearer
-# header on the uplink's WebSocket upgrade - not in a message and not in the
-# URL - so the backend verifies it before any socket exists.
-CONTINUUM_GATEWAY_TOKEN=replace-with-provisioning-token
+# The farm's edge proxy token, shown once by the console's Setup page (or
+# POST /farms/:id/edge-proxy-token). The gateway trades it for a one-hour edge
+# JWT (POST /auth/edge-token) on every connection, and the JWT - sent as an
+# Authorization: Bearer header on the WebSocket upgrade, never in a message
+# or the URL - is what the backend verifies before any socket exists.
+CONTINUUM_GATEWAY_TOKEN=replace-with-the-farms-edge-proxy-token
 
 # Cloud uplink
 CONTINUUM_UPLINK_URL=wss://api.continuum.local/ws/edge/v1

.gitignore

@@ -14,3 +14,7 @@ go2rtc.yaml
 go2rtc.yaml.tmp
 continuum.toml
 go2rtc.yaml.pid
+
+# The local store holds printer access codes.
+continuum.db
+continuum.db-*

Cargo.lock

@@ -208,6 +208,7 @@ dependencies = [
  "openssl",
  "reqwest",
  "rumqttc",
+ "rusqlite",
  "serde",
  "serde_json",
  "suppaftp",
@@ -375,6 +376,18 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "fallible-iterator"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649"
+
+[[package]]
+name = "fallible-streaming-iterator"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a"
+
 [[package]]
 name = "fastrand"
 version = "2.5.0"
@@ -415,6 +428,12 @@ dependencies = [
  "spin",
 ]
 
+[[package]]
+name = "foldhash"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb"
+
 [[package]]
 name = "foreign-types"
 version = "0.3.2"
@@ -533,6 +552,18 @@ name = "hashbrown"
 version = "0.17.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
+dependencies = [
+ "foldhash",
+]
+
+[[package]]
+name = "hashlink"
+version = "0.12.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a596f1b20ed2cc5ecac41a164aaebc7258057060f06c0cf7a2ba3991ee7990fb"
+dependencies = [
+ "hashbrown",
+]
 
 [[package]]
 name = "hmac"
@@ -875,6 +906,17 @@ version = "0.2.191"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "65cd87c9a6d60536580b316ad985726104730a5151c99845f2490428ef91240a"
 
+[[package]]
+name = "libsqlite3-sys"
+version = "0.39.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "14a5f01b87b92705d07b433c06f96cbc764637d81a0bffb5734ce17cf0f9a2b3"
+dependencies = [
+ "cc",
+ "pkg-config",
+ "vcpkg",
+]
+
 [[package]]
 name = "linux-raw-sys"
 version = "0.12.1"
@@ -1369,6 +1411,15 @@ dependencies = [
  "windows-sys 0.52.0",
 ]
 
+[[package]]
+name = "rsqlite-vfs"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "353b74f9810d5b085b6c42c9d86a5f0fdb795553235cdd48753b32e3809d18a7"
+dependencies = [
+ "thiserror 2.0.21",
+]
+
 [[package]]
 name = "rumqttc"
 version = "0.25.1"
@@ -1388,6 +1439,21 @@ dependencies = [
  "tokio-util",
 ]
 
+[[package]]
+name = "rusqlite"
+version = "0.41.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da9bba6f3506135e120a6586df05be18b48d18896600646a4bec8f993d91a594"
+dependencies = [
+ "bitflags 2.13.2",
+ "fallible-iterator",
+ "fallible-streaming-iterator",
+ "hashlink",
+ "libsqlite3-sys",
+ "smallvec",
+ "sqlite-wasm-rs",
+]
+
 [[package]]
 name = "rustc-hash"
 version = "2.1.3"
@@ -1670,6 +1736,16 @@ dependencies = [
  "lock_api",
 ]
 
+[[package]]
+name = "sqlite-wasm-rs"
+version = "0.6.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "36bd8e29bf08189a005b2df1df99823aa0861cabe9044e5f31ed54ca292ee618"
+dependencies = [
+ "cc",
+ "rsqlite-vfs",
+]
+
 [[package]]
 name = "stable_deref_trait"
 version = "1.2.1"

Cargo.toml

@@ -92,3 +92,9 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-
 # `bind_ssdp`). Windows PCs routinely have VPN/Hyper-V/WSL adapters.
 # MIT OR BSD-3-Clause - checked against docs/LICENSES.md before adding.
 if-addrs = "0.15"
+# The proxy's local store (src/store.rs): printers added from the console live
+# here, so the gateway keeps its fleet with the API unreachable. `bundled`
+# compiles SQLite from source so the appliance image needs no system library
+# (SQLite itself is public domain).
+# MIT - checked with `cargo deny check licenses` (docs/LICENSES.md policy).
+rusqlite = { version = "0.41", features = ["bundled"] }

continuum.example.toml

@@ -19,6 +19,8 @@
 
 gateway_id = "REPLACE_ME"
 farm_id = "REPLACE_ME"
+# The farm's edge proxy token from the console (Setup page). Traded for a
+# short-lived edge JWT on every connection - see src/uplink/auth.rs.
 gateway_token = "REPLACE_ME"
 
 # uplink_url = "wss://api.continuum.local/ws/edge/v1"

examples/bambu_mapping_check.rs

@@ -0,0 +1,55 @@
+//! Run with: cargo run --example bambu_mapping_check -- <printer-id> <file-on-card>
+//!
+//! **Read-only dry run of what a print would do.** Connects, reads the
+//! printer's state and loaded filament, downloads one project from the SD
+//! card root over FTPS, and prints the AMS mapping the production path would
+//! send - without sending it. Starts nothing, uploads nothing, deletes
+//! nothing. An optional third argument saves the downloaded copy locally.
+
+use std::time::Duration;
+
+use continuum_proxy::fleet;
+use continuum_proxy::printer::{GenericPrinter, PrinterHandle};
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let mut args = std::env::args().skip(1);
+    let (Some(target), Some(file)) = (args.next(), args.next()) else {
+        anyhow::bail!("usage: cargo run --example bambu_mapping_check -- <printer-id> <file-on-card> [save-to]");
+    };
+    let save_to = args.next();
+
+    let mut printers = fleet::load(std::path::Path::new("printers.toml"))?;
+    let printer = printers
+        .iter_mut()
+        .find(|p| p.base().id == target)
+        .ok_or_else(|| anyhow::anyhow!("no printer with id '{target}' in printers.toml"))?;
+    let PrinterHandle::Bambu(bambu) = printer else {
+        anyhow::bail!("'{target}' is not a Bambu printer");
+    };
+
+    tokio::time::timeout(Duration::from_secs(20), bambu.connect())
+        .await
+        .map_err(|_| anyhow::anyhow!("connect timed out"))??;
+    bambu.pushall().await.ok();
+    tokio::time::sleep(Duration::from_secs(4)).await;
+
+    let report = bambu.print_status().await.ok_or_else(|| anyhow::anyhow!("no status yet"))?;
+    println!("gcode_state: {:?}", report.gcode_state);
+    println!("{:#?}", report.ams);
+
+    let bytes = bambu.download_file(&format!("/{file}")).await?;
+    println!("downloaded {} bytes", bytes.len());
+    if let Some(path) = save_to {
+        std::fs::write(&path, &bytes)?;
+        println!("saved a copy to {path}");
+    }
+    for f in continuum_proxy::printer::plate_filaments(&bytes, 1)? {
+        println!("project filament {}: {} {} #{}", f.index, f.filament_type, f.tray_info_idx, f.color);
+    }
+    match bambu.filament_source_for(&bytes, 1).await {
+        Ok(source) => println!("mapping the production path would use: {source:?}"),
+        Err(err) => println!("the production path would REFUSE: {err}"),
+    }
+    Ok(())
+}

src/camera.rs

@@ -18,9 +18,10 @@
 //! Only the X1/H2 class publishes an `rtsp_url`. Confirmed against two real
 //! machines: an H2C reports `rtsps://<ip>:322/streaming/live/1`, and a P1P
 //! reports an `ipcam` block with no URL in it at all. A P1/A1 camera is
-//! reachable only over Bambu's proprietary port-6000 JPEG stream, which this
-//! crate does not speak, so those printers get no stream here and say so
-//! rather than advertising one that will not play.
+//! reachable only over Bambu's proprietary port-6000 JPEG stream. That is
+//! not restreamable here (go2rtc cannot make WebRTC of MJPEG without ffmpeg),
+//! so those printers get no go2rtc stream; they are served single frames by
+//! `printer::bambu_camera` instead and advertise `camera_snapshot`.
 //!
 //! That is not just an inference from the missing field - it is confirmed at
 //! the network layer (2026-10-15). The H2C accepts TCP on 322; the P1P
@@ -131,6 +132,14 @@ pub fn source_for(printer: &PrinterHandle, report_url: Option<&str>) -> Option<C
     })
 }
 
+/// Whether this printer's camera is the P1/A1 kind: it reports a camera
+/// block but no RTSP URL, so the only way to see it is single JPEG frames
+/// from port 6000 (`printer::bambu_camera`). The one rule for "snapshot
+/// camera", used by telemetry, so what is advertised is what is served.
+pub fn wants_snapshot(printer: &PrinterHandle, has_ipcam: bool, report_url: Option<&str>) -> bool {
+    matches!(printer, PrinterHandle::Bambu(_)) && has_ipcam && report_url.is_none_or(str::is_empty)
+}
+
 /// Renders a go2rtc configuration.
 ///
 /// Hand-written YAML rather than a serializer, for the same reason the

src/config.rs

@@ -45,7 +45,7 @@ use std::time::Duration;
 
 use serde::Deserialize;
 
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub struct Config {
     pub gateway_id: String,
     pub farm_id: String,
@@ -76,6 +76,9 @@ pub struct Config {
     /// `BambuPrinter::ensure_trusted`) lives entirely in that file now,
     /// not as a separate fleet-wide env-configured policy here.
     pub printers_path: String,
+    /// The gateway's own database (src/store.rs): printers added from the
+    /// console, credentials included, so the file is created owner-only.
+    pub store_path: String,
     /// How, and whether, to look for printers nobody has configured.
     ///
     /// Off unless switched on - see [`DiscoverySettings`] for why that is
@@ -104,6 +107,7 @@ pub struct ProxyFile {
     pub go2rtc_api_listen: Option<String>,
     pub stream_base_url: Option<String>,
     pub printers_path: Option<String>,
+    pub store_path: Option<String>,
     /// Only meaningful on a Proxy Box; see the module docs for why it is
     /// exposed and not applied.
     pub network: Option<NetworkSettings>,
@@ -283,6 +287,22 @@ impl fmt::Debug for WifiSettings {
     }
 }
 
+/// By hand: the derived `Debug` would print `gateway_token`, the credential
+/// that opens this farm's gateway socket, into any log line that formats the
+/// config.
+impl fmt::Debug for Config {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Config")
+            .field("gateway_id", &self.gateway_id)
+            .field("farm_id", &self.farm_id)
+            .field("gateway_token", &"<redacted>")
+            .field("uplink_url", &self.uplink_url)
+            .field("printers_path", &self.printers_path)
+            .field("store_path", &self.store_path)
+            .finish_non_exhaustive()
+    }
+}
+
 impl Config {
     /// Environment only - the behaviour from before there was a file, kept
     /// so existing callers and tests do not change.
@@ -326,6 +346,7 @@ impl Config {
                 .or_else(|| file.stream_base_url.clone())
                 .filter(|v| !v.is_empty()),
             printers_path: or_default("CONTINUUM_PRINTERS_PATH", &file.printers_path, "printers.toml"),
+            store_path: or_default("CONTINUUM_STORE_PATH", &file.store_path, "continuum.db"),
             discovery: discovery_settings(file.discovery.as_ref(), env)?,
         })
     }

src/discovery/report.rs

@@ -32,7 +32,12 @@ use crate::uplink::ClientMessage;
 /// long-running tasks. With discovery off it parks on a future that never
 /// completes, which costs nothing and keeps the "every arm runs forever"
 /// property of that `select!` true.
-pub async fn run(settings: DiscoverySettings, printers_path: String, outbound: Sender<ClientMessage>) {
+pub async fn run(
+    settings: DiscoverySettings,
+    printers_path: String,
+    store: Option<std::sync::Arc<crate::store::Store>>,
+    outbound: Sender<ClientMessage>,
+) {
     if !settings.enabled {
         tracing::info!("printer discovery is off (discovery.enabled); not scanning this network");
         std::future::pending::<()>().await;
@@ -64,7 +69,20 @@ pub async fn run(settings: DiscoverySettings, printers_path: String, outbound: S
         // restart - but someone who has *edited* the file and not restarted
         // should not keep being offered the printer they just configured.
         let configured = match crate::fleet::configured(std::path::Path::new(&printers_path)) {
-            Ok(configured) => configured,
+            Ok(mut configured) => {
+                // Printers added from the console live in the store, not the
+                // file, and are just as "already configured".
+                if let Some(store) = &store {
+                    match store.printers() {
+                        Ok(stored) => configured.add_entries(stored),
+                        Err(err) => {
+                            tracing::warn!(%err, "cannot read the local store; skipping this discovery pass");
+                            continue;
+                        }
+                    }
+                }
+                configured
+            }
             // Fail the pass rather than report against an empty set. The
             // alternative offers every printer the farm already runs as a
             // fresh find, which is worse than no report: it is a list of

src/fleet.rs

@@ -6,7 +6,7 @@
 use std::collections::HashSet;
 use std::path::Path;
 
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 
 use crate::printer::{
     BambuPrinter, BambuProtocolPreference, BambuTls, KlipperPrinter, OctoPrintFlavour, OctoPrintPrinter,
@@ -19,10 +19,14 @@ struct FleetFile {
     printers: Vec<PrinterEntry>,
 }
 
-#[derive(Debug, Deserialize)]
-struct PrinterEntry {
-    id: String,
-    name: String,
+/// One printer's connection details, as written in `printers.toml` or stored
+/// in the local store (src/store.rs). `Debug` is written by hand below: the
+/// derived one would print `access_code` and `api_key` into any log line that
+/// formats an entry.
+#[derive(Clone, Deserialize, Serialize)]
+pub struct PrinterEntry {
+    pub id: String,
+    pub name: String,
     /// One of: "bambu" (auto-detects MQTT v5-vs-v3.1.1/BambuMQTT v2-vs-v1 -
     /// the default, since a firmware update can move the same physical
     /// printer between them), "bambu_v1"/"bambu_v2" (force a specific one),
@@ -32,23 +36,38 @@ struct PrinterEntry {
     /// flavour - see `OctoPrintFlavour`. Pick `octoprint` if it is OctoPrint
     /// on a Pi and `prusa_link` if it is an MK4/XL/MINI serving its own API;
     /// discovery cannot tell, so it suggests `octoprint` and says so.
-    vendor: String,
-    host: Option<String>,
-    access_code: Option<String>,
+    pub vendor: String,
+    pub host: Option<String>,
+    pub access_code: Option<String>,
     /// The printer's serial number - Bambu's real MQTT topics are addressed
     /// by serial (`device/{sn}/report`), not by our own `id`.
-    sn: Option<String>,
-    api_key: Option<String>,
+    pub sn: Option<String>,
+    pub api_key: Option<String>,
     /// Serial device path, e.g. `/dev/ttyACM0` on Linux.
-    port: Option<String>,
+    pub port: Option<String>,
     #[serde(default = "default_baud_rate")]
-    baud_rate: u32,
+    pub baud_rate: u32,
     /// Path to a printer-specific CA certificate. Leave unset to use the
     /// bundled Bambu CA - only needed for a printer that doesn't chain to
     /// it (P1P, at least; see certs/README.md).
-    ca_cert_path: Option<String>,
+    pub ca_cert_path: Option<String>,
     #[serde(default = "default_true")]
-    require_valid_cert: bool,
+    pub require_valid_cert: bool,
+}
+
+impl std::fmt::Debug for PrinterEntry {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let secret = |value: &Option<String>| value.as_ref().map(|_| "<redacted>");
+        f.debug_struct("PrinterEntry")
+            .field("id", &self.id)
+            .field("name", &self.name)
+            .field("vendor", &self.vendor)
+            .field("host", &self.host)
+            .field("access_code", &secret(&self.access_code))
+            .field("sn", &self.sn)
+            .field("api_key", &secret(&self.api_key))
+            .finish_non_exhaustive()
+    }
 }
 
 fn default_true() -> bool {
@@ -72,6 +91,26 @@ pub fn load(path: &Path) -> anyhow::Result<Vec<PrinterHandle>> {
     file.printers.into_iter().map(build_printer).collect()
 }
 
+/// The entries in a printers file, or none if there is no file.
+///
+/// A missing file is normal now: a gateway whose printers were all added from
+/// the console has only the store. A file that exists but does not parse is
+/// still an error - that is a typo to surface, not a fleet to ignore.
+pub fn entries(path: &Path) -> anyhow::Result<Vec<PrinterEntry>> {
+    let text = match std::fs::read_to_string(path) {
+        Ok(text) => text,
+        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+        Err(err) => return Err(anyhow::anyhow!("failed to read {path:?}: {err}")),
+    };
+    Ok(toml::from_str::<FleetFile>(&text)?.printers)
+}
+
+/// Builds the client for one entry. Public so a printer added while the
+/// gateway is running goes through exactly the checks a startup one does.
+pub fn build(entry: PrinterEntry) -> anyhow::Result<PrinterHandle> {
+    build_printer(entry)
+}
+
 /// Everything the fleet file already names, for telling a printer the
 /// operator has dealt with from one they have never seen.
 ///
@@ -119,7 +158,13 @@ pub fn configured(path: &Path) -> anyhow::Result<Configured> {
     };
     let file: FleetFile = toml::from_str(&text)?;
     let mut out = Configured::default();
-    for entry in file.printers {
+    out.add_entries(file.printers);
+    Ok(out)
+}
+
+impl Configured {
+    /// Folds more entries in - the file's, and the local store's.
+    pub fn add_entries(&mut self, entries: impl IntoIterator<Item = PrinterEntry>) {
         let add = |set: &mut HashSet<String>, value: Option<String>| {
             if let Some(value) = value {
                 let value = value.trim().to_lowercase();
@@ -128,11 +173,12 @@ pub fn configured(path: &Path) -> anyhow::Result<Configured> {
                 }
             }
         };
-        add(&mut out.hosts, entry.host);
-        add(&mut out.serials, entry.sn);
-        add(&mut out.ports, entry.port);
+        for entry in entries {
+            add(&mut self.hosts, entry.host);
+            add(&mut self.serials, entry.sn);
+            add(&mut self.ports, entry.port);
+        }
     }
-    Ok(out)
 }
 
 fn build_printer(entry: PrinterEntry) -> anyhow::Result<PrinterHandle> {

src/lib.rs

@@ -16,4 +16,6 @@ pub mod config_import;
 pub mod discovery;
 pub mod fleet;
 pub mod printer;
+pub mod provision;
+pub mod store;
 pub mod uplink;

src/main.rs

@@ -68,7 +68,30 @@ async fn main() -> anyhow::Result<()> {
     // Loaded and connected here, once, at startup - nothing yet re-reads
     // printers.toml on a running daemon, so adding or removing a printer
     // still means a restart.
-    let mut fleet = fleet::load(std::path::Path::new(&config.printers_path))?;
+    // Two sources, one fleet: `printers.toml` (a hand-written file - strict,
+    // a typo stops startup) and the local store (printers added from the
+    // console, which were checked before they were saved). The store is
+    // opened first and is what lets the gateway come up with no file at all.
+    let store = std::sync::Arc::new(continuum_proxy::store::Store::open(std::path::Path::new(&config.store_path))?);
+    let mut fleet = Vec::new();
+    for entry in fleet::entries(std::path::Path::new(&config.printers_path))? {
+        fleet.push(fleet::build(entry)?);
+    }
+    for entry in store.printers()? {
+        if fleet.iter().any(|printer| printer.base().id == entry.id) {
+            tracing::warn!(printer = %entry.id, "in both printers.toml and the local store; using printers.toml");
+            continue;
+        }
+        // Skipped, not fatal: one entry the build no longer accepts should
+        // not keep the rest of a farm offline.
+        match fleet::build(entry) {
+            Ok(printer) => fleet.push(printer),
+            Err(err) => tracing::error!(%err, "a printer in the local store could not be built; skipping it"),
+        }
+    }
+    if fleet.is_empty() {
+        tracing::warn!("no printers yet - add one from the console, or list some in printers.toml");
+    }
     connect_fleet(&mut fleet).await;
 
     // The fleet is now shared with the uplink rather than just parked here
@@ -134,10 +157,11 @@ async fn main() -> anyhow::Result<()> {
         _ = shutdown_requested() => {
             info!("stop requested; exiting so go2rtc is stopped with us");
         }
-        _ = uplink::run(config.clone(), fleet.clone(), outbound.clone(), inbox) => {}
+        _ = uplink::run(config.clone(), fleet.clone(), Some(store.clone()), outbound.clone(), inbox) => {}
         _ = continuum_proxy::discovery::report::run(
             config.discovery.clone(),
             config.printers_path.clone(),
+            Some(store.clone()),
             outbound.clone(),
         ) => {}
         _ = go2rtc::watchdog(

src/printer/bambu.rs

@@ -207,6 +207,28 @@ pub enum BambuProtocolPreference {
 }
 // endregion: BambuMQTTProtocol
 
+/// One pending camera read for a P1/A1; see `bambu_camera`. Holds the access
+/// code, so it has no `Debug`.
+pub struct SnapshotJob {
+    name: String,
+    host: String,
+    access_code: String,
+    connector: native_tls::TlsConnector,
+}
+
+impl SnapshotJob {
+    /// One JPEG frame from the printer's port-6000 camera.
+    pub async fn run(self) -> Result<Vec<u8>, PrinterError> {
+        let name = self.name;
+        tokio::task::spawn_blocking(move || {
+            super::bambu_camera::snapshot(&self.host, &self.access_code, self.connector, std::time::Duration::from_secs(8))
+        })
+        .await
+        .map_err(|err| PrinterError(format!("{name}: camera thread failed: {err}")))?
+        .map_err(|err| PrinterError(format!("{name}: camera: {err}")))
+    }
+}
+
 /// 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
@@ -934,6 +956,21 @@ impl BambuPrinter {
         .map_err(|err| PrinterError(format!("{}: FTPS download failed: {err}", self.base.name)))
     }
 
+    /// Everything needed to fetch one camera frame, owned, so the caller can
+    /// drop the fleet lock before the network round trip.
+    pub fn snapshot_job(&self) -> Result<SnapshotJob, PrinterError> {
+        let connector = self
+            .tls
+            .build_connector()
+            .map_err(|err| PrinterError(format!("{}: TLS setup failed: {err}", self.base.name)))?;
+        Ok(SnapshotJob {
+            name: self.base.name.clone(),
+            host: self.base.host.clone(),
+            access_code: self.access_code.clone(),
+            connector,
+        })
+    }
+
     /// Removes one file from the printer's SD card.
     ///
     /// Takes one explicit path, because the convenient version of this is

src/printer/bambu_camera.rs

@@ -0,0 +1,134 @@
+//! The camera on a Bambu P1 or A1, which is not RTSP.
+//!
+//! Those models publish no `rtsp_url` and refuse port 322 (confirmed on a
+//! real P1P, 2026-10-15). What they have instead is a proprietary stream on
+//! TCP 6000: TLS, a single 80-byte login packet, then a continuous run of
+//! JPEG frames, each behind a 16-byte header.
+//!
+//! This reads **one frame** and hangs up. A browser cannot play that stream
+//! (it is MJPEG over a custom framing, and go2rtc cannot turn it into WebRTC
+//! without ffmpeg, which a gateway does not have), so the console shows a
+//! refreshing picture instead and a one-shot read is all it needs.
+//!
+//! Same blocking-TLS shape as `bambu_ftps`: the caller hands in the printer's
+//! own `native_tls` connector, so every trust decision stays in
+//! `BambuTls::build_connector`.
+
+use std::io::{Read, Write};
+use std::net::ToSocketAddrs;
+use std::time::Duration;
+
+pub const PORT: u16 = 6000;
+
+/// Same username as MQTT and FTPS.
+const USER: &str = "bblp";
+
+/// A frame this large is a corrupt header, not a picture. Real frames from a
+/// P1P are tens of kilobytes.
+const MAX_FRAME: usize = 4 * 1024 * 1024;
+
+/// The 80-byte login: four little-endian u32s (0x40, 0x3000, 0, 0), then the
+/// username and the access code, each zero-padded to 32 bytes.
+fn auth_packet(access_code: &str) -> [u8; 80] {
+    let mut packet = [0u8; 80];
+    packet[0..4].copy_from_slice(&0x40u32.to_le_bytes());
+    packet[4..8].copy_from_slice(&0x3000u32.to_le_bytes());
+    let user = USER.as_bytes();
+    packet[16..16 + user.len()].copy_from_slice(user);
+    // An access code is 8 characters; truncating anything longer keeps the
+    // packet shape rather than overrunning into nothing.
+    let code = access_code.as_bytes();
+    let n = code.len().min(32);
+    packet[48..48 + n].copy_from_slice(&code[..n]);
+    packet
+}
+
+/// Reads one frame: a 16-byte header whose first u32 is the payload length,
+/// then that many bytes, which must be a JPEG.
+fn read_frame<R: Read>(stream: &mut R) -> anyhow::Result<Vec<u8>> {
+    let mut header = [0u8; 16];
+    stream
+        .read_exact(&mut header)
+        .map_err(|err| anyhow::anyhow!("the camera sent no frame (wrong access code, or the stream is in use?): {err}"))?;
+    let len = u32::from_le_bytes([header[0], header[1], header[2], header[3]]) as usize;
+    if !(4..=MAX_FRAME).contains(&len) {
+        anyhow::bail!("the camera announced a {len}-byte frame, which is not a picture");
+    }
+    let mut jpeg = vec![0u8; len];
+    stream.read_exact(&mut jpeg)?;
+    if jpeg[..2] != [0xFF, 0xD8] || jpeg[len - 2..] != [0xFF, 0xD9] {
+        anyhow::bail!("the camera's frame is not a complete JPEG");
+    }
+    Ok(jpeg)
+}
+
+/// Connects, logs in and returns the first frame. Blocking; call it from
+/// `spawn_blocking`.
+pub fn snapshot(
+    host: &str,
+    access_code: &str,
+    connector: native_tls::TlsConnector,
+    timeout: Duration,
+) -> anyhow::Result<Vec<u8>> {
+    let addr = (host, PORT)
+        .to_socket_addrs()?
+        .next()
+        .ok_or_else(|| anyhow::anyhow!("{host} did not resolve"))?;
+    let tcp = std::net::TcpStream::connect_timeout(&addr, timeout)
+        .map_err(|err| anyhow::anyhow!("{host}:{PORT} did not accept a connection: {err}"))?;
+    tcp.set_read_timeout(Some(timeout))?;
+    tcp.set_write_timeout(Some(timeout))?;
+    let mut tls = connector
+        .connect(host, tcp)
+        .map_err(|err| anyhow::anyhow!("TLS handshake with the camera failed: {err}"))?;
+    tls.write_all(&auth_packet(access_code))?;
+    tls.flush()?;
+    read_frame(&mut tls)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn frame(payload: &[u8]) -> Vec<u8> {
+        let mut out = (payload.len() as u32).to_le_bytes().to_vec();
+        out.extend_from_slice(&[0; 12]);
+        out.extend_from_slice(payload);
+        out
+    }
+
+    #[test]
+    fn the_login_packet_has_the_documented_layout() {
+        let p = auth_packet("12345678");
+        assert_eq!(&p[0..8], &[0x40, 0, 0, 0, 0, 0x30, 0, 0]);
+        assert_eq!(&p[8..16], &[0; 8]);
+        assert_eq!(&p[16..20], b"bblp");
+        assert!(p[20..48].iter().all(|b| *b == 0));
+        assert_eq!(&p[48..56], b"12345678");
+        assert!(p[56..].iter().all(|b| *b == 0));
+    }
+
+    #[test]
+    fn a_whole_jpeg_frame_is_returned() {
+        let jpeg = [0xFF, 0xD8, 1, 2, 3, 0xFF, 0xD9];
+        let mut input = frame(&jpeg).into_iter().chain([9, 9, 9]).collect::<Vec<_>>();
+        assert_eq!(read_frame(&mut input.as_slice()).unwrap(), jpeg);
+        input.clear();
+    }
+
+    #[test]
+    fn nonsense_is_refused_not_returned() {
+        // not a JPEG
+        assert!(read_frame(&mut frame(&[1, 2, 3, 4]).as_slice()).is_err());
+        // length beyond the cap
+        let mut huge = (u32::MAX).to_le_bytes().to_vec();
+        huge.extend_from_slice(&[0; 12]);
+        assert!(read_frame(&mut huge.as_slice()).is_err());
+        // truncated
+        let mut cut = frame(&[0xFF, 0xD8, 0, 0xFF, 0xD9]);
+        cut.truncate(cut.len() - 2);
+        assert!(read_frame(&mut cut.as_slice()).is_err());
+        // nothing at all (what a wrong access code looks like)
+        assert!(read_frame(&mut [].as_slice()).is_err());
+    }
+}

src/printer/mod.rs

@@ -28,6 +28,7 @@
 //! stubs; that's the next piece to build past this module.
 
 mod bambu;
+mod bambu_camera;
 mod bambu_commands;
 mod bambu_ftps;
 mod bambu_mqtt;
@@ -43,7 +44,7 @@ pub use project_file::{
     ProjectFileError,
 };
 
-pub use bambu::{
+pub use bambu::{SnapshotJob, 
     BambuPrinter, BambuProtocolPreference, BambuProtocolVersion, BambuTls, FilamentSource,
     LightState, CHAMBER_LIGHT,
 };
@@ -194,6 +195,15 @@ pub enum PrinterHandle {
 }
 
 impl PrinterHandle {
+    /// A pending camera-frame read, for printers whose camera is a JPEG
+    /// stream rather than RTSP (Bambu P1/A1).
+    pub fn snapshot_job(&self) -> Result<bambu::SnapshotJob, PrinterError> {
+        match self {
+            PrinterHandle::Bambu(p) => p.snapshot_job(),
+            _ => Err(PrinterError("this printer has no snapshot camera".into())),
+        }
+    }
+
     /// The latest status this process has for the printer, if it has one.
     ///
     /// An inherent method rather than a `GenericPrinter` trait method,

src/provision.rs

@@ -0,0 +1,244 @@
+//! Adding and removing printers on a running gateway, on the console's say-so.
+//!
+//! The access code arrives in a `provision_printer` message, is checked by
+//! actually connecting, and is then written to the local store (src/store.rs).
+//! It is never logged and never sent anywhere. The cloud does not keep it -
+//! it relayed it - so this is the only place it exists.
+//!
+//! Connect first, store second: a printer that cannot be reached with the
+//! code the operator typed is a typo to report now, not a row to discover
+//! later in a fleet that quietly never comes up.
+
+use std::time::Duration;
+
+use tracing::info;
+
+use crate::fleet;
+use crate::printer::GenericPrinter;
+use crate::store::Store;
+use crate::uplink::SharedFleet;
+use crate::uplink::protocol::ProvisionPrinter;
+
+/// Bambu's protocol auto-detection can spend several seconds on a first
+/// attempt, and an operator is watching a spinner.
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Ids end up in MQTT topics, file names (pinned certificates) and log
+/// lines, so they are held to a small safe alphabet.
+fn valid_id(id: &str) -> bool {
+    !id.is_empty() && id.len() <= 64 && id.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
+}
+
+pub async fn add(fleet: &SharedFleet, store: &Store, request: ProvisionPrinter) -> Result<(), String> {
+    if !valid_id(&request.id) {
+        return Err("printer id must be 1-64 letters, digits, '-' or '_'".into());
+    }
+    if fleet.lock().await.iter().any(|p| p.base().id == request.id) {
+        return Err(format!("this gateway already has a printer with id '{}'", request.id));
+    }
+
+    let entry = fleet::PrinterEntry {
+        id: request.id,
+        name: request.name,
+        vendor: request.vendor,
+        host: request.host,
+        access_code: request.access_code,
+        sn: request.sn,
+        api_key: request.api_key,
+        port: None,
+        baud_rate: 115_200,
+        ca_cert_path: None,
+        require_valid_cert: true,
+    };
+
+    // Built (which checks the required fields are present and non-blank) and
+    // connected before the fleet lock is taken: connecting can take seconds,
+    // and the lock is what telemetry waits on.
+    let mut printer = fleet::build(entry.clone()).map_err(|err| err.to_string())?;
+    match tokio::time::timeout(CONNECT_TIMEOUT, printer.connect()).await {
+        Ok(Ok(())) => {}
+        Ok(Err(err)) => return Err(format!("could not connect to the printer: {err}")),
+        Err(_) => return Err("the printer did not answer in time - check the address and that it is on".into()),
+    }
+
+    store.save_printer(&entry).map_err(|err| format!("connected, but could not save it locally: {err}"))?;
+    info!(printer = %entry.id, "printer added from the console");
+
+    let mut fleet = fleet.lock().await;
+    // Re-checked under the lock: two requests for one id can both pass the
+    // check above while the first is still connecting.
+    if fleet.iter().any(|p| p.base().id == entry.id) {
+        return Err(format!("this gateway already has a printer with id '{}'", entry.id));
+    }
+    fleet.push(printer);
+    Ok(())
+}
+
+pub async fn remove(fleet: &SharedFleet, store: &Store, printer_id: &str) -> Result<(), String> {
+    let stored = store.remove_printer(printer_id).map_err(|err| err.to_string())?;
+    let mut fleet = fleet.lock().await;
+    let before = fleet.len();
+    fleet.retain(|p| p.base().id != printer_id);
+    if !stored && fleet.len() == before {
+        return Err(format!("no printer with id '{printer_id}' on this gateway"));
+    }
+    info!(printer = %printer_id, "printer removed from the console");
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn ids_are_a_small_safe_alphabet() {
+        assert!(valid_id("p1p"));
+        assert!(valid_id("Printer_2-b"));
+        assert!(!valid_id(""));
+        assert!(!valid_id("../etc/passwd"));
+        assert!(!valid_id("has space"));
+        assert!(!valid_id(&"a".repeat(65)));
+    }
+}
+
+#[cfg(test)]
+mod behaviour {
+    use std::sync::Arc;
+
+    use super::*;
+
+    fn request(id: &str) -> ProvisionPrinter {
+        ProvisionPrinter {
+            id: id.into(),
+            name: "Bench P1".into(),
+            vendor: "bambu".into(),
+            // Nothing listens here, so a connect attempt is refused at once.
+            host: Some("127.0.0.1".into()),
+            access_code: Some("12345678".into()),
+            sn: Some("01P00A000000001".into()),
+            api_key: None,
+        }
+    }
+
+    fn store() -> (Store, std::path::PathBuf) {
+        let path = std::env::temp_dir().join(format!("continuum-provision-test-{}-{}.db", std::process::id(), rand_suffix()));
+        (Store::open(&path).unwrap(), path)
+    }
+
+    fn rand_suffix() -> u128 {
+        std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()
+    }
+
+    fn empty_fleet() -> SharedFleet {
+        Arc::new(tokio::sync::Mutex::new(Vec::new()))
+    }
+
+    #[tokio::test]
+    async fn a_bad_id_is_refused_before_anything_happens() {
+        let (store, path) = store();
+        let err = add(&empty_fleet(), &store, request("../x")).await.unwrap_err();
+        assert!(err.contains("printer id"), "{err}");
+        assert!(store.printers().unwrap().is_empty());
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn a_blank_credential_is_named_and_nothing_is_saved() {
+        let (store, path) = store();
+        let mut req = request("p1");
+        req.access_code = Some("  ".into());
+        let err = add(&empty_fleet(), &store, req).await.unwrap_err();
+        assert!(err.contains("access_code"), "{err}");
+        assert!(store.printers().unwrap().is_empty());
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn a_printer_that_cannot_be_reached_is_not_saved() {
+        let (store, path) = store();
+        let fleet = empty_fleet();
+        let err = add(&fleet, &store, request("p1")).await.unwrap_err();
+        assert!(err.contains("could not connect") || err.contains("did not answer"), "{err}");
+        assert!(store.printers().unwrap().is_empty(), "a failed add must not leave a stored printer");
+        assert!(fleet.lock().await.is_empty());
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn the_error_never_contains_the_access_code() {
+        let (store, path) = store();
+        let err = add(&empty_fleet(), &store, request("p1")).await.unwrap_err();
+        assert!(!err.contains("12345678"), "{err}");
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn removing_forgets_a_printer_in_both_places() {
+        let (store, path) = store();
+        let entry = crate::fleet::PrinterEntry {
+            id: "p1".into(),
+            name: "Bench".into(),
+            vendor: "bambu".into(),
+            host: Some("127.0.0.1".into()),
+            access_code: Some("12345678".into()),
+            sn: Some("01P00A000000001".into()),
+            api_key: None,
+            port: None,
+            baud_rate: 115_200,
+            ca_cert_path: None,
+            require_valid_cert: true,
+        };
+        store.save_printer(&entry).unwrap();
+        let fleet = empty_fleet();
+        fleet.lock().await.push(crate::fleet::build(entry).unwrap());
+
+        remove(&fleet, &store, "p1").await.unwrap();
+        assert!(fleet.lock().await.is_empty());
+        assert!(store.printers().unwrap().is_empty());
+
+        let err = remove(&fleet, &store, "p1").await.unwrap_err();
+        assert!(err.contains("no printer"), "{err}");
+        let _ = std::fs::remove_file(path);
+    }
+
+    /// The success path, using the OctoPrint client because its `connect` is
+    /// the one that succeeds without a real machine (the driver is a stub).
+    /// What this proves is the plumbing, not a vendor: a connected printer is
+    /// saved, joins the running fleet, and is still there after a "restart"
+    /// that reopens the store and rebuilds the fleet from it.
+    #[tokio::test]
+    async fn an_added_printer_joins_the_fleet_and_survives_a_restart() {
+        let (store, path) = store();
+        let fleet = empty_fleet();
+        let mut req = request("octo1");
+        req.vendor = "octoprint".into();
+        req.sn = None;
+        req.access_code = None;
+        req.api_key = Some("key-123".into());
+
+        add(&fleet, &store, req.clone()).await.unwrap();
+        assert!(fleet.lock().await.iter().any(|p| p.base().id == "octo1"));
+
+        // Same id again: refused, and the first is untouched.
+        let again = add(&fleet, &store, req).await.unwrap_err();
+        assert!(again.contains("already has"), "{again}");
+        assert_eq!(fleet.lock().await.len(), 1);
+
+        // "Restart": reopen the store, rebuild what main would.
+        drop(store);
+        let reopened = Store::open(&path).unwrap();
+        let rebuilt: Vec<_> = reopened.printers().unwrap().into_iter().map(|e| crate::fleet::build(e).unwrap()).collect();
+        assert_eq!(rebuilt.len(), 1);
+        assert_eq!(rebuilt[0].base().id, "octo1");
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[test]
+    fn the_wire_message_parses_and_its_debug_hides_the_credential() {
+        let json = r#"{"type":"provision_printer","request_id":"r1","printer":{"id":"p1","name":"Bench","vendor":"bambu","host":"10.0.0.5","access_code":"98765432","sn":"SN1"}}"#;
+        let message: crate::uplink::ServerMessage = serde_json::from_str(json).unwrap();
+        let shown = format!("{message:?}");
+        assert!(shown.contains("p1") && shown.contains("<redacted>"), "{shown}");
+        assert!(!shown.contains("98765432"), "the access code reached a Debug string: {shown}");
+    }
+}

src/store.rs

@@ -0,0 +1,199 @@
+//! The gateway's own database: printers an operator added from the console.
+//!
+//! Why the gateway keeps these itself and not the cloud: a printer's access
+//! code is a credential for a machine on someone's LAN, and the platform's
+//! rule is that those never rest in the cloud. The console sends the code
+//! once, over the uplink, and this is where it lands. It also means the
+//! fleet survives the API being unreachable - the gateway starts, connects
+//! its printers and keeps reporting locally from this alone.
+//!
+//! SQLite, one file. The file holds credentials, so on Unix it is created
+//! owner-only (0600) before SQLite touches it; SQLite gives its journal files
+//! the same mode as the main one.
+//!
+//! `printers.toml` still works and is read alongside this; see `main.rs`.
+
+use std::path::Path;
+use std::sync::Mutex;
+
+use anyhow::Context;
+use rusqlite::{Connection, params};
+
+use crate::fleet::PrinterEntry;
+
+pub struct Store {
+    // rusqlite's `Connection` is not `Sync`, so it cannot be shared between
+    // tasks directly. Wrapping it in a `Mutex` is the plain fix: one caller
+    // at a time, which is plenty for a few writes a day.
+    conn: Mutex<Connection>,
+}
+
+impl Store {
+    pub fn open(path: &Path) -> anyhow::Result<Self> {
+        if let Some(dir) = path.parent().filter(|dir| !dir.as_os_str().is_empty()) {
+            std::fs::create_dir_all(dir).with_context(|| format!("creating {dir:?} for the local store"))?;
+        }
+        create_private(path)?;
+        let conn = Connection::open(path).with_context(|| format!("opening the local store at {path:?}"))?;
+        // Each printer is a JSON document keyed by its id. The id is the
+        // gateway's own printer id - the one telemetry is reported under -
+        // so "which printer" has a single answer everywhere.
+        conn.execute_batch(
+            "CREATE TABLE IF NOT EXISTS printers (
+                id   TEXT PRIMARY KEY,
+                json TEXT NOT NULL
+            );",
+        )?;
+        Ok(Self { conn: Mutex::new(conn) })
+    }
+
+    /// Adds a printer, or replaces the one with the same id.
+    pub fn save_printer(&self, entry: &PrinterEntry) -> anyhow::Result<()> {
+        let json = serde_json::to_string(entry)?;
+        let conn = self.conn.lock().expect("store lock poisoned");
+        conn.execute(
+            "INSERT INTO printers (id, json) VALUES (?1, ?2)
+             ON CONFLICT(id) DO UPDATE SET json = excluded.json",
+            params![entry.id, json],
+        )?;
+        Ok(())
+    }
+
+    /// Forgets a printer. True if there was one.
+    pub fn remove_printer(&self, id: &str) -> anyhow::Result<bool> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        Ok(conn.execute("DELETE FROM printers WHERE id = ?1", params![id])? > 0)
+    }
+
+    /// Every stored printer, oldest first.
+    pub fn printers(&self) -> anyhow::Result<Vec<PrinterEntry>> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        let mut statement = conn.prepare("SELECT json FROM printers ORDER BY rowid")?;
+        let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
+        let mut out = Vec::new();
+        for json in rows {
+            out.push(serde_json::from_str(&json?).context("a stored printer no longer parses")?);
+        }
+        Ok(out)
+    }
+}
+
+/// Creates the file empty and owner-only if it does not exist yet.
+#[cfg(unix)]
+fn create_private(path: &Path) -> anyhow::Result<()> {
+    use std::os::unix::fs::OpenOptionsExt;
+    std::fs::OpenOptions::new()
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .mode(0o600)
+        .open(path)
+        .with_context(|| format!("creating the local store {path:?}"))?;
+    Ok(())
+}
+
+/// Windows: the file inherits its folder's permissions, which for the
+/// per-user data directory is already that user's. Nothing to set here.
+#[cfg(not(unix))]
+fn create_private(_path: &Path) -> anyhow::Result<()> {
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn entry(id: &str, access_code: &str) -> PrinterEntry {
+        PrinterEntry {
+            id: id.into(),
+            name: format!("Printer {id}"),
+            vendor: "bambu".into(),
+            host: Some("192.168.0.50".into()),
+            access_code: Some(access_code.into()),
+            sn: Some("01P00A000000000".into()),
+            api_key: None,
+            port: None,
+            baud_rate: 115_200,
+            ca_cert_path: None,
+            require_valid_cert: true,
+        }
+    }
+
+    /// A throwaway directory, removed when dropped. Hand-rolled rather than
+    /// pulling in a crate for four tests.
+    struct TempDir(std::path::PathBuf);
+
+    impl Drop for TempDir {
+        fn drop(&mut self) {
+            let _ = std::fs::remove_dir_all(&self.0);
+        }
+    }
+
+    fn temp_db() -> (TempDir, std::path::PathBuf) {
+        use std::sync::atomic::{AtomicU32, Ordering};
+        static NEXT: AtomicU32 = AtomicU32::new(0);
+        let dir = std::env::temp_dir().join(format!(
+            "continuum-store-test-{}-{}",
+            std::process::id(),
+            NEXT.fetch_add(1, Ordering::Relaxed)
+        ));
+        let path = dir.join("nested").join("continuum.db");
+        (TempDir(dir), path)
+    }
+
+    #[test]
+    fn what_was_saved_comes_back_after_reopening() {
+        let (_dir, path) = temp_db();
+        Store::open(&path).unwrap().save_printer(&entry("a", "11111111")).unwrap();
+
+        let reopened = Store::open(&path).unwrap();
+        let printers = reopened.printers().unwrap();
+        assert_eq!(printers.len(), 1);
+        assert_eq!(printers[0].id, "a");
+        assert_eq!(printers[0].access_code.as_deref(), Some("11111111"));
+        assert_eq!(printers[0].sn.as_deref(), Some("01P00A000000000"));
+    }
+
+    #[test]
+    fn saving_the_same_id_replaces_it_and_order_is_oldest_first() {
+        let (_dir, path) = temp_db();
+        let store = Store::open(&path).unwrap();
+        store.save_printer(&entry("a", "old")).unwrap();
+        store.save_printer(&entry("b", "bbb")).unwrap();
+        store.save_printer(&entry("a", "new")).unwrap();
+
+        let printers = store.printers().unwrap();
+        let ids: Vec<_> = printers.iter().map(|p| p.id.as_str()).collect();
+        assert_eq!(ids, ["a", "b"]);
+        assert_eq!(printers[0].access_code.as_deref(), Some("new"));
+    }
+
+    #[test]
+    fn removing_reports_whether_anything_was_there() {
+        let (_dir, path) = temp_db();
+        let store = Store::open(&path).unwrap();
+        store.save_printer(&entry("a", "x")).unwrap();
+        assert!(store.remove_printer("a").unwrap());
+        assert!(!store.remove_printer("a").unwrap());
+        assert!(store.printers().unwrap().is_empty());
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn the_file_holding_credentials_is_owner_only() {
+        use std::os::unix::fs::PermissionsExt;
+        let (_dir, path) = temp_db();
+        let store = Store::open(&path).unwrap();
+        store.save_printer(&entry("a", "secret")).unwrap();
+        let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
+        assert_eq!(mode, 0o600);
+    }
+
+    #[test]
+    fn a_corrupt_file_is_an_error_not_an_empty_fleet() {
+        let (_dir, path) = temp_db();
+        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
+        std::fs::write(&path, b"this is not a sqlite database at all, just text").unwrap();
+        assert!(Store::open(&path).is_err());
+    }
+}

src/uplink/auth.rs

@@ -0,0 +1,107 @@
+//! Getting from the gateway's long-lived credential to the short-lived one
+//! the socket wants.
+//!
+//! The cloud hands a farm's owner an *edge proxy token*, which is what goes in
+//! this gateway's settings as `gateway_token`. It does not open the socket
+//! directly: the upgrade wants an *edge JWT*, which the cloud signs for an
+//! hour in exchange for the proxy token (`POST /auth/edge-token`). So every
+//! connection attempt starts by trading, which is also why a gateway that
+//! has been up for days is not locked out when the first JWT expires - each
+//! reconnect gets a fresh one.
+
+use std::time::Duration;
+
+use anyhow::Context;
+use serde::Deserialize;
+
+use crate::config::Config;
+
+const EXCHANGE_TIMEOUT: Duration = Duration::from_secs(15);
+const SOCKET_PATH: &str = "/ws/edge/v1";
+
+/// Where to trade the token, derived from the socket URL so there is one
+/// address to configure, not two that can disagree.
+pub fn exchange_url(uplink_url: &str) -> anyhow::Result<String> {
+    let (scheme, rest) = if let Some(rest) = uplink_url.strip_prefix("wss://") {
+        ("https://", rest)
+    } else if let Some(rest) = uplink_url.strip_prefix("ws://") {
+        ("http://", rest)
+    } else {
+        anyhow::bail!("the uplink URL must start with ws:// or wss://: {uplink_url}");
+    };
+    let base = rest
+        .strip_suffix(SOCKET_PATH)
+        .with_context(|| format!("the uplink URL must end with {SOCKET_PATH}: {uplink_url}"))?;
+    Ok(format!("{scheme}{base}/auth/edge-token"))
+}
+
+#[derive(Deserialize)]
+struct Exchanged {
+    token: String,
+}
+
+/// Trades `gateway_token` for an edge JWT.
+///
+/// A 401 is reported as what it means - the cloud does not accept this
+/// farm/token pair - because retrying will not fix it, and the usual cause is
+/// a token that was rotated in the console while this gateway kept the old
+/// one.
+pub async fn fetch_edge_token(config: &Config) -> anyhow::Result<String> {
+    let url = exchange_url(&config.uplink_url)?;
+    let response = reqwest::Client::builder()
+        .timeout(EXCHANGE_TIMEOUT)
+        .build()?
+        .post(&url)
+        // Built by hand: the `json` feature of reqwest is off on purpose (see
+        // Cargo.toml), and this is one small body.
+        .header("Content-Type", "application/json")
+        .body(
+            serde_json::json!({
+                "farmId": config.farm_id,
+                "edgeProxyToken": config.gateway_token,
+            })
+            .to_string(),
+        )
+        .send()
+        .await
+        .with_context(|| format!("could not reach {url} to get an edge token"))?;
+
+    match response.status() {
+        status if status.is_success() => {
+            let body = response.bytes().await?;
+            Ok(serde_json::from_slice::<Exchanged>(&body).context("the cloud's edge token reply was not understood")?.token)
+        }
+        reqwest::StatusCode::UNAUTHORIZED => {
+            anyhow::bail!("the cloud does not accept this farm_id / gateway_token - was the token rotated in the console?")
+        }
+        status => anyhow::bail!("the cloud answered {status} when asked for an edge token"),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn the_exchange_address_follows_the_socket_address() {
+        assert_eq!(
+            exchange_url("wss://api.continuum.local/ws/edge/v1").unwrap(),
+            "https://api.continuum.local/auth/edge-token"
+        );
+        assert_eq!(
+            exchange_url("ws://localhost:3040/ws/edge/v1").unwrap(),
+            "http://localhost:3040/auth/edge-token"
+        );
+        // A prefix in front of the API survives.
+        assert_eq!(
+            exchange_url("wss://example.com/api/ws/edge/v1").unwrap(),
+            "https://example.com/api/auth/edge-token"
+        );
+    }
+
+    #[test]
+    fn an_address_it_cannot_follow_is_an_error_not_a_guess() {
+        assert!(exchange_url("https://api/ws/edge/v1").is_err());
+        assert!(exchange_url("wss://api/somewhere/else").is_err());
+    }
+}

src/uplink/mod.rs

@@ -26,9 +26,10 @@
 //! readings for no consumer. See PLAN.md for what would actually be worth
 //! buffering.
 
+mod auth;
 mod download;
 mod webrtc;
-mod protocol;
+pub mod protocol;
 
 pub use protocol::{ClientMessage, DiscoveredPrinter, PrinterCommand, PrinterTelemetry, ServerMessage};
 
@@ -42,6 +43,9 @@ 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
@@ -139,6 +143,7 @@ pub fn channel() -> (
 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>,
 ) {
@@ -156,7 +161,7 @@ pub async fn run(
                 // 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, &outbound, &mut inbox).await {
+                if let Err(err) = serve(ws_stream, &config, &fleet, store.as_ref(), &outbound, &mut inbox).await {
                     warn!(?err, "uplink connection dropped, reconnecting in {delay:?}");
                 }
             }
@@ -172,6 +177,10 @@ pub async fn run(
 /// 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 -
@@ -180,7 +189,7 @@ async fn connect(config: &Config) -> anyhow::Result<WsStream> {
     // those tools already know to redact.
     let request = Request::builder()
         .uri(&config.uplink_url)
-        .header("Authorization", format!("Bearer {}", config.gateway_token))
+        .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.
@@ -200,6 +209,7 @@ 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<()> {
@@ -306,13 +316,31 @@ async fn serve(
                                     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.
-                            Err(err) => warn!(%err, payload = %text, "unparseable message from cloud, ignoring"),
+                            // 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(()),
@@ -379,6 +407,12 @@ async fn collect_telemetry(fleet: &SharedFleet) -> Vec<PrinterTelemetry> {
                 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),
         });
     }
 
@@ -456,6 +490,54 @@ fn spawn_long_command(
     });
 }
 
+/// 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
@@ -485,6 +567,48 @@ fn spawn_webrtc_negotiation(
     });
 }
 
+/// 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.

src/uplink/protocol.rs

@@ -56,6 +56,16 @@ pub enum ClientMessage {
     /// waiting on this and a viewer otherwise watches a video element that
     /// never loads with nothing anywhere saying why.
     WebrtcError { session_id: String, message: String },
+    /// The answer to a `CameraSnapshotRequest`: one JPEG frame, base64, or
+    /// the reason there is none. Always sent, success or failure - the cloud
+    /// is holding a viewer's request open for it.
+    CameraSnapshot {
+        request_id: String,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        jpeg_base64: Option<String>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        error: Option<String>,
+    },
     /// Printers found on this gateway's own network and USB ports that are
     /// *not* in its fleet file, so the console can offer them instead of
     /// making an operator walk to a machine and read its IP address off the
@@ -191,6 +201,11 @@ pub struct PrinterTelemetry {
     /// `crate::camera`.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub camera_stream: Option<String>,
+    /// True when the camera can only be read as single JPEG frames (a P1/A1)
+    /// and this gateway will answer a `CameraSnapshotRequest` for it. See
+    /// `crate::camera::wants_snapshot`.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub camera_snapshot: Option<bool>,
 }
 
 /// What `continuum-backend` sends back over the same connection.
@@ -225,6 +240,48 @@ pub enum ServerMessage {
     /// and checks its password before CORS, so the unauthenticated preflight
     /// a browser always sends is refused - measured, not assumed.
     WebrtcOffer { session_id: String, camera_stream: String, sdp: String },
+    /// Read one JPEG frame from a P1/A1-style camera and answer with a
+    /// `CameraSnapshot`. `printer_id` is our own id.
+    CameraSnapshotRequest { request_id: String, printer_id: String },
+    /// Add a printer to this gateway. Carries the printer's credential, sent
+    /// once by the console and kept only here - see `provision`. Answered
+    /// with a `CommandResult` carrying the same id.
+    ProvisionPrinter { request_id: String, printer: ProvisionPrinter },
+    /// Forget a printer. Answered like `ProvisionPrinter`.
+    RemovePrinter { request_id: String, printer_id: String },
+}
+
+/// What the console may tell a gateway about a printer to add.
+///
+/// Deliberately narrower than `fleet::PrinterEntry`: no `ca_cert_path` and no
+/// serial `port`, because both name something on the gateway's own disk or
+/// USB bus, and the cloud has no business choosing those.
+///
+/// `Debug` is by hand so the credential cannot reach a log line - the
+/// `Ok(other) => info!(?other, ...)` arm in the uplink formats whole messages.
+#[derive(Clone, Deserialize)]
+pub struct ProvisionPrinter {
+    pub id: String,
+    pub name: String,
+    pub vendor: String,
+    pub host: Option<String>,
+    pub access_code: Option<String>,
+    pub sn: Option<String>,
+    pub api_key: Option<String>,
+}
+
+impl std::fmt::Debug for ProvisionPrinter {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ProvisionPrinter")
+            .field("id", &self.id)
+            .field("name", &self.name)
+            .field("vendor", &self.vendor)
+            .field("host", &self.host)
+            .field("access_code", &self.access_code.as_ref().map(|_| "<redacted>"))
+            .field("sn", &self.sn)
+            .field("api_key", &self.api_key.as_ref().map(|_| "<redacted>"))
+            .finish()
+    }
 }
 
 /// The closed set of things the cloud is allowed to ask a printer to do.

tests/uplink_late_result.rs

@@ -18,10 +18,12 @@ use std::time::Duration;
 
 use futures_util::{SinkExt, StreamExt};
 use serde_json::{json, Value};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpListener;
 use tokio::sync::Mutex;
 use tokio_tungstenite::WebSocketStream;
 use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};
 
 use continuum_proxy::config::Config;
 use continuum_proxy::uplink;
@@ -44,6 +46,7 @@ fn config(port: u16) -> Config {
         go2rtc_api_listen: "127.0.0.1:1984".into(),
         stream_base_url: None,
         printers_path: "/nonexistent/printers.toml".into(),
+        store_path: "/nonexistent/continuum.db".into(),
         // Off, which is the default and also the only safe thing in a test:
         // an enabled one would sweep whatever network the test host is on.
         discovery: Default::default(),
@@ -69,6 +72,36 @@ where
     }
 }
 
+/// Plays the cloud's `/auth/edge-token`: reads one HTTP request, checks that
+/// the gateway sent its farm and long-lived token, and answers with a JWT.
+/// Returns the request text so the caller can assert on it.
+async fn answer_token_exchange(listener: &TcpListener) -> String {
+    let (mut stream, _) = listener.accept().await.unwrap();
+    let mut request = Vec::new();
+    let mut buf = [0u8; 4096];
+    loop {
+        let n = tokio::time::timeout(WAIT, stream.read(&mut buf)).await.unwrap().unwrap();
+        assert!(n > 0, "the gateway hung up before finishing its token request");
+        request.extend_from_slice(&buf[..n]);
+        let text = String::from_utf8_lossy(&request).into_owned();
+        if let Some((head, body)) = text.split_once("\r\n\r\n") {
+            let length = head
+                .lines()
+                .find_map(|l| l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap()))
+                .unwrap_or(0);
+            if body.len() >= length {
+                break;
+            }
+        }
+    }
+    let body = r#"{"token":"minted-edge-jwt","expiresInSeconds":3600}"#;
+    stream
+        .write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes())
+        .await
+        .unwrap();
+    String::from_utf8_lossy(&request).into_owned()
+}
+
 #[tokio::test]
 async fn a_long_command_that_fails_reports_its_reason_after_the_fact() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -76,10 +109,23 @@ async fn a_long_command_that_fails_reports_its_reason_after_the_fact() {
 
     let fleet: uplink::SharedFleet = Arc::new(Mutex::new(Vec::new()));
     let (outbound, inbox) = uplink::channel();
-    tokio::spawn(uplink::run(config(port), fleet, outbound, inbox));
+    tokio::spawn(uplink::run(config(port), fleet, None, outbound, inbox));
+
+    // The gateway trades its long-lived token for an edge JWT first, and the
+    // JWT - not the token it was configured with - is what opens the socket.
+    let exchange = answer_token_exchange(&listener).await;
+    assert!(exchange.starts_with("POST /auth/edge-token "), "{exchange}");
+    assert!(exchange.contains(r#""farmId":"test-farm""#) && exchange.contains(r#""edgeProxyToken":"test-token""#), "{exchange}");
 
     let (stream, _) = listener.accept().await.unwrap();
-    let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
+    let mut upgrade_auth = String::new();
+    let mut ws = tokio_tungstenite::accept_hdr_async(stream, |request: &Request, response: Response| {
+        upgrade_auth = request.headers().get("authorization").unwrap().to_str().unwrap().to_string();
+        Ok(response)
+    })
+    .await
+    .unwrap();
+    assert_eq!(upgrade_auth, "Bearer minted-edge-jwt", "the socket must be opened with the minted JWT");
 
     let hello = next_json(&mut ws).await;
     assert_eq!(hello["type"], "hello", "a gateway introduces itself first");

tests/uplink_protocol.rs

@@ -188,6 +188,7 @@ fn empty_telemetry_fields_are_omitted_not_null() {
             // The common case, and the third field that must be omitted: a
             // P1P reports that it has a camera without giving a URL for it.
             camera_stream: None,
+            camera_snapshot: None,
         }],
     };
 
@@ -226,6 +227,7 @@ fn a_zero_stage_is_sent_rather_than_treated_as_absent() {
             print_error: Some(0),
             active_alerts: Some(0),
             camera_stream: Some("p1p".into()),
+            camera_snapshot: None,
         }],
     };
 
@@ -299,3 +301,28 @@ fn a_hello_carries_no_credential() {
     assert!(!json.contains("token"), "the hello must carry no token, got:\n  {json}");
     assert_eq!(json, r#"{"type":"hello","gateway_id":"gw-1","farm_id":"farm-1"}"#);
 }
+
+#[test]
+fn a_snapshot_request_is_readable_and_its_answer_matches_the_schema() {
+    let request = r#"{"type":"camera_snapshot_request","request_id":"r1","printer_id":"p1p"}"#;
+    assert!(matches!(
+        serde_json::from_str::<ServerMessage>(request),
+        Ok(ServerMessage::CameraSnapshotRequest { .. })
+    ));
+
+    let ok = serde_json::to_string(&ClientMessage::CameraSnapshot {
+        request_id: "r1".into(),
+        jpeg_base64: Some("/9j/".into()),
+        error: None,
+    })
+    .unwrap();
+    assert_eq!(ok, r#"{"type":"camera_snapshot","request_id":"r1","jpeg_base64":"/9j/"}"#);
+
+    let failed = serde_json::to_string(&ClientMessage::CameraSnapshot {
+        request_id: "r1".into(),
+        jpeg_base64: None,
+        error: Some("no frame".into()),
+    })
+    .unwrap();
+    assert_eq!(failed, r#"{"type":"camera_snapshot","request_id":"r1","error":"no frame"}"#);
+}