dora-message 1.0.0-rc1

use std::collections::BTreeMap;

pub use crate::common::{
    DataMessage, LogLevel, LogMessage, NodeError, NodeErrorCause, NodeExitStatus, Timestamped,
};
use crate::{
    BuildId, DataflowId, common::DaemonId, current_crate_version, id::NodeId, versions_compatible,
};

/// Per-dataflow status reported by a daemon after (re-)registration.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct DataflowStatusEntry {
    pub dataflow_id: uuid::Uuid,
    pub running_nodes: Vec<NodeId>,
}

#[allow(clippy::large_enum_variant)]
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub enum CoordinatorRequest {
    Register(DaemonRegisterRequest),
    Event {
        daemon_id: DaemonId,
        event: DaemonEvent,
    },
}

#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub struct DaemonRegisterRequest {
    dora_version: semver::Version,
    pub machine_id: Option<String>,
    #[serde(default)]
    pub labels: BTreeMap<String, String>,
    /// Whether this daemon understands hub-sourced git nodes — the `subdir`
    /// and `hub` provenance fields on a `GitSource` (spec §10.2, P2.10).
    ///
    /// `#[serde(default)]` makes this `false` for a daemon built before the
    /// field existed, so the coordinator can refuse to route a hub node to it
    /// with a clear error. This is the capability signal the `dora_version`
    /// gate cannot provide: during the `1.0.0-rc` window a pre-hub daemon and a
    /// hub-aware coordinator report the *same* version, so version alone can't
    /// distinguish them — an explicit flag can.
    #[serde(default)]
    supports_hub_sources: bool,
}

impl DaemonRegisterRequest {
    pub fn new(machine_id: Option<String>, labels: BTreeMap<String, String>) -> Self {
        Self {
            dora_version: current_crate_version(),
            machine_id,
            labels,
            // a daemon built from this crate understands hub git sources
            supports_hub_sources: true,
        }
    }

    /// Whether the registering daemon can build/spawn hub-sourced git nodes
    /// (carrying `subdir` / `hub` provenance).
    pub fn supports_hub_sources(&self) -> bool {
        self.supports_hub_sources
    }

    pub fn check_version(&self) -> Result<(), String> {
        let crate_version = current_crate_version();
        let specified_version = &self.dora_version;

        if versions_compatible(&crate_version, specified_version)? {
            Ok(())
        } else {
            // Direction-aware remediation: `versions_compatible` rejects both
            // older and newer daemons, so the fix differs. Upgrade whichever
            // side is older.
            let remedy = if *specified_version < crate_version {
                format!(
                    "upgrade the daemon to match the coordinator (e.g. \
                     `cargo install dora-cli --version {crate_version}`) — an older daemon \
                     also lacks newer wire features such as hub `subdir`/`hub:` node sources"
                )
            } else {
                format!(
                    "upgrade the coordinator to dora v{specified_version} (or run an older \
                     daemon) so both sides match"
                )
            };
            Err(format!(
                "version mismatch: this daemon runs dora v{specified_version} but the \
                 coordinator expects v{crate_version} — these dora versions are incompatible. \
                 {remedy}.",
            ))
        }
    }
}

#[cfg(test)]
mod register_version_tests {
    use super::*;

    #[test]
    fn current_version_is_compatible() {
        assert!(
            DaemonRegisterRequest::new(None, Default::default())
                .check_version()
                .is_ok()
        );
    }

    fn request_with_version(dora_version: semver::Version) -> DaemonRegisterRequest {
        DaemonRegisterRequest {
            dora_version,
            machine_id: None,
            labels: Default::default(),
            supports_hub_sources: true,
        }
    }

    #[test]
    fn incompatible_daemon_gets_direction_aware_upgrade_advice() {
        // `versions_compatible` rejects both older and newer daemons, so the
        // remediation must name the right side. (Cross-version only — a
        // *same-version* pre-hub daemon passes this gate, which is why hub
        // capability is signalled explicitly via `supports_hub_sources`.)
        let current = current_crate_version();

        // A NEWER daemon than the coordinator → upgrade the *coordinator*.
        let err = request_with_version(semver::Version::new(current.major + 1, 0, 0))
            .check_version()
            .expect_err("newer-major daemon must be rejected");
        assert!(err.contains("version mismatch"), "{err}");
        assert!(
            err.contains("upgrade the coordinator"),
            "newer daemon should advise upgrading the coordinator: {err}"
        );

        // An OLDER daemon than the coordinator → upgrade the *daemon*.
        let err = request_with_version(semver::Version::new(0, 1, 0))
            .check_version()
            .expect_err("older daemon must be rejected");
        assert!(
            err.contains("upgrade the daemon"),
            "older daemon should advise upgrading the daemon: {err}"
        );
    }

    #[test]
    fn hub_capability_is_advertised_by_current_daemons_and_defaults_off() {
        // A daemon built from this crate advertises hub support.
        assert!(DaemonRegisterRequest::new(None, Default::default()).supports_hub_sources());

        // A daemon built before the field existed sends a request without it;
        // `#[serde(default)]` must decode that as "no hub support" so the
        // coordinator refuses to route hub nodes to it. This is the same-version
        // gap the version check cannot catch.
        let legacy = r#"{"dora_version":"1.0.0-rc1","machine_id":null,"labels":{}}"#;
        let decoded: DaemonRegisterRequest = serde_json::from_str(legacy).unwrap();
        assert!(!decoded.supports_hub_sources());
    }
}

#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub enum DaemonEvent {
    BuildResult {
        build_id: BuildId,
        result: Result<(), String>,
    },
    SpawnResult {
        dataflow_id: DataflowId,
        result: Result<(), String>,
    },
    AllNodesReady {
        dataflow_id: DataflowId,
        exited_before_subscribe: Vec<NodeId>,
    },
    AllNodesFinished {
        dataflow_id: DataflowId,
        result: DataflowDaemonResult,
    },
    Heartbeat {
        #[serde(default)]
        ft_stats: Option<FaultToleranceSnapshot>,
    },
    /// Sent by the daemon after registration to report its current state.
    /// Enables coordinator-daemon reconciliation on reconnect.
    StatusReport {
        running_dataflows: Vec<DataflowStatusEntry>,
    },
    Log(LogMessage),
    Exit,
    NodeMetrics {
        dataflow_id: DataflowId,
        metrics: BTreeMap<NodeId, NodeMetrics>,
        #[serde(default)]
        network: Option<NetworkMetrics>,
    },
    /// Topic debug payload destined for one or more active CLI subscriptions.
    ///
    /// Daemon and coordinator are co-deployed from the same build, so this
    /// multi-subscriber shape is safe to evolve within the repository.
    TopicDebugData {
        dataflow_id: DataflowId,
        subscription_ids: Vec<uuid::Uuid>,
        payload: Vec<u8>,
    },
    /// Daemon acknowledges state catch-up through a given sequence number.
    StateCatchUpAck {
        dataflow_id: DataflowId,
        ack_sequence: u64,
    },
    /// Sent by the daemon when a node has exited and the daemon will NOT
    /// restart it (e.g. `dora node stop`, a node exiting under
    /// `restart_policy: Never`, or a final-failure cascade). The
    /// coordinator uses this to invalidate its cached `node_metrics`
    /// entry so `dora node list` reflects the actual state instead of
    /// the last-reported "Running" snapshot. Without this signal the
    /// daemon's metrics-snapshot loop simply stops including the dead
    /// node and the coordinator's cache is frozen at the last
    /// pre-exit values forever.
    NodeStopped {
        dataflow_id: DataflowId,
        node_id: NodeId,
        /// `true` if the daemon called `disable_restart()` before the
        /// exit (i.e. the `stop_single_node` / `restart_single_node`
        /// path triggered by `dora node stop`/`restart`). `false` for
        /// a final-failure exit under `restart_policy: Never` or a
        /// `max_restarts` exhaustion. The coordinator uses this to
        /// pick `NodeStatus::Stopped` vs `NodeStatus::Failed`, so a
        /// crash is not silently reported as a clean teardown (which
        /// would hide it from `dora doctor`).
        #[serde(default)]
        clean_stop: bool,
    },
}

/// Health status of a node
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
pub enum NodeStatus {
    #[default]
    Running,
    Restarting,
    /// One or more inputs have timed out (circuit breaker open)
    Degraded,
    Failed,
    /// Node was cleanly stopped (e.g. via `dora node stop`) and the
    /// process has exited. Distinguishes a deliberate teardown from a
    /// crash failure. Coordinator-side entries with this status are
    /// removed after `NODE_STOPPED_GRACE_PERIOD` so `dora node list`
    /// eventually stops showing zombies.
    Stopped,
}

impl std::fmt::Display for NodeStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            NodeStatus::Running => write!(f, "Running"),
            NodeStatus::Restarting => write!(f, "Restarting"),
            NodeStatus::Degraded => write!(f, "Degraded"),
            NodeStatus::Failed => write!(f, "Failed"),
            NodeStatus::Stopped => write!(f, "Stopped"),
        }
    }
}

/// Snapshot of daemon-level fault tolerance counters
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct FaultToleranceSnapshot {
    pub restarts: u64,
    pub health_check_kills: u64,
    pub input_timeouts: u64,
    pub circuit_breaker_recoveries: u64,
}

/// Resource metrics for a node process
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct NodeMetrics {
    /// Process ID
    pub pid: u32,
    /// CPU usage percentage (0-100 per core)
    pub cpu_usage: f32,
    /// Memory usage in bytes
    pub memory_bytes: u64,
    /// Disk read bytes per second (if available)
    pub disk_read_bytes: Option<u64>,
    /// Disk write bytes per second (if available)
    pub disk_write_bytes: Option<u64>,
    /// Number of times this node has been restarted
    #[serde(default)]
    pub restart_count: u32,
    /// Input IDs that have timed out (circuit breaker open)
    #[serde(default)]
    pub broken_inputs: Vec<String>,
    /// Current health status
    #[serde(default)]
    pub status: NodeStatus,
    /// Number of pending messages in the node's input queue
    #[serde(default)]
    pub pending_messages: u64,
}

/// Per-dataflow network I/O counters for cross-daemon Zenoh traffic.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct NetworkMetrics {
    pub bytes_sent: u64,
    pub bytes_received: u64,
    pub messages_sent: u64,
    pub messages_received: u64,
    #[serde(default)]
    pub publish_failures: u64,
}

#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
pub struct DataflowDaemonResult {
    pub timestamp: uhlc::Timestamp,
    pub node_results: BTreeMap<NodeId, Result<(), NodeError>>,
}

impl DataflowDaemonResult {
    pub fn is_ok(&self) -> bool {
        self.node_results.values().all(|r| r.is_ok())
    }
}

#[derive(Debug, serde::Deserialize, serde::Serialize)]
pub enum DaemonCoordinatorReply {
    TriggerBuildResult(Result<(), String>),
    TriggerSpawnResult(Result<(), String>),
    ReloadResult(Result<(), String>),
    StopResult(Result<(), String>),
    DestroyResult {
        result: Result<(), String>,
        #[serde(skip)]
        notify: Option<tokio::sync::oneshot::Sender<()>>,
    },
    Logs(Result<Vec<u8>, String>),
    /// Reply for `DaemonCoordinatorEvent::AddNode`. Previously the daemon
    /// returned `None` and the coordinator accepted any successful TCP
    /// response as proof that AddNode applied, even a `SetParamResult` or
    /// other unrelated reply — committing state for a node the daemon
    /// may have rejected (#1682). This variant lets the coordinator
    /// pattern-match a specific reply and forward daemon errors to the
    /// CLI instead of corrupting the dataflow state. Rescue of #1757.
    AddNodeResult(Result<(), String>),
    RestartNodeResult(Result<(), String>),
    StopNodeResult(Result<(), String>),
    RemoveNodeResult(Result<(), String>),
    /// Reply for `DaemonCoordinatorEvent::AddMapping`. Previously the daemon
    /// returned `None`, which the coordinator's WS layer skipped instead
    /// of forwarding as a reply, causing `send_and_receive` to time out
    /// after 30s with `daemon dispatch failed: timeout waiting for daemon
    /// WS reply`. Same bug class as #1682's AddNode silent-reply hole;
    /// applied to mappings here.
    AddMappingResult(Result<(), String>),
    /// Reply for `DaemonCoordinatorEvent::RemoveMapping`. See
    /// `AddMappingResult` doc for the silent-reply bug class.
    RemoveMappingResult(Result<(), String>),
    SetParamResult(Result<(), String>),
    DeleteParamResult(Result<(), String>),
    StartTopicDebugStreamResult(Result<(), String>),
    StopTopicDebugStreamResult(Result<(), String>),
}