Compare commits

...
3 Commits
Author SHA1 Message Date
Ultradesu 134f89fc9d Bump libs
Build and Publish / Build and Publish Docker Image (push) Successful in 5m18s
2026-07-27 23:17:07 +01:00
Ultradesu 232c171aac Add synced listening history 2026-07-27 23:16:03 +01:00
Ultradesu 845df4e031 Fixed autoplay after fed download 2026-07-27 17:33:48 +01:00
14 changed files with 641 additions and 270 deletions
Generated
+11 -21
View File
@@ -806,15 +806,6 @@ version = "0.4.32"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cc14f565cf027a105f7a44ccf9e5b424348421a1d8952a8fc9d499d313107789"
[[package]]
name = "concurrent-queue"
version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973"
dependencies = [
"crossbeam-utils",
]
[[package]]
name = "const-oid"
version = "0.9.6"
@@ -1673,11 +1664,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",
]
@@ -1718,7 +1708,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#8de7d1292708fa0b225e5a4a9d5ab4f0676202d3"
dependencies = [
"blake3",
"data-encoding",
@@ -1845,7 +1835,7 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c"
[[package]]
name = "furumusic"
version = "0.9.1"
version = "0.9.4"
dependencies = [
"anyhow",
"async-stream",
@@ -3623,8 +3613,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#8de7d1292708fa0b225e5a4a9d5ab4f0676202d3"
dependencies = [
"async-trait",
"blake3",
@@ -5296,9 +5286,9 @@ dependencies = [
[[package]]
name = "schemars"
version = "1.2.1"
version = "1.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a2b42f36aa1cd011945615b92222f6bf73c599a102a300334cd7f8dbeec726cc"
checksum = "687274d293b6cdc6e73e0fee520bf2049650090d7164f87672d212a3c530cf4a"
dependencies = [
"dyn-clone",
"ref-cast",
@@ -5559,7 +5549,7 @@ dependencies = [
"indexmap 1.9.3",
"indexmap 2.14.0",
"schemars 0.9.0",
"schemars 1.2.1",
"schemars 1.2.2",
"serde_core",
"serde_json",
"serde_with_macros",
@@ -6616,9 +6606,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 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "furumusic"
version = "0.9.2"
version = "0.9.4"
edition = "2024"
description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL"
+141 -2
View File
@@ -11,6 +11,7 @@ use std::sync::Arc;
use std::time::Duration;
use anyhow::{Context, Result};
use music_dht::device_sync::{ListenEndReason, ListenEvent, ListenTrackMetadata};
use music_dht::{ByteStream, MusicDhtService, NetworkId, PeerTicket, SecretKey, StreamAcceptor};
use serde::{Deserialize, Serialize};
use sqlx::Row;
@@ -20,10 +21,10 @@ use crate::player::PlayerDeviceHub;
use super::{TransportStats, record_stream_transport};
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);
@@ -269,6 +270,9 @@ enum SyncOpPayload {
target_device_id: String,
command: PlaybackCommand,
},
ListenRecorded {
event: ListenEvent,
},
}
impl SyncOpPayload {
@@ -771,6 +775,41 @@ pub async fn record_track_like(
Ok(())
}
pub async fn record_track_listen(
pool: &sqlx::PgPool,
user_id: i64,
track_id: i64,
listen_id: String,
started_at_ms: i64,
listened_ms: i64,
ended_reason: ListenEndReason,
) -> Result<()> {
let content_id = track_content_id(pool, track_id)
.await?
.context("track content id is not ready")?;
let track = synced_fed_track_for_track(pool, track_id, &content_id)
.await?
.context("track metadata is not available")?;
let event = ListenEvent {
listen_id,
content_id,
started_at_ms,
listened_ms,
track_duration_ms: track.duration_seconds.map(|seconds| seconds * 1_000),
ended_reason,
track: ListenTrackMetadata {
title: track.title,
artist_names: track.artist_names,
featured_artist_names: track.featured_artist_names,
release_title: track.release_title,
},
};
if event.should_record() {
record_local_op(pool, user_id, SyncOpPayload::ListenRecorded { event }).await?;
}
Ok(())
}
/// Records a like for a content-addressed track that need not be local yet.
/// The ordinary local likes table remains a projection for materialized
/// tracks; the durable user intent lives under `content_id`.
@@ -2159,9 +2198,108 @@ async fn apply_op(
.await?;
Ok(false)
}
SyncOpPayload::ListenRecorded { event } => {
apply_listen_event(pool, user_id, &op.origin_device_id, event).await
}
}
}
async fn apply_listen_event(
pool: &sqlx::PgPool,
user_id: i64,
origin_device_id: &str,
event: &ListenEvent,
) -> 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 local_track_id: Option<i64> = sqlx::query_scalar(
"SELECT local_track_id
FROM furumusic__track_ref
WHERE content_id = $1",
)
.bind(&content_id)
.fetch_optional(pool)
.await?
.flatten();
let inserted = sqlx::query(
"INSERT INTO furumusic__listen_event
(user_id, listen_id, content_id, local_track_id, origin_device_id,
started_at_ms, listened_ms, track_duration_ms, ended_reason,
qualified, metadata_json, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
ON CONFLICT (user_id, listen_id) DO NOTHING",
)
.bind(user_id)
.bind(&event.listen_id)
.bind(&content_id)
.bind(local_track_id)
.bind(origin_device_id)
.bind(event.started_at_ms)
.bind(event.listened_ms)
.bind(event.track_duration_ms)
.bind(serde_json::to_string(&event.ended_reason)?)
.bind(event.qualifies_as_play())
.bind(serde_json::to_value(&event.track)?)
.bind(chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string())
.execute(pool)
.await?
.rows_affected()
> 0;
if !inserted || !event.qualifies_as_play() {
return Ok(inserted);
}
let Some(artist_name) = event
.track
.artist_names
.iter()
.find(|name| !name.trim().is_empty())
else {
return Ok(true);
};
let duration_seconds = event
.track_duration_ms
.unwrap_or_default()
.div_euclid(1_000)
.clamp(0, i64::from(i32::MAX)) as i32;
if duration_seconds <= 30 {
return Ok(true);
}
let listened_seconds = event
.listened_ms
.div_euclid(1_000)
.clamp(0, i64::from(i32::MAX)) as i32;
let started_at = event.started_at_ms.div_euclid(1_000);
let now = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
sqlx::query(
"INSERT INTO furumusic__lastfm_scrobble_outbox
(user_id, track_id, started_at, listened_seconds, duration_seconds,
status, created_at, updated_at, dedupe_key, track_title,
artist_name, album_title)
SELECT $1, $2, $3, $4, $5, 'pending', $6, $6, $7, $8, $9, $10
WHERE EXISTS (
SELECT 1 FROM furumusic__lastfm_account WHERE user_id = $1
)
ON CONFLICT (dedupe_key) DO NOTHING",
)
.bind(user_id)
.bind(local_track_id)
.bind(started_at)
.bind(listened_seconds)
.bind(duration_seconds)
.bind(&now)
.bind(format!("listen:{user_id}:{}", event.listen_id))
.bind(&event.track.title)
.bind(artist_name)
.bind(event.track.release_title.as_deref())
.execute(pool)
.await?;
Ok(true)
}
async fn apply_like_state(
pool: &sqlx::PgPool,
user_id: i64,
@@ -4422,6 +4560,7 @@ fn payload_kind(payload: &SyncOpPayload) -> &'static str {
SyncOpPayload::DeviceTrusted { .. } => "device_trusted",
SyncOpPayload::DeviceRevoked { .. } => "device_revoked",
SyncOpPayload::PlaybackCommand { .. } => "playback_command",
SyncOpPayload::ListenRecorded { .. } => "listen_recorded",
}
}
+100 -8
View File
@@ -229,6 +229,36 @@ impl Federation {
}
}
pub async fn prepare_discovered_content_with_progress<F>(
self: &std::sync::Arc<Self>,
content_id: &str,
progress: F,
) -> Result<PreparedTrack>
where
F: FnMut(DownloadProgress) + Send,
{
let content_id =
music_dht::normalize_content_id(content_id).context("invalid content id")?;
let service = self.service().await?;
let outcome = service.search_content_id(&content_id).await?;
let item = outcome
.local_results
.into_iter()
.chain(outcome.network_results)
.find(|item| {
item.kind == music_dht::ItemKind::Track
&& item.content_id.as_deref() == Some(content_id.as_str())
})
.context("no peer currently advertises this content id")?;
self.prepare_content_with_progress(
&content_id,
&item.owner.to_string(),
&hex_encode(item.id.as_bytes()),
progress,
)
.await
}
pub async fn prepare_content_with_progress<F>(
self: &std::sync::Arc<Self>,
content_id: &str,
@@ -295,7 +325,60 @@ impl Federation {
let dir = storage_root.join("federation");
tokio::fs::create_dir_all(&dir).await?;
let downloaded =
download(&service, owner, item_id, &content_id, &dir, &mut progress).await?;
match download(&service, owner, item_id, &content_id, &dir, &mut progress).await {
Ok(downloaded) => downloaded,
Err(primary_error) => {
progress(DownloadProgress {
phase: "discovering",
received: 0,
total: 0,
});
let outcome = service
.search_content_id(&content_id)
.await
.map_err(|err| {
anyhow::anyhow!("{primary_error:#}; content lookup also failed: {err}")
})?;
let mut last_error = primary_error;
let mut downloaded = None;
for item in outcome
.local_results
.into_iter()
.chain(outcome.network_results)
{
if item.kind != music_dht::ItemKind::Track
|| item.content_id.as_deref() != Some(content_id.as_str())
{
continue;
}
let candidate_item_id = hex_encode(item.id.as_bytes());
if item.owner == owner && candidate_item_id == item_id {
continue;
}
match download(
&service,
item.owner,
&candidate_item_id,
&content_id,
&dir,
&mut progress,
)
.await
{
Ok(candidate) => {
downloaded = Some(candidate);
break;
}
Err(err) => last_error = err,
}
}
downloaded.ok_or_else(|| {
anyhow::anyhow!(
"no reachable peer currently provides this track: {last_error:#}"
)
})?
}
};
if !save {
super::lock(&self.prepared_cache)
.insert(token.clone(), (downloaded.path, downloaded.mime));
@@ -306,15 +389,20 @@ impl Federation {
}
progress(DownloadProgress {
phase: "saving",
received: 1,
total: 1,
received: 0,
total: 0,
});
let track_id = materialize(&pool, &storage_root, &content_id, downloaded).await?;
// The normal periodic sync will publish it; this immediate sync keeps
// save-on-listen useful to the federation without waiting a minute.
if let Err(err) = self.sync_now().await {
tracing::warn!(track_id, "post-import federation publish failed: {err:#}");
}
// Materialization is the playback boundary: return the local track
// immediately so the browser can replace the pending queue entry and
// start it. Publishing must not keep the prepare stream stuck at 100%
// when a federation peer is slow or offline.
let federation = std::sync::Arc::clone(self);
tokio::spawn(async move {
if let Err(err) = federation.sync_now().await {
tracing::warn!(track_id, "post-import federation publish failed: {err:#}");
}
});
Ok(PreparedTrack {
local_track_id: Some(track_id),
stream_url: format!("/api/player/stream/{track_id}"),
@@ -362,6 +450,10 @@ fn cache_artwork(federation: &Federation, key: String, artwork: &(Vec<u8>, Strin
);
}
fn hex_encode(bytes: &[u8]) -> String {
bytes.iter().map(|byte| format!("{byte:02x}")).collect()
}
async fn download(
service: &music_dht::MusicDhtService,
owner: EndpointId,
+5 -5
View File
@@ -447,17 +447,17 @@ async fn fetch_pending_scrobbles(
o.duration_seconds,
o.attempt_count,
a.session_key::text AS session_key,
t.title::text AS title,
r.title::text AS album_title,
COALESCE(o.track_title, t.title::text) AS title,
COALESCE(o.album_title, r.title::text) AS album_title,
t.track_number,
(
COALESCE(o.artist_name, (
SELECT ar.name::text
FROM furumusic__track_artist ta
JOIN furumusic__artist ar ON ar.id = ta.artist_id
WHERE ta.track_id = t.id AND ta.role <> 'featuring'
ORDER BY ta.position
LIMIT 1
) AS artist_name,
)) AS artist_name,
(
SELECT ar.name::text
FROM furumusic__release_artist ra
@@ -468,7 +468,7 @@ async fn fetch_pending_scrobbles(
) AS album_artist_name
FROM furumusic__lastfm_scrobble_outbox o
JOIN furumusic__lastfm_account a ON a.user_id = o.user_id
JOIN furumusic__track t ON t.id = o.track_id
LEFT JOIN furumusic__track t ON t.id = o.track_id
LEFT JOIN furumusic__release r ON r.id = t.release_id
WHERE o.user_id = $1
AND o.status IN ('pending', 'retry')
+113
View File
@@ -2322,6 +2322,118 @@ pub mod db_migrations {
&[Operation::custom(create_content_addressed_music_refs).build()];
}
#[cot::db::migrations::migration_op]
async fn create_synced_listen_history(
ctx: migrations::MigrationContext<'_>,
) -> cot::db::Result<()> {
ctx.db
.raw(
"CREATE TABLE IF NOT EXISTS furumusic__listen_event (
user_id BIGINT NOT NULL,
listen_id TEXT NOT NULL,
content_id TEXT NOT NULL,
local_track_id BIGINT,
origin_device_id TEXT NOT NULL,
started_at_ms BIGINT NOT NULL,
listened_ms BIGINT NOT NULL,
track_duration_ms BIGINT,
ended_reason TEXT NOT NULL,
qualified BOOLEAN NOT NULL,
metadata_json JSONB NOT NULL,
created_at TEXT NOT NULL,
PRIMARY KEY (user_id, listen_id)
)",
)
.await?;
ctx.db
.raw(
"CREATE INDEX IF NOT EXISTS idx_listen_event_user_time
ON furumusic__listen_event (user_id, started_at_ms DESC, listen_id)",
)
.await?;
ctx.db
.raw(
"CREATE INDEX IF NOT EXISTS idx_listen_event_content
ON furumusic__listen_event (user_id, content_id)",
)
.await?;
ctx.db
.raw(
"INSERT INTO furumusic__listen_event
(user_id, listen_id, content_id, local_track_id,
origin_device_id, started_at_ms, listened_ms,
track_duration_ms, ended_reason, qualified,
metadata_json, created_at)
SELECT ph.user_id,
'legacy-web:' || ph.id::text,
tr.content_id,
ph.track_id,
ident.device_id,
(EXTRACT(EPOCH FROM ph.played_at::timestamptz) * 1000)::bigint,
COALESCE(ph.duration_listened, 0)::bigint * 1000,
(t.duration_seconds * 1000)::bigint,
CASE WHEN ph.completed THEN '\"finished\"' ELSE '\"unknown\"' END,
ph.completed,
jsonb_build_object(
'title', t.title::text,
'artist_names', COALESCE((
SELECT jsonb_agg(a.name::text ORDER BY ta.position)
FROM furumusic__track_artist ta
JOIN furumusic__artist a ON a.id = ta.artist_id
WHERE ta.track_id = t.id
AND ta.role <> 'featuring'
), '[]'::jsonb),
'featured_artist_names', COALESCE((
SELECT jsonb_agg(a.name::text ORDER BY ta.position)
FROM furumusic__track_artist ta
JOIN furumusic__artist a ON a.id = ta.artist_id
WHERE ta.track_id = t.id
AND ta.role = 'featuring'
), '[]'::jsonb),
'release_title', r.title::text
),
ph.played_at::text
FROM furumusic__play_history ph
JOIN furumusic__track t ON t.id = ph.track_id
JOIN furumusic__track_ref tr ON tr.local_track_id = ph.track_id
JOIN furumusic__fed_device_identity ident
ON ident.user_id = ph.user_id
LEFT JOIN furumusic__release r ON r.id = t.release_id
ON CONFLICT (user_id, listen_id) DO NOTHING",
)
.await?;
ctx.db
.raw(
"ALTER TABLE furumusic__lastfm_scrobble_outbox
ALTER COLUMN track_id DROP NOT NULL",
)
.await?;
ctx.db
.raw(
"ALTER TABLE furumusic__lastfm_scrobble_outbox
ADD COLUMN IF NOT EXISTS track_title TEXT,
ADD COLUMN IF NOT EXISTS artist_name TEXT,
ADD COLUMN IF NOT EXISTS album_title TEXT",
)
.await?;
Ok(())
}
#[derive(Debug, Copy, Clone)]
pub struct M0041CreateSyncedListenHistory;
impl migrations::Migration for M0041CreateSyncedListenHistory {
const APP_NAME: &'static str = "furumusic";
const MIGRATION_NAME: &'static str = "m_0041_create_synced_listen_history";
const DEPENDENCIES: &'static [migrations::MigrationDependency] =
&[migrations::MigrationDependency::migration(
"furumusic",
"m_0040_create_content_addressed_music_refs",
)];
const OPERATIONS: &'static [Operation] =
&[Operation::custom(create_synced_listen_history).build()];
}
pub const MIGRATIONS: &[&SyncDynMigration] = &[
&M0006CreateMediaFile,
&M0007CreateArtist,
@@ -2353,5 +2465,6 @@ pub mod db_migrations {
&M0038CreateFedDeviceSync,
&M0039EnsureFederationContentIdCache,
&M0040CreateContentAddressedMusicRefs,
&M0041CreateSyncedListenHistory,
];
}
+7 -5
View File
@@ -541,14 +541,16 @@ pub(super) struct UserUploadReviewUpdateRequest {
#[derive(Debug, Serialize, JsonSchema)]
pub(super) struct PlayHistoryItem {
pub(super) id: i64,
pub(super) track_id: i64,
pub(super) id: String,
pub(super) track_id: Option<i64>,
pub(super) track_title: String,
pub(super) release_title: Option<String>,
pub(super) track: TrackItem,
pub(super) track: serde_json::Value,
pub(super) played_at: String,
pub(super) duration_listened: Option<i32>,
pub(super) completed: bool,
pub(super) device_id: String,
pub(super) device_name: String,
}
#[derive(Debug, Serialize, JsonSchema)]
@@ -576,8 +578,8 @@ pub(super) struct ContentTrackMutation {
#[derive(Debug, Deserialize, JsonSchema)]
pub(super) struct PrepareFederatedTrackRequest {
pub(super) content_id: String,
pub(super) owner: String,
pub(super) item_id: String,
pub(super) owner: Option<String>,
pub(super) item_id: Option<String>,
}
#[derive(Debug, Deserialize, JsonSchema)]
+171 -190
View File
@@ -5563,139 +5563,131 @@ async fn history_list_handler(
let per_page = query.0.limit.unwrap_or(20).clamp(1, 100);
let offset = (page - 1) as i64 * per_page as i64;
let total: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM furumusic__play_history WHERE user_id = $1")
.bind(user.id)
.fetch_one(pool)
.await
.map_err(|e| cot::Error::internal(e.to_string()))?;
let rows = sqlx::query_as::<_, PlayHistoryTrackRow>(
r#"SELECT ph.id AS history_id,
ph.played_at::text AS played_at,
ph.duration_listened,
ph.completed,
t.id,
t.title::text as title,
t.track_number,
t.disc_number,
t.duration_seconds,
t.cover_file_id,
r.cover_file_id as release_cover_file_id,
t.release_id,
COALESCE(r.title::text, '') as release_title,
r.year as release_year,
COALESCE(mf.uploader_name, 'UFO')::text AS uploader_name,
mf.audio_format,
mf.audio_bitrate,
mf.audio_sample_rate,
mf.audio_bit_depth,
mf.file_size_bytes,
t.lastfm_listeners,
t.lastfm_playcount,
t.lastfm_rating,
t.lastfm_updated_at
FROM furumusic__play_history ph
JOIN furumusic__track t ON t.id = ph.track_id
LEFT JOIN furumusic__release r ON r.id = t.release_id
LEFT JOIN furumusic__media_file mf ON mf.id = t.audio_file_id
WHERE ph.user_id = $1
ORDER BY ph.played_at DESC, ph.id DESC
LIMIT $2 OFFSET $3"#,
let total: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM furumusic__listen_event
WHERE user_id = $1 AND qualified = true",
)
.bind(user.id)
.bind(per_page as i64)
.fetch_one(pool)
.await
.map_err(|e| cot::Error::internal(e.to_string()))?;
let rows = sqlx::query(
"SELECT le.listen_id, le.content_id, le.local_track_id,
le.origin_device_id, le.started_at_ms, le.listened_ms,
le.track_duration_ms, le.metadata_json,
COALESCE(NULLIF(d.name, ''), le.origin_device_id) AS device_name,
t.release_id, r.year AS release_year, t.cover_file_id,
r.cover_file_id AS release_cover_file_id
FROM furumusic__listen_event le
LEFT JOIN furumusic__fed_device d
ON d.user_id = le.user_id AND d.device_id = le.origin_device_id
LEFT JOIN furumusic__track t ON t.id = le.local_track_id
LEFT JOIN furumusic__release r ON r.id = t.release_id
WHERE le.user_id = $1 AND le.qualified = true
ORDER BY le.started_at_ms DESC, le.listen_id DESC
LIMIT $2 OFFSET $3",
)
.bind(user.id)
.bind(i64::from(per_page))
.bind(offset)
.fetch_all(pool)
.await
.map_err(|e| cot::Error::internal(e.to_string()))?;
let track_ids: Vec<i64> = rows.iter().map(|t| t.id).collect();
let track_artists = if track_ids.is_empty() {
Vec::new()
} else {
sqlx::query_as::<_, TrackArtistRow>(
r#"SELECT ta.track_id, ta.artist_id, a.name::text as artist_name, ta.role::text as role
FROM furumusic__track_artist ta
JOIN furumusic__artist a ON a.id = ta.artist_id
WHERE ta.track_id = ANY($1)
ORDER BY ta.track_id, ta.position"#,
)
.bind(&track_ids)
.fetch_all(pool)
.await
.map_err(|e| cot::Error::internal(e.to_string()))?
};
let mut track_main_artists: HashMap<i64, Vec<ArtistRef>> = HashMap::new();
let mut track_feat_artists: HashMap<i64, Vec<ArtistRef>> = HashMap::new();
for ta in &track_artists {
let artist_ref = ArtistRef {
id: ta.artist_id,
name: ta.artist_name.clone(),
};
if ta.role == "featuring" {
track_feat_artists
.entry(ta.track_id)
.or_default()
.push(artist_ref);
} else {
track_main_artists
.entry(ta.track_id)
.or_default()
.push(artist_ref);
}
}
let items = rows
.into_iter()
.map(|row| {
let listen_id: String = row.get("listen_id");
let content_id: String = row.get("content_id");
let local_track_id: Option<i64> = row.get("local_track_id");
let metadata: serde_json::Value = row.get("metadata_json");
let title = metadata
.get("title")
.and_then(serde_json::Value::as_str)
.unwrap_or("Unknown track")
.to_string();
let release_title = metadata
.get("release_title")
.and_then(serde_json::Value::as_str)
.map(ToOwned::to_owned);
let artist_values = |key: &str| {
metadata
.get(key)
.and_then(serde_json::Value::as_array)
.into_iter()
.flatten()
.filter_map(serde_json::Value::as_str)
.map(|name| serde_json::json!({"id": null, "name": name}))
.collect::<Vec<_>>()
};
let duration_ms: Option<i64> = row.get("track_duration_ms");
let release_id: Option<i64> = row.get("release_id");
let cover_url = track_cover_variant_url(
row.get("cover_file_id"),
row.get("release_cover_file_id"),
"medium",
);
let track = serde_json::json!({
"id": local_track_id.map_or_else(
|| format!("history:{content_id}"),
|id| id.to_string(),
),
"content_id": content_id,
"title": title,
"track_number": null,
"disc_number": null,
"duration_seconds": duration_ms.unwrap_or_default() as f64 / 1000.0,
"artists": artist_values("artist_names"),
"featured_artists": artist_values("featured_artist_names"),
"release_id": release_id,
"release_title": release_title.clone().unwrap_or_default(),
"release_year": row.get::<Option<i32>, _>("release_year"),
"cover_url": cover_url,
"stream_url": local_track_id
.map(|id| format!("/api/player/stream/{id}"))
.unwrap_or_default(),
"uploader_name": if local_track_id.is_some() { "Web" } else { "Federation" },
"federation_pending": local_track_id.is_none(),
"_federationTrack": if local_track_id.is_none() {
Some(serde_json::json!({
"key": {"content_id": content_id},
"metadata": metadata,
"availability": {
"state": "federated",
"local": null,
"federation": [{"owner": "", "item_id": ""}],
},
}))
} else {
None
},
});
let started_at_ms: i64 = row.get("started_at_ms");
let played_at = chrono::DateTime::from_timestamp_millis(started_at_ms)
.unwrap_or_else(chrono::Utc::now)
.to_rfc3339();
PlayHistoryItem {
id: listen_id,
track_id: local_track_id,
track_title: title,
release_title,
track,
played_at,
duration_listened: Some(
row.get::<i64, _>("listened_ms")
.div_euclid(1_000)
.clamp(0, i64::from(i32::MAX)) as i32,
),
completed: true,
device_id: row.get("origin_device_id"),
device_name: row.get("device_name"),
}
})
.collect();
Json(PlayHistoryPage {
items: rows
.into_iter()
.map(|row| PlayHistoryItem {
id: row.history_id,
track_id: row.id,
track_title: row.title.clone(),
release_title: if row.release_title.trim().is_empty() {
None
} else {
Some(row.release_title.clone())
},
track: {
let tid = row.id;
TrackItem {
id: row.id,
content_id: None,
title: row.title,
track_number: row.track_number,
disc_number: row.disc_number,
duration_seconds: row.duration_seconds,
artists: track_main_artists.remove(&tid).unwrap_or_default(),
featured_artists: track_feat_artists.remove(&tid).unwrap_or_default(),
release_id: row.release_id,
release_title: row.release_title,
release_year: row.release_year,
cover_url: track_cover_variant_url(
row.cover_file_id,
row.release_cover_file_id,
"medium",
),
stream_url: format!("/api/player/stream/{tid}"),
uploader_name: row.uploader_name,
audio_format: row.audio_format,
audio_bitrate: row.audio_bitrate,
audio_sample_rate: row.audio_sample_rate,
audio_bit_depth: row.audio_bit_depth,
file_size_bytes: row.file_size_bytes,
lastfm_listeners: row.lastfm_listeners,
lastfm_playcount: row.lastfm_playcount,
lastfm_rating: row.lastfm_rating,
lastfm_updated_at: row.lastfm_updated_at,
}
},
played_at: row.played_at,
duration_listened: row.duration_listened,
completed: row.completed,
})
.collect(),
items,
total,
page,
per_page,
@@ -5714,61 +5706,33 @@ async fn history_handler(
return Ok(json_error(StatusCode::UNAUTHORIZED, "not authenticated"));
};
let now = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
sqlx::query(
r#"INSERT INTO furumusic__play_history (user_id, track_id, played_at, duration_listened, completed)
VALUES ($1, $2, $3, $4, $5)"#,
let listened_seconds = entry.duration_listened.unwrap_or_default().max(0);
let started_at = entry
.started_at
.unwrap_or_else(|| chrono::Utc::now().timestamp() - i64::from(listened_seconds));
let listen_id = entry
.listen_id
.unwrap_or_else(|| format!("legacy-web:{}:{}:{started_at}", user.id, entry.track_id));
let ended_reason = match entry.ended_reason.as_deref() {
Some("finished") => music_dht::device_sync::ListenEndReason::Finished,
Some("skipped") => music_dht::device_sync::ListenEndReason::Skipped,
Some("replaced") => music_dht::device_sync::ListenEndReason::Replaced,
Some("stopped") => music_dht::device_sync::ListenEndReason::Stopped,
_ => music_dht::device_sync::ListenEndReason::Unknown,
};
crate::federation::devices::record_track_listen(
pool,
user.id,
entry.track_id,
listen_id,
started_at.saturating_mul(1_000),
i64::from(listened_seconds).saturating_mul(1_000),
ended_reason,
)
.bind(user.id)
.bind(entry.track_id)
.bind(&now)
.bind(entry.duration_listened)
.bind(entry.completed)
.execute(pool)
.await
.map_err(|e| cot::Error::internal(e.to_string()))?;
crate::metrics::record_play_history(entry.duration_listened, entry.completed);
if let Some(listened_seconds) = entry.duration_listened {
let (config, _) = AppConfig::load_with_db(&db).await;
match enqueue_lastfm_scrobble(
pool,
&config,
user.id,
entry.track_id,
entry.started_at,
listened_seconds,
)
.await
{
Ok(result) if result.queued => {
tracing::info!(
user_id = user.id,
track_id = entry.track_id,
sent = result.sent,
"Queued Last.fm scrobble from play history"
);
}
Ok(result) => {
tracing::debug!(
user_id = user.id,
track_id = entry.track_id,
message = ?result.message,
"Play history did not queue Last.fm scrobble"
);
}
Err(err) => {
tracing::warn!(
user_id = user.id,
track_id = entry.track_id,
error = %err,
"Failed to queue Last.fm scrobble from play history"
);
}
}
}
Json(serde_json::json!({"ok": true})).into_response()
}
@@ -6397,23 +6361,40 @@ async fn prepare_federated_track_handler(
return Ok(json_error(StatusCode::UNAUTHORIZED, "not authenticated"));
};
let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel::<serde_json::Value>();
let progress_sender = sender.clone();
tokio::spawn(async move {
let result = crate::federation::handle()
.prepare_content_with_progress(
&body.content_id,
&body.owner,
&body.item_id,
move |progress| {
let _ = progress_sender.send(serde_json::json!({
"kind": "progress",
"phase": progress.phase,
"received": progress.received,
"total": progress.total,
}));
},
)
.await;
let federation = crate::federation::handle();
let progress_sender = sender.clone();
let result = match (body.owner.as_deref(), body.item_id.as_deref()) {
(Some(owner), Some(item_id)) if !owner.is_empty() && !item_id.is_empty() => {
federation
.prepare_content_with_progress(
&body.content_id,
owner,
item_id,
move |progress| {
let _ = progress_sender.send(serde_json::json!({
"kind": "progress",
"phase": progress.phase,
"received": progress.received,
"total": progress.total,
}));
},
)
.await
}
_ => {
federation
.prepare_discovered_content_with_progress(&body.content_id, move |progress| {
let _ = progress_sender.send(serde_json::json!({
"kind": "progress",
"phase": progress.phase,
"received": progress.received,
"total": progress.total,
}));
})
.await
}
};
let event = match result {
Ok(prepared) => serde_json::json!({
"kind": "completed",
+2
View File
@@ -3,9 +3,11 @@ use serde::Deserialize;
#[derive(Debug, Deserialize)]
pub(super) struct HistoryEntry {
pub(super) track_id: i64,
pub(super) listen_id: Option<String>,
pub(super) started_at: Option<i64>,
pub(super) duration_listened: Option<i32>,
pub(super) completed: bool,
pub(super) ended_reason: Option<String>,
}
#[derive(Debug, Deserialize)]
-28
View File
@@ -254,34 +254,6 @@ pub(super) struct ReleaseUploaderRow {
pub(super) track_count: i64,
}
#[derive(sqlx::FromRow)]
pub(super) struct PlayHistoryTrackRow {
pub(super) history_id: i64,
pub(super) played_at: String,
pub(super) duration_listened: Option<i32>,
pub(super) completed: bool,
pub(super) id: i64,
pub(super) title: String,
pub(super) track_number: Option<i32>,
pub(super) disc_number: Option<i32>,
pub(super) duration_seconds: f64,
pub(super) cover_file_id: Option<i64>,
pub(super) release_cover_file_id: Option<i64>,
pub(super) release_id: i64,
pub(super) release_title: String,
pub(super) release_year: Option<i32>,
pub(super) uploader_name: String,
pub(super) audio_format: Option<String>,
pub(super) audio_bitrate: Option<i32>,
pub(super) audio_sample_rate: Option<i32>,
pub(super) audio_bit_depth: Option<i32>,
pub(super) file_size_bytes: Option<i64>,
pub(super) lastfm_listeners: Option<i64>,
pub(super) lastfm_playcount: Option<i64>,
pub(super) lastfm_rating: Option<f64>,
pub(super) lastfm_updated_at: Option<String>,
}
#[derive(sqlx::FromRow)]
pub(super) struct ReleaseInfoRow {
pub(super) id: i64,
+6 -1
View File
@@ -821,7 +821,12 @@
</template>
</button>
<div class="track-info">
<div class="track-title" x-text="item.track?.title || item.track_title"></div>
<div class="track-title">
<span x-text="item.track?.title || item.track_title"></span>
<span class="history-device-badge"
:title="item.device_id"
x-text="item.device_name"></span>
</div>
<div class="track-artists-inline">
<template x-for="(artist, artistIdx) in $store.library.trackArtistLinks(item.track)" :key="artist.label + '-' + artist.id + '-' + artistIdx">
<span>
+27 -4
View File
@@ -726,6 +726,7 @@ document.addEventListener('alpine:init', () => {
_historyRecorded: false,
_nowPlayingSent: false,
_playbackStartedAt: null,
_listenId: null,
_listenedSeconds: 0,
_lastAudioTime: 0,
_localSourceTrackId: null,
@@ -754,7 +755,7 @@ document.addEventListener('alpine:init', () => {
audio.addEventListener('ended', () => {
this._trackListenedDelta();
this._recordHistory(true);
this._recordHistory('finished');
if (Alpine.store('devices')?.shouldPlayJamLocally()) {
this.isPlaying = false;
return;
@@ -1437,21 +1438,24 @@ document.addEventListener('alpine:init', () => {
} catch {}
},
_recordHistory(completed) {
_recordHistory(endedReason) {
if (this._historyRecorded || !this.currentTrack) return;
if (!Number.isInteger(Number(this.currentTrack.id)) || this.currentTrack.federated_cache) {
return;
}
this._historyRecorded = true;
const completed = endedReason === 'finished';
const listenedSeconds = this._historyListenedSeconds(completed);
fetch('/api/player/history', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
track_id: this.currentTrack.id,
listen_id: this._listenId,
started_at: this._playbackStartedAt,
duration_listened: listenedSeconds,
completed: completed,
ended_reason: endedReason,
}),
}).catch(() => {});
},
@@ -1464,13 +1468,15 @@ document.addEventListener('alpine:init', () => {
const listened = Math.floor(Number(this._listenedSeconds || 0));
const threshold = Math.ceil(duration / 2);
if (threshold <= 0 || listened < threshold) return false;
this._recordHistory(true);
this._recordHistory('unknown');
return true;
},
_resetPlaybackTracking() {
this._nowPlayingSent = false;
this._playbackStartedAt = null;
this._listenId = (globalThis.crypto?.randomUUID?.()
|| `${Date.now()}-${Math.random().toString(16).slice(2)}`);
this._listenedSeconds = 0;
this._lastAudioTime = 0;
},
@@ -2495,6 +2501,7 @@ document.addEventListener('alpine:init', () => {
federationSearch: { loading: false, error: '', artists: [], releases: [], tracks: [] },
artistFederation: { loading: false, error: '', releases: [], tracks: [] },
federationPreparing: {},
federationFailures: {},
trackAvailability: {},
_federatedReleaseCache: {},
federationLiked: [],
@@ -3863,6 +3870,11 @@ document.addEventListener('alpine:init', () => {
return;
}
if (!contentId || sources.length === 0) return;
if (this.federationFailures[contentId]) {
const failures = { ...this.federationFailures };
delete failures[contentId];
this.federationFailures = failures;
}
if (this.federationPreparing[contentId]) {
if (options.playAfterPrepare !== false) {
this.federationPreparing = {
@@ -4023,7 +4035,13 @@ document.addEventListener('alpine:init', () => {
Alpine.store('queue').playRelease([playable], 0);
}
} catch (error) {
window.alert(error?.message || 'Federated track download failed');
this.federationFailures = {
...this.federationFailures,
[contentId]: {
message: error?.message || 'Federated track is temporarily unavailable',
failedAt: Date.now(),
},
};
} finally {
const next = { ...this.federationPreparing };
delete next[contentId];
@@ -4035,12 +4053,17 @@ document.addEventListener('alpine:init', () => {
return this.federationPreparing[contentId] || null;
},
federationFailure(contentId) {
return this.federationFailures[contentId] || null;
},
federationDownloadTooltip(contentId) {
const progress = this.federationDownload(contentId);
if (!progress) return '';
const phases = {
checking: 'Checking local library',
connecting: 'Connecting to peer',
discovering: 'Looking for another peer',
downloading: 'Downloading',
verifying: 'Verifying content ID',
saving: 'Adding to library',
+25 -5
View File
@@ -562,13 +562,17 @@
<div class="track-row federation-track-row"
@dblclick="$store.library.playFederatedTrack(track)">
<span class="track-num federation-track-status">
<template x-if="!$store.library.federationDownload(track.key.content_id) && !$store.library.isTrackLocal(track)">
<template x-if="!$store.library.federationDownload(track.key.content_id) && !$store.library.federationFailure(track.key.content_id) && !$store.library.isTrackLocal(track)">
<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="1.8">
<circle cx="12" cy="12" r="3"/>
<path d="M5.6 8.5a7.5 7.5 0 000 7M18.4 8.5a7.5 7.5 0 010 7"/>
<path d="M2.8 5.8a11 11 0 000 12.4M21.2 5.8a11 11 0 010 12.4"/>
</svg>
</template>
<template x-if="$store.library.federationFailure(track.key.content_id) && !$store.library.isTrackLocal(track)">
<span class="federation-error-badge"
:title="$store.library.federationFailure(track.key.content_id).message">!</span>
</template>
<template x-if="$store.library.federationDownload(track.key.content_id)">
<span class="federation-download-progress"
:class="{ indeterminate: !$store.library.federationDownload(track.key.content_id).total }">
@@ -905,12 +909,16 @@
<div class="track-row federation-track-row"
@dblclick="$store.library.playFederatedTrack(track)">
<span class="track-num federation-track-status">
<template x-if="!$store.library.federationDownload(track.key.content_id) && !$store.library.isTrackLocal(track)">
<template x-if="!$store.library.federationDownload(track.key.content_id) && !$store.library.federationFailure(track.key.content_id) && !$store.library.isTrackLocal(track)">
<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="1.8">
<circle cx="12" cy="12" r="3"/>
<path d="M5.6 8.5a7.5 7.5 0 000 7M18.4 8.5a7.5 7.5 0 010 7"/>
</svg>
</template>
<template x-if="$store.library.federationFailure(track.key.content_id) && !$store.library.isTrackLocal(track)">
<span class="federation-error-badge"
:title="$store.library.federationFailure(track.key.content_id).message">!</span>
</template>
<template x-if="$store.library.federationDownload(track.key.content_id)">
<span class="federation-download-progress"
:class="{ indeterminate: !$store.library.federationDownload(track.key.content_id).total }">
@@ -1158,13 +1166,17 @@
<template x-if="!track.federation_pending">
<span x-text="track.track_number || (idx + 1)"></span>
</template>
<template x-if="track.federation_pending && !$store.library.federationDownload(track.content_id) && !$store.library.isTrackLocal(track._federationTrack)">
<template x-if="track.federation_pending && !$store.library.federationDownload(track.content_id) && !$store.library.federationFailure(track.content_id) && !$store.library.isTrackLocal(track._federationTrack)">
<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="1.8">
<circle cx="12" cy="12" r="3"/>
<path d="M5.6 8.5a7.5 7.5 0 000 7M18.4 8.5a7.5 7.5 0 010 7"/>
<path d="M2.8 5.8a11 11 0 000 12.4M21.2 5.8a11 11 0 010 12.4"/>
</svg>
</template>
<template x-if="track.federation_pending && $store.library.federationFailure(track.content_id) && !$store.library.isTrackLocal(track._federationTrack)">
<span class="federation-error-badge"
:title="$store.library.federationFailure(track.content_id).message">!</span>
</template>
<template x-if="track.federation_pending && !$store.library.federationDownload(track.content_id) && $store.library.isTrackLocal(track._federationTrack)">
<span x-text="track.track_number || (idx + 1)"></span>
</template>
@@ -1312,13 +1324,17 @@
<template x-if="!track.federation_pending || $store.library.isTrackLocal(track._federationTrack)">
<span x-text="idx + 1"></span>
</template>
<template x-if="track.federation_pending && !$store.library.federationDownload(track.content_id) && !$store.library.isTrackLocal(track._federationTrack)">
<template x-if="track.federation_pending && !$store.library.federationDownload(track.content_id) && !$store.library.federationFailure(track.content_id) && !$store.library.isTrackLocal(track._federationTrack)">
<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="1.8">
<circle cx="12" cy="12" r="3"/>
<path d="M5.6 8.5a7.5 7.5 0 000 7M18.4 8.5a7.5 7.5 0 010 7"/>
<path d="M2.8 5.8a11 11 0 000 12.4M21.2 5.8a11 11 0 010 12.4"/>
</svg>
</template>
<template x-if="track.federation_pending && $store.library.federationFailure(track.content_id) && !$store.library.isTrackLocal(track._federationTrack)">
<span class="federation-error-badge"
:title="$store.library.federationFailure(track.content_id).message">!</span>
</template>
<template x-if="track.federation_pending && $store.library.federationDownload(track.content_id)">
<span class="federation-download-progress"
:class="{ indeterminate: !$store.library.federationDownload(track.content_id).total }">
@@ -1452,12 +1468,16 @@
<span class="queue-federation-status federation-track-status"
x-show="item.track.federation_pending"
x-cloak>
<template x-if="!$store.library.federationDownload(item.track.content_id)">
<template x-if="!$store.library.federationDownload(item.track.content_id) && !$store.library.federationFailure(item.track.content_id)">
<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="1.8">
<circle cx="12" cy="12" r="3"/>
<path d="M5.6 8.5a7.5 7.5 0 000 7M18.4 8.5a7.5 7.5 0 010 7"/>
</svg>
</template>
<template x-if="$store.library.federationFailure(item.track.content_id)">
<span class="federation-error-badge"
:title="$store.library.federationFailure(item.track.content_id).message">!</span>
</template>
<template x-if="$store.library.federationDownload(item.track.content_id)">
<span class="federation-download-progress"
:class="{ indeterminate: !$store.library.federationDownload(item.track.content_id).total }">
+32
View File
@@ -2729,6 +2729,26 @@ button.user-stat:hover {
height: 18px;
color: var(--accent);
}
.federation-error-badge {
display: inline-flex;
align-items: center;
justify-content: center;
width: 18px;
height: 18px;
border: 1px solid rgba(235, 105, 105, .7);
border-radius: 999px;
background: rgba(190, 55, 55, .18);
color: #ffb0b0;
font-size: 11px;
font-weight: 850;
line-height: 1;
cursor: help;
}
.queue-federation-status .federation-error-badge {
width: 16px;
height: 16px;
font-size: 10px;
}
.federation-download-progress {
position: relative;
display: inline-flex;
@@ -3681,6 +3701,18 @@ button.user-stat:hover {
}
.history-row:last-child { border-bottom: 0; }
.history-device-badge {
display: inline-block;
margin-left: .45rem;
padding: .08rem .38rem;
border: 1px solid color-mix(in srgb, currentColor 28%, transparent);
border-radius: 999px;
color: var(--text-muted);
font-size: .68rem;
font-weight: 500;
line-height: 1.2;
vertical-align: .12rem;
}
.history-cover {
width: 40px;