//! Container log operations — show logs and follow logs for a recipe.

use anyhow::{Context, Result};
use tracing::info;

use crate::config::{Cluster, Config, Node};
use crate::ssh::{run_ssh_follow, run_ssh_with_retry};

use super::shq;

pub fn logs_tail_cmd(recipe_name: &str, rank: u32, tail: usize) -> String {
    format!(
        "docker ps -q --filter {} --filter label=sm121.rank={} | xargs -r docker logs --timestamps --tail={} 2>&1",
        shq(&format!("label=sm121.recipe={}", recipe_name)),
        rank,
        tail
    )
}

pub fn logs_follow_cmd(recipe_name: &str) -> String {
    format!(
        "docker ps -q --filter {} --filter label=sm121.rank=0 | xargs -r docker logs -f --timestamps 2>&1",
        shq(&format!("label=sm121.recipe={}", recipe_name))
    )
}

pub async fn show_logs(
    config: &Config,
    cluster: &Cluster,
    recipe_name: &str,
    follow: bool,
    node_name: Option<&str>,
    dry_run: bool,
) -> Result<()> {
    // Resolve target nodes
    let targets: Vec<&Node> = match node_name {
        Some(name) => cluster.nodes.iter().filter(|n| n.name == *name).collect(),
        None => cluster.nodes.iter().collect(),
    };

    if targets.is_empty() {
        if let Some(n) = node_name {
            let name = n.to_string();
            anyhow::bail!(
                fluent_i18n::t!("exec-node-not-found", { "name" => &name, "cluster" => &cluster.name })
            );
        }
        anyhow::bail!(fluent_i18n::t!("exec-cluster-empty", { "name" => &cluster.name }));
    }

    if follow {
        // Follow mode: only head node (rank 0) or specified node with rank 0
        let target_node = targets
            .iter()
            .find(|n| n.rank == 0)
            .or(targets.first())
            .context("No nodes in cluster")?;
        info!(
            "{}",
            fluent_i18n::t!("logs-follow-head", { "name" => &target_node.name })
        );
        let cmd = logs_follow_cmd(recipe_name);
        run_ssh_follow(config, target_node, cmd, dry_run).await?;
    } else {
        let cn = cluster.name.clone();
        info!(
            "{}",
            fluent_i18n::t!("logs-header", { "name" => cn, "count" => targets.len() })
        );
        let mut handles = Vec::new();
        for node in targets {
            let cmd = logs_tail_cmd(recipe_name, node.rank, 100);
            let config = config.clone();
            let node = (*node).clone();
            let dry = dry_run;
            let name = node.name.clone();
            handles.push(crate::locale::spawn_with_locale(async move {
                (name, run_ssh_with_retry(&config, &node, cmd, dry, 1).await)
            }));
        }
        for result in futures::future::join_all(handles).await {
            if let Ok((name, Ok(output))) = result
                && !output.trim().is_empty()
            {
                println!("{}", fluent_i18n::t!("logs-section", { "name" => &name }));
                println!("{}", output);
            }
        }
    }
    Ok(())
}

/// Follow the head node's logs with an engine-liveness watch: a dead engine
/// (the API stops answering for two consecutive probes, or the log stream
/// ends while the engine is down) fails loudly and closes the recipe's run
/// in the ledger. Returns cleanly only when the follow ends with the engine
/// still up (ssh drop).
pub async fn follow_with_liveness(
    config: &Config,
    cluster: &Cluster,
    recipe_name: &str,
    port: u16,
    node_name: Option<&str>,
    dry_run: bool,
) -> Result<()> {
    let watch = crate::readiness::watch(cluster, port);
    tokio::select! {
        r = show_logs(config, cluster, recipe_name, true, node_name, dry_run) => {
            r?;
            // The log stream ended. Do NOT decide death from the one-shot API
            // probe (`alive`): the API port is closed for the whole (multi-
            // minute) weight load, so a false answer here is indistinguishable
            // from "still starting" — that mislabelled a slow boot as "engine
            // is down — containers died". A real death is a container that has
            // EXITED; a still-Up container means the engine is loading or the
            // ssh follow simply dropped. Mirror `wait_engine`'s check.
            if let Err(e) =
                crate::readiness::containers_alive(config, cluster, recipe_name).await
            {
                let _ = crate::runlock::mark_stop(recipe_name, &config.recipe_sources);
                anyhow::bail!("log stream ended and a container exited — {e}");
            }
            tracing::warn!(
                "log stream ended but the containers are still up (ssh drop, or the engine is still starting)"
            );
            Ok(())
        }
        r = watch => {
            let _ = crate::runlock::mark_stop(recipe_name, &config.recipe_sources);
            r
        }
    }
}
#[cfg(test)]
mod tests {
    use super::{logs_follow_cmd, logs_tail_cmd};

    #[test]
    fn tail_cmd_shell_quotes_recipe_name() {
        // A recipe name with shell metacharacters (`"`, `;`, spaces) must be
        // single-quoted so it can't break out of the docker --filter argument
        // or inject a command separator. (Regression for C-1.)
        let cmd = logs_tail_cmd("foo\";rm -rf /", 0, 100);
        assert!(
            cmd.contains("'label=sm121.recipe=foo\";rm -rf /'"),
            "recipe name not shell-quoted: {cmd}"
        );
        // No bare, unquoted command separator reaches the shell.
        assert!(!cmd.contains("--filter label=sm121.recipe=foo\";"));
    }

    #[test]
    fn follow_cmd_shell_quotes_recipe_name_with_single_quote() {
        // `$HOME` and spaces must stay literal (single-quoted) — not shell-
        // expanded. The inner-single-quote *escaping* of `shq` itself is
        // covered by `docker::tests::shq_escapes_inner_single_quotes`.
        let cmd = logs_follow_cmd("a b$HOME");
        assert!(
            cmd.contains("'label=sm121.recipe=a b$HOME'"),
            "recipe name not shell-quoted (would let $HOME expand): {cmd}"
        );
    }
}