feat(qbzd): scrobble-on-play engine — Last.fm / ListenBrainz (CONSOLE)
The daemon now scrobbles what it plays. A CoreEvent-bus subscriber (scrobble_engine, spawned in run() next to queue-persist): - on TrackStarted -> "now playing" to each active provider (LastFm update_now_playing / ListenBrainz submit_playing_now); - scrobbles ONCE when the position crosses the threshold from qbz_app::scrobble_timing::scrobble_delay_secs (Last.fm's half-or-4-min rule) -> LastFm scrobble / ListenBrainz submit_listen, best-effort + logged; - re-reads the canonical ScrobblerSettingsStore on each track, so `qbzd scrobble login/enable/disable` take effect on the next track with no reload signal; - holds no Arc<AppRuntime>; aborted on shutdown alongside queue-persist. Album is dropped when empty / "Unknown Album". Pure helpers (due, album_opt) are unit-tested. The ListenBrainz offline queue (ListenBrainzCache) is a follow-up — this slice submits best-effort. Verified: cargo test -p qbzd --bin qbzd -> 278 passed, 0 failed. cc
This commit is contained in:
3 files changed
+207
-1
No files matched your search
@@ -112,6 +112,13 @@ pub async fn run(roots: ProfileRoots, cfg: QbzdConfig, warns: Vec<String>) -> Re
|
||||
// clone, so it is aborted+joined ahead of `drop(booted)` (#521 ordering).
|
||||
let queue_persist = spawn_queue_persist(booted.runtime.clone(), booted.bus.subscribe());
|
||||
|
||||
// 10c. Scrobble-on-play (CONSOLE): a CoreEvent-bus subscriber that sends
|
||||
// "now playing" on TrackStarted and scrobbles once past the Last.fm
|
||||
// threshold, to whichever of Last.fm / ListenBrainz is connected +
|
||||
// enabled in the scrobbler store. Holds NO Arc<AppRuntime>, so it sits
|
||||
// outside the #521/§8.2 ordering — aborted for a clean shutdown below.
|
||||
let scrobbler = crate::scrobble_engine::spawn(roots.clone(), booted.bus.subscribe());
|
||||
|
||||
// 11. HTTP serve (02 §3) on the already-bound socket. `ApiState` carries a
|
||||
// second read-only audio-store connection (WAL) for the status audio
|
||||
// block, the tokio handle for the async queue read, and the opt-in
|
||||
@@ -187,6 +194,9 @@ pub async fn run(roots: ProfileRoots, cfg: QbzdConfig, warns: Vec<String>) -> Re
|
||||
// `Arc<AppRuntime>` clone alive past `drop(booted)` (#521 ordering).
|
||||
queue_persist.abort();
|
||||
let _ = queue_persist.await;
|
||||
// Stop the scrobble-on-play subscriber (holds no Arc<AppRuntime>; order-free).
|
||||
scrobbler.abort();
|
||||
let _ = scrobbler.await;
|
||||
// Final full session save (queue + position) now that playback is quiesced.
|
||||
playback_driver::save_session_now(booted.runtime.as_ref()).await;
|
||||
// The background auth-retry task also holds an Arc<AppRuntime> clone — abort
|
||||
|
||||
@@ -9,7 +9,7 @@ mod lock;
|
||||
mod login;
|
||||
mod paths;
|
||||
mod qconnect;
|
||||
mod scrobbler;
|
||||
mod scrobble_engine;
|
||||
mod state;
|
||||
mod tui;
|
||||
|
||||
|
||||
@@ -0,0 +1,196 @@
|
||||
// crates/qbzd/src/scrobble_engine.rs — scrobble-on-play (CONSOLE ext).
|
||||
//
|
||||
// A daemon background task subscribing the DaemonAdapter CoreEvent bus. On
|
||||
// `TrackStarted` it sends "now playing" to every ACTIVE provider; when the
|
||||
// track crosses the scrobble threshold (`qbz_app::scrobble_timing::
|
||||
// scrobble_delay_secs` — Last.fm's played-half-or-4-min rule) it scrobbles
|
||||
// ONCE. Credentials are re-read from the canonical `ScrobblerSettingsStore` on
|
||||
// each track start, so `qbzd scrobble …` changes take effect on the next track
|
||||
// with no reload signal. Best-effort + logged.
|
||||
//
|
||||
// Providers: Last.fm (LastFmClient::update_now_playing / scrobble) and
|
||||
// ListenBrainz (submit_playing_now / submit_listen). Both backends are
|
||||
// qbz-integrations (Slint-free). A persistent ListenBrainz offline queue
|
||||
// (ListenBrainzCache) is a follow-up; this slice submits best-effort.
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
use qbz_app::settings::scrobblers::{ScrobblerSettings, ScrobblerSettingsStore};
|
||||
use qbz_integrations::lastfm::LastFmClient;
|
||||
use qbz_integrations::listenbrainz::{ListenBrainzClient, ListenBrainzConfig};
|
||||
use qbz_models::{CoreEvent, QueueTrack};
|
||||
use tokio::sync::broadcast;
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
use crate::paths::ProfileRoots;
|
||||
|
||||
/// The track currently being timed for a scrobble.
|
||||
struct Playing {
|
||||
track: QueueTrack,
|
||||
/// Unix seconds when it started — Last.fm's scrobble timestamp.
|
||||
started_at: u64,
|
||||
/// Seconds into the track at which it becomes scrobble-eligible; `None`
|
||||
/// means "too short to scrobble" (`scrobble_delay_secs` returned None).
|
||||
threshold: Option<u64>,
|
||||
scrobbled: bool,
|
||||
}
|
||||
|
||||
/// Spawn the scrobble-on-play task. Holds NO `Arc<AppRuntime>` (only the roots,
|
||||
/// its own store, and the bus receiver), so it is outside the §8.2 audio
|
||||
/// clock-release ordering — the caller aborts it for a clean shutdown.
|
||||
pub fn spawn(roots: ProfileRoots, mut rx: broadcast::Receiver<CoreEvent>) -> JoinHandle<()> {
|
||||
use broadcast::error::RecvError;
|
||||
tokio::spawn(async move {
|
||||
let store = match ScrobblerSettingsStore::new_at(&roots.data) {
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
log::warn!("[scrobbler] store open failed; scrobbling disabled: {e}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let mut playing: Option<Playing> = None;
|
||||
|
||||
loop {
|
||||
match rx.recv().await {
|
||||
Ok(CoreEvent::TrackStarted { track, .. }) => {
|
||||
let settings = store.get_settings().unwrap_or_default();
|
||||
if !settings.enabled {
|
||||
playing = None;
|
||||
continue;
|
||||
}
|
||||
now_playing(&settings, &track).await;
|
||||
playing = Some(Playing {
|
||||
threshold: qbz_app::scrobble_timing::scrobble_delay_secs(track.duration_secs),
|
||||
started_at: now_unix(),
|
||||
track,
|
||||
scrobbled: false,
|
||||
});
|
||||
}
|
||||
Ok(CoreEvent::PositionUpdated { position_secs, .. }) => {
|
||||
if let Some(p) = playing.as_mut() {
|
||||
if due(position_secs, p.threshold, p.scrobbled) {
|
||||
let settings = store.get_settings().unwrap_or_default();
|
||||
scrobble(&settings, &p.track, p.started_at).await;
|
||||
p.scrobbled = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(RecvError::Lagged(_)) => continue,
|
||||
Err(RecvError::Closed) => return,
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// ============================ internals ============================
|
||||
|
||||
/// Whether the current track is due to scrobble now: it has a threshold, has
|
||||
/// been played to it, and hasn't been scrobbled yet. Pure — unit-tested.
|
||||
fn due(position_secs: u64, threshold: Option<u64>, scrobbled: bool) -> bool {
|
||||
!scrobbled && threshold.is_some_and(|t| position_secs >= t)
|
||||
}
|
||||
|
||||
async fn now_playing(s: &ScrobblerSettings, t: &QueueTrack) {
|
||||
let album = album_opt(t);
|
||||
if s.lastfm_active() {
|
||||
let c = LastFmClient::with_session_key(s.lastfm_session_key.clone());
|
||||
if let Err(e) = c.update_now_playing(&t.artist, &t.title, album).await {
|
||||
log::debug!("[scrobbler] last.fm now-playing failed: {e}");
|
||||
}
|
||||
}
|
||||
if s.listenbrainz_active() {
|
||||
let c = lb_client(s);
|
||||
if let Err(e) = c.submit_playing_now(&t.artist, &t.title, album, None).await {
|
||||
log::debug!("[scrobbler] listenbrainz now-playing failed: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn scrobble(s: &ScrobblerSettings, t: &QueueTrack, started_at: u64) {
|
||||
let album = album_opt(t);
|
||||
if s.lastfm_active() {
|
||||
let c = LastFmClient::with_session_key(s.lastfm_session_key.clone());
|
||||
match c.scrobble(&t.artist, &t.title, album, started_at).await {
|
||||
Ok(()) => log::info!("[scrobbler] last.fm scrobbled: {} — {}", t.artist, t.title),
|
||||
Err(e) => log::warn!("[scrobbler] last.fm scrobble failed: {e}"),
|
||||
}
|
||||
}
|
||||
if s.listenbrainz_active() {
|
||||
let c = lb_client(s);
|
||||
match c.submit_listen(&t.artist, &t.title, album, started_at as i64, None).await {
|
||||
Ok(()) => log::info!("[scrobbler] listenbrainz submitted: {} — {}", t.artist, t.title),
|
||||
Err(e) => log::warn!("[scrobbler] listenbrainz submit failed: {e}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A ListenBrainz client bound to the stored token, with its own enabled flag
|
||||
/// ON (submit_* early-returns if the client config is disabled — our gate is
|
||||
/// `ScrobblerSettings::listenbrainz_active`, checked before calling).
|
||||
fn lb_client(s: &ScrobblerSettings) -> ListenBrainzClient {
|
||||
ListenBrainzClient::with_config(ListenBrainzConfig {
|
||||
enabled: true,
|
||||
token: Some(s.listenbrainz_token.clone()),
|
||||
user_name: Some(s.listenbrainz_username.clone()),
|
||||
})
|
||||
}
|
||||
|
||||
/// The album name, unless it's empty or the queue-track "Unknown Album"
|
||||
/// placeholder (both scrobble better as "no album" than a fake one).
|
||||
fn album_opt(t: &QueueTrack) -> Option<&str> {
|
||||
if t.album.is_empty() || t.album == "Unknown Album" {
|
||||
None
|
||||
} else {
|
||||
Some(&t.album)
|
||||
}
|
||||
}
|
||||
|
||||
fn now_unix() -> u64 {
|
||||
SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn due_only_when_past_threshold_and_not_yet_scrobbled() {
|
||||
assert!(!due(10, Some(120), false)); // not there yet
|
||||
assert!(due(120, Some(120), false)); // exactly at threshold
|
||||
assert!(due(200, Some(120), false)); // past it
|
||||
assert!(!due(200, Some(120), true)); // already scrobbled
|
||||
assert!(!due(999, None, false)); // too short to scrobble (no threshold)
|
||||
}
|
||||
|
||||
fn qt(album: &str) -> QueueTrack {
|
||||
QueueTrack {
|
||||
id: 1,
|
||||
title: "Spain".into(),
|
||||
version: None,
|
||||
artist: "Chick Corea".into(),
|
||||
album: album.into(),
|
||||
album_version: None,
|
||||
duration_secs: 300,
|
||||
artwork_url: None,
|
||||
hires: false,
|
||||
bit_depth: None,
|
||||
sample_rate: None,
|
||||
is_local: false,
|
||||
album_id: None,
|
||||
artist_id: None,
|
||||
streamable: true,
|
||||
source: None,
|
||||
parental_warning: false,
|
||||
source_item_id_hint: None,
|
||||
context_kind: None,
|
||||
context_id: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn album_opt_drops_empty_and_unknown() {
|
||||
assert_eq!(album_opt(&qt("Light as a Feather")), Some("Light as a Feather"));
|
||||
assert_eq!(album_opt(&qt("")), None);
|
||||
assert_eq!(album_opt(&qt("Unknown Album")), None);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user