From 9352b493a11617b6febc53ad9c5bf06d8adce785 Mon Sep 17 00:00:00 2001 From: Ultradesu Date: Fri, 24 Jul 2026 04:11:01 +0300 Subject: [PATCH] Connected Devices: added remote control --- src/app/action.rs | 6 +- src/app/event.rs | 4 + src/app/mod.rs | 405 ++++++++++++++++++++++++++++++- src/app/popup.rs | 125 +++++++++- src/app/state.rs | 176 +++++++++++++- src/app/update.rs | 8 +- src/config/default_keymap.toml | 3 +- src/devices.rs | 431 ++++++++++++++++++++++++++++++++- src/library/mod.rs | 18 ++ src/ui/federation.rs | 71 +++++- src/ui/mod.rs | 18 +- src/ui/popup.rs | 118 ++++++++- 12 files changed, 1349 insertions(+), 34 deletions(-) diff --git a/src/app/action.rs b/src/app/action.rs index 52e2cb8..e681b96 100644 --- a/src/app/action.rs +++ b/src/app/action.rs @@ -42,6 +42,7 @@ pub enum Action { DownloadSelected, RemoveFromQueue, ClearQueue, + OpenConnectedDevices, GoToRelease, AddToPlaylist, NewPlaylist, @@ -99,7 +100,8 @@ impl Action { | Action::VolumeDown | Action::ToggleShuffle | Action::CycleRepeat - | Action::ToggleVisualizer => Category::Playback, + | Action::ToggleVisualizer + | Action::OpenConnectedDevices => Category::Playback, Action::QueueAddNext | Action::QueueAddLast | Action::DownloadSelected @@ -147,6 +149,7 @@ impl Action { Action::ToggleShuffle => Some(":shuffle"), Action::CycleRepeat => Some(":repeat [off|one|all]"), Action::ClearQueue => Some(":clear"), + Action::OpenConnectedDevices => None, Action::ToggleHelp => Some(":help"), Action::OpenSearch => Some("/text"), _ => None, @@ -188,6 +191,7 @@ impl Action { Action::DownloadSelected => "Federation: download to library".into(), Action::RemoveFromQueue => "Queue: remove selected".into(), Action::ClearQueue => "Queue: clear".into(), + Action::OpenConnectedDevices => "Connected devices…".into(), Action::GoToRelease => "Open the track's release".into(), Action::AddToPlaylist => "Add track to a playlist…".into(), Action::NewPlaylist => "Create a playlist".into(), diff --git a/src/app/event.rs b/src/app/event.rs index 58a4f5a..3475dea 100644 --- a/src/app/event.rs +++ b/src/app/event.rs @@ -137,4 +137,8 @@ pub enum AppEvent { DeviceConnectResult(Result), /// Incoming pairing request that passed the invite-secret check. DevicePairingRequest(crate::devices::PendingPairing), + /// Trusted device playback state, delivered by personal-device sync. + DevicePlayback(crate::devices::PlaybackSnapshot), + /// Playback command addressed to this device. + PlaybackCommand(crate::devices::PlaybackCommand), } diff --git a/src/app/mod.rs b/src/app/mod.rs index 109d5f3..c569b4c 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -3,7 +3,7 @@ mod cmdline; pub mod command; pub mod event; pub mod input; -mod popup; +pub(crate) mod popup; pub mod state; pub mod update; @@ -98,6 +98,12 @@ pub async fn run( let federation = crate::federation::Federation::new(Arc::clone(&library), Arc::clone(&devices)); state.federation.settings = federation.settings(); state.federation.devices = Some(devices.status()); + if let Ok((device_id, device_name)) = devices.identity_summary() { + state.device_playback.self_device_id = device_id.clone(); + state.device_playback.self_device_name = device_name.clone(); + state.device_playback.active_device_id = Some(device_id); + state.device_playback.active_device_name = Some(device_name); + } let player_events = event_tx.clone(); let mut runtime = Runtime { event_tx, @@ -169,11 +175,212 @@ pub async fn run( } fn sync_player_shared(state: &mut AppState, runtime: &Runtime) { - if state.player.current.is_some() && !runtime.player_start_pending { + if state.device_playback.is_control() { + extrapolate_control_position(state); + } else if state.player.current.is_some() && !runtime.player_start_pending { state.player.position_secs = runtime.player.shared.position().as_secs_f64(); state.player.paused = runtime.player.shared.paused(); } - state.player.audio_analysis = runtime.player.shared.audio_analysis(); + state.player.audio_analysis = if state.device_playback.is_control() { + player::AudioAnalysisSnapshot::default() + } else { + runtime.player.shared.audio_analysis() + }; + publish_playback_snapshot(state, runtime); +} + +fn unix_time_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) +} + +fn playback_repeat_to_wire(mode: state::RepeatMode) -> crate::devices::PlaybackRepeat { + match mode { + state::RepeatMode::Off => crate::devices::PlaybackRepeat::Off, + state::RepeatMode::One => crate::devices::PlaybackRepeat::One, + state::RepeatMode::All => crate::devices::PlaybackRepeat::All, + } +} + +fn playback_repeat_from_wire(mode: crate::devices::PlaybackRepeat) -> state::RepeatMode { + match mode { + crate::devices::PlaybackRepeat::Off => state::RepeatMode::Off, + crate::devices::PlaybackRepeat::One => state::RepeatMode::One, + crate::devices::PlaybackRepeat::All => state::RepeatMode::All, + } +} + +fn playback_state_from_ui(state: &AppState) -> crate::devices::PlaybackStateWire { + crate::devices::PlaybackStateWire { + queue: state + .player + .queue + .iter() + .map(crate::devices::PlaybackTrack::from_track) + .collect(), + queue_pos: state.player.queue_pos, + playing: state.player.playing, + paused: state.player.paused, + position_secs: state.player.position_secs, + volume: state.player.volume, + shuffle: state.player.shuffle, + repeat: playback_repeat_to_wire(state.player.repeat), + } +} + +fn playback_track_to_ui( + wire: &crate::devices::PlaybackTrack, + library: Option<&Library>, +) -> crate::library::models::TrackItem { + if let (Some(library), Some(content_id)) = (library, wire.content_id.as_deref()) + && let Ok(Some(track)) = library.track_by_content_id(content_id) + { + return track; + } + wire.to_track_item() +} + +fn apply_playback_state_to_ui( + state: &mut AppState, + wire: &crate::devices::PlaybackStateWire, + library: Option<&Library>, +) { + state.player.queue = wire + .queue + .iter() + .map(|track| playback_track_to_ui(track, library)) + .collect(); + state.player.queue_pos = wire + .queue_pos + .min(state.player.queue.len().saturating_sub(1)); + state.player.playing = wire.playing && !state.player.queue.is_empty(); + state.player.paused = wire.paused; + state.player.position_secs = wire.position_secs.max(0.0); + state.player.volume = wire.volume.min(100); + state.player.shuffle = wire.shuffle; + state.player.repeat = playback_repeat_from_wire(wire.repeat); + state.player.prefetched_pos = None; + state.player.original_order = None; + state.player.current = state + .player + .playing + .then(|| state.player.queue.get(state.player.queue_pos).cloned()) + .flatten(); + if !state.player.playing { + state.player.current = None; + state.player.paused = false; + } + state.queue_tab.cursor = state + .queue_tab + .cursor + .min(state.player.queue.len().saturating_sub(1)); +} + +fn publish_playback_snapshot(state: &mut AppState, runtime: &Runtime) { + let Ok((device_id, device_name)) = runtime.devices.identity_summary() else { + return; + }; + state.device_playback.self_device_id = device_id.clone(); + state.device_playback.self_device_name = device_name.clone(); + if state.device_playback.role == state::DevicePlaybackRole::Active { + state.device_playback.active_device_id = Some(device_id.clone()); + state.device_playback.active_device_name = Some(device_name.clone()); + } + let snapshot = crate::devices::PlaybackSnapshot { + device_id, + device_name, + active: state.device_playback.role == state::DevicePlaybackRole::Active, + updated_at_ms: unix_time_ms(), + state: playback_state_from_ui(state), + }; + runtime.devices.publish_playback(snapshot); +} + +fn extrapolate_control_position(state: &mut AppState) { + let Some(snapshot) = state.device_playback.last_remote_snapshot.as_ref() else { + return; + }; + if !snapshot.state.playing { + return; + } + let elapsed = if snapshot.state.paused { + 0.0 + } else { + (unix_time_ms().saturating_sub(snapshot.updated_at_ms) as f64 / 1000.0).max(0.0) + }; + let duration = state + .player + .current + .as_ref() + .map(|track| track.duration_seconds) + .unwrap_or(0.0); + let position = snapshot.state.position_secs + elapsed; + state.player.position_secs = if duration > 0.0 { + position.min(duration) + } else { + position + }; +} + +pub(crate) fn become_control_device( + state: &mut AppState, + runtime: &Runtime, + snapshot: crate::devices::PlaybackSnapshot, +) { + if state.device_playback.role == state::DevicePlaybackRole::Active { + runtime.player.stop(); + } + state.device_playback.role = state::DevicePlaybackRole::Control; + state.device_playback.active_device_id = Some(snapshot.device_id.clone()); + state.device_playback.active_device_name = Some(snapshot.device_name.clone()); + state.device_playback.last_remote_snapshot = Some(snapshot.clone()); + state + .device_playback + .remote + .insert(snapshot.device_id.clone(), snapshot.clone()); + apply_playback_state_to_ui(state, &snapshot.state, Some(runtime.library.as_ref())); + extrapolate_control_position(state); + state.status_message = Some(format!("controlling {}", snapshot.device_name)); +} + +pub(crate) fn become_active_device(state: &mut AppState, runtime: &mut Runtime, start_audio: bool) { + let was_control = state.device_playback.role == state::DevicePlaybackRole::Control; + state.device_playback.role = state::DevicePlaybackRole::Active; + let Ok((device_id, device_name)) = runtime.devices.identity_summary() else { + return; + }; + state.device_playback.self_device_id = device_id.clone(); + state.device_playback.self_device_name = device_name.clone(); + state.device_playback.active_device_id = Some(device_id); + state.device_playback.active_device_name = Some(device_name); + state.device_playback.last_remote_snapshot = None; + if was_control && start_audio && state.player.playing { + start_current_audio( + state, + runtime, + state.player.position_secs, + state.player.paused, + ); + } + publish_playback_snapshot(state, runtime); +} + +fn record_control_playback_state(state: &mut AppState, runtime: &Runtime) { + if !state.device_playback.is_control() { + return; + } + let Some(target) = state.device_playback.active_device_id.clone() else { + return; + }; + let command = crate::devices::PlaybackCommand::SetState { + state: playback_state_from_ui(state), + }; + if let Err(err) = runtime.devices.record_playback_command(&target, command) { + tracing::warn!(%err, target, "recording playback command failed"); + state.status_message = Some(format!("device command failed: {err:#}")); + } } const ARTISTS_PREFETCH_MARGIN: usize = 24; @@ -381,6 +588,10 @@ fn spawn_art_fetch(runtime: &Runtime, key: String, path: String, width: u16, hei /// Execute a side effect requested by update(). fn perform_effect(state: &mut AppState, runtime: &mut Runtime, effect: Effect) { + if state.device_playback.is_control() && is_controlled_playback_effect(&effect) { + perform_control_playback_effect(state, runtime, effect); + return; + } match effect { Effect::PlayCurrent => { play_current(state, runtime); @@ -412,6 +623,7 @@ fn perform_effect(state: &mut AppState, runtime: &mut Runtime, effect: Effect) { save_app_settings(state); } Effect::SetOptions => {} + Effect::PlaybackQueueChanged => {} Effect::EnqueueRelease { id, next } => { let library = Arc::clone(&runtime.library); let tx = runtime.event_tx.clone(); @@ -688,6 +900,48 @@ fn perform_effect(state: &mut AppState, runtime: &mut Runtime, effect: Effect) { } } +fn is_controlled_playback_effect(effect: &Effect) -> bool { + matches!( + effect, + Effect::PlayCurrent + | Effect::TogglePause + | Effect::StopPlayback + | Effect::SeekBy(_) + | Effect::SetVolume(_) + | Effect::SetOptions + | Effect::RemoveQueueIndices { .. } + | Effect::PlaybackQueueChanged + ) +} + +fn perform_control_playback_effect(state: &mut AppState, runtime: &mut Runtime, effect: Effect) { + match effect { + Effect::PlayCurrent => { + state.player.current = state.player.queue.get(state.player.queue_pos).cloned(); + state.player.playing = state.player.current.is_some(); + state.player.paused = false; + state.player.position_secs = 0.0; + } + Effect::TogglePause => {} + Effect::StopPlayback => { + state.player.playing = false; + state.player.current = None; + state.player.paused = false; + state.player.position_secs = 0.0; + } + Effect::SeekBy(delta) => { + state.player.position_secs = (state.player.position_secs + delta as f64).max(0.0); + } + Effect::SetVolume(volume) => { + state.player.volume = volume.min(100); + save_app_settings(state); + } + Effect::SetOptions | Effect::RemoveQueueIndices { .. } | Effect::PlaybackQueueChanged => {} + _ => {} + } + record_control_playback_state(state, runtime); +} + fn clamp_settings_cursor(state: &mut AppState) { let last = state::settings_rows(state).len().saturating_sub(1); state.settings_cursor = state.settings_cursor.min(last); @@ -1375,6 +1629,97 @@ fn apply_queue_refresh( } } +fn handle_device_playback_snapshot( + state: &mut AppState, + runtime: &mut Runtime, + snapshot: crate::devices::PlaybackSnapshot, +) { + if snapshot.device_id == state.device_playback.self_device_id { + return; + } + state + .device_playback + .remote + .insert(snapshot.device_id.clone(), snapshot.clone()); + let now = unix_time_ms(); + state.device_playback.online_devices = state + .device_playback + .remote + .values() + .filter(|snapshot| { + now.saturating_sub(snapshot.updated_at_ms) <= state::DEVICE_ONLINE_TTL_MS + }) + .count() + + 1; + + if !snapshot.active { + return; + } + if snapshot.state.playing { + let was_active = state.device_playback.role == state::DevicePlaybackRole::Active; + let was_paused = state.player.playing && state.player.paused; + become_control_device(state, runtime, snapshot.clone()); + if was_active && was_paused { + state.popup = Some(state::Popup::FedText { + title: "Active device moved".to_string(), + text: format!("Playback is now controlled by {}.", snapshot.device_name), + }); + } + } else if state.device_playback.is_control() + && state.device_playback.active_device_id.as_deref() == Some(snapshot.device_id.as_str()) + { + become_active_device(state, runtime, false); + state.status_message = Some("active playback moved to this device".into()); + } +} + +fn handle_playback_command( + state: &mut AppState, + runtime: &mut Runtime, + command: crate::devices::PlaybackCommand, +) { + match command { + crate::devices::PlaybackCommand::SetState { state: wire } => { + let old_current_id = state.player.current.as_ref().map(|track| track.id); + let old_playing = state.player.playing; + let old_paused = state.player.paused; + become_active_device(state, runtime, false); + apply_playback_state_to_ui(state, &wire, Some(runtime.library.as_ref())); + runtime + .player + .set_volume(player::amplitude(state.player.volume)); + if !state.player.playing { + runtime.player_start_pending = false; + runtime.player.stop(); + push_media_update(state, runtime, true); + publish_playback_snapshot(state, runtime); + return; + } + let current_id = state.player.current.as_ref().map(|track| track.id); + if !old_playing || old_current_id != current_id { + start_current_audio( + state, + runtime, + state.player.position_secs, + state.player.paused, + ); + push_media_metadata(state, runtime); + } else { + runtime.player.seek(std::time::Duration::from_secs_f64( + state.player.position_secs, + )); + if state.player.paused && !old_paused { + runtime.player.pause(); + } else if !state.player.paused && old_paused { + runtime.player.resume(); + } + } + push_media_update(state, runtime, true); + publish_playback_snapshot(state, runtime); + } + } +} + fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent) { match event { AppEvent::StatusMessage(message) => state.status_message = Some(message), @@ -1383,6 +1728,45 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent } AppEvent::DeviceSyncStatus(status) => { state.federation.devices = Some(status); + let now = unix_time_ms(); + state.device_playback.online_devices = state + .federation + .devices + .as_ref() + .map(|status| { + status + .devices + .iter() + .filter(|device| { + state::device_presence_section(state, device, now) + == state::DevicePresenceSection::Online + }) + .count() + }) + .unwrap_or(1) + .max(1); + let active_revoked = state.device_playback.is_control() + && state + .device_playback + .active_device_id + .as_ref() + .is_some_and(|active| { + state + .federation + .devices + .as_ref() + .and_then(|status| { + status + .devices + .iter() + .find(|device| device.device_id == *active) + }) + .is_some_and(|device| device.revoked) + }); + if active_revoked { + become_active_device(state, runtime, false); + state.status_message = Some("active playback moved to this device".into()); + } clamp_settings_cursor(state); } AppEvent::DeviceInvite(result) => match result { @@ -1415,6 +1799,12 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent }); state.federation.devices = Some(runtime.devices.status()); } + AppEvent::DevicePlayback(snapshot) => { + handle_device_playback_snapshot(state, runtime, snapshot); + } + AppEvent::PlaybackCommand(command) => { + handle_playback_command(state, runtime, command); + } AppEvent::FedSearchLoaded { seq, result } => { if runtime.search_seq.load(std::sync::atomic::Ordering::SeqCst) != seq { return; @@ -1769,6 +2159,7 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent AppEvent::EnqueueTracks { tracks, next } => { let count = tracks.len(); update::enqueue_tracks(state, tracks, next); + record_control_playback_state(state, runtime); state.status_message = Some(if next { format!("{count} tracks queued next") } else { @@ -1849,8 +2240,12 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent state.player.playing = false; state.player.paused = false; state.player.current = None; - runtime.player.stop(); - push_media_update(state, runtime, true); + if state.device_playback.is_control() { + record_control_playback_state(state, runtime); + } else { + runtime.player.stop(); + push_media_update(state, runtime, true); + } return; } }; diff --git a/src/app/popup.rs b/src/app/popup.rs index 16b9d9d..f16ba9e 100644 --- a/src/app/popup.rs +++ b/src/app/popup.rs @@ -12,11 +12,89 @@ use crossterm::event::{KeyCode, KeyEvent}; use crate::app::Runtime; use crate::app::event::AppEvent; use crate::app::state::{ - AppState, DeleteTarget, EditField, EditTarget, FedInputField, Loadable, Popup, - addable_playlists, + self, AppState, DeleteTarget, DevicePresenceSection, EditField, EditTarget, FedInputField, + Loadable, Popup, addable_playlists, }; use crate::library::models::{ReleaseEdit, TrackEdit, TrackItem}; +#[derive(Clone)] +pub(crate) struct ConnectedDevicePopupRow { + pub device_id: String, + pub name: String, + pub is_self: bool, + pub online: bool, + pub revoked: bool, + pub section: DevicePresenceSection, + pub active: bool, + pub playing: bool, + pub paused: bool, + pub queue_len: usize, +} + +pub(crate) fn connected_device_rows(state: &AppState) -> Vec { + let mut rows = Vec::new(); + if let Some(status) = &state.federation.devices { + let now = state::unix_time_ms(); + for index in state::device_status_order(state) { + let Some(device) = status.devices.get(index) else { + continue; + }; + let snapshot = state.device_playback.remote.get(&device.device_id); + let is_self = + device.is_self || device.device_id == state.device_playback.self_device_id; + let section = state::device_presence_section(state, device, now); + let online = section == DevicePresenceSection::Online; + rows.push(ConnectedDevicePopupRow { + device_id: device.device_id.clone(), + name: state::device_display_name(device), + is_self, + online, + revoked: device.revoked, + section, + active: state::device_status_active(state, &device.device_id), + playing: if is_self { + state.player.playing + } else { + snapshot.is_some_and(|snapshot| snapshot.state.playing) + }, + paused: if is_self { + state.player.paused + } else { + snapshot.is_some_and(|snapshot| snapshot.state.paused) + }, + queue_len: if is_self { + state.player.queue.len() + } else { + snapshot + .map(|snapshot| snapshot.state.queue.len()) + .unwrap_or(0) + }, + }); + } + } + if rows + .iter() + .all(|row| row.device_id != state.device_playback.self_device_id) + { + rows.insert( + 0, + ConnectedDevicePopupRow { + device_id: state.device_playback.self_device_id.clone(), + name: state.device_playback.self_device_name.clone(), + is_self: true, + online: true, + revoked: false, + section: DevicePresenceSection::Online, + active: state.device_playback.role == crate::app::state::DevicePlaybackRole::Active, + playing: state.player.playing, + paused: state.player.paused, + queue_len: state.player.queue.len(), + }, + ); + } + rows +} + pub fn handle_key(state: &mut AppState, runtime: &mut Runtime, key: KeyEvent) { let Some(popup) = state.popup.take() else { return; @@ -82,6 +160,49 @@ pub fn handle_key(state: &mut AppState, runtime: &mut Runtime, key: KeyEvent) { Popup::ConfirmDeviceRevoke { device_id, name } => { handle_device_revoke(state, runtime, device_id, name, key); } + Popup::ConnectedDevices { cursor } => { + handle_connected_devices(state, runtime, cursor, key); + } + } +} + +fn handle_connected_devices( + state: &mut AppState, + runtime: &mut Runtime, + cursor: usize, + key: KeyEvent, +) { + let rows = connected_device_rows(state); + let last = rows.len().saturating_sub(1); + match key.code { + KeyCode::Esc | KeyCode::Char('q') => {} + KeyCode::Up | KeyCode::Char('k') => { + state.popup = Some(Popup::ConnectedDevices { + cursor: cursor.saturating_sub(1), + }); + } + KeyCode::Down | KeyCode::Char('j') => { + state.popup = Some(Popup::ConnectedDevices { + cursor: (cursor + 1).min(last), + }); + } + KeyCode::Enter => { + if let Some(row) = rows.get(cursor.min(last)) { + if row.revoked { + state.status_message = Some("revoked device cannot be controlled".into()); + } else if row.is_self { + super::become_active_device(state, runtime, true); + state.status_message = Some("active playback moved to this device".into()); + } else if let Some(snapshot) = + state.device_playback.remote.get(&row.device_id).cloned() + { + super::become_control_device(state, runtime, snapshot); + } else { + state.status_message = Some("device has no playback snapshot yet".into()); + } + } + } + _ => state.popup = Some(Popup::ConnectedDevices { cursor }), } } diff --git a/src/app/state.rs b/src/app/state.rs index e157002..b565135 100644 --- a/src/app/state.rs +++ b/src/app/state.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::sync::Arc; use crate::app::input::LineEdit; @@ -542,6 +542,8 @@ pub enum Popup { }, /// Confirmation before revoking a trusted device. ConfirmDeviceRevoke { device_id: String, name: String }, + /// Connected playback devices and their current role/status. + ConnectedDevices { cursor: usize }, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -602,6 +604,128 @@ pub enum SettingsRow { VisualizationEdit, } +pub const DEVICE_ONLINE_TTL_MS: i64 = 45_000; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DevicePresenceSection { + Online, + Offline, + Revoked, +} + +impl DevicePresenceSection { + pub fn title(self) -> &'static str { + match self { + DevicePresenceSection::Online => "Online devices", + DevicePresenceSection::Offline => "Offline devices", + DevicePresenceSection::Revoked => "Revoked devices", + } + } + + fn sort_rank(self) -> u8 { + match self { + DevicePresenceSection::Online => 0, + DevicePresenceSection::Offline => 1, + DevicePresenceSection::Revoked => 2, + } + } +} + +pub fn unix_time_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) +} + +pub fn device_display_name(device: &crate::devices::DeviceStatusRow) -> String { + if device.name.trim().is_empty() { + device.device_id.chars().take(10).collect() + } else { + device.name.clone() + } +} + +pub fn device_status_active(state: &AppState, device_id: &str) -> bool { + state + .device_playback + .active_device_id + .as_deref() + .is_some_and(|active| active == device_id) + || state + .device_playback + .remote + .get(device_id) + .is_some_and(|snapshot| snapshot.active) +} + +pub fn device_status_online( + state: &AppState, + device: &crate::devices::DeviceStatusRow, + now_ms: i64, +) -> bool { + device.is_self + || device.device_id == state.device_playback.self_device_id + || device + .last_seen_ms + .is_some_and(|seen| now_ms.saturating_sub(seen) <= DEVICE_ONLINE_TTL_MS) + || state + .device_playback + .remote + .get(&device.device_id) + .is_some_and(|snapshot| { + now_ms.saturating_sub(snapshot.updated_at_ms) <= DEVICE_ONLINE_TTL_MS + }) +} + +pub fn device_presence_section( + state: &AppState, + device: &crate::devices::DeviceStatusRow, + now_ms: i64, +) -> DevicePresenceSection { + if device.revoked { + DevicePresenceSection::Revoked + } else if device_status_online(state, device, now_ms) { + DevicePresenceSection::Online + } else { + DevicePresenceSection::Offline + } +} + +pub fn device_status_order(state: &AppState) -> Vec { + let Some(status) = &state.federation.devices else { + return Vec::new(); + }; + let now = unix_time_ms(); + let mut indices: Vec = status + .devices + .iter() + .enumerate() + .filter_map(|(index, device)| (!device.revoked).then_some(index)) + .collect(); + indices.sort_by(|left, right| { + let left_device = &status.devices[*left]; + let right_device = &status.devices[*right]; + let left_section = device_presence_section(state, left_device, now).sort_rank(); + let right_section = device_presence_section(state, right_device, now).sort_rank(); + ( + left_section, + !device_status_active(state, &left_device.device_id), + !left_device.is_self, + device_display_name(left_device).to_ascii_lowercase(), + left_device.device_id.as_str(), + ) + .cmp(&( + right_section, + !device_status_active(state, &right_device.device_id), + !right_device.is_self, + device_display_name(right_device).to_ascii_lowercase(), + right_device.device_id.as_str(), + )) + }); + indices +} + pub fn settings_rows(state: &AppState) -> Vec { let mut rows = Vec::new(); rows.extend(FedRow::ALL.into_iter().map(SettingsRow::Federation)); @@ -609,9 +733,11 @@ pub fn settings_rows(state: &AppState) -> Vec { rows.push(SettingsRow::DeviceInvite); rows.push(SettingsRow::DeviceConnect); rows.push(SettingsRow::DeviceSyncNow); - if let Some(status) = &state.federation.devices { - rows.extend((0..status.devices.len()).map(SettingsRow::Device)); - } + rows.extend( + device_status_order(state) + .into_iter() + .map(SettingsRow::Device), + ); rows.push(SettingsRow::VisualizationClock); rows.extend( state @@ -800,6 +926,47 @@ impl Default for PlayerBar { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum DevicePlaybackRole { + #[default] + Active, + Control, +} + +impl DevicePlaybackRole { + pub fn label(self) -> &'static str { + match self { + DevicePlaybackRole::Active => "active", + DevicePlaybackRole::Control => "control", + } + } +} + +#[derive(Debug, Default)] +pub struct DevicePlaybackState { + pub role: DevicePlaybackRole, + pub self_device_id: String, + pub self_device_name: String, + pub active_device_id: Option, + pub active_device_name: Option, + pub online_devices: usize, + pub remote: BTreeMap, + pub last_remote_snapshot: Option, +} + +impl DevicePlaybackState { + pub fn is_control(&self) -> bool { + self.role == DevicePlaybackRole::Control + } + + pub fn active_label(&self) -> String { + self.active_device_name + .clone() + .or_else(|| self.active_device_id.clone()) + .unwrap_or_else(|| "this device".to_string()) + } +} + /// Single source of truth for the UI. Mutated only by `update()` and the /// event handlers in the main loop; views render from `&AppState`. #[derive(Debug, Default)] @@ -814,6 +981,7 @@ pub struct AppState { pub status_message: Option, pub settings_cursor: usize, pub player: PlayerBar, + pub device_playback: DevicePlaybackState, pub visualizer: crate::visualizer::VisualizerState, pub global: GlobalTab, pub artist_views: HashMap>, diff --git a/src/app/update.rs b/src/app/update.rs index 039eaff..5c6bfa8 100644 --- a/src/app/update.rs +++ b/src/app/update.rs @@ -44,6 +44,8 @@ pub enum Effect { restart_paused: Option, stop: bool, }, + /// Queue/options changed without a direct audio engine action. + PlaybackQueueChanged, /// Persist the federation settings and start/stop the node. FedApplySettings, /// Force an immediate library publish into the DHT. @@ -102,6 +104,10 @@ pub fn update(state: &mut AppState, action: Action) -> Option { state.help_visible = !state.help_visible; None } + Action::OpenConnectedDevices => { + state.popup = Some(super::state::Popup::ConnectedDevices { cursor: 0 }); + None + } Action::NextTab => { switch_tab(state, state.active_tab.next()); None @@ -642,7 +648,7 @@ fn delete_selected(state: &mut AppState) -> Option { } } if state.active_tab != Tab::Global { - return None; + return Some(Effect::PlaybackQueueChanged); } if state.global.stack.is_empty() { let artist = state.global.artists.get(state.global.selected).cloned()?; diff --git a/src/config/default_keymap.toml b/src/config/default_keymap.toml index 39a243e..77b189b 100644 --- a/src/config/default_keymap.toml +++ b/src/config/default_keymap.toml @@ -65,8 +65,7 @@ command = "QueueAddLast" [[keymaps]] key_sequence = "shift-c" -command = "ClearQueue" -context = "queue" +command = "OpenConnectedDevices" [[keymaps]] key_sequence = "d" diff --git a/src/devices.rs b/src/devices.rs index f473f49..0607423 100644 --- a/src/devices.rs +++ b/src/devices.rs @@ -19,6 +19,7 @@ use tokio::io::{AsyncRead, AsyncReadExt}; use crate::app::event::AppEvent; use crate::library::Library; +use crate::library::models::{ArtistRef, TrackItem}; pub const SYNC_ALPN: &[u8] = b"furumi/sync/1"; const CLIENT_VERSION: &str = env!("CARGO_PKG_VERSION"); @@ -27,7 +28,7 @@ const INVITE_TTL_MS: i64 = 10 * 60 * 1000; const PAIRING_WAIT_MS: i64 = 5 * 60 * 1000; const PAIRING_RETRY_DELAY: Duration = Duration::from_secs(1); const RESPONSE_DRAIN_TIMEOUT: Duration = Duration::from_secs(2); -const SYNC_INTERVAL: Duration = Duration::from_secs(30); +const SYNC_INTERVAL: Duration = Duration::from_secs(2); const MAX_LINE: usize = 8 * 1024 * 1024; const MAX_OPS_PER_BATCH: usize = 1000; @@ -79,6 +80,148 @@ pub struct DeviceSync { conn: Arc>, library: Arc, event_tx: Arc>>>, + playback: Arc>, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "snake_case")] +pub enum PlaybackRepeat { + #[default] + Off, + One, + All, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct PlaybackTrack { + pub id: i64, + pub title: String, + pub track_number: Option, + pub disc_number: Option, + pub duration_seconds: f64, + #[serde(default)] + pub artist_names: Vec, + #[serde(default)] + pub featured_artist_names: Vec, + pub release_id: i64, + pub release_title: String, + pub release_year: Option, + #[serde(default)] + pub file_path: String, + pub content_id: Option, + pub audio_format: Option, + pub audio_bitrate: Option, + pub audio_sample_rate: Option, + pub audio_bit_depth: Option, + pub file_size_bytes: Option, + #[serde(default)] + pub play_count: i64, + #[serde(default)] + pub fed: Option, +} + +impl PlaybackTrack { + pub fn from_track(track: &TrackItem) -> Self { + Self { + id: track.id, + title: track.title.clone(), + track_number: track.track_number, + disc_number: track.disc_number, + duration_seconds: track.duration_seconds, + artist_names: track + .artists + .iter() + .map(|artist| artist.name.clone()) + .collect(), + featured_artist_names: track + .featured_artists + .iter() + .map(|artist| artist.name.clone()) + .collect(), + release_id: track.release_id, + release_title: track.release_title.clone(), + release_year: track.release_year, + file_path: track.file_path.clone(), + content_id: track.content_id.clone(), + audio_format: track.audio_format.clone(), + audio_bitrate: track.audio_bitrate, + audio_sample_rate: track.audio_sample_rate, + audio_bit_depth: track.audio_bit_depth, + file_size_bytes: track.file_size_bytes, + play_count: track.play_count, + fed: track.fed.as_ref().and_then(SyncedFedTrack::from_fed), + } + } + + pub fn to_track_item(&self) -> TrackItem { + let refs = |names: &[String]| -> Vec { + names + .iter() + .map(|name| ArtistRef { + id: -1, + name: name.clone(), + }) + .collect() + }; + TrackItem { + id: self.id, + title: self.title.clone(), + track_number: self.track_number, + disc_number: self.disc_number, + duration_seconds: self.duration_seconds, + artists: refs(&self.artist_names), + featured_artists: refs(&self.featured_artist_names), + release_id: self.release_id, + release_title: self.release_title.clone(), + release_year: self.release_year, + file_path: self.file_path.clone(), + content_id: self.content_id.clone(), + cover_path: None, + audio_format: self.audio_format.clone(), + audio_bitrate: self.audio_bitrate, + audio_sample_rate: self.audio_sample_rate, + audio_bit_depth: self.audio_bit_depth, + file_size_bytes: self.file_size_bytes, + play_count: self.play_count, + fed: self.fed.as_ref().map(SyncedFedTrack::to_fed_track), + } + } +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct PlaybackStateWire { + #[serde(default)] + pub queue: Vec, + #[serde(default)] + pub queue_pos: usize, + pub playing: bool, + pub paused: bool, + pub position_secs: f64, + #[serde(default)] + pub volume: u8, + pub shuffle: bool, + pub repeat: PlaybackRepeat, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct PlaybackSnapshot { + pub device_id: String, + pub device_name: String, + pub active: bool, + pub updated_at_ms: i64, + pub state: PlaybackStateWire, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum PlaybackCommand { + SetState { state: PlaybackStateWire }, +} + +#[derive(Debug, Clone, Default)] +struct PlaybackShared { + local: Option, + remote: BTreeMap, } #[derive(Debug, Clone)] @@ -172,6 +315,10 @@ pub enum SyncOpPayload { target_device_id: String, target_max_seq_seen: i64, }, + PlaybackCommand { + target_device_id: String, + command: PlaybackCommand, + }, } impl SyncOpPayload { @@ -334,6 +481,8 @@ enum WireMessage { vector: BTreeMap, ops: Vec, snapshot: SyncSnapshot, + #[serde(default)] + playback: Option, }, PairResponse { accepted: bool, @@ -353,6 +502,8 @@ enum WireMessage { ops: Vec, #[serde(default)] snapshot: SyncSnapshot, + #[serde(default)] + playback: Option, }, Hello { group_id: String, @@ -361,6 +512,8 @@ enum WireMessage { vector: BTreeMap, ops: Vec, snapshot: SyncSnapshot, + #[serde(default)] + playback: Option, }, SyncResponse { accepted: bool, @@ -374,6 +527,8 @@ enum WireMessage { ops: Vec, #[serde(default)] snapshot: SyncSnapshot, + #[serde(default)] + playback: Option, }, } @@ -419,6 +574,7 @@ impl DeviceSync { conn: Arc::new(std::sync::Mutex::new(conn)), library, event_tx: Arc::new(std::sync::Mutex::new(None)), + playback: Arc::new(std::sync::Mutex::new(PlaybackShared::default())), }); sync.ensure_identity()?; sync.repair_like_order_from_sync_state()?; @@ -429,6 +585,32 @@ impl DeviceSync { *lock(&self.event_tx) = Some(tx); } + pub fn identity_summary(&self) -> Result<(String, String)> { + let identity = self.ensure_identity()?; + Ok((identity.device_id, identity.name)) + } + + pub fn publish_playback(&self, mut snapshot: PlaybackSnapshot) { + if snapshot.updated_at_ms <= 0 { + snapshot.updated_at_ms = now_ms(); + } + lock(&self.playback).local = Some(snapshot); + } + + pub fn record_playback_command( + &self, + target_device_id: &str, + command: PlaybackCommand, + ) -> Result<()> { + if target_device_id.trim().is_empty() { + return Ok(()); + } + self.record_local_op(SyncOpPayload::PlaybackCommand { + target_device_id: target_device_id.to_string(), + command, + }) + } + pub fn set_device_name(&self, name: &str, endpoint_ticket: Option<&str>) -> Result<()> { let name = if name.trim().is_empty() { "furumi".to_string() @@ -546,6 +728,7 @@ impl DeviceSync { let vector = self.vector()?; let ops = self.ops_for_peer(&invite.device_id)?; let snapshot = self.snapshot()?; + let playback = self.local_playback_snapshot(); let mut stream = service.open_stream(peer, SYNC_ALPN).await?; write_msg( &mut stream, @@ -559,6 +742,7 @@ impl DeviceSync { vector, ops, snapshot, + playback, }, ) .await?; @@ -575,11 +759,15 @@ impl DeviceSync { vector, ops, snapshot, + playback, .. } => { self.set_group_id(&group_id)?; if let Some(profile) = profile { self.apply_device_profile(&profile, true)?; + if let Some(playback) = playback { + self.apply_playback_snapshot(playback)?; + } } self.apply_device_profiles(&devices)?; self.apply_snapshot(snapshot)?; @@ -851,6 +1039,7 @@ impl DeviceSync { let vector = self.vector()?; let ops = self.ops_for_peer(&device.device_id)?; let snapshot = self.snapshot()?; + let playback = self.local_playback_snapshot(); let mut stream = service.open_stream(peer, SYNC_ALPN).await?; write_msg( &mut stream, @@ -861,6 +1050,7 @@ impl DeviceSync { vector, ops, snapshot, + playback, }, ) .await?; @@ -875,9 +1065,13 @@ impl DeviceSync { vector, ops, snapshot, + playback, .. } => { self.apply_device_profiles(&devices)?; + if let Some(playback) = playback { + self.apply_playback_snapshot(playback)?; + } self.apply_snapshot(snapshot)?; self.apply_ops(ops)?; self.note_peer_vector(&device.device_id, &vector)?; @@ -1192,10 +1386,44 @@ impl DeviceSync { &op.origin_device_id, *target_max_seq_seen, )?, + SyncOpPayload::PlaybackCommand { + target_device_id, + command, + } => { + self.apply_playback_command(target_device_id, command, &op.op_id)?; + false + } }; Ok(changed) } + fn apply_playback_command( + &self, + target_device_id: &str, + command: &PlaybackCommand, + op_id: &str, + ) -> Result<()> { + let identity = self.ensure_identity()?; + if target_device_id != identity.device_id { + return Ok(()); + } + let inserted = { + let conn = lock(&self.conn); + conn.execute( + "INSERT OR IGNORE INTO sync_playback_applied (op_id, applied_at_ms) + VALUES (?1, ?2)", + params![op_id, now_ms()], + )? + }; + if inserted == 0 { + return Ok(()); + } + if let Some(tx) = lock(&self.event_tx).as_ref() { + let _ = tx.send(AppEvent::PlaybackCommand(command.clone())); + } + Ok(()) + } + fn apply_device_trusted(&self, target_device_id: &str, hlc_ms: i64) -> Result { let was_revoked = { let conn = lock(&self.conn); @@ -1451,8 +1679,8 @@ impl DeviceSync { let conn = lock(&self.conn); conn.query_row( "SELECT present, position, hlc_ms, op_id - FROM sync_state_playlist_items - WHERE playlist_id = ?1 AND content_id = ?2", + FROM sync_state_playlist_items + WHERE playlist_id = ?1 AND content_id = ?2", params![playlist_id, content_id], |row| { Ok(( @@ -2082,6 +2310,34 @@ impl DeviceSync { } } + fn local_playback_snapshot(&self) -> Option { + lock(&self.playback).local.clone() + } + + fn apply_playback_snapshot(&self, snapshot: PlaybackSnapshot) -> Result<()> { + let identity = self.ensure_identity()?; + if snapshot.device_id == identity.device_id { + return Ok(()); + } + let should_send = { + let mut playback = lock(&self.playback); + let changed = playback + .remote + .get(&snapshot.device_id) + .is_none_or(|current| snapshot.updated_at_ms > current.updated_at_ms); + if changed { + playback + .remote + .insert(snapshot.device_id.clone(), snapshot.clone()); + } + changed + }; + if should_send && let Some(tx) = lock(&self.event_tx).as_ref() { + let _ = tx.send(AppEvent::DevicePlayback(snapshot)); + } + Ok(()) + } + fn notify_library_changed(&self) { if let Some(tx) = lock(&self.event_tx).as_ref() { let _ = tx.send(AppEvent::LibraryChanged { message: None }); @@ -2093,6 +2349,7 @@ impl DeviceSync { let conn = lock(&self.conn); let compactable = compactable_tombstone_ids(&conn)?; for (op_id, origin, seq) in compactable { + let revoked_target = tombstone_revoke_target(&conn, &op_id)?; conn.execute( "INSERT INTO sync_compacted (origin_device_id, max_seq) VALUES (?1, ?2) @@ -2101,6 +2358,9 @@ impl DeviceSync { params![origin, seq], )?; conn.execute("DELETE FROM sync_ops WHERE op_id = ?1", [op_id])?; + if let Some(device_id) = revoked_target { + delete_revoked_device_if_fully_compacted(&conn, &device_id)?; + } } conn.execute( "DELETE FROM sync_state_likes @@ -2145,6 +2405,9 @@ pub async fn sync_loop(sync: Arc, service: Arc) { if let Err(err) = sync.sync_once(Arc::clone(&service)).await { tracing::debug!("personal sync tick failed: {err:#}"); } + if let Some(tx) = lock(&sync.event_tx).as_ref() { + let _ = tx.send(AppEvent::DeviceSyncStatus(sync.status())); + } } } @@ -2164,6 +2427,7 @@ async fn serve_one( vector, ops, snapshot, + playback, } => { handle_pair_request( stream, @@ -2178,6 +2442,7 @@ async fn serve_one( vector, ops, snapshot, + playback, ) .await } @@ -2188,9 +2453,10 @@ async fn serve_one( vector, ops, snapshot, + playback, } => { handle_hello( - stream, sync, service, group_id, profile, devices, vector, ops, snapshot, + stream, sync, service, group_id, profile, devices, vector, ops, snapshot, playback, ) .await } @@ -2212,6 +2478,7 @@ async fn handle_pair_request( vector: BTreeMap, ops: Vec, snapshot: SyncSnapshot, + playback: Option, ) -> Result<()> { profile.endpoint_id = stream.peer_id.to_string(); let request_id = pair_request_id(&invite_id, &profile.device_id); @@ -2233,6 +2500,7 @@ async fn handle_pair_request( vector: BTreeMap::new(), ops: Vec::new(), snapshot: SyncSnapshot::default(), + playback: None, }, ) .await?; @@ -2303,6 +2571,7 @@ async fn handle_pair_request( vector: BTreeMap::new(), ops: Vec::new(), snapshot: SyncSnapshot::default(), + playback: None, }, ) .await?; @@ -2323,6 +2592,7 @@ async fn handle_pair_request( vector: BTreeMap::new(), ops: Vec::new(), snapshot: SyncSnapshot::default(), + playback: None, }, ) .await?; @@ -2347,6 +2617,9 @@ async fn handle_pair_request( sync.apply_device_profiles(&requester_group_devices)?; } sync.apply_device_profile(&profile, true)?; + if let Some(playback) = playback { + sync.apply_playback_snapshot(playback)?; + } sync.apply_snapshot(snapshot)?; sync.apply_ops(ops)?; sync.note_peer_vector(&profile.device_id, &vector)?; @@ -2362,6 +2635,7 @@ async fn handle_pair_request( let vector = sync.vector()?; let ops = sync.ops_for_peer(&profile.device_id)?; let snapshot = sync.snapshot()?; + let playback = sync.local_playback_snapshot(); write_msg( &mut stream, &WireMessage::PairResponse { @@ -2374,6 +2648,7 @@ async fn handle_pair_request( vector, ops, snapshot, + playback, }, ) .await?; @@ -2392,6 +2667,7 @@ async fn handle_hello( vector: BTreeMap, ops: Vec, snapshot: SyncSnapshot, + playback: Option, ) -> Result<()> { let identity = sync.ensure_identity()?; if group_id != identity.group_id { @@ -2404,6 +2680,7 @@ async fn handle_hello( vector: BTreeMap::new(), ops: Vec::new(), snapshot: SyncSnapshot::default(), + playback: None, }, ) .await?; @@ -2420,6 +2697,7 @@ async fn handle_hello( vector: BTreeMap::new(), ops: Vec::new(), snapshot: SyncSnapshot::default(), + playback: None, }, ) .await?; @@ -2429,6 +2707,9 @@ async fn handle_hello( profile.endpoint_id = stream.peer_id.to_string(); sync.apply_device_profile(&profile, false)?; sync.apply_device_profiles(&devices)?; + if let Some(playback) = playback { + sync.apply_playback_snapshot(playback)?; + } sync.apply_snapshot(snapshot)?; sync.apply_ops(ops)?; sync.note_peer_vector(&profile.device_id, &vector)?; @@ -2441,6 +2722,7 @@ async fn handle_hello( let vector = sync.vector()?; let ops = sync.ops_for_peer(&profile.device_id)?; let snapshot = sync.snapshot()?; + let playback = sync.local_playback_snapshot(); write_msg( &mut stream, &WireMessage::SyncResponse { @@ -2454,6 +2736,7 @@ async fn handle_hello( vector, ops, snapshot, + playback, }, ) .await?; @@ -2632,7 +2915,11 @@ CREATE TABLE IF NOT EXISTS sync_state_playlist_items ( op_id TEXT NOT NULL, PRIMARY KEY (playlist_id, content_id) ); -"#, +CREATE TABLE IF NOT EXISTS sync_playback_applied ( + op_id TEXT PRIMARY KEY, + applied_at_ms INTEGER NOT NULL +); + "#, )?; ensure_column( conn, @@ -2717,6 +3004,7 @@ fn payload_kind(payload: &SyncOpPayload) -> &'static str { SyncOpPayload::DeviceProfileSet { .. } => "device_profile_set", SyncOpPayload::DeviceTrusted { .. } => "device_trusted", SyncOpPayload::DeviceRevoked { .. } => "device_revoked", + SyncOpPayload::PlaybackCommand { .. } => "playback_command", } } @@ -2793,6 +3081,71 @@ fn compactable_tombstone_ids(conn: &Connection) -> Result Result> { + let payload_json: Option = conn + .query_row( + "SELECT payload_json FROM sync_ops WHERE op_id = ?1 AND kind = 'device_revoked'", + [op_id], + |row| row.get(0), + ) + .optional()?; + let Some(payload_json) = payload_json else { + return Ok(None); + }; + let payload: SyncOpPayload = serde_json::from_str(&payload_json)?; + Ok(match payload { + SyncOpPayload::DeviceRevoked { + target_device_id, .. + } => Some(target_device_id), + _ => None, + }) +} + +fn has_revoke_op_for_target(conn: &Connection, device_id: &str) -> Result { + let mut stmt = + conn.prepare("SELECT payload_json FROM sync_ops WHERE kind = 'device_revoked'")?; + let payloads = stmt + .query_map([], |row| row.get::<_, String>(0))? + .collect::>>()?; + for payload_json in payloads { + let Ok(SyncOpPayload::DeviceRevoked { + target_device_id, .. + }) = serde_json::from_str::(&payload_json) + else { + continue; + }; + if target_device_id == device_id { + return Ok(true); + } + } + Ok(false) +} + +fn delete_revoked_device_if_fully_compacted(conn: &Connection, device_id: &str) -> Result<()> { + let own_device_id = get_meta(conn, "device_id")?.unwrap_or_default(); + if device_id == own_device_id { + return Ok(()); + } + let origin_ops: i64 = conn.query_row( + "SELECT COUNT(*) FROM sync_ops WHERE origin_device_id = ?1", + [device_id], + |row| row.get(0), + )?; + if origin_ops > 0 || has_revoke_op_for_target(conn, device_id)? { + return Ok(()); + } + conn.execute( + "DELETE FROM sync_devices + WHERE device_id = ?1 AND revoked_at_ms IS NOT NULL", + [device_id], + )?; + conn.execute( + "DELETE FROM sync_peer_acks WHERE peer_device_id = ?1", + [device_id], + )?; + Ok(()) +} + fn peer_ack_floor_label(conn: &Connection) -> Result { let count: i64 = conn.query_row("SELECT COUNT(*) FROM sync_peer_acks", [], |row| row.get(0))?; if count == 0 { @@ -2964,18 +3317,23 @@ fn base64url_decode(value: &str) -> Result> { mod tests { use super::*; + static NEXT_TEST_DB: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); + fn test_sync() -> DeviceSync { let conn = Connection::open_in_memory().unwrap(); init_schema(&conn).unwrap(); + let unique = NEXT_TEST_DB.fetch_add(1, std::sync::atomic::Ordering::Relaxed); let library_path = std::env::temp_dir().join(format!( - "furumi-devices-test-{}-{}.sqlite3", + "furumi-devices-test-{}-{}-{}.sqlite3", std::process::id(), - now_ms() + now_ms(), + unique )); let sync = DeviceSync { conn: Arc::new(std::sync::Mutex::new(conn)), library: Arc::new(Library::open(&library_path).unwrap()), event_tx: Arc::new(std::sync::Mutex::new(None)), + playback: Arc::new(std::sync::Mutex::new(PlaybackShared::default())), }; sync.ensure_identity().unwrap(); sync @@ -2996,6 +3354,18 @@ mod tests { != 0 } + fn device_known(sync: &DeviceSync, device_id: &str) -> bool { + let conn = lock(&sync.conn); + conn.query_row( + "SELECT 1 FROM sync_devices WHERE device_id = ?1", + [device_id], + |row| row.get::<_, i64>(0), + ) + .optional() + .unwrap() + .is_some() + } + fn test_fed_track(content_id: &str) -> crate::federation::FedTrack { crate::federation::FedTrack { item_id: "fed_item_1".to_string(), @@ -3042,6 +3412,53 @@ mod tests { ); } + #[test] + fn compacted_device_revoke_removes_device_row() { + let sync = test_sync(); + let device_id = "dev_old"; + + sync.apply_device_trusted(device_id, 10).unwrap(); + assert!(device_known(&sync, device_id)); + + sync.revoke_device(device_id).unwrap(); + assert!(!device_known(&sync, device_id)); + } + + #[test] + fn playback_command_is_targeted_and_deduplicated() { + let sync = test_sync(); + let identity = sync.ensure_identity().unwrap(); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + sync.set_event_tx(tx); + let command = PlaybackCommand::SetState { + state: PlaybackStateWire { + queue: Vec::new(), + queue_pos: 0, + playing: false, + paused: false, + position_secs: 0.0, + volume: 42, + shuffle: false, + repeat: PlaybackRepeat::Off, + }, + }; + + sync.apply_playback_command("dev_other", &command, "op_other") + .unwrap(); + assert!(rx.try_recv().is_err()); + + sync.apply_playback_command(&identity.device_id, &command, "op_1") + .unwrap(); + assert!(matches!( + rx.try_recv().unwrap(), + crate::app::event::AppEvent::PlaybackCommand(_) + )); + + sync.apply_playback_command(&identity.device_id, &command, "op_1") + .unwrap(); + assert!(rx.try_recv().is_err()); + } + #[test] fn newer_device_trust_reactivates_revoked_device() { let sync = test_sync(); diff --git a/src/library/mod.rs b/src/library/mod.rs index ae3b07b..f0eca91 100644 --- a/src/library/mod.rs +++ b/src/library/mod.rs @@ -601,6 +601,24 @@ impl Library { Ok(tracks) } + pub fn track_by_content_id(&self, content_id: &str) -> Result> { + let Some(content_id) = music_dht::normalize_content_id(content_id) else { + return Ok(None); + }; + let conn = self.lock(); + let mut tracks = query_tracks( + &conn, + &format!( + "SELECT {TRACK_COLUMNS} FROM tracks t + JOIN releases r ON r.id = t.release_id + WHERE t.content_id = ?1 + LIMIT 1" + ), + params![content_id], + )?; + Ok(tracks.pop()) + } + // ----------------------------------------------------------------- // Playlists & likes // ----------------------------------------------------------------- diff --git a/src/ui/federation.rs b/src/ui/federation.rs index 943256f..8a9077e 100644 --- a/src/ui/federation.rs +++ b/src/ui/federation.rs @@ -6,7 +6,7 @@ use ratatui::text::{Line, Span}; use ratatui::widgets::{Block, Paragraph}; use super::theme; -use crate::app::state::{AppState, FedRow, settings_rows}; +use crate::app::state::{AppState, DevicePresenceSection, FedRow, settings_rows}; pub fn draw(frame: &mut Frame, area: Rect, state: &AppState) { let block = Block::bordered() @@ -16,7 +16,8 @@ pub fn draw(frame: &mut Frame, area: Rect, state: &AppState) { let inner = block.inner(area); frame.render_widget(block, area); - let rows_height = (settings_rows(state).len() + 5) as u16; + let rows_height = + (settings_rows(state).len() + 5 + device_presence_sections(state).len()) as u16; let [rows_area, _, status_area] = Layout::vertical([ Constraint::Length(rows_height.min(inner.height)), Constraint::Length(1), @@ -28,6 +29,24 @@ pub fn draw(frame: &mut Frame, area: Rect, state: &AppState) { draw_status(frame, status_area, state); } +fn device_presence_sections(state: &AppState) -> Vec { + let Some(status) = &state.federation.devices else { + return Vec::new(); + }; + let now = crate::app::state::unix_time_ms(); + let mut sections = Vec::new(); + for index in crate::app::state::device_status_order(state) { + let Some(device) = status.devices.get(index) else { + continue; + }; + let section = crate::app::state::device_presence_section(state, device, now); + if sections.last().copied() != Some(section) { + sections.push(section); + } + } + sections +} + fn draw_settings_rows(frame: &mut Frame, area: Rect, state: &AppState) { let settings = &state.federation.settings; let on_off = |on: bool| if on { "on" } else { "off" }; @@ -134,13 +153,24 @@ fn draw_settings_rows(frame: &mut Frame, area: Rect, state: &AppState) { ); cursor += 1; if let Some(status) = devices { - for device in &status.devices { + let now = crate::app::state::unix_time_ms(); + let mut current_section = None; + for index in crate::app::state::device_status_order(state) { + let Some(device) = status.devices.get(index) else { + continue; + }; + let section = crate::app::state::device_presence_section(state, device, now); + if current_section != Some(section) { + draw_subsection(frame, area, &mut y, section.title()); + current_section = Some(section); + } + let name = crate::app::state::device_display_name(device); let label = if device.is_self { - format!("* {}", device.name) + format!("* {name}") } else if device.revoked { - format!(" {} (revoked)", device.name) + format!(" {name} (revoked)") } else { - format!(" {}", device.name) + format!(" {name}") }; let version = if device.client_version.is_empty() { "unknown".to_string() @@ -148,10 +178,15 @@ fn draw_settings_rows(frame: &mut Frame, area: Rect, state: &AppState) { format!("v{}", device.client_version) }; let can_revoke = connected_devices_enabled && !device.is_self && !device.revoked; + let presence = match section { + DevicePresenceSection::Online => "online", + DevicePresenceSection::Offline => "offline", + DevicePresenceSection::Revoked => "revoked", + }; let value = if can_revoke { - format!("{version} · revoke ↵") + format!("{version} · {presence} · revoke ↵") } else if connected_devices_enabled { - version + format!("{version} · {presence}") } else { disabled_value.clone() }; @@ -252,6 +287,26 @@ fn draw_section(frame: &mut Frame, area: Rect, y: &mut u16, title: &'static str) *y = (*y).saturating_add(1); } +fn draw_subsection(frame: &mut Frame, area: Rect, y: &mut u16, title: &'static str) { + if *y >= area.y + area.height { + return; + } + let rect = Rect { + x: area.x, + y: *y, + width: area.width, + height: 1, + }; + frame.render_widget( + Paragraph::new(Line::from(vec![ + Span::raw(" "), + Span::styled(title, theme::dim()), + ])), + rect, + ); + *y = (*y).saturating_add(1); +} + fn draw_row( frame: &mut Frame, area: Rect, diff --git a/src/ui/mod.rs b/src/ui/mod.rs index 1f8873b..329fa13 100644 --- a/src/ui/mod.rs +++ b/src/ui/mod.rs @@ -142,7 +142,7 @@ fn draw_queue(frame: &mut Frame, area: Rect, state: &AppState) { let player = &state.player; let block = Block::bordered() .title(format!( - " Queue — {} tracks · enter: play · d: remove · shift-v: select · shift-c: clear ", + " Queue — {} tracks · enter: play · d: remove · shift-v: select · :clear ", player.queue.len() )) .title_style(theme::header()) @@ -212,7 +212,8 @@ fn format_secs(secs: f64) -> String { /// Playback time, progress bar, queue position, volume and mode flags. /// Wider consoles get a longer bar and full flags; narrow ones drop pieces. -fn player_right_line(player: &crate::app::state::PlayerBar, width: u16) -> Line<'static> { +fn player_right_line(state: &AppState, width: u16) -> Line<'static> { + let player = &state.player; let mut spans: Vec> = Vec::new(); if let Some(track) = &player.current && player.playing @@ -268,6 +269,17 @@ fn player_right_line(player: &crate::app::state::PlayerBar, width: u16) -> Line< } else { spans.push(Span::styled(format!(" {}%", player.volume), theme::dim())); } + if width >= 70 { + spans.push(Span::raw(" ")); + spans.push(Span::styled( + format!( + "{} · online {}", + state.device_playback.role.label(), + state.device_playback.online_devices.max(1) + ), + theme::dim(), + )); + } // Keep a gap between the flags and the username block to the right. spans.push(Span::raw(" ")); Line::from(spans) @@ -281,7 +293,7 @@ fn draw_status(frame: &mut Frame, area: Rect, state: &AppState) { // Layout: track title left, time/progress/flags on the right. The // right block is built first and gets a fixed width; the title // truncates into whatever is left. - let center = player_right_line(player, area.width); + let center = player_right_line(state, area.width); let center_width = (center.width() as u16).min(area.width); let [title_area, right_area] = Layout::horizontal([Constraint::Min(8), Constraint::Length(center_width)]) diff --git a/src/ui/popup.rs b/src/ui/popup.rs index d742971..ad71d9b 100644 --- a/src/ui/popup.rs +++ b/src/ui/popup.rs @@ -4,7 +4,9 @@ use ratatui::text::{Line, Span}; use ratatui::widgets::{Block, Clear, Paragraph, Wrap}; use super::theme; -use crate::app::state::{AppState, EditField, Loadable, Popup, addable_playlists}; +use crate::app::state::{ + AppState, DevicePresenceSection, EditField, Loadable, Popup, addable_playlists, +}; use crate::library::models::{ArtistRef, TrackItem}; pub fn draw(frame: &mut Frame, state: &AppState) { @@ -54,10 +56,124 @@ pub fn draw(frame: &mut Frame, state: &AppState) { Some(Popup::ConfirmDeviceRevoke { device_id, name }) => { draw_device_revoke(frame, device_id, name) } + Some(Popup::ConnectedDevices { cursor }) => draw_connected_devices(frame, state, *cursor), None => {} } } +fn draw_connected_devices(frame: &mut Frame, state: &AppState, cursor: usize) { + let rows = crate::app::popup::connected_device_rows(state); + enum DisplayLine { + Section(DevicePresenceSection), + Row(usize), + } + let mut display_lines = Vec::new(); + let mut last_section = None; + for (index, row) in rows.iter().enumerate() { + if last_section != Some(row.section) { + display_lines.push(DisplayLine::Section(row.section)); + last_section = Some(row.section); + } + display_lines.push(DisplayLine::Row(index)); + } + let height = + (display_lines.len() as u16 + 5).clamp(7, frame.area().height.saturating_sub(2).max(7)); + let area = centered(frame.area(), 76, height); + let block = Block::bordered() + .title(" Connected devices ") + .title_style(theme::header()) + .border_style(theme::accent()); + let inner = block.inner(area); + frame.render_widget(Clear, area); + frame.render_widget(block, area); + + let [summary_area, list_area, hint_area] = Layout::vertical([ + Constraint::Length(1), + Constraint::Min(1), + Constraint::Length(1), + ]) + .areas(inner); + let active = state.device_playback.active_label(); + frame.render_widget( + Paragraph::new(Line::from(vec![ + Span::styled("Role ", theme::dim()), + Span::styled(state.device_playback.role.label(), theme::accent()), + Span::raw(" "), + Span::styled("Active ", theme::dim()), + Span::raw(active), + ])), + summary_area, + ); + + let visible = usize::from(list_area.height.max(1)); + let selected = cursor.min(rows.len().saturating_sub(1)); + let selected_line = display_lines + .iter() + .position(|line| matches!(line, DisplayLine::Row(index) if *index == selected)) + .unwrap_or(0); + let first = selected_line + .saturating_sub(visible / 2) + .min(display_lines.len().saturating_sub(visible)); + for (line_index, line) in display_lines.iter().enumerate().skip(first).take(visible) { + let area = Rect { + x: list_area.x, + y: list_area.y + (line_index - first) as u16, + width: list_area.width, + height: 1, + }; + let DisplayLine::Row(index) = line else { + let DisplayLine::Section(section) = line else { + continue; + }; + frame.render_widget( + Paragraph::new(Line::styled(section.title(), theme::header())), + area, + ); + continue; + }; + let row = &rows[*index]; + let role = if row.revoked { + "revoked" + } else if row.active { + "active" + } else if row.is_self + && state.device_playback.role == crate::app::state::DevicePlaybackRole::Control + { + "control" + } else { + "device" + }; + let play = if row.playing && row.paused { + "paused" + } else if row.playing { + "playing" + } else { + "stopped" + }; + let online = if row.online { "online" } else { "offline" }; + let marker = if row.is_self { "*" } else { " " }; + let line = Line::from(vec![ + Span::styled(format!("{marker} "), theme::accent()), + Span::raw(row.name.clone()), + Span::styled(format!(" {role} · {online} · {play}"), theme::dim()), + Span::styled(format!(" · {} queued", row.queue_len), theme::dim()), + ]); + frame.render_widget(Paragraph::new(line), area); + if *index == selected { + frame.buffer_mut().set_style(area, theme::tab_active()); + } + } + + frame.render_widget( + Paragraph::new(Line::styled( + "enter: control selected / move active here · esc close", + theme::dim(), + )) + .alignment(Alignment::Center), + hint_area, + ); +} + fn draw_library_filters(frame: &mut Frame, state: &AppState, cursor: usize) { let area = centered(frame.area(), 46, 6); let block = Block::bordered()