continuum-proxy/src/main.rs

// Everything but go2rtc comes from the library crate (src/lib.rs) rather than
// being re-declared here. Declaring `mod printer;` in both places compiles the
// same files twice, once per crate, and the binary copy warned about every
// `pub use` re-export and every method only examples or tests call - "unused"
// is true inside a binary, where nothing is externally reachable, and wrong
// about the library's actual public API. One copy, no false warnings.
mod go2rtc;

use tracing::info;
use tracing_subscriber::EnvFilter;

use continuum_proxy::bootstrap::{self, Bootstrap};
use continuum_proxy::config::Config;
use continuum_proxy::config_import;
use continuum_proxy::printer::{GenericPrinter, PrinterHandle};
use continuum_proxy::{camera, completion, fleet, uplink};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")))
        .init();

    dotenvy::dotenv().ok();

    // Order matters: import first, so a stick present at boot is already
    // the installed config by the time it is read. A rejected stick is
    // logged inside and does not stop startup - the box runs on what it has.
    let bootstrap = Bootstrap::discover()?;
    // Plaintext bundles are refused on a Proxy Box and signatures are
    // mandatory once a signing key is provisioned; see the config_import
    // module docs. An unprovisioned box is allowed to start - it runs on its
    // installed config - but says loudly what it cannot check.
    //
    // Loaded only on a box: on a PC nothing imports from removable media, and
    // a stray or malformed file under /etc/continuum must not stop a PC user's
    // proxy starting over provisioning it never uses.
    let policy = if bootstrap.mode == bootstrap::Mode::ProxyBox {
        config_import::load_provisioning()?
    } else {
        config_import::BundlePolicy::unencrypted_only()
    };
    if bootstrap.mode == bootstrap::Mode::ProxyBox {
        if policy.signing_keys.is_empty() {
            tracing::warn!("no signing key provisioned: config bundles will NOT be authenticated");
        }
        if policy.box_id.is_none() {
            tracing::warn!("no box id provisioned: signed config bundles cannot be accepted");
        }
    }
    for outcome in config_import::import_from_pendrives(&bootstrap, &policy) {
        config_import::log_outcome(&outcome);
    }
    let file = bootstrap.load_config_file()?;
    let config = Config::from_file_and_env(file.as_ref())?;
    info!(mode = ?bootstrap.mode, config_file = ?file.as_ref().map(|_| &bootstrap.config_path), "configuration loaded");

    // BOUNDARY: network settings are parsed and validated, and stop here.
    // Applying them (static IP, DNS, wifi) is the appliance image's job, not
    // this process's - see the module docs of `config`. Logged so that a box
    // given a network block does not look like it acted on it.
    if let Some(network) = file.as_ref().and_then(|f| f.network.as_ref()) {
        tracing::warn!(?network, "network settings are present but this build does NOT apply them");
    }

    tracing::info!(gateway_id = %config.gateway_id, farm_id = %config.farm_id, "starting continuum-proxy");

    // 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.
    // 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
    // keeping connections alive: the uplink reads each printer's status to
    // send telemetry up, and looks printers up by id to dispatch commands
    // down. `Arc<Mutex<..>>` is how two tasks share one mutable thing -
    // see `uplink::SharedFleet` for why that type and what it costs.
    let fleet: uplink::SharedFleet = std::sync::Arc::new(tokio::sync::Mutex::new(fleet));

    // Five long-running tasks: the cloud uplink, the go2rtc subprocess
    // watchdog, the camera-config builder that tells it what to serve, the
    // print-completion watcher that blinks a printer's lights when it
    // finishes, and printer discovery. Every printer's MQTT session is a
    // sixth thing running concurrently, but implicitly - each one has its own
    // background poll task spawned by `connect()`, kept alive by the fleet
    // above. More will join these later (the plate changer).
    //
    // Discovery parks on a future that never completes unless it is switched
    // on, which it is not by default - see `discovery::report::run`. So this
    // arm is usually a task that costs nothing, and that is the intended
    // steady state, not a disabled feature waiting to be finished.
    //
    // The completion watcher is deliberately its own task rather than part
    // of the uplink: it must keep working with the cloud connection down,
    // because the person it is for is standing next to the printer. See
    // `completion::run`.
    //
    // The camera builder and the watchdog are two tasks and not one because
    // they answer different questions - what should be served, and keeping
    // the thing that serves it alive. They meet at one `Notify`: the builder
    // rings it after writing a new config, because go2rtc only reads its
    // configuration at startup.
    //
    // `tokio::select!` stops as soon as any returns; all four loop forever,
    // so in practice this runs until the process is killed.
    let restart_go2rtc = std::sync::Arc::new(tokio::sync::Notify::new());

    // Built once, here, rather than inside the rebuild loop: the config is
    // re-rendered every few seconds and compared byte-for-byte to decide whether to
    // restart go2rtc, so a credential that changed per render would restart
    // it - and drop every viewer - on every tick.
    let camera_auth = camera::ApiAuth::new()?;
    info!(
        user = %camera_auth.username,
        config = %config.go2rtc_config,
        "go2rtc's API requires basic auth; the password is in its config file"
    );

    // Created here rather than inside the uplink so discovery can hand it
    // messages without the uplink knowing what discovery is. It outlives any
    // one connection, so a scan that finishes while the cloud is unreachable
    // waits for the next socket instead of being thrown away.
    let (outbound, inbox) = uplink::channel();

    tokio::select! {
        // Ends the process through a normal return, so the supervisors below
        // are dropped and go2rtc's `kill_on_drop` actually fires. Without this
        // arm SIGTERM - what `kill`, `docker stop` and a service manager all
        // send - ends the process on the spot with no destructor run, and
        // go2rtc is orphaned: it keeps :1984 and :8555 bound, so the restarted
        // proxy cannot start its own, and it keeps a printer's access code in
        // memory. Found when a test run left one behind.
        _ = shutdown_requested() => {
            info!("stop requested; exiting so go2rtc is stopped with us");
        }
        _ = 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(
            config.go2rtc_bin.clone(),
            config.go2rtc_config.clone(),
            restart_go2rtc.clone(),
        ) => {}
        _ = camera::run(
            fleet.clone(),
            config.go2rtc_config.clone(),
            config.go2rtc_api_listen.clone(),
            camera_auth,
            restart_go2rtc.clone(),
        ) => {}
        _ = completion::run(fleet.clone()) => {}
        // Returns only after installing a new config from a stick inserted
        // while running. Settings are read once at startup, so the process
        // ends here and relies on its supervisor (systemd `Restart=always`
        // on a box) to start it again on the new config. Not exercised end
        // to end - there is no supervisor or real stick in the dev
        // environment.
        _ = config_import::watch(bootstrap.clone(), policy.clone()) => {
            info!("exiting so the supervisor restarts the proxy on the newly installed config");
        }
    }

    Ok(())
}

/// Resolves when the process is asked to stop: SIGTERM or SIGINT on Unix,
/// Ctrl-C elsewhere.
async fn shutdown_requested() {
    #[cfg(unix)]
    {
        use tokio::signal::unix::{SignalKind, signal};
        if let Ok(mut terminate) = signal(SignalKind::terminate()) {
            tokio::select! {
                _ = terminate.recv() => {}
                _ = tokio::signal::ctrl_c() => {}
            }
            return;
        }
    }
    let _ = tokio::signal::ctrl_c().await;
}

/// Connects every configured printer concurrently (not one at a time -
/// Bambu's auto-detect fallback alone can take several seconds per
/// printer if it has to time out on MQTT v5 before trying v3.1.1;
/// serially, a fleet of twenty would take twenty times as long to start).
/// Logs success/failure per printer instead of failing startup over one -
/// a farm shouldn't refuse to start because one printer is unplugged.
async fn connect_fleet(fleet: &mut [PrinterHandle]) {
    let results =
        futures_util::future::join_all(fleet.iter_mut().map(|printer| async {
            let name = printer.display_name();
            (name, printer.connect().await)
        }))
        .await;

    for (name, result) in results {
        match result {
            Ok(()) => tracing::info!(printer = %name, "connected"),
            Err(err) => tracing::warn!(printer = %name, %err, "failed to connect"),
        }
    }
}