From 232c171aac358d79a61cdc9f890a34c276a5c95b Mon Sep 17 00:00:00 2001 From: Ultradesu Date: Mon, 27 Jul 2026 23:16:03 +0100 Subject: [PATCH] Add synced listening history --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/federation/devices.rs | 143 +++++++++++++- src/federation/receive.rs | 30 +++ src/lastfm.rs | 10 +- src/music/mod.rs | 113 +++++++++++ src/player/dto.rs | 12 +- src/player/mod.rs | 361 ++++++++++++++++------------------ src/player/queries.rs | 2 + src/player/rows.rs | 28 --- templates/player/modals.html | 7 +- templates/player/scripts.html | 12 +- templates/player/styles.html | 12 ++ 13 files changed, 498 insertions(+), 236 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ee8bc1e..e2ad50b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1845,7 +1845,7 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" [[package]] name = "furumusic" -version = "0.9.3" +version = "0.9.4" dependencies = [ "anyhow", "async-stream", diff --git a/Cargo.toml b/Cargo.toml index afc0a90..16b6638 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "furumusic" -version = "0.9.3" +version = "0.9.4" edition = "2024" description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL" diff --git a/src/federation/devices.rs b/src/federation/devices.rs index 15c3e58..941ad7d 100644 --- a/src/federation/devices.rs +++ b/src/federation/devices.rs @@ -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 { + 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 = 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", } } diff --git a/src/federation/receive.rs b/src/federation/receive.rs index a72444a..c05c507 100644 --- a/src/federation/receive.rs +++ b/src/federation/receive.rs @@ -229,6 +229,36 @@ impl Federation { } } + pub async fn prepare_discovered_content_with_progress( + self: &std::sync::Arc, + content_id: &str, + progress: F, + ) -> Result + 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( self: &std::sync::Arc, content_id: &str, diff --git a/src/lastfm.rs b/src/lastfm.rs index 0316728..23b3234 100644 --- a/src/lastfm.rs +++ b/src/lastfm.rs @@ -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') diff --git a/src/music/mod.rs b/src/music/mod.rs index 9b44fe4..880b0a9 100644 --- a/src/music/mod.rs +++ b/src/music/mod.rs @@ -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, ]; } diff --git a/src/player/dto.rs b/src/player/dto.rs index a145d1e..15c9dc6 100644 --- a/src/player/dto.rs +++ b/src/player/dto.rs @@ -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, pub(super) track_title: String, pub(super) release_title: Option, - pub(super) track: TrackItem, + pub(super) track: serde_json::Value, pub(super) played_at: String, pub(super) duration_listened: Option, 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, + pub(super) item_id: Option, } #[derive(Debug, Deserialize, JsonSchema)] diff --git a/src/player/mod.rs b/src/player/mod.rs index 4e7266a..fb7d4a7 100644 --- a/src/player/mod.rs +++ b/src/player/mod.rs @@ -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 = 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> = HashMap::new(); - let mut track_feat_artists: HashMap> = 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 = 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::>() + }; + let duration_ms: Option = row.get("track_duration_ms"); + let release_id: Option = 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::, _>("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::("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::(); - 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", diff --git a/src/player/queries.rs b/src/player/queries.rs index 268e960..a80edf2 100644 --- a/src/player/queries.rs +++ b/src/player/queries.rs @@ -3,9 +3,11 @@ use serde::Deserialize; #[derive(Debug, Deserialize)] pub(super) struct HistoryEntry { pub(super) track_id: i64, + pub(super) listen_id: Option, pub(super) started_at: Option, pub(super) duration_listened: Option, pub(super) completed: bool, + pub(super) ended_reason: Option, } #[derive(Debug, Deserialize)] diff --git a/src/player/rows.rs b/src/player/rows.rs index 4496b7c..ad453ba 100644 --- a/src/player/rows.rs +++ b/src/player/rows.rs @@ -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, - pub(super) completed: bool, - pub(super) id: i64, - pub(super) title: String, - pub(super) track_number: Option, - pub(super) disc_number: Option, - pub(super) duration_seconds: f64, - pub(super) cover_file_id: Option, - pub(super) release_cover_file_id: Option, - pub(super) release_id: i64, - pub(super) release_title: String, - pub(super) release_year: Option, - pub(super) uploader_name: String, - pub(super) audio_format: Option, - pub(super) audio_bitrate: Option, - pub(super) audio_sample_rate: Option, - pub(super) audio_bit_depth: Option, - pub(super) file_size_bytes: Option, - pub(super) lastfm_listeners: Option, - pub(super) lastfm_playcount: Option, - pub(super) lastfm_rating: Option, - pub(super) lastfm_updated_at: Option, -} - #[derive(sqlx::FromRow)] pub(super) struct ReleaseInfoRow { pub(super) id: i64, diff --git a/templates/player/modals.html b/templates/player/modals.html index a4efdab..f691662 100644 --- a/templates/player/modals.html +++ b/templates/player/modals.html @@ -821,7 +821,12 @@
-
+
+ + +