openrtc 1.0.2

use std::collections::HashMap;
use std::sync::Arc;

use crate::connection_manager::{ConnectionRecord, ConnectionState};
use crate::route_policy::{KnownRoute, RouteImplementation};

/// Exact installed instance of one independent optional route.
///
/// Every adapter callback must retain this tuple. The peer-session owner rejects
/// it after physical replacement, route replacement, or instance replacement.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct NativeOptionalRouteGeneration {
    pub transport_stable_id: u64,
    pub transport_generation: u64,
    pub route_generation: u64,
    /// Address of the installed `Arc` target. A retired callback retains its
    /// old allocation, so it cannot match a replacement route even when the
    /// logical connection and physical Iroh generation are unchanged.
    pub route_instance_id: usize,
}

/// Outcome of a terminal callback from an optional native route instance.
///
/// Handling the current instance and immediately restarting it are separate
/// facts: recovery can be intentionally deferred while the authoritative Iroh
/// path is still unknown or already optimal.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum NativeOptionalRouteTerminalDisposition {
    Stale,
    Handled { recovery_started: bool },
}

/// Generation registry for independently instantiable route adapters.
///
/// The generated route catalog is the taxonomy authority. Adding a future
/// adapter does not require another generation field or lifecycle switch.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub(crate) struct NativeOptionalRouteGenerations {
    routes: HashMap<KnownRoute, NativeOptionalRouteGeneration>,
}

pub(crate) type NativeRouteStartGate = Arc<tokio::sync::Mutex<()>>;
pub(crate) type NativeRouteStartGuard = tokio::sync::OwnedMutexGuard<()>;

/// Serializes mechanism construction per logical connection and route.
///
/// This is an operation gate, not a retry owner. The peer-session actor decides
/// whether and when to request another attempt.
#[derive(Default)]
pub(crate) struct NativeRouteStartGateRegistry {
    gates: tokio::sync::Mutex<HashMap<(String, KnownRoute), NativeRouteStartGate>>,
}

impl NativeRouteStartGateRegistry {
    pub(crate) async fn acquire(
        &self,
        connection_id: &str,
        route: KnownRoute,
    ) -> Option<(NativeRouteStartGate, NativeRouteStartGuard)> {
        if connection_id.trim().is_empty() || !is_independent_route_adapter(route) {
            return None;
        }
        let key = (connection_id.to_string(), route);
        let gate = {
            let mut gates = self.gates.lock().await;
            gates
                .entry(key)
                .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
                .clone()
        };
        let guard = gate.clone().lock_owned().await;
        Some((gate, guard))
    }

    pub(crate) async fn release(
        &self,
        connection_id: &str,
        route: KnownRoute,
        gate: NativeRouteStartGate,
        guard: NativeRouteStartGuard,
    ) {
        drop(guard);
        let key = (connection_id.to_string(), route);
        let mut gates = self.gates.lock().await;
        if gates
            .get(&key)
            .is_some_and(|current| Arc::ptr_eq(current, &gate))
            // The map and this release call are the only remaining owners, so
            // no waiter can still acquire this per-route start lock.
            && Arc::strong_count(&gate) == 2
        {
            gates.remove(&key);
        }
    }

    #[cfg(test)]
    pub(crate) async fn contains(&self, connection_id: &str, route: KnownRoute) -> bool {
        self.gates
            .lock()
            .await
            .contains_key(&(connection_id.to_string(), route))
    }
}

impl NativeOptionalRouteGenerations {
    pub(crate) fn get(&self, route: KnownRoute) -> Option<NativeOptionalRouteGeneration> {
        self.routes.get(&route).copied()
    }

    pub(crate) fn set(
        &mut self,
        route: KnownRoute,
        generation: Option<NativeOptionalRouteGeneration>,
    ) -> bool {
        if !is_independent_route_adapter(route) {
            return false;
        }
        match generation {
            Some(generation) => {
                self.routes.insert(route, generation);
            }
            None => {
                self.routes.remove(&route);
            }
        }
        true
    }

    pub(crate) fn is_empty(&self) -> bool {
        self.routes.is_empty()
    }

    pub(crate) fn sync_route_generation(
        &mut self,
        transport_stable_id: u64,
        transport_generation: u64,
        route_generation: u64,
    ) {
        for generation in self.routes.values_mut() {
            if generation.transport_stable_id == transport_stable_id
                && generation.transport_generation == transport_generation
            {
                generation.route_generation = route_generation;
            }
        }
    }
}

pub(crate) fn is_independent_route_adapter(route: KnownRoute) -> bool {
    let descriptor = route.descriptor();
    descriptor.implementation == RouteImplementation::RouteAdapter
        && descriptor.independently_instantiable
}

pub(crate) fn native_optional_route_generation_matches(
    record: Option<&ConnectionRecord>,
    expected: Option<NativeOptionalRouteGeneration>,
    route_instance_id: usize,
) -> bool {
    let (Some(record), Some(expected)) = (record, expected) else {
        return false;
    };
    expected.route_instance_id == route_instance_id
        && record.transport_stable_id == Some(expected.transport_stable_id)
        && record.transport_generation == expected.transport_generation
        && record.route_generation == expected.route_generation
        && matches!(
            record.state,
            ConnectionState::Connecting | ConnectionState::Connected
        )
}

pub(crate) fn accepted_native_route_event_generation(
    record: Option<&ConnectionRecord>,
    expected: Option<NativeOptionalRouteGeneration>,
    route_instance_id: usize,
) -> Option<super::NativePeerDataGeneration> {
    if !native_optional_route_generation_matches(record, expected, route_instance_id) {
        return None;
    }
    let expected = expected?;
    Some(super::NativePeerDataGeneration {
        transport_stable_id: expected.transport_stable_id,
        transport_generation: expected.transport_generation,
        route_generation: expected.route_generation,
    })
}

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

    fn record(state: ConnectionState) -> ConnectionRecord {
        ConnectionRecord {
            connection_id: "connection-a".to_string(),
            node_id: Some("node-b".to_string()),
            device_id: Some("device-b".to_string()),
            device_id_hint: None,
            endpoint_id: Some("node-b".to_string()),
            transport_generation: 3,
            route_generation: 7,
            transport_stable_id: Some(11),
            transport_source: Some("iroh".to_string()),
            active_transport: "iroh-relay".to_string(),
            parallel_transport: None,
            last_state_change_at_ms: 1,
            last_transport_change_at_ms: 1,
            last_route_change_at_ms: 1,
            state,
            status_reason: None,
            transition_count: 1,
            connecting_transition_count: 1,
            replacement_count: 0,
            retire_count: 0,
            last_disconnect_reason: None,
            last_reconnect_reason: None,
            created_at_ms: 1,
            updated_at_ms: 1,
        }
    }

    fn generation() -> NativeOptionalRouteGeneration {
        NativeOptionalRouteGeneration {
            transport_stable_id: 11,
            transport_generation: 3,
            route_generation: 7,
            route_instance_id: 13,
        }
    }

    #[test]
    fn registry_uses_the_generated_adapter_taxonomy() {
        let mut routes = NativeOptionalRouteGenerations::default();
        assert!(routes.set(KnownRoute::WebRtc, Some(generation())));
        assert!(routes.set(KnownRoute::Moq, Some(generation())));
        assert!(!routes.set(KnownRoute::WebRtcLan, Some(generation())));
        assert!(!routes.set(KnownRoute::IrohRelay, Some(generation())));
        assert_eq!(routes.get(KnownRoute::WebRtc), Some(generation()));
        assert_eq!(routes.get(KnownRoute::Moq), Some(generation()));
    }

    #[test]
    fn callback_fence_rejects_every_retired_generation_domain() {
        let current = record(ConnectionState::Connected);
        assert!(native_optional_route_generation_matches(
            Some(&current),
            Some(generation()),
            13,
        ));

        let mut stale = current.clone();
        stale.transport_stable_id = Some(12);
        assert!(!native_optional_route_generation_matches(
            Some(&stale),
            Some(generation()),
            13,
        ));
        stale = current.clone();
        stale.transport_generation += 1;
        assert!(!native_optional_route_generation_matches(
            Some(&stale),
            Some(generation()),
            13,
        ));
        stale = current.clone();
        stale.route_generation += 1;
        assert!(!native_optional_route_generation_matches(
            Some(&stale),
            Some(generation()),
            13,
        ));
        stale = record(ConnectionState::Closed);
        assert!(!native_optional_route_generation_matches(
            Some(&stale),
            Some(generation()),
            13,
        ));
        assert!(!native_optional_route_generation_matches(
            Some(&current),
            Some(generation()),
            14,
        ));
        assert_eq!(
            accepted_native_route_event_generation(Some(&current), Some(generation()), 13),
            Some(super::super::NativePeerDataGeneration {
                transport_stable_id: 11,
                transport_generation: 3,
                route_generation: 7,
            }),
        );
        assert_eq!(
            accepted_native_route_event_generation(Some(&current), Some(generation()), 14),
            None,
        );
    }

    #[test]
    fn route_generation_sync_updates_all_current_adapter_instances() {
        let mut routes = NativeOptionalRouteGenerations::default();
        routes.set(KnownRoute::WebRtc, Some(generation()));
        routes.set(KnownRoute::Moq, Some(generation()));
        routes.sync_route_generation(11, 3, 8);
        assert_eq!(routes.get(KnownRoute::WebRtc).unwrap().route_generation, 8);
        assert_eq!(routes.get(KnownRoute::Moq).unwrap().route_generation, 8);

        routes.sync_route_generation(99, 3, 9);
        assert_eq!(routes.get(KnownRoute::WebRtc).unwrap().route_generation, 8);
    }

    #[tokio::test]
    async fn route_start_gate_serializes_each_adapter_without_lost_wakeup() {
        let gates = Arc::new(NativeRouteStartGateRegistry::default());
        assert!(gates
            .acquire("connection-a", KnownRoute::IrohRelay)
            .await
            .is_none());
        let (gate, guard) = gates
            .acquire("connection-a", KnownRoute::Moq)
            .await
            .expect("MoQ adapter gate");
        let waiting_gates = gates.clone();
        let waiter =
            tokio::spawn(
                async move { waiting_gates.acquire("connection-a", KnownRoute::Moq).await },
            );

        tokio::task::yield_now().await;
        gates
            .release("connection-a", KnownRoute::Moq, gate, guard)
            .await;
        let (waiter_gate, waiter_guard) =
            tokio::time::timeout(std::time::Duration::from_secs(1), waiter)
                .await
                .expect("adapter waiter must wake")
                .expect("adapter waiter task")
                .expect("adapter waiter gate");
        gates
            .release("connection-a", KnownRoute::Moq, waiter_gate, waiter_guard)
            .await;
        assert!(!gates.contains("connection-a", KnownRoute::Moq).await);
    }
}