nautalid 0.1.0

//! KWT **`jti`** replay for `/v1/protected/*` — [AetherDB](https://crates.io/crates/aetherdb) only.
//!
//! Embedded **`kv-local`** when **`KWT_REPLAY_AETHERDB_URL`** is unset; **`tcp://…`** for a
//! shared distributed node (Hickory DNS + TLS/mTLS). Disable with **`KWT_REPLAY=0`**.

pub(crate) mod bus_api;
mod connect;
pub(crate) mod kdl_build;
pub mod runtime;
pub(crate) mod topology;

#[cfg(all(feature = "kwt", feature = "filter-control"))]
pub mod aetherdb_control;

#[cfg(all(feature = "kwt", feature = "filter-control"))]
pub mod wt_aether_control;

#[cfg(all(feature = "kwt", feature = "filter-control"))]
pub mod aether_runtime_control;

#[cfg(all(feature = "kwt", feature = "filter-control"))]
pub mod control;

pub use crate::aether_tunnel::WireTls;
pub use connect::{parse_url, ConnectError, ConnectMode, shared_replay_required};
pub use runtime::{
    format_cluster_response, format_dma_response, safety_config_from_kdl, AetherRuntime,
    HostStatus, RuntimeError,
};
pub use topology::{DeployMode, Topology};

use std::cell::Cell;
use std::sync::Mutex;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use aetherdb::{Key, Value};
use thiserror::Error;
use uuid::Uuid;

const DEFAULT_PREFIX: &str = "nautalid:kwt:jti:";

#[derive(Debug, Error)]
pub enum OpenError {
    #[error(transparent)]
    Connect(#[from] ConnectError),
}

enum State {
    Disabled,
    Db(Aetherdb),
}

struct Aetherdb {
    db: aetherdb::DatabaseRef,
    prefix: String,
    ttl_secs: u64,
    shared: bool,
    max_jtis: usize,
}

const COUNT_KEY_SUFFIX: &str = "__count__";

/// Tracks seen token JTIs for replay rejection via AetherDB.
#[derive(Clone)]
pub struct JtiStore {
    state: std::sync::Arc<Mutex<State>>,
}

impl JtiStore {
    pub fn open_from_env() -> Result<Self, OpenError> {
        if replay_disabled() {
            return Ok(Self::disabled());
        }
        let ttl_secs = replay_ttl_secs();
        let spec = connect::spec_from_env();
        Self::open_with_connect(spec, ttl_secs, None)
    }

    /// Open replay storage from an explicit connect spec (embedded or wire).
    pub fn open_with_connect(
        spec: connect::ConnectMode,
        ttl_secs: u64,
        prefix: Option<&str>,
    ) -> Result<Self, OpenError> {
        let shared = matches!(
            spec,
            connect::ConnectMode::Wire { .. } | connect::ConnectMode::Colocated { .. }
        );
        let db = Aetherdb::open(ttl_secs, &spec, prefix)?;
        tracing::info!(shared, "KWT replay using AetherDB");
        Ok(Self {
            state: std::sync::Arc::new(Mutex::new(State::Db(db))),
        })
    }

    /// Hot-swap JTI storage to an in-process database (unified data plane).
    pub fn reload_colocated(&self, db: aetherdb::DatabaseRef) -> Result<(), OpenError> {
        let inner = Aetherdb::from_database(db, true)?;
        *self.state.lock().expect("jti store lock poisoned") = State::Db(inner);
        tracing::info!("KWT replay attached to colocated AetherDB");
        Ok(())
    }

    /// Hot-swap JTI storage from a new connect spec (local / wire / file).
    pub fn reload_from_connect(&self, spec: connect::ConnectMode) -> Result<(), OpenError> {
        if replay_disabled() {
            *self.state.lock().expect("jti store lock poisoned") = State::Disabled;
            return Ok(());
        }
        let ttl_secs = replay_ttl_secs();
        let inner = Aetherdb::open(ttl_secs, &spec, None)?;
        tracing::info!(shared = inner.shared, "KWT replay reconnected");
        *self.state.lock().expect("jti store lock poisoned") = State::Db(inner);
        Ok(())
    }

    /// Re-read `KWT_REPLAY_AETHERDB_URL` and reconnect.
    pub fn reload_from_env(&self) -> Result<(), OpenError> {
        self.reload_from_connect(connect::spec_from_env())
    }

    #[must_use]
    pub fn disabled() -> Self {
        Self {
            state: std::sync::Arc::new(Mutex::new(State::Disabled)),
        }
    }

    #[must_use]
    pub fn is_enabled(&self) -> bool {
        !matches!(
            *self.state.lock().expect("jti store lock poisoned"),
            State::Disabled
        )
    }

    #[must_use]
    pub fn backend_label(&self) -> &'static str {
        match *self.state.lock().expect("jti store lock poisoned") {
            State::Disabled => "disabled",
            State::Db(_) => "aetherdb",
        }
    }

    #[must_use]
    pub fn is_shared(&self) -> bool {
        match *self.state.lock().expect("jti store lock poisoned") {
            State::Disabled => false,
            State::Db(ref db) => db.shared,
        }
    }

    /// Record **`jti`** or return [`kwt::KwtError::Replayed`].
    pub fn check_and_record(&self, jti: &Uuid) -> Result<(), kwt::KwtError> {
        let guard = self.state.lock().expect("jti store lock poisoned");
        match &*guard {
            State::Disabled => Ok(()),
            State::Db(db) => db.check_and_record(jti),
        }
    }
}

impl Aetherdb {
    fn open(
        ttl_secs: u64,
        spec: &connect::ConnectMode,
        prefix: Option<&str>,
    ) -> Result<Self, OpenError> {
        let prefix = prefix
            .map(str::to_string)
            .or_else(|| {
                std::env::var("KWT_REPLAY_AETHERDB_PREFIX")
                    .ok()
                    .filter(|s| !s.trim().is_empty())
            })
            .unwrap_or_else(|| DEFAULT_PREFIX.into());
        let shared = matches!(
            spec,
            connect::ConnectMode::Wire { .. } | connect::ConnectMode::Colocated { .. }
        );
        let db = match spec {
            connect::ConnectMode::Colocated { db } => db.clone(),
            _ => connect::open(spec)?,
        };
        Ok(Self {
            db,
            prefix,
            ttl_secs,
            shared,
            max_jtis: replay_max_jtis(),
        })
    }

    fn from_database(db: aetherdb::DatabaseRef, shared: bool) -> Result<Self, OpenError> {
        let prefix = std::env::var("KWT_REPLAY_AETHERDB_PREFIX")
            .ok()
            .filter(|s| !s.trim().is_empty())
            .unwrap_or_else(|| DEFAULT_PREFIX.into());
        Ok(Self {
            db,
            prefix,
            ttl_secs: replay_ttl_secs(),
            shared,
            max_jtis: replay_max_jtis(),
        })
    }

    fn count_key(&self) -> Key {
        Key::from_bytes(format!("{}{COUNT_KEY_SUFFIX}", self.prefix).into_bytes())
    }

    fn load_count(&self, txn: &aetherdb::Transaction<'_>) -> Result<usize, aetherdb::Error> {
        let key = self.count_key();
        match txn.get(&key) {
            Ok(Some(value)) => {
                let bytes = value.as_bytes();
                if bytes.len() >= 8 {
                    let mut buf = [0u8; 8];
                    buf.copy_from_slice(&bytes[..8]);
                    Ok(usize::try_from(u64::from_be_bytes(buf)).unwrap_or(usize::MAX))
                } else {
                    Ok(0)
                }
            }
            Ok(None) => Ok(0),
            Err(err) => Err(err),
        }
    }

    fn store_count(
        &self,
        txn: &aetherdb::Transaction<'_>,
        count: usize,
    ) -> Result<(), aetherdb::Error> {
        let key = self.count_key();
        let value = Value::from_bytes((count as u64).to_be_bytes().to_vec());
        txn.put(key, value)
    }

    fn key(&self, jti: &Uuid) -> Key {
        Key::from_bytes(format!("{}{jti}", self.prefix).into_bytes())
    }

    fn expiry_value(&self) -> Value {
        let exp = SystemTime::now()
            .checked_add(Duration::from_secs(self.ttl_secs))
            .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
            .map_or(0, |d| d.as_secs());
        Value::from_bytes(exp.to_be_bytes().to_vec())
    }

    fn is_expired(value: &Value) -> bool {
        let bytes = value.as_bytes();
        if bytes.len() < 8 {
            return true;
        }
        let mut buf = [0u8; 8];
        buf.copy_from_slice(&bytes[..8]);
        let exp = u64::from_be_bytes(buf);
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .is_ok_and(|now| now.as_secs() >= exp)
    }

    fn check_and_record(&self, jti: &Uuid) -> Result<(), kwt::KwtError> {
        let key = self.key(jti);
        let value = self.expiry_value();
        let replay = Cell::new(false);
        let at_capacity = Cell::new(false);
        self.db
            .transaction(|txn| {
                if let Some(existing) = txn.get(&key)? {
                    if !Self::is_expired(&existing) {
                        replay.set(true);
                        return Ok(());
                    }
                    let refresh = self.expiry_value();
                    txn.put(self.key(jti), refresh)?;
                    return Ok(());
                }
                let count = self.load_count(txn)?;
                if count >= self.max_jtis {
                    at_capacity.set(true);
                    return Ok(());
                }
                txn.put(key, value)?;
                self.store_count(txn, count + 1)?;
                Ok(())
            })
            .map_err(|_| kwt::KwtError::Replayed)?;
        if at_capacity.get() {
            return Err(kwt::KwtError::Replayed);
        }
        if replay.get() {
            return Err(kwt::KwtError::Replayed);
        }
        Ok(())
    }
}

fn replay_disabled() -> bool {
    if crate::security::kwt_jti_replay_required() {
        return false;
    }
    matches!(
        std::env::var("KWT_REPLAY").ok().as_deref(),
        Some("0") | Some("false") | Some("no")
    )
}

fn replay_ttl_secs() -> u64 {
    std::env::var("KWT_REPLAY_TTL_SECS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(86_400)
}

fn replay_max_jtis() -> usize {
    std::env::var("KWT_REPLAY_MAX_JTIS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(1_000_000)
}

/// Warm-up Hickory resolution for a configured wire endpoint (non-fatal).
pub async fn warm_up_wire_dns() {
    let spec = connect::spec_from_env();
    let connect::ConnectMode::Wire { endpoint, .. } = spec else {
        return;
    };
    match connect::resolve_wire_endpoint(&endpoint).await {
        Ok(resolved) => {
            tracing::info!(configured = %endpoint, %resolved, "AetherDB wire endpoint resolved");
        }
        Err(e) => {
            tracing::warn!(configured = %endpoint, %e, "AetherDB wire DNS lookup failed (non-fatal)");
        }
    }
}

/// Leaked AetherDB for async HTTP tests (avoids dropping the DB runtime inside `#[tokio::test]`).
#[cfg(test)]
pub fn test_jti_store() -> JtiStore {
    use std::sync::OnceLock;
    static STORE: OnceLock<JtiStore> = OnceLock::new();
    STORE
        .get_or_init(|| JtiStore::open_from_env().expect("test AetherDB"))
        .clone()
}

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

    fn test_store() -> JtiStore {
        let db = Aetherdb {
            db: open_bundled("kv-local").expect("kv-local"),
            prefix: "test:jti:".into(),
            ttl_secs: 3600,
            shared: false,
            max_jtis: 1_000_000,
        };
        JtiStore {
            state: std::sync::Arc::new(Mutex::new(State::Db(db))),
        }
    }

    #[test]
    fn rejects_replayed_jti() {
        let store = test_store();
        let jti = Uuid::now_v7();
        store.check_and_record(&jti).expect("first use");
        assert!(matches!(
            store.check_and_record(&jti),
            Err(kwt::KwtError::Replayed)
        ));
    }

    #[test]
    fn disabled_allows_reuse() {
        let store = JtiStore::disabled();
        let jti = Uuid::now_v7();
        store.check_and_record(&jti).unwrap();
        store.check_and_record(&jti).unwrap();
    }

    #[test]
    fn rejects_at_capacity() {
        let db = Aetherdb {
            db: open_bundled("kv-local").expect("kv-local"),
            prefix: "test:jti:cap:".into(),
            ttl_secs: 3600,
            shared: false,
            max_jtis: 2,
        };
        let store = JtiStore {
            state: std::sync::Arc::new(Mutex::new(State::Db(db))),
        };
        let jti1 = Uuid::now_v7();
        let jti2 = Uuid::now_v7();
        let jti3 = Uuid::now_v7();
        store.check_and_record(&jti1).expect("first");
        store.check_and_record(&jti2).expect("second");
        assert!(matches!(
            store.check_and_record(&jti3),
            Err(kwt::KwtError::Replayed)
        ));
    }

    #[test]
    fn open_from_env_uses_aetherdb() {
        let store = JtiStore::open_from_env().expect("open");
        assert_eq!(store.backend_label(), "aetherdb");
        assert!(!store.is_shared());
    }
}