feat(listen-log): per-user listen event store, tracker and reading rules
New qbz-app::listen_log (contract §12.1): listen_events with PK = event (never the track), snapshot text per row, source + native item id, context, stream facts; listen_meta with schema_version, origin_id, paused, retention_days (unlimited). The tracker is a pure state machine: played_ms accumulates only while playing from monotonic deltas <= 5 s, so pauses, seeks and stalls add nothing; flush every 10 s of play; rows close as natural/skip/stop/error/shutdown and orphans close as shutdown on the next open. No threshold at write time: is_play / is_skip are pure reader rules with boundary tests. ListenLogger is the async facade (spawn_blocking everywhere); paused writes nothing, clear is DELETE + VACUUM + checkpoint. 29 tests. cc
This commit is contained in:
6 files changed
+1581
No files matched your search
@@ -3,6 +3,7 @@ pub mod diagnostics;
|
||||
pub mod memory_watchdog;
|
||||
pub mod offline_mode;
|
||||
pub mod graphics_autoconfig;
|
||||
pub mod listen_log;
|
||||
pub mod playback_context;
|
||||
pub mod playback_driver;
|
||||
pub mod qconnect_identity;
|
||||
|
||||
@@ -0,0 +1,385 @@
|
||||
//! The async facade both hosts drive: a [`ListenTracker`] plus a
|
||||
//! [`ListenStore`], every store call inside `spawn_blocking`.
|
||||
//!
|
||||
//! Hosts call the five verbs (`track_started`, `tick`, `ended_naturally`,
|
||||
//! `stopped`, `shutdown`) and never touch SQLite or the state machine
|
||||
//! themselves. "Listening history OFF" (`paused`) is honoured HERE, in one
|
||||
//! place: a paused logger opens no row, so nothing downstream has to ask.
|
||||
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
use super::store::ListenStore;
|
||||
use super::tracker::{ListenMeta, ListenTracker};
|
||||
|
||||
/// Where this process's rows come from (`origin_id`).
|
||||
pub enum Origin {
|
||||
/// The desktop app: a uuid generated once per install and kept in
|
||||
/// `listen_meta`.
|
||||
Install,
|
||||
/// The daemon: `qbzd:<hostname>`, fixed.
|
||||
Daemon { hostname: String },
|
||||
}
|
||||
|
||||
pub struct ListenLogger {
|
||||
store: Arc<Mutex<ListenStore>>,
|
||||
tracker: Mutex<ListenTracker>,
|
||||
app_session_id: String,
|
||||
origin_id: String,
|
||||
/// Mirror of `listen_meta.paused`, so the hot path (a tick per second)
|
||||
/// never opens the database to ask.
|
||||
paused: std::sync::atomic::AtomicBool,
|
||||
}
|
||||
|
||||
pub fn now_unix() -> i64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_secs() as i64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
impl ListenLogger {
|
||||
/// Open the per-user store, close any rows a previous run left open
|
||||
/// (`shutdown`), and mint this run's `app_session_id`.
|
||||
pub async fn open(user_dir: std::path::PathBuf, origin: Origin) -> Result<Arc<Self>, String> {
|
||||
let (store, origin_id, paused, orphans) = tokio::task::spawn_blocking(move || {
|
||||
let store = ListenStore::open_at(&user_dir)?;
|
||||
let origin_id = match origin {
|
||||
Origin::Install => store.origin_id_or_init(|| uuid::Uuid::new_v4().to_string())?,
|
||||
Origin::Daemon { hostname } => format!("qbzd:{hostname}"),
|
||||
};
|
||||
let paused = store.is_paused()?;
|
||||
let orphans = store.close_orphans_as_shutdown()?;
|
||||
Ok::<_, String>((store, origin_id, paused, orphans))
|
||||
})
|
||||
.await
|
||||
.map_err(|e| format!("listen log open task failed: {e}"))??;
|
||||
if orphans > 0 {
|
||||
log::info!(
|
||||
"[listen-log] closed {orphans} row(s) left open by a previous run as shutdown"
|
||||
);
|
||||
}
|
||||
log::info!("[listen-log] open (origin {origin_id}, paused={paused})");
|
||||
Ok(Arc::new(Self {
|
||||
store: Arc::new(Mutex::new(store)),
|
||||
tracker: Mutex::new(ListenTracker::new()),
|
||||
app_session_id: uuid::Uuid::new_v4().to_string(),
|
||||
origin_id,
|
||||
paused: std::sync::atomic::AtomicBool::new(paused),
|
||||
}))
|
||||
}
|
||||
|
||||
pub fn is_paused(&self) -> bool {
|
||||
self.paused.load(std::sync::atomic::Ordering::Relaxed)
|
||||
}
|
||||
|
||||
/// The "Listening history" toggle. Turning it OFF closes the row in
|
||||
/// flight (as `stop`) so nothing is written after the user said no.
|
||||
pub async fn set_paused(&self, paused: bool) -> Result<(), String> {
|
||||
self.paused
|
||||
.store(paused, std::sync::atomic::Ordering::Relaxed);
|
||||
if paused {
|
||||
let closed = self.tracker.lock().unwrap().stopped(now_unix(), false);
|
||||
if let Some(c) = closed {
|
||||
self.with_store(move |s| s.close_event(&c)).await?;
|
||||
}
|
||||
}
|
||||
self.with_store(move |s| s.set_paused(paused)).await
|
||||
}
|
||||
|
||||
/// "Clear listening history": DELETE + VACUUM. The row in flight is
|
||||
/// closed first so it does not survive the clear.
|
||||
pub async fn clear(&self) -> Result<(), String> {
|
||||
let closed = self.tracker.lock().unwrap().stopped(now_unix(), false);
|
||||
self.with_store(move |s| {
|
||||
if let Some(c) = closed {
|
||||
s.close_event(&c)?;
|
||||
}
|
||||
s.clear()
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn count(&self) -> Result<u64, String> {
|
||||
self.with_store(|s| s.count()).await
|
||||
}
|
||||
|
||||
/// A new track started (the host's de-duped track edge). Closes the row
|
||||
/// in flight as skip (or, with `infer_end`, as natural when it sat within
|
||||
/// 2 s of its end — for hosts without an explicit end edge).
|
||||
pub async fn track_started(&self, meta: ListenMeta, infer_end: bool) {
|
||||
let now = now_unix();
|
||||
if self.is_paused() {
|
||||
// Still close whatever was open before the pause landed.
|
||||
let closed = self.tracker.lock().unwrap().stopped(now, infer_end);
|
||||
if let Some(c) = closed {
|
||||
let _ = self.with_store(move |s| s.close_event(&c)).await;
|
||||
}
|
||||
return;
|
||||
}
|
||||
let (closed, meta) = {
|
||||
let mut t = self.tracker.lock().unwrap();
|
||||
if infer_end {
|
||||
t.track_started_inferring_end(meta, now)
|
||||
} else {
|
||||
t.track_started(meta, now)
|
||||
}
|
||||
};
|
||||
let session = self.app_session_id.clone();
|
||||
let origin = self.origin_id.clone();
|
||||
let result = self
|
||||
.with_store(move |s| {
|
||||
if let Some(c) = closed {
|
||||
s.close_event(&c)?;
|
||||
}
|
||||
s.open_event(&session, &origin, &meta, now)
|
||||
})
|
||||
.await;
|
||||
match result {
|
||||
Ok(id) => self.tracker.lock().unwrap().opened(id),
|
||||
Err(e) => log::warn!("[listen-log] open failed: {e}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// One playback observation (1 Hz). Flushes to disk every ~10 s of play.
|
||||
pub async fn tick(&self, position_ms: u64, playing: bool) {
|
||||
let flush = self.tracker.lock().unwrap().tick(position_ms, playing);
|
||||
if let Some(f) = flush {
|
||||
if let Err(e) = self.with_store(move |s| s.flush_progress(&f)).await {
|
||||
log::warn!("[listen-log] flush failed: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn ended_naturally(&self) {
|
||||
let closed = self.tracker.lock().unwrap().ended_naturally(now_unix());
|
||||
self.close(closed).await;
|
||||
}
|
||||
|
||||
pub async fn stopped(&self, infer_natural: bool) {
|
||||
let closed = self
|
||||
.tracker
|
||||
.lock()
|
||||
.unwrap()
|
||||
.stopped(now_unix(), infer_natural);
|
||||
self.close(closed).await;
|
||||
}
|
||||
|
||||
pub async fn errored(&self) {
|
||||
let closed = self.tracker.lock().unwrap().errored(now_unix());
|
||||
self.close(closed).await;
|
||||
}
|
||||
|
||||
/// Orderly exit: close the row in flight as `shutdown`. SYNCHRONOUS on
|
||||
/// purpose — the hosts call it after their event loop is gone.
|
||||
pub fn shutdown_blocking(&self) {
|
||||
let closed = self.tracker.lock().unwrap().shutdown(now_unix());
|
||||
if let Some(c) = closed {
|
||||
if let Ok(store) = self.store.lock() {
|
||||
if let Err(e) = store.close_event(&c) {
|
||||
log::warn!("[listen-log] shutdown close failed: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether a row is in flight (tests + diagnostics).
|
||||
pub fn has_open_row(&self) -> bool {
|
||||
self.tracker.lock().unwrap().open().is_some()
|
||||
}
|
||||
|
||||
/// Developer inspection: every row in id order.
|
||||
pub async fn rows(&self) -> Result<Vec<super::store::ListenRow>, String> {
|
||||
self.with_store(|s| s.rows()).await
|
||||
}
|
||||
|
||||
async fn close(&self, closed: Option<super::tracker::Closed>) {
|
||||
if let Some(c) = closed {
|
||||
if let Err(e) = self.with_store(move |s| s.close_event(&c)).await {
|
||||
log::warn!("[listen-log] close failed: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn with_store<T, F>(&self, f: F) -> Result<T, String>
|
||||
where
|
||||
T: Send + 'static,
|
||||
F: FnOnce(&ListenStore) -> Result<T, String> + Send + 'static,
|
||||
{
|
||||
let store = Arc::clone(&self.store);
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = store
|
||||
.lock()
|
||||
.map_err(|_| "listen log store lock poisoned".to_string())?;
|
||||
f(&guard)
|
||||
})
|
||||
.await
|
||||
.map_err(|e| format!("listen log task failed: {e}"))?
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::super::rules::EndReason;
|
||||
use super::*;
|
||||
|
||||
fn meta(id: &str, duration_ms: u64) -> ListenMeta {
|
||||
ListenMeta {
|
||||
source: "qobuz".into(),
|
||||
source_item_id: id.into(),
|
||||
title: format!("T{id}"),
|
||||
artist: "A".into(),
|
||||
duration_ms,
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn play_through_then_skip_then_shutdown() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let l = ListenLogger::open(dir.path().to_path_buf(), Origin::Install)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(l.origin_id.len(), 36);
|
||||
|
||||
l.track_started(meta("1", 30_000), false).await;
|
||||
for s in 0..=29u64 {
|
||||
l.tick(s * 1_000, true).await;
|
||||
}
|
||||
l.ended_naturally().await;
|
||||
|
||||
l.track_started(meta("2", 200_000), false).await;
|
||||
for s in 0..=10u64 {
|
||||
l.tick(s * 1_000, true).await;
|
||||
}
|
||||
l.track_started(meta("3", 200_000), false).await;
|
||||
for s in 0..=25u64 {
|
||||
l.tick(s * 1_000, true).await;
|
||||
}
|
||||
l.shutdown_blocking();
|
||||
|
||||
let rows = l.rows().await.unwrap();
|
||||
assert_eq!(rows.len(), 3);
|
||||
assert_eq!(rows[0].end_reason, Some(EndReason::Natural));
|
||||
assert_eq!(rows[0].played_ms, 29_000);
|
||||
assert_eq!(rows[1].end_reason, Some(EndReason::Skip));
|
||||
assert_eq!(rows[1].played_ms, 10_000);
|
||||
assert_eq!(rows[2].end_reason, Some(EndReason::Shutdown));
|
||||
assert_eq!(rows[2].played_ms, 25_000);
|
||||
assert!(rows.iter().all(|r| r.app_session_id == l.app_session_id));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn crash_leaves_flushed_progress_and_reopen_closes_it() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let origin = {
|
||||
let l = ListenLogger::open(dir.path().to_path_buf(), Origin::Install)
|
||||
.await
|
||||
.unwrap();
|
||||
l.track_started(meta("1", 200_000), false).await;
|
||||
for s in 0..=23u64 {
|
||||
l.tick(s * 1_000, true).await;
|
||||
}
|
||||
// No shutdown: the process died here. The last flush was at 20 s.
|
||||
l.origin_id.clone()
|
||||
};
|
||||
let l = ListenLogger::open(dir.path().to_path_buf(), Origin::Install)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(l.origin_id, origin, "origin id survives restarts");
|
||||
let rows = l.rows().await.unwrap();
|
||||
assert_eq!(rows.len(), 1);
|
||||
assert_eq!(rows[0].end_reason, Some(EndReason::Shutdown));
|
||||
assert_eq!(rows[0].played_ms, 20_000);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn paused_writes_nothing_and_resumes() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let l = ListenLogger::open(dir.path().to_path_buf(), Origin::Install)
|
||||
.await
|
||||
.unwrap();
|
||||
l.set_paused(true).await.unwrap();
|
||||
l.track_started(meta("1", 100_000), false).await;
|
||||
l.tick(0, true).await;
|
||||
l.tick(15_000, true).await;
|
||||
l.ended_naturally().await;
|
||||
assert_eq!(l.count().await.unwrap(), 0);
|
||||
assert!(!l.has_open_row());
|
||||
|
||||
// The flag is persisted: a reopen stays paused.
|
||||
let l2 = ListenLogger::open(dir.path().to_path_buf(), Origin::Install)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(l2.is_paused());
|
||||
l2.set_paused(false).await.unwrap();
|
||||
l2.track_started(meta("2", 100_000), false).await;
|
||||
l2.tick(0, true).await;
|
||||
l2.tick(1_000, true).await;
|
||||
l2.stopped(false).await;
|
||||
let rows = l2.rows().await.unwrap();
|
||||
assert_eq!(rows.len(), 1);
|
||||
assert_eq!(rows[0].end_reason, Some(EndReason::Stop));
|
||||
assert_eq!(rows[0].played_ms, 1_000);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn pausing_mid_track_closes_the_open_row_and_stops_writing() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let l = ListenLogger::open(dir.path().to_path_buf(), Origin::Install)
|
||||
.await
|
||||
.unwrap();
|
||||
l.track_started(meta("1", 100_000), false).await;
|
||||
l.tick(0, true).await;
|
||||
l.tick(1_000, true).await;
|
||||
l.set_paused(true).await.unwrap();
|
||||
l.tick(2_000, true).await;
|
||||
l.track_started(meta("2", 100_000), false).await;
|
||||
let rows = l.rows().await.unwrap();
|
||||
assert_eq!(rows.len(), 1);
|
||||
assert_eq!(rows[0].end_reason, Some(EndReason::Stop));
|
||||
assert_eq!(rows[0].played_ms, 1_000);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn clear_removes_everything_including_the_open_row() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let l = ListenLogger::open(dir.path().to_path_buf(), Origin::Install)
|
||||
.await
|
||||
.unwrap();
|
||||
l.track_started(meta("1", 100_000), false).await;
|
||||
l.ended_naturally().await;
|
||||
l.track_started(meta("2", 100_000), false).await;
|
||||
l.clear().await.unwrap();
|
||||
assert_eq!(l.count().await.unwrap(), 0);
|
||||
assert!(!l.has_open_row());
|
||||
// Logging continues after a clear.
|
||||
l.track_started(meta("3", 100_000), false).await;
|
||||
assert_eq!(l.count().await.unwrap(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn daemon_origin_and_inferred_end() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let l = ListenLogger::open(
|
||||
dir.path().to_path_buf(),
|
||||
Origin::Daemon {
|
||||
hostname: "pi".into(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
l.track_started(meta("1", 100_000), true).await;
|
||||
l.tick(0, true).await;
|
||||
l.tick(99_000, true).await;
|
||||
// The bus only says "another track started": infer natural.
|
||||
l.track_started(meta("2", 100_000), true).await;
|
||||
l.tick(0, true).await;
|
||||
l.tick(5_000, true).await;
|
||||
l.stopped(true).await;
|
||||
let rows = l.rows().await.unwrap();
|
||||
assert_eq!(rows[0].origin_id, "qbzd:pi");
|
||||
assert_eq!(rows[0].end_reason, Some(EndReason::Natural));
|
||||
assert_eq!(rows[1].end_reason, Some(EndReason::Stop));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,154 @@
|
||||
//! Listen log — the per-user record of what was actually heard.
|
||||
//!
|
||||
//! Contract: `qbz-nix-docs/research/2026-08-28-listening-log-offline-reco-analysis.md`
|
||||
//! §12.1. Capture only: this module has NO consumer in the app. No rail,
|
||||
//! mix, insight or suggestion reads it; the older stores
|
||||
//! (`recently_played.json`, `album_play_history.db`,
|
||||
//! `playlist_play_history.db`) keep running unchanged beside it.
|
||||
//!
|
||||
//! - [`store`] — the SQLite schema (PK = event) and its migrations.
|
||||
//! - [`tracker`] — the pure state machine (accumulator, end reasons).
|
||||
//! - [`rules`] — the reading thresholds (`is_play`, `is_skip`).
|
||||
//! - [`logger`] — the async facade the Qt poll loop and `qbzd` drive.
|
||||
|
||||
pub mod logger;
|
||||
pub mod rules;
|
||||
pub mod store;
|
||||
pub mod tracker;
|
||||
|
||||
pub use logger::{now_unix, ListenLogger, Origin};
|
||||
pub use rules::{is_play, is_skip, EndReason};
|
||||
pub use store::{ListenRow, ListenStore};
|
||||
pub use tracker::{ListenMeta, ListenTracker};
|
||||
|
||||
/// Ephemeral ids (CD / ad-hoc folder rows) live above 2^48 and never reach
|
||||
/// the log (owner rule, mirrored from `recently_qt.rs:329`).
|
||||
pub const EPHEMERAL_ID_FLOOR: u64 = 1 << 48;
|
||||
|
||||
/// Build the row snapshot from a queue track plus what the host knows about
|
||||
/// the stream. `source_item_id` is the NATIVE id when the queue carries one
|
||||
/// (media-server item ids, Plex rating keys, offline row ids travel in
|
||||
/// `source_item_id_hint` for local-tier rows); for Qobuz the hint is a
|
||||
/// container id and the track id IS the native id.
|
||||
pub fn meta_from_queue_track(
|
||||
track: &qbz_models::QueueTrack,
|
||||
bit_depth: Option<u32>,
|
||||
sample_rate_hz: Option<u32>,
|
||||
output_backend: Option<String>,
|
||||
) -> ListenMeta {
|
||||
let source = track
|
||||
.source
|
||||
.clone()
|
||||
.filter(|s| !s.is_empty())
|
||||
.unwrap_or_else(|| {
|
||||
if track.is_local {
|
||||
"local".into()
|
||||
} else {
|
||||
"qobuz".into()
|
||||
}
|
||||
});
|
||||
// "qobuz_download" is the offline cache's own source word in the local
|
||||
// library; the contract's vocabulary calls that tier `offline`.
|
||||
let source = if source == "qobuz_download" {
|
||||
"offline".to_string()
|
||||
} else {
|
||||
source
|
||||
};
|
||||
let source_item_id = match (&track.source_item_id_hint, track.is_local) {
|
||||
(Some(hint), true) if !hint.is_empty() => hint.clone(),
|
||||
_ => track.id.to_string(),
|
||||
};
|
||||
ListenMeta {
|
||||
source,
|
||||
source_item_id,
|
||||
track_id: Some(track.id as i64),
|
||||
album_id: track.album_id.clone().filter(|a| !a.is_empty()),
|
||||
artist_id: track.artist_id.map(|id| id.to_string()),
|
||||
isrc: track.isrc.clone().filter(|s| !s.is_empty()),
|
||||
recording_mbid: track.recording_mbid.clone().filter(|s| !s.is_empty()),
|
||||
title: track.title.clone(),
|
||||
artist: track.artist.clone(),
|
||||
album: Some(track.album.clone()).filter(|a| !a.is_empty()),
|
||||
album_artist: None,
|
||||
artwork_key: track.artwork_url.clone().filter(|a| !a.is_empty()),
|
||||
duration_ms: track.duration_secs * 1_000,
|
||||
context_kind: track.context_kind.clone().unwrap_or_default(),
|
||||
context_id: track.context_id.clone().unwrap_or_default(),
|
||||
bit_depth: bit_depth.or(track.bit_depth),
|
||||
sample_rate: sample_rate_hz
|
||||
.or_else(|| track.sample_rate.map(|khz| (khz * 1000.0).round() as u32)),
|
||||
output_backend,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn track() -> qbz_models::QueueTrack {
|
||||
qbz_models::QueueTrack {
|
||||
id: 7,
|
||||
title: "Song".into(),
|
||||
version: None,
|
||||
artist: "Band".into(),
|
||||
album: "LP".into(),
|
||||
album_version: None,
|
||||
duration_secs: 200,
|
||||
artwork_url: Some("http://art".into()),
|
||||
hires: true,
|
||||
bit_depth: Some(24),
|
||||
sample_rate: Some(96.0),
|
||||
is_local: false,
|
||||
album_id: Some("alb".into()),
|
||||
artist_id: Some(9),
|
||||
streamable: true,
|
||||
source: Some("qobuz".into()),
|
||||
parental_warning: false,
|
||||
source_item_id_hint: Some("alb".into()),
|
||||
context_kind: Some("playlist".into()),
|
||||
context_id: Some("pl-1".into()),
|
||||
isrc: Some("USRC17607839".into()),
|
||||
recording_mbid: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn qobuz_uses_the_track_id_not_the_container_hint() {
|
||||
let m = meta_from_queue_track(&track(), None, None, None);
|
||||
assert_eq!(m.source, "qobuz");
|
||||
assert_eq!(m.source_item_id, "7");
|
||||
assert_eq!(m.track_id, Some(7));
|
||||
assert_eq!(m.duration_ms, 200_000);
|
||||
assert_eq!(m.context_kind, "playlist");
|
||||
assert_eq!(m.context_id, "pl-1");
|
||||
assert_eq!(m.isrc.as_deref(), Some("USRC17607839"));
|
||||
assert_eq!(m.sample_rate, Some(96_000));
|
||||
assert_eq!(m.bit_depth, Some(24));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn local_tier_rows_use_the_native_hint_and_offline_is_renamed() {
|
||||
let mut t = track();
|
||||
t.is_local = true;
|
||||
t.source = Some("jellyfin".into());
|
||||
t.source_item_id_hint = Some("item-abc".into());
|
||||
let m = meta_from_queue_track(&t, Some(16), Some(44_100), Some("alsa".into()));
|
||||
assert_eq!(m.source, "jellyfin");
|
||||
assert_eq!(m.source_item_id, "item-abc");
|
||||
assert_eq!(m.bit_depth, Some(16));
|
||||
assert_eq!(m.sample_rate, Some(44_100));
|
||||
assert_eq!(m.output_backend.as_deref(), Some("alsa"));
|
||||
|
||||
t.source = Some("qobuz_download".into());
|
||||
assert_eq!(
|
||||
meta_from_queue_track(&t, None, None, None).source,
|
||||
"offline"
|
||||
);
|
||||
|
||||
t.source = None;
|
||||
t.source_item_id_hint = None;
|
||||
let m = meta_from_queue_track(&t, None, None, None);
|
||||
assert_eq!(m.source, "local");
|
||||
assert_eq!(m.source_item_id, "7");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
//! Pure reading rules over listen events.
|
||||
//!
|
||||
//! The store applies NO threshold when it writes (LMS/Astra): a skip is a row
|
||||
//! with little `played_ms`, never an absent row. Whatever wants to count
|
||||
//! "plays" or "skips" applies its bar here, so two consumers can never
|
||||
//! disagree about what a play is because one of them wrote and the other
|
||||
//! read (the Amperfy/Retro trap).
|
||||
|
||||
/// Why a listen event closed. `Open` is the in-flight state (NULL on disk).
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum EndReason {
|
||||
/// The track played to its end.
|
||||
Natural,
|
||||
/// Replaced by another track before its end (next / previous / play X).
|
||||
Skip,
|
||||
/// Playback stopped with nothing following.
|
||||
Stop,
|
||||
/// The stream failed.
|
||||
Error,
|
||||
/// The app went away with the row open (closed on the next start, or on
|
||||
/// an orderly exit).
|
||||
Shutdown,
|
||||
/// Playback moved to another device (cast / peer). Reserved — v1 never
|
||||
/// records those tracks, but the word is fixed so readers can rely on it.
|
||||
Handoff,
|
||||
}
|
||||
|
||||
/// ONE list, read by both directions: `as_str` and `parse` must never drift
|
||||
/// (the `AudioFormat::Dsd` → `Unknown` lesson: a writer and a reader that each
|
||||
/// spell the enum are two lists).
|
||||
const END_REASONS: &[(EndReason, &str)] = &[
|
||||
(EndReason::Natural, "natural"),
|
||||
(EndReason::Skip, "skip"),
|
||||
(EndReason::Stop, "stop"),
|
||||
(EndReason::Error, "error"),
|
||||
(EndReason::Shutdown, "shutdown"),
|
||||
(EndReason::Handoff, "handoff"),
|
||||
];
|
||||
|
||||
impl EndReason {
|
||||
pub fn as_str(self) -> &'static str {
|
||||
END_REASONS
|
||||
.iter()
|
||||
.find(|(r, _)| *r == self)
|
||||
.map(|(_, s)| *s)
|
||||
.expect("every EndReason is in END_REASONS")
|
||||
}
|
||||
|
||||
pub fn parse(s: &str) -> Option<Self> {
|
||||
END_REASONS.iter().find(|(_, t)| *t == s).map(|(r, _)| *r)
|
||||
}
|
||||
}
|
||||
|
||||
/// Last.fm / MusicBee bar: the track is longer than 30 s and at least half of
|
||||
/// it (capped at 4 minutes) was audible.
|
||||
pub fn is_play(played_ms: u64, duration_ms: u64) -> bool {
|
||||
duration_ms > 30_000 && played_ms >= (duration_ms / 2).min(240_000)
|
||||
}
|
||||
|
||||
/// beets bar: it was replaced before 85 % of it was heard. A track that was
|
||||
/// replaced at 90 % is a play that happened to end early, not a skip.
|
||||
pub fn is_skip(end_reason: Option<EndReason>, played_ms: u64, duration_ms: u64) -> bool {
|
||||
end_reason == Some(EndReason::Skip) && (played_ms as f64) < 0.85 * duration_ms as f64
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn end_reason_round_trips_every_variant() {
|
||||
for (reason, word) in END_REASONS {
|
||||
assert_eq!(reason.as_str(), *word);
|
||||
assert_eq!(EndReason::parse(word), Some(*reason));
|
||||
}
|
||||
assert_eq!(EndReason::parse("nope"), None);
|
||||
assert_eq!(END_REASONS.len(), 6);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn is_play_thirty_second_floor() {
|
||||
// 30 s exactly is NOT longer than 30 s.
|
||||
assert!(!is_play(30_000, 30_000));
|
||||
// 30.001 s: half is 15 s (integer ms).
|
||||
assert!(is_play(15_000, 30_001));
|
||||
assert!(!is_play(14_999, 30_001));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn is_play_half_or_four_minutes() {
|
||||
// 100 s track: half = 50 s.
|
||||
assert!(is_play(50_000, 100_000));
|
||||
assert!(!is_play(49_999, 100_000));
|
||||
// 10 min track: half = 300 s, capped at 240 s.
|
||||
assert!(is_play(240_000, 600_000));
|
||||
assert!(!is_play(239_999, 600_000));
|
||||
// Exactly 480 s: half == cap.
|
||||
assert!(is_play(240_000, 480_000));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn is_play_unknown_duration_never_counts() {
|
||||
assert!(!is_play(1_000_000, 0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn is_skip_needs_skip_reason_and_under_85_percent() {
|
||||
let d = 200_000;
|
||||
assert!(is_skip(Some(EndReason::Skip), 10_000, d));
|
||||
assert!(is_skip(Some(EndReason::Skip), 169_999, d));
|
||||
// 85 % exactly is not a skip.
|
||||
assert!(!is_skip(Some(EndReason::Skip), 170_000, d));
|
||||
assert!(!is_skip(Some(EndReason::Natural), 10_000, d));
|
||||
assert!(!is_skip(Some(EndReason::Stop), 10_000, d));
|
||||
assert!(!is_skip(None, 10_000, d));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn is_skip_zero_duration_is_never_a_skip() {
|
||||
// 0 < 0.85 * 0 is false: an unknown-duration row cannot be a skip.
|
||||
assert!(!is_skip(Some(EndReason::Skip), 0, 0));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,521 @@
|
||||
//! The per-user listen-event store: `<user_dir>/listen/listen_log.db`.
|
||||
//!
|
||||
//! One row per listen EVENT (PK = event, never the track — the Tempo trap:
|
||||
//! a track-keyed table with REPLACE can only ever count to one). Counters
|
||||
//! are derived by readers, never written here.
|
||||
//!
|
||||
//! Synchronous `rusqlite`; the async facade (`logger.rs`) wraps every call
|
||||
//! in `spawn_blocking`. WAL + `synchronous=NORMAL` (ADR-002).
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use rusqlite::{params, Connection, OptionalExtension};
|
||||
|
||||
use super::rules::EndReason;
|
||||
use super::tracker::{Closed, Flush, ListenMeta};
|
||||
|
||||
pub const SCHEMA_VERSION: i64 = 1;
|
||||
|
||||
/// Sub-directory + file under the user dir.
|
||||
pub const DB_SUBDIR: &str = "listen";
|
||||
pub const DB_FILE: &str = "listen_log.db";
|
||||
|
||||
pub struct ListenStore {
|
||||
conn: Connection,
|
||||
#[allow(dead_code)]
|
||||
path: Option<PathBuf>,
|
||||
}
|
||||
|
||||
/// One persisted row, as readers see it (tests + the developer query).
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct ListenRow {
|
||||
pub id: i64,
|
||||
pub app_session_id: String,
|
||||
pub origin_id: String,
|
||||
pub source: String,
|
||||
pub source_item_id: String,
|
||||
pub track_id: Option<i64>,
|
||||
pub isrc: Option<String>,
|
||||
pub recording_mbid: Option<String>,
|
||||
pub title: String,
|
||||
pub artist: String,
|
||||
pub duration_ms: u64,
|
||||
pub started_at: i64,
|
||||
pub ended_at: Option<i64>,
|
||||
pub played_ms: u64,
|
||||
pub end_position_ms: Option<u64>,
|
||||
pub end_reason: Option<EndReason>,
|
||||
pub context_kind: String,
|
||||
pub context_id: String,
|
||||
}
|
||||
|
||||
impl ListenStore {
|
||||
/// Open (or create) the store under `user_dir`.
|
||||
pub fn open_at(user_dir: &Path) -> Result<Self, String> {
|
||||
let dir = user_dir.join(DB_SUBDIR);
|
||||
std::fs::create_dir_all(&dir).map_err(|e| format!("Failed to create listen dir: {e}"))?;
|
||||
let path = dir.join(DB_FILE);
|
||||
let conn =
|
||||
Connection::open(&path).map_err(|e| format!("Failed to open listen log: {e}"))?;
|
||||
conn.execute_batch(
|
||||
"PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA busy_timeout=2000;",
|
||||
)
|
||||
.map_err(|e| format!("Failed to set listen log pragmas: {e}"))?;
|
||||
let store = Self {
|
||||
conn,
|
||||
path: Some(path),
|
||||
};
|
||||
store.migrate()?;
|
||||
Ok(store)
|
||||
}
|
||||
|
||||
pub fn open_in_memory() -> Result<Self, String> {
|
||||
let conn = Connection::open_in_memory()
|
||||
.map_err(|e| format!("Failed to open in-memory listen log: {e}"))?;
|
||||
let store = Self { conn, path: None };
|
||||
store.migrate()?;
|
||||
Ok(store)
|
||||
}
|
||||
|
||||
/// Idempotent: creates the schema on a fresh file and records
|
||||
/// `schema_version`; a later version adds its `ALTER`s behind a version
|
||||
/// check here (a table is only new once — see `playlist_play_history`).
|
||||
fn migrate(&self) -> Result<(), String> {
|
||||
self.conn
|
||||
.execute_batch(
|
||||
r#"
|
||||
CREATE TABLE IF NOT EXISTS listen_meta (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS listen_events (
|
||||
id INTEGER PRIMARY KEY,
|
||||
app_session_id TEXT NOT NULL,
|
||||
origin_id TEXT NOT NULL,
|
||||
source TEXT NOT NULL,
|
||||
source_item_id TEXT NOT NULL,
|
||||
track_id INTEGER,
|
||||
album_id TEXT,
|
||||
artist_id TEXT,
|
||||
isrc TEXT,
|
||||
recording_mbid TEXT,
|
||||
title TEXT NOT NULL,
|
||||
artist TEXT NOT NULL,
|
||||
album TEXT,
|
||||
album_artist TEXT,
|
||||
artwork_key TEXT,
|
||||
duration_ms INTEGER NOT NULL DEFAULT 0,
|
||||
started_at INTEGER NOT NULL,
|
||||
ended_at INTEGER,
|
||||
played_ms INTEGER NOT NULL DEFAULT 0,
|
||||
end_position_ms INTEGER,
|
||||
end_reason TEXT,
|
||||
context_kind TEXT NOT NULL DEFAULT '',
|
||||
context_id TEXT NOT NULL DEFAULT '',
|
||||
bit_depth INTEGER,
|
||||
sample_rate INTEGER,
|
||||
output_backend TEXT
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS ix_le_started ON listen_events(started_at);
|
||||
CREATE INDEX IF NOT EXISTS ix_le_item ON listen_events(source, source_item_id, started_at);
|
||||
CREATE INDEX IF NOT EXISTS ix_le_isrc ON listen_events(isrc) WHERE isrc IS NOT NULL;
|
||||
CREATE INDEX IF NOT EXISTS ix_le_open ON listen_events(id) WHERE end_reason IS NULL;
|
||||
"#,
|
||||
)
|
||||
.map_err(|e| format!("Failed to create listen log schema: {e}"))?;
|
||||
|
||||
let current = self
|
||||
.meta("schema_version")?
|
||||
.and_then(|v| v.parse::<i64>().ok());
|
||||
match current {
|
||||
None => self.set_meta("schema_version", &SCHEMA_VERSION.to_string())?,
|
||||
Some(v) if v > SCHEMA_VERSION => {
|
||||
return Err(format!(
|
||||
"listen log schema {v} is newer than this build ({SCHEMA_VERSION})"
|
||||
));
|
||||
}
|
||||
// Future: `Some(v) if v < SCHEMA_VERSION` → ALTERs, then bump.
|
||||
Some(_) => {}
|
||||
}
|
||||
// Defaults that must exist from day one so readers never special-case
|
||||
// an absent key: unlimited retention (owner decision §10.4).
|
||||
if self.meta("retention_days")?.is_none() {
|
||||
self.set_meta("retention_days", "0")?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ----- meta -----------------------------------------------------------
|
||||
|
||||
pub fn meta(&self, key: &str) -> Result<Option<String>, String> {
|
||||
self.conn
|
||||
.query_row(
|
||||
"SELECT value FROM listen_meta WHERE key = ?1",
|
||||
params![key],
|
||||
|row| row.get::<_, Option<String>>(0),
|
||||
)
|
||||
.optional()
|
||||
.map(|v| v.flatten())
|
||||
.map_err(|e| format!("Failed to read listen meta {key}: {e}"))
|
||||
}
|
||||
|
||||
pub fn set_meta(&self, key: &str, value: &str) -> Result<(), String> {
|
||||
self.conn
|
||||
.execute(
|
||||
"INSERT INTO listen_meta (key, value) VALUES (?1, ?2)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||||
params![key, value],
|
||||
)
|
||||
.map(|_| ())
|
||||
.map_err(|e| format!("Failed to write listen meta {key}: {e}"))
|
||||
}
|
||||
|
||||
pub fn schema_version(&self) -> Result<i64, String> {
|
||||
Ok(self
|
||||
.meta("schema_version")?
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(0))
|
||||
}
|
||||
|
||||
/// The install's origin id, generated once and kept in `listen_meta`.
|
||||
/// `qbzd` passes its own fixed id instead (`qbzd:<host>`).
|
||||
pub fn origin_id_or_init(&self, generate: impl FnOnce() -> String) -> Result<String, String> {
|
||||
if let Some(id) = self.meta("origin_id")? {
|
||||
return Ok(id);
|
||||
}
|
||||
let id = generate();
|
||||
self.set_meta("origin_id", &id)?;
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
/// `paused = 1` means "Listening history" is OFF: nothing is written
|
||||
/// (and nothing is deleted).
|
||||
pub fn is_paused(&self) -> Result<bool, String> {
|
||||
Ok(self.meta("paused")?.as_deref() == Some("1"))
|
||||
}
|
||||
|
||||
pub fn set_paused(&self, paused: bool) -> Result<(), String> {
|
||||
self.set_meta("paused", if paused { "1" } else { "0" })
|
||||
}
|
||||
|
||||
pub fn retention_days(&self) -> Result<u32, String> {
|
||||
Ok(self
|
||||
.meta("retention_days")?
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(0))
|
||||
}
|
||||
|
||||
// ----- events ---------------------------------------------------------
|
||||
|
||||
/// Insert an open row; returns its id.
|
||||
pub fn open_event(
|
||||
&self,
|
||||
app_session_id: &str,
|
||||
origin_id: &str,
|
||||
meta: &ListenMeta,
|
||||
started_at: i64,
|
||||
) -> Result<i64, String> {
|
||||
self.conn
|
||||
.execute(
|
||||
"INSERT INTO listen_events (
|
||||
app_session_id, origin_id, source, source_item_id, track_id,
|
||||
album_id, artist_id, isrc, recording_mbid,
|
||||
title, artist, album, album_artist, artwork_key,
|
||||
duration_ms, started_at, context_kind, context_id,
|
||||
bit_depth, sample_rate, output_backend
|
||||
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21)",
|
||||
params![
|
||||
app_session_id,
|
||||
origin_id,
|
||||
meta.source,
|
||||
meta.source_item_id,
|
||||
meta.track_id,
|
||||
meta.album_id,
|
||||
meta.artist_id,
|
||||
meta.isrc,
|
||||
meta.recording_mbid,
|
||||
meta.title,
|
||||
meta.artist,
|
||||
meta.album,
|
||||
meta.album_artist,
|
||||
meta.artwork_key,
|
||||
meta.duration_ms as i64,
|
||||
started_at,
|
||||
meta.context_kind,
|
||||
meta.context_id,
|
||||
meta.bit_depth,
|
||||
meta.sample_rate,
|
||||
meta.output_backend,
|
||||
],
|
||||
)
|
||||
.map_err(|e| format!("Failed to open listen event: {e}"))?;
|
||||
Ok(self.conn.last_insert_rowid())
|
||||
}
|
||||
|
||||
/// Persist the accumulator of an OPEN row (a crash then loses at most
|
||||
/// one flush interval). Never touches a closed row.
|
||||
pub fn flush_progress(&self, f: &Flush) -> Result<(), String> {
|
||||
self.conn
|
||||
.execute(
|
||||
"UPDATE listen_events SET played_ms = ?1, end_position_ms = ?2
|
||||
WHERE id = ?3 AND end_reason IS NULL",
|
||||
params![
|
||||
f.played_ms as i64,
|
||||
f.end_position_ms.map(|v| v as i64),
|
||||
f.event_id
|
||||
],
|
||||
)
|
||||
.map(|_| ())
|
||||
.map_err(|e| format!("Failed to flush listen progress: {e}"))
|
||||
}
|
||||
|
||||
/// Close a row. Idempotent on an already-closed row (no-op).
|
||||
pub fn close_event(&self, c: &Closed) -> Result<(), String> {
|
||||
self.conn
|
||||
.execute(
|
||||
"UPDATE listen_events
|
||||
SET ended_at = ?1, played_ms = ?2, end_position_ms = ?3, end_reason = ?4
|
||||
WHERE id = ?5 AND end_reason IS NULL",
|
||||
params![
|
||||
c.ended_at,
|
||||
c.played_ms as i64,
|
||||
c.end_position_ms.map(|v| v as i64),
|
||||
c.reason.as_str(),
|
||||
c.event_id
|
||||
],
|
||||
)
|
||||
.map(|_| ())
|
||||
.map_err(|e| format!("Failed to close listen event: {e}"))
|
||||
}
|
||||
|
||||
/// Rows left open by a crash (or a kill) close as `shutdown` with the
|
||||
/// `played_ms` that reached disk, at startup. `ended_at` is unknowable —
|
||||
/// it is set to `started_at + played_ms` as the best lower bound.
|
||||
pub fn close_orphans_as_shutdown(&self) -> Result<usize, String> {
|
||||
self.conn
|
||||
.execute(
|
||||
"UPDATE listen_events
|
||||
SET end_reason = ?1, ended_at = started_at + (played_ms / 1000)
|
||||
WHERE end_reason IS NULL",
|
||||
params![EndReason::Shutdown.as_str()],
|
||||
)
|
||||
.map_err(|e| format!("Failed to close orphan listen events: {e}"))
|
||||
}
|
||||
|
||||
/// "Clear listening history": every event goes, the file is compacted.
|
||||
/// `listen_meta` (origin id, paused flag, retention) is kept.
|
||||
pub fn clear(&self) -> Result<(), String> {
|
||||
// Under WAL the vacuumed image lands in the -wal file first; the
|
||||
// TRUNCATE checkpoint is what makes the main file actually shrink
|
||||
// (otherwise "Clear" leaves the old size on disk until a later
|
||||
// checkpoint happens to run).
|
||||
self.conn
|
||||
.execute_batch("DELETE FROM listen_events; VACUUM; PRAGMA wal_checkpoint(TRUNCATE);")
|
||||
.map_err(|e| format!("Failed to clear listen log: {e}"))
|
||||
}
|
||||
|
||||
pub fn count(&self) -> Result<u64, String> {
|
||||
self.conn
|
||||
.query_row("SELECT COUNT(*) FROM listen_events", [], |row| {
|
||||
row.get::<_, i64>(0)
|
||||
})
|
||||
.map(|n| n as u64)
|
||||
.map_err(|e| format!("Failed to count listen events: {e}"))
|
||||
}
|
||||
|
||||
/// Every row in id order (tests + developer inspection; never a rail).
|
||||
pub fn rows(&self) -> Result<Vec<ListenRow>, String> {
|
||||
let mut stmt = self
|
||||
.conn
|
||||
.prepare(
|
||||
"SELECT id, app_session_id, origin_id, source, source_item_id, track_id,
|
||||
isrc, recording_mbid, title, artist, duration_ms, started_at,
|
||||
ended_at, played_ms, end_position_ms, end_reason,
|
||||
context_kind, context_id
|
||||
FROM listen_events ORDER BY id",
|
||||
)
|
||||
.map_err(|e| format!("Failed to prepare listen query: {e}"))?;
|
||||
let rows = stmt
|
||||
.query_map([], |row| {
|
||||
Ok(ListenRow {
|
||||
id: row.get(0)?,
|
||||
app_session_id: row.get(1)?,
|
||||
origin_id: row.get(2)?,
|
||||
source: row.get(3)?,
|
||||
source_item_id: row.get(4)?,
|
||||
track_id: row.get(5)?,
|
||||
isrc: row.get(6)?,
|
||||
recording_mbid: row.get(7)?,
|
||||
title: row.get(8)?,
|
||||
artist: row.get(9)?,
|
||||
duration_ms: row.get::<_, i64>(10)? as u64,
|
||||
started_at: row.get(11)?,
|
||||
ended_at: row.get(12)?,
|
||||
played_ms: row.get::<_, i64>(13)? as u64,
|
||||
end_position_ms: row.get::<_, Option<i64>>(14)?.map(|v| v as u64),
|
||||
end_reason: row
|
||||
.get::<_, Option<String>>(15)?
|
||||
.as_deref()
|
||||
.and_then(EndReason::parse),
|
||||
context_kind: row.get(16)?,
|
||||
context_id: row.get(17)?,
|
||||
})
|
||||
})
|
||||
.map_err(|e| format!("Failed to query listen events: {e}"))?;
|
||||
Ok(rows.flatten().collect())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn meta() -> ListenMeta {
|
||||
ListenMeta {
|
||||
source: "qobuz".into(),
|
||||
source_item_id: "42".into(),
|
||||
track_id: Some(42),
|
||||
title: "Song".into(),
|
||||
artist: "Band".into(),
|
||||
album: Some("LP".into()),
|
||||
duration_ms: 200_000,
|
||||
context_kind: "album".into(),
|
||||
context_id: "abc".into(),
|
||||
isrc: Some("USRC17607839".into()),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn schema_version_is_recorded_and_reopen_is_idempotent() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
{
|
||||
let s = ListenStore::open_at(dir.path()).unwrap();
|
||||
assert_eq!(s.schema_version().unwrap(), SCHEMA_VERSION);
|
||||
assert_eq!(s.retention_days().unwrap(), 0);
|
||||
assert!(!s.is_paused().unwrap());
|
||||
}
|
||||
let s = ListenStore::open_at(dir.path()).unwrap();
|
||||
assert_eq!(s.schema_version().unwrap(), SCHEMA_VERSION);
|
||||
assert!(dir.path().join(DB_SUBDIR).join(DB_FILE).exists());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_newer_schema_refuses_to_open() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
{
|
||||
let s = ListenStore::open_at(dir.path()).unwrap();
|
||||
s.set_meta("schema_version", &(SCHEMA_VERSION + 1).to_string())
|
||||
.unwrap();
|
||||
}
|
||||
assert!(ListenStore::open_at(dir.path()).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn origin_id_is_generated_once() {
|
||||
let s = ListenStore::open_in_memory().unwrap();
|
||||
let a = s.origin_id_or_init(|| "first".into()).unwrap();
|
||||
let b = s.origin_id_or_init(|| "second".into()).unwrap();
|
||||
assert_eq!(a, "first");
|
||||
assert_eq!(b, "first");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn open_flush_close_round_trip() {
|
||||
let s = ListenStore::open_in_memory().unwrap();
|
||||
let id = s.open_event("sess", "origin", &meta(), 1_000).unwrap();
|
||||
s.flush_progress(&Flush {
|
||||
event_id: id,
|
||||
played_ms: 10_000,
|
||||
end_position_ms: Some(10_000),
|
||||
})
|
||||
.unwrap();
|
||||
let row = &s.rows().unwrap()[0];
|
||||
assert_eq!(row.played_ms, 10_000);
|
||||
assert_eq!(row.end_reason, None);
|
||||
assert_eq!(row.isrc.as_deref(), Some("USRC17607839"));
|
||||
assert_eq!(row.context_kind, "album");
|
||||
s.close_event(&Closed {
|
||||
event_id: id,
|
||||
reason: EndReason::Skip,
|
||||
played_ms: 12_000,
|
||||
end_position_ms: Some(12_000),
|
||||
ended_at: 1_012,
|
||||
})
|
||||
.unwrap();
|
||||
let row = &s.rows().unwrap()[0];
|
||||
assert_eq!(row.end_reason, Some(EndReason::Skip));
|
||||
assert_eq!(row.played_ms, 12_000);
|
||||
assert_eq!(row.ended_at, Some(1_012));
|
||||
// A late flush must not reopen or alter a closed row.
|
||||
s.flush_progress(&Flush {
|
||||
event_id: id,
|
||||
played_ms: 99_000,
|
||||
end_position_ms: None,
|
||||
})
|
||||
.unwrap();
|
||||
assert_eq!(s.rows().unwrap()[0].played_ms, 12_000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn orphans_close_as_shutdown_with_the_flushed_progress() {
|
||||
let s = ListenStore::open_in_memory().unwrap();
|
||||
let a = s.open_event("sess", "o", &meta(), 1_000).unwrap();
|
||||
s.flush_progress(&Flush {
|
||||
event_id: a,
|
||||
played_ms: 30_000,
|
||||
end_position_ms: Some(30_000),
|
||||
})
|
||||
.unwrap();
|
||||
let b = s.open_event("sess", "o", &meta(), 2_000).unwrap();
|
||||
s.close_event(&Closed {
|
||||
event_id: b,
|
||||
reason: EndReason::Natural,
|
||||
played_ms: 200_000,
|
||||
end_position_ms: Some(200_000),
|
||||
ended_at: 2_200,
|
||||
})
|
||||
.unwrap();
|
||||
assert_eq!(s.close_orphans_as_shutdown().unwrap(), 1);
|
||||
let rows = s.rows().unwrap();
|
||||
assert_eq!(rows[0].end_reason, Some(EndReason::Shutdown));
|
||||
assert_eq!(rows[0].played_ms, 30_000);
|
||||
assert_eq!(rows[0].ended_at, Some(1_030));
|
||||
assert_eq!(rows[1].end_reason, Some(EndReason::Natural));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn clear_empties_events_and_keeps_meta() {
|
||||
let s = ListenStore::open_in_memory().unwrap();
|
||||
s.set_paused(false).unwrap();
|
||||
let _ = s.origin_id_or_init(|| "keep-me".into()).unwrap();
|
||||
s.open_event("sess", "o", &meta(), 1).unwrap();
|
||||
s.open_event("sess", "o", &meta(), 2).unwrap();
|
||||
assert_eq!(s.count().unwrap(), 2);
|
||||
s.clear().unwrap();
|
||||
assert_eq!(s.count().unwrap(), 0);
|
||||
assert_eq!(s.meta("origin_id").unwrap().as_deref(), Some("keep-me"));
|
||||
assert_eq!(s.schema_version().unwrap(), SCHEMA_VERSION);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn vacuum_compacts_the_file() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let s = ListenStore::open_at(dir.path()).unwrap();
|
||||
for i in 0..2_000 {
|
||||
s.open_event("sess", "o", &meta(), i).unwrap();
|
||||
}
|
||||
let path = dir.path().join(DB_SUBDIR).join(DB_FILE);
|
||||
// Checkpoint so the rows are in the main file, not only the WAL.
|
||||
s.conn
|
||||
.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")
|
||||
.unwrap();
|
||||
let before = std::fs::metadata(&path).unwrap().len();
|
||||
s.clear().unwrap();
|
||||
let after = std::fs::metadata(&path).unwrap().len();
|
||||
assert!(
|
||||
after < before,
|
||||
"vacuum did not shrink the file ({before} -> {after})"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,397 @@
|
||||
//! The listen tracker: a pure state machine over playback observations.
|
||||
//!
|
||||
//! No clock, no I/O — every input carries its own `now`, and every output is
|
||||
//! a value the caller persists. That is what makes the rules testable and
|
||||
//! what keeps the two hosts (the Qt poll loop and the `qbzd` event bus)
|
||||
//! from drifting: both feed the same machine.
|
||||
//!
|
||||
//! The one rule that matters (Codex §9, Feishin): `played_ms` accumulates
|
||||
//! ONLY while playing, from position deltas that are monotonic and at most
|
||||
//! [`MAX_DELTA_MS`]. A pause adds nothing; a seek (forward or back) is a
|
||||
//! delta outside that window and adds nothing; a stall that skips reporting
|
||||
//! for a while adds nothing. Wall-clock and the final position are never
|
||||
//! consulted — pauses, seeks, repeat and the old scrobble bug all made them
|
||||
//! lie.
|
||||
|
||||
use super::rules::EndReason;
|
||||
|
||||
/// A position jump larger than this is a seek or a stall, not audible time.
|
||||
/// The hosts observe at 1 Hz, so a genuine tick is ~1 s.
|
||||
pub const MAX_DELTA_MS: u64 = 5_000;
|
||||
|
||||
/// Persist the accumulator at least this often (in audible ms) so a crash
|
||||
/// loses at most this much of the open row.
|
||||
pub const FLUSH_EVERY_MS: u64 = 10_000;
|
||||
|
||||
/// Snapshot of the track a row is opened for. Text is captured here on
|
||||
/// purpose (Astra): the row must still make sense after the track leaves
|
||||
/// Qobuz or the Plex server it came from is gone.
|
||||
#[derive(Debug, Clone, Default, PartialEq)]
|
||||
pub struct ListenMeta {
|
||||
pub source: String,
|
||||
pub source_item_id: String,
|
||||
pub track_id: Option<i64>,
|
||||
pub album_id: Option<String>,
|
||||
pub artist_id: Option<String>,
|
||||
pub isrc: Option<String>,
|
||||
pub recording_mbid: Option<String>,
|
||||
pub title: String,
|
||||
pub artist: String,
|
||||
pub album: Option<String>,
|
||||
pub album_artist: Option<String>,
|
||||
pub artwork_key: Option<String>,
|
||||
pub duration_ms: u64,
|
||||
pub context_kind: String,
|
||||
pub context_id: String,
|
||||
pub bit_depth: Option<u32>,
|
||||
pub sample_rate: Option<u32>,
|
||||
pub output_backend: Option<String>,
|
||||
}
|
||||
|
||||
/// The open row, as the machine sees it.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct OpenListen {
|
||||
/// Store row id, set by the host once the row is inserted.
|
||||
pub event_id: i64,
|
||||
pub meta: ListenMeta,
|
||||
pub started_at: i64,
|
||||
pub played_ms: u64,
|
||||
/// Last position observed, whatever the play state.
|
||||
pub last_position_ms: Option<u64>,
|
||||
/// `played_ms` at the last flush the host acknowledged.
|
||||
flushed_ms: u64,
|
||||
}
|
||||
|
||||
/// A row to close, with everything the store needs.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct Closed {
|
||||
pub event_id: i64,
|
||||
pub reason: EndReason,
|
||||
pub played_ms: u64,
|
||||
pub end_position_ms: Option<u64>,
|
||||
pub ended_at: i64,
|
||||
}
|
||||
|
||||
/// A progress checkpoint the host should persist.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub struct Flush {
|
||||
pub event_id: i64,
|
||||
pub played_ms: u64,
|
||||
pub end_position_ms: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
pub struct ListenTracker {
|
||||
open: Option<OpenListen>,
|
||||
}
|
||||
|
||||
impl ListenTracker {
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
pub fn open(&self) -> Option<&OpenListen> {
|
||||
self.open.as_ref()
|
||||
}
|
||||
|
||||
/// A new track surfaced. Closes the previous row (if any) as a SKIP —
|
||||
/// unless the host already closed it as natural — and returns the meta
|
||||
/// the host must insert; call [`Self::opened`] with the row id.
|
||||
pub fn track_started(&mut self, meta: ListenMeta, now: i64) -> (Option<Closed>, ListenMeta) {
|
||||
let closed = self.close_with(EndReason::Skip, now);
|
||||
self.open = Some(OpenListen {
|
||||
event_id: 0,
|
||||
meta: meta.clone(),
|
||||
started_at: now,
|
||||
played_ms: 0,
|
||||
last_position_ms: None,
|
||||
flushed_ms: 0,
|
||||
});
|
||||
(closed, meta)
|
||||
}
|
||||
|
||||
/// The host inserted the row for the track from the last
|
||||
/// [`Self::track_started`]; from now on progress has somewhere to go.
|
||||
pub fn opened(&mut self, event_id: i64) {
|
||||
if let Some(open) = self.open.as_mut() {
|
||||
open.event_id = event_id;
|
||||
}
|
||||
}
|
||||
|
||||
/// One playback observation. Returns a flush when the accumulator has
|
||||
/// advanced by [`FLUSH_EVERY_MS`] since the last one.
|
||||
pub fn tick(&mut self, position_ms: u64, playing: bool) -> Option<Flush> {
|
||||
let open = self.open.as_mut()?;
|
||||
if let (true, Some(last)) = (playing, open.last_position_ms) {
|
||||
if position_ms > last {
|
||||
let delta = position_ms - last;
|
||||
if delta <= MAX_DELTA_MS {
|
||||
open.played_ms += delta;
|
||||
}
|
||||
}
|
||||
}
|
||||
open.last_position_ms = Some(position_ms);
|
||||
if open.event_id != 0 && open.played_ms.saturating_sub(open.flushed_ms) >= FLUSH_EVERY_MS {
|
||||
open.flushed_ms = open.played_ms;
|
||||
return Some(Flush {
|
||||
event_id: open.event_id,
|
||||
played_ms: open.played_ms,
|
||||
end_position_ms: open.last_position_ms,
|
||||
});
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
/// The host's own natural-end predicate fired.
|
||||
pub fn ended_naturally(&mut self, now: i64) -> Option<Closed> {
|
||||
self.close_with(EndReason::Natural, now)
|
||||
}
|
||||
|
||||
/// Playback stopped with nothing following. A host without an explicit
|
||||
/// natural-end edge (qbzd) passes `infer_natural = true`: a stop whose
|
||||
/// last position sat within 2 s of the duration is the track ending.
|
||||
pub fn stopped(&mut self, now: i64, infer_natural: bool) -> Option<Closed> {
|
||||
let reason = if infer_natural && self.near_end() {
|
||||
EndReason::Natural
|
||||
} else {
|
||||
EndReason::Stop
|
||||
};
|
||||
self.close_with(reason, now)
|
||||
}
|
||||
|
||||
pub fn errored(&mut self, now: i64) -> Option<Closed> {
|
||||
self.close_with(EndReason::Error, now)
|
||||
}
|
||||
|
||||
pub fn shutdown(&mut self, now: i64) -> Option<Closed> {
|
||||
self.close_with(EndReason::Shutdown, now)
|
||||
}
|
||||
|
||||
/// For hosts that only see "a different track started": was the previous
|
||||
/// one within 2 s of its end? Mirrors the desktop's
|
||||
/// `seen_position + 2 >= duration` (playback_driver.rs:226-264).
|
||||
pub fn near_end(&self) -> bool {
|
||||
let Some(open) = self.open.as_ref() else {
|
||||
return false;
|
||||
};
|
||||
let duration = open.meta.duration_ms;
|
||||
match open.last_position_ms {
|
||||
Some(pos) if duration > 0 => pos + 2_000 >= duration,
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Close with an inferred reason: natural when near the end, else skip.
|
||||
/// The qbzd bus has no TrackEnded (nobody emits it), so this is its
|
||||
/// track-edge close.
|
||||
pub fn track_started_inferring_end(
|
||||
&mut self,
|
||||
meta: ListenMeta,
|
||||
now: i64,
|
||||
) -> (Option<Closed>, ListenMeta) {
|
||||
let reason = if self.near_end() {
|
||||
EndReason::Natural
|
||||
} else {
|
||||
EndReason::Skip
|
||||
};
|
||||
let closed = self.close_with(reason, now);
|
||||
self.open = Some(OpenListen {
|
||||
event_id: 0,
|
||||
meta: meta.clone(),
|
||||
started_at: now,
|
||||
played_ms: 0,
|
||||
last_position_ms: None,
|
||||
flushed_ms: 0,
|
||||
});
|
||||
(closed, meta)
|
||||
}
|
||||
|
||||
fn close_with(&mut self, reason: EndReason, now: i64) -> Option<Closed> {
|
||||
let open = self.open.take()?;
|
||||
// A row the host never managed to insert has nothing to close.
|
||||
if open.event_id == 0 {
|
||||
return None;
|
||||
}
|
||||
Some(Closed {
|
||||
event_id: open.event_id,
|
||||
reason,
|
||||
played_ms: open.played_ms,
|
||||
end_position_ms: open.last_position_ms,
|
||||
ended_at: now,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn meta(duration_ms: u64) -> ListenMeta {
|
||||
ListenMeta {
|
||||
source: "qobuz".into(),
|
||||
source_item_id: "1".into(),
|
||||
title: "T".into(),
|
||||
artist: "A".into(),
|
||||
duration_ms,
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
fn started(t: &mut ListenTracker, id: i64) {
|
||||
let (_, _) = t.track_started(meta(200_000), 1_000);
|
||||
t.opened(id);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accumulates_only_monotonic_small_deltas_while_playing() {
|
||||
let mut t = ListenTracker::new();
|
||||
started(&mut t, 7);
|
||||
assert_eq!(t.tick(0, true), None); // first observation: no delta
|
||||
t.tick(1_000, true);
|
||||
t.tick(2_000, true);
|
||||
assert_eq!(t.open().unwrap().played_ms, 2_000);
|
||||
// Seek forward 60 s: ignored, but the position is remembered.
|
||||
t.tick(62_000, true);
|
||||
assert_eq!(t.open().unwrap().played_ms, 2_000);
|
||||
t.tick(63_000, true);
|
||||
assert_eq!(t.open().unwrap().played_ms, 3_000);
|
||||
// Seek back: negative delta ignored.
|
||||
t.tick(10_000, true);
|
||||
assert_eq!(t.open().unwrap().played_ms, 3_000);
|
||||
t.tick(11_000, true);
|
||||
assert_eq!(t.open().unwrap().played_ms, 4_000);
|
||||
// Exactly MAX_DELTA counts; one more does not.
|
||||
t.tick(16_000, true);
|
||||
assert_eq!(t.open().unwrap().played_ms, 9_000);
|
||||
t.tick(21_001, true);
|
||||
assert_eq!(t.open().unwrap().played_ms, 9_000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn paused_ticks_add_nothing_and_survive_the_pause() {
|
||||
let mut t = ListenTracker::new();
|
||||
started(&mut t, 7);
|
||||
t.tick(0, true);
|
||||
t.tick(1_000, true);
|
||||
// Two minutes paused at the same position.
|
||||
for _ in 0..120 {
|
||||
t.tick(1_000, false);
|
||||
}
|
||||
assert_eq!(t.open().unwrap().played_ms, 1_000);
|
||||
// Resume: the next delta is 1 s again.
|
||||
t.tick(2_000, true);
|
||||
assert_eq!(t.open().unwrap().played_ms, 2_000);
|
||||
// A paused tick whose position moved (a seek while paused) counts nothing.
|
||||
t.tick(50_000, false);
|
||||
t.tick(51_000, true);
|
||||
assert_eq!(t.open().unwrap().played_ms, 3_000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn flushes_every_ten_audible_seconds() {
|
||||
let mut t = ListenTracker::new();
|
||||
started(&mut t, 7);
|
||||
t.tick(0, true);
|
||||
let mut flushes = Vec::new();
|
||||
for s in 1..=25u64 {
|
||||
if let Some(f) = t.tick(s * 1_000, true) {
|
||||
flushes.push(f);
|
||||
}
|
||||
}
|
||||
assert_eq!(
|
||||
flushes,
|
||||
vec![
|
||||
Flush {
|
||||
event_id: 7,
|
||||
played_ms: 10_000,
|
||||
end_position_ms: Some(10_000)
|
||||
},
|
||||
Flush {
|
||||
event_id: 7,
|
||||
played_ms: 20_000,
|
||||
end_position_ms: Some(20_000)
|
||||
},
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn next_track_closes_the_previous_as_skip_with_its_accumulator() {
|
||||
let mut t = ListenTracker::new();
|
||||
started(&mut t, 7);
|
||||
t.tick(0, true);
|
||||
for s in 1..=10u64 {
|
||||
t.tick(s * 1_000, true);
|
||||
}
|
||||
let (closed, next) = t.track_started(meta(100_000), 5_000);
|
||||
assert_eq!(
|
||||
closed,
|
||||
Some(Closed {
|
||||
event_id: 7,
|
||||
reason: EndReason::Skip,
|
||||
played_ms: 10_000,
|
||||
end_position_ms: Some(10_000),
|
||||
ended_at: 5_000,
|
||||
})
|
||||
);
|
||||
assert_eq!(next.duration_ms, 100_000);
|
||||
assert_eq!(t.open().unwrap().event_id, 0);
|
||||
t.opened(8);
|
||||
assert_eq!(t.open().unwrap().event_id, 8);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn natural_end_then_next_track_opens_without_a_second_close() {
|
||||
let mut t = ListenTracker::new();
|
||||
started(&mut t, 7);
|
||||
t.tick(0, true);
|
||||
t.tick(199_000, true);
|
||||
let closed = t.ended_naturally(9_000).unwrap();
|
||||
assert_eq!(closed.reason, EndReason::Natural);
|
||||
let (closed_again, _) = t.track_started(meta(100_000), 9_001);
|
||||
assert_eq!(closed_again, None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stop_shutdown_error_reasons() {
|
||||
let mut t = ListenTracker::new();
|
||||
started(&mut t, 1);
|
||||
assert_eq!(t.stopped(2, false).unwrap().reason, EndReason::Stop);
|
||||
started(&mut t, 2);
|
||||
assert_eq!(t.shutdown(3).unwrap().reason, EndReason::Shutdown);
|
||||
started(&mut t, 3);
|
||||
assert_eq!(t.errored(4).unwrap().reason, EndReason::Error);
|
||||
assert_eq!(t.shutdown(5), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn inferred_end_uses_the_two_second_window() {
|
||||
let mut t = ListenTracker::new();
|
||||
started(&mut t, 1);
|
||||
t.tick(0, true);
|
||||
t.tick(198_000, true);
|
||||
assert!(t.near_end());
|
||||
let (closed, _) = t.track_started_inferring_end(meta(50_000), 9);
|
||||
assert_eq!(closed.unwrap().reason, EndReason::Natural);
|
||||
t.opened(2);
|
||||
t.tick(0, true);
|
||||
t.tick(10_000, true);
|
||||
assert!(!t.near_end());
|
||||
assert_eq!(t.stopped(10, true).unwrap().reason, EndReason::Stop);
|
||||
started(&mut t, 3);
|
||||
t.tick(198_500, true);
|
||||
assert_eq!(t.stopped(11, true).unwrap().reason, EndReason::Natural);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_row_that_was_never_inserted_closes_to_nothing() {
|
||||
let mut t = ListenTracker::new();
|
||||
let _ = t.track_started(meta(1), 1);
|
||||
// no `opened` call: the insert failed or paused
|
||||
t.tick(0, true);
|
||||
t.tick(1_000, true);
|
||||
assert_eq!(t.tick(11_000, true), None);
|
||||
assert_eq!(t.ended_naturally(2), None);
|
||||
assert!(t.open().is_none());
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user