Simplify boilerplate for learning: drop real protocol code, minimal main.rs

Removed adapters/ (rumqttc MQTT, suppaftp FTPS, reqwest HTTP), plate_changer/ (tokio-serial), discovery/, and cache.rs (rusqlite) — src/printer/ already covers 'talk to a printer' as simple stubs, so keeping a second, more complex version alongside it was redundant. main.rs goes from 6 concurrent tasks to 2 (uplink + go2rtc). uplink/ drops channels, jitter, and Send-bound futures. Verified with cargo build + cargo run --example printer_polymorphism.
2026-08-28 18:22:38 +00:00
parent 5705f36c01
commit d0a17ee50e
16 changed files with 80 additions and 799 deletions
@@ -5,25 +5,9 @@ CONTINUUM_GATEWAY_TOKEN=replace-with-provisioning-token
 # Cloud uplink
 CONTINUUM_UPLINK_URL=wss://api.continuum.local/ws/edge/v1
 CONTINUUM_API_BASE=https://api.continuum.local
 # Local state
 CONTINUUM_SQLITE_PATH=./data/edge-cache.db
 # go2rtc process management
 CONTINUUM_GO2RTC_BIN=/usr/local/bin/go2rtc
 CONTINUUM_GO2RTC_CONFIG=./go2rtc.yaml
 CONTINUUM_GO2RTC_API_PORT=1984
 # Bambu Lab printers (LAN mode) — comma-separated "serial:ip:access_code" triples,
 # normally discovered via SSDP but overridable for static configs.
 CONTINUUM_BAMBU_STATIC_PRINTERS=
 # PrusaLink / Moonraker discovery
 CONTINUUM_DISCOVERY_INTERVAL_SECS=30
 # plate changer hardware
 CONTINUUM_PLATE_CHANGER_SERIAL_PORT=/dev/ttyUSB0
 CONTINUUM_PLATE_CHANGER_BAUD=115200
 RUST_LOG=info,continuum_proxy=debug
@@ -20,30 +20,15 @@ path = "examples/printer_polymorphism.rs"
 [dependencies]
 tokio = { version = "1.40", features = ["full"] }
 tokio-tungstenite = { version = "0.24", features = ["native-tls"] }
 tokio-util = { version = "0.7", features = ["codec", "compat"] }
 futures-util = "0.3"
 rumqttc = "0.24"
 reqwest = { version = "0.12", features = ["stream", "json"] }
 suppaftp = { version = "6", features = ["async-native-tls"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 rusqlite = { version = "0.32", features = ["bundled"] }
 tokio-serial = "5.4"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
 anyhow = "1"
 thiserror = "1"
 uuid = { version = "1", features = ["v4", "serde"] }
 rand = "0.8"
 dotenvy = "0.15"
-# continuum-common's rust-core crate (error types, tracing init, G-code helpers)
+# Real printer protocol clients (MQTT for Bambu, HTTP for PrusaLink/
-# lives in the sibling `continuum-common` repo. Once that crate is published to
+# Moonraker, FTPS for gcode transfer) go here once you're ready to build
-# a registry or exposed over git, add it here, e.g.:
+# past the stubs in src/printer/ — see that module's doc comment.
 #   continuum-common = { git = "https://git.octoturge.com/Continuum/continuum-common.git", package = "continuum-core" }
 [profile.release]
 opt-level = "z"
 lto = true
 codegen-units = 1
 panic = "abort"
@@ -1,51 +1,52 @@
 # continuum-proxy
-Edge gateway daemon for the Continuum print farm platform. Runs on a Linux SBC
+Edge gateway daemon for the Continuum print farm platform. Runs on a Linux
-on-site, discovers and talks to printers over the LAN, and maintains a
+SBC on-site, talks to printers over the LAN, and keeps a connection to the
-resilient uplink to the cloud control plane (`continuum-backend`).
+cloud control plane (`continuum-backend`).
-## Responsibilities
+## This is a learning-stage boilerplate
- **Uplink** (`src/uplink/`): persistent WebSocket connection to
+This repo is deliberately minimal right now — real printer protocol clients
-  `wss://api.domain.com/ws/edge/v1` with exponential-backoff reconnects,
+(Bambu MQTT+FTPS, PrusaLink REST, Klipper/Moonraker WebSocket) are **not**
-  heartbeat ping/pong, and inbound task dispatch.
+implemented yet. Each vendor is a stub that just prints what it would do
- **Printer adapters** (`src/adapters/`): protocol clients per printer vendor —
+(`src/printer/bambu.rs`, `prusa.rs`, `klipper.rs`). The idea is to learn
-  Bambu Lab (MQTT 8883 + FTPS 990), PrusaLink (REST), Klipper (Moonraker WS).
+Rust's polymorphism pattern (trait + enum, since Rust has no class
- **Plate changer** (`src/plate_changer/`): mechanical cycle execution and
+inheritance) on something simple before adding real networking on top.
  sensor validation over GPIO/serial for automated plate-swap hardware.
 - **Local cache** (`rusqlite`): buffers telemetry and job state through
  connectivity gaps so nothing is lost if the uplink drops.
 - **go2rtc watchdog**: supervises the bundled `go2rtc` process for camera
  restreaming (RTSP/USB → WebRTC).
 ## Getting started
 ```bash
 cp .env.example .env
-cargo run
+cargo run                                  # the daemon: cloud uplink + go2rtc watchdog
 cargo run --example printer_polymorphism   # standalone demo, no network/env needed
 ```
 ## Structure
 ```
 src/
-  main.rs            Multi-threaded async engine: uplink, discovery, go2rtc watchdog
+  main.rs      Runs the uplink and the go2rtc watchdog side by side
-  uplink/             Cloud WebSocket client (reconnect, heartbeat, dispatch)
+  config.rs    Loads settings from environment variables
-  adapters/           bambu.rs, prusalink.rs, moonraker.rs — wire-protocol clients
+  uplink/      WebSocket client to continuum-backend: connect, heartbeat, reconnect on drop
-  printer/            Domain model: Printer trait + PrinterBase + PrinterHandle enum
+  printer/     Printer trait + PrinterBase + PrinterHandle enum + one stub per vendor
-  plate_changer/      Mechanical cycle control + sensor validation
+  go2rtc.rs    Restarts the go2rtc camera-restreaming process if it dies
-  discovery/          LAN printer discovery (SSDP / mDNS / static config)
+examples/
  printer_polymorphism.rs   Runs all three printer stubs through one `connect()` call site
 ```
-`src/adapters/` talks the raw wire protocols (MQTT, FTPS, HTTP, WS).
+`src/printer/` is where the "inheritance" question lives — see that
-`src/printer/` is the vendor-agnostic domain model layered on top: a
+module's doc comment for the trait+enum pattern this project uses instead
-`Printer` trait plus one struct per vendor (`BambuPrinter`, `PrusaPrinter`,
+of class inheritance, and run the example above to see it work.
 `KlipperPrinter`) wrapped in a closed `PrinterHandle` enum. Rust has no class
 inheritance, so this trait+enum combo is what stands in for a
 `GenericPrinter -> BambuPrinter` hierarchy — composition for shared fields,
 a trait for shared/overridable behavior, an enum for the closed set of
 vendors. See `examples/printer_polymorphism.rs`:
-```bash
+## What's not here yet (on purpose)
-cargo run --example printer_polymorphism
+
-```
+- Real MQTT/FTPS/HTTP/WebSocket printer clients — `src/printer/*.rs` has a
  `println!` where each of these will go.
 - Local SQLite buffering for telemetry across connectivity gaps.
 - LAN printer discovery (SSDP/mDNS).
 - The mechanical plate-changer interface (serial/GPIO).
 Add these back in one at a time as you get comfortable with the Rust
 underneath them — each is its own small lesson (async I/O, a new crate's
 API, error handling for a real protocol) rather than something to absorb
 all at once.
@@ -1,144 +0,0 @@
 use std::path::Path;
 use std::time::Duration;
 use rumqttc::{AsyncClient, Event, MqttOptions, Packet, QoS, TlsConfiguration, Transport};
 use serde::{Deserialize, Serialize};
 use suppaftp::async_native_tls::TlsConnector;
 use suppaftp::{AsyncNativeTlsConnector, AsyncNativeTlsFtpStream};
 use tokio::sync::mpsc;
 use tokio_util::compat::TokioAsyncReadCompatExt;
 use tracing::{debug, info, warn};
 const MQTT_PORT: u16 = 8883;
 const FTPS_PORT: u16 = 990;
 #[derive(Debug, Clone)]
 pub struct BambuPrinter {
    pub serial: String,
    pub host: String,
    pub access_code: String,
 }
 /// Telemetry pushed from the printer's `report` MQTT topic, decoded into the
 /// subset of fields the control plane cares about.
 #[derive(Debug, Clone, Deserialize)]
 pub struct BambuReport {
    #[serde(default)]
    pub nozzle_temper: Option<f32>,
    #[serde(default)]
    pub bed_temper: Option<f32>,
    #[serde(default)]
    pub mc_percent: Option<u8>,
    #[serde(default)]
    pub gcode_state: Option<String>,
 }
 #[derive(Debug, Serialize)]
 struct BambuCommandEnvelope<'a> {
    print: BambuCommand<'a>,
 }
 #[derive(Debug, Serialize)]
 struct BambuCommand<'a> {
    sequence_id: &'a str,
    command: &'a str,
 }
 /// Connects over local-LAN MQTTS to a single Bambu printer and streams
 /// decoded telemetry reports out on `tx`. Bambu's LAN-mode broker uses a
 /// self-signed cert, so a permissive TLS config is required.
 pub async fn run_telemetry(printer: BambuPrinter, tx: mpsc::Sender<(String, BambuReport)>) {
    loop {
        if let Err(err) = telemetry_session(&printer, &tx).await {
            warn!(serial = %printer.serial, ?err, "bambu MQTT session ended, reconnecting");
        }
        tokio::time::sleep(Duration::from_secs(5)).await;
    }
 }
 async fn telemetry_session(
    printer: &BambuPrinter,
    tx: &mpsc::Sender<(String, BambuReport)>,
 ) -> anyhow::Result<()> {
    let mut opts = MqttOptions::new(
        format!("continuum-proxy-{}", printer.serial),
        printer.host.clone(),
        MQTT_PORT,
    );
    opts.set_credentials("bblp", printer.access_code.clone());
    opts.set_keep_alive(Duration::from_secs(20));
    // Bambu LAN-mode uses a self-signed certificate; the client trusts it
    // explicitly because the connection never leaves the local network.
    opts.set_transport(Transport::Tls(TlsConfiguration::default()));
    let (client, mut event_loop) = AsyncClient::new(opts, 16);
    let report_topic = format!("device/{}/report", printer.serial);
    client.subscribe(&report_topic, QoS::AtMostOnce).await?;
    loop {
        match event_loop.poll().await? {
            Event::Incoming(Packet::Publish(publish)) if publish.topic == report_topic => {
                match serde_json::from_slice::<serde_json::Value>(&publish.payload) {
                    Ok(value) => {
                        if let Some(print) = value.get("print") {
                            if let Ok(report) = serde_json::from_value::<BambuReport>(print.clone()) {
                                debug!(serial = %printer.serial, ?report, "bambu telemetry");
                                if tx.send((printer.serial.clone(), report)).await.is_err() {
                                    return Ok(());
                                }
                            }
                        }
                    }
                    Err(err) => warn!(?err, "failed to decode bambu report payload"),
                }
            }
            Event::Incoming(Packet::Disconnect) => {
                return Err(anyhow::anyhow!("printer closed MQTT connection"));
            }
            _ => {}
        }
    }
 }
 /// Sends a G-code print-control command (pause/resume/stop/etc.) to the
 /// printer over its LAN MQTT channel.
 pub async fn send_command(printer: &BambuPrinter, sequence_id: &str, command: &str) -> anyhow::Result<()> {
    let mut opts = MqttOptions::new(format!("continuum-proxy-cmd-{}", printer.serial), printer.host.clone(), MQTT_PORT);
    opts.set_credentials("bblp", printer.access_code.clone());
    opts.set_transport(Transport::Tls(TlsConfiguration::default()));
    let (client, mut event_loop) = AsyncClient::new(opts, 4);
    let request_topic = format!("device/{}/request", printer.serial);
    let envelope = BambuCommandEnvelope {
        print: BambuCommand { sequence_id, command },
    };
    client
        .publish(&request_topic, QoS::AtLeastOnce, false, serde_json::to_vec(&envelope)?)
        .await?;
    // Pump the event loop once so the publish actually flushes before we drop the client.
    let _ = tokio::time::timeout(Duration::from_secs(3), event_loop.poll()).await;
    Ok(())
 }
 /// Uploads a sliced `.gcode.3mf` project file to the printer's local storage
 /// over FTPS (port 990, implicit TLS) ahead of a print job.
 pub async fn dispatch_file(printer: &BambuPrinter, local_path: &Path, remote_name: &str) -> anyhow::Result<()> {
    info!(serial = %printer.serial, remote_name, "dispatching file via FTPS");
    let ftp = AsyncNativeTlsFtpStream::connect(format!("{}:{FTPS_PORT}", printer.host)).await?;
    let mut ftp = ftp
        .into_secure(AsyncNativeTlsConnector::from(TlsConnector::new()), &printer.host)
        .await?;
    ftp.login("bblp", &printer.access_code).await?;
    // suppaftp wants a `futures::io::AsyncRead`; tokio::fs::File only
    // implements tokio's own AsyncRead, so bridge it with tokio-util's
    // `.compat()` adapter rather than pulling in a second async-fs stack.
    let mut file = tokio::fs::File::open(local_path).await?.compat();
    ftp.put_file(remote_name, &mut file).await?;
    ftp.quit().await?;
    Ok(())
 }
@@ -1,3 +0,0 @@
 pub mod bambu;
 pub mod moonraker;
 pub mod prusalink;
@@ -1,74 +0,0 @@
 use futures_util::{SinkExt, StreamExt};
 use serde::{Deserialize, Serialize};
 use tokio_tungstenite::tungstenite::Message;
 use tracing::warn;
 #[derive(Debug, Clone)]
 pub struct MoonrakerPrinter {
    pub host: String,
 }
 #[derive(Debug, Serialize)]
 struct JsonRpcRequest<'a> {
    jsonrpc: &'static str,
    method: &'a str,
    params: serde_json::Value,
    id: u64,
 }
 #[derive(Debug, Clone, Deserialize)]
 pub struct MoonrakerNotification {
    pub method: String,
    #[serde(default)]
    pub params: Vec<serde_json::Value>,
 }
 /// Subscribes to a Klipper/Moonraker printer's `printer.objects.subscribe`
 /// WebSocket feed and forwards decoded status notifications upstream.
 pub async fn run(printer: MoonrakerPrinter, tx: tokio::sync::mpsc::Sender<MoonrakerNotification>) {
    loop {
        if let Err(err) = session(&printer, &tx).await {
            warn!(host = %printer.host, ?err, "moonraker session ended, reconnecting");
        }
        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
    }
 }
 async fn session(
    printer: &MoonrakerPrinter,
    tx: &tokio::sync::mpsc::Sender<MoonrakerNotification>,
 ) -> anyhow::Result<()> {
    let url = format!("ws://{}/websocket", printer.host);
    let (ws_stream, _) = tokio_tungstenite::connect_async(&url).await?;
    let (mut write, mut read) = ws_stream.split();
    let subscribe = JsonRpcRequest {
        jsonrpc: "2.0",
        method: "printer.objects.subscribe",
        params: serde_json::json!({
            "objects": {
                "extruder": ["temperature", "target"],
                "heater_bed": ["temperature", "target"],
                "print_stats": ["state", "progress"],
            }
        }),
        id: 1,
    };
    write.send(Message::Text(serde_json::to_string(&subscribe)?)).await?;
    while let Some(frame) = read.next().await {
        match frame? {
            Message::Text(text) => {
                if let Ok(notification) = serde_json::from_str::<MoonrakerNotification>(&text) {
                    if tx.send(notification).await.is_err() {
                        return Ok(());
                    }
                }
            }
            Message::Close(_) => return Ok(()),
            _ => {}
        }
    }
    Ok(())
 }
@@ -1,56 +0,0 @@
 use serde::Deserialize;
 use tracing::warn;
 #[derive(Debug, Clone)]
 pub struct PrusaLinkPrinter {
    pub host: String,
    pub api_key: String,
 }
 #[derive(Debug, Clone, Deserialize)]
 pub struct PrusaLinkStatus {
    pub printer: PrusaLinkPrinterState,
 }
 #[derive(Debug, Clone, Deserialize)]
 pub struct PrusaLinkPrinterState {
    pub state: String,
    pub temp_nozzle: f32,
    pub temp_bed: f32,
 }
 /// Polls PrusaLink's REST API for the current printer state.
 /// PrusaLink has no push/streaming transport, so the proxy polls it on a
 /// short interval and forwards deltas upstream as synthetic telemetry.
 pub async fn fetch_status(client: &reqwest::Client, printer: &PrusaLinkPrinter) -> anyhow::Result<PrusaLinkStatus> {
    let url = format!("http://{}/api/v1/status", printer.host);
    let response = client
        .get(url)
        .header("X-Api-Key", &printer.api_key)
        .send()
        .await?
        .error_for_status()?;
    let status = response.json::<PrusaLinkStatus>().await?;
    Ok(status)
 }
 pub async fn poll_loop(
    client: reqwest::Client,
    printer: PrusaLinkPrinter,
    interval: std::time::Duration,
    tx: tokio::sync::mpsc::Sender<(String, PrusaLinkStatus)>,
 ) {
    let mut ticker = tokio::time::interval(interval);
    loop {
        ticker.tick().await;
        match fetch_status(&client, &printer).await {
            Ok(status) => {
                if tx.send((printer.host.clone(), status)).await.is_err() {
                    return;
                }
            }
            Err(err) => warn!(host = %printer.host, ?err, "prusalink poll failed"),
        }
    }
 }
@@ -1,59 +0,0 @@
 use rusqlite::Connection;
 use tracing::info;
 /// Local SQLite cache that buffers telemetry and job state across uplink
 /// outages so nothing is lost between the printer and the cloud.
 pub struct EdgeCache {
    conn: Connection,
 }
 impl EdgeCache {
    pub fn open(path: &str) -> anyhow::Result<Self> {
        if let Some(parent) = std::path::Path::new(path).parent() {
            std::fs::create_dir_all(parent)?;
        }
        let conn = Connection::open(path)?;
        conn.execute_batch(
            "CREATE TABLE IF NOT EXISTS pending_telemetry (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                printer_id TEXT NOT NULL,
                payload TEXT NOT NULL,
                created_at TEXT NOT NULL DEFAULT (datetime('now'))
            );
            CREATE TABLE IF NOT EXISTS job_state (
                job_id TEXT PRIMARY KEY,
                status TEXT NOT NULL,
                updated_at TEXT NOT NULL DEFAULT (datetime('now'))
            );",
        )?;
        info!(path, "opened edge cache");
        Ok(Self { conn })
    }
    pub fn enqueue_telemetry(&self, printer_id: &str, payload: &str) -> anyhow::Result<()> {
        self.conn.execute(
            "INSERT INTO pending_telemetry (printer_id, payload) VALUES (?1, ?2)",
            (printer_id, payload),
        )?;
        Ok(())
    }
    pub fn drain_telemetry(&self, limit: u32) -> anyhow::Result<Vec<(i64, String, String)>> {
        let mut stmt = self
            .conn
            .prepare("SELECT id, printer_id, payload FROM pending_telemetry ORDER BY id ASC LIMIT ?1")?;
        let rows = stmt
            .query_map([limit], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }
    pub fn ack_telemetry(&self, ids: &[i64]) -> anyhow::Result<()> {
        for id in ids {
            self.conn.execute("DELETE FROM pending_telemetry WHERE id = ?1", [id])?;
        }
        Ok(())
    }
 }
@@ -6,12 +6,8 @@ pub struct Config {
    pub farm_id: String,
    pub gateway_token: String,
    pub uplink_url: String,
    pub api_base: String,
    pub sqlite_path: String,
    pub go2rtc_bin: String,
    pub go2rtc_config: String,
    pub go2rtc_api_port: u16,
    pub discovery_interval_secs: u64,
 }
 impl Config {
@@ -21,12 +17,8 @@ impl Config {
            farm_id: require("CONTINUUM_FARM_ID")?,
            gateway_token: require("CONTINUUM_GATEWAY_TOKEN")?,
            uplink_url: env_or("CONTINUUM_UPLINK_URL", "wss://api.continuum.local/ws/edge/v1"),
            api_base: env_or("CONTINUUM_API_BASE", "https://api.continuum.local"),
            sqlite_path: env_or("CONTINUUM_SQLITE_PATH", "./data/edge-cache.db"),
            go2rtc_bin: env_or("CONTINUUM_GO2RTC_BIN", "go2rtc"),
            go2rtc_config: env_or("CONTINUUM_GO2RTC_CONFIG", "./go2rtc.yaml"),
            go2rtc_api_port: env_or("CONTINUUM_GO2RTC_API_PORT", "1984").parse()?,
            discovery_interval_secs: env_or("CONTINUUM_DISCOVERY_INTERVAL_SECS", "30").parse()?,
        })
    }
 }
@@ -1,44 +0,0 @@
 use std::time::Duration;
 use serde::{Deserialize, Serialize};
 use tracing::{debug, info};
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum PrinterVendor {
    Bambu,
    Prusa,
    Klipper,
 }
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DiscoveredPrinter {
    pub vendor: PrinterVendor,
    pub host: String,
    pub serial: Option<String>,
 }
 /// Periodically sweeps the LAN for printers (SSDP for Bambu, mDNS for
 /// PrusaLink/Moonraker) and reports newly-seen devices upstream.
 ///
 /// This is intentionally a stub: production discovery would bind a UDP
 /// multicast socket per protocol. It's structured so each protocol's probe
 /// can be dropped in independently without touching the polling loop.
 pub async fn run(interval: Duration, tx: tokio::sync::mpsc::Sender<DiscoveredPrinter>) {
    let mut ticker = tokio::time::interval(interval);
    loop {
        ticker.tick().await;
        debug!("running LAN discovery sweep");
        for printer in sweep().await {
            info!(host = %printer.host, ?printer.vendor, "discovered printer");
            if tx.send(printer).await.is_err() {
                return;
            }
        }
    }
 }
 async fn sweep() -> Vec<DiscoveredPrinter> {
    // TODO: SSDP probe (Bambu), mDNS `_prusalink._tcp` / `_moonraker._tcp` probes.
    Vec::new()
 }
@@ -1,118 +1,33 @@
 mod adapters;
 mod cache;
 mod config;
 mod discovery;
 mod go2rtc;
-mod plate_changer;
+mod printer;
 mod uplink;
 use std::time::Duration;
 use tokio::sync::mpsc;
 use tracing::{info, warn};
 use tracing_subscriber::EnvFilter;
 use crate::cache::EdgeCache;
 use crate::config::Config;
 use crate::uplink::ClientMessage;
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")))
        .json()
        .init();
-    dotenvy_load();
+    dotenvy::dotenv().ok();
    let config = Config::from_env()?;
    info!(gateway_id = %config.gateway_id, farm_id = %config.farm_id, "starting continuum-proxy");
-    let cache = EdgeCache::open(&config.sqlite_path)?;
+    tracing::info!(gateway_id = %config.gateway_id, farm_id = %config.farm_id, "starting continuum-proxy");
    // Channels wiring the printer adapters -> uplink, and uplink -> task
    // handlers for cloud-issued commands.
    let (telemetry_tx, telemetry_rx) = mpsc::channel::<ClientMessage>(256);
    let (task_tx, mut task_rx) = mpsc::channel::<uplink::ServerMessage>(64);
    let (discovered_tx, mut discovered_rx) = mpsc::channel::<discovery::DiscoveredPrinter>(32);
    let uplink_handle = tokio::spawn(uplink::run(config.clone(), telemetry_rx, task_tx));
    let go2rtc_handle = tokio::spawn(go2rtc::watchdog(config.go2rtc_bin.clone(), config.go2rtc_config.clone()));
    let discovery_handle = tokio::spawn(discovery::run(
        Duration::from_secs(config.discovery_interval_secs),
        discovered_tx,
    ));
    // Cloud-issued task dispatcher: pause/resume/plate-change/etc. commands
    // arriving over the uplink get routed to the right adapter here.
    let task_dispatcher = tokio::spawn(async move {
        while let Some(msg) = task_rx.recv().await {
            match msg {
                uplink::ServerMessage::Task { task_id, kind, payload } => {
                    info!(task_id, kind, ?payload, "received task from cloud");
                    // TODO: route to adapters::bambu / prusalink / moonraker or plate_changer
                    // based on `kind`, then send a ClientMessage::TaskResult back upstream.
                }
                uplink::ServerMessage::HelloAck { session_id } => {
                    info!(session_id, "uplink session established");
                }
                uplink::ServerMessage::HeartbeatAck { seq } => {
                    tracing::debug!(seq, "heartbeat acked");
                }
            }
        }
    });
    // Newly-discovered printers get their telemetry adapters spawned on the fly.
    let telemetry_tx_for_discovery = telemetry_tx.clone();
    let discovery_dispatcher = tokio::spawn(async move {
        while let Some(printer) = discovered_rx.recv().await {
            info!(host = %printer.host, "spawning adapter for discovered printer");
            let _ = &telemetry_tx_for_discovery;
            // TODO: match printer.vendor and spawn adapters::bambu::run_telemetry / moonraker::run / prusalink::poll_loop
        }
    });
    // Periodically flush anything the SQLite cache buffered while the uplink was down.
    let cache_flusher = tokio::spawn(async move {
        let mut ticker = tokio::time::interval(Duration::from_secs(10));
        loop {
            ticker.tick().await;
            match cache.drain_telemetry(100) {
                Ok(rows) if !rows.is_empty() => {
                    let ids: Vec<i64> = rows.iter().map(|(id, _, _)| *id).collect();
                    for (_, printer_id, payload) in &rows {
                        let payload: serde_json::Value = serde_json::from_str(payload).unwrap_or_default();
                        let _ = telemetry_tx
                            .send(ClientMessage::Telemetry { printer_id: printer_id.clone(), payload })
                            .await;
                    }
                    if let Err(err) = cache.ack_telemetry(&ids) {
                        warn!(?err, "failed to ack drained telemetry rows");
                    }
                }
                Ok(_) => {}
                Err(err) => warn!(?err, "failed to drain edge cache"),
            }
        }
    });
    // Two things run at once for now: the cloud uplink, and the go2rtc
    // camera-restreaming process. More will join them later — printer
    // discovery, the plate changer — as you build past src/printer/'s
    // stubs. `tokio::select!` runs both and stops as soon as either one
    // returns; both loop forever below, so in practice this runs until the
    // process is killed.
    tokio::select! {
-        res = uplink_handle => warn!(?res, "uplink task exited"),
+        _ = uplink::run(config.clone()) => {}
-        res = go2rtc_handle => warn!(?res, "go2rtc watchdog exited"),
+        _ = go2rtc::watchdog(config.go2rtc_bin.clone(), config.go2rtc_config.clone()) => {}
        res = discovery_handle => warn!(?res, "discovery task exited"),
        res = task_dispatcher => warn!(?res, "task dispatcher exited"),
        res = discovery_dispatcher => warn!(?res, "discovery dispatcher exited"),
        res = cache_flusher => warn!(?res, "cache flusher exited"),
    }
    Ok(())
 }
 /// Loads `.env` if present; a missing file is fine (env vars may already be
 /// set by the process supervisor), so the error is deliberately ignored.
 fn dotenvy_load() {
    let _ = dotenvy::dotenv();
 }
@@ -1,20 +0,0 @@
 /// Snapshot of the plate changer's discrete sensor inputs, read after a
 /// cycle completes to confirm the mechanism actually did what it reported.
 #[derive(Debug, Clone, Copy, Default)]
 pub struct SensorState {
    pub plate_present: bool,
    pub bed_clear: bool,
 }
 /// Reads the plate-present and bed-clear sensors.
 ///
 /// On the target hardware (Raspberry Pi / SBC GPIO header) this would read
 /// two debounced digital inputs via `rppal` or sysfs GPIO. Stubbed here so
 /// the crate builds without hardware access; swap in a real backend behind
 /// this same function signature.
 pub async fn read_sensor_state() -> SensorState {
    SensorState {
        plate_present: true,
        bed_clear: true,
    }
 }
@@ -1,68 +0,0 @@
 mod gpio;
 mod serial;
 use thiserror::Error;
 use tracing::{info, warn};
 #[derive(Debug, Error)]
 pub enum PlateChangerError {
    #[error("plate changer hardware not responding")]
    NotResponding,
    #[error("sensor validation failed: {0}")]
    SensorMismatch(String),
    #[error("serial transport error: {0}")]
    Serial(#[from] serial::SerialError),
 }
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum CycleOutcome {
    Success,
    Retried,
 }
 /// Drives a mechanical plate-swap cycle: signal the changer over serial,
 /// wait for it to report completion, then cross-check the physical sensors
 /// (plate-present, bed-clear) before releasing the print queue to continue.
 pub struct PlateChanger {
    port: serial::SerialPort,
 }
 impl PlateChanger {
    pub fn open(path: &str, baud: u32) -> Result<Self, PlateChangerError> {
        Ok(Self {
            port: serial::SerialPort::open(path, baud)?,
        })
    }
    pub async fn run_cycle(&mut self, slot: u8) -> Result<CycleOutcome, PlateChangerError> {
        info!(slot, "starting plate change cycle");
        self.port.send_command(&serial::Command::Eject).await?;
        self.port.await_ack(std::time::Duration::from_secs(30)).await?;
        self.port.send_command(&serial::Command::LoadSlot(slot)).await?;
        self.port.await_ack(std::time::Duration::from_secs(30)).await?;
        match self.validate_sensors().await {
            Ok(()) => Ok(CycleOutcome::Success),
            Err(err) => {
                warn!(?err, slot, "sensor validation failed, retrying cycle once");
                self.port.send_command(&serial::Command::LoadSlot(slot)).await?;
                self.port.await_ack(std::time::Duration::from_secs(30)).await?;
                self.validate_sensors().await?;
                Ok(CycleOutcome::Retried)
            }
        }
    }
    async fn validate_sensors(&mut self) -> Result<(), PlateChangerError> {
        let state = gpio::read_sensor_state().await;
        if !state.plate_present {
            return Err(PlateChangerError::SensorMismatch("plate not detected on bed".into()));
        }
        if !state.bed_clear {
            return Err(PlateChangerError::SensorMismatch("bed obstruction detected".into()));
        }
        Ok(())
    }
 }
@@ -1,64 +0,0 @@
 use std::time::Duration;
 use thiserror::Error;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio_serial::SerialPortBuilderExt;
 #[derive(Debug, Error)]
 pub enum SerialError {
    #[error("failed to open serial port: {0}")]
    Open(#[from] tokio_serial::Error),
    #[error("i/o error: {0}")]
    Io(#[from] std::io::Error),
    #[error("timed out waiting for hardware acknowledgement")]
    Timeout,
    #[error("hardware reported a fault: {0}")]
    Fault(String),
 }
 pub enum Command {
    Eject,
    LoadSlot(u8),
 }
 impl Command {
    fn encode(&self) -> String {
        match self {
            Command::Eject => "EJECT\n".to_string(),
            Command::LoadSlot(slot) => format!("LOAD {slot}\n"),
        }
    }
 }
 /// Thin line-protocol wrapper around the plate changer's serial control
 /// board (an ASCII command set over a USB-serial link, e.g. an Arduino/RP2040
 /// running the changer firmware).
 pub struct SerialPort {
    inner: tokio_serial::SerialStream,
 }
 impl SerialPort {
    pub fn open(path: &str, baud: u32) -> Result<Self, SerialError> {
        let inner = tokio_serial::new(path, baud).timeout(Duration::from_millis(500)).open_native_async()?;
        Ok(Self { inner })
    }
    pub async fn send_command(&mut self, command: &Command) -> Result<(), SerialError> {
        self.inner.write_all(command.encode().as_bytes()).await?;
        Ok(())
    }
    pub async fn await_ack(&mut self, timeout: Duration) -> Result<(), SerialError> {
        let mut buf = [0u8; 64];
        let read = tokio::time::timeout(timeout, self.inner.read(&mut buf))
            .await
            .map_err(|_| SerialError::Timeout)??;
        let response = String::from_utf8_lossy(&buf[..read]);
        if response.trim() == "OK" {
            Ok(())
        } else {
            Err(SerialError::Fault(response.trim().to_string()))
        }
    }
 }
@@ -1,3 +1,12 @@
 //! The cloud connection: a WebSocket to `continuum-backend`'s `/ws/edge/v1`.
 //!
 //! This is intentionally the simplest version of "a client that reconnects
 //! when it drops": no backoff curve, no channels routing messages in from
 //! other parts of the program yet — just connect, say hello, send a
 //! heartbeat on a timer, and print whatever comes back. Once you're
 //! comfortable with this, growing it (real backoff, forwarding printer
 //! telemetry, dispatching cloud-issued commands) is additive, not a rewrite.
 mod protocol;
 pub use protocol::{ClientMessage, ServerMessage};
@@ -5,109 +14,61 @@ pub use protocol::{ClientMessage, ServerMessage};
 use std::time::Duration;
 use futures_util::{SinkExt, StreamExt};
 use tokio::sync::mpsc;
 use tokio_tungstenite::tungstenite::Message;
 use tracing::{info, warn};
 use crate::config::Config;
-const MIN_BACKOFF: Duration = Duration::from_secs(1);
+const RETRY_DELAY: Duration = Duration::from_secs(5);
 const MAX_BACKOFF: Duration = Duration::from_secs(60);
 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);
-/// Runs the resilient uplink loop forever: connect, authenticate, exchange
+/// Runs forever: connect, serve the connection until it drops, wait, retry.
-/// heartbeats and telemetry/task messages, and reconnect with exponential
+pub async fn run(config: Config) {
 /// backoff (plus jitter) whenever the connection drops.
 pub async fn run(
    config: Config,
    mut telemetry_rx: mpsc::Receiver<ClientMessage>,
    task_tx: mpsc::Sender<ServerMessage>,
 ) {
    let mut backoff = MIN_BACKOFF;
    loop {
        info!(url = %config.uplink_url, "connecting to cloud uplink");
-        match connect_and_serve(&config, &mut telemetry_rx, &task_tx).await {
+        if let Err(err) = connect_and_serve(&config).await {
-            Ok(()) => {
+            warn!(?err, "uplink connection failed, retrying in {RETRY_DELAY:?}");
                info!("uplink connection closed cleanly");
                backoff = MIN_BACKOFF;
        }
-            Err(err) => {
+
-                warn!(?err, backoff_secs = backoff.as_secs(), "uplink connection failed, retrying");
+        tokio::time::sleep(RETRY_DELAY).await;
    }
 }
-        let jitter = Duration::from_millis(rand::random::<u64>() % 500);
+async fn connect_and_serve(config: &Config) -> anyhow::Result<()> {
-        tokio::time::sleep(backoff + jitter).await;
+    let (ws_stream, _response) = tokio_tungstenite::connect_async(&config.uplink_url).await?;
        backoff = (backoff * 2).min(MAX_BACKOFF);
    }
 }
 async fn connect_and_serve(
    config: &Config,
    telemetry_rx: &mut mpsc::Receiver<ClientMessage>,
    task_tx: &mpsc::Sender<ServerMessage>,
 ) -> anyhow::Result<()> {
    let (ws_stream, _resp) = tokio_tungstenite::connect_async(&config.uplink_url).await?;
    let (mut write, mut read) = ws_stream.split();
-    send(&mut write, &ClientMessage::Hello {
+    let hello = ClientMessage::Hello {
        gateway_id: config.gateway_id.clone(),
        farm_id: config.farm_id.clone(),
        token: config.gateway_token.clone(),
-        version: env!("CARGO_PKG_VERSION"),
+    };
-    })
+    write.send(Message::Text(serde_json::to_string(&hello)?)).await?;
    .await?;
    let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL);
    let mut seq: u64 = 0;
    heartbeat.tick().await; // consume the immediate first tick
    loop {
        tokio::select! {
            _ = heartbeat.tick() => {
                seq += 1;
-                send(&mut write, &ClientMessage::Heartbeat { seq }).await?;
+                let msg = ClientMessage::Heartbeat { seq };
-            }
+                write.send(Message::Text(serde_json::to_string(&msg)?)).await?;
            Some(msg) = telemetry_rx.recv() => {
                send(&mut write, &msg).await?;
            }
            frame = read.next() => {
                match frame {
                    Some(Ok(Message::Text(text))) => {
-                        match serde_json::from_str::<ServerMessage>(&text) {
+                        if let Ok(server_msg) = serde_json::from_str::<ServerMessage>(&text) {
-                            Ok(server_msg) => {
+                            info!(?server_msg, "received from cloud");
                                if task_tx.send(server_msg).await.is_err() {
                                    return Ok(());
                        }
                    }
-                            Err(err) => warn!(?err, "failed to decode server message"),
+                    Some(Ok(Message::Close(_))) | None => return Ok(()),
                        }
                    }
                    Some(Ok(Message::Ping(payload))) => {
                        write.send(Message::Pong(payload)).await?;
                    }
                    Some(Ok(Message::Close(frame))) => {
                        info!(?frame, "server closed uplink");
                        return Ok(());
                    }
                    Some(Ok(_)) => {}
                    Some(Err(err)) => return Err(err.into()),
-                    None => return Ok(()),
+                    _ => {}
                }
            }
        }
    }
 }
 async fn send(
    write: &mut (impl SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin),
    msg: &ClientMessage,
 ) -> anyhow::Result<()> {
    let payload = serde_json::to_string(msg)?;
    write.send(Message::Text(payload)).await?;
    Ok(())
 }
@@ -1,42 +1,17 @@
 use serde::{Deserialize, Serialize};
-/// Messages sent from this gateway up to `continuum-backend`'s `/ws/edge/v1` route.
+/// What this gateway sends up to `continuum-backend`'s `/ws/edge/v1` route.
 #[derive(Debug, Clone, Serialize)]
 #[serde(tag = "type", rename_all = "snake_case")]
 pub enum ClientMessage {
-    Hello {
+    Hello { gateway_id: String, farm_id: String, token: String },
-        gateway_id: String,
+    Heartbeat { seq: u64 },
        farm_id: String,
        token: String,
        version: &'static str,
    },
    Heartbeat {
        seq: u64,
    },
    Telemetry {
        printer_id: String,
        payload: serde_json::Value,
    },
    TaskResult {
        task_id: String,
        ok: bool,
        error: Option<String>,
    },
 }
-/// Messages received from `continuum-backend` over the same connection.
+/// What `continuum-backend` sends back over the same connection.
 #[derive(Debug, Clone, Deserialize)]
 #[serde(tag = "type", rename_all = "snake_case")]
 pub enum ServerMessage {
-    HelloAck {
+    HelloAck { session_id: String },
-        session_id: String,
+    HeartbeatAck { seq: u64 },
    },
    HeartbeatAck {
        seq: u64,
    },
    Task {
        task_id: String,
        kind: String,
        payload: serde_json::Value,
    },
 }