openrtc 1.0.4

use super::types::{ConnectionHealth, ConnectionRecord, ConnectionState};

pub(super) fn default_health_for_state(state: &ConnectionState) -> ConnectionHealth {
    match state {
        ConnectionState::Failed | ConnectionState::Closed | ConnectionState::Closing => {
            ConnectionHealth::Stale
        }
        ConnectionState::Connected | ConnectionState::Connecting | ConnectionState::Pending => {
            ConnectionHealth::Unknown
        }
    }
}

pub(super) fn is_active_state(state: &ConnectionState) -> bool {
    matches!(
        state,
        ConnectionState::Pending | ConnectionState::Connecting | ConnectionState::Connected
    )
}

pub(super) fn connection_state_rank(state: &ConnectionState) -> u8 {
    match state {
        ConnectionState::Connected => 5,
        ConnectionState::Connecting => 4,
        ConnectionState::Pending => 3,
        ConnectionState::Failed => 2,
        ConnectionState::Closing => 1,
        ConnectionState::Closed => 0,
    }
}

pub(super) fn strongest_state(records: &[ConnectionRecord]) -> ConnectionState {
    if records.is_empty() {
        return ConnectionState::Closed;
    }

    // Data-plane truth: any Connected record with a live transport (stable
    // transport id assigned) reflects an actual working data path and beats
    // any Failed records on the same peer. Failed records describe past dial
    // attempts that did not establish a transport, so they must not override a
    // live transport on the same peer pair (e.g. when the local side's
    // outbound dial failed but the remote successfully dialed in, or when a
    // parallel transport replacement succeeded after one attempt failed).
    let has_connected_with_transport = records.iter().any(|record| {
        matches!(record.state, ConnectionState::Connected) && record.transport_stable_id.is_some()
    });
    if has_connected_with_transport {
        return ConnectionState::Connected;
    }

    let has_failed = records
        .iter()
        .any(|record| matches!(record.state, ConnectionState::Failed));
    let has_connected = records
        .iter()
        .any(|record| matches!(record.state, ConnectionState::Connected));
    if has_failed && has_connected {
        let max_failed_at = records
            .iter()
            .filter(|record| matches!(record.state, ConnectionState::Failed))
            .map(|record| record.updated_at_ms)
            .max()
            .unwrap_or(0);
        let max_connected_at = records
            .iter()
            .filter(|record| matches!(record.state, ConnectionState::Connected))
            .map(|record| record.updated_at_ms)
            .max()
            .unwrap_or(0);
        // Stale "connected" rows (e.g. backfill / ghost without a live transport) must not win
        // over a newer failed dial (e.g. ensure_connected_addr timeout + set_failed on the
        // in-flight record). The data-plane-truth branch above already handles the case where
        // a live transport exists; here both candidates lack a stable transport.
        if max_failed_at >= max_connected_at {
            return ConnectionState::Failed;
        }
    }

    if records
        .iter()
        .any(|record| matches!(record.state, ConnectionState::Connected))
    {
        return ConnectionState::Connected;
    }
    if records
        .iter()
        .any(|record| matches!(record.state, ConnectionState::Connecting))
    {
        return ConnectionState::Connecting;
    }
    if records
        .iter()
        .any(|record| matches!(record.state, ConnectionState::Pending))
    {
        return ConnectionState::Pending;
    }
    if records
        .iter()
        .any(|record| matches!(record.state, ConnectionState::Failed))
    {
        return ConnectionState::Failed;
    }
    if records
        .iter()
        .any(|record| matches!(record.state, ConnectionState::Closing))
    {
        return ConnectionState::Closing;
    }
    ConnectionState::Closed
}

pub(super) fn merge_upsert_state(
    existing: &ConnectionState,
    incoming: &ConnectionState,
) -> ConnectionState {
    if is_active_state(existing) && is_active_state(incoming) {
        if connection_state_rank(existing) >= connection_state_rank(incoming) {
            return existing.clone();
        }
    }
    incoming.clone()
}

pub(super) fn merge_upsert_reason(
    existing: &ConnectionRecord,
    merged_state: &ConnectionState,
    incoming_state: &ConnectionState,
    incoming_reason: Option<String>,
) -> Option<String> {
    if merged_state == &existing.state && merged_state != incoming_state {
        return existing.status_reason.clone();
    }

    if incoming_reason.is_some() {
        incoming_reason
    } else if merged_state == &existing.state {
        existing.status_reason.clone()
    } else {
        None
    }
}

#[cfg(test)]
mod tests {
    //! Desktop connection-lifecycle resolution under churn.
    //!
    //! Wireless peers reconnect, fail, and re-dial constantly. The store can hold
    //! several `ConnectionRecord`s for one peer at once (an outbound dial that
    //! timed out, a successful inbound dial, a ghost backfill row, a parallel
    //! transport replacement). These tests assert that `strongest_state` /
    //! `merge_upsert_state` resolve those conflicting records to the data-plane
    //! truth — a live transport always wins, and a stale "connected" row never
    //! erases a newer failure (audit V5: non-authoritative state must not override
    //! stronger/fresher state).
    use super::*;
    use crate::connection_manager::types::{ConnectionRecord, ConnectionState};

    fn record(id: &str, state: ConnectionState, updated_at_ms: i64) -> ConnectionRecord {
        let mut r = ConnectionRecord::new(id.to_string(), Some("node-1".to_string()), None, None);
        r.state = state;
        r.updated_at_ms = updated_at_ms;
        r
    }

    fn with_live_transport(mut r: ConnectionRecord) -> ConnectionRecord {
        r.transport_stable_id = Some(42);
        r
    }

    #[test]
    fn empty_records_resolve_to_closed() {
        assert_eq!(strongest_state(&[]), ConnectionState::Closed);
    }

    #[test]
    fn live_connected_transport_beats_a_newer_failed_dial() {
        // Local outbound dial failed at t=200, but the remote dialed in and we hold
        // a live transport (connected at t=100). The working data path must win.
        let records = vec![
            with_live_transport(record("inbound", ConnectionState::Connected, 100)),
            record("outbound", ConnectionState::Failed, 200),
        ];
        assert_eq!(strongest_state(&records), ConnectionState::Connected);
    }

    #[test]
    fn stale_ghost_connected_does_not_override_a_newer_failed_dial() {
        // A "connected" row without a live transport (ghost/backfill) is older than
        // a fresh failed dial — the failure is the truth.
        let records = vec![
            record("ghost", ConnectionState::Connected, 100),
            record("dial", ConnectionState::Failed, 200),
        ];
        assert_eq!(strongest_state(&records), ConnectionState::Failed);
    }

    #[test]
    fn newer_connected_without_transport_beats_older_failed() {
        let records = vec![
            record("dial", ConnectionState::Failed, 100),
            record("reconnect", ConnectionState::Connected, 200),
        ];
        assert_eq!(strongest_state(&records), ConnectionState::Connected);
    }

    #[test]
    fn active_states_rank_above_terminal_states() {
        let records = vec![
            record("a", ConnectionState::Closing, 300),
            record("b", ConnectionState::Connecting, 100),
            record("c", ConnectionState::Pending, 50),
        ];
        // Connecting outranks Pending and Closing during a flapping reconnect.
        assert_eq!(strongest_state(&records), ConnectionState::Connecting);
    }

    #[test]
    fn merge_keeps_stronger_active_state_against_weaker_update() {
        // A late "connecting" update must not downgrade an established "connected".
        assert_eq!(
            merge_upsert_state(&ConnectionState::Connected, &ConnectionState::Connecting),
            ConnectionState::Connected,
        );
    }

    #[test]
    fn merge_upgrades_to_stronger_active_state() {
        assert_eq!(
            merge_upsert_state(&ConnectionState::Connecting, &ConnectionState::Connected),
            ConnectionState::Connected,
        );
    }

    #[test]
    fn merge_accepts_terminal_failure_over_active_state() {
        // A real failure/close is terminal and must be applied even though the
        // existing state was "active".
        assert_eq!(
            merge_upsert_state(&ConnectionState::Connected, &ConnectionState::Failed),
            ConnectionState::Failed,
        );
        assert_eq!(
            merge_upsert_state(&ConnectionState::Connecting, &ConnectionState::Closed),
            ConnectionState::Closed,
        );
    }

    #[test]
    fn terminal_states_map_to_stale_health() {
        for state in [
            ConnectionState::Failed,
            ConnectionState::Closed,
            ConnectionState::Closing,
        ] {
            assert_eq!(default_health_for_state(&state), ConnectionHealth::Stale);
        }
        for state in [
            ConnectionState::Connected,
            ConnectionState::Connecting,
            ConnectionState::Pending,
        ] {
            assert_eq!(default_health_for_state(&state), ConnectionHealth::Unknown);
        }
    }
}