Watch Together

const handleJoinRoom = async () => {
  const joinedRoom = await wtJoinRoom(
    roomCode.trim().toUpperCase(),
    selectedMedia.id,
    selectedMedia.title,
    buildMediaMatchKey(selectedMedia),
    nickname.trim()
  );
  
  // Auto-launch MPV if playback already started
  if (joinedRoom.is_playing) {
    await launchMpv(joinedRoom.current_position || 0);
  }
};

pub async fn join_room(
    &self,
    room_code: String,
    media_id: i64,
    media_title: Option<String>,
    media_match_key: Option<String>,
    nickname: String,
) -> Result<RoomInfo, String> {
    let mut session = WatchSession::new(media_id, false);
    
    let join_msg = ClientMessage::Join {
        room_code: room_code.to_uppercase(),
        nickname,
        client_id: session.client_id.clone(),
        media_id,
        media_title,
        media_match_key,
    };
    
    // Send join and wait for confirmation
    let room = send_and_wait_for_room(&mut write, &mut read, join_msg).await?;
    Ok(room)
}

const handleInviteFriend = async (friend: Friend) => {
  const inviteText = `Join my Watch Together room for "${room.media_title}". Room code: ${room.code}`;
  await sendChatMessage(friend.id, inviteText);
};

pub enum SyncAction {
    Play { position: f64 },
    Pause { position: f64 },
    Seek { position: f64 },
}

pub fn launch_mpv_wt(
    mpv_path: &str,
    file_or_url: &str,
    media_id: i64,
    session_id: &str,
    start_position: f64,
    auth_header: Option<&str>,
    is_host: bool,
) -> Result<(u32, WatchTogetherController), String> {
    let controller = WatchTogetherController::new(session_id, is_host);
    
    let mut cmd = std::process::Command::new(mpv_path);
    cmd.arg(controller.get_ipc_arg()); // --input-ipc-server=\\.\pipe\mpv-wt-{session}
    cmd.arg(format!("--start={}", start_position as i64));
    cmd.arg("--pause=yes"); // Start paused until everyone ready
    cmd.arg(file_or_url);
    
    let child = cmd.spawn()?;
    let pid = child.id();
    
    Ok((pid, controller))
}

const SEEK_THRESHOLD: f64 = 2.0;           // Hard seek if drift > 2s
const SLOWDOWN_KICKIN: f64 = 0.35;         // Start speed correction at 0.35s
const SLOWDOWN_RESET: f64 = 0.1;           // Reset to 1x speed when < 0.1s drift
const SLOWDOWN_RATE: f64 = 0.95;           // Slow to 95% speed when ahead
const SPEEDUP_RATE: f64 = 1.05;            // Speed up to 105% when behind
const REWIND_THRESHOLD: f64 = 4.0;         // Hard seek if 4s ahead
const FASTFORWARD_THRESHOLD: f64 = 5.0;    // Hard seek if 5s behind
const SEEK_COOLDOWN_MS: u64 = 1200;        // Prevent seek spam

pub async fn apply_sync(&self, server_position: f64, server_paused: bool) -> Result<(), String> {
    let state = self.local_state.write().await;
    let elapsed = state.last_update.elapsed().as_secs_f64();
    let estimated_position = if state.paused {
        state.position
    } else {
        state.position + (elapsed * state.speed)
    };
    
    let diff = estimated_position - server_position;
    
    // Tiered correction strategy
    if diff > REWIND_THRESHOLD || diff < -FASTFORWARD_THRESHOLD {
        // Very large drift - hard seek
        self.seek_to(server_position).await?;
        self.set_speed(1.0).await?;
    } else if diff.abs() > SEEK_THRESHOLD {
        // Medium drift - seek
        self.seek_to(server_position).await?;
        self.set_speed(1.0).await?;
    } else if diff > SLOWDOWN_KICKIN {
        // Slightly ahead - slow down
        self.set_speed(SLOWDOWN_RATE).await?;
    } else if diff < -SLOWDOWN_KICKIN {
        // Slightly behind - speed up
        self.set_speed(SPEEDUP_RATE).await?;
    } else if diff.abs() < SLOWDOWN_RESET {
        // Close enough - reset to normal
        self.set_speed(1.0).await?;
    }
    
    Ok(())
}

pub async fn set_paused(&self, paused: bool) -> Result<(), String> {
    // Increment counter to ignore the echo event
    self.ignoring_on_the_fly.fetch_add(1, Ordering::SeqCst);
    
    self.send_command(vec![
        "set_property".into(),
        "pause".into(),
        paused.into(),
    ]).await?;
    
    Ok(())
}

useEffect(() => {
  const unlisten = listen<WatchEvent>('wt-event', (event) => {
    const data = event.payload;
    
    switch (data.type) {
      case 'room_updated':
      case 'participant_changed':
        onSessionChange(data.room, sessionId, data.room.is_playing);
        break;
        
      case 'sync_command':
        // Applied in backend, just update UI
        setLastSyncTime(Date.now());
        break;
        
      case 'state_update':
        // Periodic authoritative position from server
        setLastSyncTime(Date.now());
        setIsConnected(true);
        break;
        
      case 'playback_started':
        setView('playing');
        launchMpv(data.position || 0);
        break;
        
      case 'disconnected':
        setIsConnected(false);
        onSessionChange(null, '', false);
        break;
    }
  });
  
  return () => { unlisten.then(fn => fn()); };
}, []);

fn get_relay_server_url() -> String {
    std::env::var("STREAMVAULT_WS_URL")
        .unwrap_or_else(|_| {
            if cfg!(debug_assertions) {
                "ws://localhost:3001/ws/watchtogether".to_string()
            } else {
                "wss://streamvault-backend-server.onrender.com/ws/watchtogether".to_string()
            }
        })
}

// Check RTT (Round Trip Time)
ServerMessage::StateUpdate { your_rtt, .. } => {
    if your_rtt > 500.0 {
        // High latency warning
    }
}

if (timeSinceSync > 10000) {
  // Show reconnect prompt
  <Button onClick={handleReconnect}>Reconnect</Button>
}