diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 6d835ac..d685566 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -89,6 +89,20 @@ joining another user's trusted-device group. Keeping these layers separate prevents discovery convenience from silently becoming a synchronization trust decision. +### Federation Jam control + +Jam is a third, deliberately narrow authority boundary. A host creates an +opaque `frid://j/...` runtime capability and remains the only node producing +audio. Other TUI peers use a dedicated Jam ALPN to submit the same portable +playback commands used by connected-device control and receive the host's +playback snapshot. They receive queue metadata, not audio. + +Jam never exchanges trusted membership, likes, playlists, or listening +history. Volume remains local. Commands carry unique IDs and are retried until +the host acknowledges them, while inactive participants expire from the +runtime session. Regenerating the capability or restarting the host invalidates +the previous link. + ## Discovery and direct communication Furumi separates finding content from transferring it. diff --git a/Cargo.toml b/Cargo.toml index 26dfc40..c3ee4b6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "furumi_tui" -version = "0.2.1" +version = "0.2.2" edition = "2024" rust-version = "1.97" description = "A federated P2P player for personal music libraries" diff --git a/src/app/event.rs b/src/app/event.rs index 71be605..ef8f023 100644 --- a/src/app/event.rs +++ b/src/app/event.rs @@ -166,4 +166,14 @@ pub enum AppEvent { DevicePlayback(crate::devices::PlaybackSnapshot), /// Playback command addressed to this device. PlaybackCommand(crate::devices::PlaybackCommand), + /// Current lifecycle/status of the federation Jam. + JamStatus(crate::jam::JamStatus), + /// Host playback snapshot received by a Jam participant. + JamPlayback(crate::devices::PlaybackSnapshot), + /// Participant command accepted by this Jam host. + JamCommand(crate::devices::PlaybackCommand), + /// Result of creating/regenerating a Jam capability. + JamInvite(Result), + /// Result of joining a Jam capability. + JamJoined(Result), } diff --git a/src/app/mod.rs b/src/app/mod.rs index cf17749..a10eca2 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -38,6 +38,7 @@ pub struct Runtime { pub event_tx: mpsc::UnboundedSender, pub library: Arc, pub devices: Arc, + pub jam: Arc, pub federation: Arc, /// When the last Federation-tab status snapshot was requested. pub fed_status_at: Option, @@ -249,7 +250,12 @@ pub async fn run( let devices = crate::devices::DeviceSync::new(Arc::clone(&library))?; devices.set_event_tx(event_tx.clone()); - let federation = crate::federation::Federation::new(Arc::clone(&library), Arc::clone(&devices)); + let jam = crate::jam::JamManager::new(event_tx.clone()); + let federation = crate::federation::Federation::new( + Arc::clone(&library), + Arc::clone(&devices), + Arc::clone(&jam), + ); state.federation.settings = federation.settings(); state.federation.devices = Some(devices.status()); if let Ok((device_id, device_name)) = devices.identity_summary() { @@ -263,6 +269,7 @@ pub async fn run( event_tx, library, devices, + jam, federation, fed_status_at: None, library_network_refresh_at: None, @@ -360,7 +367,7 @@ fn sync_player_shared(state: &mut AppState, runtime: &Runtime) { } else { runtime.player.shared.audio_analysis() }; - if state.device_playback.role == state::DevicePlaybackRole::Active { + if state.device_playback.is_audio_owner() { publish_playback_snapshot(state, runtime); } } @@ -474,7 +481,7 @@ fn apply_playback_state_to_ui( } fn publish_playback_snapshot(state: &mut AppState, runtime: &Runtime) { - if state.device_playback.role != state::DevicePlaybackRole::Active { + if !state.device_playback.is_audio_owner() { return; } publish_playback_snapshot_with_active(state, runtime, true); @@ -503,6 +510,19 @@ fn publish_playback_snapshot_with_active(state: &mut AppState, runtime: &Runtime state: playback_state_from_ui(state), }; runtime.devices.publish_playback(snapshot); + if state.device_playback.role == state::DevicePlaybackRole::Jam + && state.device_playback.jam_host + { + runtime + .jam + .publish_host_playback(crate::devices::PlaybackSnapshot { + device_id: state.device_playback.self_device_id.clone(), + device_name: state.device_playback.self_device_name.clone(), + active: true, + updated_at_ms: unix_time_ms(), + state: playback_state_from_ui(state), + }); + } } fn update_local_idle_since(state: &mut AppState) { @@ -532,7 +552,7 @@ fn active_idle_lease_expired(snapshot: &crate::devices::PlaybackSnapshot, now: i } fn local_active_lease_protected(state: &mut AppState, now: i64) -> bool { - if state.device_playback.role != state::DevicePlaybackRole::Active || !state.player.playing { + if !state.device_playback.is_audio_owner() || !state.player.playing { return false; } if !state.player.paused { @@ -576,7 +596,7 @@ pub(crate) fn become_control_device( runtime: &Runtime, snapshot: crate::devices::PlaybackSnapshot, ) { - if state.device_playback.role == state::DevicePlaybackRole::Active { + if state.device_playback.is_audio_owner() { runtime.player.stop(); publish_inactive_playback_snapshot(state, runtime); } @@ -596,6 +616,7 @@ pub(crate) fn become_control_device( 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; + state.device_playback.jam_host = false; let Ok((device_id, device_name)) = runtime.devices.identity_summary() else { return; }; @@ -617,7 +638,7 @@ pub(crate) fn become_active_device(state: &mut AppState, runtime: &mut Runtime, } pub(crate) fn transfer_active_to_this_device(state: &mut AppState, runtime: &mut Runtime) { - if state.device_playback.role == state::DevicePlaybackRole::Active { + if state.device_playback.is_audio_owner() { publish_playback_snapshot(state, runtime); request_urgent_device_sync(runtime); return; @@ -722,6 +743,12 @@ fn record_control_playback_state(state: &mut AppState, runtime: &Runtime, seek: state: playback_state_from_ui(state), seek, }; + if state.device_playback.role == state::DevicePlaybackRole::Jam { + if let Err(err) = runtime.jam.submit_command(command) { + state.status_message = Some(format!("Jam command failed: {err:#}")); + } + return; + } record_playback_command_async(runtime, target, command, "device command"); } @@ -1479,10 +1506,41 @@ fn perform_effect(state: &mut AppState, runtime: &mut Runtime, effect: Effect) { | Effect::DeviceSetName(_) | Effect::DeviceRevoke(_) | Effect::DeviceLeaveGroup + | Effect::JamCreate + | Effect::JamJoin(_) if !state.connected_devices_enabled() => { state.status_message = Some("enable federation before using connected devices".into()); } + Effect::JamCreate => { + let federation = Arc::clone(&runtime.federation); + let tx = runtime.event_tx.clone(); + tokio::spawn(async move { + let result = federation + .create_jam() + .await + .map_err(|err| format!("{err:#}")); + let _ = tx.send(AppEvent::JamInvite(result)); + }); + } + Effect::JamJoin(invite) => { + let result = runtime + .federation + .join_jam(&invite) + .map(|()| "joined Jam; waiting for host state".to_string()) + .map_err(|err| format!("{err:#}")); + let _ = runtime.event_tx.send(AppEvent::JamJoined(result)); + let _ = runtime + .event_tx + .send(AppEvent::JamStatus(runtime.jam.status())); + } + Effect::JamLeave => { + runtime.jam.leave(); + state.jam = runtime.jam.status(); + state.device_playback.role = state::DevicePlaybackRole::Active; + state.device_playback.jam_host = false; + state.status_message = Some("left Jam".into()); + } Effect::DeviceShowInvite => { let fed = Arc::clone(&runtime.federation); let devices = Arc::clone(&runtime.devices); @@ -1679,6 +1737,8 @@ fn is_controlled_playback_effect(effect: &Effect) -> bool { fn perform_control_playback_effect(state: &mut AppState, runtime: &mut Runtime, effect: Effect) { let mut seek = false; + let local_only_volume = state.device_playback.role == state::DevicePlaybackRole::Jam + && matches!(effect, Effect::SetVolume(_)); match effect { Effect::PlayCurrent => { state.player.current = state.player.queue.get(state.player.queue_pos).cloned(); @@ -1708,6 +1768,9 @@ fn perform_control_playback_effect(state: &mut AppState, runtime: &mut Runtime, | Effect::LoadListenHistory => {} _ => {} } + if local_only_volume { + return; + } record_control_playback_state(state, runtime, seek); } @@ -2753,7 +2816,7 @@ fn handle_device_playback_snapshot( let already_controls_this_device = state.device_playback.is_control() && state.device_playback.active_device_id.as_deref() == Some(snapshot.device_id.as_str()); if !lease_expired || already_controls_this_device { - let was_active = state.device_playback.role == state::DevicePlaybackRole::Active; + let was_active = state.device_playback.is_audio_owner(); let was_paused = state.player.playing && state.player.paused; become_control_device(state, runtime, snapshot.clone()); if was_active && was_paused { @@ -2829,7 +2892,7 @@ fn handle_playback_command( if active_device_id == state.device_playback.self_device_id { return; } - let was_active = state.device_playback.role == state::DevicePlaybackRole::Active; + let was_active = state.device_playback.is_audio_owner(); let snapshot = crate::devices::PlaybackSnapshot { device_id: active_device_id, device_name: active_device_name, @@ -2962,6 +3025,83 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent AppEvent::PlaybackCommand(command) => { handle_playback_command(state, runtime, command); } + AppEvent::JamStatus(status) => { + state.jam = status.clone(); + match status.role { + crate::jam::JamRole::Host => { + state.device_playback.role = state::DevicePlaybackRole::Jam; + state.device_playback.jam_host = true; + publish_playback_snapshot(state, runtime); + } + crate::jam::JamRole::Participant => { + if state.device_playback.is_audio_owner() { + runtime.player.stop(); + runtime.player_start_pending = false; + } + state.device_playback.role = state::DevicePlaybackRole::Jam; + state.device_playback.jam_host = false; + } + crate::jam::JamRole::None => { + if state.device_playback.role == state::DevicePlaybackRole::Jam { + state.device_playback.role = state::DevicePlaybackRole::Active; + state.device_playback.jam_host = false; + } + } + } + } + AppEvent::JamPlayback(snapshot) => { + let local_volume = state.player.volume; + become_control_device(state, runtime, snapshot); + state.device_playback.role = state::DevicePlaybackRole::Jam; + state.device_playback.jam_host = false; + state.player.volume = local_volume; + state.status_message = Some(format!( + "Jam · controlling {}", + state.device_playback.active_label() + )); + } + AppEvent::JamCommand(command) => { + let command = match command { + crate::devices::PlaybackCommand::SetState { + state: mut wire, + seek, + } => { + wire.volume = state.player.volume; + crate::devices::PlaybackCommand::SetState { state: wire, seek } + } + crate::devices::PlaybackCommand::ActiveChanged { .. } => { + state.status_message = + Some("Jam cannot transfer audio away from its host".into()); + return; + } + }; + handle_playback_command(state, runtime, command); + state.device_playback.role = state::DevicePlaybackRole::Jam; + state.device_playback.jam_host = true; + publish_playback_snapshot(state, runtime); + } + AppEvent::JamInvite(result) => match result { + Ok(invite) => { + if !state.device_playback.is_audio_owner() { + transfer_active_to_this_device(state, runtime); + } + state.jam = runtime.jam.status(); + state.device_playback.role = state::DevicePlaybackRole::Jam; + state.device_playback.jam_host = true; + publish_playback_snapshot(state, runtime); + state.popup = Some(state::Popup::FedCopyText { + title: "Jam invite".to_string(), + text: invite, + help: "Copied capability lets federation peers control this host player until restart or regeneration.".to_string(), + }); + state.status_message = Some("Jam started".into()); + } + Err(error) => state.status_message = Some(format!("Jam: {error}")), + }, + AppEvent::JamJoined(result) => match result { + Ok(message) => state.status_message = Some(message), + Err(error) => state.status_message = Some(format!("Jam: {error}")), + }, AppEvent::FedSearchLoaded { seq, result } => { if runtime.search_seq.load(std::sync::atomic::Ordering::SeqCst) != seq { return; diff --git a/src/app/popup.rs b/src/app/popup.rs index 22c1600..f20683b 100644 --- a/src/app/popup.rs +++ b/src/app/popup.rs @@ -82,7 +82,7 @@ pub(crate) fn connected_device_rows(state: &AppState) -> Vec {} + KeyCode::Char('h') => { + super::perform_effect(state, runtime, crate::app::update::Effect::JamCreate); + } + KeyCode::Char('J') => { + state.popup = Some(Popup::FedInput { + field: FedInputField::JamInvite, + input: crate::app::input::LineEdit::default(), + }); + } + KeyCode::Char('l') if state.jam.role != crate::jam::JamRole::None => { + super::perform_effect(state, runtime, crate::app::update::Effect::JamLeave); + } + KeyCode::Char('c') => { + if let Some(invite) = state.jam.invite.as_deref() { + match copy_to_clipboard(invite) { + Ok(()) => state.status_message = Some("Jam invite copied".into()), + Err(error) => { + state.status_message = Some(format!("clipboard: {error}")); + state.popup = Some(Popup::ConnectedDevices { cursor }); + } + } + } else { + state.status_message = Some("only the Jam host has an invite".into()); + state.popup = Some(Popup::ConnectedDevices { cursor }); + } + } KeyCode::Up | KeyCode::Char('k') => { state.popup = Some(Popup::ConnectedDevices { cursor: cursor.saturating_sub(1), @@ -499,6 +525,17 @@ fn handle_fed_input( ); } } + FedInputField::JamInvite => { + if value.is_empty() { + state.status_message = Some("Jam invite is empty".into()); + } else { + super::perform_effect( + state, + runtime, + crate::app::update::Effect::JamJoin(value), + ); + } + } } } _ => { diff --git a/src/app/state.rs b/src/app/state.rs index 35080f1..134561e 100644 --- a/src/app/state.rs +++ b/src/app/state.rs @@ -805,6 +805,7 @@ pub enum FedInputField { ConnectTicket, DeviceName, ConnectInvite, + JamInvite, } impl FedInputField { @@ -814,6 +815,7 @@ impl FedInputField { FedInputField::ConnectTicket => "Connect to peer (paste ticket)", FedInputField::DeviceName => "Device name", FedInputField::ConnectInvite => "Connect device (paste frid://i invite)", + FedInputField::JamInvite => "Join Jam (paste frid://j invite)", } } @@ -831,6 +833,9 @@ impl FedInputField { FedInputField::ConnectInvite => { "Paste a frid:// invite generated by another client to add this device to its sync group." } + FedInputField::JamInvite => { + "Paste a frid://j capability to control the host player for this Jam only." + } } } } @@ -1210,6 +1215,7 @@ pub enum DevicePlaybackRole { #[default] Active, Control, + Jam, } impl DevicePlaybackRole { @@ -1217,6 +1223,7 @@ impl DevicePlaybackRole { match self { DevicePlaybackRole::Active => "active", DevicePlaybackRole::Control => "control", + DevicePlaybackRole::Jam => "jam", } } } @@ -1232,11 +1239,18 @@ pub struct DevicePlaybackState { pub local_idle_since_ms: Option, pub remote: BTreeMap, pub last_remote_snapshot: Option, + pub jam_host: bool, } impl DevicePlaybackState { pub fn is_control(&self) -> bool { self.role == DevicePlaybackRole::Control + || (self.role == DevicePlaybackRole::Jam && !self.jam_host) + } + + pub fn is_audio_owner(&self) -> bool { + self.role == DevicePlaybackRole::Active + || (self.role == DevicePlaybackRole::Jam && self.jam_host) } pub fn active_label(&self) -> String { @@ -1264,6 +1278,7 @@ pub struct AppState { pub settings_cursor: usize, pub player: PlayerBar, pub device_playback: DevicePlaybackState, + pub jam: crate::jam::JamStatus, 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 9150788..1c341d6 100644 --- a/src/app/update.rs +++ b/src/app/update.rs @@ -68,6 +68,12 @@ pub enum Effect { DeviceRevoke(String), /// Leave the current personal-device group after publishing self-revoke. DeviceLeaveGroup, + /// Create or regenerate the runtime Jam capability. + JamCreate, + /// Join a federation Jam by capability. + JamJoin(String), + /// Leave the current Jam without changing personal-device state. + JamLeave, /// Assemble the federated artist card (fan-out to the owning peers). FedOpenArtist(String), /// Download federated tracks into the local library, one by one. diff --git a/src/federation/mod.rs b/src/federation/mod.rs index e86fbb1..56f95a5 100644 --- a/src/federation/mod.rs +++ b/src/federation/mod.rs @@ -416,6 +416,7 @@ struct Running { pub struct Federation { library: Arc, devices: Arc, + jam: Arc, data_dir: PathBuf, cache_dir: PathBuf, media_dir: PathBuf, @@ -515,7 +516,11 @@ async fn dht_record_payload_bytes(data_dir: PathBuf, now_ms: u64) -> Result } impl Federation { - pub fn new(library: Arc, devices: Arc) -> Arc { + pub fn new( + library: Arc, + devices: Arc, + jam: Arc, + ) -> Arc { let dirs = crate::config::project_dirs(); let data_dir = dirs .as_ref() @@ -539,6 +544,7 @@ impl Federation { Arc::new(Self { library, devices, + jam, data_dir, cache_dir, media_dir, @@ -555,6 +561,27 @@ impl Federation { lock(&self.settings).clone() } + pub async fn create_jam(&self) -> Result { + let service = { + let running = self.running.lock().await; + Arc::clone( + &running + .as_ref() + .context("enable federation before creating a Jam")? + .service, + ) + }; + let (device_id, device_name) = self.devices.identity_summary()?; + self.jam + .create_or_regenerate(service, device_id, device_name) + .await + } + + pub fn join_jam(&self, invite: &str) -> Result<()> { + let (device_id, device_name) = self.devices.identity_summary()?; + self.jam.join(invite, device_id, device_name) + } + fn cached_metadata_snapshot(&self) -> Vec { lock(&self.metadata_cache).values().cloned().collect() } @@ -627,6 +654,8 @@ impl Federation { .stream_protocol(CATALOG_ALPN) // Personal-device sync (likes, playlists, trusted devices). .stream_protocol(crate::devices::SYNC_ALPN) + // Capability-scoped shared playback control. + .stream_protocol(crate::jam::JAM_ALPN) .build() .map_err(|err| anyhow::anyhow!("invalid federation config: {err}"))?; let (service, mut events) = MusicDhtService::start(config) @@ -690,6 +719,15 @@ impl Federation { let device_tick_task = tokio::spawn(async move { crate::devices::sync_loop(device_sync, device_service, device_transport).await; }); + let jam_acceptor = service + .stream_acceptor(crate::jam::JAM_ALPN) + .map_err(|err| anyhow::anyhow!("failed to take the Jam acceptor: {err}"))?; + let jam_serve_task = + tokio::spawn(crate::jam::serve_peers(jam_acceptor, Arc::clone(&self.jam))); + let jam_poll_task = tokio::spawn(crate::jam::poll_loop( + Arc::clone(&self.jam), + Arc::clone(&service), + )); *guard = Some(Running { service, @@ -702,6 +740,8 @@ impl Federation { catalog_task, device_sync_task, device_tick_task, + jam_serve_task, + jam_poll_task, ], }); self.set_error(None); diff --git a/src/jam.rs b/src/jam.rs new file mode 100644 index 0000000..495f865 --- /dev/null +++ b/src/jam.rs @@ -0,0 +1,732 @@ +//! Capability-based Jam playback control for independent federation peers. + +use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use anyhow::{Context, Result}; +use music_dht::{ByteStream, MusicDhtService, PeerTicket, StreamAcceptor}; +use serde::{Deserialize, Serialize}; +use tokio::sync::mpsc; + +use crate::app::event::AppEvent; +use crate::devices::{PlaybackCommand, PlaybackSnapshot}; + +pub const JAM_ALPN: &[u8] = b"furumi/jam/1"; +const PROTOCOL_VERSION: u16 = 1; +const MAX_LINE: usize = 8 * 1024 * 1024; +const MAX_COMMANDS: usize = 128; +const PARTICIPANT_TTL_MS: i64 = 30 * 60 * 1_000; +const POLL_INTERVAL: Duration = Duration::from_millis(500); + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum JamRole { + None, + Host, + Participant, +} + +#[derive(Debug, Clone)] +pub struct JamStatus { + pub role: JamRole, + pub jam_id: Option, + pub host_name: Option, + pub invite: Option, + pub participants: Vec, + pub connected: bool, + pub last_error: Option, +} + +impl Default for JamStatus { + fn default() -> Self { + Self { + role: JamRole::None, + jam_id: None, + host_name: None, + invite: None, + participants: Vec::new(), + connected: false, + last_error: None, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct JamParticipant { + pub participant_id: String, + pub name: String, + pub last_seen_ms: i64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct JamInvite { + v: u16, + #[serde(rename = "t")] + ticket: String, + #[serde(rename = "j")] + jam_id: String, + #[serde(rename = "s")] + secret: String, + #[serde(rename = "d")] + host_device_id: String, + #[serde(rename = "n")] + host_name: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct JamCommand { + command_id: String, + participant_id: String, + command: PlaybackCommand, + sent_at_ms: i64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum WireMessage { + Poll { + version: u16, + jam_id: String, + secret: String, + participant: JamParticipant, + #[serde(default)] + commands: Vec, + }, + Snapshot { + accepted: bool, + #[serde(default)] + error: Option, + #[serde(default)] + acknowledged_command_ids: Vec, + #[serde(default)] + playback: Option, + #[serde(default)] + participants: Vec, + host_time_ms: i64, + }, + Leave { + version: u16, + jam_id: String, + secret: String, + participant_id: String, + }, +} + +#[derive(Default)] +struct HostState { + invite: Option, + invite_uri: Option, + playback: Option, + participants: HashMap, + seen_commands: HashSet, + seen_order: VecDeque, +} + +struct JoinedState { + invite: JamInvite, + participant: JamParticipant, + pending: VecDeque, + participants: Vec, + connected: bool, + last_error: Option, + last_activity_ms: i64, +} + +#[derive(Default)] +struct State { + host: HostState, + joined: Option, +} + +pub struct JamManager { + state: Mutex, + event_tx: mpsc::UnboundedSender, +} + +impl JamManager { + pub fn new(event_tx: mpsc::UnboundedSender) -> Arc { + Arc::new(Self { + state: Mutex::new(State::default()), + event_tx, + }) + } + + pub async fn create_or_regenerate( + &self, + service: Arc, + host_device_id: String, + host_name: String, + ) -> Result { + let invite = JamInvite { + v: PROTOCOL_VERSION, + ticket: service.ticket().await?.to_string(), + jam_id: format!("jam_{}", random_hex(12)), + secret: random_hex(32), + host_device_id, + host_name, + }; + let uri = encode_invite(&invite)?; + let mut state = lock(&self.state); + state.joined = None; + state.host = HostState { + invite: Some(invite), + invite_uri: Some(uri.clone()), + ..HostState::default() + }; + Ok(uri) + } + + pub fn join(&self, uri: &str, participant_id: String, name: String) -> Result<()> { + let invite = parse_invite(uri)?; + let mut state = lock(&self.state); + state.host = HostState::default(); + state.joined = Some(JoinedState { + invite, + participant: JamParticipant { + participant_id, + name, + last_seen_ms: now_ms(), + }, + pending: VecDeque::new(), + participants: Vec::new(), + connected: false, + last_error: None, + last_activity_ms: now_ms(), + }); + Ok(()) + } + + pub fn leave(&self) { + let mut state = lock(&self.state); + state.joined = None; + state.host = HostState::default(); + } + + pub fn status(&self) -> JamStatus { + let state = lock(&self.state); + if let Some(joined) = &state.joined { + return JamStatus { + role: JamRole::Participant, + jam_id: Some(joined.invite.jam_id.clone()), + host_name: Some(joined.invite.host_name.clone()), + invite: None, + participants: joined.participants.clone(), + connected: joined.connected, + last_error: joined.last_error.clone(), + }; + } + if let Some(invite) = &state.host.invite { + return JamStatus { + role: JamRole::Host, + jam_id: Some(invite.jam_id.clone()), + host_name: Some(invite.host_name.clone()), + invite: state.host.invite_uri.clone(), + participants: state.host.participants.values().cloned().collect(), + connected: true, + last_error: None, + }; + } + JamStatus::default() + } + + pub fn publish_host_playback(&self, snapshot: PlaybackSnapshot) { + let mut state = lock(&self.state); + if state.host.invite.is_some() { + state.host.playback = Some(snapshot); + } + } + + pub fn submit_command(&self, command: PlaybackCommand) -> Result<()> { + let mut state = lock(&self.state); + let joined = state + .joined + .as_mut() + .context("this player is not controlling a Jam")?; + if joined.pending.len() >= MAX_COMMANDS { + joined.pending.pop_front(); + } + joined.pending.push_back(JamCommand { + command_id: format!("cmd_{}", random_hex(16)), + participant_id: joined.participant.participant_id.clone(), + command, + sent_at_ms: now_ms(), + }); + joined.last_activity_ms = now_ms(); + Ok(()) + } + + async fn poll_once(&self, service: Arc) -> Result<()> { + let (invite, participant, commands) = { + let state = lock(&self.state); + let joined = state.joined.as_ref().context("not joined")?; + ( + joined.invite.clone(), + joined.participant.clone(), + joined.pending.iter().cloned().collect::>(), + ) + }; + let ticket: PeerTicket = invite.ticket.parse().context("invalid Jam host ticket")?; + let peer = service.connect(ticket).await?; + let mut stream = service.open_stream(peer, JAM_ALPN).await?; + write_message( + &mut stream, + &WireMessage::Poll { + version: PROTOCOL_VERSION, + jam_id: invite.jam_id.clone(), + secret: invite.secret.clone(), + participant, + commands, + }, + ) + .await?; + stream.send.finish()?; + let response = read_message(&mut stream).await?; + let WireMessage::Snapshot { + accepted, + error, + acknowledged_command_ids, + playback, + participants, + .. + } = response + else { + anyhow::bail!("unexpected Jam response"); + }; + anyhow::ensure!( + accepted, + "{}", + error.unwrap_or_else(|| "Jam refused".into()) + ); + { + let mut state = lock(&self.state); + let Some(joined) = state.joined.as_mut() else { + return Ok(()); + }; + if joined.invite.jam_id != invite.jam_id { + return Ok(()); + } + let acknowledged = acknowledged_command_ids.into_iter().collect::>(); + joined + .pending + .retain(|command| !acknowledged.contains(&command.command_id)); + joined.participants = participants; + joined.connected = true; + joined.last_error = None; + joined.participant.last_seen_ms = now_ms(); + if playback + .as_ref() + .is_some_and(|snapshot| snapshot.state.playing && !snapshot.state.paused) + { + joined.last_activity_ms = now_ms(); + } + } + if let Some(playback) = playback { + let _ = self.event_tx.send(AppEvent::JamPlayback(playback)); + } + let _ = self.event_tx.send(AppEvent::JamStatus(self.status())); + Ok(()) + } +} + +pub async fn serve_peers(mut acceptor: StreamAcceptor, manager: Arc) { + while let Some(stream) = acceptor.accept().await { + let manager = Arc::clone(&manager); + tokio::spawn(async move { + if let Err(err) = serve_one(stream, manager).await { + tracing::debug!("Jam stream failed: {err:#}"); + } + }); + } +} + +async fn serve_one(mut stream: ByteStream, manager: Arc) -> Result<()> { + match read_message(&mut stream).await? { + WireMessage::Poll { + version, + jam_id, + secret, + mut participant, + commands, + } => { + let (accepted, error, acknowledged, playback, participants) = { + let mut state = lock(&manager.state); + let valid = + version == PROTOCOL_VERSION + && state.host.invite.as_ref().is_some_and(|invite| { + invite.jam_id == jam_id && invite.secret == secret + }); + if !valid { + ( + false, + Some("invalid Jam capability".to_string()), + vec![], + None, + vec![], + ) + } else { + let now = now_ms(); + state.host.participants.retain(|_, row| { + now.saturating_sub(row.last_seen_ms) <= PARTICIPANT_TTL_MS + }); + participant.last_seen_ms = now; + state + .host + .participants + .insert(participant.participant_id.clone(), participant); + let mut acknowledged = Vec::new(); + for command in commands.into_iter().take(MAX_COMMANDS) { + acknowledged.push(command.command_id.clone()); + if state.host.seen_commands.insert(command.command_id.clone()) { + state.host.seen_order.push_back(command.command_id.clone()); + let _ = manager.event_tx.send(AppEvent::JamCommand(command.command)); + } + } + while state.host.seen_order.len() > 4096 { + if let Some(id) = state.host.seen_order.pop_front() { + state.host.seen_commands.remove(&id); + } + } + ( + true, + None, + acknowledged, + state.host.playback.clone(), + state.host.participants.values().cloned().collect(), + ) + } + }; + write_message( + &mut stream, + &WireMessage::Snapshot { + accepted, + error, + acknowledged_command_ids: acknowledged, + playback, + participants, + host_time_ms: now_ms(), + }, + ) + .await?; + stream.send.finish()?; + let _ = tokio::time::timeout(Duration::from_secs(2), stream.send.stopped()).await; + let _ = manager.event_tx.send(AppEvent::JamStatus(manager.status())); + } + WireMessage::Leave { + version, + jam_id, + secret, + participant_id, + } => { + let mut state = lock(&manager.state); + if version == PROTOCOL_VERSION + && state + .host + .invite + .as_ref() + .is_some_and(|invite| invite.jam_id == jam_id && invite.secret == secret) + { + state.host.participants.remove(&participant_id); + } + drop(state); + let _ = manager.event_tx.send(AppEvent::JamStatus(manager.status())); + } + WireMessage::Snapshot { .. } => anyhow::bail!("unexpected Jam snapshot"), + } + Ok(()) +} + +pub async fn poll_loop(manager: Arc, service: Arc) { + let mut interval = tokio::time::interval(POLL_INTERVAL); + loop { + interval.tick().await; + if manager.status().role != JamRole::Participant { + continue; + } + let expired = { + let state = lock(&manager.state); + state.joined.as_ref().is_some_and(|joined| { + now_ms().saturating_sub(joined.last_activity_ms) > PARTICIPANT_TTL_MS + }) + }; + if expired { + manager.leave(); + let _ = manager.event_tx.send(AppEvent::JamStatus(manager.status())); + let _ = manager.event_tx.send(AppEvent::StatusMessage( + "Jam ended after 30 minutes without playback or commands".into(), + )); + continue; + } + if let Err(err) = manager.poll_once(Arc::clone(&service)).await { + { + let mut state = lock(&manager.state); + if let Some(joined) = state.joined.as_mut() { + joined.connected = false; + joined.last_error = Some(format!("{err:#}")); + } + } + let _ = manager.event_tx.send(AppEvent::JamStatus(manager.status())); + } + } +} + +async fn write_message(stream: &mut ByteStream, message: &WireMessage) -> Result<()> { + let mut bytes = serde_json::to_vec(message)?; + anyhow::ensure!(bytes.len() <= MAX_LINE, "Jam message is too large"); + bytes.push(b'\n'); + stream.send.write_all(&bytes).await?; + Ok(()) +} + +async fn read_message(stream: &mut ByteStream) -> Result { + let mut bytes = Vec::new(); + let mut byte = [0_u8; 1]; + loop { + let read = stream.recv.read(&mut byte).await?; + if read.unwrap_or(0) == 0 || byte[0] == b'\n' { + break; + } + bytes.push(byte[0]); + anyhow::ensure!(bytes.len() <= MAX_LINE, "Jam message is too large"); + } + anyhow::ensure!(!bytes.is_empty(), "empty Jam message"); + Ok(serde_json::from_slice(&bytes)?) +} + +fn encode_invite(invite: &JamInvite) -> Result { + Ok(format!( + "frid://j/{}", + base64url_encode(&serde_json::to_vec(invite)?) + )) +} + +fn parse_invite(uri: &str) -> Result { + let encoded = uri + .trim() + .strip_prefix("frid://j/") + .context("expected frid://j invite")?; + let invite: JamInvite = serde_json::from_slice(&base64url_decode(encoded)?)?; + anyhow::ensure!( + invite.v == PROTOCOL_VERSION, + "unsupported Jam invite version" + ); + anyhow::ensure!( + !invite.ticket.is_empty() + && !invite.jam_id.is_empty() + && invite.secret.len() >= 16 + && !invite.host_device_id.is_empty(), + "incomplete Jam invite" + ); + Ok(invite) +} + +fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_millis() as i64) + .unwrap_or(0) +} + +fn random_hex(bytes: usize) -> String { + use std::sync::atomic::{AtomicU64, Ordering}; + static COUNTER: AtomicU64 = AtomicU64::new(1); + let seed = format!( + "{}:{}:{}", + now_ms(), + std::process::id(), + COUNTER.fetch_add(1, Ordering::Relaxed) + ); + let hash = blake3::hash(seed.as_bytes()).to_hex().to_string(); + hash[..(bytes * 2).min(hash.len())].to_string() +} + +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +fn base64url_encode(bytes: &[u8]) -> String { + const TABLE: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"; + let mut out = String::new(); + for chunk in bytes.chunks(3) { + let b0 = chunk[0]; + let b1 = chunk.get(1).copied().unwrap_or(0); + let b2 = chunk.get(2).copied().unwrap_or(0); + out.push(TABLE[(b0 >> 2) as usize] as char); + out.push(TABLE[(((b0 & 3) << 4) | (b1 >> 4)) as usize] as char); + if chunk.len() > 1 { + out.push(TABLE[(((b1 & 15) << 2) | (b2 >> 6)) as usize] as char); + } + if chunk.len() > 2 { + out.push(TABLE[(b2 & 63) as usize] as char); + } + } + out +} + +fn base64url_decode(value: &str) -> Result> { + fn decode(byte: u8) -> Option { + match byte { + b'A'..=b'Z' => Some(byte - b'A'), + b'a'..=b'z' => Some(byte - b'a' + 26), + b'0'..=b'9' => Some(byte - b'0' + 52), + b'-' => Some(62), + b'_' => Some(63), + _ => None, + } + } + let bytes = value.as_bytes(); + anyhow::ensure!(bytes.len() % 4 != 1, "invalid base64url Jam invite"); + let mut out = Vec::with_capacity(bytes.len() * 3 / 4); + let mut i = 0; + while i < bytes.len() { + let a = decode(bytes[i]).context("invalid base64url Jam invite")?; + let b = decode(*bytes.get(i + 1).context("truncated Jam invite")?) + .context("invalid base64url Jam invite")?; + let c = bytes.get(i + 2).and_then(|byte| decode(*byte)); + let d = bytes.get(i + 3).and_then(|byte| decode(*byte)); + out.push((a << 2) | (b >> 4)); + if let Some(c) = c { + out.push((b << 4) | (c >> 2)); + if let Some(d) = d { + out.push((c << 6) | d); + } + } + i += 4; + } + Ok(out) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn invite_round_trips_and_is_separate_from_pairing() { + let invite = JamInvite { + v: PROTOCOL_VERSION, + ticket: "ticket".into(), + jam_id: "jam_test".into(), + secret: "0123456789abcdef".into(), + host_device_id: "dev_host".into(), + host_name: "Host".into(), + }; + let uri = encode_invite(&invite).unwrap(); + assert!(uri.starts_with("frid://j/")); + assert_eq!(parse_invite(&uri).unwrap().jam_id, "jam_test"); + assert!(parse_invite("frid://i/abcd").is_err()); + } + + #[tokio::test] + async fn participant_receives_host_state_and_host_receives_command() { + let unique = random_hex(8); + let host_dir = std::env::temp_dir().join(format!("furumi-jam-host-{unique}")); + let guest_dir = std::env::temp_dir().join(format!("furumi-jam-guest-{unique}")); + std::fs::create_dir_all(&host_dir).unwrap(); + std::fs::create_dir_all(&guest_dir).unwrap(); + let network = music_dht::NetworkId::from_name(&format!("jam-test-{unique}")); + let host_config = music_dht::MusicDhtConfig::builder() + .data_dir(&host_dir) + .network_id(network) + .stream_protocol(JAM_ALPN) + .build() + .unwrap(); + let guest_config = music_dht::MusicDhtConfig::builder() + .data_dir(&guest_dir) + .network_id(network) + .stream_protocol(JAM_ALPN) + .build() + .unwrap(); + let (host_service, mut host_events) = music_dht::MusicDhtService::start(host_config) + .await + .unwrap(); + let (guest_service, mut guest_events) = music_dht::MusicDhtService::start(guest_config) + .await + .unwrap(); + let host_service = Arc::new(host_service); + let guest_service = Arc::new(guest_service); + let host_drain = tokio::spawn(async move { while host_events.recv().await.is_some() {} }); + let guest_drain = tokio::spawn(async move { while guest_events.recv().await.is_some() {} }); + + let (host_tx, mut host_rx) = mpsc::unbounded_channel(); + let (guest_tx, mut guest_rx) = mpsc::unbounded_channel(); + let host = JamManager::new(host_tx); + let guest = JamManager::new(guest_tx); + let invite = host + .create_or_regenerate(Arc::clone(&host_service), "dev_host".into(), "Host".into()) + .await + .unwrap(); + guest + .join(&invite, "dev_guest".into(), "Guest".into()) + .unwrap(); + + let playback_state = crate::devices::PlaybackStateWire { + queue: Vec::new(), + queue_pos: 0, + playing: false, + paused: false, + idle_since_ms: Some(now_ms()), + position_secs: 0.0, + volume: 73, + shuffle: false, + repeat: crate::devices::PlaybackRepeat::Off, + }; + host.publish_host_playback(PlaybackSnapshot { + device_id: "dev_host".into(), + device_name: "Host".into(), + active: true, + updated_at_ms: now_ms(), + state: playback_state.clone(), + }); + let acceptor = host_service.stream_acceptor(JAM_ALPN).unwrap(); + let server = tokio::spawn(serve_peers(acceptor, Arc::clone(&host))); + + guest.poll_once(Arc::clone(&guest_service)).await.unwrap(); + let playback = tokio::time::timeout(Duration::from_secs(3), async { + loop { + if let Some(AppEvent::JamPlayback(snapshot)) = guest_rx.recv().await { + break snapshot; + } + } + }) + .await + .unwrap(); + assert_eq!(playback.device_id, "dev_host"); + assert_eq!(playback.state.volume, 73); + + guest + .submit_command(PlaybackCommand::SetState { + state: crate::devices::PlaybackStateWire { + paused: true, + ..playback_state + }, + seek: false, + }) + .unwrap(); + guest.poll_once(Arc::clone(&guest_service)).await.unwrap(); + let command = tokio::time::timeout(Duration::from_secs(3), async { + loop { + if let Some(AppEvent::JamCommand(command)) = host_rx.recv().await { + break command; + } + } + }) + .await + .unwrap(); + assert!(matches!( + command, + PlaybackCommand::SetState { + state: crate::devices::PlaybackStateWire { paused: true, .. }, + seek: false + } + )); + + server.abort(); + host_drain.abort(); + guest_drain.abort(); + host_service.shutdown().await.unwrap(); + guest_service.shutdown().await.unwrap(); + let _ = std::fs::remove_dir_all(host_dir); + let _ = std::fs::remove_dir_all(guest_dir); + } +} diff --git a/src/main.rs b/src/main.rs index 0d7132b..f186197 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,6 +3,7 @@ mod art; mod config; mod devices; mod federation; +mod jam; mod library; mod media; mod player; diff --git a/src/ui/popup.rs b/src/ui/popup.rs index 812cca2..63e64e4 100644 --- a/src/ui/popup.rs +++ b/src/ui/popup.rs @@ -440,7 +440,7 @@ fn draw_connected_devices(frame: &mut Frame, state: &AppState, cursor: usize) { display_lines.push(DisplayLine::Row(index)); } let height = - (display_lines.len() as u16 + 9).clamp(11, frame.area().height.saturating_sub(2).max(11)); + (display_lines.len() as u16 + 14).clamp(16, frame.area().height.saturating_sub(2).max(16)); let area = centered(frame.area(), 76, height); let block = Block::bordered() .title(" Connected devices ") @@ -450,9 +450,10 @@ fn draw_connected_devices(frame: &mut Frame, state: &AppState, cursor: usize) { frame.render_widget(Clear, area); frame.render_widget(block, area); - let [summary_area, this_area, other_area, hint_area] = Layout::vertical([ + let [summary_area, this_area, jam_area, other_area, hint_area] = Layout::vertical([ Constraint::Length(1), Constraint::Length(3), + Constraint::Length(4), Constraint::Min(1), Constraint::Length(2), ]) @@ -479,12 +480,11 @@ fn draw_connected_devices(frame: &mut Frame, state: &AppState, cursor: usize) { width: this_area.width, height: 1, }; - let action_label = - if state.device_playback.role == crate::app::state::DevicePlaybackRole::Active { - "This device is active" - } else { - "Make this device active" - }; + let action_label = if state.device_playback.is_audio_owner() { + "This device is active" + } else { + "Make this device active" + }; let self_name = self_row .map(|row| row.name.as_str()) .unwrap_or(state.device_playback.self_device_name.as_str()); @@ -501,6 +501,60 @@ fn draw_connected_devices(frame: &mut Frame, state: &AppState, cursor: usize) { state, ); + render_subtitle(frame, jam_area, state, "Jam"); + let jam_status = match state.jam.role { + crate::jam::JamRole::None => "inactive · h host · J join".to_string(), + crate::jam::JamRole::Host => format!( + "HOST {} · {} participant(s) · c copy invite · h regenerate · l leave", + state + .jam + .jam_id + .as_deref() + .unwrap_or_default() + .chars() + .take(12) + .collect::(), + state.jam.participants.len() + ), + crate::jam::JamRole::Participant => format!( + "{} · {} · {} participant(s) · l leave", + if state.jam.connected { + "CONNECTED" + } else { + "RECONNECTING" + }, + state.jam.host_name.as_deref().unwrap_or("Jam host"), + state.jam.participants.len() + ), + }; + frame.render_widget( + Paragraph::new(Line::styled( + format!(" {jam_status}"), + if state.jam.role == crate::jam::JamRole::None { + theme::dim() + } else { + theme::accent_for(state) + }, + )), + Rect { + x: jam_area.x, + y: jam_area.y + 1, + width: jam_area.width, + height: 1, + }, + ); + if let Some(error) = state.jam.last_error.as_deref() { + frame.render_widget( + Paragraph::new(Line::styled(format!(" {error}"), theme::dim())), + Rect { + x: jam_area.x, + y: jam_area.y + 2, + width: jam_area.width, + height: 1, + }, + ); + } + render_subtitle(frame, other_area, state, "Other devices"); let list_area = Rect { x: other_area.x, @@ -560,7 +614,7 @@ fn draw_connected_devices(frame: &mut Frame, state: &AppState, cursor: usize) { ); frame.render_widget( Paragraph::new(Line::styled( - "enter: activate selected device / control active selected · esc close", + "enter device · h host Jam · J join · c copy · l leave · esc close", theme::dim(), )) .alignment(Alignment::Center), diff --git a/src/ui/theme.rs b/src/ui/theme.rs index c8a281b..f6c5165 100644 --- a/src/ui/theme.rs +++ b/src/ui/theme.rs @@ -4,6 +4,7 @@ use crate::app::state::{AppState, DevicePlaybackRole}; pub const ACCENT: Color = Color::Cyan; pub const CONTROL_ACCENT: Color = Color::Yellow; +pub const JAM_ACCENT: Color = Color::Magenta; pub const DIM: Color = Color::DarkGray; pub fn accent() -> Style { @@ -38,7 +39,12 @@ pub fn selection() -> Style { pub fn selection_for(state: &AppState) -> Style { if state.device_playback.is_control() { - Style::new().fg(Color::White).bg(Color::Rgb(92, 72, 0)) + let background = if state.device_playback.role == DevicePlaybackRole::Jam { + Color::Rgb(80, 24, 96) + } else { + Color::Rgb(92, 72, 0) + }; + Style::new().fg(Color::White).bg(background) } else { selection() } @@ -64,6 +70,7 @@ pub fn role_pill(role: DevicePlaybackRole) -> Style { let bg = match role { DevicePlaybackRole::Active => Color::Green, DevicePlaybackRole::Control => CONTROL_ACCENT, + DevicePlaybackRole::Jam => JAM_ACCENT, }; Style::new() .fg(Color::Black) @@ -72,7 +79,9 @@ pub fn role_pill(role: DevicePlaybackRole) -> Style { } fn accent_color_for(state: &AppState) -> Color { - if state.device_playback.is_control() { + if state.device_playback.role == DevicePlaybackRole::Jam { + JAM_ACCENT + } else if state.device_playback.is_control() { CONTROL_ACCENT } else { ACCENT