volli-core 0.1.11

#![cfg_attr(test, allow(unused_crate_dependencies))]

use crate::env_config::config_dir_env;
use serde::{Deserialize, Serialize};
use std::path::PathBuf;

/// Return the base directory for all configuration files.
pub fn config_dir() -> PathBuf {
    if let Some(dir) = config_dir_env() {
        PathBuf::from(dir)
    } else if let Some(mut path) = dirs_next::config_dir() {
        path.push("volli");
        path
    } else {
        dirs_next::home_dir()
            .unwrap_or_else(|| PathBuf::from("."))
            .join(".volli")
    }
}

/// Default port used for TCP connections.
pub const DEFAULT_TCP_PORT: u16 = 4242;
/// Default port used for QUIC connections.
pub const DEFAULT_QUIC_PORT: u16 = 4242;
pub mod codec;
pub mod env_config;
pub mod handshake;
pub mod namegen;
pub mod peer_db;
pub mod peer_store;
pub mod profile;
pub mod token;
pub mod util;
pub mod worker;

pub use env_config::{
    ConfigDirGuard, ConfigGuard, EnvironmentConfig, env_config, override_config_dir,
    override_env_config, override_env_config_patch,
};
pub use worker::{Protocol, Role, WorkerConfig};

/// Basic message type used between worker and server.
#[derive(Debug, Serialize, Deserialize)]
pub enum Message {
    Ping {
        version: u64,
    },
    Pong {
        mac: String,
        version: u64,
    },
    Auth {
        token: String,
        #[serde(default)]
        worker_id: Option<String>,
        #[serde(default)]
        worker_name: Option<String>,
    },
    Join {
        token: String,
    },
    AuthOk,
    AuthErr,
    TokenRefreshRequest {
        token: String,
    },
    TokenRefreshOk {
        token: String,
    },
    TokenRefreshErr {
        reason: String,
    },
    Hello {
        manager_id: String,
        nonce: [u8; 32],
        sig: Vec<u8>,
    },
    Welcome {
        manager_id: String,
        nonce: [u8; 32],
        sig: Vec<u8>,
    },
    ClusterKey {
        ver: u32,
        csk: [u8; 32],
    },
    /// Join response with cluster key and peer info
    JoinResponse {
        ver: u32,
        csk: [u8; 32],
        peer: Box<ManagerPeerEntry>,
    },
    /// Client authentication using cluster keys
    ClientAuth {
        token: String,
    },
    /// Client command request for distributed execution
    ClientCommandRequest {
        request_id: String,
        command: String,
        args: Vec<String>,
        target: String, // Serialized CommandTarget
        timeout_secs: u64,
        options: String, // Serialized CommandOptions
    },
    /// Client command response
    ClientCommandResponse {
        request_id: String,
        worker_id: String,
        worker_name: Option<String>,
        success: bool,
        duration_millis: u64,
        output: String,
    },
    /// Client streaming: header frame (manager -> client)
    ClientCommandHeader {
        request_id: String,
        worker_id: String,
        worker_name: Option<String>,
        payload: Vec<u8>,
    },
    /// Client streaming: progress frame (manager -> client)
    ClientCommandStream {
        request_id: String,
        worker_id: String,
        worker_name: Option<String>,
        payload: Vec<u8>,
    },
    /// Client streaming: footer/completion frame per worker (manager -> client)
    ClientCommandFooter {
        request_id: String,
        worker_id: String,
        worker_name: Option<String>,
        payload: Vec<u8>,
        duration_millis: u64,
        success: bool,
    },
    /// Client command completion
    ClientCommandComplete {
        request_id: String,
        total_workers: u32,
    },
    /// Client command error
    ClientCommandError {
        request_id: String,
        error: String,
    },
    /// Client command cancel request
    ClientCommandCancel {
        request_id: String,
    },
    /// Worker command request (manager -> worker)
    WorkerCommandRequest {
        request_id: String,
        command: String,
        args: Vec<String>,
        timeout_secs: u64,
        options: String, // Serialized CommandOptions
    },
    /// Worker command response with results (worker -> manager)
    WorkerCommandResponse {
        request_id: String,
        worker_id: String,
        success: bool,
        duration_millis: u64,
        output: String,
    },
    /// Worker streaming header (binary payload of CommandHeader)
    WorkerCommandHeader {
        request_id: String,
        worker_id: String,
        payload: Vec<u8>,
    },
    /// Worker streaming frame (binary payload of CommandStream)
    WorkerCommandStream {
        request_id: String,
        worker_id: String,
        payload: Vec<u8>,
    },
    /// Worker streaming footer (binary payload of CommandFooter)
    WorkerCommandFooter {
        request_id: String,
        worker_id: String,
        payload: Vec<u8>,
        duration_millis: u64,
        success: bool,
    },
    /// Worker command error (worker -> manager)
    WorkerCommandError {
        request_id: String,
        error: String,
    },
    /// Cancel a running command on a worker (manager -> worker)
    WorkerCommandCancel {
        request_id: String,
    },
    /// Manager heartbeat and peer list gossip.
    ///
    /// `version` is a monotonically increasing counter for the sender's
    /// peer table. `peers` contains the full list of known peers when
    /// `version` has advanced since the last heartbeat; otherwise it may
    /// be empty to indicate no changes.
    Announce {
        meta: Box<ManagerPeerEntry>,
        version: u64,
        peers: Vec<ManagerPeerEntry>,
        /// List of known workers when `version` has advanced. Empty otherwise.
        #[serde(default)]
        workers: Vec<WorkerEntry>,
    },
    /// Worker indicates it is intentionally closing this session (e.g., after migration)
    Goodbye,
}

#[cfg(test)]
mod message_tests {
    use super::*;
    use crate::codec::Codec;

    #[test]
    fn bincode_announce_roundtrip() {
        let meta = ManagerPeerEntry {
            manager_id: "m1".into(),
            manager_name: "m1".into(),
            tenant: "t".into(),
            cluster: "c".into(),
            host: "h1".into(),
            tcp_port: 1,
            quic_port: 1,
            pub_fp: String::new(),
            csk_ver: 0,
            tls_cert: "cert1".into(),
            tls_fp: "fp1".into(),
            health: None,
        };
        let extra = ManagerPeerEntry {
            host: "h2".into(),
            tcp_port: 2,
            quic_port: 2,
            tls_cert: "cert2".into(),
            tls_fp: "fp2".into(),
            manager_id: "m2".into(),
            manager_name: "m2".into(),
            tenant: "t".into(),
            cluster: "c".into(),
            pub_fp: String::new(),
            csk_ver: 0,
            health: None,
        };
        let msg = Message::Announce {
            meta: Box::new(meta),
            version: 1,
            peers: vec![extra],
            workers: Vec::new(),
        };
        let bytes = crate::codec::JsonCodec::encode(&msg);
        let decoded = crate::codec::JsonCodec::decode(&bytes).unwrap();
        match decoded {
            Message::Announce { version, peers, .. } => {
                assert_eq!(version, 1);
                assert_eq!(peers.len(), 1);
            }
            other => panic!("unexpected: {:?}", other),
        }
    }
}

#[cfg(test)]
mod bincode_smoke {
    use super::*;
    use crate::codec::Codec;

    #[test]
    fn manager_peer_entry_bincode_roundtrip() {
        let e = ManagerPeerEntry {
            manager_id: "m1".into(),
            manager_name: "m1".into(),
            tenant: "t".into(),
            cluster: "c".into(),
            host: "h1".into(),
            tcp_port: 1,
            quic_port: 1,
            pub_fp: String::new(),
            csk_ver: 0,
            tls_cert: "cert1".into(),
            tls_fp: "fp1".into(),
            health: None,
        };
        let msg = Message::Announce {
            meta: Box::new(e.clone()),
            version: 0,
            peers: vec![e.clone()],
            workers: Vec::new(),
        };
        let bytes = crate::codec::BincodeCodec::encode(&msg);
        let decoded = crate::codec::BincodeCodec::decode(&bytes).unwrap();
        match decoded {
            Message::Announce { version, peers, .. } => {
                assert_eq!(version, 0);
                assert_eq!(peers.len(), 1);
            }
            other => panic!("unexpected: {:?}", other),
        }
    }

    #[derive(Debug, Serialize, Deserialize, PartialEq)]
    struct Wrapper {
        meta: ManagerPeerEntry,
        peers: Vec<ManagerPeerEntry>,
    }

    #[test]
    fn wrapper_bincode_roundtrip() {
        let e = ManagerPeerEntry {
            manager_id: "m1".into(),
            manager_name: "m1".into(),
            tenant: "t".into(),
            cluster: "c".into(),
            host: "h1".into(),
            tcp_port: 1,
            quic_port: 1,
            pub_fp: String::new(),
            csk_ver: 0,
            tls_cert: "cert1".into(),
            tls_fp: "fp1".into(),
            health: None,
        };
        let w = Wrapper {
            meta: e.clone(),
            peers: vec![e.clone()],
        };
        let bytes = crate::codec::BincodeCodec::encode(&Message::Ping { version: 42 });
        let _ = bytes.len();
        // Also ensure serde serialization of Wrapper works via serde_json
        let json = serde_json::to_string(&w).unwrap();
        let w2: Wrapper = serde_json::from_str(&json).unwrap();
        assert_eq!(w.peers.len(), w2.peers.len());
    }

    #[test]
    fn peer_entry_alone_bincode_roundtrip() {
        let e = ManagerPeerEntry {
            manager_id: "m1".into(),
            manager_name: "m1".into(),
            tenant: "t".into(),
            cluster: "c".into(),
            host: "h1".into(),
            tcp_port: 1,
            quic_port: 1,
            pub_fp: String::new(),
            csk_ver: 0,
            tls_cert: "cert1".into(),
            tls_fp: "fp1".into(),
            health: None,
        };
        let bytes = crate::codec::BincodeCodec::encode(&Message::Hello {
            manager_id: "x".into(),
            nonce: [0u8; 32],
            sig: vec![],
        });
        let _ = bytes.len();
        let v = vec![e.clone(), e];
        // Serialize vec<ManagerPeerEntry> with serde bincode directly to confirm
        let bv = bincode::serialize(&v).unwrap();
        let v2: Vec<ManagerPeerEntry> = bincode::deserialize(&bv).unwrap();
        assert_eq!(v2.len(), 2);
    }

    #[test]
    fn sanity_bincode_vec_string() {
        let v = vec!["a".to_string(), "b".to_string()];
        let bytes = bincode::serialize(&v).unwrap();
        let vv: Vec<String> = bincode::deserialize(&bytes).unwrap();
        assert_eq!(vv.len(), 2);
    }
}

/// Metadata advertised by managers in heartbeats and peer storage
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
pub struct ManagerPeerEntry {
    pub manager_id: String,
    pub manager_name: String,
    pub tenant: String,
    pub cluster: String,
    pub host: String,
    pub tcp_port: u16,
    pub quic_port: u16,
    pub pub_fp: String,
    pub csk_ver: u32,
    pub tls_cert: String,
    pub tls_fp: String,

    // Optional health metrics; always serialized for bincode compatibility
    #[serde(default)]
    pub health: Option<HealthMetrics>,
}

/// Health and load metrics for manager nodes
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct HealthMetrics {
    pub health_score: f32,        // 0.0-1.0, higher is better
    pub load_percentage: f32,     // 0.0-100.0, current utilization
    pub max_workers: Option<u32>, // None = unlimited
    pub current_workers: u32,     // Current worker count
    pub avg_cpu: Option<f32>,     // Optional system metrics
    pub avg_memory: Option<f32>,  // Optional system metrics
    pub last_health_update: u64,  // Unix timestamp
}

impl Default for HealthMetrics {
    fn default() -> Self {
        Self {
            health_score: 1.0, // Start with perfect health
            load_percentage: 0.0,
            max_workers: None, // Unlimited by default
            current_workers: 0,
            avg_cpu: None,
            avg_memory: None,
            last_health_update: 0,
        }
    }
}

impl ManagerPeerEntry {
    /// Calculate load factor based on reported load percentage (0.0 ..= 1.0)
    pub fn calculate_load_factor(&self) -> f32 {
        match &self.health {
            Some(health) => (health.load_percentage / 100.0).clamp(0.0, 1.0),
            None => 0.0,
        }
    }

    /// Calculate weighted score combining health_score, load_percentage, and RTT
    pub fn weighted_score(&self, rtt_ms: Option<f64>) -> f64 {
        match &self.health {
            Some(health) => {
                let health_factor = health.health_score as f64;
                // Use reported load percentage rather than worker counts to avoid leaking worker totals
                let load_factor = 1.0 - self.calculate_load_factor() as f64;
                let rtt_factor = rtt_ms.map(|rtt| 1.0 / (1.0 + rtt / 100.0)).unwrap_or(1.0);

                // Weighted combination: 40% health, 40% load, 20% RTT
                0.4 * health_factor + 0.4 * load_factor + 0.2 * rtt_factor
            }
            None => 0.5, // Default score when no health data available
        }
    }
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
pub enum ConnectionState {
    #[default]
    Inactive,
    Client,
    Server,
}

/// Worker presence advertised across the mesh.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct WorkerEntry {
    pub worker_id: String,
    pub manager_id: String,
    #[serde(default)]
    pub worker_name: Option<String>,
    #[serde(default)]
    pub last_seen: Option<u64>,
    #[serde(default)]
    pub connected_since: Option<u64>,
    #[serde(default)]
    pub disconnected_at: Option<u64>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct AliveEntry {
    pub meta: ManagerPeerEntry,
    pub last_seen: u64,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct PingRequest {
    pub host: String,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct PingResponse {
    pub success: bool,
    pub latency_ms: u32,
}