mt_coord 0.9.0

use std::collections::{HashMap, HashSet};

use log::{debug, error, info, warn};
use mt_sea::net::{self, Packet};
use mt_sea::{coordinator::CoordinatorImpl, net::PacketKind};

use mt_net::{ActionPlan, COMPARE_NODE_NAME, RatPubRegisterKind, Rules, VariableHuman};
use mt_sea::{Coordinator, DISCONNECT_TIMEOUT_MS, Qos, ShipKind, WindData};

const EMBEDDED_COORDINATOR_READY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

fn take_subscription_override_vars(
    overrides: &mut HashMap<(String, String), Qos>,
    disconnected_ship: &str,
) -> Vec<String> {
    let vars = overrides
        .keys()
        .filter(|(ship, _)| ship == disconnected_ship)
        .map(|(_, var)| var.clone())
        .collect::<Vec<_>>();
    overrides.retain(|(ship, _), _| ship != disconnected_ship);
    vars
}

#[derive(Debug)]
enum MinotTask {
    AppendRule {
        variable: String,
        commands: Vec<VariableHuman>,
    },
    LockNext {
        unlock_first: bool,
    },
    RulesClear,
    SendRatAction {
        ship_name: String,
        var_name: String,
        data: ActionPlan,
    },
    WindDynamicVarReq(String),
    Unlock,
    RegisterShipAtVar {
        ship: String,
        var: String,
        kind: net::RatPubRegisterKind,
        node_mode: Qos,
    },
    UnregisterShipAtVar {
        ship: String,
        var: String,
        kind: net::RatPubRegisterKind,
    },
    /// A peer monitor on another node declared this ship dead.
    PeerDead {
        ship: String,
    },
    /// Push updated routes for a variable to all involved ships (async, runs in coord_rx).
    PushRoutesForVar {
        var: String,
    },
    /// Forward a Torpedo to all connected ships and then stop this coordinator.
    BroadcastTorpedo {
        dead_clients: Vec<String>,
    },
}

#[derive(Debug)]
enum WindTaskKind {
    Fix(Vec<WindData>),
    Dynamic,
}

#[derive(Debug)]
struct WindTask {
    kind: WindTaskKind,
    already_seen: bool,
}

pub fn run_coordinator(
    locked_start: bool,
    clients: HashSet<String>,
    rules: Rules,
    torpedo_tx: Option<tokio::sync::mpsc::Sender<()>>,
) {
    run_coordinator_with_ready(locked_start, clients, rules, torpedo_tx, None, None);
}

fn run_coordinator_with_ready(
    locked_start: bool,
    clients: HashSet<String>,
    rules: Rules,
    torpedo_tx: Option<tokio::sync::mpsc::Sender<()>>,
    ready_tx: Option<tokio::sync::oneshot::Sender<Result<(), String>>>,
    startup_failed_tx: Option<tokio::sync::oneshot::Sender<()>>,
) {
    let coordinator_shutting_down = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let coordinator_shutting_down_in_coord = std::sync::Arc::clone(&coordinator_shutting_down);
    #[cfg(feature = "embed-scope")]
    let coordinator_shutting_down_in_scope = std::sync::Arc::clone(&coordinator_shutting_down);

    // Zenoh handles coordinator discovery and port management
    let clients_wait_for_ack = std::sync::Arc::new(std::sync::RwLock::new(!clients.is_empty()));
    let rules = std::sync::Arc::new(std::sync::RwLock::new(rules));
    let rules_changer = std::sync::Arc::clone(&rules);

    let winds_var = std::sync::Arc::new(std::sync::RwLock::new(
        HashMap::<String, Vec<WindTask>>::new(),
    ));
    let winds_changer = std::sync::Arc::clone(&winds_var);

    // Bounded queue to apply backpressure when wind forwarding is slower than producers.
    let (wind_tx, mut wind_rx) = tokio::sync::mpsc::channel::<Vec<WindData>>(64);
    let wind_clients: std::sync::Arc<std::sync::RwLock<HashSet<String>>> =
        std::sync::Arc::new(std::sync::RwLock::new(HashSet::new()));

    // Shared set of currently connected client names (for Sonar responses)
    let connected_clients: std::sync::Arc<std::sync::RwLock<HashSet<String>>> =
        std::sync::Arc::new(std::sync::RwLock::new(HashSet::new()));

    // Delivery mode per connected client. Read from sync contexts that cannot
    // touch the async `rat_qs` lock, so the mode is mirrored here rather than
    // re-derived. Absent means the client is gone.
    let client_qos: std::sync::Arc<std::sync::RwLock<HashMap<String, Qos>>> =
        std::sync::Arc::new(std::sync::RwLock::new(HashMap::new()));

    // Task for centralized coordinator task work that is generated by the tasks implemented after this
    let (coord_tx, mut coord_rx) = tokio::sync::mpsc::unbounded_channel::<MinotTask>();
    let coord_tx_new_client = coord_tx.clone();
    let cwa_write = std::sync::Arc::clone(&clients_wait_for_ack);
    tokio::spawn(async move {
        let coordinator = match CoordinatorImpl::new(None, clients_wait_for_ack).await {
            Ok(coordinator) => std::sync::Arc::new(coordinator),
            Err(error) => {
                if let Some(ready_tx) = ready_tx {
                    let _ = ready_tx.send(Err(error.to_string()));
                }
                if let Some(startup_failed_tx) = startup_failed_tx {
                    let _ = startup_failed_tx.send(());
                }
                error!("Failed to start coordinator: {error}");
                return;
            }
        };
        if let Some(ready_tx) = ready_tx {
            let _ = ready_tx.send(Ok(()));
        }
        #[cfg(feature = "embed-scope")]
        {
            let coordinator_shutting_down_for_scope =
                std::sync::Arc::clone(&coordinator_shutting_down_in_scope);
            tokio::spawn(async move {
                if let Err(e) = mt_scope::Scope::create(mt_scope::ScopeConfig {
                    name: "minot_scope".to_string(),
                    mode: mt_scope::Qos::Reliable,
                })
                .await
                {
                    if coordinator_shutting_down_for_scope.load(std::sync::atomic::Ordering::SeqCst)
                    {
                        debug!("Embedded scope stopped during coordinator shutdown: {}", e);
                    } else if e.to_string().contains("Receiver task failed to start") {
                        debug!(
                            "Embedded scope stopped (likely due to program shutdown): {}",
                            e
                        );
                    } else {
                        error!("Embedded scope exited with error: {}", e);
                    }
                }
            });
        }
        let coordinator_for_wind_dispatch = std::sync::Arc::clone(&coordinator);
        let wind_clients_for_dispatch = std::sync::Arc::clone(&wind_clients);

        tokio::spawn(async move {
            let mut waiting_for_wind = false;
            while let Some(data) = wind_rx.recv().await {
                let targets: Vec<String> = loop {
                    let current = wind_clients_for_dispatch
                        .read()
                        .unwrap()
                        .iter()
                        .cloned()
                        .collect::<Vec<_>>();
                    if !current.is_empty() {
                        if waiting_for_wind {
                            info!("Wind client connected; resuming queued wind forwarding.");
                            waiting_for_wind = false;
                        }
                        break current;
                    }
                    if !waiting_for_wind {
                        warn!(
                            "Received wind data but no connected winds; pausing forwarding until a wind client connects."
                        );
                        waiting_for_wind = true;
                    }
                    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
                };

                for ship_name in targets {
                    let ship_known = coordinator_for_wind_dispatch
                        .rat_qs
                        .read()
                        .await
                        .contains_key(&ship_name);
                    if !ship_known {
                        warn!(
                            "Skipping stale wind target '{}' while forwarding wind data.",
                            ship_name
                        );
                        wind_clients_for_dispatch
                            .write()
                            .unwrap()
                            .remove(&ship_name);
                        continue;
                    }
                    if let Err(e) = coordinator_for_wind_dispatch
                        .blow_wind(ship_name.clone(), data.clone())
                        .await
                    {
                        error!("Error while blowing wind to {}: {}", ship_name, e);
                        wind_clients_for_dispatch
                            .write()
                            .unwrap()
                            .remove(&ship_name);
                    }
                }
            }
            debug!("wind_rx closed");
        });

        // unlock clients to send us stuff when all our expected clients are connected
        let total_clients_check_rats = std::sync::Arc::clone(&coordinator.rat_qs);
        let new_client_notify = coordinator.new_client_notify.clone();
        let (clients_collected_tx, mut clients_collected_rx) = tokio::sync::mpsc::channel(1);
        tokio::spawn(async move {
            debug!("waiting for all clients to be connected");
            CoordinatorImpl::ensure_clients_connected(
                total_clients_check_rats.clone(),
                clients,
                new_client_notify,
            )
            .await;
            debug!("all clients connected, sending ack to clients.");
            clients_collected_tx.send(()).await.unwrap();
            let clients = total_clients_check_rats.read().await;
            let ack = net::Packet {
                header: net::Header::default(),
                data: net::PacketKind::Acknowledge,
            };
            for (_client_name, client) in clients.iter() {
                if let Err(e) = client.sender.send(ack.clone()).await {
                    error!("Could not send ack to unlock client: {e}");
                }
            }
            // allow other clients to join after all predefined clients are connected
            let mut cwa = cwa_write.write().unwrap();
            *cwa = false;
        });

        let mut new_clients_chan = coordinator.sea.network_clients_chan.subscribe();
        let minot_client_connected = std::sync::Arc::new(std::sync::RwLock::new(false));
        let tasks_minot_connect = std::sync::Arc::clone(&minot_client_connected);
        let rules_change_for_disconnect = std::sync::Arc::clone(&rules_changer);
        let sub_overrides_for_disconnect = std::sync::Arc::clone(&coordinator.sub_qos_overrides);
        let cc_for_spawn = std::sync::Arc::clone(&connected_clients);
        let qos_for_spawn = std::sync::Arc::clone(&client_qos);
        let wind_clients_for_spawn = std::sync::Arc::clone(&wind_clients);
        // Keep a clone for the coord_rx loop; torpedo_tx itself is moved into the inner spawn.
        let torpedo_tx_for_coord = torpedo_tx.clone();
        let client_handlers: std::sync::Arc<
            std::sync::Mutex<HashMap<String, tokio::task::AbortHandle>>,
        > = std::sync::Arc::new(std::sync::Mutex::new(HashMap::new()));
        tokio::spawn(async move {
            loop {
                match new_clients_chan.recv().await {
                    Ok(client) => {
                        let wind_tx_inner = wind_tx.clone();
                        match client.name {
                            ShipKind::Rat(name) => {
                                if name == COMPARE_NODE_NAME {
                                    let mut lock = tasks_minot_connect.write().unwrap();
                                    *lock = true;
                                    // New run starting: clear stale wind state so setup packets
                                    // from this run don't accumulate on top of previous ones.
                                    winds_changer.write().unwrap().clear();
                                }
                                // Abort any existing handler for this client name so the old
                                // zenoh subscriber (keyed by name) also exits and stops
                                // delivering stale packets to the old handler.
                                if let Some(old) = client_handlers.lock().unwrap().remove(&name) {
                                    old.abort();
                                }
                                cc_for_spawn.write().unwrap().insert(name.clone());
                                qos_for_spawn
                                    .write()
                                    .unwrap()
                                    .insert(name.clone(), client.node_mode);
                                let mut client_news = client.recv.subscribe();
                                let inner_name = name.clone();
                                let rat_rules = std::sync::Arc::clone(&rules);
                                let rat_coord_tx = coord_tx_new_client.clone();
                                let winds_inner = std::sync::Arc::clone(&winds_changer);
                                let rules_change_for_disconnect_inner =
                                    std::sync::Arc::clone(&rules_change_for_disconnect);
                                let sub_overrides_for_disconnect_inner =
                                    std::sync::Arc::clone(&sub_overrides_for_disconnect);
                                let connected_clients_inner = std::sync::Arc::clone(&cc_for_spawn);
                                let client_qos_inner = std::sync::Arc::clone(&qos_for_spawn);
                                let handlers_for_task = std::sync::Arc::clone(&client_handlers);
                                let name_for_task = name.clone();
                                // spawn task to handle variable requests for this rat
                                let handle = tokio::spawn(async move {
                                    let minot_compare_name = COMPARE_NODE_NAME.to_string();
                                    match inner_name.as_str() {
                                        COMPARE_NODE_NAME => loop {
                                            match tokio::time::timeout(
                                                tokio::time::Duration::from_millis(DISCONNECT_TIMEOUT_MS),
                                                client_news.recv(),
                                            )
                                            .await
                                            {
                                                Ok(Ok((packet, _))) => match packet.data {
                                                    PacketKind::RuleAppend {
                                                        variable,
                                                        commands,
                                                    } => {
                                                        rat_coord_tx
                                                            .send(MinotTask::AppendRule {
                                                                variable,
                                                                commands,
                                                            })
                                                            .unwrap();
                                                    }
                                                    PacketKind::RulesClear => {
                                                        rat_coord_tx
                                                            .send(MinotTask::RulesClear)
                                                            .unwrap();
                                                    }
                                                    PacketKind::LockNext { unlock_first } => {
                                                        rat_coord_tx
                                                            .send(MinotTask::LockNext {
                                                                unlock_first,
                                                            })
                                                            .unwrap();
                                                    }
                                                    PacketKind::Unlock => {
                                                        rat_coord_tx
                                                            .send(MinotTask::Unlock)
                                                            .unwrap();
                                                    }
                                                    PacketKind::WindDynamic(var) => {
                                                        let mut w = winds_inner.write().unwrap();
                                                        match w.get_mut(&var) {
                                                            Some(wts) => {
                                                                if !wts.iter().any(|task| {
                                                                    matches!(task.kind, WindTaskKind::Dynamic)
                                                                }) {
                                                                    wts.push(WindTask {
                                                                        already_seen: false,
                                                                        kind: WindTaskKind::Dynamic,
                                                                    });
                                                                }
                                                            }
                                                            None => {
                                                                w.insert(
                                                                    var,
                                                                    vec![WindTask {
                                                                        already_seen: false,
                                                                        kind: WindTaskKind::Dynamic,
                                                                    }],
                                                                );
                                                            }
                                                        }
                                                    }
                                                    PacketKind::Wind(bwd) => {
                                                        let mut send_now =
                                                            Vec::with_capacity(bwd.len());
                                                        for wd in bwd {
                                                            match wd.at_var {
                                                                Some(var) => {
                                                                    let mut w = winds_inner
                                                                        .write()
                                                                        .unwrap();
                                                                    match w.get_mut(&var) {
                                                                        Some(wt) => {
                                                                            wt.push(WindTask { kind: WindTaskKind::Fix(vec![wd.data]), already_seen: false });
                                                                        }
                                                                        None => {
                                                                            w.insert(var,
                                                                            vec![WindTask {
                                                                                already_seen: false,
                                                                                kind: WindTaskKind::Fix(vec![wd.data])
                                                                            }]);
                                                                        }
                                                                    }
                                                                }
                                                                None => {
                                                                    send_now.push(wd.data);
                                                                }
                                                            }
                                                        }

                                                        if !send_now.is_empty() {
                                                            if let Err(e) = wind_tx_inner.send(send_now).await {
                                                                error!(
                                                                    "Could not enqueue wind data from Minot TUI for forwarding: {}",
                                                                    e
                                                                );
                                                            }
                                                        }
                                                    }

                                                    PacketKind::Heartbeat => {
                                                        debug!(
                                                            "Received heartbeat from Minot TUI"
                                                        );
                                                    }
                                                    PacketKind::PeerDead { ship } => {
                                                        rat_coord_tx
                                                            .send(MinotTask::PeerDead { ship })
                                                            .unwrap();
                                                    }
                                                    _ => {
                                                        warn!(
                                                            "Received unexpected packet from Minot TUI"
                                                        );
                                                    }
                                                },
                                                Ok(Err(
                                                    tokio::sync::broadcast::error::RecvError::Lagged(
                                                        n,
                                                    ),
                                                )) => {
                                                    warn!(
                                                        "Minot TUI receiver lagged by {} messages",
                                                        n
                                                    );
                                                }
                                                Ok(Err(
                                                    tokio::sync::broadcast::error::RecvError::Closed,
                                                )) => {
                                                    info!("Minot TUI disconnected");
                                                    connected_clients_inner
                                                        .write()
                                                        .unwrap()
                                                        .remove(COMPARE_NODE_NAME);
                                                    return;
                                                }
                                                Err(_elapsed) => {
                                                    info!(
                                                        "Minot TUI timed out (no heartbeat), treating as disconnected"
                                                    );
                                                    connected_clients_inner
                                                        .write()
                                                        .unwrap()
                                                        .remove(COMPARE_NODE_NAME);
                                                    return;
                                                }
                                            }
                                        },
                                        _ => {
                                            loop {
                                                match tokio::time::timeout(
                                                    tokio::time::Duration::from_millis(
                                                        DISCONNECT_TIMEOUT_MS,
                                                    ),
                                                    client_news.recv(),
                                                )
                                                .await
                                                {
                                                    Ok(Ok((packet, _))) => {
                                                        match packet.data {
                                                            PacketKind::RegisterShipAtVar {
                                                                ship,
                                                                var,
                                                                kind,
                                                                node_mode,
                                                            } => {
                                                                // Acknowledge is sent by coord_rx after validation
                                                                rat_coord_tx.send(MinotTask::RegisterShipAtVar {ship,var,kind,node_mode},).unwrap();
                                                            }
                                                            PacketKind::UnregisterShipAtVar {
                                                                ship,
                                                                var,
                                                                kind,
                                                            } => {
                                                                rat_coord_tx.send(MinotTask::UnregisterShipAtVar {ship,var,kind},).unwrap();
                                                            }
                                                            PacketKind::VariableTaskRequest(
                                                                variable,
                                                            ) => {
                                                                let (minot_actions, mut my_actions) = {
                                                                    let rat_rules_ref =
                                                                        rat_rules.read().unwrap();

                                                                    (
                                                                        mt_sea::get_strategies(
                                                                            &rat_rules_ref,
                                                                            &minot_compare_name,
                                                                            variable.clone(),
                                                                            Some(&inner_name),
                                                                        ),
                                                                        mt_sea::get_strategies(
                                                                            &rat_rules_ref,
                                                                            &inner_name,
                                                                            variable.clone(),
                                                                            None,
                                                                        ),
                                                                    )
                                                                };
                                                                for action in minot_actions {
                                                                    if !matches!(
                                                                        action,
                                                                        ActionPlan::Sail
                                                                    ) {
                                                                        rat_coord_tx.send(
                                                                    MinotTask::SendRatAction {
                                                                        ship_name: minot_compare_name
                                                                            .clone(),
                                                                        var_name: variable.clone(),
                                                                        data: action,
                                                                    },
                                                                )
                                                                .unwrap();
                                                                    }
                                                                }

                                                                if my_actions.is_empty() {
                                                                    my_actions
                                                                        .push(ActionPlan::Sail);
                                                                }

                                                                debug!(
                                                                    "sending actions: {:?}",
                                                                    my_actions
                                                                );

                                                                // answer the client that asked
                                                                for action in my_actions {
                                                                    rat_coord_tx.send(MinotTask::SendRatAction {
                                                                        ship_name: inner_name.clone(),
                                                                        var_name: variable.clone(),
                                                                        data: action,
                                                                    })
                                                                    .unwrap();
                                                                }

                                                                let mut pending_fixed =
                                                                    Vec::<Vec<WindData>>::new();
                                                                {
                                                                    let mut winds =
                                                                        winds_inner.write().unwrap();

                                                                    let mut asked_for_dynamic = false;
                                                                    if let Some(wt) =
                                                                        winds.get_mut(&variable)
                                                                    {
                                                                        for wte in wt.iter_mut() {
                                                                            match &mut wte.kind {
                                                                                WindTaskKind::Fix(wind_datas)
                                                                                    if !wte.already_seen =>
                                                                                {
                                                                                    pending_fixed.push(wind_datas.clone());
                                                                                    wte.already_seen = true;
                                                                                }
                                                                                WindTaskKind::Dynamic => {
                                                                                    if asked_for_dynamic {
                                                                                        continue;
                                                                                    }
                                                                                    asked_for_dynamic = true;

                                                                                    debug!(
                                                                                        "asking Minot TUI for dyn for {}",
                                                                                        &variable
                                                                                    );
                                                                                    let ret = rat_coord_tx.send(
                                                                                            MinotTask::WindDynamicVarReq(
                                                                                                variable.clone(),
                                                                                            ),
                                                                                        );
                                                                                    match ret {
                                                                                        Ok(_) => {}
                                                                                        Err(e) => {
                                                                                            error!("Could not find connected Minot TUI for asking dynamic wind. {e}");
                                                                                        }
                                                                                    }
                                                                                }
                                                                                WindTaskKind::Fix(_) => {}
                                                                            }
                                                                        }
                                                                    }

                                                                    // set all other to be sent again
                                                                    for (var, wind) in winds.iter_mut() {
                                                                        if var == &variable {
                                                                            continue;
                                                                        }

                                                                        wind.iter_mut().for_each(
                                                                            |wte| {
                                                                                wte.already_seen =
                                                                                    false;
                                                                            },
                                                                        );
                                                                    }
                                                                }

                                                                for wind_data in pending_fixed {
                                                                    if let Err(e) =
                                                                        wind_tx_inner.send(wind_data).await
                                                                    {
                                                                        error!(
                                                                            "Could not enqueue stored wind data for forwarding: {}",
                                                                            e
                                                                        );
                                                                    }
                                                                }
                                                            }
                                                            PacketKind::Heartbeat => {
                                                                debug!(
                                                                    "Received heartbeat from {}",
                                                                    inner_name
                                                                );
                                                                let _ = client
                                                                    .send
                                                                    .send(Packet {
                                                                        header: net::Header::default(),
                                                                        data: PacketKind::Heartbeat,
                                                                    })
                                                                    .await;
                                                            }
                                                            PacketKind::Sonar => {
                                                                let reliable = connected_clients_inner
                                                                    .read()
                                                                    .unwrap()
                                                                    .clone();
                                                                let best_effort = client_qos_inner
                                                                    .read()
                                                                    .unwrap()
                                                                    .iter()
                                                                    .filter(|(_, mode)| **mode == Qos::BestEffort)
                                                                    .map(|(name, _)| name.clone())
                                                                    .collect::<HashSet<_>>();
                                                                let _ = client
                                                                    .send
                                                                    .send(Packet {
                                                                        header: net::Header::default(),
                                                                        data: PacketKind::ClientsHash { reliable, best_effort },
                                                                    })
                                                                    .await;
                                                            }
                                                            PacketKind::Wind(bwd) => {
                                                                let mut send_now =
                                                                    Vec::with_capacity(bwd.len());
                                                                for wd in bwd {
                                                                    match wd.at_var {
                                                                        Some(var) => {
                                                                            let mut w = winds_inner
                                                                                .write()
                                                                                .unwrap();
                                                                            match w.get_mut(&var) {
                                                                                Some(wt) => {
                                                                                    wt.push(WindTask { kind: WindTaskKind::Fix(vec![wd.data]), already_seen: false });
                                                                                }
                                                                                None => {
                                                                                    w.insert(var,
                                                                                    vec![WindTask {
                                                                                        already_seen: false,
                                                                                        kind: WindTaskKind::Fix(vec![wd.data])
                                                                                    }]);
                                                                                }
                                                                            }
                                                                        }
                                                                        None => {
                                                                            send_now.push(wd.data);
                                                                        }
                                                                    }
                                                                }
                                                                if !send_now.is_empty() {
                                                                    if let Err(e) = wind_tx_inner.send(send_now).await {
                                                                        error!(
                                                                            "Could not enqueue wind data from client '{}' for forwarding: {}",
                                                                            inner_name,
                                                                            e
                                                                        );
                                                                    }
                                                                }
                                                            }
                                                            PacketKind::Torpedo(dead_clients) => {
                                                                debug!(
                                                                    "Received Torpedo: {:?}",
                                                                    dead_clients
                                                                );
                                                                let _ = client
                                                                    .send
                                                                    .send(Packet {
                                                                        header: net::Header::default(),
                                                                        data: PacketKind::Acknowledge,
                                                                    })
                                                                    .await;
                                                                rat_coord_tx
                                                                    .send(MinotTask::BroadcastTorpedo {
                                                                        dead_clients,
                                                                    })
                                                                    .unwrap();
                                                            }
                                                            PacketKind::PeerDead { ship } => {
                                                                debug!(
                                                                    "PeerDead reported by {}: {}",
                                                                    inner_name, ship
                                                                );
                                                                rat_coord_tx
                                                                    .send(MinotTask::PeerDead { ship })
                                                                    .unwrap();
                                                            }
                                                            _ => {
                                                                warn!(
                                                                    "Received unexpected packet {:?}",
                                                                    packet.data
                                                                );
                                                            }
                                                        }
                                                    }
                                                    result @ (Ok(Err(
                                                        tokio::sync::broadcast::error::RecvError::Closed,
                                                    ))
                                                    | Err(_)) => {
                                                        if result.is_err() {
                                                            // The client stopped speaking for
                                                            // DISCONNECT_TIMEOUT_MS. Its handler is
                                                            // the only thing answering variable
                                                            // requests, so tearing it down silently
                                                            // makes the client hang on its next
                                                            // bacon() with no diagnostic.
                                                            warn!(
                                                                "Client {} timed out (no packet or heartbeat within {}ms), treating as disconnected",
                                                                inner_name, DISCONNECT_TIMEOUT_MS
                                                            );
                                                        } else {
                                                            info!(
                                                                "Client {} disconnected",
                                                                inner_name
                                                            );
                                                        }
                                                        let best_effort_vars =
                                                            sub_overrides_for_disconnect_inner
                                                                .read()
                                                                .unwrap()
                                                                .keys()
                                                                .filter(|(ship, _)| ship == &inner_name)
                                                                .map(|(_, var)| var.clone())
                                                                .collect::<Vec<_>>();

                                                        let mut affected_vars =
                                                            best_effort_vars.clone();
                                                        if client.remove_rules_on_disconnect
                                                            || !best_effort_vars.is_empty()
                                                        {
                                                            affected_vars.extend(
                                                                rules_change_for_disconnect_inner
                                                                .read()
                                                                .unwrap()
                                                                .all_vars_for_ship(&inner_name),
                                                            );
                                                            rules_change_for_disconnect_inner
                                                                .write()
                                                                .unwrap()
                                                                .remove_client(&inner_name);
                                                        }
                                                        take_subscription_override_vars(
                                                            &mut sub_overrides_for_disconnect_inner
                                                                .write()
                                                                .unwrap(),
                                                            &inner_name,
                                                        );
                                                        affected_vars.sort();
                                                        affected_vars.dedup();
                                                        for var in affected_vars {
                                                            rat_coord_tx
                                                                .send(MinotTask::PushRoutesForVar { var })
                                                                .ok();
                                                        }
                                                        connected_clients_inner
                                                            .write()
                                                            .unwrap()
                                                            .remove(&inner_name);
                                                        client_qos_inner
                                                            .write()
                                                            .unwrap()
                                                            .remove(&inner_name);
                                                        return;
                                                    }
                                                    Ok(Err(
                                                        tokio::sync::broadcast::error::RecvError::Lagged(
                                                            n,
                                                        ),
                                                    )) => {
                                                        warn!(
                                                            "Client {} receiver lagged by {} messages",
                                                            inner_name, n
                                                        );
                                                    }
                                                }
                                            }
                                        }
                                    };
                                });
                                handlers_for_task
                                    .lock()
                                    .unwrap()
                                    .insert(name_for_task, handle.abort_handle());
                            }
                            ShipKind::Wind(name) => {
                                if let Some(old) = client_handlers.lock().unwrap().remove(&name) {
                                    old.abort();
                                }
                                wind_clients_for_spawn.write().unwrap().insert(name.clone());
                                let mut client_news = client.recv.subscribe();
                                let inner_name_rx = name.clone();
                                let winds_on_disconnect = std::sync::Arc::clone(&winds_changer);
                                let wind_clients_on_disconnect =
                                    std::sync::Arc::clone(&wind_clients_for_spawn);
                                let wind_recv_handle = tokio::spawn(async move {
                                    loop {
                                        match client_news.recv().await {
                                            Ok((packet, _)) => match packet.data {
                                                PacketKind::Acknowledge => {
                                                    debug!(
                                                        "Received wind ack from {inner_name_rx}"
                                                    );
                                                }
                                                PacketKind::Heartbeat => {
                                                    debug!(
                                                        "Received heartbeat from wind {inner_name_rx}"
                                                    );
                                                }
                                                _ => {
                                                    warn!(
                                                        "Unexpected packet from wind: {packet:?}"
                                                    );
                                                }
                                            },
                                            Err(e) => {
                                                match e {
                                                    tokio::sync::broadcast::error::RecvError::Closed => {},
                                                    _ => {
                                                        error!(
                                                            "Could not receive packet from wind client {inner_name_rx}: {e}"
                                                        );
                                                    }
                                                }
                                                info!(
                                                    "Wind client {inner_name_rx} disconnected, clearing wind state"
                                                );
                                                wind_clients_on_disconnect
                                                    .write()
                                                    .unwrap()
                                                    .remove(&inner_name_rx);
                                                winds_on_disconnect.write().unwrap().clear();
                                                return;
                                            }
                                        }
                                    }
                                });
                                client_handlers
                                    .lock()
                                    .unwrap()
                                    .insert(name, wind_recv_handle.abort_handle());
                            }
                        }
                    }
                    Err(e) => {
                        error!("Could not receive new client: {}", e);
                    }
                }
            }
        });

        // wait here until all clients are connected.
        clients_collected_rx.recv().await;

        let mut lock_next = locked_start;
        while let Some(task) = coord_rx.recv().await {
            match task {
                MinotTask::WindDynamicVarReq(var) => {
                    {
                        let minot_tui_connected = minot_client_connected.read().unwrap();
                        if !*minot_tui_connected {
                            error!(
                                "Wants to ask Minot TUI for wind for var {var} but is not connected."
                            );
                            continue;
                        }
                    }

                    let ret = coordinator
                        .rat_send(
                            COMPARE_NODE_NAME.to_owned(),
                            PacketKind::VariableTaskRequest(var),
                        )
                        .await;
                    if let Err(e) = ret {
                        error!("Could not send Dynamic Var Request to LH TUI: {e}");
                    }
                }
                MinotTask::AppendRule { variable, commands } => {
                    let mut current_rules = rules_changer.write().unwrap();
                    current_rules.insert(variable, commands);
                    info!("added new rule");
                }
                MinotTask::RulesClear => {
                    let mut current_rules = rules_changer.write().unwrap();
                    current_rules.clear();
                    info!("cleared all rules");
                }
                MinotTask::LockNext { unlock_first } => {
                    info!("locking next");
                    if unlock_first {
                        let clients = coordinator.rat_qs.read().await;
                        for (c, _) in clients.iter() {
                            if let Err(e) = coordinator
                                .rat_send(c.clone(), PacketKind::Acknowledge)
                                .await
                            {
                                error!("Error while sending ack for unlocking to Rat {}: {}", c, e);
                                std::process::exit(1);
                            }
                        }
                    }

                    lock_next = true;
                }
                MinotTask::Unlock => {
                    let clients = coordinator.rat_qs.read().await;
                    for (c, _) in clients.iter() {
                        if let Err(e) = coordinator
                            .rat_send(c.clone(), PacketKind::Acknowledge)
                            .await
                        {
                            error!("Error while sending ack for unlocking to Rat {}: {}", c, e);
                            std::process::exit(1);
                        }
                    }
                    lock_next = false;
                }
                MinotTask::SendRatAction {
                    ship_name,
                    var_name,
                    mut data,
                } => {
                    {
                        let minot_tui_connected = minot_client_connected.read().unwrap();
                        if !*minot_tui_connected {
                            if ship_name == COMPARE_NODE_NAME {
                                continue;
                            }
                            data = match &data {
                                ActionPlan::Sail => data.clone(),
                                ActionPlan::Shoot { target, id } => {
                                    let ntargets = target
                                        .iter()
                                        .filter(|&t| t != COMPARE_NODE_NAME)
                                        .cloned()
                                        .collect::<Vec<_>>();
                                    if ntargets.is_empty() {
                                        ActionPlan::Sail
                                    } else {
                                        ActionPlan::Shoot {
                                            target: ntargets,
                                            id: *id,
                                        }
                                    }
                                }
                                ActionPlan::Catch { source, id: _ } => {
                                    if source == COMPARE_NODE_NAME {
                                        ActionPlan::Sail
                                    } else {
                                        data.clone()
                                    }
                                }
                            };
                        }
                    }

                    let lock_next = lock_next && ship_name != COMPARE_NODE_NAME;

                    // Resilient targets get a spawned, failure-tolerant dispatch.
                    // The inline branch below aborts the whole coordinator when a
                    // send fails, which is only correct for a node whose loss is
                    // fatal anyway. A per-topic override wins over the node mode.
                    let effective_mode = coordinator
                        .sub_qos_overrides
                        .read()
                        .unwrap()
                        .get(&(ship_name.clone(), var_name.clone()))
                        .copied()
                        .or_else(|| client_qos.read().unwrap().get(&ship_name).copied())
                        .unwrap_or_default();

                    if !effective_mode.fires_torpedo() {
                        let coord_inner = coordinator.clone();
                        let var_name_inner = var_name.clone();
                        let ship_name_inner = ship_name.clone();
                        tokio::spawn(async move {
                            let _ = coord_inner
                                .rat_action_send(
                                    ship_name_inner,
                                    var_name_inner,
                                    data,
                                    lock_next,
                                    true,
                                )
                                .await;
                        });
                    } else if let Err(e) = coordinator
                        .rat_action_send(ship_name.clone(), var_name, data, lock_next, false)
                        .await
                    {
                        error!("Error while sending action to Rat {}: {}", ship_name, e);
                        std::process::exit(1);
                    }
                }
                MinotTask::UnregisterShipAtVar { ship, var, kind } => {
                    let mt_net_kind = match kind {
                        net::RatPubRegisterKind::Publish => RatPubRegisterKind::Publish,
                        net::RatPubRegisterKind::Subscribe | net::RatPubRegisterKind::Scope => {
                            RatPubRegisterKind::Subscribe
                        }
                    };
                    let removed = {
                        let mut current_rules = rules_changer.write().unwrap();
                        current_rules.unregister(&var, &ship, mt_net_kind.clone())
                    };
                    if !removed {
                        debug!("Ship {ship} was not registered for var {var}, nothing to remove");
                        let _ = coordinator
                            .rat_send(ship, net::PacketKind::Acknowledge)
                            .await;
                        continue;
                    }
                    // A best-effort subscription is tracked separately, so it
                    // has to be forgotten too or the topic stays marked as
                    // having a best-effort subscriber that no longer exists.
                    coordinator
                        .sub_qos_overrides
                        .write()
                        .unwrap()
                        .remove(&(ship.clone(), var.clone()));
                    info!(
                        "Unregistering from topics ({:?}, {:?}, {:?})",
                        &ship, &var, &mt_net_kind
                    );
                    let rules_snapshot = rules_changer.read().unwrap().clone();
                    if let Err(e) = coordinator.push_routes_for_var(&var, &rules_snapshot).await {
                        error!("Failed to push routes after unregistering {var}: {e}");
                    }
                    let _ = coordinator
                        .rat_send(ship, net::PacketKind::Acknowledge)
                        .await;
                }
                MinotTask::RegisterShipAtVar {
                    ship,
                    var,
                    kind,
                    node_mode,
                } => {
                    // Track BE publishers
                    if kind == net::RatPubRegisterKind::Publish && node_mode == Qos::BestEffort {
                        coordinator
                            .be_publisher_vars
                            .write()
                            .unwrap()
                            .insert(var.clone());

                        // Notify any existing reliable subscribers that this var now has a BE publisher.
                        let reliable_subs: Vec<String> = {
                            let current_rules = rules_changer.read().unwrap();
                            let overrides = coordinator.sub_qos_overrides.read().unwrap();
                            current_rules
                                .subscriber_ships(&var)
                                .into_iter()
                                .filter(|s| {
                                    overrides
                                        .get(&(s.clone(), var.clone()))
                                        .is_none_or(|mode| mode.expects_reliable_delivery())
                                })
                                .collect()
                        };
                        if !reliable_subs.is_empty() {
                            let error_msg = format!(
                                "Topic '{}' has a best-effort publisher. A reliable subscriber cannot connect to a best-effort publisher. \
                                 Pass Qos::BestEffort to create_subscriber to opt in to best-effort delivery.",
                                var
                            );
                            for sub_ship in reliable_subs {
                                error!(
                                    "Notifying existing reliable subscriber '{}' of BE publisher on '{}'",
                                    sub_ship, var
                                );
                                let _ = coordinator
                                    .rat_send(
                                        sub_ship,
                                        net::PacketKind::RegistrationError(error_msg.clone()),
                                    )
                                    .await;
                            }
                        }
                    }

                    // Reject reliable subscribers connecting to BE-published topics
                    if (kind == net::RatPubRegisterKind::Subscribe
                        || kind == net::RatPubRegisterKind::Scope)
                        && node_mode.expects_reliable_delivery()
                        && coordinator.be_publisher_vars.read().unwrap().contains(&var)
                    {
                        let msg = format!(
                            "Topic '{}' has a best-effort publisher. A reliable subscriber cannot connect to a best-effort publisher. \
                             Pass Qos::BestEffort to create_subscriber to opt in to best-effort delivery.",
                            var
                        );
                        error!("{}", &msg);
                        let _ = coordinator
                            .rat_send(ship, net::PacketKind::RegistrationError(msg))
                            .await;
                        continue;
                    }

                    if node_mode != Qos::Reliable && kind == net::RatPubRegisterKind::Subscribe {
                        coordinator
                            .sub_qos_overrides
                            .write()
                            .unwrap()
                            .insert((ship.clone(), var.clone()), node_mode);
                    }
                    let mt_net_kind = match kind {
                        net::RatPubRegisterKind::Publish => RatPubRegisterKind::Publish,
                        net::RatPubRegisterKind::Subscribe | net::RatPubRegisterKind::Scope => {
                            RatPubRegisterKind::Subscribe
                        }
                    };
                    let already_registered = {
                        let current_rules = rules_changer.read().unwrap();
                        current_rules
                            .cache
                            .get(&var)
                            .map(|s| s.contains(&(ship.clone(), mt_net_kind.clone())))
                            .unwrap_or(false)
                    };
                    // Single-subscription prevention
                    if already_registered {
                        debug!(
                            "Ship {} already registered for var {} as {:?}, skipping",
                            ship, var, mt_net_kind
                        );
                        // Still send Acknowledge for duplicate registrations
                        let _ = coordinator
                            .rat_send(ship, net::PacketKind::Acknowledge)
                            .await;
                        continue;
                    }
                    info!(
                        "Registering for topics ({:?}, {:?}, {:?})",
                        &ship, &var, &mt_net_kind
                    );
                    {
                        let mut current_rules = rules_changer.write().unwrap();
                        current_rules.register(var.clone(), ship.clone(), mt_net_kind);
                    }
                    // Wait for the registering ship to be present in rat_qs before
                    // pushing routes, so convert_action can resolve its address.
                    coordinator.wait_for_client(&ship).await;
                    // Push updated routes to all affected ships after registration.
                    let rules_snapshot = rules_changer.read().unwrap().clone();
                    coordinator
                        .push_routes_for_var(&var, &rules_snapshot)
                        .await
                        .ok();
                    let _ = coordinator
                        .rat_send(ship, net::PacketKind::Acknowledge)
                        .await;
                }
                MinotTask::PushRoutesForVar { var } => {
                    let rules_snapshot = rules_changer.read().unwrap().clone();
                    if let Err(e) = coordinator.push_routes_for_var(&var, &rules_snapshot).await {
                        debug!("PushRoutesForVar for {}: {}", var, e);
                    }
                }
                MinotTask::BroadcastTorpedo { dead_clients } => {
                    info!("Broadcasting Torpedo: {:?}", dead_clients);
                    let targets: Vec<(String, tokio::sync::mpsc::Sender<Packet>)> = {
                        let clients = coordinator.rat_qs.read().await;
                        clients
                            .iter()
                            .map(|(name, info)| (name.clone(), info.sender.clone()))
                            .collect()
                    };

                    for (name, sender) in targets {
                        if let Err(e) = sender
                            .send(Packet {
                                header: net::Header::default(),
                                data: PacketKind::Torpedo(dead_clients.clone()),
                            })
                            .await
                        {
                            debug!("Could not forward Torpedo to {}: {}", name, e);
                        }
                    }

                    coordinator_shutting_down_in_coord
                        .store(true, std::sync::atomic::Ordering::SeqCst);
                    if let Some(tx) = &torpedo_tx_for_coord {
                        let _ = tx.send(()).await;
                    }
                }
                MinotTask::PeerDead { ship } => {
                    info!("PeerDead: {}", ship);
                    let is_reliable = client_qos
                        .read()
                        .unwrap()
                        .get(&ship)
                        .copied()
                        .unwrap_or_default()
                        .fires_torpedo();
                    let affected_vars = rules_changer.read().unwrap().all_vars_for_ship(&ship);
                    rules_changer.write().unwrap().remove_client(&ship);
                    take_subscription_override_vars(
                        &mut coordinator.sub_qos_overrides.write().unwrap(),
                        &ship,
                    );
                    let rules_snapshot = rules_changer.read().unwrap().clone();
                    for var in &affected_vars {
                        coordinator
                            .push_routes_for_var(var, &rules_snapshot)
                            .await
                            .ok();
                    }
                    connected_clients.write().unwrap().remove(&ship);
                    client_qos.write().unwrap().remove(&ship);
                    if is_reliable {
                        let dead_clients = vec![ship.clone()];
                        info!("Reliable peer lost, firing Torpedo: {:?}", dead_clients);
                        let targets: Vec<(String, tokio::sync::mpsc::Sender<Packet>)> = {
                            let clients = coordinator.rat_qs.read().await;
                            clients
                                .iter()
                                .map(|(name, info)| (name.clone(), info.sender.clone()))
                                .collect()
                        };

                        for (name, sender) in targets {
                            if let Err(e) = sender
                                .send(Packet {
                                    header: net::Header::default(),
                                    data: PacketKind::Torpedo(dead_clients.clone()),
                                })
                                .await
                            {
                                debug!("Could not forward Torpedo to {}: {}", name, e);
                            }
                        }

                        coordinator_shutting_down_in_coord
                            .store(true, std::sync::atomic::Ordering::SeqCst);
                        if let Some(tx) = &torpedo_tx_for_coord {
                            let _ = tx.send(()).await;
                        }
                    }
                }
            }
        }
        debug!("coord_rx closed");
    });
}

/// Attempt to start a coordinator with the given configuration.
///
/// Acquires an exclusive lock file at `/tmp/minot-coord.lock`.
/// Returns `true` if the coordinator was started, `false` if one was already running.
/// Idempotent: safe to call concurrently — only one will win the lock.
pub fn try_start_with_rules(
    locked_start: bool,
    clients: HashSet<String>,
    rules: Rules,
    torpedo_tx: Option<tokio::sync::mpsc::Sender<()>>,
) -> bool {
    try_start_with_rules_and_ready(locked_start, clients, rules, torpedo_tx, None)
}

fn try_start_with_rules_and_ready(
    locked_start: bool,
    clients: HashSet<String>,
    rules: Rules,
    torpedo_tx: Option<tokio::sync::mpsc::Sender<()>>,
    ready_tx: Option<tokio::sync::oneshot::Sender<Result<(), String>>>,
) -> bool {
    let lock_file_path =
        std::env::temp_dir().join(format!("minot-coord_{}.lock", current_user_id()));
    let lock_file = match std::fs::File::create(&lock_file_path) {
        Ok(f) => f,
        Err(e) => {
            log::error!("Failed to create coordinator lock file: {}", e);
            return false;
        }
    };
    if fs2::FileExt::try_lock_exclusive(&lock_file).is_err() {
        log::debug!("Coordinator already running, skipping embedded start.");
        return false;
    }
    log::info!("Starting embedded coordinator...");
    let (startup_failed_tx, startup_failed_rx) = tokio::sync::oneshot::channel();
    run_coordinator_with_ready(
        locked_start,
        clients,
        rules,
        torpedo_tx,
        ready_tx,
        Some(startup_failed_tx),
    );
    // Keep the lock alive until runtime exit, but release it if coordinator
    // initialization fails so a subsequent attempt can recover.
    tokio::spawn(async move {
        let _lock = lock_file;
        let _ = startup_failed_rx.await;
    });
    true
}

fn current_user_id() -> u64 {
    #[cfg(unix)]
    {
        // SAFETY: `getuid` takes no arguments and has no preconditions.
        unsafe { libc::getuid() as u64 }
    }

    #[cfg(not(unix))]
    {
        0
    }
}

/// Start a coordinator with empty rules.
///
/// Idempotent: does nothing if a coordinator is already running.
/// Safe to call from any async context — spawns background tasks and returns immediately.
pub fn start_default() {
    try_start_with_rules(false, HashSet::new(), Rules::new(), None);
}

pub fn start_default_with_torpedo(torpedo_tx: Option<tokio::sync::mpsc::Sender<()>>) {
    try_start_with_rules(false, HashSet::new(), Rules::new(), torpedo_tx);
}

/// Ensure this process has started the default coordinator and wait until its
/// network join subscriber can accept registrations.
///
/// Returns `Ok(false)` when another process already owns the coordinator lock.
/// Callers should still retry registration: the lock owner may be finishing startup.
pub async fn ensure_default_coordinator_ready(
    torpedo_tx: Option<tokio::sync::mpsc::Sender<()>>,
) -> anyhow::Result<bool> {
    let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
    let started = try_start_with_rules_and_ready(
        false,
        HashSet::new(),
        Rules::new(),
        torpedo_tx,
        Some(ready_tx),
    );
    if started {
        tokio::time::timeout(EMBEDDED_COORDINATOR_READY_TIMEOUT, ready_rx)
            .await
            .map_err(|_| anyhow::anyhow!("timed out waiting for embedded coordinator readiness"))?
            .map_err(|_| anyhow::anyhow!("embedded coordinator exited before becoming ready"))?
            .map_err(anyhow::Error::msg)?;
    }
    Ok(started)
}

#[deprecated(note = "use ensure_default_coordinator_ready")]
pub async fn start_default_with_torpedo_ready(
    torpedo_tx: Option<tokio::sync::mpsc::Sender<()>>,
) -> anyhow::Result<bool> {
    ensure_default_coordinator_ready(torpedo_tx).await
}

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

    #[test]
    fn disconnect_removes_only_the_dead_clients_subscription_overrides() {
        let mut overrides = HashMap::from([
            (
                ("scope".to_string(), "registered".to_string()),
                Qos::BestEffort,
            ),
            (("scope".to_string(), "map".to_string()), Qos::TryReliable),
            (
                ("monitor".to_string(), "registered".to_string()),
                Qos::BestEffort,
            ),
        ]);

        let mut affected = take_subscription_override_vars(&mut overrides, "scope");
        affected.sort();

        assert_eq!(affected, ["map", "registered"]);
        assert_eq!(
            overrides,
            HashMap::from([(
                ("monitor".to_string(), "registered".to_string()),
                Qos::BestEffort
            )])
        );
    }
}