nautalid 0.1.0

//! Nautalid glue for AetherDB wire routes not exposed on [`aetherdb::WireSessionPool`]
//! (e.g. **`replicate`**) and loopback dispatch when in-process handlers are unavailable.

use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use aetherdb::{
    EngineRuntime, Envelope, MessageKind, ZmqConfig, ZmqPattern, ZmqTlsConfig, ZmqTransport,
};
use aetherdb::fwt::Inbox;
use bytes::Bytes;

const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);

const HOST_CLIENT_ROUTES: &[&str] = &["storage", "command", "dma", "control", "replicate"];

/// Cached ZMQ wire clients (all standard host routes).
pub struct WireRoutePool {
    runtime: Arc<EngineRuntime>,
    tls: Option<ZmqTlsConfig>,
    connections: Mutex<HashMap<String, WireRouteConnection>>,
}

struct WireRouteConnection {
    transport: ZmqTransport,
    remote_routes: HashMap<String, u32>,
    inboxes: HashMap<String, Inbox>,
    next_correlation: AtomicU64,
}

impl WireRoutePool {
    #[must_use]
    pub fn new(runtime: Arc<EngineRuntime>) -> Self {
        Self::with_tls(runtime, None)
    }

    #[must_use]
    pub fn with_tls(runtime: Arc<EngineRuntime>, tls: Option<ZmqTlsConfig>) -> Self {
        Self {
            runtime,
            tls,
            connections: Mutex::new(HashMap::new()),
        }
    }

    pub fn request(
        &self,
        endpoint: &str,
        route: &str,
        payload: Bytes,
    ) -> Result<Bytes, String> {
        let mut guard = self.connections.lock().expect("wire route pool lock");
        if !guard.contains_key(endpoint) {
            let conn = WireRouteConnection::connect(&self.runtime, endpoint, self.tls.clone())?;
            guard.insert(endpoint.to_string(), conn);
        }
        guard
            .get(endpoint)
            .expect("wire route connection")
            .request(route, payload)
    }
}

impl WireRouteConnection {
    fn connect(
        runtime: &Arc<EngineRuntime>,
        endpoint: &str,
        tls: Option<ZmqTlsConfig>,
    ) -> Result<Self, String> {
        let transport = ZmqTransport::start(
            Arc::clone(runtime),
            ZmqConfig {
                bind: None,
                connect: Some(endpoint.to_string()),
                pattern: ZmqPattern::RouterDealer,
                tls,
            },
        )
        .map_err(|err| err.to_string())?;
        let mut inboxes = HashMap::new();
        for name in HOST_CLIENT_ROUTES {
            inboxes.insert((*name).to_string(), transport.bind(*name));
        }
        std::thread::sleep(Duration::from_millis(100));
        Ok(Self {
            transport,
            remote_routes: HashMap::new(),
            inboxes,
            next_correlation: AtomicU64::new(0),
        })
    }

    fn request(&self, route: &str, payload: Bytes) -> Result<Bytes, String> {
        let outbox = self.transport.outbox();
        let route_addr = self
            .remote_routes
            .get(route)
            .copied()
            .or_else(|| outbox.route_address(route))
            .ok_or_else(|| format!("unknown wire route `{route}`"))?;
        let inbox = self
            .inboxes
            .get(route)
            .ok_or_else(|| format!("unknown wire inbox `{route}`"))?;
        let correlation_id = self.next_correlation.fetch_add(1, Ordering::Relaxed) + 1;
        outbox
            .send(
                Envelope::new(MessageKind::Request, route_addr, 0, payload)
                    .with_correlation_id(correlation_id),
            )
            .map_err(|err| err.to_string())?;

        let deadline = Instant::now() + DEFAULT_TIMEOUT;
        loop {
            if Instant::now() > deadline {
                return Err(format!("{route} wire request timeout"));
            }
            match inbox.try_recv() {
                Ok(Some(env))
                    if env.correlation_id == correlation_id
                        && env.kind == MessageKind::Response =>
                {
                    return Ok(env.payload);
                }
                Ok(Some(_)) => {}
                Ok(None) => std::thread::sleep(Duration::from_millis(10)),
                Err(err) => return Err(err.to_string()),
            }
        }
    }
}