1use std::sync::Arc;23use serde::Serialize;4use sigil_protocol::{5 ClientControlEnvelopeV2, FocusCommandActionV2, FocusCommandV2, ServerControlEnvelopeV2,6 SessionSnapshotV2, read_server_control_v2, write_client_control_v2,7};8use tauri::{AppHandle, Emitter};910use crate::commands::state::{SessionControlSender, SessionSnapshotState};11use crate::media::network_diagnostics::{NetworkSessionDiagnostics, lock_network_diagnostics};1213pub(crate) const SESSION_CONTROL_COMMAND_CAPACITY: usize = 4;141516pub(crate) struct SessionStatePayload {17 pub(crate) native_generation: u64,18 pub(crate) snapshot: SessionSnapshotV2,19}202122pub(crate) struct FocusCommandResultPayload {23 pub(crate) native_generation: u64,24 pub(crate) request_id: u64,25 pub(crate) accepted: bool,26 pub(crate) revision: u64,27 pub(crate) message: Option<String>,28}2930pub(crate) async fn install_initial_snapshot(31 app: &AppHandle,32 state: &Arc<tokio::sync::Mutex<Option<SessionSnapshotState>>>,33 native_generation: u64,34 snapshot: SessionSnapshotV2,35) -> Result<(), String> {36 snapshot37 .validate()38 .map_err(|error| format!("Invalid initial session snapshot: {error}"))?;39 let payload = SessionStatePayload {40 native_generation,41 snapshot: snapshot.clone(),42 };43 *state.lock().await = Some((native_generation, snapshot));44 app.emit("session-state", payload)45 .map_err(|error| format!("Failed to publish initial session snapshot: {error}"))46}474849pub(crate) async fn run_session_control(50 app: AppHandle,51 native_generation: u64,52 media_generation_id: u64,53 mut send: iroh::endpoint::SendStream,54 mut recv: iroh::endpoint::RecvStream,55 mut commands: tokio::sync::mpsc::Receiver<ClientControlEnvelopeV2>,56 state: Arc<tokio::sync::Mutex<Option<SessionSnapshotState>>>,57 diagnostics: Arc<std::sync::Mutex<NetworkSessionDiagnostics>>,58) {59 loop {60 tokio::select! {61 command = commands.recv() => {62 let Some(command) = command else { break; };63 if let Err(error) = write_client_control_v2(&mut send, &command).await {64 eprintln!("[client] control v2 command write failed: {error}");65 break;66 }67 }68 envelope = read_server_control_v2(&mut recv) => {69 match envelope {70 Ok(Some(ServerControlEnvelopeV2::Snapshot { snapshot })) => {71 if snapshot.media.generation_id != media_generation_id {72 eprintln!("[client] control v2 snapshot changed media generation");73 break;74 }75 let mut guard = state.lock().await;76 let is_current_generation = guard.as_ref().is_some_and(|(generation, _)| {77 *generation == native_generation78 });79 if !is_current_generation {80 break;81 }82 let current_revision = guard83 .as_ref()84 .map(|(_, snapshot)| snapshot.revision)85 .unwrap_or(0);86 if snapshot.revision <= current_revision {87 let _ = lock_network_diagnostics(&diagnostics)88 .observe_session_snapshot(&snapshot, std::time::Instant::now());89 continue;90 }91 if let Err(error) = lock_network_diagnostics(&diagnostics)92 .observe_session_snapshot(&snapshot, std::time::Instant::now())93 {94 eprintln!("[client] invalid session diagnostics snapshot: {error}");95 break;96 }97 *guard = Some((native_generation, snapshot.clone()));98 drop(guard);99 let _ = app.emit(100 "session-state",101 SessionStatePayload { native_generation, snapshot },102 );103 }104 Ok(Some(ServerControlEnvelopeV2::FocusResult { result })) => {105 let _ = app.emit(106 "focus-command-result",107 FocusCommandResultPayload {108 native_generation,109 request_id: result.request_id,110 accepted: result.accepted,111 revision: result.revision,112 message: result.message,113 },114 );115 }116 Ok(None) => break,117 Err(error) => {118 lock_network_diagnostics(&diagnostics).mark_session_snapshot_invalid();119 eprintln!("[client] invalid control v2 message: {error}");120 break;121 }122 }123 }124 }125 }126}127128pub(crate) async fn send_focus_command(129 sender: &tokio::sync::Mutex<Option<(u64, SessionControlSender)>>,130 native_generation: u64,131 request_id: u64,132 action: FocusCommandActionV2,133 expected_revision: u64,134 expected_focus_generation: Option<u64>,135 expected_proposal_id: Option<u64>,136) -> Result<bool, String> {137 let sender = sender138 .lock()139 .await140 .as_ref()141 .filter(|(generation, _)| *generation == native_generation)142 .map(|(_, sender)| sender.clone())143 .ok_or_else(|| "No matching control v2 session is active".to_string())?;144 let command = ClientControlEnvelopeV2::Focus {145 command: FocusCommandV2 {146 request_id,147 action,148 slot: sigil_protocol::ControllerSlot::ZERO,149 expected_revision,150 expected_focus_generation,151 expected_proposal_id,152 },153 };154 command155 .validate()156 .map_err(|error| format!("Invalid focus command: {error}"))?;157 match sender.try_send(command) {158 Ok(()) => Ok(true),159 Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => Ok(false),160 Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {161 Err("Control v2 command channel closed".to_string())162 }163 }164}165166167pub(crate) async fn latest_snapshot(168 state: &tokio::sync::Mutex<Option<SessionSnapshotState>>,169 native_generation: u64,170) -> Option<SessionSnapshotV2> {171 state172 .lock()173 .await174 .as_ref()175 .filter(|(generation, _)| *generation == native_generation)176 .map(|(_, snapshot)| snapshot.clone())177}178179180mod tests {181 use super::*;182 use sigil_protocol::{183 ControllerSlot, FocusStateV2, FocusTransitionReasonV2, MediaGenerationDescriptorV2,184 ViewerPresenceId, ViewerPresenceV2,185 };186187 fn snapshot(revision: u64) -> SessionSnapshotV2 {188 let presence = ViewerPresenceId::new("viewer-1").unwrap();189 SessionSnapshotV2 {190 revision,191 self_presence_id: presence.clone(),192 viewers: vec![ViewerPresenceV2 {193 presence_id: presence,194 session_id: 7,195 input_capable: true,196 you: true,197 }],198 focus: FocusStateV2::Vacant {199 slot: ControllerSlot::ZERO,200 },201 focus_proposal: None,202 self_is_focus_owner: false,203 transition_reason: FocusTransitionReasonV2::Initial,204 media: MediaGenerationDescriptorV2 {205 generation_id: 7,206 broadcast_name: sigil_protocol::media_moq_broadcast_name(7).unwrap(),207 },208 }209 }210211 212 async fn latest_snapshot_is_owned_by_native_generation() {213 let state = tokio::sync::Mutex::new(Some((3, snapshot(2))));214 assert_eq!(latest_snapshot(&state, 3).await.unwrap().revision, 2);215 assert!(latest_snapshot(&state, 4).await.is_none());216 }217218 219 async fn focus_commands_are_bounded_and_revision_bound() {220 let (sender, mut receiver) = tokio::sync::mpsc::channel(1);221 let state = tokio::sync::Mutex::new(Some((5, sender)));222 assert!(223 send_focus_command(&state, 5, 1, FocusCommandActionV2::Request, 2, None, None)224 .await225 .unwrap()226 );227 assert!(228 !send_focus_command(&state, 5, 2, FocusCommandActionV2::Request, 2, None, None)229 .await230 .unwrap()231 );232 assert!(matches!(233 receiver.recv().await,234 Some(ClientControlEnvelopeV2::Focus { .. })235 ));236 assert!(237 send_focus_command(&state, 4, 3, FocusCommandActionV2::Request, 2, None, None)238 .await239 .is_err()240 );241 }242}