12 Commits
Author SHA1 Message Date
Ultradesu 34153cca9d Added jam 2026-07-28 18:58:38 +01:00
Ultradesu d4c219837b Added JAM feature 2026-07-28 18:41:08 +01:00
Ultradesu 1c721b68f5 Added JAM feature 2026-07-28 18:35:59 +01:00
Ultradesu 29466d70e8 Added --help 2026-07-28 11:18:14 +01:00
Ultradesu abf9ad782c Added --status to get simple status 2026-07-28 11:16:38 +01:00
Ultradesu 28189bae95 Added playback history window 2026-07-27 23:37:46 +01:00
Ultradesu a7f41ff205 Bump libs 2026-07-27 23:20:21 +01:00
Ultradesu 40c318a1d0 Update frid dependency for listening history 2026-07-27 23:19:45 +01:00
Ultradesu 0d6c89761f Sync listening history across devices 2026-07-27 23:15:57 +01:00
Ultradesu f55571ab34 Fixed status icon background 2026-07-27 17:38:14 +01:00
Ultradesu eb97c6469e Fixed liked track sorting 2026-07-27 17:16:39 +01:00
Ultradesu 7ebf07161e Fixed liked track sorting 2026-07-27 17:09:33 +01:00
25 changed files with 1879 additions and 111 deletions
+5 -1
View File
@@ -10,6 +10,7 @@ permissions:
env:
CARGO_TERM_COLOR: always
CARGO_NET_RETRY: 10
jobs:
build:
@@ -43,8 +44,11 @@ jobs:
- name: Show Rust version
run: rustc --version && cargo --version
- name: Fetch locked dependencies
run: cargo fetch --locked
- name: Build
run: cargo build --release --locked
run: cargo build --release --locked --offline
- name: Package
shell: bash
+14
View File
@@ -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.
Generated
+12 -13
View File
@@ -240,7 +240,7 @@ version = "3.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311"
dependencies = [
"event-listener 5.4.1",
"event-listener 5.4.2",
"event-listener-strategy",
"pin-project-lite",
]
@@ -1376,11 +1376,10 @@ dependencies = [
[[package]]
name = "event-listener"
version = "5.4.1"
version = "5.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e13b66accf52311f30a0db42147dadea9850cb48cd070028831ae5f5d4b856ab"
checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2"
dependencies = [
"concurrent-queue",
"parking",
"pin-project-lite",
]
@@ -1391,7 +1390,7 @@ version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93"
dependencies = [
"event-listener 5.4.1",
"event-listener 5.4.2",
"pin-project-lite",
]
@@ -1456,7 +1455,7 @@ dependencies = [
[[package]]
name = "federation-net"
version = "0.1.0"
source = "git+https://gt.hexor.cy/ab/frid.git#a9012351dcdbdf8dbaa1f5dd71e498b4bc678d99"
source = "git+https://gt.hexor.cy/ab/frid.git?rev=8de7d1292708fa0b225e5a4a9d5ab4f0676202d3#8de7d1292708fa0b225e5a4a9d5ab4f0676202d3"
dependencies = [
"blake3",
"data-encoding",
@@ -1567,7 +1566,7 @@ dependencies = [
[[package]]
name = "furumi_tui"
version = "0.1.6"
version = "0.2.2"
dependencies = [
"anyhow",
"blake3",
@@ -2989,8 +2988,8 @@ dependencies = [
[[package]]
name = "music-dht"
version = "0.1.0"
source = "git+https://gt.hexor.cy/ab/frid.git#a9012351dcdbdf8dbaa1f5dd71e498b4bc678d99"
version = "0.2.0"
source = "git+https://gt.hexor.cy/ab/frid.git?rev=8de7d1292708fa0b225e5a4a9d5ab4f0676202d3#8de7d1292708fa0b225e5a4a9d5ab4f0676202d3"
dependencies = [
"async-trait",
"blake3",
@@ -5370,9 +5369,9 @@ dependencies = [
[[package]]
name = "thin-vec"
version = "0.2.18"
version = "0.2.19"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b0f7e269b48f0a7dd0146680fa24b50cc67fc0373f086a5b2f99bd084639b482"
checksum = "79def32ffcd477db1ff26f76dab9e3a91f0bd42a85ca96577089b24623056f9d"
[[package]]
name = "thiserror"
@@ -5632,9 +5631,9 @@ dependencies = [
[[package]]
name = "toml_parser"
version = "1.1.2+spec-1.1.0"
version = "1.1.3+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526"
checksum = "1d38ac1cf9b95face32296c0a3ede1fdc270627c9d9c02a7274dd6d960dc4d56"
dependencies = [
"winnow 1.0.4",
]
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "furumi_tui"
version = "0.1.7"
version = "0.2.2"
edition = "2024"
rust-version = "1.97"
description = "A federated P2P player for personal music libraries"
@@ -21,7 +21,7 @@ image = { version = "0.25.10", default-features = false, features = ["jpeg", "pn
lofty = "0.22"
# P2P federation: library index in a shared DHT + audio streaming between
# peers (same protocol as furumi-fd).
music-dht = { git = "https://gt.hexor.cy/ab/frid.git" }
music-dht = { git = "https://gt.hexor.cy/ab/frid.git", rev = "8de7d1292708fa0b225e5a4a9d5ab4f0676202d3" }
ratatui = "0.30.1"
rhai = { version = "1", features = ["sync"] }
rodio = { version = "0.22.2", default-features = false, features = ["playback", "mp3", "flac", "vorbis", "wav", "symphonia-aac", "symphonia-isomp4", "symphonia-alac"] }
+23
View File
@@ -114,6 +114,29 @@ Import a music directory from Furumi's command line:
Federation, trusted-device pairing, and key bindings are configured directly
inside the player.
### Now playing in tmux
While Furumi is running, a second invocation can print a cheap, single-line
playback snapshot without opening the TUI or library:
```bash
furumi --status
# ▶ Artist — Track 1:23/4:05
```
For example, add this to `.tmux.conf`:
```tmux
set -g status-interval 1
set -g status-right '#(furumi --status) | %H:%M'
```
`furumi --status-json` returns the same snapshot as JSON, including playback
state, title, artist, album, position, duration, and volume. Both commands
print nothing when Furumi is stopped or no track is loaded. On Linux, Furumi
also exposes the existing MPRIS player `cy.hexor.furumi`, which can be queried
with tools such as `playerctl`.
## Architecture
Furumi is a Rust application built with:
+5 -1
View File
@@ -43,6 +43,7 @@ pub enum Action {
RemoveFromQueue,
ClearQueue,
OpenConnectedDevices,
OpenListenHistory,
GoToRelease,
AddToPlaylist,
NewPlaylist,
@@ -102,7 +103,8 @@ impl Action {
| Action::ToggleShuffle
| Action::CycleRepeat
| Action::ToggleVisualizer
| Action::OpenConnectedDevices => Category::Playback,
| Action::OpenConnectedDevices
| Action::OpenListenHistory => Category::Playback,
Action::QueueAddNext
| Action::QueueAddLast
| Action::DownloadSelected
@@ -152,6 +154,7 @@ impl Action {
Action::CycleRepeat => Some(":repeat [off|one|all]"),
Action::ClearQueue => Some(":clear"),
Action::OpenConnectedDevices => None,
Action::OpenListenHistory => None,
Action::ToggleHelp => Some(":help"),
Action::OpenSearch => Some("/text"),
_ => None,
@@ -194,6 +197,7 @@ impl Action {
Action::RemoveFromQueue => "Queue: remove selected".into(),
Action::ClearQueue => "Queue: clear".into(),
Action::OpenConnectedDevices => "Connected devices…".into(),
Action::OpenListenHistory => "Listening history…".into(),
Action::GoToRelease => "Open the track's release".into(),
Action::AddToPlaylist => "Add track to a playlist…".into(),
Action::NewPlaylist => "Create a playlist".into(),
+11
View File
@@ -56,6 +56,7 @@ pub enum AppEvent {
LocalContentIdsLoaded(Result<Vec<String>, String>),
/// Counts and storage footprint of the local library/database.
LocalLibraryStatsLoaded(Result<crate::library::LocalLibraryStats, String>),
ListenHistoryLoaded(Result<Vec<crate::library::ListenHistoryEntry>, String>),
/// One content id became available locally while the UI is open.
LocalContentAvailable {
content_id: String,
@@ -165,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<String, String>),
/// Result of joining a Jam capability.
JamJoined(Result<String, String>),
}
+227 -28
View File
@@ -38,6 +38,7 @@ pub struct Runtime {
pub event_tx: mpsc::UnboundedSender<AppEvent>,
pub library: Arc<Library>,
pub devices: Arc<crate::devices::DeviceSync>,
pub jam: Arc<crate::jam::JamManager>,
pub federation: Arc<crate::federation::Federation>,
/// When the last Federation-tab status snapshot was requested.
pub fed_status_at: Option<std::time::Instant>,
@@ -66,6 +67,7 @@ pub struct Runtime {
pub player_start_pending: bool,
pub media_tx: std::sync::mpsc::Sender<crate::media::MediaUpdate>,
pub last_media_push: Option<std::time::Instant>,
pub status_publisher: crate::status::Publisher,
}
#[derive(Debug, Clone, Copy)]
@@ -248,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() {
@@ -262,6 +269,7 @@ pub async fn run(
event_tx,
library,
devices,
jam,
federation,
fed_status_at: None,
library_network_refresh_at: None,
@@ -286,6 +294,7 @@ pub async fn run(
player_start_pending: false,
media_tx,
last_media_push: None,
status_publisher: crate::status::Publisher::spawn(),
};
spawn_content_id_backfill(&runtime);
@@ -325,6 +334,9 @@ pub async fn run(
state.advance_spinner();
expire_quit_confirmation(&mut state);
sync_player_shared(&mut state, &runtime);
runtime
.status_publisher
.publish(crate::status::PlaybackStatus::from_player(&state.player));
maybe_prefetch_next(&mut state, &runtime);
push_media_update(&state, &mut runtime, false);
}
@@ -355,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);
}
}
@@ -469,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);
@@ -498,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) {
@@ -527,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 {
@@ -571,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);
}
@@ -591,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;
};
@@ -612,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;
@@ -717,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");
}
@@ -1332,6 +1364,18 @@ fn perform_effect(state: &mut AppState, runtime: &mut Runtime, effect: Effect) {
}
});
}
Effect::LoadListenHistory => {
let library = Arc::clone(&runtime.library);
let devices = Arc::clone(&runtime.devices);
let tx = runtime.event_tx.clone();
tokio::task::spawn_blocking(move || {
let result = library
.listen_history(500)
.map_err(|err| format!("{err:#}"));
let _ = tx.send(AppEvent::ListenHistoryLoaded(result));
let _ = tx.send(AppEvent::DeviceSyncStatus(devices.status()));
});
}
Effect::ToggleLikes {
track_ids,
fed_tracks,
@@ -1462,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);
@@ -1662,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();
@@ -1685,9 +1762,15 @@ fn perform_control_playback_effect(state: &mut AppState, runtime: &mut Runtime,
state.player.volume = volume.min(100);
save_app_settings(state);
}
Effect::SetOptions | Effect::RemoveQueueIndices { .. } | Effect::PlaybackQueueChanged => {}
Effect::SetOptions
| Effect::RemoveQueueIndices { .. }
| Effect::PlaybackQueueChanged
| Effect::LoadListenHistory => {}
_ => {}
}
if local_only_volume {
return;
}
record_control_playback_state(state, runtime, seek);
}
@@ -1769,6 +1852,7 @@ fn start_current_audio(
state.player.position_secs = position_secs.max(0.0);
state.player.audio_analysis = player::AudioAnalysisSnapshot::default();
state.player.track_started_at = Some(now_epoch_seconds());
state.player.listen_id = Some(runtime.devices.new_listen_id());
state.player.prefetched_pos = None;
runtime.player_start_pending = true;
runtime.player.stop();
@@ -1796,16 +1880,19 @@ fn start_current_audio(
}
// The track that was playing until now was cut short by this switch.
let previous_started_at = state.player.track_started_at;
let previous_listen_id = state.player.listen_id.take();
let next_key = track_playback_key(&track);
let mut same_track = false;
let same_track_started_at = if let Some(previous) = state.player.current.take() {
let same_track = track_playback_key(&previous) == next_key;
same_track = track_playback_key(&previous) == next_key;
if state.player.playing && !same_track {
report_history(
runtime,
previous.id,
&previous,
previous_listen_id.as_deref(),
state.player.track_started_at,
state.player.position_secs.round() as i32,
false,
(state.player.position_secs * 1_000.0).round() as i64,
music_dht::device_sync::ListenEndReason::Replaced,
);
}
same_track.then_some(previous_started_at).flatten()
@@ -1818,6 +1905,11 @@ fn start_current_audio(
state.player.position_secs = position_secs.max(0.0);
state.player.audio_analysis = player::AudioAnalysisSnapshot::default();
state.player.track_started_at = same_track_started_at.or_else(|| Some(now_epoch_seconds()));
state.player.listen_id = if same_track {
previous_listen_id
} else {
Some(runtime.devices.new_listen_id())
};
state.player.prefetched_pos = None;
state.status_message = Some(format!("{}{}", track.title, track.artist_line()));
@@ -1972,18 +2064,28 @@ fn maybe_prefetch_next(state: &mut AppState, runtime: &Runtime) {
/// than 5s are noise.
fn report_history(
runtime: &Runtime,
track_id: i64,
track: &crate::library::models::TrackItem,
listen_id: Option<&str>,
started_at: Option<i64>,
listened: i32,
completed: bool,
listened_ms: i64,
ended_reason: music_dht::device_sync::ListenEndReason,
) {
// Ephemeral federated tracks are not library rows; no history for them.
if listened < 5 || track_id < 0 {
let Some(listen_id) = listen_id else {
return;
}
let library = Arc::clone(&runtime.library);
};
let Some(event) = runtime.devices.listen_event_for_track(
listen_id.to_string(),
track,
started_at.unwrap_or_else(now_epoch_seconds) * 1_000,
listened_ms,
ended_reason,
) else {
tracing::warn!(title = %track.title, "history skipped: track has no content id");
return;
};
let devices = Arc::clone(&runtime.devices);
tokio::task::spawn_blocking(move || {
if let Err(err) = library.add_history(track_id, started_at, listened, completed) {
if let Err(err) = devices.record_listen(event) {
tracing::warn!(%err, "history write failed");
}
});
@@ -2682,6 +2784,12 @@ fn handle_device_playback_snapshot(
runtime: &mut Runtime,
snapshot: crate::devices::PlaybackSnapshot,
) {
// Personal-device reconciliation must never change Jam ownership. Jam
// has its own authority and lifecycle even when the same TUI also belongs
// to a trusted-device group.
if state.device_playback.role == state::DevicePlaybackRole::Jam {
return;
}
if snapshot.device_id == state.device_playback.self_device_id {
return;
}
@@ -2711,10 +2819,10 @@ fn handle_device_playback_snapshot(
return;
}
let lease_expired = active_idle_lease_expired(&snapshot, now);
let already_controls_this_device = state.device_playback.is_control()
let already_controls_this_device = state.device_playback.is_personal_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 {
@@ -2726,7 +2834,7 @@ fn handle_device_playback_snapshot(
return;
}
if state.device_playback.is_control() {
if state.device_playback.is_personal_control() {
return;
}
become_active_device(state, runtime, false);
@@ -2790,7 +2898,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,
@@ -2812,6 +2920,12 @@ fn handle_playback_command(
fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent) {
match event {
AppEvent::StatusMessage(message) => state.status_message = Some(message),
AppEvent::ListenHistoryLoaded(result) => {
state.listen_history = Some(match result {
Ok(entries) => state::Loadable::Ready(entries),
Err(err) => state::Loadable::Failed(err),
});
}
AppEvent::FederationStatus(status) => {
state.federation.status = Some(status);
}
@@ -2834,7 +2948,7 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent
})
.unwrap_or(1)
.max(1);
let active_revoked = state.device_playback.is_control()
let active_revoked = state.device_playback.is_personal_control()
&& state
.device_playback
.active_device_id
@@ -2852,7 +2966,7 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent
})
.is_some_and(|device| device.revoked)
});
let active_missing = state.device_playback.is_control()
let active_missing = state.device_playback.is_personal_control()
&& state
.device_playback
.active_device_id
@@ -2914,9 +3028,91 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent
AppEvent::DevicePlayback(snapshot) => {
handle_device_playback_snapshot(state, runtime, snapshot);
}
AppEvent::PlaybackCommand(_)
if state.device_playback.role == state::DevicePlaybackRole::Jam =>
{
tracing::debug!("ignored personal-device playback command while Jam is active");
}
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;
@@ -3292,10 +3488,11 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent
if let Some(finished) = state.player.current.clone() {
report_history(
runtime,
finished.id,
&finished,
state.player.listen_id.as_deref(),
state.player.track_started_at,
finished.duration_seconds.round() as i32,
true,
(finished.duration_seconds * 1_000.0).round() as i64,
music_dht::device_sync::ListenEndReason::Finished,
);
}
if has_next {
@@ -3309,10 +3506,12 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent
state.player.current = state.player.queue.get(state.player.queue_pos).cloned();
state.player.position_secs = 0.0;
state.player.track_started_at = Some(now_epoch_seconds());
state.player.listen_id = Some(runtime.devices.new_listen_id());
push_media_metadata(state, runtime);
push_media_update(state, runtime, true);
} else {
state.player.current = None;
state.player.listen_id = None;
state.player.prefetched_pos = None;
if let Some(effect) = update::advance_after_finish(state) {
perform_effect(state, runtime, effect);
+63 -1
View File
@@ -82,7 +82,7 @@ pub(crate) fn connected_device_rows(state: &AppState) -> Vec<ConnectedDevicePopu
is_self: true,
online: true,
section: DevicePresenceSection::Online,
active: state.device_playback.role == crate::app::state::DevicePlaybackRole::Active,
active: state.device_playback.is_audio_owner(),
playing: state.player.playing,
paused: state.player.paused,
queue_len: state.player.queue.len(),
@@ -196,9 +196,28 @@ pub fn handle_key(state: &mut AppState, runtime: &mut Runtime, key: KeyEvent) {
Popup::ConnectedDevices { cursor } => {
handle_connected_devices(state, runtime, cursor, key);
}
Popup::ListenHistory { cursor } => handle_listen_history(state, cursor, key),
}
}
fn handle_listen_history(state: &mut AppState, cursor: usize, key: KeyEvent) {
let len = match state.listen_history.as_ref() {
Some(crate::app::state::Loadable::Ready(entries)) => entries.len(),
_ => 0,
};
let cursor = match key.code {
KeyCode::Esc | KeyCode::Char('q') => return,
KeyCode::Up | KeyCode::Char('k') => cursor.saturating_sub(1),
KeyCode::Down | KeyCode::Char('j') => (cursor + 1).min(len.saturating_sub(1)),
KeyCode::PageUp => cursor.saturating_sub(10),
KeyCode::PageDown => (cursor + 10).min(len.saturating_sub(1)),
KeyCode::Home | KeyCode::Char('g') => 0,
KeyCode::End | KeyCode::Char('G') => len.saturating_sub(1),
_ => cursor,
};
state.popup = Some(Popup::ListenHistory { cursor });
}
fn handle_federation_status_details(
state: &mut AppState,
mut parent: FederationStatusPopupState,
@@ -367,6 +386,32 @@ fn handle_connected_devices(
let cursor = cursor.min(last);
match key.code {
KeyCode::Esc | KeyCode::Char('q') => {}
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),
@@ -378,6 +423,12 @@ fn handle_connected_devices(
});
}
KeyCode::Enter => {
if state.device_playback.role == crate::app::state::DevicePlaybackRole::Jam {
state.status_message =
Some("leave the current Jam before switching personal devices".into());
state.popup = Some(Popup::ConnectedDevices { cursor });
return;
}
if cursor == 0 {
super::transfer_active_to_this_device(state, runtime);
state.status_message = Some("active playback moved to this device".into());
@@ -480,6 +531,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),
);
}
}
}
}
_ => {
+25
View File
@@ -732,6 +732,8 @@ pub enum Popup {
ConfirmDeviceLeave,
/// Connected playback devices and their current role/status.
ConnectedDevices { cursor: usize },
/// Qualified listening history from every trusted device.
ListenHistory { cursor: usize },
/// Full federation, transport and device status details.
FederationStatusDetails {
focus: StatusDetailFocus,
@@ -803,6 +805,7 @@ pub enum FedInputField {
ConnectTicket,
DeviceName,
ConnectInvite,
JamInvite,
}
impl FedInputField {
@@ -812,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)",
}
}
@@ -829,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."
}
}
}
}
@@ -1170,6 +1177,8 @@ pub struct PlayerBar {
pub audio_analysis: crate::player::AudioAnalysisSnapshot,
/// Epoch seconds when the current track started (for history reports).
pub track_started_at: Option<i64>,
/// Stable id reused for every report of the current playback session.
pub listen_id: Option<String>,
/// Queue index already enqueued in the audio thread for gapless play.
pub prefetched_pos: Option<usize>,
pub volume: u8,
@@ -1191,6 +1200,7 @@ impl Default for PlayerBar {
position_secs: 0.0,
audio_analysis: crate::player::AudioAnalysisSnapshot::default(),
track_started_at: None,
listen_id: None,
prefetched_pos: None,
original_order: None,
volume: 80,
@@ -1205,6 +1215,7 @@ pub enum DevicePlaybackRole {
#[default]
Active,
Control,
Jam,
}
impl DevicePlaybackRole {
@@ -1212,6 +1223,7 @@ impl DevicePlaybackRole {
match self {
DevicePlaybackRole::Active => "active",
DevicePlaybackRole::Control => "control",
DevicePlaybackRole::Jam => "jam",
}
}
}
@@ -1227,11 +1239,22 @@ pub struct DevicePlaybackState {
pub local_idle_since_ms: Option<i64>,
pub remote: BTreeMap<String, crate::devices::PlaybackSnapshot>,
pub last_remote_snapshot: Option<crate::devices::PlaybackSnapshot>,
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_personal_control(&self) -> bool {
self.role == DevicePlaybackRole::Control
}
pub fn is_audio_owner(&self) -> bool {
self.role == DevicePlaybackRole::Active
|| (self.role == DevicePlaybackRole::Jam && self.jam_host)
}
pub fn active_label(&self) -> String {
@@ -1259,6 +1282,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<i64, Loadable<ArtistDetail>>,
@@ -1277,6 +1301,7 @@ pub struct AppState {
pub likes_loaded: bool,
pub local_content_ids_loaded: bool,
pub local_library_stats: Option<Loadable<crate::library::LocalLibraryStats>>,
pub listen_history: Option<Loadable<Vec<crate::library::ListenHistoryEntry>>>,
pub logs: LogsTab,
pub queue_tab: QueueTab,
pub federation: FederationTab,
+14
View File
@@ -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.
@@ -82,6 +88,8 @@ pub enum Effect {
OpenVisualizerEditor {
path: std::path::PathBuf,
},
/// Load qualified listening history without blocking the UI thread.
LoadListenHistory,
}
pub fn update(state: &mut AppState, action: Action) -> Option<Effect> {
@@ -114,6 +122,11 @@ pub fn update(state: &mut AppState, action: Action) -> Option<Effect> {
state.popup = Some(super::state::Popup::ConnectedDevices { cursor: 0 });
None
}
Action::OpenListenHistory => {
state.popup = Some(super::state::Popup::ListenHistory { cursor: 0 });
state.listen_history = Some(Loadable::Loading);
Some(Effect::LoadListenHistory)
}
Action::NextTab => {
switch_tab(state, state.active_tab.next());
None
@@ -1092,6 +1105,7 @@ fn remove_queue_indices(state: &mut AppState, indices: &[usize]) -> QueueRemoval
state.player.current = state.player.queue.get(state.player.queue_pos).cloned();
state.player.position_secs = 0.0;
state.player.track_started_at = None;
state.player.listen_id = None;
state.queue_tab.cursor = state.queue_tab.cursor.min(state.player.queue.len() - 1);
return QueueRemovalOutcome {
restart_paused: was_loaded.then_some(was_paused),
+17
View File
@@ -16,6 +16,23 @@ fn with_artists(n: usize) -> AppState {
state
}
#[test]
fn listening_history_popup_requests_a_background_load() {
let mut state = AppState::default();
assert_eq!(
update(&mut state, Action::OpenListenHistory),
Some(Effect::LoadListenHistory)
);
assert!(matches!(
state.popup,
Some(crate::app::state::Popup::ListenHistory { cursor: 0 })
));
assert!(matches!(
state.listen_history,
Some(crate::app::state::Loadable::Loading)
));
}
fn test_track(id: i64) -> TrackItem {
TrackItem {
id,
+4
View File
@@ -197,6 +197,10 @@ command = "CycleRepeat"
key_sequence = "shift-l"
command = "ToggleVisualizer"
[[keymaps]]
key_sequence = "shift-h"
command = "OpenListenHistory"
[[keymaps]]
key_sequence = "x"
command = "ToggleLike"
+9
View File
@@ -596,4 +596,13 @@ mod tests {
KeyResolution::Action(Action::SeekForward { seconds: 10 })
);
}
#[test]
fn default_listening_history_key_resolves() {
let mut km = keymap_from(DEFAULT_KEYMAP);
assert_eq!(
km.resolve(key!(shift - h), KeyContext::Library),
KeyResolution::Action(Action::OpenListenHistory)
);
}
}
+60 -2
View File
@@ -12,6 +12,7 @@ use std::sync::Arc;
use std::time::Duration;
use anyhow::{Context as _, Result};
use music_dht::device_sync::{ListenEvent, ListenTrackMetadata};
use music_dht::{ByteStream, MusicDhtService, NetworkId, PeerTicket, SecretKey, StreamAcceptor};
use rusqlite::{Connection, OptionalExtension, params};
use serde::{Deserialize, Serialize};
@@ -21,9 +22,9 @@ use crate::app::event::AppEvent;
use crate::library::Library;
use crate::library::models::{ArtistRef, TrackItem};
pub const SYNC_ALPN: &[u8] = b"furumi/sync/1";
pub const SYNC_ALPN: &[u8] = b"furumi/sync/2";
const CLIENT_VERSION: &str = env!("CARGO_PKG_VERSION");
const PROTOCOL_VERSION: u16 = 1;
const PROTOCOL_VERSION: u16 = 2;
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);
@@ -366,6 +367,9 @@ pub enum SyncOpPayload {
target_device_id: String,
command: PlaybackCommand,
},
ListenRecorded {
event: ListenEvent,
},
}
impl SyncOpPayload {
@@ -637,6 +641,10 @@ impl DeviceSync {
Ok((identity.device_id, identity.name))
}
pub fn new_listen_id(&self) -> String {
format!("{}-{}", now_ms(), random_hex(12))
}
pub fn publish_playback(&self, mut snapshot: PlaybackSnapshot) {
if snapshot.updated_at_ms <= 0 {
snapshot.updated_at_ms = now_ms();
@@ -1073,6 +1081,52 @@ impl DeviceSync {
Ok(())
}
pub fn record_listen(&self, event: ListenEvent) -> Result<()> {
if !event.should_record() {
return Ok(());
}
self.record_local_op(SyncOpPayload::ListenRecorded { event })
}
pub fn listen_event_for_track(
&self,
listen_id: String,
track: &TrackItem,
started_at_ms: i64,
listened_ms: i64,
ended_reason: music_dht::device_sync::ListenEndReason,
) -> Option<ListenEvent> {
let content_id = track
.content_id
.as_deref()
.or_else(|| track.fed.as_ref()?.content_id.as_deref())
.and_then(music_dht::normalize_content_id)?;
Some(ListenEvent {
listen_id,
content_id,
started_at_ms,
listened_ms,
track_duration_ms: (track.duration_seconds > 0.0)
.then_some((track.duration_seconds * 1_000.0).round() as i64),
ended_reason,
track: ListenTrackMetadata {
title: track.title.clone(),
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_title: (!track.release_title.trim().is_empty())
.then(|| track.release_title.clone()),
},
})
}
pub fn record_playlist_created(&self, playlist_id: i64, title: &str) -> Result<()> {
let playlist_id = self.library.ensure_playlist_sync_id(playlist_id)?;
self.record_local_op(SyncOpPayload::PlaylistCreated {
@@ -1584,6 +1638,9 @@ impl DeviceSync {
self.apply_playback_command(target_device_id, command, &op.op_id)?;
false
}
SyncOpPayload::ListenRecorded { event } => self
.library
.apply_listen_event(event, &op.origin_device_id)?,
};
Ok(changed)
}
@@ -3229,6 +3286,7 @@ fn payload_kind(payload: &SyncOpPayload) -> &'static str {
SyncOpPayload::DeviceTrusted { .. } => "device_trusted",
SyncOpPayload::DeviceRevoked { .. } => "device_revoked",
SyncOpPayload::PlaybackCommand { .. } => "playback_command",
SyncOpPayload::ListenRecorded { .. } => "listen_recorded",
}
}
+41 -1
View File
@@ -416,6 +416,7 @@ struct Running {
pub struct Federation {
library: Arc<Library>,
devices: Arc<crate::devices::DeviceSync>,
jam: Arc<crate::jam::JamManager>,
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<u64>
}
impl Federation {
pub fn new(library: Arc<Library>, devices: Arc<crate::devices::DeviceSync>) -> Arc<Self> {
pub fn new(
library: Arc<Library>,
devices: Arc<crate::devices::DeviceSync>,
jam: Arc<crate::jam::JamManager>,
) -> Arc<Self> {
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<String> {
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<CachedTrackMetadata> {
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);
+732
View File
@@ -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<String>,
pub host_name: Option<String>,
pub invite: Option<String>,
pub participants: Vec<JamParticipant>,
pub connected: bool,
pub last_error: Option<String>,
}
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<JamCommand>,
},
Snapshot {
accepted: bool,
#[serde(default)]
error: Option<String>,
#[serde(default)]
acknowledged_command_ids: Vec<String>,
#[serde(default)]
playback: Option<PlaybackSnapshot>,
#[serde(default)]
participants: Vec<JamParticipant>,
host_time_ms: i64,
},
Leave {
version: u16,
jam_id: String,
secret: String,
participant_id: String,
},
}
#[derive(Default)]
struct HostState {
invite: Option<JamInvite>,
invite_uri: Option<String>,
playback: Option<PlaybackSnapshot>,
participants: HashMap<String, JamParticipant>,
seen_commands: HashSet<String>,
seen_order: VecDeque<String>,
}
struct JoinedState {
invite: JamInvite,
participant: JamParticipant,
pending: VecDeque<JamCommand>,
participants: Vec<JamParticipant>,
connected: bool,
last_error: Option<String>,
last_activity_ms: i64,
}
#[derive(Default)]
struct State {
host: HostState,
joined: Option<JoinedState>,
}
pub struct JamManager {
state: Mutex<State>,
event_tx: mpsc::UnboundedSender<AppEvent>,
}
impl JamManager {
pub fn new(event_tx: mpsc::UnboundedSender<AppEvent>) -> Arc<Self> {
Arc::new(Self {
state: Mutex::new(State::default()),
event_tx,
})
}
pub async fn create_or_regenerate(
&self,
service: Arc<MusicDhtService>,
host_device_id: String,
host_name: String,
) -> Result<String> {
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<MusicDhtService>) -> 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::<Vec<_>>(),
)
};
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::<HashSet<_>>();
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<JamManager>) {
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<JamManager>) -> 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<JamManager>, service: Arc<MusicDhtService>) {
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<WireMessage> {
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<String> {
Ok(format!(
"frid://j/{}",
base64url_encode(&serde_json::to_vec(invite)?)
))
}
fn parse_invite(uri: &str) -> Result<JamInvite> {
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<T>(mutex: &Mutex<T>) -> 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<Vec<u8>> {
fn decode(byte: u8) -> Option<u8> {
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);
}
}
+143 -21
View File
@@ -26,6 +26,16 @@ use models::{
pub const LIKES_PLAYLIST_ID: i64 = -1;
const NETWORK_ARTIST_CACHE_TTL_MS: i64 = 7 * 24 * 60 * 60 * 1000;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ListenHistoryEntry {
pub listen_id: String,
pub content_id: String,
pub title: String,
pub artist: String,
pub origin_device_id: String,
pub started_at_ms: i64,
}
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS artists (
id INTEGER PRIMARY KEY,
@@ -126,10 +136,43 @@ CREATE TABLE IF NOT EXISTS history (
completed INTEGER NOT NULL DEFAULT 0,
played_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS listen_events (
listen_id TEXT PRIMARY KEY,
content_id TEXT NOT NULL,
local_track_id INTEGER REFERENCES tracks(id) ON DELETE SET NULL,
origin_device_id TEXT NOT NULL,
started_at_ms INTEGER NOT NULL,
listened_ms INTEGER NOT NULL,
track_duration_ms INTEGER,
ended_reason TEXT NOT NULL,
qualified INTEGER NOT NULL,
metadata_json TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_tracks_release ON tracks(release_id);
CREATE INDEX IF NOT EXISTS idx_track_artists_artist ON track_artists(artist_id);
CREATE INDEX IF NOT EXISTS idx_release_artists_artist ON release_artists(artist_id);
CREATE INDEX IF NOT EXISTS idx_history_track ON history(track_id);
CREATE INDEX IF NOT EXISTS idx_listen_events_content
ON listen_events(content_id, started_at_ms DESC);
CREATE INDEX IF NOT EXISTS idx_listen_events_local_track
ON listen_events(local_track_id, qualified);
CREATE TRIGGER IF NOT EXISTS reconcile_listen_events_after_track_insert
AFTER INSERT ON tracks
WHEN NEW.content_id IS NOT NULL
BEGIN
UPDATE listen_events
SET local_track_id = NEW.id
WHERE content_id = NEW.content_id;
END;
CREATE TRIGGER IF NOT EXISTS reconcile_listen_events_after_track_content_id
AFTER UPDATE OF content_id ON tracks
WHEN NEW.content_id IS NOT NULL
BEGIN
UPDATE listen_events
SET local_track_id = NEW.id
WHERE content_id = NEW.content_id;
END;
CREATE INDEX IF NOT EXISTS idx_playlist_tracks_playlist ON playlist_tracks(playlist_id);
CREATE INDEX IF NOT EXISTS idx_fed_playlist_tracks_playlist
ON fed_playlist_tracks(playlist_sync_id, position);
@@ -159,7 +202,9 @@ const TRACK_COLUMNS: &str = "
t.file_path, t.audio_format, t.audio_bitrate, t.audio_sample_rate,
t.audio_bit_depth, t.file_size_bytes,
t.content_id,
(SELECT COUNT(*) FROM history h WHERE h.track_id = t.id AND h.completed = 1)
((SELECT COUNT(*) FROM history h WHERE h.track_id = t.id AND h.completed = 1)
+ (SELECT COUNT(*) FROM listen_events le
WHERE le.local_track_id = t.id AND le.qualified = 1))
";
/// Plain rows handed to the federation for publishing (see
@@ -798,9 +843,14 @@ impl Library {
|row| row.get(0),
)?;
let total_play_count: i64 = conn.query_row(
"SELECT COUNT(*) FROM history h
WHERE h.completed = 1 AND h.track_id IN
(SELECT track_id FROM track_artists WHERE artist_id = ?1)",
"SELECT
(SELECT COUNT(*) FROM history h
WHERE h.completed = 1 AND h.track_id IN
(SELECT track_id FROM track_artists WHERE artist_id = ?1))
+
(SELECT COUNT(*) FROM listen_events le
WHERE le.qualified = 1 AND le.local_track_id IN
(SELECT track_id FROM track_artists WHERE artist_id = ?1))",
[id],
|row| row.get(0),
)?;
@@ -1201,7 +1251,11 @@ impl Library {
.get(&track.id)
.copied()
.unwrap_or_default();
(liked_at, track.title.clone(), track)
let stable_key = track
.content_id
.clone()
.unwrap_or_else(|| format!("local:{:020}", track.id));
(liked_at, stable_key, track)
})
.collect();
@@ -1218,11 +1272,11 @@ impl Library {
})?;
for row in fed_rows {
let (liked_at, fed) = row?;
entries.push((
liked_at,
fed.title.clone(),
crate::federation::pending_track(&fed),
));
let stable_key = fed
.content_id
.clone()
.unwrap_or_else(|| format!("fed:{}", fed.item_id));
entries.push((liked_at, stable_key, crate::federation::pending_track(&fed)));
}
entries.sort_by(|left, right| right.0.cmp(&left.0).then_with(|| left.1.cmp(&right.1)));
return Ok(PlaylistDetail {
@@ -2192,20 +2246,88 @@ impl Library {
Ok(true)
}
pub fn add_history(
/// Idempotently materialize a portable trusted-device listen event.
pub fn apply_listen_event(
&self,
track_id: i64,
started_at: Option<i64>,
listened_seconds: i32,
completed: bool,
) -> Result<()> {
event: &music_dht::device_sync::ListenEvent,
origin_device_id: &str,
) -> Result<bool> {
if !event.should_record() || origin_device_id.trim().is_empty() {
return Ok(false);
}
let content_id = music_dht::normalize_content_id(&event.content_id)
.context("invalid listen content id")?;
let conn = self.lock();
conn.execute(
"INSERT INTO history (track_id, started_at, listened_seconds, completed)
VALUES (?1, ?2, ?3, ?4)",
params![track_id, started_at, listened_seconds, completed],
let local_track_id: Option<i64> = conn
.query_row(
"SELECT id FROM tracks WHERE content_id = ?1 LIMIT 1",
[&content_id],
|row| row.get(0),
)
.optional()?;
let inserted = conn.execute(
"INSERT OR IGNORE INTO listen_events
(listen_id, content_id, local_track_id, origin_device_id,
started_at_ms, listened_ms, track_duration_ms, ended_reason,
qualified, metadata_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
params![
event.listen_id,
content_id,
local_track_id,
origin_device_id,
event.started_at_ms,
event.listened_ms,
event.track_duration_ms,
serde_json::to_string(&event.ended_reason)?,
i64::from(event.qualifies_as_play()),
serde_json::to_string(&event.track)?,
],
)?;
Ok(())
Ok(inserted > 0)
}
/// Most recent qualified listens, including tracks that are not present
/// in this device's local library.
pub fn listen_history(&self, limit: usize) -> Result<Vec<ListenHistoryEntry>> {
let conn = self.lock();
let mut stmt = conn.prepare(
"SELECT listen_id, content_id, origin_device_id, started_at_ms, metadata_json
FROM listen_events
WHERE qualified = 1
ORDER BY started_at_ms DESC, listen_id DESC
LIMIT ?1",
)?;
let rows = stmt
.query_map([limit.min(i64::MAX as usize) as i64], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, i64>(3)?,
row.get::<_, String>(4)?,
))
})?
.collect::<rusqlite::Result<Vec<_>>>()?;
rows.into_iter()
.map(
|(listen_id, content_id, origin_device_id, started_at_ms, metadata_json)| {
let metadata: music_dht::device_sync::ListenTrackMetadata =
serde_json::from_str(&metadata_json)
.context("invalid listen history metadata")?;
let mut artists = metadata.artist_names;
artists.extend(metadata.featured_artist_names);
Ok(ListenHistoryEntry {
listen_id,
content_id,
title: metadata.title,
artist: artists.join(", "),
origin_device_id,
started_at_ms,
})
},
)
.collect()
}
// -----------------------------------------------------------------
+49 -2
View File
@@ -502,8 +502,55 @@ fn delete_track_drops_empty_release() {
fn history_counts_completed_plays() {
let lib = test_library();
let track_id = add_track(&lib, "Song", "Artist", "Album");
lib.add_history(track_id, None, 60, true).unwrap();
lib.add_history(track_id, None, 10, false).unwrap();
let content_id = lib
.tracks_by_ids(&[track_id])
.unwrap()
.remove(0)
.content_id
.unwrap();
let event = music_dht::device_sync::ListenEvent {
listen_id: "listen-1".to_string(),
content_id,
started_at_ms: 1_700_000_000_000,
listened_ms: 60_000,
track_duration_ms: Some(60_000),
ended_reason: music_dht::device_sync::ListenEndReason::Finished,
track: music_dht::device_sync::ListenTrackMetadata {
title: "Song".to_string(),
artist_names: vec!["Artist".to_string()],
featured_artist_names: Vec::new(),
release_title: Some("Album".to_string()),
},
};
assert!(lib.apply_listen_event(&event, "device-a").unwrap());
assert!(!lib.apply_listen_event(&event, "device-a").unwrap());
let track = lib.tracks_by_ids(&[track_id]).unwrap().remove(0);
assert_eq!(track.play_count, 1);
let history = lib.listen_history(20).unwrap();
assert_eq!(history.len(), 1);
assert_eq!(history[0].listen_id, "listen-1");
assert_eq!(history[0].title, "Song");
assert_eq!(history[0].artist, "Artist");
assert_eq!(history[0].origin_device_id, "device-a");
}
#[test]
fn listen_history_hides_unqualified_events_and_keeps_remote_metadata() {
let lib = test_library();
let event = music_dht::device_sync::ListenEvent {
listen_id: "remote-listen".to_string(),
content_id: format!("b3:{}", "a".repeat(64)),
started_at_ms: 1_700_000_000_000,
listened_ms: 10_000,
track_duration_ms: Some(120_000),
ended_reason: music_dht::device_sync::ListenEndReason::Skipped,
track: music_dht::device_sync::ListenTrackMetadata {
title: "Remote song".to_string(),
artist_names: vec!["Remote artist".to_string()],
featured_artist_names: vec!["Guest".to_string()],
release_title: None,
},
};
assert!(lib.apply_listen_event(&event, "remote-device").unwrap());
assert!(lib.listen_history(20).unwrap().is_empty());
}
+31 -1
View File
@@ -3,10 +3,12 @@ mod art;
mod config;
mod devices;
mod federation;
mod jam;
mod library;
mod media;
mod player;
mod share;
mod status;
mod streaming;
mod ui;
mod visualizer;
@@ -19,11 +21,39 @@ use crossterm::event::{
PopKeyboardEnhancementFlags, PushKeyboardEnhancementFlags,
};
const HELP: &str = "\
Furumi federated terminal music player
Usage:
furumi [OPTION]
Options:
-h, --help Show this help
-V, --version Show version
--status Print a one-line now-playing status
--status-json
Print now-playing status as JSON
tmux:
set -g status-right '#(furumi --status)'
";
fn main() -> Result<()> {
if std::env::args_os().any(|arg| arg == "--version" || arg == "-V") {
let args: Vec<_> = std::env::args_os().collect();
if args.iter().any(|arg| arg == "--help" || arg == "-h") {
print!("{HELP}");
return Ok(());
}
if args.iter().any(|arg| arg == "--version" || arg == "-V") {
println!("furumi {}", env!("CARGO_PKG_VERSION"));
return Ok(());
}
if args
.iter()
.any(|arg| arg == "--status" || arg == "--status-json")
{
return status::print(args.iter().any(|arg| arg == "--status-json"));
}
let mut startup_warning = None;
if let Err(err) = config::logging::init() {
+190
View File
@@ -0,0 +1,190 @@
//! Cheap now-playing export for status bars and other polling clients.
//!
//! The UI only sends snapshots through a bounded channel. A dedicated thread
//! performs the filesystem writes so a slow filesystem cannot stall drawing
//! or input handling.
use std::fs;
use std::io;
use std::path::PathBuf;
use std::sync::mpsc::{self, SyncSender, TrySendError};
use std::time::{SystemTime, UNIX_EPOCH};
use serde::{Deserialize, Serialize};
use crate::app::state::PlayerBar;
const STALE_AFTER_MS: u64 = 5_000;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct PlaybackStatus {
pub playing: bool,
pub paused: bool,
pub title: String,
pub artist: String,
pub album: String,
pub position_secs: f64,
pub duration_secs: f64,
pub volume: u8,
pub updated_at_ms: u64,
}
impl PlaybackStatus {
pub fn from_player(player: &PlayerBar) -> Option<Self> {
let track = player.current.as_ref()?;
player.playing.then(|| Self {
playing: true,
paused: player.paused,
title: track.title.clone(),
artist: track.artist_line(),
album: track.release_title.clone(),
position_secs: player.position_secs.max(0.0),
duration_secs: track.duration_seconds.max(0.0),
volume: player.volume,
updated_at_ms: now_ms(),
})
}
pub fn one_line(&self) -> String {
let icon = if self.paused { "" } else { "" };
let track = if self.artist.is_empty() {
self.title.clone()
} else {
format!("{}{}", self.artist, self.title)
};
format!(
"{icon} {track} {}/{}",
duration(self.position_secs),
duration(self.duration_secs)
)
}
}
pub struct Publisher {
tx: Option<SyncSender<Option<PlaybackStatus>>>,
}
impl Publisher {
pub fn spawn() -> Self {
let (tx, rx) = mpsc::sync_channel(1);
std::thread::Builder::new()
.name("playback-status".to_string())
.spawn(move || {
while let Ok(snapshot) = rx.recv() {
if let Some(snapshot) = snapshot {
if let Err(err) = write(&snapshot) {
tracing::debug!(?err, "writing playback status failed");
}
} else {
remove();
}
}
remove();
})
.ok();
Self { tx: Some(tx) }
}
pub fn publish(&self, snapshot: Option<PlaybackStatus>) {
let Some(tx) = &self.tx else { return };
match tx.try_send(snapshot) {
Ok(()) | Err(TrySendError::Full(_)) => {}
Err(TrySendError::Disconnected(_)) => {}
}
}
}
impl Drop for Publisher {
fn drop(&mut self) {
self.tx.take();
}
}
pub fn print(json: bool) -> anyhow::Result<()> {
let Some(status) = read()? else {
return Ok(());
};
if json {
println!("{}", serde_json::to_string(&status)?);
} else {
println!("{}", status.one_line());
}
Ok(())
}
fn read() -> anyhow::Result<Option<PlaybackStatus>> {
let Some(path) = path() else {
return Ok(None);
};
let bytes = match fs::read(path) {
Ok(bytes) => bytes,
Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(None),
Err(err) => return Err(err.into()),
};
let status: PlaybackStatus = serde_json::from_slice(&bytes)?;
if now_ms().saturating_sub(status.updated_at_ms) > STALE_AFTER_MS {
return Ok(None);
}
Ok(Some(status))
}
fn write(status: &PlaybackStatus) -> anyhow::Result<()> {
let Some(path) = path() else {
return Ok(());
};
let Some(dir) = path.parent() else {
return Ok(());
};
fs::create_dir_all(dir)?;
let temporary = dir.join("playback-status.tmp");
fs::write(&temporary, serde_json::to_vec(status)?)?;
#[cfg(windows)]
if path.exists() {
let _ = fs::remove_file(&path);
}
fs::rename(temporary, path)?;
Ok(())
}
fn remove() {
if let Some(path) = path() {
let _ = fs::remove_file(path);
}
}
fn path() -> Option<PathBuf> {
crate::config::project_dirs().map(|dirs| dirs.cache_dir().join("playback-status.json"))
}
fn now_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis() as u64)
.unwrap_or(0)
}
fn duration(seconds: f64) -> String {
let total = seconds.max(0.0).round() as u64;
format!("{}:{:02}", total / 60, total % 60)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn one_line_contains_state_metadata_and_progress() {
let status = PlaybackStatus {
playing: true,
paused: true,
title: "Track".into(),
artist: "Artist".into(),
album: "Release".into(),
position_secs: 83.0,
duration_secs: 245.0,
volume: 80,
updated_at_ms: 0,
};
assert_eq!(status.one_line(), "⏸ Artist — Track 1:23/4:05");
}
}
+2 -1
View File
@@ -152,7 +152,8 @@ fn draw_tile_meta(
availability: Option<Availability>,
selected: bool,
) {
let marker = availability.map(|availability| availability_marker(availability, selected));
let selected_style = selected.then(|| theme::tab_active_for(state));
let marker = availability.map(|availability| availability_marker(availability, selected_style));
let marker_width = marker
.map(|(label, _)| UnicodeWidthStr::width(label) as u16)
.unwrap_or(0)
+18 -7
View File
@@ -19,25 +19,36 @@ use crate::library::models::Availability;
pub(crate) fn availability_marker(
availability: Availability,
selected: bool,
selected_style: Option<Style>,
) -> (&'static str, Style) {
let (label, style) = match availability {
Availability::Local => ("", Style::new().fg(Color::Green)),
Availability::Mixed => ("", Style::new().fg(Color::Yellow)),
Availability::Remote => ("", theme::accent()),
};
if selected {
(label, theme::tab_active())
} else {
(label, style)
}
(label, selected_style.unwrap_or(style))
}
pub(crate) fn availability_prefix(availability: Availability) -> Span<'static> {
let (label, style) = availability_marker(availability, false);
let (label, style) = availability_marker(availability, None);
Span::styled(format!("{label} "), style)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn selected_availability_marker_uses_the_supplied_mode_style() {
let client_style = Style::new().fg(Color::Black).bg(Color::Yellow);
let (label, style) = availability_marker(Availability::Local, Some(client_style));
assert_eq!(label, "");
assert_eq!(style, client_style);
}
}
pub fn draw(frame: &mut Frame, state: &AppState, keymap: &Keymap) {
if state.visualizer.active {
crate::visualizer::draw(frame, state);
+168 -11
View File
@@ -1,7 +1,7 @@
use ratatui::Frame;
use ratatui::layout::{Alignment, Constraint, Flex, Layout, Rect};
use ratatui::text::{Line, Span};
use ratatui::widgets::{Block, Clear, Paragraph, Wrap};
use ratatui::widgets::{Block, Cell, Clear, Paragraph, Row, Table, TableState, Wrap};
use unicode_width::{UnicodeWidthChar, UnicodeWidthStr};
use super::theme;
@@ -88,10 +88,121 @@ pub fn draw(frame: &mut Frame, state: &AppState) {
}
Some(Popup::ConfirmDeviceLeave) => draw_device_leave(frame, state),
Some(Popup::ConnectedDevices { cursor }) => draw_connected_devices(frame, state, *cursor),
Some(Popup::ListenHistory { cursor }) => draw_listen_history(frame, state, *cursor),
None => {}
}
}
fn draw_listen_history(frame: &mut Frame, state: &AppState, cursor: usize) {
let area = centered(
frame.area(),
100,
frame.area().height.saturating_sub(4).clamp(10, 30),
);
let block = Block::bordered()
.title(" Listening history ")
.title_style(theme::header_for(state))
.border_style(theme::strong_border_for(state));
let inner = block.inner(area);
frame.render_widget(Clear, area);
frame.render_widget(block, area);
let [body, footer] = Layout::vertical([Constraint::Min(1), Constraint::Length(1)]).areas(inner);
match state.listen_history.as_ref() {
Some(Loadable::Loading) | None => {
frame.render_widget(
Paragraph::new(format!("{} Loading history…", state.spinner()))
.alignment(Alignment::Center),
body,
);
}
Some(Loadable::Failed(err)) => {
frame.render_widget(
Paragraph::new(format!("History unavailable: {err}"))
.style(theme::dim())
.wrap(Wrap { trim: true }),
body,
);
}
Some(Loadable::Ready(entries)) if entries.is_empty() => {
frame.render_widget(
Paragraph::new("No qualified listens yet.")
.style(theme::dim())
.alignment(Alignment::Center),
body,
);
}
Some(Loadable::Ready(entries)) => {
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|duration| duration.as_millis().min(i64::MAX as u128) as i64)
.unwrap_or_default();
let rows = entries.iter().map(|entry| {
Row::new(vec![
Cell::from(entry.title.clone()),
Cell::from(entry.artist.clone()),
Cell::from(relative_listen_time(entry.started_at_ms, now_ms)),
Cell::from(history_device_name(state, &entry.origin_device_id)),
])
});
let mut table_state =
TableState::default().with_selected(cursor.min(entries.len() - 1));
let table = Table::new(
rows,
[
Constraint::Percentage(34),
Constraint::Percentage(28),
Constraint::Length(12),
Constraint::Percentage(26),
],
)
.header(Row::new(["Track", "Artist", "When", "Device"]).style(theme::header_for(state)))
.row_highlight_style(theme::selection_for(state))
.highlight_symbol(" ");
frame.render_stateful_widget(table, body, &mut table_state);
}
}
frame.render_widget(
Paragraph::new("j/k scroll · pgup/pgdn page · esc close")
.style(theme::dim())
.alignment(Alignment::Center),
footer,
);
}
fn history_device_name(state: &AppState, device_id: &str) -> String {
state
.federation
.devices
.as_ref()
.and_then(|status| {
status
.devices
.iter()
.find(|device| device.device_id == device_id)
.map(|device| device.name.clone())
})
.filter(|name| !name.trim().is_empty())
.unwrap_or_else(|| {
if device_id == state.device_playback.self_device_id {
state.device_playback.self_device_name.clone()
} else {
device_id.chars().take(10).collect()
}
})
}
fn relative_listen_time(started_at_ms: i64, now_ms: i64) -> String {
let elapsed = now_ms.saturating_sub(started_at_ms).max(0) / 1_000;
match elapsed {
0..=59 => "now".to_string(),
60..=3_599 => format!("{}m ago", elapsed / 60),
3_600..=86_399 => format!("{}h ago", elapsed / 3_600),
86_400..=604_799 => format!("{}d ago", elapsed / 86_400),
_ => format!("{}w ago", elapsed / 604_800),
}
}
fn draw_federation_status_details(
frame: &mut Frame,
state: &AppState,
@@ -329,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 ")
@@ -339,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),
])
@@ -368,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());
@@ -390,7 +501,53 @@ fn draw_connected_devices(frame: &mut Frame, state: &AppState, cursor: usize) {
state,
);
render_subtitle(frame, other_area, state, "Other devices");
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 · {} peer(s) · c copy · h new · l leave",
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, "My devices");
let list_area = Rect {
x: other_area.x,
y: other_area.y + 1,
@@ -449,7 +606,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),
+14 -19
View File
@@ -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 {
@@ -18,13 +19,6 @@ pub fn dim() -> Style {
Style::new().fg(DIM)
}
pub fn tab_active() -> Style {
Style::new()
.fg(Color::Black)
.bg(ACCENT)
.add_modifier(Modifier::BOLD)
}
pub fn tab_active_for(state: &AppState) -> Style {
Style::new()
.fg(Color::Black)
@@ -44,10 +38,10 @@ 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))
} else {
selection()
match state.device_playback.role {
DevicePlaybackRole::Jam => Style::new().fg(Color::White).bg(Color::Rgb(80, 24, 96)),
DevicePlaybackRole::Control => Style::new().fg(Color::White).bg(Color::Rgb(92, 72, 0)),
DevicePlaybackRole::Active => selection(),
}
}
@@ -56,10 +50,10 @@ pub fn header_for(state: &AppState) -> Style {
}
pub fn border_for(state: &AppState) -> Style {
if state.device_playback.is_control() {
Style::new().fg(CONTROL_ACCENT)
} else {
dim()
match state.device_playback.role {
DevicePlaybackRole::Jam => Style::new().fg(JAM_ACCENT),
DevicePlaybackRole::Control => Style::new().fg(CONTROL_ACCENT),
DevicePlaybackRole::Active => dim(),
}
}
@@ -71,6 +65,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)
@@ -79,9 +74,9 @@ pub fn role_pill(role: DevicePlaybackRole) -> Style {
}
fn accent_color_for(state: &AppState) -> Color {
if state.device_playback.is_control() {
CONTROL_ACCENT
} else {
ACCENT
match state.device_playback.role {
DevicePlaybackRole::Jam => JAM_ACCENT,
DevicePlaybackRole::Control => CONTROL_ACCENT,
DevicePlaybackRole::Active => ACCENT,
}
}