use std::collections::{HashMap, HashSet};
#[cfg(not(target_arch = "wasm32"))]
use std::time::{SystemTime, UNIX_EPOCH};
use super::state_machine::{connection_state_rank, default_health_for_state, strongest_state};
use super::types::{ConnectionHealth, ConnectionRecord, ConnectionStore, PeerSnapshot};
pub(super) fn unix_ms_now() -> i64 {
#[cfg(target_arch = "wasm32")]
{
return js_sys::Date::now() as i64;
}
#[cfg(not(target_arch = "wasm32"))]
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis() as i64)
.unwrap_or_default()
}
pub(super) fn add_index(
index: &mut HashMap<String, HashSet<String>>,
key: Option<&str>,
connection_id: &str,
) {
if let Some(key) = key {
index
.entry(key.to_string())
.or_default()
.insert(connection_id.to_string());
}
}
pub(super) fn remove_index(
index: &mut HashMap<String, HashSet<String>>,
key: Option<&str>,
connection_id: &str,
) {
if let Some(key) = key {
if let Some(values) = index.get_mut(key) {
values.remove(connection_id);
if values.is_empty() {
index.remove(key);
}
}
}
}
pub(super) fn add_indexes(store: &mut ConnectionStore, record: &ConnectionRecord) {
add_index(
&mut store.by_node_id,
record.node_id.as_deref(),
&record.connection_id,
);
add_index(
&mut store.by_device_id,
record.device_id.as_deref(),
&record.connection_id,
);
add_index(
&mut store.by_device_hint_id,
record.device_id_hint.as_deref(),
&record.connection_id,
);
add_index(
&mut store.by_endpoint_id,
record.endpoint_id.as_deref(),
&record.connection_id,
);
}
pub(super) fn canonical_peer_id(record: &ConnectionRecord) -> String {
record
.device_id
.clone()
.or_else(|| record.device_id_hint.clone())
.unwrap_or_else(|| record.connection_id.clone())
}
pub(super) fn collect_peer_ids(store: &ConnectionStore) -> HashSet<String> {
let mut peer_ids = HashSet::new();
for record in store.by_id.values() {
peer_ids.insert(canonical_peer_id(record));
}
peer_ids.extend(store.health_by_peer.keys().cloned());
peer_ids.extend(store.scopes_by_peer.keys().cloned());
peer_ids
}
pub(super) fn resolve_peer_id(store: &ConnectionStore, id: &str) -> Option<String> {
if let Some(record) = store.by_id.get(id) {
return Some(canonical_peer_id(record));
}
if store.health_by_peer.contains_key(id) || store.scopes_by_peer.contains_key(id) {
return Some(id.to_string());
}
if let Some(connection_ids) = store.by_device_id.get(id) {
if let Some(record) = select_preferred_record(store, connection_ids) {
return Some(canonical_peer_id(record));
}
}
if let Some(connection_ids) = store.by_device_hint_id.get(id) {
if let Some(record) = select_preferred_record(store, connection_ids) {
return Some(canonical_peer_id(record));
}
}
if let Some(connection_ids) = store.by_node_id.get(id) {
if let Some(record) = select_preferred_record(store, connection_ids) {
return Some(canonical_peer_id(record));
}
}
None
}
fn compare_record_priority(
left: &ConnectionRecord,
right: &ConnectionRecord,
) -> std::cmp::Ordering {
connection_state_rank(&right.state)
.cmp(&connection_state_rank(&left.state))
.then_with(|| right.updated_at_ms.cmp(&left.updated_at_ms))
.then_with(|| left.connection_id.cmp(&right.connection_id))
}
fn compare_active_record_priority(
left: &ConnectionRecord,
right: &ConnectionRecord,
) -> std::cmp::Ordering {
let left_connected = matches!(left.state, super::types::ConnectionState::Connected);
let right_connected = matches!(right.state, super::types::ConnectionState::Connected);
let left_live = left_connected && left.transport_stable_id.is_some();
let right_live = right_connected && right.transport_stable_id.is_some();
left_live
.cmp(&right_live)
.then_with(|| left_connected.cmp(&right_connected))
.then_with(|| connection_state_rank(&left.state).cmp(&connection_state_rank(&right.state)))
.then_with(|| left.transport_generation.cmp(&right.transport_generation))
.then_with(|| left.updated_at_ms.cmp(&right.updated_at_ms))
.then_with(|| right.connection_id.cmp(&left.connection_id))
}
fn select_preferred_record<'a>(
store: &'a ConnectionStore,
connection_ids: &HashSet<String>,
) -> Option<&'a ConnectionRecord> {
let mut records: Vec<&ConnectionRecord> = connection_ids
.iter()
.filter_map(|connection_id| store.by_id.get(connection_id))
.collect();
records.sort_by(|left, right| compare_record_priority(left, right));
records.into_iter().next()
}
pub(super) fn migrate_peer_state(
store: &mut ConnectionStore,
previous_peer_id: &str,
next_peer_id: &str,
) {
if previous_peer_id == next_peer_id {
return;
}
if let Some(previous_health) = store.health_by_peer.remove(previous_peer_id) {
let merged = merge_health(store.health_by_peer.remove(next_peer_id), previous_health);
store
.health_by_peer
.insert(next_peer_id.to_string(), merged);
}
if let Some(previous_scopes) = store.scopes_by_peer.remove(previous_peer_id) {
store
.scopes_by_peer
.entry(next_peer_id.to_string())
.or_default()
.extend(previous_scopes);
}
}
pub(super) fn merge_health(
current: Option<ConnectionHealth>,
incoming: ConnectionHealth,
) -> ConnectionHealth {
match current {
None | Some(ConnectionHealth::Unknown) => incoming,
Some(current) if incoming == ConnectionHealth::Unknown => current,
Some(_) => incoming,
}
}
pub(super) fn build_peer_snapshot(store: &ConnectionStore, peer_id: &str) -> Option<PeerSnapshot> {
let mut records: Vec<ConnectionRecord> = store
.by_id
.values()
.filter(|record| canonical_peer_id(record) == peer_id)
.cloned()
.collect();
records.sort_by(|left, right| compare_record_priority(left, right));
if records.is_empty()
&& !store.health_by_peer.contains_key(peer_id)
&& !store.scopes_by_peer.contains_key(peer_id)
{
return None;
}
let active_record = records
.iter()
.max_by(|left, right| compare_active_record_priority(left, right));
let device_id = records.iter().find_map(|record| record.device_id.clone());
let device_id_hint = records
.iter()
.find_map(|record| record.device_id_hint.clone());
let node_id = active_record
.and_then(|record| record.node_id.clone())
.or_else(|| records.iter().find_map(|record| record.node_id.clone()));
let active_transport_stable_id = active_record.and_then(|record| record.transport_stable_id);
let active_transport_generation = active_record
.map(|record| record.transport_generation)
.unwrap_or(0);
let active_route_generation = active_record
.map(|record| record.route_generation)
.unwrap_or(0);
let active_transport = active_record
.map(|record| record.active_transport.clone())
.unwrap_or_else(|| "iroh".to_string());
let parallel_transport = active_record.and_then(|record| record.parallel_transport.clone());
let last_route_change_at_ms = active_record
.map(|record| record.last_route_change_at_ms)
.unwrap_or_else(unix_ms_now);
let active_connection_id = active_record.map(|record| record.connection_id.clone());
let connection_ids = active_connection_id
.into_iter()
.chain(
records
.iter()
.filter(|record| {
active_record
.map(|active| active.connection_id != record.connection_id)
.unwrap_or(true)
})
.map(|record| record.connection_id.clone()),
)
.collect();
let status = strongest_state(&records);
let health = store
.health_by_peer
.get(peer_id)
.cloned()
.unwrap_or_else(|| default_health_for_state(&status));
let scopes = store
.scopes_by_peer
.get(peer_id)
.map(|scopes| scopes.iter().cloned().collect())
.unwrap_or_default();
let transition_count = records
.iter()
.map(|record| record.transition_count)
.max()
.unwrap_or(0);
let connecting_transition_count = records
.iter()
.map(|record| record.connecting_transition_count)
.max()
.unwrap_or(0);
let replacement_count = records
.iter()
.map(|record| record.replacement_count)
.max()
.unwrap_or(0);
let retire_count = records
.iter()
.map(|record| record.retire_count)
.max()
.unwrap_or(0);
let last_disconnect_reason = records
.iter()
.rev()
.find_map(|record| record.last_disconnect_reason.clone());
let last_reconnect_reason = records
.iter()
.rev()
.find_map(|record| record.last_reconnect_reason.clone());
let last_seen_at_ms = records
.iter()
.map(|record| record.updated_at_ms)
.max()
.unwrap_or_else(unix_ms_now);
let last_lifecycle_transition_at_ms = records
.iter()
.map(|record| {
record
.last_state_change_at_ms
.max(record.last_transport_change_at_ms)
.max(record.last_route_change_at_ms)
})
.max()
.unwrap_or_else(unix_ms_now);
let error = records
.iter()
.rev()
.find_map(|record| record.status_reason.clone());
Some(PeerSnapshot {
peer_id: peer_id.to_string(),
device_id,
device_id_hint,
node_id,
connection_ids,
active_transport_stable_id,
active_transport_generation,
active_route_generation,
active_transport,
parallel_transport,
last_route_change_at_ms,
last_lifecycle_transition_at_ms,
status,
health,
transition_count,
connecting_transition_count,
replacement_count,
retire_count,
last_disconnect_reason,
last_reconnect_reason,
scopes,
last_seen_at_ms,
error,
})
}