1use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};2use std::sync::{Arc, Mutex as StdMutex};3use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};45use iroh::{Endpoint, SecretKey, endpoint::presets};6use serde::Serialize;7use sigil_protocol::{8 InputEvent, InvitationGrants, KeyframeRequestReasonV3, PointerSurfaceDimensions,9 SignedSubscriptionCapability,10};11use tauri::{12 AppHandle, Emitter,13 ipc::{Channel, Response},14};1516use crate::commands::auth::derive_iroh_secret_from_key;17use crate::commands::enrollment::{connection_enrollment, mark_invitation_redeemed};18use crate::commands::state::{AppState, MediaFeedbackSender, development_direct_node_available};19use crate::media::adaptive_feedback::{20 CLIENT_MEDIA_FEEDBACK_CONNECT_TIMEOUT, open_negotiated_feedback_stream,21 run_media_feedback_session,22};23use crate::media::audio_delivery::{24 AudioStartRequest, MoqAudioStartRequest, lock_audio_deliveries, next_audio_generation,25 try_start_audio, try_start_moq_audio,26};27use crate::media::frame_channel::{close_generation_connection, next_media_generation};28use crate::media::input_delivery::{29 CLIENT_INPUT_QUEUE_CAPACITY, InputSession, PointerFeedbackPayload, V2InputContext,30 open_input_session, run_input_feedback, run_input_forwarder, v2_input_eligibility,31};32use crate::media::media_control::run_media_control_writer_v3;33use crate::media::moq_receiver::{MoqAuthenticationExpectation, open_upstream_moq_media};34use crate::media::network_diagnostics::{35 MediaDeliveryRole, NetworkSessionDiagnostics, lock_network_diagnostics,36};37use crate::media::session_control::{38 SESSION_CONTROL_COMMAND_CAPACITY, install_initial_snapshot, run_session_control,39};40use crate::media::transport::{41 CLIENT_ENDPOINT_CLOSE_TIMEOUT, ControlProtocol, MediaTransport, NegotiatedMediaStream,42 open_negotiated_media_stream,43};44use crate::media::video_delivery::{VideoDeliveryRequest, run_video_delivery};4546struct ClientConnectGuard {47 active: Arc<AtomicBool>,48 committed: bool,49}5051impl ClientConnectGuard {52 fn acquire(active: Arc<AtomicBool>) -> Result<Self, String> {53 active54 .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)55 .map_err(|_| "A client connection is already active or in progress".to_string())?;56 Ok(Self {57 active,58 committed: false,59 })60 }6162 fn commit(mut self) {63 self.committed = true;64 }65}6667impl Drop for ClientConnectGuard {68 fn drop(&mut self) {69 if !self.committed {70 self.active.store(false, Ordering::SeqCst);71 }72 }73}747576pub struct ConnectResult {77 pub connected: bool,78 pub host_node_id: Option<String>,79 pub development_mode: bool,80 pub media_transport: &'static str,81 pub control_protocol: &'static str,82 pub session_snapshot: Option<sigil_protocol::SessionSnapshotV2>,83 pub pointer_surface_dimensions: Option<PointerSurfaceDimensions>,84 pub relative_pointer_available: bool,85 pub pointer_position_feedback_available: bool,86 pub absolute_pointer_available: bool,87 pub keyboard_available: bool,88 pub text_available: bool,89 pub gamepad_available: bool,90 pub control_available: bool,91 pub audio_available: bool,92 pub audio_generation: Option<u64>,93 pub audio_error: Option<String>,94 pub media_generation: u64,95 pub adaptive_feedback_available: bool,96 pub adaptive_feedback_error: Option<String>,97 pub error: Option<String>,98}99100pub(crate) async fn connect_client(101 app: AppHandle,102 state: &AppState,103 pin: String,104 frame_channel: Channel<Response>,105 audio_channel: Channel<Response>,106 pointer_channel: Channel<PointerFeedbackPayload>,107 audio_supported: bool,108) -> Result<ConnectResult, String> {109 let _connection_serial = state.client_connection_serial.try_lock().map_err(|_| {110 "Another client connection or disconnection operation is in progress".to_string()111 })?;112 let connect_guard = ClientConnectGuard::acquire(Arc::clone(&state.client_connection_active))?;113114 // Binary WebCodecs delivery is the only video path; fail before any115 // network work so a webview without H.264 WebCodecs support surfaces a116 // clear error instead of a black stream.117 if !state.webcodecs.load(Ordering::SeqCst) {118 return Err(119 "WebCodecs H.264 decoding is unavailable in this webview; Portal cannot stream video"120 .to_string(),121 );122 }123124 let (client_secret, development_mode) = if let Some(node_id) = state.dev_connect_node_id {125 if !development_direct_node_available() {126 return Err(127 "Development direct-node routing requires a debug build or the explicit demo-direct-node feature"128 .to_string(),129 );130 }131 let _ = app.emit(132 "dev-connect-routing",133 serde_json::json!({134 "host_node_id": node_id.to_string(),135 "warning": "Passkey identity lookup skipped; this is not client authorization."136 }),137 );138 (SecretKey::generate(), true)139 } else {140 // FIDO2 derivation — 30s timeout so a missing/stuck key surfaces quickly.141 let client_secret = tokio::time::timeout(142 Duration::from_secs(30),143 tokio::task::spawn_blocking(move || derive_iroh_secret_from_key(&pin)),144 )145 .await146 .map_err(|_| "Security key timed out (30s). Make sure your key is connected.".to_string())?147 .map_err(|e| format!("Task failed: {}", e))?148 .map_err(|e| format!("FIDO2 error: {:?}", e))?;149150 // Key has been tapped — relay connection is next; update the UI overlay.151 let _ = app.emit("fido-done", ());152 (client_secret, false)153 };154155 let (host_node_id, grants, invitation) = if development_mode {156 (157 state158 .dev_connect_node_id159 .ok_or_else(|| "Development host routing disappeared".to_string())?,160 InvitationGrants::ALL,161 None,162 )163 } else {164 let enrollment = connection_enrollment(&app, client_secret.public())?;165 (166 enrollment.host_node_id,167 enrollment.grants,168 enrollment.pending_invitation,169 )170 };171 let mut handshake_nonce = [0_u8; 16];172 getrandom::fill(&mut handshake_nonce)173 .map_err(|error| format!("Failed to generate handshake nonce: {error}"))?;174 let endpoint = Endpoint::builder(presets::N0)175 .secret_key(client_secret)176 .bind()177 .await178 .map_err(|e| format!("Failed to bind endpoint: {}", e))?;179180 let _ = tokio::time::timeout(Duration::from_secs(10), endpoint.online()).await;181182 // Use just the node ID — the presets::N0 relay map handles geographic183 // routing and fallback across all N0 relays automatically.184 let addr = iroh::EndpointAddr::new(host_node_id);185186 let first_attempt =187 open_negotiated_media_stream(&endpoint, &addr, handshake_nonce, invitation.as_deref())188 .await;189 let (190 frame_conn,191 frame_recv,192 media_control_stream,193 media_negotiation,194 media_transport,195 control_protocol,196 ) = match first_attempt {197 Ok(result) => result,198 Err(invitation_error)199 if invitation.is_some()200 && invitation_error.contains("Portal peer is not authorized") =>201 {202 // Recover only the narrow crash window where Sigil durably203 // consumed the invitation but Portal did not durably clear it.204 // The replay itself remains rejected; a second, ticket-free205 // connection can succeed only as the already-enrolled Iroh206 // peer authenticated by the exact invited host.207 open_negotiated_media_stream(&endpoint, &addr, handshake_nonce, None)208 .await209 .map_err(|retry_error| {210 format!(211 "{invitation_error}; ticket-free enrollment recovery also failed: {retry_error}"212 )213 })?214 }215 Err(error) => return Err(error),216 };217 let media_session_id = media_negotiation.session_id();218 let pointer_surface_dimensions = media_negotiation.pointer_surface_dimensions();219 let initial_session_snapshot = match &media_negotiation {220 NegotiatedMediaStream::V2(negotiation) => Some(negotiation.initial_snapshot.clone()),221 NegotiatedMediaStream::V1(_) => None,222 };223 let advertised_media = initial_session_snapshot224 .as_ref()225 .map(|snapshot| snapshot.media.clone());226 let moq_authentication = match &media_negotiation {227 NegotiatedMediaStream::V2(negotiation) => Some(MoqAuthenticationExpectation {228 expected_host: *host_node_id.as_bytes(),229 subscription_capability: negotiation.media_subscription_capability.clone(),230 authorization_revision: negotiation.media_authorization_revision,231 now_unix: SystemTime::now()232 .duration_since(UNIX_EPOCH)233 .map_err(|_| "System clock is before the Unix epoch".to_string())?234 .as_secs(),235 }),236 NegotiatedMediaStream::V1(_) => None,237 };238 let subscription_expires_at_unix = match &media_negotiation {239 NegotiatedMediaStream::V2(negotiation) => Some(240 SignedSubscriptionCapability::decode(&negotiation.media_subscription_capability)241 .map_err(|error| format!("Invalid media subscription capability: {error}"))?242 .claims243 .expires_at_unix,244 ),245 NegotiatedMediaStream::V1(_) => None,246 };247 let media_generation = next_media_generation(&state.client_media_generation)?;248 let network_diagnostics = Arc::new(StdMutex::new(NetworkSessionDiagnostics::new(249 Instant::now(),250 false,251 )));252 if moq_authentication.is_some() {253 let mut diagnostics = lock_network_diagnostics(&network_diagnostics);254 diagnostics.configure_authenticated_media(MediaDeliveryRole::DirectHost);255 diagnostics.configure_v2_session(256 initial_session_snapshot257 .as_ref()258 .ok_or_else(|| "Control v2 omitted its diagnostics snapshot".to_string())?,259 subscription_expires_at_unix260 .ok_or_else(|| "Control v2 omitted its subscription expiry".to_string())?,261 Instant::now(),262 )?;263 }264 let media_verification_failures =265 lock_network_diagnostics(&network_diagnostics).media_verification_failure_counter();266 let (upstream_moq_media, upstream_moq_audio, frame_connection_for_stats) = if media_transport267 == MediaTransport::UpstreamMoq268 {269 let (receiver, audio_receiver, diagnostics_connection) = match open_upstream_moq_media(270 &endpoint,271 &addr,272 media_session_id,273 advertised_media.as_ref().map(|media| media.generation_id),274 advertised_media275 .as_ref()276 .map(|media| media.broadcast_name.as_str()),277 moq_authentication,278 media_verification_failures,279 )280 .await281 {282 Ok(media) => media,283 Err(error) => {284 // CONTROL already authenticated and owns the host's285 // one-client lease. A post-auth MoQ failure is286 // terminal, and must explicitly release that lease;287 // it must never fall through to a legacy media ALPN.288 frame_conn.close(1_u32.into(), b"upstream MoQ setup failed");289 let _ = tokio::time::timeout(CLIENT_ENDPOINT_CLOSE_TIMEOUT, endpoint.close()).await;290 return Err(error);291 }292 };293 (Some(receiver), audio_receiver, diagnostics_connection)294 } else {295 (None, None, frame_conn.clone())296 };297 if !development_mode && let Some(expected_invitation) = invitation.as_deref() {298 // An accepted media hello means Sigil durably committed the one-time299 // enrollment before returning. Future PIN/tap/play sessions send no300 // bearer credential and authenticate by the stable Iroh peer instead.301 mark_invitation_redeemed(&app, expected_invitation)?;302 }303304 // Feedback is a v3 sidecar for both the preferred upstream-MoQ transport305 // and the grouped-v3 compatibility path. Unsupported ALPN is normal306 // compatibility with older Sigil hosts; all other failures remain visible307 // diagnostics but never downgrade the authenticated media session.308 let (adaptive_feedback_stream, adaptive_feedback_error) = match tokio::time::timeout(309 CLIENT_MEDIA_FEEDBACK_CONNECT_TIMEOUT,310 open_negotiated_feedback_stream(&endpoint, &addr, handshake_nonce, media_session_id),311 )312 .await313 {314 Ok(Ok(stream)) => (stream, None),315 Ok(Err(error)) => {316 eprintln!("[client] adaptive feedback unavailable: {error}");317 (None, Some(error))318 }319 Err(_) => {320 let error = "Adaptive feedback negotiation timed out".to_string();321 eprintln!("[client] adaptive feedback unavailable: {error}");322 (None, Some(error))323 }324 };325326 let mut input_connection_for_stats = None;327 let (input_availability, v1_input_forwarder) = if control_protocol == ControlProtocol::V1 {328 let InputSession {329 connection: input_connection,330 send: input_send,331 recv: input_recv,332 capabilities: input_capabilities,333 availability: input_availability,334 } = open_input_session(&endpoint, &addr, handshake_nonce, media_session_id, grants).await?;335 lock_network_diagnostics(&network_diagnostics)336 .set_input_ack_negotiated(input_availability.input_ack);337 input_connection_for_stats = Some(input_connection.clone());338 if input_availability.pointer_position_feedback || input_availability.input_ack {339 tokio::spawn(run_input_feedback(340 input_recv,341 Arc::clone(&network_diagnostics),342 input_availability.pointer_position_feedback,343 input_availability.input_ack,344 pointer_channel,345 ));346 } else {347 drop(input_recv);348 }349 (input_availability, Some((input_send, input_capabilities)))350 } else {351 *state.v2_input_context.lock().await = Some(V2InputContext {352 endpoint: endpoint.clone(),353 address: addr.clone(),354 nonce: handshake_nonce,355 media_session_id,356 native_generation: media_generation,357 grants,358 pointer_channel,359 network_diagnostics: Arc::clone(&network_diagnostics),360 });361 (v2_input_eligibility(grants), None)362 };363364 let audio_generation = next_audio_generation(&state.audio_connection_generation)?;365 lock_audio_deliveries(&state.audio_deliveries).begin_generation(audio_generation)?;366 let audio_result = if control_protocol == ControlProtocol::V2 {367 let receiver = upstream_moq_audio368 .ok_or_else(|| "Host did not advertise an authenticated MoQ Opus track".to_string());369 match receiver {370 Ok(receiver) => try_start_moq_audio(371 app.clone(),372 receiver,373 MoqAudioStartRequest {374 audio_supported,375 audio_channel,376 audio_deliveries: Arc::clone(&state.audio_deliveries),377 connection_generation: Arc::clone(&state.audio_connection_generation),378 generation: audio_generation,379 },380 )381 .map(|()| None),382 Err(error) => Err(error),383 }384 } else {385 try_start_audio(386 app.clone(),387 &endpoint,388 AudioStartRequest {389 address: addr.clone(),390 handshake_nonce,391 media_session_id,392 audio_supported,393 audio_channel,394 audio_deliveries: Arc::clone(&state.audio_deliveries),395 connection_generation: Arc::clone(&state.audio_connection_generation),396 generation: audio_generation,397 },398 )399 .await400 .map(Some)401 };402 let mut audio_connection_for_stats = None;403 let (audio_available, connected_audio_generation, audio_error) = match audio_result {404 Ok(connection) => {405 if let Some(connection) = connection {406 audio_connection_for_stats = Some(connection.clone());407 *state.audio_connection.lock().await = Some((audio_generation, connection));408 }409 (true, Some(audio_generation), None)410 }411 Err(error) => {412 lock_audio_deliveries(&state.audio_deliveries).cancel_generation(audio_generation);413 (false, None, Some(error))414 }415 };416 let mut v1_input_receiver = None;417 if let Some((input_send, input_capabilities)) = v1_input_forwarder {418 let (tx, rx) = tokio::sync::mpsc::channel::<InputEvent>(CLIENT_INPUT_QUEUE_CAPACITY);419 *state.input_send.lock().await = Some(tx);420 v1_input_receiver = Some((input_send, rx, input_capabilities));421 }422423 {424 let mut ce = state.client_endpoint.lock().await;425 *ce = Some(endpoint.clone());426 }427 *state.media_connection.lock().await = Some((media_generation, frame_conn.clone()));428 let adaptive_feedback_available = adaptive_feedback_stream.is_some();429 if let Some((connection, send, recv)) = adaptive_feedback_stream {430 let (feedback_tx, feedback_rx) = tokio::sync::watch::channel(None);431 let feedback_sender: MediaFeedbackSender = feedback_tx;432 *state.media_feedback.lock().await =433 Some((media_generation, connection, feedback_sender.clone()));434 tokio::spawn(run_media_feedback_session(435 app.clone(),436 media_generation,437 send,438 recv,439 feedback_rx,440 Arc::clone(&network_diagnostics),441 ));442 } else {443 *state.media_feedback.lock().await = None;444 }445 let media_control_requests = if control_protocol == ControlProtocol::V1 {446 let (control_tx, control_rx) = tokio::sync::mpsc::channel(1);447 *state.media_control.lock().await = Some((media_generation, control_tx.clone()));448 tokio::spawn(run_media_control_writer_v3(449 media_control_stream,450 control_rx,451 ));452 if media_transport == MediaTransport::GroupedObjectsV3 {453 let _ = control_tx.try_send((KeyframeRequestReasonV3::Join, None));454 }455 Some(control_tx)456 } else {457 let initial_snapshot = initial_session_snapshot458 .clone()459 .ok_or_else(|| "Control v2 omitted its initial session snapshot".to_string())?;460 install_initial_snapshot(461 &app,462 &state.session_snapshot,463 media_generation,464 initial_snapshot.clone(),465 )466 .await?;467 let (control_tx, control_rx) = tokio::sync::mpsc::channel(SESSION_CONTROL_COMMAND_CAPACITY);468 *state.session_control.lock().await = Some((media_generation, control_tx));469 tokio::spawn(run_session_control(470 app.clone(),471 media_generation,472 initial_snapshot.media.generation_id,473 media_control_stream,474 frame_recv,475 control_rx,476 Arc::clone(&state.session_snapshot),477 Arc::clone(&network_diagnostics),478 ));479 None480 };481 let frame_events_in_flight = Arc::new(AtomicUsize::new(0));482 *state.frame_delivery.lock().await =483 Some((media_generation, Arc::clone(&frame_events_in_flight)));484485 // Input forwarder: absolute motion is latest-value state and may be486 // dropped at the 60 Hz boundary. Relative motion is displacement, so it487 // owns a separate accumulator and timer that coalesces rather than drops.488 let input_send_diagnostics = Arc::clone(&network_diagnostics);489 if let Some((input_send, rx, input_capabilities)) = v1_input_receiver {490 tokio::spawn(run_input_forwarder(491 input_send,492 rx,493 input_capabilities,494 input_send_diagnostics,495 ));496 }497498 tokio::spawn(run_video_delivery(VideoDeliveryRequest {499 app,500 endpoint,501 frame_connection: frame_connection_for_stats,502 input_connection: input_connection_for_stats,503 audio_connection: audio_connection_for_stats,504 network_diagnostics,505 media_control_requests,506 upstream_moq_media,507 frame_events_in_flight,508 frame_channel,509 media_transport,510 media_generation,511 connected_audio_generation,512 }));513514 let result = ConnectResult {515 connected: true,516 host_node_id: Some(host_node_id.to_string()),517 development_mode,518 media_transport: media_transport.diagnostic_name(),519 control_protocol: control_protocol.diagnostic_name(),520 session_snapshot: initial_session_snapshot,521 pointer_surface_dimensions,522 relative_pointer_available: input_availability.relative_pointer,523 pointer_position_feedback_available: input_availability.pointer_position_feedback,524 absolute_pointer_available: input_availability.absolute_pointer,525 keyboard_available: input_availability.keyboard,526 text_available: input_availability.text,527 gamepad_available: input_availability.gamepad,528 control_available: input_availability.control,529 audio_available,530 audio_generation: connected_audio_generation,531 audio_error,532 media_generation,533 adaptive_feedback_available,534 adaptive_feedback_error,535 error: None,536 };537 connect_guard.commit();538 Ok(result)539}540541pub(crate) async fn disconnect_client(state: &AppState) -> Result<bool, String> {542 let _connection_serial = state.client_connection_serial.lock().await;543 next_audio_generation(&state.audio_connection_generation)?;544 lock_audio_deliveries(&state.audio_deliveries).clear();545 *state.media_control.lock().await = None;546 *state.session_control.lock().await = None;547 *state.session_snapshot.lock().await = None;548 *state.v2_input_context.lock().await = None;549 if let Some((_generation, connection, _sender)) = state.media_feedback.lock().await.take() {550 connection.close(0_u32.into(), b"client disconnected");551 }552 // Do not rely on endpoint shutdown alone to retire the session. The frame553 // reader and diagnostics task both own connection clones, and a surviving554 // media connection keeps the host's encoder and one-client lease alive.555 // Closing the media connection explicitly gives the host an immediate,556 // protocol-level session boundary.557 close_generation_connection(state.media_connection.lock().await.take(), |connection| {558 connection.close(0_u32.into(), b"client disconnected");559 });560 close_generation_connection(state.audio_connection.lock().await.take(), |connection| {561 connection.close(0_u32.into(), b"client disconnected");562 });563 close_generation_connection(state.input_connection.lock().await.take(), |connection| {564 connection.close(0_u32.into(), b"client disconnected");565 });566 {567 let mut ce = state.client_endpoint.lock().await;568 if let Some(endpoint) = ce.take()569 && tokio::time::timeout(CLIENT_ENDPOINT_CLOSE_TIMEOUT, endpoint.close())570 .await571 .is_err()572 {573 eprintln!(574 "[client] timed out waiting for endpoint shutdown after explicit connection close"575 );576 }577 }578 {579 let mut input_send = state.input_send.lock().await;580 *input_send = None;581 }582 *state.frame_delivery.lock().await = None;583 state584 .client_connection_active585 .store(false, Ordering::SeqCst);586 Ok(true)587}588589590mod tests {591 use super::*;592593 594 fn client_connect_guard_rejects_overlap_and_resets_only_failed_attempts() {595 let active = Arc::new(AtomicBool::new(false));596 let attempt = ClientConnectGuard::acquire(Arc::clone(&active)).unwrap();597 assert!(ClientConnectGuard::acquire(Arc::clone(&active)).is_err());598 drop(attempt);599 assert!(!active.load(Ordering::SeqCst));600601 ClientConnectGuard::acquire(Arc::clone(&active))602 .unwrap()603 .commit();604 assert!(active.load(Ordering::SeqCst));605 assert!(ClientConnectGuard::acquire(Arc::clone(&active)).is_err());606 active.store(false, Ordering::SeqCst);607 }608}