diff --git a/mtp-type-maps b/mtp-type-maps index f753cfd..486541b 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit f753cfd406fc0050a8ccd76a90897205efceacd8 +Subproject commit 486541b9483356ff49ff3ec7016f87d3ecbeaa0e diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs index a333260..4712af0 100644 --- a/src/anonymous_clients/anonymous_client_connection.rs +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -8,7 +8,6 @@ use uuid::Uuid; use crate::anonymous_clients::anonymous_manager::{self, generate_username}; use crate::app_state::AppState; use crate::calls::call_group::call_invite_secret_from_cv; -use crate::data::user::UserStatus; use crate::rho::connection::{GeneralConnection, MtpReceiver, MtpSender}; use crate::util::data_type_id; use crate::util::logger::PrintType; @@ -222,9 +221,13 @@ impl AnonymousClientConnection { .await; } - // Handle client status changes - if cv.is_type(CommunicationType::ClientChanged) { - self.handle_client_changed(cv).await; + // Presence is account-scoped and anonymous sessions have no + // persisted account preference to change. + if cv.is_type(CommunicationType::SetUserState) + || cv.is_type(CommunicationType::ClientChanged) + { + self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoUserId) + .await; return; } @@ -321,22 +324,6 @@ impl AnonymousClientConnection { }); } - /// Handle client status change - async fn handle_client_changed(self: Arc, cv: CommunicationValue) { - if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) { - let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online); - self.state - .omega - .client_changed( - self.user_id as i64, - self.user_id as i64, - self.user_id as i64, - user_status, - ) - .await; - } - } - /// Handle call invite async fn handle_call_invite(self: Arc, cv: CommunicationValue) { let receiver_id: i64 = cv.get_data(DataType::ReceiverId).as_number().unwrap_or(0) as i64; diff --git a/src/config.rs b/src/config.rs index 7c10773..484da67 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,10 +1,12 @@ -use std::env; +use std::{env, time::Duration}; use thiserror::Error; const DEFAULT_RHO_PORT: u16 = 443; const DEFAULT_OMEGA_HOST: &str = "tensamin.net"; const DEFAULT_OMEGA_PORT: u16 = 9187; +const DEFAULT_OMEGA_SYNC_TIMEOUT_SECONDS: u64 = 20; +const DEFAULT_OMEGA_SYNC_RETRIES: u32 = 3; #[derive(Clone, Debug, Eq, PartialEq)] pub struct LiveKitConfig { @@ -19,6 +21,11 @@ pub struct Config { pub omega_host: String, pub omega_port: u16, pub omikron_id: u64, + /// Maximum duration of one route or subscription synchronization request. + pub omega_sync_timeout: Duration, + /// Number of synchronization requests before the transport is closed and + /// the normal reconnect loop starts. + pub omega_sync_retries: u32, pub livekit: Option, } @@ -38,6 +45,15 @@ impl Config { let rho_port = parse_or_default("RHO_PORT", DEFAULT_RHO_PORT)?; let omega_port = parse_or_default("OMEGA_PORT", DEFAULT_OMEGA_PORT)?; let omikron_id = parse_or_default("ID", 0_u64)?; + // Each synchronization request is bounded by this timeout. After + // omega_sync_retries attempts, Omikron closes the authenticated + // transport so the normal reconnect loop can establish a clean state. + let omega_sync_timeout = Duration::from_secs(parse_or_default( + "OMEGA_SYNC_TIMEOUT_SECONDS", + DEFAULT_OMEGA_SYNC_TIMEOUT_SECONDS, + )?); + let omega_sync_retries = + parse_or_default("OMEGA_SYNC_RETRIES", DEFAULT_OMEGA_SYNC_RETRIES)?.max(1); let omega_host = env::var("OMEGA_HOST") .unwrap_or_else(|_| DEFAULT_OMEGA_HOST.to_string()) .trim() @@ -55,6 +71,8 @@ impl Config { omega_host, omega_port, omikron_id, + omega_sync_timeout, + omega_sync_retries, livekit: livekit_from_environment()?, }) } diff --git a/src/data/user.rs b/src/data/user.rs index 8587ffe..1eda0ba 100644 --- a/src/data/user.rs +++ b/src/data/user.rs @@ -1,7 +1,4 @@ -use strum::IntoEnumIterator; -use strum_macros::EnumIter; - -#[derive(Debug, Clone, PartialEq, EnumIter, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] #[allow(unused, non_camel_case_types)] pub enum UserStatus { user_offline, @@ -20,12 +17,40 @@ impl UserStatus { pub fn to_string(&self) -> String { format!("{:?}", self) } - pub fn from_str(s: &str) -> Option { - for sel in UserStatus::iter() { - if &sel.to_string() == s { - return Some(sel); - } + pub fn from_client_preference(s: &str) -> Option { + match s { + "user_online" => Some(Self::user_online), + "user_idle" => Some(Self::user_idle), + "user_dnd" => Some(Self::user_dnd), + "user_wc" => Some(Self::user_wc), + "user_invisible" => Some(Self::user_invisible), + _ => None, } - None + } + + pub fn public_value(&self) -> Self { + match self { + Self::user_invisible => Self::user_offline, + value => value.clone(), + } + } + + pub fn from_str(s: &str) -> Option { + Self::from_client_preference(s) + } +} + +#[cfg(test)] +mod tests { + use super::UserStatus; + + #[test] + fn network_parser_rejects_derived_states() { + assert_eq!(UserStatus::from_str("user_offline"), None); + assert_eq!(UserStatus::from_str("iota_offline"), None); + assert_eq!( + UserStatus::from_str("user_online"), + Some(UserStatus::user_online) + ); } } diff --git a/src/omega/capabilities.rs b/src/omega/capabilities.rs new file mode 100644 index 0000000..b658068 --- /dev/null +++ b/src/omega/capabilities.rs @@ -0,0 +1,145 @@ +use std::collections::BTreeSet; + +const OMIKRON_PREFIX: &str = "omikron;caps="; +const OMEGA_PREFIX: &str = "omega;caps="; +const SET_USER_STATE: &str = "set_user_state_v1"; +const STATE_SUBSCRIBE: &str = "state_subscribe_v1"; +const SESSION_SNAPSHOT: &str = "session_snapshot_v1"; +const CLIENT_STATE_PUSH: &str = "client_state_push_v1"; + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct PeerCapabilities { + pub set_user_state_v1: bool, + pub state_subscribe_v1: bool, + pub session_snapshot_v1: bool, + pub client_state_push_v1: bool, +} + +impl PeerCapabilities { + pub fn current() -> Self { + Self { + set_user_state_v1: true, + state_subscribe_v1: true, + session_snapshot_v1: true, + client_state_push_v1: true, + } + } + + pub fn identification_description(&self) -> String { + let mut names = Vec::new(); + if self.set_user_state_v1 { + names.push(SET_USER_STATE); + } + if self.state_subscribe_v1 { + names.push(STATE_SUBSCRIBE); + } + if self.session_snapshot_v1 { + names.push(SESSION_SNAPSHOT); + } + if self.client_state_push_v1 { + names.push(CLIENT_STATE_PUSH); + } + format!("{OMIKRON_PREFIX}{}", names.join(",")) + } + + pub fn from_identification_description(description: Option<&str>) -> Result { + parse_capabilities(description, OMIKRON_PREFIX) + } + + pub fn from_omega_identification_description(description: Option<&str>) -> Result { + // A missing Omega descriptor selects the documented legacy wire + // contracts: tuple route snapshots, GetStates-only subscriptions, + // and ClientChanged pushes. + parse_capabilities(description, OMEGA_PREFIX) + } +} + +fn parse_capabilities(description: Option<&str>, prefix: &str) -> Result { + let Some(description) = description else { + return Ok(PeerCapabilities::default()); + }; + if description == "omega" { + return Ok(PeerCapabilities::default()); + } + let Some(capabilities) = description.strip_prefix(prefix) else { + return Err(()); + }; + let mut seen = BTreeSet::new(); + for capability in capabilities.split(',') { + if capability.is_empty() || !seen.insert(capability) { + return Err(()); + } + } + if seen.iter().any(|capability| { + !matches!( + *capability, + SET_USER_STATE | STATE_SUBSCRIBE | SESSION_SNAPSHOT | CLIENT_STATE_PUSH + ) + }) { + return Err(()); + } + Ok(PeerCapabilities { + set_user_state_v1: seen.contains(SET_USER_STATE), + state_subscribe_v1: seen.contains(STATE_SUBSCRIBE), + session_snapshot_v1: seen.contains(SESSION_SNAPSHOT), + client_state_push_v1: seen.contains(CLIENT_STATE_PUSH), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn capability_description_round_trips() { + let capabilities = PeerCapabilities::current(); + assert_eq!( + PeerCapabilities::from_identification_description(Some( + &capabilities.identification_description() + )), + Ok(capabilities) + ); + } + + #[test] + fn malformed_capabilities_are_rejected() { + assert!( + PeerCapabilities::from_identification_description(Some("omikron;caps=unknown_v9")) + .is_err() + ); + assert!( + PeerCapabilities::from_identification_description(Some( + "omikron;caps=state_subscribe_v1,state_subscribe_v1" + )) + .is_err() + ); + assert!(PeerCapabilities::from_identification_description(Some("omikron;caps=")).is_err()); + } + + #[test] + fn missing_capabilities_mean_legacy_peer() { + assert_eq!( + PeerCapabilities::from_identification_description(None), + Ok(PeerCapabilities::default()) + ); + assert_eq!( + PeerCapabilities::from_omega_identification_description(Some("omega")), + Ok(PeerCapabilities::default()) + ); + } + + #[test] + fn omega_capabilities_use_the_omega_namespace() { + assert!( + PeerCapabilities::from_omega_identification_description(Some( + "omega;caps=set_user_state_v1" + )) + .unwrap() + .set_user_state_v1 + ); + assert!( + PeerCapabilities::from_identification_description(Some("omega;caps=set_user_state_v1")) + .is_err() + ); + } +} diff --git a/src/omega/mod.rs b/src/omega/mod.rs index cbe77d9..4108cc6 100644 --- a/src/omega/mod.rs +++ b/src/omega/mod.rs @@ -1 +1,2 @@ +pub mod capabilities; pub mod omega_connection; diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index 78a2d38..e66e62c 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -1,6 +1,9 @@ +use super::capabilities::PeerCapabilities; use crate::{ - config::Config, data::user::UserStatus, log_cv_in, log_cv_out, log_err, log_in, log_out, - rho::rho_manager::RhoManager, util::logger::PrintType, + config::Config, + log_cv_in, log_cv_out, log_err, log_in, log_out, + rho::rho_manager::RhoManager, + util::{data_type_id, logger::PrintType}, }; use dashmap::DashMap; use mtp::client::{Client, MTPConnection, Sender}; @@ -13,7 +16,7 @@ use mtp::{ use mtp_transport::ConnectionHandle; use std::{sync::Arc, time::Duration}; use tokio::{ - sync::{Mutex, RwLock, mpsc, watch}, + sync::{Mutex, RwLock, mpsc, oneshot, watch}, task::JoinHandle, time::{Instant, sleep}, }; @@ -23,6 +26,7 @@ use uuid::Uuid; const RECONNECT_DELAY: Duration = Duration::from_secs(5); const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(300); const CONNECTION_TIMEOUT: Duration = Duration::from_secs(10); +const CAPABILITY_NEGOTIATION_TIMEOUT: Duration = Duration::from_secs(1); const PING_INTERVAL: Duration = Duration::from_secs(5); const TASK_CLEANUP_INTERVAL: Duration = Duration::from_secs(60); const TASK_MAX_AGE: Duration = Duration::from_secs(60); @@ -30,6 +34,37 @@ const MAX_CONCURRENT_REQUESTS: usize = 128; const CIRCUIT_BREAKER_FAILURE_THRESHOLD: u32 = 3; const CIRCUIT_BREAKER_COOLDOWN: Duration = Duration::from_secs(30); +fn client_changed_target(value: &CommunicationValue) -> Option<(i64, i64)> { + if !value.is_type(CommunicationType::ClientChanged) { + return None; + } + let receiver = i64::try_from(value.get_receiver()) + .ok() + .filter(|id| *id > 0)?; + let session_id = value + .get_data(DataType::SessionId) + .as_signed_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| *id > 0)?; + value + .get_data(DataType::UserId) + .as_signed_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| *id > 0)?; + value.get_data(DataType::UserState).as_str()?; + Some((receiver, session_id)) +} + +fn parse_omega_capability_response(value: &CommunicationValue) -> Result { + if !value.is_type(CommunicationType::IdentificationResponse) { + return Err("not an identification response".to_string()); + } + PeerCapabilities::from_omega_identification_description( + value.get_data(DataType::Description).as_str(), + ) + .map_err(|_| "Omega advertised an invalid capability set".to_string()) +} + // ============================================================================ // Waiting Task System // ============================================================================ @@ -85,24 +120,33 @@ pub fn start_task_cleanup_loop(omega: Arc) { pub enum ConnectionState { Disconnected, Connecting, - Connected { identified: bool }, + Identifying, + SynchronizingRoutes, + SynchronizingSubscriptions, + Ready, } #[allow(unused_variables)] impl ConnectionState { pub fn is_connected(&self) -> bool { - match self { - ConnectionState::Connected { identified } => true, - _ => false, - } + !matches!( + self, + ConnectionState::Disconnected | ConnectionState::Connecting + ) } #[allow(dead_code)] pub fn is_identified(&self) -> bool { - match self { - ConnectionState::Connected { identified: true } => true, - _ => false, - } + matches!( + self, + ConnectionState::SynchronizingRoutes + | ConnectionState::SynchronizingSubscriptions + | ConnectionState::Ready + ) + } + + pub fn is_ready(&self) -> bool { + matches!(self, ConnectionState::Ready) } } @@ -117,6 +161,8 @@ pub struct OmegaConnection { host: String, port: u16, omikron_id: u64, + sync_timeout: Duration, + sync_retries: u32, pub connection_id: Uuid, shutdown_tx: Arc>>>, // Track if we should reconnect on close @@ -126,6 +172,14 @@ pub struct OmegaConnection { keyring: mtp::crypto::Keyring, rho: Arc, waiting_tasks: DashMap, + session_presence: DashMap<(i64, i64), SessionPresenceState>, + peer_capabilities: Arc>, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct SessionPresenceState { + pub subscribed_user_ids: Vec, + pub pending: bool, } impl OmegaConnection { @@ -143,6 +197,8 @@ impl OmegaConnection { host: config.omega_host.clone(), port: config.omega_port, omikron_id: config.omikron_id, + sync_timeout: config.omega_sync_timeout, + sync_retries: config.omega_sync_retries, connection_id: Uuid::new_v4(), shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))), reconnect_on_close: Arc::new(RwLock::new(true)), @@ -154,6 +210,8 @@ impl OmegaConnection { keyring, rho, waiting_tasks: DashMap::new(), + session_presence: DashMap::new(), + peer_capabilities: Arc::new(RwLock::new(PeerCapabilities::default())), } } @@ -267,6 +325,7 @@ impl OmegaConnection { let client_config = ClientConfig::new(format!("https://{}:{}", self.host, self.port)) .with_client_id(self.omikron_id) + .with_description(PeerCapabilities::current().identification_description()) .with_policy( Policy::default() .with_send_mode(SendMode::SingleStreamPerMessage) @@ -302,18 +361,43 @@ impl OmegaConnection { // Store sender let sender_arc = Arc::new(connection.sender.clone()); *self.sender.write().await = Some(sender_arc.clone()); - *self.state.write().await = ConnectionState::Connected { identified: false }; + // A new authenticated transport gets a fresh request circuit. A + // previous transport can have failed while synchronizing and must not + // prevent this connection from making its initial requests. + self.circuit_breaker.lock().await.record_success(); + *self.state.write().await = ConnectionState::Identifying; // Get handle for close monitoring let sender_handle = sender_arc.handle().clone(); let connection = Arc::new(connection); + let (capability_tx, capability_rx) = oneshot::channel(); // Start read loop let read_self = self.clone(); let read_handle = tokio::spawn(async move { - read_self.read_loop(connection, sender_handle).await; + read_self + .read_loop(connection, sender_handle, Some(capability_tx)) + .await; }); + let negotiated_capabilities = + match tokio::time::timeout(CAPABILITY_NEGOTIATION_TIMEOUT, capability_rx).await { + Ok(Ok(Ok(capabilities))) => capabilities, + Ok(Ok(Err(error))) => { + self.fail_synchronization("capability negotiation", error) + .await; + let _ = read_handle.await; + return Err("Omega capability negotiation failed".to_string()); + } + Ok(Err(_)) | Err(_) => { + // An Omega from before capability negotiation sends no second + // identification response. Fall back to its documented + // legacy contracts rather than guessing modern support. + PeerCapabilities::default() + } + }; + *self.peer_capabilities.write().await = negotiated_capabilities; + // Tell omega our current state now that we're actually connected - // doing this after teardown (as before) sent into a sender that had // already been cleared, silently dropping the sync every time. @@ -354,8 +438,10 @@ impl OmegaConnection { } } - async fn sync_client_iota_status(self: Arc) { + async fn sync_client_iota_status(&self) { + *self.state.write().await = ConnectionState::SynchronizingRoutes; let mut connected_iota_ids: Vec = Vec::new(); + let mut connected_sessions: Vec = Vec::new(); let mut connected_user_ids: Vec = Vec::new(); let iota_ids = self.rho.iota_ids().await; @@ -367,21 +453,120 @@ impl OmegaConnection { for rho in rho_connections { for client_conn in rho.get_client_connections().await { - connected_user_ids.push(DataValue::SignedNumber( - client_conn.get_user_id().await.into(), - )); + let type_map = mtp::codec::TypeMap::latest(); + let user_id = DataValue::SignedNumber(client_conn.get_user_id().await.into()); + let session_id = DataValue::SignedNumber((client_conn.session_id as i128).into()); + let iota_id = DataValue::SignedNumber((rho.get_iota_id().await as i128).into()); + let peer_capabilities = self.peer_capabilities.read().await.clone(); + if peer_capabilities.session_snapshot_v1 { + connected_sessions.push(DataValue::Container(vec![ + (data_type_id(DataType::UserId, &type_map), user_id), + (data_type_id(DataType::SessionId, &type_map), session_id), + (data_type_id(DataType::IotaId, &type_map), iota_id), + ])); + } else { + connected_user_ids.push(user_id); + } } } - let sync_msg = CommunicationValue::new(CommunicationType::SyncClientIotaStatus) - .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids)) - .add_typed_default(DataType::UserIds, DataValue::Array(connected_user_ids)) - .add_typed_default( - DataType::RhoConnections, - DataValue::SignedNumber(self.rho.connection_count().await as i128), - ); + let peer_capabilities = self.peer_capabilities.read().await.clone(); + let sync_msg = if peer_capabilities.session_snapshot_v1 { + CommunicationValue::new(CommunicationType::SyncClientIotaStatus) + .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids)) + .add_typed_default(DataType::UserStates, DataValue::Array(connected_sessions)) + .add_typed_default( + DataType::RhoConnections, + DataValue::SignedNumber(self.rho.connection_count().await as i128), + ) + } else { + // Legacy Omega has no request/response contract for this message. + // Its exact snapshot is IotaIds + UserIds + RhoConnections. + CommunicationValue::new(CommunicationType::SyncClientIotaStatus) + .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids)) + .add_typed_default(DataType::UserIds, DataValue::Array(connected_user_ids)) + .add_typed_default( + DataType::RhoConnections, + DataValue::SignedNumber(self.rho.connection_count().await as i128), + ) + }; - self.send_message(&sync_msg).await; + if !peer_capabilities.session_snapshot_v1 { + if let Err(error) = self.send_message_result(&sync_msg).await { + self.fail_synchronization("legacy route synchronization", error) + .await; + return; + } + *self.state.write().await = ConnectionState::SynchronizingSubscriptions; + if let Err(error) = self.restore_presence_subscriptions().await { + self.fail_synchronization("subscription restoration", error) + .await; + return; + } + *self.state.write().await = ConnectionState::Ready; + return; + } + + let mut response = Err("route synchronization did not start".to_string()); + for attempt in 0..self.sync_retries { + response = self + .await_response(&sync_msg, Some(self.sync_timeout)) + .await; + match &response { + Ok(value) if value.is_type(CommunicationType::Success) => break, + Ok(value) if value.is_type(CommunicationType::ErrorInvalidData) => break, + _ if attempt + 1 < self.sync_retries => { + sleep(Duration::from_millis(250 * 2u64.pow(attempt))).await + } + _ => break, + } + } + match response { + Ok(response) if response.is_type(CommunicationType::Success) => { + *self.state.write().await = ConnectionState::SynchronizingSubscriptions; + if let Err(error) = self.restore_presence_subscriptions().await { + log_err!( + 0, + PrintType::Omega, + "Presence subscription restoration failed: {}", + error + ); + self.fail_synchronization("subscription restoration", error) + .await; + return; + } + *self.state.write().await = ConnectionState::Ready; + } + Ok(response) => { + let error = format!( + "Omega rejected presence snapshot with {}", + response.get_type() + ); + self.fail_synchronization("route synchronization", error) + .await; + return; + } + Err(error) => { + self.fail_synchronization("route synchronization", error) + .await; + return; + } + } + } + + async fn fail_synchronization(&self, phase: &str, error: String) { + log_err!( + 0, + PrintType::Omega, + "Omega {} failed after {} attempts: {}; closing transport for reconnect", + phase, + self.sync_retries, + error + ); + *self.state.write().await = ConnectionState::Disconnected; + if let Some(sender) = self.sender.read().await.as_ref() { + sender.close().await; + } } // ------------------------------------------------------------------------- @@ -392,6 +577,7 @@ impl OmegaConnection { self: Arc, connection: Arc, sender_handle: Arc, + mut capability_tx: Option>>, ) { // Monitor both receiver and sender handle for close let mut close_rx = sender_handle.subscribe_close(); @@ -403,6 +589,14 @@ impl OmegaConnection { Ok(cv) => { log_cv_in!(PrintType::Omega, &cv); + if cv.is_type(CommunicationType::IdentificationResponse) { + if let Some(tx) = capability_tx.take() { + let result = parse_omega_capability_response(&cv); + let _ = tx.send(result); + } + continue; + } + let msg_id = cv.get_id(); if let Some((_, task)) = self.waiting_tasks.remove(&msg_id) { if (task.task)(self.clone(), cv.clone()) { @@ -430,6 +624,20 @@ impl OmegaConnection { } } } + if cv.is_type(CommunicationType::ClientChanged) { + let Some((receiver, session_id)) = client_changed_target(&cv) + else { + continue; + }; + if let Some(client) = self + .rho + .get_client_connection(receiver, session_id) + .await + { + client.send_message(&cv).await; + } + continue; + } } Err(e) => { log_err!(0, PrintType::Omega, "Receive error: {}", e); @@ -488,6 +696,14 @@ impl OmegaConnection { let _ = self.try_send_message(cv).await; } + pub async fn send_message_result(&self, cv: &CommunicationValue) -> Result<(), String> { + self.try_send_message(cv).await + } + + pub async fn supports_set_user_state(&self) -> bool { + self.peer_capabilities.read().await.set_user_state_v1 + } + pub async fn await_connection(&self, timeout_duration: Option) -> Result<(), String> { if self.state.read().await.is_connected() { return Ok(()); @@ -607,72 +823,371 @@ impl OmegaConnection { self.state.read().await.is_identified() } + pub async fn is_ready(&self) -> bool { + self.state.read().await.is_ready() + } + #[allow(dead_code)] - pub async fn close_iota(&self, iota_id: i64) { + pub async fn close_iota(&self, iota_id: i64) -> Result<(), String> { let cv = CommunicationValue::new(CommunicationType::IotaDisconnected) .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())); - self.send_message(&cv).await; + let request_id = cv.get_id(); + let result = self.lifecycle_request(&cv).await; + if let Err(error) = &result { + self.log_lifecycle_failure("IotaDisconnected", iota_id, None, request_id, error); + self.sync_client_iota_status().await; + } + result } - pub async fn client_changed( - &self, - _iota_id: i64, - user_id: i64, - session_id: i64, - state: UserStatus, - ) { - let cv = CommunicationValue::new(CommunicationType::ClientChanged) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) - .add_typed_default( - DataType::SessionId, - DataValue::SignedNumber(session_id.into()), - ) - .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())); - self.send_message(&cv).await; - } - - pub async fn client_disconnected(&self, user_id: i64, session_id: i64) { + pub async fn client_disconnected(&self, user_id: i64, session_id: i64) -> Result<(), String> { + self.session_presence.remove(&(user_id, session_id)); let cv = CommunicationValue::new(CommunicationType::UserDisconnected) .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) .add_typed_default( DataType::SessionId, DataValue::SignedNumber(session_id.into()), ); - self.send_message(&cv).await; + let request_id = cv.get_id(); + let result = self.lifecycle_request(&cv).await; + if let Err(error) = &result { + self.log_lifecycle_failure( + "UserDisconnected", + user_id, + Some(session_id), + request_id, + error, + ); + self.sync_client_iota_status().await; + } + result } - pub async fn user_states(&self, user_id: i64, user_ids: Vec) { - let user_ids = user_ids - .iter() - .map(|v| DataValue::SignedNumber((*v).into())) - .collect(); - - let cv = CommunicationValue::new(CommunicationType::GetStates) - .add_typed_default(DataType::UserIds, DataValue::Array(user_ids)); - let msg_id = cv.get_id(); - - let rho_manager = self.rho.clone(); - self.waiting_tasks.insert( - msg_id, - WaitingTask { - task: Box::new( - move |_: Arc, response: CommunicationValue| { - let rho_manager = rho_manager.clone(); - tokio::spawn(async move { - let rho = rho_manager.get_for_user(user_id).await; - if let Some(rho) = rho { - for client in rho.get_client_connections_for_user(user_id).await { - client.send_message(&response).await; - } - } - }); - true - }, + pub async fn user_connected( + &self, + user_id: i64, + session_id: i64, + iota_id: i64, + ) -> Result<(), String> { + let request = CommunicationValue::new(CommunicationType::UserConnected) + .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(session_id.into()), + ) + .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())) + .add_typed_default( + DataType::UserState, + DataValue::Str("user_online".to_string()), + ) + .add_typed_default( + DataType::UpdatedAt, + DataValue::SignedNumber( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i128, ), - inserted_at: Instant::now(), + ); + let request_id = request.get_id(); + let result = self.lifecycle_request(&request).await; + if let Err(error) = &result { + self.log_lifecycle_failure( + "UserConnected", + user_id, + Some(session_id), + request_id, + error, + ); + self.sync_client_iota_status().await; + } + result + } + + pub async fn iota_connected(&self, iota_id: i64) -> Result<(), String> { + let request = CommunicationValue::new(CommunicationType::IotaConnected) + .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())); + let request_id = request.get_id(); + let result = self.lifecycle_request(&request).await; + if let Err(error) = &result { + self.log_lifecycle_failure("IotaConnected", iota_id, None, request_id, error); + self.sync_client_iota_status().await; + } + result + } + + pub async fn reconcile_routes(&self) { + self.sync_client_iota_status().await; + } + + async fn lifecycle_request(&self, request: &CommunicationValue) -> Result<(), String> { + if !self.is_ready().await { + return Err("Omega is not synchronized".to_string()); + } + if !self.peer_capabilities.read().await.session_snapshot_v1 { + // Legacy Omega handles lifecycle messages as notifications and + // never returns Success. A transport-level send result is the + // only acknowledgement available under that contract. + return self.send_message_result(request).await; + } + let response = self + .await_response(request, Some(Duration::from_secs(20))) + .await?; + if response.is_type(CommunicationType::Success) { + Ok(()) + } else { + Err(format!("Omega rejected {}", response.get_type())) + } + } + + fn log_lifecycle_failure( + &self, + operation: &str, + entity_id: i64, + session_id: Option, + request_id: u32, + error: &str, + ) { + log_err!( + 0, + PrintType::Omega, + "Lifecycle operation={} entity_id={} session_id={:?} request_id={} failed: {}", + operation, + entity_id, + session_id, + request_id, + error + ); + } + + pub async fn replace_state_subscription( + &self, + user_id: i64, + session_id: i64, + mut user_ids: Vec, + ) -> Result<(), String> { + if user_id <= 0 || session_id <= 0 { + return Err("invalid user or session id".to_string()); + } + + user_ids.sort_unstable(); + user_ids.dedup(); + + let key = (user_id, session_id); + // Store the latest authoritative snapshot before checking readiness. + // A client can synchronize while the transport is reconnecting. + self.session_presence.insert( + key, + SessionPresenceState { + subscribed_user_ids: user_ids.clone(), + pending: true, }, ); - self.send_message(&cv).await; + if !self.is_ready().await { + return Err("Omega is not synchronized".to_string()); + } + + self.deliver_state_subscription(key, user_ids).await + } + + async fn deliver_state_subscription( + &self, + key: (i64, i64), + user_ids: Vec, + ) -> Result<(), String> { + let (user_id, session_id) = key; + + if !self.peer_capabilities.read().await.state_subscribe_v1 { + let delivered_ids = user_ids.clone(); + let result = self + .request_user_states(user_id, session_id, user_ids) + .await; + if result.is_ok() { + self.mark_presence_delivered(key, &delivered_ids).await; + } + return result; + } + + let subscription = CommunicationValue::new(CommunicationType::StateSubscribe) + .with_sender(user_id as u64) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(session_id.into()), + ) + .add_typed_default( + DataType::UserIds, + DataValue::Array( + user_ids + .iter() + .map(|user_id| DataValue::SignedNumber((*user_id).into())) + .collect(), + ), + ); + let response = self + .await_response(&subscription, Some(Duration::from_secs(20))) + .await?; + if !response.is_type(CommunicationType::Success) { + return Err(format!( + "StateSubscribe rejected with {}", + response.get_type() + )); + } + + let delivered_ids = user_ids.clone(); + self.request_user_states(user_id, session_id, user_ids) + .await?; + self.mark_presence_delivered(key, &delivered_ids).await; + Ok(()) + } + + async fn restore_presence_subscriptions(&self) -> Result<(), String> { + let mut active = Vec::new(); + for rho in self.rho.connections().await { + active.extend(rho.get_client_connections().await); + } + let active_keys: std::collections::HashSet<_> = active + .iter() + .map(|client| (client.user_id as i64, client.session_id as i64)) + .collect(); + + // Do not replay subscriptions for sessions that disappeared while the + // transport was down. + self.session_presence + .retain(|key, _| active_keys.contains(key)); + + for key in active_keys { + let Some(snapshot) = self.session_presence.get(&key).map(|entry| entry.clone()) else { + continue; + }; + if !snapshot.pending { + continue; + } + self.deliver_state_subscription(key, snapshot.subscribed_user_ids) + .await?; + } + Ok(()) + } + + async fn mark_presence_delivered(&self, key: (i64, i64), delivered_ids: &[i64]) { + if let Some(mut entry) = self.session_presence.get_mut(&key) + && entry.subscribed_user_ids == delivered_ids + { + entry.pending = false; + } + } + + async fn request_user_states( + &self, + user_id: i64, + session_id: i64, + user_ids: Vec, + ) -> Result<(), String> { + let request = CommunicationValue::new(CommunicationType::GetStates) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(session_id.into()), + ) + .add_typed_default( + DataType::UserIds, + DataValue::Array( + user_ids + .iter() + .map(|user_id| DataValue::SignedNumber((*user_id).into())) + .collect(), + ), + ); + let response = self + .await_response(&request, Some(Duration::from_secs(20))) + .await?; + let legacy_peer = !self.peer_capabilities.read().await.client_state_push_v1; + let response = + if legacy_peer && response.get_data(DataType::SessionId).as_number().is_none() { + response.add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(session_id.into()), + ) + } else { + response + }; + if !legacy_peer + && response.get_data(DataType::SessionId).as_number() != Some(session_id as i128) + { + return Err("GetStates response had the wrong session".to_string()); + } + + if let Some(client) = self.rho.get_client_connection(user_id, session_id).await { + client.send_message(&response).await; + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::{PeerCapabilities, client_changed_target, parse_omega_capability_response}; + use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; + + fn notification() -> CommunicationValue { + CommunicationValue::new(CommunicationType::ClientChanged) + .with_receiver(7) + .add_typed_default(DataType::SessionId, DataValue::SignedNumber(11)) + .add_typed_default(DataType::UserId, DataValue::SignedNumber(20)) + .add_typed_default(DataType::UserState, DataValue::Str("user_online".into())) + } + + #[test] + fn client_changed_target_requires_all_routing_fields() { + assert_eq!(client_changed_target(¬ification()), Some((7, 11))); + + let mut missing_session = notification(); + missing_session.remove_data(DataType::SessionId); + assert_eq!(client_changed_target(&missing_session), None); + + let mut missing_state = notification(); + missing_state.remove_data(DataType::UserState); + assert_eq!(client_changed_target(&missing_state), None); + } + + #[test] + fn client_changed_target_rejects_non_positive_ids() { + let invalid = CommunicationValue::new(CommunicationType::ClientChanged) + .with_receiver(0) + .add_typed_default(DataType::SessionId, DataValue::SignedNumber(11)) + .add_typed_default(DataType::UserId, DataValue::SignedNumber(20)) + .add_typed_default(DataType::UserState, DataValue::Str("user_online".into())); + assert_eq!(client_changed_target(&invalid), None); + } + + #[test] + fn capability_response_negotiates_new_omega() { + let response = CommunicationValue::new(CommunicationType::IdentificationResponse) + .add_typed_default( + DataType::Description, + DataValue::Str( + "omega;caps=set_user_state_v1,state_subscribe_v1,session_snapshot_v1,client_state_push_v1" + .into(), + ), + ); + let capabilities = parse_omega_capability_response(&response).unwrap(); + assert!(capabilities.state_subscribe_v1); + assert!(capabilities.client_state_push_v1); + } + + #[test] + fn missing_capability_response_is_legacy() { + let response = CommunicationValue::new(CommunicationType::IdentificationResponse); + let capabilities = parse_omega_capability_response(&response).unwrap(); + assert_eq!(capabilities, PeerCapabilities::default()); + } + + #[test] + fn unsupported_capability_response_is_rejected() { + let response = CommunicationValue::new(CommunicationType::IdentificationResponse) + .add_typed_default( + DataType::Description, + DataValue::Str("omega;caps=unsupported_v9".into()), + ); + assert!(parse_omega_capability_response(&response).is_err()); } } diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 64cb052..c5c7dee 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -24,7 +24,6 @@ pub struct ClientConnection { pub receiver: Arc, pub_key: Arc>>>, pub rho_connection: Arc>>>, - pub interested_users: Arc>>, is_open: Arc>, message_slots: Arc, } @@ -35,7 +34,6 @@ impl ClientConnection { state: general.state.clone(), pub_key: Arc::new(RwLock::new(None)), rho_connection: general.rho_connection.clone(), - interested_users: Arc::new(RwLock::new(Vec::new())), is_open: Arc::new(RwLock::new(true)), sender: general.sender.clone(), receiver: general.receiver.clone(), @@ -104,9 +102,26 @@ impl ClientConnection { let mut cv = cv; - // Handle client status changes - if cv.is_type(CommunicationType::ClientChanged) { - self.handle_client_changed(cv).await; + if cv.is_type(CommunicationType::SetUserState) { + self.handle_set_user_state(cv).await; + return; + } + + // Compatibility for clients predating SetUserState. The target + // user fields, if present, are deliberately ignored: an + // authenticated connection may only change its own state. + if cv.is_type(CommunicationType::ClientChanged) + && cv.get_data_opt(DataType::UserState).is_some() + { + self.handle_set_user_state( + CommunicationValue::new(CommunicationType::SetUserState) + .with_id(cv.get_id()) + .add_typed_default( + DataType::UserState, + cv.get_data(DataType::UserState).clone(), + ), + ) + .await; return; } @@ -191,6 +206,46 @@ impl ClientConnection { || cv.is_type(CommunicationType::GetIotaData) || cv.is_type(CommunicationType::DeleteUser) { + if cv.is_type(CommunicationType::ChangeUserData) + && cv.get_data_opt(DataType::OnlineStatus).is_some() + { + let mut profile_request = cv.clone(); + let preference = profile_request.remove_data(DataType::OnlineStatus); + let state_request = CommunicationValue::new(CommunicationType::SetUserState) + .with_id(cv.get_id()) + .add_typed_default(DataType::UserState, preference.unwrap()); + let state_response = match self.request_set_user_state(state_request).await { + Ok(response) => response, + Err(error_type) => { + self.send_error_response(cv.get_id(), error_type).await; + return; + } + }; + if !state_response.is_type(CommunicationType::Success) { + self.send_message(&state_response).await; + return; + } + + if profile_request.data_len() == 0 { + self.send_message(&state_response).await; + return; + } + + match self + .await_omega_response(profile_request.with_sender(self.user_id)) + .await + { + Ok(response) if response.is_type(CommunicationType::Success) => { + self.send_message(&response).await; + } + Ok(response) => self.send_message(&response).await, + Err(_) => { + self.send_error_response(cv.get_id(), CommunicationType::ErrorInternal) + .await; + } + } + return; + } let sender = self.get_user_id().await; self.handle_omega_forward(cv.with_sender(sender as u64)) .await; @@ -290,34 +345,91 @@ impl ClientConnection { }); } async fn handle_omega_forward(self: Arc, cv: CommunicationValue) { - let client_for_closure = self.clone(); - let response_cv = self - .state - .omega - .clone() - .await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20))) - .await; - if let Ok(response_cv) = response_cv { - client_for_closure.send_message(&response_cv).await; + let request_id = cv.get_id(); + match self.await_omega_response(cv).await { + Ok(response_cv) => self.send_message(&response_cv).await, + Err(_) => { + self.send_error_response(request_id, CommunicationType::ErrorInternal) + .await; + } } } - /// Handle client status change - async fn handle_client_changed(self: Arc, cv: CommunicationValue) { - let user_id = self.get_user_id().await; - if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) { - let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online); - if let Some(rho_conn) = self.get_rho_connection().await { - self.state - .omega - .client_changed( - rho_conn.get_iota_id().await as i64, - user_id as i64, - self.session_id as i64, - user_status, - ) - .await; - } + async fn await_omega_response(&self, cv: CommunicationValue) -> Result { + self.state + .omega + .clone() + .await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20))) + .await + .map_err(|_| ()) + } + + async fn request_set_user_state( + &self, + cv: CommunicationValue, + ) -> Result { + if !self.state.omega.is_ready().await { + return Err(CommunicationType::ErrorInternal); + } + let Some(state) = cv + .get_data(DataType::UserState) + .as_str() + .and_then(UserStatus::from_client_preference) + else { + return Err(CommunicationType::ErrorInvalidData); + }; + let Some(rho) = self.get_rho_connection().await else { + return Err(CommunicationType::ErrorNoIota); + }; + let request = if self.state.omega.supports_set_user_state().await { + CommunicationValue::new(CommunicationType::SetUserState) + .with_id(cv.get_id()) + .with_sender(self.user_id) + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber(self.user_id as i128), + ) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(self.session_id as i128), + ) + .add_typed_default( + DataType::IotaId, + DataValue::SignedNumber(rho.get_iota_id().await as i128), + ) + .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())) + } else { + // Legacy Omega accepts the original ClientChanged request. It is + // intentionally limited to the authenticated user's ID and a + // writable preference, with no derived connectivity state. + CommunicationValue::new(CommunicationType::ClientChanged) + .with_id(cv.get_id()) + .with_sender(self.user_id) + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber(self.user_id as i128), + ) + .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())) + }; + if !self.state.omega.supports_set_user_state().await { + self.state + .omega + .send_message_result(&request) + .await + .map_err(|_| CommunicationType::ErrorInternal)?; + return Ok(CommunicationValue::new(CommunicationType::Success) + .with_id(cv.get_id()) + .add_typed_default(DataType::UserState, DataValue::Str(state.to_string()))); + } + self.await_omega_response(request) + .await + .map_err(|_| CommunicationType::ErrorInternal) + } + + async fn handle_set_user_state(self: Arc, cv: CommunicationValue) { + match self.request_set_user_state(cv.clone()).await { + Ok(response) => self.send_message(&response).await, + Err(error_type) => self.send_error_response(cv.get_id(), error_type).await, } } @@ -830,35 +942,6 @@ impl ClientConnection { let _ = self.sender.close(); } - /// Set interested users list - pub async fn set_interested_users(self: Arc, interested_ids: Vec) { - let mut interested_guard = self.interested_users.write().await; - *interested_guard = interested_ids; - } - #[allow(dead_code)] - pub async fn get_interested_users(self: Arc) -> Vec { - let interested_guard = self.interested_users.read().await; - interested_guard.clone() - } - - /// Check if interested in a user and send notification - #[allow(dead_code)] - pub async fn are_you_interested(self: Arc, user_id: i64, user_status: &str) { - let interested_guard = self.clone().get_interested_users().await; - if interested_guard.contains(&user_id) { - let status = if user_status == "user_invisible" { - "user_offline" - } else { - user_status - }; - let notification = CommunicationValue::new(CommunicationType::ClientChanged) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) - .add_typed_default(DataType::UserState, DataValue::Str(status.to_string())); - - self.send_message(¬ification).await; - } - } - /// Handle connection close pub async fn handle_close(self: Arc) { let user_id = self.get_user_id().await; @@ -880,7 +963,6 @@ impl Clone for ClientConnection { client_version: self.client_version.clone(), pub_key: Arc::clone(&self.pub_key), rho_connection: Arc::clone(&self.rho_connection), - interested_users: Arc::clone(&self.interested_users), is_open: Arc::clone(&self.is_open), message_slots: Arc::clone(&self.message_slots), } diff --git a/src/rho/connection.rs b/src/rho/connection.rs index 137929b..cede813 100755 --- a/src/rho/connection.rs +++ b/src/rho/connection.rs @@ -136,10 +136,23 @@ impl GeneralConnection { if let Some(rho_conn) = rho { rho_conn.bind_user_id(user_id).await; rho_conn.add_client_connection(client.clone()).await; - self.notify_user_connected(user_id, rho_conn.get_iota_id().await as i64) - .await; - self.send_initial_client_state_request(&rho_conn, user_id) - .await; + match self + .notify_user_connected(user_id, rho_conn.get_iota_id().await as i64) + .await + { + Ok(()) => { + self.send_initial_client_state_request(&rho_conn, user_id) + .await; + } + Err(error) => { + log_err!( + user_id, + PrintType::Client, + "UserConnected acknowledgement failed: {}", + error + ); + } + } } else { log_err!( user_id, @@ -155,29 +168,13 @@ impl GeneralConnection { true } - async fn notify_user_connected(&self, user_id: i64, iota_id: i64) { + async fn notify_user_connected(&self, user_id: i64, iota_id: i64) -> Result<(), String> { let session_id = *self.session_id.read().await as i64; - let notify = CommunicationValue::new(CommunicationType::UserConnected) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) - .add_typed_default( - DataType::SessionId, - DataValue::SignedNumber(session_id.into()), - ) - .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())) - .add_typed_default( - DataType::UserState, - DataValue::Str("user_online".to_string()), - ) - .add_typed_default( - DataType::UpdatedAt, - DataValue::SignedNumber( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i128, - ), - ); - self.state.omega.clone().send_message(¬ify).await; + self.state + .omega + .clone() + .user_connected(user_id, session_id, iota_id) + .await } async fn find_user_rho(&self, user_id: i64) -> Option> { @@ -227,9 +224,6 @@ impl GeneralConnection { async fn migrate_iota(self: &Arc) { let id = self.id; log_in!(id as i64, PrintType::Iota, "Iota {} connected", id); - let notify = CommunicationValue::new(CommunicationType::IotaConnected) - .add_typed_default(DataType::IotaId, DataValue::SignedNumber(id.into())); - self.state.omega.clone().send_message(¬ify).await; let iota = IotaConnection::from_general(self.clone(), id).await; let previous = self.state.rho.get_by_iota(id as i64).await; @@ -246,6 +240,14 @@ impl GeneralConnection { } } self.state.rho.add(rho).await; + if let Err(error) = self.state.omega.clone().iota_connected(id as i64).await { + log_err!( + id as i64, + PrintType::Iota, + "IotaConnected acknowledgement failed: {}", + error + ); + } self.load_iota_users(id).await; iota.start(); } diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 7f7086c..047f718 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -23,6 +23,47 @@ use tokio::sync::mpsc; use super::rho_connection::RhoConnection; +fn contact_snapshot(value: &CommunicationValue) -> Option<(i64, i64, Vec)> { + if !value.is_type(CommunicationType::ClientStateSync) + && !value.is_type(CommunicationType::AddConversation) + { + return None; + } + + let user_id = i64::try_from(value.get_receiver()) + .ok() + .filter(|id| *id > 0)?; + let session_id = value + .get_data(DataType::SessionId) + .as_signed_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| *id > 0)?; + let DataValue::Array(values) = value.get_data(DataType::UserIds) else { + return None; + }; + + let user_id_type = data_type_id(DataType::UserId, &TypeMap::latest()); + let mut contact_ids = Vec::with_capacity(values.len()); + for value in values { + let contact_id = match value { + DataValue::SignedNumber(id) => i64::try_from(*id).ok(), + DataValue::Container(entries) => entries.iter().find_map(|(key, value)| { + (*key == user_id_type) + .then(|| value.as_signed_number()) + .flatten() + .and_then(|id| i64::try_from(id).ok()) + }), + _ => None, + } + .filter(|id| *id > 0)?; + contact_ids.push(contact_id); + } + contact_ids.sort_unstable(); + contact_ids.dedup(); + + Some((user_id, session_id, contact_ids)) +} + #[allow(dead_code)] pub struct IotaConnection { pub state: Arc, @@ -172,6 +213,29 @@ impl IotaConnection { } } + async fn replace_state_subscription( + &self, + user_id: i64, + session_id: i64, + contact_ids: Vec, + ) -> Result<(), String> { + let Some(rho_connection) = self.get_rho_connection().await else { + return Err("Rho connection is unavailable".to_string()); + }; + if rho_connection + .get_client_connection(user_id, session_id) + .await + .is_none() + { + return Err("target client session is no longer connected".to_string()); + } + + self.state + .omega + .replace_state_subscription(user_id, session_id, contact_ids) + .await + } + /// Send a CommunicationValue to the Iota pub async fn send_message(&self, cv: &CommunicationValue) { log_cv_out!(PrintType::Iota, cv); @@ -206,6 +270,34 @@ impl IotaConnection { cv }; + // Iota's UserIds field is an authoritative replacement snapshot for + // the requesting client session, including contact mutations. + if let Some((user_id, session_id, contact_ids)) = contact_snapshot(&cv) { + if let Err(error) = self + .replace_state_subscription(user_id, session_id, contact_ids) + .await + { + log_err!( + self.iota_id as i64, + PrintType::Omega, + "Failed to replace presence subscription for user {} session {}: {}", + user_id, + session_id, + error + ); + } + } + + if cv.is_type(CommunicationType::StateSubscribe) { + self.send_error_response( + cv.get_id(), + CommunicationType::ErrorInvalidData, + Some("StateSubscribe must come from an authoritative contact snapshot"), + ) + .await; + return; + } + // Handle GET_CHATS if cv.is_type(CommunicationType::GetChats) { self.handle_get_chats(cv).await; @@ -517,7 +609,20 @@ impl IotaConnection { } let mut interested_ids: Vec = Vec::new(); - let session_id = cv.get_data(DataType::SessionId).as_signed_number(); + let Some(session_id) = cv + .get_data(DataType::SessionId) + .as_signed_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| *id > 0) + else { + self.forward_to_client( + CommunicationValue::new(CommunicationType::ErrorInvalidData) + .with_id(cv.get_id()) + .with_receiver(user_id), + ) + .await; + return; + }; let tm = TypeMap::latest(); // Presence interest is the complete contact set, independent of @@ -658,20 +763,18 @@ impl IotaConnection { // ============================ // Notify Omega // ============================ - self.state - .omega - .user_states(user_id as i64, interested_ids.clone()) - .await; - - // ============================ - // Notify Rho - // ============================ - if let Some(rho_conn) = self.get_rho_connection().await { - if let Some(session_id) = session_id.and_then(|id| i64::try_from(id).ok()) { - rho_conn - .set_interested(user_id as i64, session_id, interested_ids) - .await; - } + if let Err(error) = self + .replace_state_subscription(user_id as i64, session_id, interested_ids.clone()) + .await + { + log_err!( + self.iota_id as i64, + PrintType::Omega, + "Failed to replace presence subscription for user {} session {}: {}", + user_id, + session_id, + error + ); } // ============================ @@ -779,3 +882,48 @@ impl std::fmt::Debug for IotaConnection { .finish() } } + +#[cfg(test)] +mod tests { + use super::contact_snapshot; + use crate::util::data_type_id; + use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue, TypeMap}; + + fn snapshot(user_ids: DataValue) -> CommunicationValue { + CommunicationValue::new(CommunicationType::ClientStateSync) + .with_receiver(7) + .add_typed_default(DataType::SessionId, DataValue::SignedNumber(11)) + .add_typed_default(DataType::UserIds, user_ids) + } + + #[test] + fn contact_snapshot_deduplicates_flat_and_typed_ids() { + let user_id_type = data_type_id(DataType::UserId, &TypeMap::latest()); + let parsed = contact_snapshot(&snapshot(DataValue::Array(vec![ + DataValue::SignedNumber(20), + DataValue::Container(vec![(user_id_type, DataValue::SignedNumber(21))]), + DataValue::SignedNumber(20), + ]))) + .unwrap(); + + assert_eq!(parsed, (7, 11, vec![20, 21])); + } + + #[test] + fn contact_snapshot_accepts_empty_contact_sets() { + assert_eq!( + contact_snapshot(&snapshot(DataValue::Array(Vec::new()))), + Some((7, 11, Vec::new())) + ); + } + + #[test] + fn contact_snapshot_rejects_malformed_contact_sets() { + assert_eq!( + contact_snapshot(&snapshot(DataValue::Array(vec![DataValue::Str( + "bad".into(), + )]))), + None + ); + } +} diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 46f0330..4c60560 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -1,6 +1,6 @@ use super::{client_connection::ClientConnection, iota_connection::IotaConnection}; -use crate::{data::user::UserStatus, rho::app_connection::AppConnection}; +use crate::{log_err, rho::app_connection::AppConnection}; use dashmap::DashMap; use mtp::codec::{CommunicationValue, DataType}; use std::sync::Arc; @@ -178,11 +178,21 @@ impl RhoConnection { return; } self.client_connections.remove(&key); - self.iota_connection + if let Err(error) = self + .iota_connection .state .omega .client_disconnected(target_user_id, target_session_id) - .await; + .await + { + log_err!( + target_user_id, + crate::util::logger::PrintType::Client, + "UserDisconnected acknowledgement failed for session {}: {}", + target_session_id, + error + ); + } let remaining_for_user = self .client_connections .iter() @@ -194,16 +204,6 @@ impl RhoConnection { .rho .remove_user_binding(target_user_id, self.get_iota_id().await as i64) .await; - self.iota_connection - .state - .omega - .client_changed( - self.get_iota_id().await as i64, - target_user_id, - target_session_id, - UserStatus::user_offline, - ) - .await; } } @@ -228,7 +228,14 @@ impl RhoConnection { } // Notify OmegaConnection - self.iota_connection.state.omega.close_iota(iota_id).await; + if let Err(error) = self.iota_connection.state.omega.close_iota(iota_id).await { + log_err!( + iota_id, + crate::util::logger::PrintType::Iota, + "IotaDisconnected acknowledgement failed: {}", + error + ); + } } /// Send message from Iota to specific client @@ -255,25 +262,6 @@ impl RhoConnection { self.iota_connection.send_message(&cv).await; } - /// Set interested users for a specific client - pub async fn set_interested(&self, user_id: i64, session_id: i64, interested_ids: Vec) { - if let Some(connection) = self.get_client_connection(user_id, session_id).await { - connection.set_interested_users(interested_ids).await; - } - } - - /// Check if clients are interested in a user - #[allow(dead_code)] - pub async fn are_they_interested(&self, user_id: i64, user_status: &str) { - let connections = self.get_client_connections().await; - for connection in connections.iter() { - connection - .clone() - .are_you_interested(user_id, user_status) - .await; - } - } - /// Check if this RhoConnection contains a specific user ID #[allow(dead_code)] pub async fn contains_user(&self, user_id: &i64) -> bool { diff --git a/src/rho/rho_manager.rs b/src/rho/rho_manager.rs index 0345ae8..b86e205 100644 --- a/src/rho/rho_manager.rs +++ b/src/rho/rho_manager.rs @@ -135,4 +135,17 @@ impl RhoManager { .map(|entry| entry.value().clone()) .collect() } + + pub async fn get_client_connection( + &self, + user_id: i64, + session_id: i64, + ) -> Option> { + for rho in self.connections().await { + if let Some(client) = rho.get_client_connection(user_id, session_id).await { + return Some(client); + } + } + None + } }