mt_sea 0.7.0

pub mod client;
pub mod coordinator;
pub mod net;
pub mod ship;

pub use net::Qos;

// ---------------------------------------------------------------------------
// Timing constants — these values are interdependent.
// When changing one, check all others marked with the same group tag.
// ---------------------------------------------------------------------------

/// [heartbeat] How often nodes send heartbeat messages to keep their connection alive.
pub const HEARTBEAT_INTERVAL_MS: u64 = 400;

/// [heartbeat] Heartbeat is skipped if data was sent within this window.
/// Must be ≤ HEARTBEAT_INTERVAL_MS so a heartbeat is always sent before
/// DISCONNECT_TIMEOUT_MS elapses after the last real data.
pub const HEARTBEAT_SUPPRESS_MS: u64 = HEARTBEAT_INTERVAL_MS / 2;

/// [heartbeat] Coordinator drops a client after this long with no message.
/// Must be > HEARTBEAT_INTERVAL_MS + HEARTBEAT_SUPPRESS_MS to avoid false disconnects.
pub const DISCONNECT_TIMEOUT_MS: u64 = HEARTBEAT_INTERVAL_MS * 2; // 800 ms

/// [registration] How long to wait for a coordinator before giving up on registration.
pub const REGISTRATION_TIMEOUT_MS: u64 = 2000;

/// [peer-heartbeat] Consecutive ping failures before declaring a peer dead.
pub const PEER_DEAD_THRESHOLD: u32 = 3;

/// [coordinator] Last-resort timeout for coordinator-side per-client handler.
/// Nodes no longer heartbeat the coordinator directly; this only fires for
/// truly isolated or zombie clients that sent no packet for this long.
pub const COORD_CLIENT_IDLE_TIMEOUT_MS: u64 = 30_000;

/// [registration] How long to wait for an embedded coordinator to start before
/// retrying registration.
pub const COORDINATOR_STARTUP_WAIT_MS: u64 = 1000;

use mt_net::{ActionPlan, BagMsg, Rules, VariableHuman};

/// Initialize logging with zenoh logs filtered to warn level regardless of RUST_LOG setting.
/// Uses RUST_LOG env var for other crates, defaulting to `info` if not set.
pub fn init_logging() {
    use env_logger::Env;
    let env = Env::new().filter_or("RUST_LOG", "info");
    env_logger::Builder::from_env(env)
        .filter_module("zenoh", log::LevelFilter::Warn)
        .filter_module("zenoh::api::admin", log::LevelFilter::Off)
        .filter_module("zenoh::api::session", log::LevelFilter::Off)
        .filter_module("zenoh_transport", log::LevelFilter::Warn)
        .filter_module("zenoh_link", log::LevelFilter::Warn)
        .filter_module("zenoh_protocol", log::LevelFilter::Warn)
        .init();
}
use rkyv::{
    Archive, Deserialize, Serialize,
    api::high::{HighSerializer, HighValidator},
    bytecheck::CheckBytes,
    de::Pool,
    rancor::Strategy,
    ser::allocator::ArenaHandle,
    util::AlignedVec,
};

#[derive(Debug, Clone, PartialEq, Archive, Serialize, Deserialize, Hash, Eq, PartialOrd, Ord)]
pub enum ShipKind {
    Rat(String),
    Wind(String),
}

pub type ShipName = i128;

#[derive(Debug, Clone, Serialize, Deserialize, Archive, PartialEq, Eq)]
pub struct NetworkShipAddress {
    ip: [u8; 4],
    pub port: u16,
    ship: ShipName,
    pub kind: ShipKind,
    pub node_mode: net::Qos,
}

#[derive(Debug, Archive, Clone, Default, Serialize, Deserialize)]
pub enum Action {
    #[default]
    Sail,
    Shoot {
        target: Vec<NetworkShipAddress>,
        id: u32,
    },
    Catch {
        source: NetworkShipAddress,
        id: u32,
    },
}

#[derive(Clone, Debug)]
pub struct Variable {
    pub ship: ShipName,
    pub strategy: Option<Action>,
}

pub fn get_strategies(
    haystack: &Rules,
    rat_ship: &str,
    variable: String,
    indirect_parent_rat: Option<&str>,
) -> Vec<ActionPlan> {
    match haystack.raw().get(&variable) {
        None => vec![ActionPlan::default()],
        Some(plans) => {
            // directly because rule was set
            let directly = plans
                .iter()
                .filter(|plan| plan.ship == rat_ship)
                .filter_map(|el| el.strategy.clone())
                .collect::<Vec<_>>();

            // as other part of a rule
            let mut indirect = plans
                .iter()
                .filter(|plan| indirect_parent_rat.is_none_or(|parent_rat| plan.ship == parent_rat))
                .filter_map(|plan| match plan.strategy.as_ref()? {
                    ActionPlan::Sail => None,
                    ActionPlan::Shoot { target, id } => target
                        .iter()
                        .find(|shoot_target| *shoot_target == rat_ship)
                        .map(|_| ActionPlan::Catch {
                            source: plan.ship.clone(),
                            id: *id,
                        }),
                    ActionPlan::Catch { source, id } => {
                        if source == rat_ship {
                            Some(ActionPlan::Shoot {
                                target: vec![source.clone()],
                                id: *id,
                            })
                        } else {
                            None
                        }
                    }
                })
                .collect::<Vec<_>>();

            indirect.extend(directly);
            indirect
        }
    }
}

#[async_trait::async_trait]
pub trait Ship: Send + Sync + 'static {
    /// Indicate a trigger point and ask the link pilot what to do with the variable.
    async fn ask_for_action(&self, variable_name: &str) -> anyhow::Result<(Action, bool)>;

    // async fn wait_for_action(&self) -> anyhow::Result<crate::Action>;

    async fn wait_for_wind(&self) -> anyhow::Result<Vec<WindData>>;

    fn get_cannon(&self) -> &impl Cannon;
}

#[derive(Archive, Serialize, Deserialize, Debug, Clone, Copy, Default)]
pub enum VariableType {
    #[default]
    StaticOnly, // statically supported but no dynamic conversion implemented
    U8,
    I32,
    F32,
    F64,
}

impl From<u8> for VariableType {
    fn from(value: u8) -> Self {
        match value {
            1 => Self::U8,
            2 => Self::I32,
            3 => Self::F32,
            4 => Self::F64,
            _ => Self::default(),
        }
    }
}

impl From<VariableType> for u8 {
    fn from(value: VariableType) -> Self {
        match value {
            VariableType::StaticOnly => 0,
            VariableType::U8 => 1,
            VariableType::I32 => 2,
            VariableType::F32 => 3,
            VariableType::F64 => 4,
        }
    }
}

use rkyv::rancor::Error as RkyvError;

// Trait for types that can be Sent (Serialized).
// Requires Sized, Send, Sync, 'static, and the specific rkyv Serialize bound.
pub trait Sendable: Sized + Send + Sync + 'static
where
    Self: for<'b> Serialize<HighSerializer<AlignedVec, ArenaHandle<'b>, RkyvError>>,
    Self: Archive<
        Archived: for<'a> CheckBytes<HighValidator<'a, rkyv::rancor::Error>>
                      + Deserialize<Self, Strategy<Pool, rkyv::rancor::Error>>,
    >,
{
}
// Blanket implementation for Sendable. Any type meeting the bounds is Sendable.
impl<T> Sendable for T
where
    T: Sized + Send + Sync + 'static,
    T: for<'b> Serialize<HighSerializer<AlignedVec, ArenaHandle<'b>, RkyvError>>,
    T: Archive<
        Archived: for<'a> CheckBytes<HighValidator<'a, rkyv::rancor::Error>>
                      + Deserialize<T, Strategy<Pool, rkyv::rancor::Error>>,
    >,
{
}

#[async_trait::async_trait]
pub trait Cannon: Send + Sync + 'static {
    // Initialize a 1:1 connection to the target. Ports are shared using the sea network internally.

    /// Dump the data to the target.
    async fn shoot<'b, T: Sendable>(
        &self,
        targets: &'b [crate::NetworkShipAddress],
        id: u32,
        data: &T,
        variable_type: VariableType,
        variable_name: &str,
    ) -> anyhow::Result<()>;

    /// Catch the dumped data from the source.
    /// The returning Vec can contain previously missed entities of T from existing sync connections.
    /// The first item of T is the newest, followed by incremental older ones.
    async fn catch<T: Sendable>(&self, id: u32) -> anyhow::Result<Vec<T>>;

    async fn catch_dyn(&self, id: u32) -> anyhow::Result<Vec<(String, VariableType, String)>>;
}

#[derive(Clone, Debug, Default, Copy, Archive, Serialize, Deserialize, PartialEq)]
pub struct TimeMsg {
    pub sec: i32,
    pub nanosec: u32,
}

#[derive(Clone, Debug, Default, PartialEq, Archive, Serialize, Deserialize)]
pub struct Header {
    pub seq: u32,
    pub stamp: TimeMsg,
    pub frame_id: String,
}

pub type WindData = BagMsg;

#[async_trait::async_trait]
pub trait Coordinator: Send + Sync + 'static {
    async fn rat_action_request_queue(
        &self,
        ship: String,
    ) -> anyhow::Result<tokio::sync::broadcast::Receiver<String>>;

    async fn blow_wind(&self, ship: String, data: Vec<WindData>) -> anyhow::Result<()>;

    async fn rat_action_send(
        &self,
        ship: String,
        variable: String,
        action: ActionPlan,
        lock_until_ack: bool,
        best_effort: bool,
    ) -> anyhow::Result<()>;

    /// Push updated routes for `variable` to all ships currently involved in it.
    /// Called after topology changes (register / disconnect / PeerDead).
    async fn push_routes_for_var(&self, variable: &str, rules: &Rules) -> anyhow::Result<()>;
}