launcher/src/docker/push.rs

//! Post-build registry push (`[registry]` config section).
//!
//! Images are pushed as **version tags**, not `:local`: the version is the
//! image's build checksum (recursive SHA-256 of the image dir, the same
//! value stamped as the `sm121.checksum` label) — identical content always
//! maps to the identical tag, so re-publishing "the same thing" cannot
//! create a duplicate; changed content is a new version. Each push also
//! refreshes the mutable `latest` tag.
//!
//! A local publication ledger (`published.json` next to the global config)
//! records what was published and with which image digest: a re-push with
//! an unchanged digest is skipped (with a note) unless forced. The push
//! itself runs in a tmux session on the build node — a 30 GB push must not
//! ride on a single SSH session — polled with the same 30 s / 90 min loop
//! as the build. The launcher stores no registry credentials: the node must
//! be `docker login`-ed once.

use super::build::{DEFAULT_MAX_BUILD_SECS, format_elapsed};
use super::shq;
use crate::config::{Config, Node};
use crate::ssh::run_ssh;
use anyhow::Result;
use std::collections::HashMap;
use tracing::info;

/// One published image: its version tag, the local image digest it was
/// published from, and when. Keyed by the repo name (the image name without
/// its tag).
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
pub struct PublishedEntry {
    pub version: String,
    pub image_id: String,
    pub published_at: String,
}

/// Ledger file: `<global config dir>/published.json`.
fn ledger_path() -> Option<std::path::PathBuf> {
    crate::config::default_config_dir().map(|d| d.join("published.json"))
}

fn read_published() -> HashMap<String, PublishedEntry> {
    let Some(p) = ledger_path() else {
        return HashMap::new();
    };
    std::fs::read_to_string(p)
        .ok()
        .and_then(|s| serde_json::from_str(&s).ok())
        .unwrap_or_default()
}

fn record_published(name: &str, entry: &PublishedEntry) {
    let Some(p) = ledger_path() else {
        return;
    };
    let mut m = read_published();
    m.insert(name.to_string(), entry.clone());
    if let Some(parent) = p.parent() {
        let _ = std::fs::create_dir_all(parent);
    }
    let _ = std::fs::write(&p, serde_json::to_string_pretty(&m).unwrap_or_default());
}

/// Map a local image to the two pushed refs: `{base}/{name}:{version}` and
/// `{base}/{name}:latest`. `None` when the mapping is meaningless (empty
/// base, image name, or version).
pub(crate) fn registry_refs(
    base: &str,
    image_tag: &str,
    version: &str,
) -> Option<(String, String)> {
    let base = base.trim().trim_end_matches('/');
    let name = image_tag.trim();
    let version = version.trim();
    if base.is_empty() || name.is_empty() || version.is_empty() {
        return None;
    }
    let name_part = name.rsplit_once(':').map(|(n, _)| n).unwrap_or(name);
    let repo = name_part.rsplit('/').next()?.trim();
    if repo.is_empty() {
        return None;
    }
    Some((
        format!("{base}/{repo}:{version}"),
        format!("{base}/{repo}:latest"),
    ))
}

/// Push a built image to the configured `[registry]` (no-op when the section
/// is absent). `version` = the build checksum (first 12 hex chars); both the
/// version tag and `latest` are pushed. Skips (with a note) when the ledger
/// records this exact image digest under the same version, unless `force`.
pub async fn push_image(
    config: &Config,
    node: &Node,
    image_tag: &str,
    version: &str,
    force: bool,
    label: &str,
) -> Result<()> {
    let Some(registry) = config.registry.as_ref() else {
        return Ok(());
    };
    let Some((version_ref, latest_ref)) = registry_refs(&registry.base, image_tag, version) else {
        anyhow::bail!(
            "{}",
            fluent_i18n::t!("push-mapping-failed", {
                "base" => registry.base.clone(),
                "image" => image_tag.to_string()
            })
        );
    };
    let host = registry.base.split('/').next().unwrap_or(&registry.base);
    let repo_name = latest_ref
        .rsplit_once(":latest")
        .map(|(repo, _)| repo.to_string())
        .unwrap_or_else(|| image_tag.to_string());

    // Duplicate guard: ledger says this exact image was already published
    // under this version → skip unless explicitly forced.
    let image_id = run_ssh(
        config,
        node,
        format!(
            "docker image inspect --format '{{{{.Id}}}}' {} 2>/dev/null || echo MISSING",
            shq(image_tag)
        ),
        false,
    )
    .await
    .unwrap_or_default()
    .trim()
    .to_string();
    if image_id.is_empty() || image_id == "MISSING" {
        let node_name = node.name.clone();
        anyhow::bail!("image '{image_tag}' not found on '{node_name}' — build it first");
    }
    if !force
        && let Some(e) = read_published().get(&repo_name)
        && e.image_id == image_id
        && e.version == version.trim()
    {
        info!(
            "{}",
            fluent_i18n::t!("push-already-published", {
                "ref" => version_ref.clone(),
                "label" => label.to_string()
            })
        );
        return Ok(());
    }

    let session_name = format!("sm121-push-{label}");
    let logfile = format!("/tmp/{label}-push.log");
    let marker = format!("Push done: {version_ref}");

    run_ssh(
        config,
        node,
        format!(
            "tmux kill-session -t {s} 2>/dev/null || true",
            s = shq(&session_name)
        ),
        false,
    )
    .await?;
    let tmux_cmd = format!(
        "tmux new-session -d -s {s} \"docker push {a} > {log} 2>&1 && docker push {b} >> {log} 2>&1 \
         && echo '{m}' >> {log}\"",
        s = shq(&session_name),
        a = shq(&version_ref),
        b = shq(&latest_ref),
        m = shq(&marker),
        log = shq(&logfile)
    );
    info!(
        "{}",
        fluent_i18n::t!("pushing-image", { "ref" => version_ref.clone(), "session" => session_name.clone() })
    );
    run_ssh(config, node, tmux_cmd, false).await?;

    let start = std::time::Instant::now();
    loop {
        tokio::time::sleep(std::time::Duration::from_secs(30)).await;
        let elapsed = start.elapsed();
        if elapsed.as_secs() >= DEFAULT_MAX_BUILD_SECS {
            anyhow::bail!(
                "{}",
                fluent_i18n::t!("push-timeout", {
                    "ref" => version_ref.clone(),
                    "node" => node.name.clone(),
                    "elapsed" => format_elapsed(elapsed),
                    "session" => session_name.clone(),
                    "log" => logfile.clone()
                })
            );
        }
        let alive = run_ssh(
            config,
            node,
            format!(
                "tmux has-session -t {s} 2>&1 || echo DEAD",
                s = shq(&session_name)
            ),
            false,
        )
        .await?;
        let dead = alive.trim().contains("DEAD");
        if let Ok(logs) = run_ssh(
            config,
            node,
            format!(
                "tail -10 {log} 2>/dev/null || echo '(no log yet)'",
                log = shq(&logfile)
            ),
            false,
        )
        .await
        {
            for line in logs.lines().filter(|l| !l.is_empty()) {
                info!("  [push] {}", line.trim());
            }
        }
        if dead {
            let ok = run_ssh(
                config,
                node,
                format!(
                    "grep -F -- {m} {log} >/dev/null 2>&1 && echo OK || echo NO",
                    m = shq(&marker),
                    log = shq(&logfile)
                ),
                false,
            )
            .await
            .map(|s| s.trim() == "OK")
            .unwrap_or(false);
            if ok {
                let digest = run_ssh(
                    config,
                    node,
                    format!(
                        "grep -oE 'digest: sha256:[0-9a-f]+ size: [0-9]+' {log} | tail -1",
                        log = shq(&logfile)
                    ),
                    false,
                )
                .await
                .unwrap_or_default();
                let digest_suffix = digest
                    .trim()
                    .lines()
                    .last()
                    .map(|d| format!(" — {d}"))
                    .unwrap_or_default();
                info!(
                    "{}{}",
                    fluent_i18n::t!("pushed-image", {
                        "ref" => version_ref.clone(),
                        "latest" => latest_ref.clone(),
                        "elapsed" => format_elapsed(elapsed)
                    }),
                    digest_suffix
                );
                record_published(
                    &repo_name,
                    &PublishedEntry {
                        version: version.trim().to_string(),
                        image_id,
                        published_at: crate::stamp::utc_stamp(),
                    },
                );
            } else {
                let exit_log = run_ssh(
                    config,
                    node,
                    format!("tail -20 {log}", log = shq(&logfile)),
                    false,
                )
                .await
                .unwrap_or_default();
                anyhow::bail!(
                    "{}",
                    fluent_i18n::t!("push-failed", {
                        "ref" => version_ref.clone(),
                        "node" => node.name.clone(),
                        "elapsed" => format_elapsed(elapsed),
                        "log" => exit_log.trim().to_string(),
                        "host" => host.to_string()
                    })
                );
            }
            return Ok(());
        }
        info!(
            "{}",
            fluent_i18n::t!("push-running", { "elapsed" => format_elapsed(elapsed) })
        );
    }
}