Compare commits
3
Commits
d61d7a6bac
...
v0.9.8
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
05d68724f3 | ||
|
|
d71845509d | ||
|
|
4efcfdc539 |
No files matched your search
Generated
+1
-1
@@ -1836,7 +1836,7 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c"
|
||||
|
||||
[[package]]
|
||||
name = "furumusic"
|
||||
version = "0.9.6"
|
||||
version = "0.9.7"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"async-stream",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "furumusic"
|
||||
version = "0.9.7"
|
||||
version = "0.9.8"
|
||||
edition = "2024"
|
||||
description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL"
|
||||
|
||||
|
||||
@@ -32,6 +32,9 @@ const RESPONSE_DRAIN_TIMEOUT: Duration = Duration::from_secs(2);
|
||||
const DEVICE_SYNC_INTERVAL: Duration = Duration::from_secs(2);
|
||||
const LOCAL_SEED_RECHECK_MS: i64 = 60 * 1000;
|
||||
const MAX_LINE: usize = 8 * 1024 * 1024;
|
||||
/// Playback commands carry queue state but are useful only briefly and only
|
||||
/// to their addressed device. They must not inflate a new peer's catch-up.
|
||||
const PLAYBACK_COMMAND_TTL_MS: i64 = 5 * 60 * 1_000;
|
||||
const MAX_OPS_PER_BATCH: i64 = 1000;
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
@@ -84,6 +87,7 @@ struct Identity {
|
||||
#[derive(Debug, Clone)]
|
||||
struct StoredDevice {
|
||||
device_id: String,
|
||||
endpoint_id: String,
|
||||
endpoint_ticket: String,
|
||||
}
|
||||
|
||||
@@ -1573,8 +1577,7 @@ async fn sync_device(
|
||||
user_id: i64,
|
||||
device: &StoredDevice,
|
||||
) -> Result<()> {
|
||||
let ticket: PeerTicket = device.endpoint_ticket.parse()?;
|
||||
let peer = service.connect(ticket).await?;
|
||||
let peer = resolve_device_peer(&service, device).await?;
|
||||
let own_ticket = service.ticket().await?.to_string();
|
||||
let identity = ensure_identity(pool, user_id, "").await?;
|
||||
let profile = own_profile(pool, user_id, "", &own_ticket).await?;
|
||||
@@ -3420,12 +3423,16 @@ async fn local_playback_snapshot(
|
||||
user_id: i64,
|
||||
identity: &Identity,
|
||||
) -> Option<PlaybackSnapshot> {
|
||||
let state = hub.current_playback_state_json(user_id)?;
|
||||
// Keep publishing an inactive snapshot after a handoff. Omitting the
|
||||
// snapshot left the last `active: true` value alive on trusted peers until
|
||||
// its TTL elapsed, allowing the always-on web peer to reclaim playback.
|
||||
let active = hub.federation_playback_is_local(user_id);
|
||||
let state = hub.playback_state_json_for_commands(user_id)?;
|
||||
let wire = playback_state_from_browser_json(pool, state).await.ok()?;
|
||||
Some(PlaybackSnapshot {
|
||||
device_id: identity.device_id.clone(),
|
||||
device_name: identity.name.clone(),
|
||||
active: true,
|
||||
active,
|
||||
updated_at_ms: now_ms(),
|
||||
state: wire,
|
||||
})
|
||||
@@ -4118,7 +4125,7 @@ pub async fn playlist_content_ids_for_removal(
|
||||
async fn active_remote_devices(pool: &sqlx::PgPool, user_id: i64) -> Result<Vec<StoredDevice>> {
|
||||
let identity = ensure_identity(pool, user_id, "").await?;
|
||||
let rows = sqlx::query(
|
||||
"SELECT device_id, endpoint_ticket
|
||||
"SELECT device_id, endpoint_id, endpoint_ticket
|
||||
FROM furumusic__fed_device
|
||||
WHERE user_id = $1
|
||||
AND trusted_at_ms IS NOT NULL
|
||||
@@ -4134,11 +4141,31 @@ async fn active_remote_devices(pool: &sqlx::PgPool, user_id: i64) -> Result<Vec<
|
||||
.into_iter()
|
||||
.map(|row| StoredDevice {
|
||||
device_id: row.get("device_id"),
|
||||
endpoint_id: row.get("endpoint_id"),
|
||||
endpoint_ticket: row.get("endpoint_ticket"),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
async fn resolve_device_peer(
|
||||
service: &MusicDhtService,
|
||||
device: &StoredDevice,
|
||||
) -> Result<music_dht::EndpointId> {
|
||||
if let Ok(peer) = device.endpoint_id.parse::<music_dht::EndpointId>()
|
||||
&& (service.connected_peers().contains(&peer)
|
||||
|| service
|
||||
.known_peers()
|
||||
.iter()
|
||||
.any(|contact| contact.peer_id == peer))
|
||||
{
|
||||
// Prefer the live/current-schema DHT contact over a persisted ticket
|
||||
// that may have been issued before a schema upgrade.
|
||||
return Ok(peer);
|
||||
}
|
||||
let ticket: PeerTicket = device.endpoint_ticket.parse()?;
|
||||
service.connect(ticket).await.map_err(Into::into)
|
||||
}
|
||||
|
||||
async fn active_device_count(pool: &sqlx::PgPool, user_id: i64) -> Result<i64> {
|
||||
sqlx::query_scalar(
|
||||
"SELECT COUNT(*) FROM furumusic__fed_device
|
||||
@@ -4202,11 +4229,19 @@ async fn ops_for_peer(
|
||||
AND a.origin_device_id = o.origin_device_id
|
||||
WHERE o.user_id = $1
|
||||
AND o.seq > COALESCE(a.max_seq, 0)
|
||||
AND (
|
||||
o.kind != 'playback_command'
|
||||
OR (
|
||||
o.hlc_ms >= $3
|
||||
AND o.payload_json->>'target_device_id' = $2
|
||||
)
|
||||
)
|
||||
ORDER BY o.hlc_ms, o.op_id
|
||||
LIMIT $3",
|
||||
LIMIT $4",
|
||||
)
|
||||
.bind(user_id)
|
||||
.bind(peer_device_id)
|
||||
.bind(now_ms().saturating_sub(PLAYBACK_COMMAND_TTL_MS))
|
||||
.bind(MAX_OPS_PER_BATCH)
|
||||
.fetch_all(pool)
|
||||
.await?;
|
||||
|
||||
+176
-12
@@ -54,6 +54,7 @@ struct LocalUploadResponse {
|
||||
}
|
||||
|
||||
const PLAYER_DEVICE_TTL_MS: i64 = 30_000;
|
||||
const PLAYER_DEVICE_RETURN_TAKEOVER_MS: i64 = 30 * 60 * 1_000;
|
||||
const PLAYER_DEVICE_COMMAND_TTL_MS: i64 = 20_000;
|
||||
const PLAYER_DEVICE_MAX_COMMANDS: usize = 32;
|
||||
const PLAYER_JAM_IDLE_TTL_MS: i64 = 4 * 60 * 60 * 1000;
|
||||
@@ -104,6 +105,7 @@ struct PlayerJamSession {
|
||||
#[derive(Debug, Default)]
|
||||
struct PlayerDeviceHubState {
|
||||
devices_by_user: HashMap<i64, HashMap<String, PlayerDevice>>,
|
||||
device_last_seen_ms: HashMap<(i64, String), i64>,
|
||||
active_device_by_user: HashMap<i64, String>,
|
||||
commands_by_device: HashMap<(i64, String), VecDeque<PendingPlayerDeviceCommand>>,
|
||||
playback_state_by_user: HashMap<i64, PlayerDevicePlaybackStateDto>,
|
||||
@@ -153,19 +155,12 @@ impl PlayerDeviceHub {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn current_playback_state_json(&self, user_id: i64) -> Option<serde_json::Value> {
|
||||
pub(crate) fn federation_playback_is_local(&self, user_id: i64) -> bool {
|
||||
let state = self.state.lock().expect("player device hub lock");
|
||||
if state
|
||||
!state
|
||||
.active_device_by_user
|
||||
.get(&user_id)
|
||||
.is_some_and(|id| is_fed_virtual_device_id(id))
|
||||
{
|
||||
return None;
|
||||
}
|
||||
state
|
||||
.playback_state_by_user
|
||||
.get(&user_id)
|
||||
.and_then(|playback| serde_json::to_value(playback).ok())
|
||||
}
|
||||
|
||||
pub(crate) fn playback_state_json_for_commands(
|
||||
@@ -264,14 +259,58 @@ impl PlayerDeviceHub {
|
||||
user_agent: Option<&str>,
|
||||
current_jam_id: Option<&str>,
|
||||
playback_state: Option<PlayerDevicePlaybackStateDto>,
|
||||
) -> PlayerDevicesResponse {
|
||||
) -> (PlayerDevicesResponse, Option<String>) {
|
||||
let now = current_millis();
|
||||
let mut state = self.state.lock().expect("player device hub lock");
|
||||
self.prune_locked(&mut state, now);
|
||||
let is_new_or_returning = state
|
||||
.device_last_seen_ms
|
||||
.get(&(user_id, device_id.to_string()))
|
||||
.is_none_or(|last_seen| {
|
||||
now.saturating_sub(*last_seen) >= PLAYER_DEVICE_RETURN_TAKEOVER_MS
|
||||
});
|
||||
let previous_active_id = state.active_device_by_user.get(&user_id).cloned();
|
||||
self.touch_locked(&mut state, user_id, device_id, user_agent, now);
|
||||
let active_is_playing = state
|
||||
.playback_state_by_user
|
||||
.get(&user_id)
|
||||
.is_some_and(|playback| playback.track.is_some() && !playback.paused);
|
||||
let should_claim_idle_playback = is_new_or_returning
|
||||
&& previous_active_id.as_deref() != Some(device_id)
|
||||
&& !active_is_playing;
|
||||
if should_claim_idle_playback {
|
||||
let transfer_state = state
|
||||
.playback_state_by_user
|
||||
.get(&user_id)
|
||||
.cloned()
|
||||
.map(|playback| playback_state_at(playback, now));
|
||||
state
|
||||
.active_device_by_user
|
||||
.insert(user_id, device_id.to_string());
|
||||
if let Some(transfer_state) = transfer_state {
|
||||
state
|
||||
.playback_state_by_user
|
||||
.insert(user_id, transfer_state.clone());
|
||||
if let Ok(payload) = serde_json::to_value(transfer_state) {
|
||||
self.enqueue_command_locked(
|
||||
&mut state,
|
||||
user_id,
|
||||
device_id,
|
||||
"transfer_state",
|
||||
payload,
|
||||
now,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
self.update_playback_state_locked(&mut state, user_id, device_id, playback_state, now);
|
||||
self.touch_jam_locked(&mut state, user_id, device_id, current_jam_id, now);
|
||||
self.snapshot_locked(&state, user_id, device_id, current_jam_id, now)
|
||||
(
|
||||
self.snapshot_locked(&state, user_id, device_id, current_jam_id, now),
|
||||
should_claim_idle_playback
|
||||
.then_some(previous_active_id)
|
||||
.flatten(),
|
||||
)
|
||||
}
|
||||
|
||||
fn poll(
|
||||
@@ -448,6 +487,9 @@ impl PlayerDeviceHub {
|
||||
last_seen_ms: now,
|
||||
};
|
||||
devices.insert(device_id.to_string(), device);
|
||||
state
|
||||
.device_last_seen_ms
|
||||
.insert((user_id, device_id.to_string()), now);
|
||||
|
||||
let active_online = state
|
||||
.active_device_by_user
|
||||
@@ -868,6 +910,9 @@ impl PlayerDeviceHub {
|
||||
}
|
||||
|
||||
fn prune_locked(&self, state: &mut PlayerDeviceHubState, now: i64) {
|
||||
state.device_last_seen_ms.retain(|_, last_seen| {
|
||||
now.saturating_sub(*last_seen) <= PLAYER_DEVICE_RETURN_TAKEOVER_MS
|
||||
});
|
||||
state
|
||||
.jams_by_id
|
||||
.retain(|_, jam| now.saturating_sub(jam.host_last_seen_ms) <= PLAYER_JAM_IDLE_TTL_MS);
|
||||
@@ -1185,6 +1230,109 @@ mod device_tests {
|
||||
.is_some_and(|devices| devices.contains_key("fed:remote"))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn new_browser_claims_an_idle_federated_player() {
|
||||
let hub = PlayerDeviceHub::default();
|
||||
let user_id = 8;
|
||||
hub.apply_fed_playback_state_json(
|
||||
user_id,
|
||||
"remote",
|
||||
"Remote",
|
||||
true,
|
||||
serde_json::json!({
|
||||
"track": {"id": 2},
|
||||
"tracks": [],
|
||||
"index": 0,
|
||||
"position_seconds": 12.0,
|
||||
"duration_seconds": 100.0,
|
||||
"paused": true,
|
||||
"shuffle": false,
|
||||
"repeat_mode": "off",
|
||||
"volume": 0.7
|
||||
}),
|
||||
)
|
||||
.expect("valid snapshot");
|
||||
|
||||
let (response, previous) = hub.heartbeat(user_id, "browser", None, None, None);
|
||||
|
||||
assert_eq!(response.active_device_id.as_deref(), Some("browser"));
|
||||
assert_eq!(previous.as_deref(), Some("fed:remote"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn new_browser_does_not_claim_a_playing_federated_player() {
|
||||
let hub = PlayerDeviceHub::default();
|
||||
let user_id = 9;
|
||||
hub.apply_fed_playback_state_json(
|
||||
user_id,
|
||||
"remote",
|
||||
"Remote",
|
||||
true,
|
||||
serde_json::json!({
|
||||
"track": {"id": 2},
|
||||
"tracks": [],
|
||||
"index": 0,
|
||||
"position_seconds": 12.0,
|
||||
"duration_seconds": 100.0,
|
||||
"paused": false,
|
||||
"shuffle": false,
|
||||
"repeat_mode": "off",
|
||||
"volume": 0.7
|
||||
}),
|
||||
)
|
||||
.expect("valid snapshot");
|
||||
|
||||
let (response, previous) = hub.heartbeat(user_id, "browser", None, None, None);
|
||||
|
||||
assert_eq!(response.active_device_id.as_deref(), Some("fed:remote"));
|
||||
assert_eq!(previous, None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn refreshed_control_browser_keeps_its_control_role() {
|
||||
let hub = PlayerDeviceHub::default();
|
||||
let user_id = 10;
|
||||
hub.apply_fed_playback_state_json(
|
||||
user_id,
|
||||
"remote",
|
||||
"Remote",
|
||||
true,
|
||||
serde_json::json!({
|
||||
"track": {"id": 2},
|
||||
"tracks": [],
|
||||
"index": 0,
|
||||
"position_seconds": 12.0,
|
||||
"duration_seconds": 100.0,
|
||||
"paused": true,
|
||||
"shuffle": false,
|
||||
"repeat_mode": "off",
|
||||
"volume": 0.7
|
||||
}),
|
||||
)
|
||||
.expect("valid snapshot");
|
||||
{
|
||||
let mut state = hub.state.lock().expect("device hub");
|
||||
let now = current_millis();
|
||||
state.devices_by_user.entry(user_id).or_default().insert(
|
||||
"browser".to_string(),
|
||||
PlayerDevice {
|
||||
id: "browser".to_string(),
|
||||
name: "Browser".to_string(),
|
||||
kind: "computer".to_string(),
|
||||
last_seen_ms: now,
|
||||
},
|
||||
);
|
||||
state
|
||||
.device_last_seen_ms
|
||||
.insert((user_id, "browser".to_string()), now);
|
||||
}
|
||||
|
||||
let (response, previous) = hub.heartbeat(user_id, "browser", None, None, None);
|
||||
|
||||
assert_eq!(response.active_device_id.as_deref(), Some("fed:remote"));
|
||||
assert_eq!(previous, None);
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, sqlx::FromRow)]
|
||||
@@ -4846,7 +4994,7 @@ async fn devices_heartbeat_handler(
|
||||
return Ok(json_error(StatusCode::BAD_REQUEST, "invalid device id"));
|
||||
};
|
||||
|
||||
let response = hub.heartbeat(
|
||||
let (response, previous_active_id) = hub.heartbeat(
|
||||
user.id,
|
||||
&device_id,
|
||||
dto.user_agent.as_deref(),
|
||||
@@ -4856,6 +5004,22 @@ async fn devices_heartbeat_handler(
|
||||
.as_deref(),
|
||||
dto.playback_state,
|
||||
);
|
||||
if let Some(previous_fed_device_id) = previous_active_id
|
||||
.as_deref()
|
||||
.and_then(fed_device_id_from_virtual)
|
||||
&& let Some(playback_state) = response.playback_state.clone()
|
||||
{
|
||||
let state = match serde_json::to_value(playback_state) {
|
||||
Ok(state) => state,
|
||||
Err(err) => return Ok(json_error(StatusCode::BAD_REQUEST, &format!("{err}"))),
|
||||
};
|
||||
if let Err(err) = crate::federation::handle()
|
||||
.fed_device_web_active_takeover(user.id, previous_fed_device_id, state)
|
||||
.await
|
||||
{
|
||||
return Ok(json_error(StatusCode::BAD_REQUEST, &format!("{err}")));
|
||||
}
|
||||
}
|
||||
Json(response).into_response()
|
||||
}
|
||||
|
||||
|
||||
@@ -757,8 +757,22 @@
|
||||
</div>
|
||||
</div>
|
||||
<template x-if="$store.library.currentArtist.top_tracks && $store.library.currentArtist.top_tracks.length > 0">
|
||||
<section class="artist-release-group">
|
||||
<h2 class="artist-release-group-title">{{ t.player_top_tracks }}</h2>
|
||||
<section class="artist-release-group" x-data="{ expanded: false }">
|
||||
<div class="artist-release-group-heading">
|
||||
<h2 class="artist-release-group-title">{{ t.player_top_tracks }}</h2>
|
||||
<button class="artist-top-tracks-toggle"
|
||||
type="button"
|
||||
x-show="$store.library.currentArtist.top_tracks.length > 5"
|
||||
@click="expanded = !expanded"
|
||||
:aria-expanded="expanded"
|
||||
x-cloak>
|
||||
<span x-text="expanded ? '{{ t.player_collapse }}' : '{{ t.player_expand_all }}'"></span>
|
||||
<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"
|
||||
:class="{ expanded: expanded }">
|
||||
<polyline points="6 9 12 15 18 9"/>
|
||||
</svg>
|
||||
</button>
|
||||
</div>
|
||||
<div class="track-list-header">
|
||||
<span>#</span>
|
||||
<span>{{ t.player_title }}</span>
|
||||
@@ -766,7 +780,9 @@
|
||||
<span></span>
|
||||
<span style="text-align:right">{{ t.player_duration }}</span>
|
||||
</div>
|
||||
<template x-for="(track, idx) in $store.library.currentArtist.top_tracks" :key="track.id">
|
||||
<template x-for="(track, idx) in (expanded
|
||||
? $store.library.currentArtist.top_tracks
|
||||
: $store.library.currentArtist.top_tracks.slice(0, 5))" :key="track.id">
|
||||
<div class="track-row artist-appearance-row"
|
||||
:class="{ playing: $store.player.currentTrack && $store.player.currentTrack.id === track.id }"
|
||||
@dblclick="$store.queue.playRelease($store.library.currentArtist.top_tracks, idx)">
|
||||
|
||||
@@ -676,6 +676,42 @@ button.user-stat:hover {
|
||||
margin-bottom: 14px;
|
||||
text-transform: capitalize;
|
||||
}
|
||||
.artist-release-group-heading {
|
||||
display: flex;
|
||||
align-items: center;
|
||||
justify-content: space-between;
|
||||
gap: 16px;
|
||||
margin-bottom: 14px;
|
||||
}
|
||||
.artist-release-group-heading .artist-release-group-title {
|
||||
margin-bottom: 0;
|
||||
}
|
||||
.artist-top-tracks-toggle {
|
||||
border: 0;
|
||||
background: transparent;
|
||||
color: var(--text-subdued);
|
||||
display: inline-flex;
|
||||
align-items: center;
|
||||
gap: 6px;
|
||||
padding: 5px 7px;
|
||||
border-radius: 5px;
|
||||
font: inherit;
|
||||
font-size: 12px;
|
||||
font-weight: 700;
|
||||
cursor: pointer;
|
||||
}
|
||||
.artist-top-tracks-toggle:hover {
|
||||
color: var(--text-primary);
|
||||
background: var(--bg-hover);
|
||||
}
|
||||
.artist-top-tracks-toggle svg {
|
||||
width: 15px;
|
||||
height: 15px;
|
||||
transition: transform 0.18s ease;
|
||||
}
|
||||
.artist-top-tracks-toggle svg.expanded {
|
||||
transform: rotate(180deg);
|
||||
}
|
||||
|
||||
/* Release detail header */
|
||||
.release-header {
|
||||
|
||||
Reference in new issue
Block a user