dora-cli 1.0.0

use std::io::Write;

use clap::Args;
use colored::Colorize;
use serde::Serialize;
use tabwriter::TabWriter;

use crate::{
    command::{
        Executable, default_tracing,
        topic::selector::{node_topic_inputs, node_topic_outputs},
    },
    common::{
        CoordinatorOptions, expect_reply, resolve_dataflow_identifier_interactive,
        send_control_request,
    },
    formatting::OutputFormat,
    ws_client::WsSession,
};
use dora_core::config::InputMapping;
use dora_message::{
    cli_to_coordinator::ControlRequest, coordinator_to_cli::NodeInfo, descriptor::Descriptor,
    id::NodeId,
};

/// Show detailed information about a specific node.
///
/// Displays inputs, outputs, subscribers, restart policy, and runtime metrics.
///
/// Examples:
///
/// Show info for a node:
///   dora node info camera_node
///
/// Show info for a node in a specific dataflow:
///   dora node info camera_node --dataflow my-dataflow
///
/// Output as JSON:
///   dora node info camera_node --format json
#[derive(Debug, Args)]
#[clap(verbatim_doc_comment)]
pub struct Info {
    /// Node ID to inspect
    #[clap(value_name = "NODE")]
    pub node: String,

    /// Filter by dataflow name or UUID
    #[clap(long, short = 'd', value_name = "NAME_OR_UUID")]
    pub dataflow: Option<String>,

    /// Output format
    ///
    /// `json` emits a single pretty-printed JSON document.
    #[clap(long, short = 'f', value_name = "FORMAT", default_value_t = OutputFormat::Table)]
    pub format: OutputFormat,

    #[clap(flatten)]
    coordinator: CoordinatorOptions,
}

impl Executable for Info {
    fn execute(self) -> eyre::Result<()> {
        default_tracing()?;

        let session = self.coordinator.connect()?;
        info(&session, &self.node, self.dataflow.as_deref(), self.format)
    }
}

#[derive(Serialize)]
struct NodeInfoOutput {
    node_id: String,
    dataflow: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    name: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    description: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    path: Option<String>,
    outputs: Vec<OutputInfo>,
    inputs: Vec<InputInfo>,
    restart_policy: String,
    max_restarts: u32,
    #[serde(skip_serializing_if = "Option::is_none")]
    restart_delay: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    health_check_timeout: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    finish_grace_secs: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    metrics: Option<MetricsOutput>,
}

#[derive(Serialize)]
struct OutputInfo {
    id: String,
    subscribers: Vec<String>,
}

#[derive(Serialize)]
struct InputInfo {
    id: String,
    source: String,
}

#[derive(Serialize)]
struct MetricsOutput {
    status: String,
    pid: u32,
    cpu: String,
    memory: String,
    restart_count: u32,
    pending_messages: u64,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    broken_inputs: Vec<String>,
}

fn info(
    session: &WsSession,
    node_name: &str,
    dataflow_filter: Option<&str>,
    format: OutputFormat,
) -> eyre::Result<()> {
    let node_id: NodeId = node_name
        .parse::<NodeId>()
        .map_err(|e| eyre::eyre!("invalid node ID: {e}"))?;

    // Resolve the dataflow
    let dataflow_uuid = resolve_dataflow_identifier_interactive(session, dataflow_filter)?;

    // Get descriptor
    let reply = send_control_request(session, &ControlRequest::Info { dataflow_uuid })?;
    let (dataflow_name, descriptor) = expect_reply!(reply, DataflowInfo { name, descriptor })?;

    // Find the node in the descriptor
    let node_desc = descriptor
        .nodes
        .iter()
        .find(|n| n.id == node_id)
        .ok_or_else(|| {
            let available: Vec<_> = descriptor.nodes.iter().map(|n| n.id.to_string()).collect();
            eyre::eyre!(
                "node `{node_name}` not found in dataflow\n\n  \
                 hint: available nodes: {}",
                available.join(", ")
            )
        })?;

    // Get runtime metrics
    let metrics = fetch_node_metrics(session, dataflow_uuid, &node_id)?;

    // Build output info with subscribers
    let outputs = build_output_info(node_desc, &descriptor);
    let inputs = build_input_info(node_desc);

    let dataflow_display = dataflow_name
        .clone()
        .unwrap_or_else(|| dataflow_uuid.to_string());

    let output = NodeInfoOutput {
        node_id: node_name.to_string(),
        dataflow: dataflow_display.clone(),
        name: node_desc.name.clone(),
        description: node_desc.description.clone(),
        path: node_desc.path.clone(),
        outputs,
        inputs,
        restart_policy: format!("{:?}", node_desc.restart_policy),
        max_restarts: node_desc.max_restarts,
        restart_delay: node_desc.restart_delay,
        health_check_timeout: node_desc.health_check_timeout,
        finish_grace_secs: node_desc.finish_grace_secs,
        metrics: metrics.and_then(|m| m.metrics).map(|m| MetricsOutput {
            status: m.status.to_string(),
            pid: m.pid,
            cpu: format!("{:.1}%", m.cpu_usage),
            memory: format!("{:.0} MB", m.memory_mb),
            restart_count: m.restart_count,
            pending_messages: m.pending_messages,
            broken_inputs: m.broken_inputs,
        }),
    };

    match format {
        OutputFormat::Table => print_table(&output),
        OutputFormat::Json => {
            println!("{}", serde_json::to_string_pretty(&output)?);
        }
    }

    Ok(())
}

fn fetch_node_metrics(
    session: &WsSession,
    dataflow_uuid: uuid::Uuid,
    node_id: &NodeId,
) -> eyre::Result<Option<NodeInfo>> {
    let reply = send_control_request(session, &ControlRequest::GetNodeInfo)?;
    let node_infos = expect_reply!(reply, NodeInfoList(infos))?;

    Ok(node_infos
        .into_iter()
        .find(|n| n.dataflow_id == dataflow_uuid && n.node_id == *node_id))
}

fn build_output_info(
    node_desc: &dora_message::descriptor::Node,
    descriptor: &Descriptor,
) -> Vec<OutputInfo> {
    // Use `node_topic_outputs` / `node_topic_inputs` rather than the raw
    // `node.outputs` / `node.inputs` fields: those are empty for `operator:` and
    // `operators:` nodes, which declare their ports under `operator.config` /
    // `operators[].config` (dora-rs/dora#2893). Reading the raw fields makes
    // `dora node info` report no inputs/outputs for an operator-backed node, and
    // omits every operator-node consumer from an output's subscriber list.
    node_topic_outputs(node_desc)
        .iter()
        .map(|output_id| {
            let mut subscribers = Vec::new();
            for other_node in &descriptor.nodes {
                for (input_id, input) in node_topic_inputs(other_node) {
                    if let InputMapping::User(user) = &input.mapping
                        && user.source == node_desc.id
                        && user.output == *output_id
                    {
                        subscribers.push(format!("{}/{}", other_node.id, input_id));
                    }
                }
            }
            OutputInfo {
                id: output_id.to_string(),
                subscribers,
            }
        })
        .collect()
}

fn build_input_info(node_desc: &dora_message::descriptor::Node) -> Vec<InputInfo> {
    node_topic_inputs(node_desc)
        .into_iter()
        .map(|(input_id, input)| InputInfo {
            id: input_id.to_string(),
            // Use `InputMapping`'s canonical `Display` rather than re-deriving the
            // source string here. The previous hand-rolled
            // `dora/timer/millis/{interval.as_millis()}` truncated any
            // sub-millisecond timer (e.g. a `dora/timer/hz/2000` = 500µs input)
            // to `dora/timer/millis/0`, a nonsensical zero-interval source;
            // `Display` picks the coarsest exact unit and round-trips through the
            // descriptor parser.
            source: input.mapping.to_string(),
        })
        .collect()
}

fn print_table(output: &NodeInfoOutput) {
    println!("{}: {}", "Node".bold(), output.node_id);
    println!("{}: {}", "Dataflow".bold(), output.dataflow);
    if let Some(name) = &output.name {
        println!("{}: {}", "Name".bold(), name);
    }
    if let Some(desc) = &output.description {
        println!("{}: {}", "Description".bold(), desc);
    }
    if let Some(path) = &output.path {
        println!("{}: {}", "Path".bold(), path);
    }
    println!();

    // Restart policy
    println!("{}", "Restart Policy:".bold());
    println!("  Policy: {}", output.restart_policy);
    println!("  Max restarts: {}", output.max_restarts);
    if let Some(delay) = output.restart_delay {
        println!("  Restart delay: {delay}s");
    }
    if let Some(timeout) = output.health_check_timeout {
        println!("  Health check timeout: {timeout}s");
    }
    if let Some(grace) = output.finish_grace_secs {
        println!("  Finish grace: {grace}s");
    }
    println!();

    // Inputs
    println!("{}", "Inputs:".bold());
    if output.inputs.is_empty() {
        println!("  <none>");
    } else {
        let mut tw = TabWriter::new(std::io::stdout().lock());
        for input in &output.inputs {
            let _ = tw.write_all(format!("  {}\t<- {}\n", input.id, input.source).as_bytes());
        }
        let _ = tw.flush();
    }
    println!();

    // Outputs
    println!("{}", "Outputs:".bold());
    if output.outputs.is_empty() {
        println!("  <none>");
    } else {
        for out in &output.outputs {
            if out.subscribers.is_empty() {
                println!("  {} (no subscribers)", out.id);
            } else {
                println!("  {} -> {}", out.id, out.subscribers.join(", "));
            }
        }
    }
    println!();

    // Runtime metrics
    println!("{}", "Runtime Metrics:".bold());
    if let Some(metrics) = &output.metrics {
        let mut tw = TabWriter::new(std::io::stdout().lock());
        let _ = tw.write_all(format!("  Status:\t{}\n", metrics.status).as_bytes());
        let _ = tw.write_all(format!("  PID:\t{}\n", metrics.pid).as_bytes());
        let _ = tw.write_all(format!("  CPU:\t{}\n", metrics.cpu).as_bytes());
        let _ = tw.write_all(format!("  Memory:\t{}\n", metrics.memory).as_bytes());
        let _ = tw.write_all(format!("  Restarts:\t{}\n", metrics.restart_count).as_bytes());
        let _ = tw.write_all(format!("  Pending msgs:\t{}\n", metrics.pending_messages).as_bytes());
        if !metrics.broken_inputs.is_empty() {
            let _ = tw.write_all(
                format!("  Broken inputs:\t{}\n", metrics.broken_inputs.join(", ")).as_bytes(),
            );
        }
        let _ = tw.flush();
    } else {
        println!("  <not running or metrics unavailable>");
    }
}

#[cfg(test)]
mod tests {
    use super::{build_input_info, build_output_info};
    use dora_message::descriptor::{Descriptor, Node};

    fn descriptor() -> Descriptor {
        serde_yaml::from_str(
            "\
nodes:
  - id: camera
    path: camera
    outputs:
      - frame
  - id: detect
    operator:
      python: detect.py
      inputs:
        image: camera/frame
      outputs:
        - boxes
  - id: sink
    path: sink
    inputs:
      boxes: detect/boxes
",
        )
        .expect("parse descriptor")
    }

    fn node<'a>(descriptor: &'a Descriptor, id: &str) -> &'a Node {
        descriptor
            .nodes
            .iter()
            .find(|n| n.id.to_string() == id)
            .expect("node in fixture")
    }

    // Regression: `dora node info` read the raw `node.inputs`, which is empty
    // for `operator:` nodes (their ports live under `operator.config`), so it
    // reported no inputs for operator-backed nodes (dora-rs/dora#2893).
    #[test]
    fn operator_node_inputs_are_reported() {
        let d = descriptor();
        let ids: Vec<_> = build_input_info(node(&d, "detect"))
            .into_iter()
            .map(|i| i.id)
            .collect();
        assert!(
            ids.iter().any(|id| id == "image"),
            "operator input missing: {ids:?}"
        );
    }

    // The output listing and its subscriber scan must likewise see operator
    // ports: `detect/boxes` is produced by an operator and consumed by `sink`.
    #[test]
    fn operator_node_outputs_and_subscribers_are_reported() {
        let d = descriptor();
        let outputs = build_output_info(node(&d, "detect"), &d);
        let boxes = outputs
            .iter()
            .find(|o| o.id == "boxes")
            .expect("operator output `boxes` should be listed");
        assert!(
            boxes.subscribers.iter().any(|s| s == "sink/boxes"),
            "operator-node subscriber missing: {:?}",
            boxes.subscribers
        );
    }
}