From 737adf0890bbed16490bf671d6daac02d7bf7e2e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= <55464917+jlucaso1@users.noreply.github.com> Date: Sun, 26 Jul 2026 18:52:24 -0300 Subject: [PATCH 01/63] feat(voip): add group calls and call links --- src/client/voip.rs | 953 +++++++++++++ src/handlers/call.rs | 440 +++++- src/request.rs | 23 +- src/voip/facade.rs | 1252 +++++++++++++++- src/voip/mod.rs | 12 +- src/voip/transport.rs | 3 +- wacore/src/stanza/call.rs | 69 +- wacore/src/stanza/group_call.rs | 1658 ++++++++++++++++++++++ wacore/src/stanza/mod.rs | 1 + wacore/src/types/call.rs | 57 +- wacore/src/types/group_call.rs | 305 ++++ wacore/src/types/mod.rs | 1 + wacore/src/voip/app_data.rs | 215 +++ wacore/src/voip/audio.rs | 7 + wacore/src/voip/demux.rs | 107 ++ wacore/src/voip/driver.rs | 108 ++ wacore/src/voip/e2e_srtp.rs | 15 + wacore/src/voip/engine.rs | 1326 ++++++++++++++++- wacore/src/voip/group.rs | 305 ++++ wacore/src/voip/group_audio.rs | 220 +++ wacore/src/voip/group_media.rs | 837 +++++++++++ wacore/src/voip/h264.rs | 11 + wacore/src/voip/mod.rs | 27 +- wacore/src/voip/registry.rs | 504 ++++++- wacore/src/voip/rtp.rs | 2 + wacore/src/voip/session.rs | 151 +- wacore/src/voip/ssrc.rs | 34 + wacore/src/voip/stun.rs | 246 +++- wacore/tests/group_call_builders_test.rs | 88 ++ 29 files changed, 8878 insertions(+), 99 deletions(-) create mode 100644 wacore/src/stanza/group_call.rs create mode 100644 wacore/src/types/group_call.rs create mode 100644 wacore/src/voip/app_data.rs create mode 100644 wacore/src/voip/group.rs create mode 100644 wacore/src/voip/group_audio.rs create mode 100644 wacore/src/voip/group_media.rs create mode 100644 wacore/tests/group_call_builders_test.rs diff --git a/src/client/voip.rs b/src/client/voip.rs index a707e7275..e8c5f6bb9 100644 --- a/src/client/voip.rs +++ b/src/client/voip.rs @@ -3,11 +3,36 @@ #[cfg(feature = "voip-runtime")] use std::sync::Arc; +#[cfg(feature = "voip-runtime")] +use std::time::Duration; use wacore::stanza::call::{TerminateParams, build_reject, build_terminate}; +#[cfg(feature = "voip-runtime")] +use wacore::stanza::group_call::{ + build_active_group_accept, build_active_group_preaccept, build_call_link_create, + build_call_link_join_with_capability, build_call_link_query, build_raise_hand, + build_screen_share, build_waiting_room_admit, build_waiting_room_deny, + build_waiting_room_heartbeat, build_waiting_room_toggle, parse_call_link_create_ack, + parse_call_link_join_ack, parse_call_link_query_ack, parse_waiting_room_admit_ack, + parse_waiting_room_deny_ack, parse_waiting_room_toggle_ack, +}; +#[cfg(feature = "voip-runtime")] +use wacore::types::call::CallAction; use wacore::types::call::IncomingCall; +#[cfg(feature = "voip-runtime")] +use wacore::types::group_call::{ + CallLink, CallLinkJoin, CallLinkMedia, CallLinkPreview, ScreenShare, ScreenShareState, +}; +#[cfg(feature = "voip-runtime")] +use wacore::voip::{AudioFormat, CallEvent, CallPhase, CallSession, VideoControl}; use wacore_binary::Jid; +#[cfg(feature = "voip-runtime")] +use wacore_binary::Node; +#[cfg(feature = "voip-runtime")] +use wacore_binary::Server; +#[cfg(feature = "voip-runtime")] +use super::ResponseWaiter; use super::{Client, ClientError}; /// Opaque call-control handle obtained via [`Client::voip`]. Borrows the client; @@ -106,6 +131,14 @@ pub enum CallError { #[cfg(feature = "voip-runtime")] #[error("offer pkmsg requires (account is None)")] MissingDeviceIdentity, + /// A call-service response was malformed or rejected. + #[cfg(feature = "voip-runtime")] + #[error("call service response failed: {0}")] + Response(String), + /// The call service did not answer within its bounded request window. + #[cfg(feature = "voip-runtime")] + #[error("call service request timed out")] + ResponseTimeout, } impl Voip<'_> { @@ -165,6 +198,439 @@ impl Voip<'_> { crate::voip::facade::OutgoingCall::new(self.client, peer) } + /// Begin a native group call to two or more selected users. + #[cfg(feature = "voip-runtime")] + pub fn group_call<'b>(&'b self, targets: &'b [Jid]) -> crate::voip::OutgoingGroupCall<'b> { + crate::voip::facade::OutgoingGroupCall::new(self.client, targets) + } + + /// Begin a native call bound to an existing group. The current roster is resolved at + /// [`start`](crate::voip::GroupBoundCall::start), with this account excluded automatically. + #[cfg(feature = "voip-runtime")] + pub fn group_call_by_id<'b>(&'b self, group_jid: &'b Jid) -> crate::voip::GroupBoundCall<'b> { + crate::voip::facade::GroupBoundCall::new(self.client, group_jid) + } + + /// Join a reusable call link and attach group media after admission. + #[cfg(feature = "voip-runtime")] + pub fn call_link<'b>( + &'b self, + token_or_url: &'b str, + media: CallLinkMedia, + ) -> crate::voip::CallLinkCall<'b> { + crate::voip::facade::CallLinkCall::new(self.client, token_or_url, media) + } + + /// Send the eager preparation response for an active group-call invitation. + #[cfg(feature = "voip-runtime")] + pub async fn preaccept_group_invite(&self, incoming: &IncomingCall) -> Result<(), CallError> { + let CallAction::Offer { + call_id, + call_creator, + is_video, + .. + } = &incoming.action + else { + return Err(CallError::NotAnOffer); + }; + if incoming.group.is_none() { + return Err(CallError::Media("offer is not an active group invitation")); + } + let node = build_active_group_preaccept( + call_id, + call_creator, + &self.client.generate_request_id(), + *is_video, + ) + .map_err(|error| CallError::Response(error.to_string()))?; + self.client.send_node(node).await?; + Ok(()) + } + + /// Immediately accept an active group-call invitation using call-scoped signaling. + #[cfg(feature = "voip-runtime")] + pub async fn accept_group_invite(&self, incoming: &IncomingCall) -> Result<(), CallError> { + let CallAction::Offer { + call_id, + call_creator, + .. + } = &incoming.action + else { + return Err(CallError::NotAnOffer); + }; + if incoming.group.is_none() { + return Err(CallError::Media("offer is not an active group invitation")); + } + let node = + build_active_group_accept(call_id, call_creator, &self.client.generate_request_id()) + .map_err(|error| CallError::Response(error.to_string()))?; + self.client.send_node(node).await?; + self.client.call_registry().take_ringing(call_id); + self.client + .call_registry() + .transition(call_id, CallPhase::Connecting); + Ok(()) + } + + /// Create a reusable audio or video call link. + #[cfg(feature = "voip-runtime")] + pub async fn create_call_link(&self, media: CallLinkMedia) -> Result { + let request_id = self.client.generate_request_id(); + let request = build_call_link_create(media, &request_id) + .map_err(|error| CallError::Response(error.to_string()))?; + execute_call_service_request( + self.client, + &request_id, + request, + parse_call_link_create_ack, + ) + .await + } + + /// Inspect a call link without joining it. + #[cfg(feature = "voip-runtime")] + pub async fn preview_call_link( + &self, + token_or_url: &str, + media: CallLinkMedia, + ) -> Result { + let token = normalize_call_link_token(token_or_url, media)?; + let request_id = self.client.generate_request_id(); + let request = build_call_link_query(&token, media, &request_id) + .map_err(|error| CallError::Response(error.to_string()))?; + execute_call_service_request(self.client, &request_id, request, parse_call_link_query_ack) + .await + } + + /// Join a call link. The result explicitly reports whether this endpoint was admitted or placed + /// in the waiting room; media starts only after an admitted authoritative group snapshot. + #[cfg(feature = "voip-runtime")] + pub async fn join_call_link( + &self, + token_or_url: &str, + media: CallLinkMedia, + ) -> Result { + self.join_call_link_with_audio(token_or_url, media, AudioFormat::MLOW_16KHZ_60MS) + .await + } + + #[cfg(feature = "voip-runtime")] + pub(crate) async fn join_call_link_with_audio( + &self, + token_or_url: &str, + media: CallLinkMedia, + audio_format: AudioFormat, + ) -> Result { + let own_lid = self.client.lid().ok_or(CallError::Media("no own LID"))?; + let token = normalize_call_link_token(token_or_url, media)?; + let request_id = self.client.generate_request_id(); + let capability = crate::voip::facade::offer_capability(false, audio_format); + let request = build_call_link_join_with_capability(&token, media, &request_id, capability) + .map_err(|error| CallError::Response(error.to_string()))?; + let mut join = execute_call_service_request( + self.client, + &request_id, + request, + parse_call_link_join_ack, + ) + .await?; + if join.token.is_empty() { + join.token.clone_from(&token); + } + if join.media != media { + return Err(CallError::Response( + "call-link response changed the requested media mode".to_string(), + )); + } + + let mut session = CallSession::new_outgoing( + &join.call_id, + Jid::new(&join.call_id, Server::Call), + join.call_creator.clone(), + ); + session.audio_format = Some(audio_format); + session.is_video = media == CallLinkMedia::Video; + session.group = join.group.clone(); + let _ = session.transition_to(CallPhase::Calling); + let _ = session.transition_to(if join.in_waiting_room { + CallPhase::WaitingRoom + } else { + CallPhase::Connecting + }); + let registry = self.client.call_registry(); + let generation = registry.insert(session); + + if join.in_waiting_room { + let Some(room) = join.pending_waiting_room() else { + registry.remove_if_current(&join.call_id, generation); + return Err(CallError::Response( + "call-link join omitted its waiting-room state".to_string(), + )); + }; + if registry.apply_waiting_room(room) != wacore::voip::GroupStateApply::Applied { + registry.remove_if_current(&join.call_id, generation); + return Err(CallError::Response( + "call-link waiting-room identity was rejected".to_string(), + )); + } + if let Err(error) = self + .waiting_room_heartbeat(&join.call_id, &join.call_creator) + .await + { + registry.remove_if_current(&join.call_id, generation); + return Err(error); + } + self.start_waiting_room_heartbeat( + join.call_id.clone(), + join.call_creator.clone(), + generation, + ); + } else if let Some(update) = join.group.as_ref() + && update.rekey_requested + { + let raw_epoch = match crate::voip::facade::fanout_group_epoch(self.client, update).await + { + Ok(raw_epoch) => raw_epoch, + Err(error) => { + registry.remove_if_current(&join.call_id, generation); + return Err(error); + } + }; + if !registry.send_group_epoch(&join.call_id, update.transaction_id, raw_epoch) { + registry.remove_if_current(&join.call_id, generation); + return Err(CallError::Media( + "call-link group epoch could not be retained", + )); + } + } + + registry.set_group_invite_self_device( + &join.call_id, + generation, + wacore::types::group_call::GroupCallDevice::new(own_lid).with_capability(1, capability), + ); + Ok(join) + } + + /// Enable or disable approval for a live call-link waiting room. + #[cfg(feature = "voip-runtime")] + pub async fn set_approval_required( + &self, + call_id: &str, + call_creator: &Jid, + enabled: bool, + ) -> Result<(), CallError> { + self.ensure_waiting_room_admin(call_id)?; + let request_id = self.client.generate_request_id(); + execute_call_service_request( + self.client, + &request_id, + build_waiting_room_toggle(call_id, call_creator, enabled, &request_id), + parse_waiting_room_toggle_ack, + ) + .await?; + self.client + .call_registry() + .set_waiting_room_enabled(call_id, enabled); + Ok(()) + } + + /// Keep a pending call-link admission alive. + #[cfg(feature = "voip-runtime")] + pub async fn waiting_room_heartbeat( + &self, + call_id: &str, + call_creator: &Jid, + ) -> Result<(), CallError> { + self.send_group_control( + call_id, + build_waiting_room_heartbeat(call_id, call_creator, &self.client.generate_request_id()), + ) + .await + } + + /// Admit one user from a call-link waiting room. + #[cfg(feature = "voip-runtime")] + pub async fn admit_waiting_user( + &self, + call_id: &str, + call_creator: &Jid, + user: &Jid, + ) -> Result<(), CallError> { + self.ensure_waiting_room_admin(call_id)?; + let request_id = self.client.generate_request_id(); + execute_call_service_request( + self.client, + &request_id, + build_waiting_room_admit(call_id, call_creator, user, &request_id), + parse_waiting_room_admit_ack, + ) + .await + } + + /// Deny one user from a call-link waiting room. + #[cfg(feature = "voip-runtime")] + pub async fn deny_waiting_user( + &self, + call_id: &str, + call_creator: &Jid, + user: &Jid, + ) -> Result<(), CallError> { + self.ensure_waiting_room_admin(call_id)?; + let request_id = self.client.generate_request_id(); + execute_call_service_request( + self.client, + &request_id, + build_waiting_room_deny(call_id, call_creator, user, &request_id), + parse_waiting_room_deny_ack, + ) + .await + } + + /// Publish the local persistent raise/lower-hand state. + #[cfg(feature = "voip-runtime")] + pub async fn set_hand_raised( + &self, + call_id: &str, + call_creator: &Jid, + raised: bool, + ) -> Result<(), CallError> { + let participant = self + .client + .lid() + .ok_or(CallError::Media("no own LID"))? + .to_non_ad(); + let target = Jid::new(call_id, Server::Call); + self.send_group_control( + call_id, + build_raise_hand( + call_id, + &target, + call_creator, + &self.client.generate_request_id(), + raised, + ), + ) + .await?; + let registry = self.client.call_registry(); + if registry.set_raised_hand(call_id, &participant, raised) { + registry.send_call_event( + call_id, + CallEvent::HandRaised { + participant, + raised, + }, + ); + } + Ok(()) + } + + /// Publish a screen-share start/stop transition. + #[cfg(feature = "voip-runtime")] + pub async fn set_screen_share( + &self, + call_id: &str, + call_creator: &Jid, + state: ScreenShareState, + screen_share_id: Option, + ) -> Result<(), CallError> { + let participant = self + .client + .lid() + .ok_or(CallError::Media("no own LID"))? + .to_non_ad(); + let target = Jid::new(call_id, Server::Call); + self.send_group_control( + call_id, + build_screen_share( + call_id, + &target, + call_creator, + &self.client.generate_request_id(), + state, + screen_share_id, + ), + ) + .await?; + let screen_share = ScreenShare::new(state, screen_share_id); + let registry = self.client.call_registry(); + if registry.set_screen_share(call_id, &participant, screen_share.clone()) { + registry.send_call_event( + call_id, + CallEvent::ScreenShareChanged { + participant, + screen_share, + }, + ); + } + if state == ScreenShareState::Started + && let Some(generation) = registry.generation_of(call_id) + { + registry.send_video_ctl(call_id, generation, VideoControl::RequireKeyframe); + } + Ok(()) + } + + #[cfg(feature = "voip-runtime")] + async fn send_group_control(&self, call_id: &str, node: Node) -> Result<(), CallError> { + if call_id.is_empty() { + return Err(CallError::EmptyCallId); + } + self.client.send_node(node).await?; + Ok(()) + } + + #[cfg(feature = "voip-runtime")] + fn ensure_waiting_room_admin(&self, call_id: &str) -> Result<(), CallError> { + let room = self + .client + .call_registry() + .group_state(call_id) + .and_then(|state| state.waiting_room().cloned()) + .ok_or(CallError::Media("call has no waiting-room state"))?; + if !room.is_admin { + return Err(CallError::Media( + "waiting-room control requires an administrator", + )); + } + Ok(()) + } + + #[cfg(feature = "voip-runtime")] + fn start_waiting_room_heartbeat(&self, call_id: String, call_creator: Jid, generation: u64) { + let weak_client = self.client.self_weak.get().cloned().unwrap_or_default(); + let runtime = self.client.runtime.clone(); + let sleeper = runtime.clone(); + let heartbeat_call_id = call_id.clone(); + let task = runtime.spawn(Box::pin(async move { + loop { + sleeper.sleep(Duration::from_secs(10)).await; + let Some(client) = weak_client.upgrade() else { + break; + }; + if client.call_registry().phase(&heartbeat_call_id) != Some(CallPhase::WaitingRoom) + { + break; + } + let request_id = client.generate_request_id(); + if client + .send_node(build_waiting_room_heartbeat( + &heartbeat_call_id, + &call_creator, + &request_id, + )) + .await + .is_err() + { + break; + } + } + })); + self.client + .call_registry() + .set_waiting_room_task(&call_id, generation, task); + } + /// Terminate an active call. pub async fn terminate( &self, @@ -195,15 +661,86 @@ impl Voip<'_> { } } +#[cfg(feature = "voip-runtime")] +fn normalize_call_link_token( + token_or_url: &str, + expected_media: CallLinkMedia, +) -> Result { + let value = token_or_url.trim(); + if value.is_empty() { + return Err(CallError::Response( + "call-link token is required".to_string(), + )); + } + const PREFIX: &str = "https://call.whatsapp.com/"; + if let Some(path) = value.strip_prefix(PREFIX) { + let mut parts = path.split('/'); + let media = parts.next(); + let token = parts.next(); + if parts.next().is_some() + || token.is_none_or(str::is_empty) + || media != Some(expected_media.as_str()) + { + return Err(CallError::Response( + "invalid call-link URL or media mode".to_string(), + )); + } + return Ok(token.unwrap_or_default().to_string()); + } + if value.contains("://") || value.contains('/') { + return Err(CallError::Response("invalid call-link token".to_string())); + } + Ok(value.to_string()) +} + +#[cfg(feature = "voip-runtime")] +async fn execute_call_service_request( + client: &Client, + request_id: &str, + request: Node, + parse: fn(&wacore_binary::NodeRef<'_>) -> anyhow::Result, +) -> Result { + let (tx, response) = futures::channel::oneshot::channel(); + let cleanup_generation = client + .response_waiters_guard() + .try_insert_guarded(request_id.to_string(), ResponseWaiter::Iq(tx)) + .ok_or_else(|| CallError::Response("duplicate call-service request id".to_string()))?; + let _waiter_guard = crate::request::ResponseWaiterGuard::new( + client.response_waiters.clone(), + request_id.to_string(), + cleanup_generation, + ); + if let Err(error) = client.send_node(request).await { + return Err(error.into()); + } + let response = + match wacore::runtime::timeout(&*client.runtime, Duration::from_secs(10), response).await { + Ok(Ok(response)) => response, + Ok(Err(_)) => return Err(CallError::Response("response channel closed".to_string())), + Err(_) => return Err(CallError::ResponseTimeout), + }; + parse(response.get()).map_err(|error| CallError::Response(error.to_string())) +} + #[cfg(test)] mod tests { use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; + #[cfg(feature = "voip-runtime")] + use std::time::Duration; use async_trait::async_trait; use bytes::Bytes; use wacore::handshake::NoiseCipher; use wacore::types::call::{CallAction, IncomingCall}; + #[cfg(feature = "voip-runtime")] + use wacore::types::group_call::{CallLinkMedia, ScreenShareState}; + #[cfg(feature = "voip-runtime")] + use wacore::voip::{ + AudioFormat, CallEvent, CallPhase, CallSession, VideoControl, video_control_channel, + }; + #[cfg(feature = "voip-runtime")] + use wacore_binary::builder::NodeBuilder; use wacore_binary::{Jid, Server}; use crate::client::Client; @@ -414,4 +951,420 @@ mod tests { }; assert!(client.voip().reject(&call).await.is_err()); } + + #[cfg(feature = "voip-runtime")] + #[tokio::test] + async fn local_group_controls_commit_state_events_and_screen_keyframe_gate() { + let (client, transport) = crate::test_utils::create_iq_test_client().await; + let own_device = Jid::new("111111111111111", Server::Lid).with_device(1); + client + .persistence_manager() + .process_command(crate::store::commands::DeviceCommand::SetLid(Some( + own_device, + ))) + .await; + let participant = Jid::new("111111111111111", Server::Lid); + let creator = participant.clone(); + let call_id = "TEST-GROUP-CONTROLS"; + let registry = client.call_registry(); + let generation = registry.insert(CallSession::new_outgoing( + call_id, + Jid::new(call_id, Server::Call), + creator.clone(), + )); + let (event_tx, event_rx) = async_channel::bounded(4); + let (video_tx, video_rx) = video_control_channel(); + registry.set_video_channels(call_id, generation, event_tx, video_tx, Box::new(|| {})); + + client + .voip() + .set_hand_raised(call_id, &creator, true) + .await + .expect("raise hand"); + assert!( + registry + .group_state(call_id) + .expect("group state") + .raised_hands() + .contains(&participant) + ); + assert!(matches!( + event_rx.try_recv(), + Ok(CallEvent::HandRaised { + participant: event_participant, + raised: true, + }) if event_participant == participant + )); + + client + .voip() + .set_screen_share(call_id, &creator, ScreenShareState::Started, Some(7)) + .await + .expect("start screen share"); + let share = registry + .group_state(call_id) + .expect("group state") + .screen_shares() + .get(&participant) + .cloned() + .expect("local screen share"); + assert_eq!(share.state, ScreenShareState::Started); + assert_eq!(share.version, 2); + assert_eq!(share.screen_share_id, Some(7)); + assert!(matches!( + event_rx.try_recv(), + Ok(CallEvent::ScreenShareChanged { + participant: event_participant, + screen_share, + }) if event_participant == participant && screen_share == share + )); + assert_eq!( + video_rx.try_recv(), + Ok(VideoControl::RequireKeyframe), + "starting a replacement screen source must re-arm the H.264 recovery gate" + ); + + client + .voip() + .set_screen_share(call_id, &creator, ScreenShareState::Stopped, None) + .await + .expect("stop screen share"); + assert!( + registry + .group_state(call_id) + .expect("group state") + .screen_shares() + .is_empty() + ); + assert!(matches!( + event_rx.try_recv(), + Ok(CallEvent::ScreenShareChanged { + participant: event_participant, + screen_share, + }) if event_participant == participant + && screen_share.state == ScreenShareState::Stopped + )); + assert_eq!(video_rx.try_recv(), Err(async_channel::TryRecvError::Empty)); + assert_eq!(transport.sent_count(), 3); + registry.remove_if_current(call_id, generation); + } + + #[cfg(feature = "voip-runtime")] + #[tokio::test(start_paused = true)] + async fn call_link_requests_round_trip_through_bounded_response_waiters() { + async fn wait_for_frames( + transport: &crate::transport::mock::CapturingMockTransport, + expected: usize, + ) { + for _ in 0..10_000 { + if transport.sent_count() >= expected { + return; + } + tokio::task::yield_now().await; + } + panic!("timed out waiting for {expected} captured call frames"); + } + + let (client, transport) = crate::test_utils::create_iq_test_client().await; + client + .persistence_manager() + .process_command(crate::store::commands::DeviceCommand::SetLid(Some( + Jid::new("111111111111111", Server::Lid).with_device(1), + ))) + .await; + let creator = Jid::new("333333333333333", Server::Lid); + + let sent = client.wait_for_sent_node(crate::client::NodeFilter::tag("call")); + let create_client = client.clone(); + let create = tokio::spawn(async move { + create_client + .voip() + .create_call_link(CallLinkMedia::Video) + .await + }); + let request = sent.await.expect("link_create request"); + let request_id = request + .as_node_ref() + .attrs() + .optional_string("id") + .expect("request id") + .into_owned(); + crate::test_utils::answer_iq( + &client, + &request_id, + &NodeBuilder::new("ack") + .attr("class", "call") + .attr("type", "link_create") + .attr("id", request_id.as_str()) + .children([NodeBuilder::new("link_create") + .attr("token", "TEST-CALL-LINK") + .attr("media", "video") + .build()]) + .build(), + ) + .await; + let link = create.await.expect("create task").expect("create response"); + assert_eq!(link.token, "TEST-CALL-LINK"); + assert_eq!(link.media, CallLinkMedia::Video); + + let sent = client.wait_for_sent_node(crate::client::NodeFilter::tag("call")); + let preview_client = client.clone(); + let preview = tokio::spawn(async move { + preview_client + .voip() + .preview_call_link("TEST-CALL-LINK", CallLinkMedia::Video) + .await + }); + let request = sent.await.expect("link_query request"); + let request_id = request + .as_node_ref() + .attrs() + .optional_string("id") + .expect("request id") + .into_owned(); + crate::test_utils::answer_iq( + &client, + &request_id, + &NodeBuilder::new("ack") + .attr("class", "call") + .attr("type", "link_query") + .attr("id", request_id.as_str()) + .children([NodeBuilder::new("link_query") + .attr("token", "TEST-CALL-LINK") + .attr("media", "video") + .attr("link_creator", creator.clone()) + .children([NodeBuilder::new("waiting_room") + .attr("enabled", "1") + .attr("is_admin", "0") + .build()]) + .build()]) + .build(), + ) + .await; + let preview = preview + .await + .expect("preview task") + .expect("preview response"); + assert_eq!(preview.creator, creator); + assert!(preview.waiting_room_enabled); + assert!(!preview.is_admin); + + let sent = client.wait_for_sent_node(crate::client::NodeFilter::tag("call")); + let join_client = client.clone(); + let join = tokio::spawn(async move { + join_client + .voip() + .join_call_link_with_audio( + "TEST-CALL-LINK", + CallLinkMedia::Video, + AudioFormat::OPUS_16KHZ_60MS, + ) + .await + }); + let request = sent.await.expect("link_join request"); + let request_ref = request.as_node_ref(); + let action = &request_ref.children().expect("join action children")[0]; + assert_eq!( + action + .get_optional_child("capability") + .expect("join capability") + .content_bytes(), + Some(wacore::stanza::call::CAPABILITY_STANDARD_OPUS_OFFER.as_slice()) + ); + let request_id = request + .as_node_ref() + .attrs() + .optional_string("id") + .expect("request id") + .into_owned(); + crate::test_utils::answer_iq( + &client, + &request_id, + &NodeBuilder::new("ack") + .attr("class", "call") + .attr("type", "link_join") + .attr("id", request_id.as_str()) + .children([NodeBuilder::new("waiting_room") + .attr("call-id", "TEST-CALL-ID") + .attr("call-creator", creator.clone()) + .attr("link-token", "TEST-CALL-LINK") + .attr("media", "video") + .attr("enabled", "1") + .attr("is_admin", "0") + .attr("transaction-id", "7") + .children([NodeBuilder::new("user") + .attr("jid", Jid::new("444444444444444", Server::Lid)) + .attr("state", "pending") + .build()]) + .build()]) + .build(), + ) + .await; + let join = join.await.expect("join task").expect("join response"); + assert!(join.in_waiting_room); + assert!(join.waiting_room_enabled); + assert_eq!(join.call_id, "TEST-CALL-ID"); + assert!(join.group.is_none()); + assert_eq!( + client.call_registry().phase("TEST-CALL-ID"), + Some(CallPhase::WaitingRoom) + ); + let room = client + .call_registry() + .group_state("TEST-CALL-ID") + .and_then(|state| state.waiting_room().cloned()) + .expect("waiting-room state retained"); + assert_eq!(room.transaction_id, Some(7)); + assert_eq!(room.users.len(), 1); + + wait_for_frames(&transport, 4).await; + let immediate = crate::test_utils::decode_sent_iq(&transport, 3).await; + let heartbeat = &immediate.get().children().expect("heartbeat action")[0]; + assert_eq!(heartbeat.tag, "heartbeat"); + assert_eq!( + heartbeat.attrs().optional_string("type").as_deref(), + Some("waiting_room") + ); + + tokio::time::advance(Duration::from_secs(10)).await; + wait_for_frames(&transport, 5).await; + let scheduled = crate::test_utils::decode_sent_iq(&transport, 4).await; + assert_eq!( + scheduled.get().children().expect("heartbeat action")[0].tag, + "heartbeat" + ); + + let admitted = NodeBuilder::new("group_update") + .attr("call-id", "TEST-CALL-ID") + .attr("call-creator", creator) + .children([NodeBuilder::new("group_info") + .attr("transaction-id", "8") + .attr("connected-limit", "32") + .attr("media", "video") + .build()]) + .build(); + let update = wacore::stanza::group_call::parse_group_update(&admitted.as_node_ref()) + .expect("admitted group snapshot"); + assert_eq!( + client.call_registry().apply_group_update(update), + wacore::voip::GroupStateApply::Applied + ); + assert_eq!( + client.call_registry().phase("TEST-CALL-ID"), + Some(CallPhase::Connecting) + ); + let heartbeat_count = transport.sent_count(); + tokio::time::advance(Duration::from_secs(20)).await; + tokio::task::yield_now().await; + assert_eq!( + transport.sent_count(), + heartbeat_count, + "admission must cancel the repeating heartbeat" + ); + let generation = client + .call_registry() + .generation_of("TEST-CALL-ID") + .expect("registered call-link generation"); + client + .call_registry() + .remove_if_current("TEST-CALL-ID", generation); + } + + #[cfg(feature = "voip-runtime")] + #[tokio::test] + async fn cancelling_call_link_request_removes_response_waiter() { + let (client, _transport) = crate::test_utils::create_iq_test_client().await; + let sent = client.wait_for_sent_node(crate::client::NodeFilter::tag("call")); + let request_client = client.clone(); + let request = tokio::spawn(async move { + request_client + .voip() + .create_call_link(CallLinkMedia::Audio) + .await + }); + let node = sent.await.expect("link_create request"); + let request_id = node + .as_node_ref() + .attrs() + .optional_string("id") + .expect("request id") + .into_owned(); + assert!( + client.response_waiters_guard().contains_key(&request_id), + "the request must register its ACK waiter before sending" + ); + + request.abort(); + assert!( + request + .await + .expect_err("request should be cancelled") + .is_cancelled() + ); + tokio::task::yield_now().await; + assert!( + !client.response_waiters_guard().contains_key(&request_id), + "cancelling a call-service request must not leak its waiter" + ); + } + + #[cfg(feature = "voip-runtime")] + #[tokio::test] + async fn immediately_admitted_call_link_preserves_requested_token() { + let (client, _transport) = crate::test_utils::create_iq_test_client().await; + client + .persistence_manager() + .process_command(crate::store::commands::DeviceCommand::SetLid(Some( + Jid::new("111111111111111", Server::Lid).with_device(1), + ))) + .await; + let creator = Jid::new("333333333333333", Server::Lid); + let sent = client.wait_for_sent_node(crate::client::NodeFilter::tag("call")); + let join_client = client.clone(); + let join = tokio::spawn(async move { + join_client + .voip() + .join_call_link_with_audio( + "REQUESTED-CALL-LINK", + CallLinkMedia::Video, + AudioFormat::OPUS_16KHZ_60MS, + ) + .await + }); + let request = sent.await.expect("link_join request"); + let request_id = request + .as_node_ref() + .attrs() + .optional_string("id") + .expect("request id") + .into_owned(); + crate::test_utils::answer_iq( + &client, + &request_id, + &NodeBuilder::new("ack") + .attr("class", "call") + .attr("type", "link_join") + .attr("id", request_id.as_str()) + .children([NodeBuilder::new("group_info") + .attr("call-id", "ADMITTED-CALL-ID") + .attr("call-creator", creator) + .attr("transaction-id", "1") + .attr("connected-limit", "32") + .attr("media", "video") + .build()]) + .build(), + ) + .await; + + let admitted = join.await.expect("join task").expect("join response"); + assert_eq!(admitted.token, "REQUESTED-CALL-LINK"); + assert!(!admitted.in_waiting_room); + let generation = client + .call_registry() + .generation_of("ADMITTED-CALL-ID") + .expect("registered admitted call"); + client + .call_registry() + .remove_if_current("ADMITTED-CALL-ID", generation); + } } diff --git a/src/handlers/call.rs b/src/handlers/call.rs index d21453123..b43534433 100644 --- a/src/handlers/call.rs +++ b/src/handlers/call.rs @@ -3,17 +3,26 @@ use std::sync::Arc; use async_trait::async_trait; use log::{debug, warn}; #[cfg(feature = "voip-runtime")] +use wacore::message_processing::EncType; +#[cfg(feature = "voip-runtime")] +use wacore::messages::MessageUtils; +#[cfg(feature = "voip-runtime")] use wacore::stanza::call::{ REJECT_REASON_BUSY, TERMINATE_REASON_ACCEPTED_ELSEWHERE, TERMINATE_REASON_GROUP_CALL_ENDED, TERMINATE_REASON_REJECTED_ELSEWHERE, TERMINATE_REASON_TIMEOUT, TerminateParams, VideoStateParams, build_call_video_ack, build_terminate, build_video_state, }; use wacore::stanza::call::{build_offer_ack_receipt, parse_call_stanza}; +use wacore::stanza::group_call::build_call_control_ack; use wacore::types::call::{CallAction, IncomingCall, MissedCall, MissedReason}; #[cfg(feature = "voip-runtime")] use wacore::types::call::{CallEndedElsewhere, ElsewhereOutcome, VideoState}; use wacore::types::events::Event; #[cfg(feature = "voip-runtime")] +use wacore::types::group_call::{GroupCallDevice, GroupCallEncRekey}; +#[cfg(feature = "voip-runtime")] +use wacore::voip::GroupStateApply; +#[cfg(feature = "voip-runtime")] use wacore::voip::{CallEvent, PeerVideoTransition, VideoControl}; #[cfg(feature = "voip-runtime")] use wacore_binary::Jid; @@ -46,12 +55,54 @@ impl StanzaHandler for CallHandler { node: Arc, cancelled: &mut bool, ) -> bool { - // Silence the unused-parameter warning on the no-voip build (only the video arm cancels). - #[cfg(not(feature = "voip-runtime"))] - let _ = &cancelled; let nr = node.get(); + let typed_control_type = nr + .children() + .and_then(|children| children.first()) + .map(|child| child.tag.as_ref()) + .filter(|tag| matches!(*tag, "waiting_room_update" | "user_action" | "screen_share")); + let typed_control_ack = + typed_control_type.and_then(|action_type| build_call_control_ack(nr, action_type)); + if typed_control_type.is_some() { + // These actions require a typed ACK. Never let the router send its generic ACK, + // including when parsing fails or the typed send itself fails. + *cancelled = true; + } match parse_call_stanza(nr) { Ok(Some(call)) => { + #[cfg(feature = "voip-runtime")] + let group_transition_lock = matches!( + call.action, + CallAction::GroupUpdate { .. } + | CallAction::EncRekey { .. } + | CallAction::WaitingRoomUpdate { .. } + | CallAction::RaiseHand { .. } + | CallAction::ScreenShare { .. } + ) + .then(|| { + client + .call_registry() + .group_transition_lock(call.action.call_id()) + }) + .flatten(); + #[cfg(feature = "voip-runtime")] + let _group_transition_guard = if let Some(lock) = group_transition_lock.as_ref() { + Some(lock.lock().await) + } else { + None + }; + if let Some(action_type) = typed_control_type { + let Some(ack) = typed_control_ack else { + warn!( + "call: {action_type} stanza has no routable id; leaving it uncommitted" + ); + return true; + }; + if let Err(error) = client.send_node(ack).await { + warn!("call: failed to send typed {action_type} ack: {error}"); + return true; + } + } // Diagnostic: every recognized action we receive (offer/accept/reject/ // terminate/transport/relaylatency...). Lets us see whether the caller actually gets a // peer device's (which drives the sibling dismiss). @@ -90,6 +141,17 @@ impl StanzaHandler for CallHandler { client .call_registry() .mark_incoming_ringing(call.action.call_id()); + if let Some(group) = call.group.as_deref() { + let mut session = wacore::voip::CallSession::new_incoming( + call.action.call_id(), + call.from.clone(), + call.action.call_creator().clone(), + ); + session.is_video = + matches!(&call.action, CallAction::Offer { is_video: true, .. }); + session.group = Some(group.clone()); + client.call_registry().insert_ringing_group(session); + } } if is_offer && let Err(e) = send_offer_ack_receipt(&client, &call).await { warn!("call: failed to send offer ack receipt: {e}"); @@ -134,6 +196,28 @@ impl StanzaHandler for CallHandler { } return true; } + #[cfg(feature = "voip-runtime")] + if matches!( + &call.action, + CallAction::PreAccept { .. } | CallAction::Accept { .. } + ) && let Some(capability) = nr + .children() + .and_then(|children| children.first()) + .and_then(|action| action.get_optional_child("capability")) + && let Some(bytes) = + capability.content_bytes().filter(|bytes| !bytes.is_empty()) + { + let mut attrs = capability.attrs(); + let version = attrs + .optional_u64("ver") + .and_then(|version| u32::try_from(version).ok()) + .unwrap_or(1); + client.call_registry().set_group_invite_peer_device( + call.action.call_id(), + GroupCallDevice::new(call.from.clone()) + .with_capability(version, bytes.to_vec()), + ); + } // Caller-side: key our recv path to the device that actually answered. We dial the // base callee LID, but a companion answers from `:N` and encrypts under its own // device id; without this every inbound frame decrypts to garbage. One-shot, and a @@ -213,6 +297,152 @@ impl StanzaHandler for CallHandler { let mut dispatch_call = true; #[cfg(not(feature = "voip-runtime"))] let dispatch_call = true; + #[cfg(feature = "voip-runtime")] + match &call.action { + CallAction::GroupUpdate { update } => { + dispatch_call = match client + .call_registry() + .apply_group_update(update.clone()) + { + GroupStateApply::Applied => { + client + .call_registry() + .send_group_update(&update.call_id, update.clone()); + if update.rekey_requested { + match crate::voip::facade::fanout_group_epoch( + &client, update, + ) + .await + { + Ok(raw_epoch) => { + if !client.call_registry().send_group_epoch( + &update.call_id, + update.transaction_id, + raw_epoch, + ) { + warn!( + "call: local group epoch consumer closed for {}", + update.call_id + ); + } + } + Err(error) => { + warn!( + "call: failed to distribute requested group epoch for {}: {error}", + update.call_id + ); + } + } + } + client.call_registry().send_call_event( + &update.call_id, + CallEvent::GroupUpdated(Box::new(update.clone())), + ); + true + } + GroupStateApply::Stale | GroupStateApply::UnknownCall => false, + GroupStateApply::IdentityMismatch + | GroupStateApply::InvalidSnapshot => { + warn!( + "call: rejected invalid group snapshot for {}", + update.call_id + ); + false + } + _ => false, + }; + } + CallAction::EncRekey { rekey } => { + dispatch_call = false; + let sender = routed_call_sender(&call); + if !client.call_registry().group_sender_authorized( + &rekey.call_id, + &rekey.call_creator, + &sender, + ) { + warn!( + "call: rejected group epoch from unauthorized sender for {}", + rekey.call_id + ); + } else { + match decrypt_group_epoch(&client, rekey, &sender).await { + Ok(raw_epoch) => { + if !client.call_registry().send_group_epoch( + &rekey.call_id, + rekey.transaction_id, + raw_epoch, + ) { + debug!( + "call: group epoch for {} has no active media consumer", + rekey.call_id + ); + } + } + Err(error) => { + warn!( + "call: rejected encrypted group epoch for {}: {error}", + rekey.call_id + ); + } + } + } + } + CallAction::WaitingRoomUpdate { room } => { + dispatch_call = + match client.call_registry().apply_waiting_room(room.clone()) { + GroupStateApply::Applied => { + client.call_registry().send_call_event( + &room.call_id, + CallEvent::WaitingRoomUpdated(Box::new(room.clone())), + ); + true + } + GroupStateApply::Stale | GroupStateApply::UnknownCall => false, + GroupStateApply::IdentityMismatch + | GroupStateApply::InvalidSnapshot => { + warn!( + "call: rejected invalid waiting-room snapshot for {}", + room.call_id + ); + false + } + _ => false, + }; + } + CallAction::RaiseHand { raised, .. } => { + let participant = routed_call_sender(&call).to_non_ad(); + if client.call_registry().set_raised_hand( + call.action.call_id(), + &participant, + *raised, + ) { + client.call_registry().send_call_event( + call.action.call_id(), + CallEvent::HandRaised { + participant, + raised: *raised, + }, + ); + } + } + CallAction::ScreenShare { screen_share, .. } => { + let participant = routed_call_sender(&call).to_non_ad(); + if client.call_registry().set_screen_share( + call.action.call_id(), + &participant, + screen_share.clone(), + ) { + client.call_registry().send_call_event( + call.action.call_id(), + CallEvent::ScreenShareChanged { + participant, + screen_share: screen_share.clone(), + }, + ); + } + } + _ => {} + } // In-call