1use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};2use std::sync::{Arc, Mutex as StdMutex};3use std::time::{Duration, Instant};45use iroh::{Endpoint, SecretKey, endpoint::presets};6use serde::Serialize;7use sigil_protocol::{8 InputEvent, InvitationGrants, KeyframeRequestReasonV3, PointerSurfaceDimensions,9};10use tauri::{11 AppHandle, Emitter,12 ipc::{Channel, Response},13};1415use crate::commands::auth::derive_iroh_secret_from_key;16use crate::commands::enrollment::{connection_enrollment, mark_invitation_redeemed};17use crate::commands::state::{AppState, MediaFeedbackSender, development_direct_node_available};18use crate::media::adaptive_feedback::{19 CLIENT_MEDIA_FEEDBACK_CONNECT_TIMEOUT, open_negotiated_feedback_stream,20 run_media_feedback_session,21};22use crate::media::audio_delivery::{23 AudioStartRequest, lock_audio_deliveries, next_audio_generation, try_start_audio,24};25use crate::media::frame_channel::{close_generation_connection, next_media_generation};26use crate::media::input_delivery::{27 CLIENT_INPUT_QUEUE_CAPACITY, InputSession, PointerFeedbackPayload, open_input_session,28 run_input_feedback, run_input_forwarder,29};30use crate::media::media_control::run_media_control_writer_v3;31use crate::media::moq_receiver::open_upstream_moq_media;32use crate::media::network_diagnostics::NetworkSessionDiagnostics;33use crate::media::transport::{34 CLIENT_ENDPOINT_CLOSE_TIMEOUT, MediaTransport, open_negotiated_media_stream,35};36use crate::media::video_delivery::{VideoDeliveryRequest, run_video_delivery};3738struct ClientConnectGuard {39 active: Arc<AtomicBool>,40 committed: bool,41}4243impl ClientConnectGuard {44 fn acquire(active: Arc<AtomicBool>) -> Result<Self, String> {45 active46 .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)47 .map_err(|_| "A client connection is already active or in progress".to_string())?;48 Ok(Self {49 active,50 committed: false,51 })52 }5354 fn commit(mut self) {55 self.committed = true;56 }57}5859impl Drop for ClientConnectGuard {60 fn drop(&mut self) {61 if !self.committed {62 self.active.store(false, Ordering::SeqCst);63 }64 }65}666768pub struct ConnectResult {69 pub connected: bool,70 pub host_node_id: Option<String>,71 pub development_mode: bool,72 pub media_transport: &'static str,73 pub pointer_surface_dimensions: Option<PointerSurfaceDimensions>,74 pub relative_pointer_available: bool,75 pub pointer_position_feedback_available: bool,76 pub absolute_pointer_available: bool,77 pub keyboard_available: bool,78 pub text_available: bool,79 pub gamepad_available: bool,80 pub control_available: bool,81 pub audio_available: bool,82 pub audio_generation: Option<u64>,83 pub audio_error: Option<String>,84 pub media_generation: u64,85 pub adaptive_feedback_available: bool,86 pub adaptive_feedback_error: Option<String>,87 pub error: Option<String>,88}8990pub(crate) async fn connect_client(91 app: AppHandle,92 state: &AppState,93 pin: String,94 frame_channel: Channel<Response>,95 audio_channel: Channel<Response>,96 pointer_channel: Channel<PointerFeedbackPayload>,97 audio_supported: bool,98) -> Result<ConnectResult, String> {99 let _connection_serial = state.client_connection_serial.try_lock().map_err(|_| {100 "Another client connection or disconnection operation is in progress".to_string()101 })?;102 let connect_guard = ClientConnectGuard::acquire(Arc::clone(&state.client_connection_active))?;103104 // Binary WebCodecs delivery is the only video path; fail before any105 // network work so a webview without H.264 WebCodecs support surfaces a106 // clear error instead of a black stream.107 if !state.webcodecs.load(Ordering::SeqCst) {108 return Err(109 "WebCodecs H.264 decoding is unavailable in this webview; Portal cannot stream video"110 .to_string(),111 );112 }113114 let (client_secret, development_mode) = if let Some(node_id) = state.dev_connect_node_id {115 if !development_direct_node_available() {116 return Err(117 "Development direct-node routing requires a debug build or the explicit demo-direct-node feature"118 .to_string(),119 );120 }121 let _ = app.emit(122 "dev-connect-routing",123 serde_json::json!({124 "host_node_id": node_id.to_string(),125 "warning": "Passkey identity lookup skipped; this is not client authorization."126 }),127 );128 (SecretKey::generate(), true)129 } else {130 // FIDO2 derivation — 30s timeout so a missing/stuck key surfaces quickly.131 let client_secret = tokio::time::timeout(132 Duration::from_secs(30),133 tokio::task::spawn_blocking(move || derive_iroh_secret_from_key(&pin)),134 )135 .await136 .map_err(|_| "Security key timed out (30s). Make sure your key is connected.".to_string())?137 .map_err(|e| format!("Task failed: {}", e))?138 .map_err(|e| format!("FIDO2 error: {:?}", e))?;139140 // Key has been tapped — relay connection is next; update the UI overlay.141 let _ = app.emit("fido-done", ());142 (client_secret, false)143 };144145 let (host_node_id, grants, invitation) = if development_mode {146 (147 state148 .dev_connect_node_id149 .ok_or_else(|| "Development host routing disappeared".to_string())?,150 InvitationGrants::ALL,151 None,152 )153 } else {154 let enrollment = connection_enrollment(&app, client_secret.public())?;155 (156 enrollment.host_node_id,157 enrollment.grants,158 enrollment.pending_invitation,159 )160 };161 let mut handshake_nonce = [0_u8; 16];162 getrandom::fill(&mut handshake_nonce)163 .map_err(|error| format!("Failed to generate handshake nonce: {error}"))?;164 let endpoint = Endpoint::builder(presets::N0)165 .secret_key(client_secret)166 .bind()167 .await168 .map_err(|e| format!("Failed to bind endpoint: {}", e))?;169170 let _ = tokio::time::timeout(Duration::from_secs(10), endpoint.online()).await;171172 // Use just the node ID — the presets::N0 relay map handles geographic173 // routing and fallback across all N0 relays automatically.174 let addr = iroh::EndpointAddr::new(host_node_id);175176 let first_attempt =177 open_negotiated_media_stream(&endpoint, &addr, handshake_nonce, invitation.as_deref())178 .await;179 let (frame_conn, frame_recv, media_control_stream, media_negotiation, media_transport) =180 match first_attempt {181 Ok(result) => result,182 Err(invitation_error)183 if invitation.is_some()184 && invitation_error.contains("Portal peer is not authorized") =>185 {186 // Recover only the narrow crash window where Sigil durably187 // consumed the invitation but Portal did not durably clear it.188 // The replay itself remains rejected; a second, ticket-free189 // connection can succeed only as the already-enrolled Iroh190 // peer authenticated by the exact invited host.191 open_negotiated_media_stream(&endpoint, &addr, handshake_nonce, None)192 .await193 .map_err(|retry_error| {194 format!(195 "{invitation_error}; ticket-free enrollment recovery also failed: {retry_error}"196 )197 })?198 }199 Err(error) => return Err(error),200 };201 let media_session_id = media_negotiation.session_id;202 let (upstream_moq_media, frame_connection_for_stats) = if media_transport203 == MediaTransport::UpstreamMoq204 {205 let (receiver, diagnostics_connection) =206 match open_upstream_moq_media(&endpoint, &addr, media_session_id).await {207 Ok(media) => media,208 Err(error) => {209 // CONTROL already authenticated and owns the host's210 // one-client lease. A post-auth MoQ failure is211 // terminal, and must explicitly release that lease;212 // it must never fall through to a legacy media ALPN.213 frame_conn.close(1_u32.into(), b"upstream MoQ setup failed");214 let _ =215 tokio::time::timeout(CLIENT_ENDPOINT_CLOSE_TIMEOUT, endpoint.close()).await;216 return Err(error);217 }218 };219 (Some(receiver), diagnostics_connection)220 } else {221 (None, frame_conn.clone())222 };223 let pointer_surface_dimensions = media_negotiation.pointer_surface_dimensions;224 if !development_mode && let Some(expected_invitation) = invitation.as_deref() {225 // An accepted media hello means Sigil durably committed the one-time226 // enrollment before returning. Future PIN/tap/play sessions send no227 // bearer credential and authenticate by the stable Iroh peer instead.228 mark_invitation_redeemed(&app, expected_invitation)?;229 }230231 // Feedback is a v3 sidecar for both the preferred upstream-MoQ transport232 // and the grouped-v3 compatibility path. Unsupported ALPN is normal233 // compatibility with older Sigil hosts; all other failures remain visible234 // diagnostics but never downgrade the authenticated media session.235 let (adaptive_feedback_stream, adaptive_feedback_error) = match tokio::time::timeout(236 CLIENT_MEDIA_FEEDBACK_CONNECT_TIMEOUT,237 open_negotiated_feedback_stream(&endpoint, &addr, handshake_nonce, media_session_id),238 )239 .await240 {241 Ok(Ok(stream)) => (stream, None),242 Ok(Err(error)) => {243 eprintln!("[client] adaptive feedback unavailable: {error}");244 (None, Some(error))245 }246 Err(_) => {247 let error = "Adaptive feedback negotiation timed out".to_string();248 eprintln!("[client] adaptive feedback unavailable: {error}");249 (None, Some(error))250 }251 };252253 let InputSession {254 connection: input_connection,255 send: input_send,256 recv: input_recv,257 capabilities: input_capabilities,258 availability: input_availability,259 } = open_input_session(&endpoint, &addr, handshake_nonce, media_session_id, grants).await?;260261 let network_diagnostics = Arc::new(StdMutex::new(NetworkSessionDiagnostics::new(262 Instant::now(),263 input_availability.input_ack,264 )));265266 if input_availability.pointer_position_feedback || input_availability.input_ack {267 let feedback_diagnostics = Arc::clone(&network_diagnostics);268 let pointer_feedback_enabled = input_availability.pointer_position_feedback;269 let input_ack_enabled = input_availability.input_ack;270 tokio::spawn(run_input_feedback(271 input_recv,272 feedback_diagnostics,273 pointer_feedback_enabled,274 input_ack_enabled,275 pointer_channel,276 ));277 } else {278 drop(input_recv);279 }280281 let audio_generation = next_audio_generation(&state.audio_connection_generation)?;282 lock_audio_deliveries(&state.audio_deliveries).begin_generation(audio_generation)?;283 let audio_result = try_start_audio(284 app.clone(),285 &endpoint,286 AudioStartRequest {287 address: addr.clone(),288 handshake_nonce,289 media_session_id,290 audio_supported,291 audio_channel,292 audio_deliveries: Arc::clone(&state.audio_deliveries),293 connection_generation: Arc::clone(&state.audio_connection_generation),294 generation: audio_generation,295 },296 )297 .await;298 let mut audio_connection_for_stats = None;299 let (audio_available, connected_audio_generation, audio_error) = match audio_result {300 Ok(connection) => {301 audio_connection_for_stats = Some(connection.clone());302 *state.audio_connection.lock().await = Some((audio_generation, connection));303 (true, Some(audio_generation), None)304 }305 Err(error) => {306 lock_audio_deliveries(&state.audio_deliveries).cancel_generation(audio_generation);307 (false, None, Some(error))308 }309 };310 let media_generation = next_media_generation(&state.client_media_generation)?;311312 let (tx, rx) = tokio::sync::mpsc::channel::<InputEvent>(CLIENT_INPUT_QUEUE_CAPACITY);313 {314 let mut input_send_guard = state.input_send.lock().await;315 *input_send_guard = Some(tx);316 }317318 {319 let mut ce = state.client_endpoint.lock().await;320 *ce = Some(endpoint.clone());321 }322 *state.media_connection.lock().await = Some((media_generation, frame_conn.clone()));323 let adaptive_feedback_available = adaptive_feedback_stream.is_some();324 if let Some((connection, send, recv)) = adaptive_feedback_stream {325 let (feedback_tx, feedback_rx) = tokio::sync::watch::channel(None);326 let feedback_sender: MediaFeedbackSender = feedback_tx;327 *state.media_feedback.lock().await =328 Some((media_generation, connection, feedback_sender.clone()));329 tokio::spawn(run_media_feedback_session(330 app.clone(),331 media_generation,332 send,333 recv,334 feedback_rx,335 ));336 } else {337 *state.media_feedback.lock().await = None;338 }339 let media_control_requests = {340 let (control_tx, control_rx) = tokio::sync::mpsc::channel(1);341 *state.media_control.lock().await = Some((media_generation, control_tx.clone()));342 tokio::spawn(run_media_control_writer_v3(343 media_control_stream,344 control_rx,345 ));346 if media_transport == MediaTransport::GroupedObjectsV3 {347 let _ = control_tx.try_send((KeyframeRequestReasonV3::Join, None));348 }349 Some(control_tx)350 };351 let frame_events_in_flight = Arc::new(AtomicUsize::new(0));352 *state.frame_delivery.lock().await =353 Some((media_generation, Arc::clone(&frame_events_in_flight)));354355 // Input forwarder: absolute motion is latest-value state and may be356 // dropped at the 60 Hz boundary. Relative motion is displacement, so it357 // owns a separate accumulator and timer that coalesces rather than drops.358 let input_send_diagnostics = Arc::clone(&network_diagnostics);359 tokio::spawn(run_input_forwarder(360 input_send,361 rx,362 input_capabilities,363 input_send_diagnostics,364 ));365366 tokio::spawn(run_video_delivery(VideoDeliveryRequest {367 app,368 endpoint,369 frame_recv,370 frame_connection: frame_connection_for_stats,371 input_connection,372 audio_connection: audio_connection_for_stats,373 network_diagnostics,374 media_control_requests,375 upstream_moq_media,376 frame_events_in_flight,377 frame_channel,378 media_transport,379 media_generation,380 connected_audio_generation,381 }));382383 let result = ConnectResult {384 connected: true,385 host_node_id: Some(host_node_id.to_string()),386 development_mode,387 media_transport: media_transport.diagnostic_name(),388 pointer_surface_dimensions,389 relative_pointer_available: input_availability.relative_pointer,390 pointer_position_feedback_available: input_availability.pointer_position_feedback,391 absolute_pointer_available: input_availability.absolute_pointer,392 keyboard_available: input_availability.keyboard,393 text_available: input_availability.text,394 gamepad_available: input_availability.gamepad,395 control_available: input_availability.control,396 audio_available,397 audio_generation: connected_audio_generation,398 audio_error,399 media_generation,400 adaptive_feedback_available,401 adaptive_feedback_error,402 error: None,403 };404 connect_guard.commit();405 Ok(result)406}407408pub(crate) async fn disconnect_client(state: &AppState) -> Result<bool, String> {409 let _connection_serial = state.client_connection_serial.lock().await;410 next_audio_generation(&state.audio_connection_generation)?;411 lock_audio_deliveries(&state.audio_deliveries).clear();412 *state.media_control.lock().await = None;413 if let Some((_generation, connection, _sender)) = state.media_feedback.lock().await.take() {414 connection.close(0_u32.into(), b"client disconnected");415 }416 // Do not rely on endpoint shutdown alone to retire the session. The frame417 // reader and diagnostics task both own connection clones, and a surviving418 // media connection keeps the host's encoder and one-client lease alive.419 // Closing the media connection explicitly gives the host an immediate,420 // protocol-level session boundary.421 close_generation_connection(state.media_connection.lock().await.take(), |connection| {422 connection.close(0_u32.into(), b"client disconnected");423 });424 close_generation_connection(state.audio_connection.lock().await.take(), |connection| {425 connection.close(0_u32.into(), b"client disconnected");426 });427 {428 let mut ce = state.client_endpoint.lock().await;429 if let Some(endpoint) = ce.take()430 && tokio::time::timeout(CLIENT_ENDPOINT_CLOSE_TIMEOUT, endpoint.close())431 .await432 .is_err()433 {434 eprintln!(435 "[client] timed out waiting for endpoint shutdown after explicit connection close"436 );437 }438 }439 {440 let mut input_send = state.input_send.lock().await;441 *input_send = None;442 }443 *state.frame_delivery.lock().await = None;444 state445 .client_connection_active446 .store(false, Ordering::SeqCst);447 Ok(true)448}449450451mod tests {452 use super::*;453454 455 fn client_connect_guard_rejects_overlap_and_resets_only_failed_attempts() {456 let active = Arc::new(AtomicBool::new(false));457 let attempt = ClientConnectGuard::acquire(Arc::clone(&active)).unwrap();458 assert!(ClientConnectGuard::acquire(Arc::clone(&active)).is_err());459 drop(attempt);460 assert!(!active.load(Ordering::SeqCst));461462 ClientConnectGuard::acquire(Arc::clone(&active))463 .unwrap()464 .commit();465 assert!(active.load(Ordering::SeqCst));466 assert!(ClientConnectGuard::acquire(Arc::clone(&active)).is_err());467 active.store(false, Ordering::SeqCst);468 }469}