diff --git a/android/app/src/main/java/app/hocket/ui/screens/home/HomeScreen.kt b/android/app/src/main/java/app/hocket/ui/screens/home/HomeScreen.kt index dc931ab..f43447d 100644 --- a/android/app/src/main/java/app/hocket/ui/screens/home/HomeScreen.kt +++ b/android/app/src/main/java/app/hocket/ui/screens/home/HomeScreen.kt @@ -122,7 +122,9 @@ fun HomeScreen(nav: NavHostController) { if (recentList.isNotEmpty()) { val ids = recentList.map { it.track.id }.distinct() item(key = "h.continue", contentType = "header") { SectionHeader(stringResource(R.string.home_continue), Modifier.animateItem()) } - items(recentList.take(5), key = { "r" + it.playedAt + it.track.id }, contentType = { "track" }) { entry -> + // One row per play (start + track is its identity, and a lazy list throws on a + // repeated key): older databases can hold the same play twice. + items(recentList.distinctBy { it.playedAt to it.track.id }.take(5), key = { "r" + it.playedAt + it.track.id }, contentType = { "track" }) { entry -> TrackRow(entry.track, onClick = { client.dispatch(Commands.playTracks(serverId, ids, ids.indexOf(entry.track.id), continueLabel)) }, modifier = Modifier.animateItem(), trailing = { Text(entryAgo(entry), style = MaterialTheme.typography.labelSmall, color = MaterialTheme.colorScheme.onSurfaceVariant, modifier = Modifier.padding(start = 8.dp)) }) diff --git a/android/playback/src/main/java/app/hocket/playback/ConnectRouteProvider.kt b/android/playback/src/main/java/app/hocket/playback/ConnectRouteProvider.kt index 7f499d2..82caefc 100644 --- a/android/playback/src/main/java/app/hocket/playback/ConnectRouteProvider.kt +++ b/android/playback/src/main/java/app/hocket/playback/ConnectRouteProvider.kt @@ -48,6 +48,12 @@ class ConnectRouteProvider : MediaRoute2ProviderService() { private var state = ConnectRoutes.State() /** The device a switcher pick asked for, and when, until the core's state shows it playing. */ private var pending: Pair? = null + /** + * When "Stop casting" asked for playback back here, until the core's state shows it here. The + * session stays released meanwhile: recreating it while the other device still reports playing + * would bring the chip back mid-handoff, and each further tap would start another handoff. + */ + private var returningSince: Long? = null private var sessionRoute: String? = null override fun onCreate() { @@ -67,6 +73,10 @@ class ConnectRouteProvider : MediaRoute2ProviderService() { /** The device the session should show as selected: the core's, or a pick still in flight. */ private fun target(): DeviceInfo? { + returningSince?.let { since -> + if (state.remote == null || SystemClock.elapsedRealtime() - since > PENDING_MS) returningSince = null + else return null + } state.remote?.let { remote -> if (pending?.first == remote.id) pending = null; return remote } val (id, at) = pending ?: return null if (SystemClock.elapsedRealtime() - at > PENDING_MS) { pending = null; return null } @@ -115,6 +125,7 @@ class ConnectRouteProvider : MediaRoute2ProviderService() { } private fun handoffTo(deviceId: String) { + returningSince = null pending = deviceId to SystemClock.elapsedRealtime() dispatch(Commands.handoffTo(deviceId)) } @@ -138,7 +149,10 @@ class ConnectRouteProvider : MediaRoute2ProviderService() { sessionRoute = null notifySessionReleased(sessionId) val self = state.selfId - if (state.remote != null && self != null) dispatch(Commands.handoffTo(self)) + if (state.remote != null && self != null) { + returningSince = SystemClock.elapsedRealtime() + dispatch(Commands.handoffTo(self)) + } } override fun onTransferToRoute(requestId: Long, sessionId: String, routeId: String) { diff --git a/crates/hocket-core/src/db/migrations/0004_play_history_unique.sql b/crates/hocket-core/src/db/migrations/0004_play_history_unique.sql new file mode 100644 index 0000000..df31224 --- /dev/null +++ b/crates/hocket-core/src/db/migrations/0004_play_history_unique.sql @@ -0,0 +1,41 @@ +-- 0004: a play is recorded once. A play's identity is its track and its +-- original start (`played_at`, the session clock's `startedAt`, carried +-- unchanged across handoffs). A play that moved back to a device that had +-- already recorded it (quick handoffs back and forth) reached its threshold +-- there again and was recorded twice, and the Android Home screen, keyed by +-- start and track, crashed on the pair. +-- +-- Keep the first row of each play (scrobbled if any copy was), take the +-- extra local play count bumps back, drop the copies, and let the index +-- refuse any new one. +UPDATE play_history SET scrobbled = 1 +WHERE scrobbled = 0 AND EXISTS ( + SELECT 1 FROM play_history o + WHERE o.server_id = play_history.server_id AND o.track_id = play_history.track_id + AND o.played_at = play_history.played_at AND o.scrobbled = 1 +); + +UPDATE tracks SET local_play_count = MAX(0, local_play_count - ( + SELECT count(*) FROM play_history d + WHERE d.server_id = tracks.server_id AND d.track_id = tracks.id + AND EXISTS ( + SELECT 1 FROM play_history k + WHERE k.server_id = d.server_id AND k.track_id = d.track_id + AND k.played_at = d.played_at AND k.id < d.id + ) +)) +WHERE EXISTS ( + SELECT 1 FROM play_history d JOIN play_history k + ON k.server_id = d.server_id AND k.track_id = d.track_id + AND k.played_at = d.played_at AND k.id < d.id + WHERE d.server_id = tracks.server_id AND d.track_id = tracks.id +); + +DELETE FROM play_history +WHERE EXISTS ( + SELECT 1 FROM play_history k + WHERE k.server_id = play_history.server_id AND k.track_id = play_history.track_id + AND k.played_at = play_history.played_at AND k.id < play_history.id +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_play_history_play ON play_history(server_id, track_id, played_at); diff --git a/crates/hocket-core/src/db/mod.rs b/crates/hocket-core/src/db/mod.rs index a99d19e..ef8d0b6 100644 --- a/crates/hocket-core/src/db/mod.rs +++ b/crates/hocket-core/src/db/mod.rs @@ -40,10 +40,15 @@ pub const MIGRATIONS: &[(u32, &str, &str)] = &[ "0003_stream_cache_spans", include_str!("migrations/0003_stream_cache_spans.sql"), ), + ( + 4, + "0004_play_history_unique", + include_str!("migrations/0004_play_history_unique.sql"), + ), ]; /// Current schema version (the last migration number). -pub const SCHEMA_VERSION: u32 = 3; +pub const SCHEMA_VERSION: u32 = 4; /// Tables that are a cache of the server and may be dropped and re-synced. pub const MIRROR_TABLES: &[&str] = &[ @@ -507,6 +512,67 @@ mod tests { assert_eq!(v as u32, SCHEMA_VERSION); } + #[test] + fn migration_4_drops_duplicate_plays() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("hocket.db"); + { + // A version 3 database holding one play recorded twice. + let db = Db::open(&path, &dir.path().join("backups")).unwrap(); + db.with_conn(|c| { + c.execute_batch( + "DROP INDEX idx_play_history_play; + DELETE FROM schema_version WHERE version = 4; + INSERT INTO tracks(id, server_id, title, local_play_count) VALUES ('t', 's', 'T', 3), ('u', 's', 'U', 1); + INSERT INTO play_history(server_id, track_id, played_at, played_ms, scrobbled, device_id) VALUES + ('s', 't', 1790294961606.5, 126260, 0, 'd'), + ('s', 't', 1790294961606.5, 132388, 1, 'd'), + ('s', 't', 1790294000000.0, 126260, 0, 'd'), + ('s', 'u', 1790294961606.5, 126260, 0, 'd');", + )?; + Ok(()) + }) + .unwrap(); + } + let db = Db::open(&path, &dir.path().join("backups")).unwrap(); + assert_eq!(db.schema_version().unwrap(), SCHEMA_VERSION); + let rows: Vec<(String, f64, i64, i64)> = db + .with_conn(|c| { + let mut st = c.prepare( + "SELECT track_id, played_at, played_ms, scrobbled FROM play_history ORDER BY id", + )?; + let rows = st.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)))?; + Ok(rows.collect::, _>>()?) + }) + .unwrap(); + assert_eq!( + rows, + vec![ + ("t".into(), 1790294961606.5, 126260, 1), + ("t".into(), 1790294000000.0, 126260, 0), + ("u".into(), 1790294961606.5, 126260, 0), + ] + ); + let counts: Vec = db + .with_conn(|c| { + let mut st = c.prepare("SELECT local_play_count FROM tracks ORDER BY id")?; + let rows = st.query_map([], |r| r.get(0))?; + Ok(rows.collect::, _>>()?) + }) + .unwrap(); + assert_eq!(counts, vec![2, 1]); + // The index refuses a copy from now on. + assert!(db + .with_conn(|c| { + c.execute( + "INSERT INTO play_history(server_id, track_id, played_at, played_ms, scrobbled, device_id) VALUES ('s', 'u', 1790294961606.5, 1, 0, 'd')", + [], + )?; + Ok(()) + }) + .is_err()); + } + #[test] fn newer_database_is_refused() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/hocket-core/src/db/queries.rs b/crates/hocket-core/src/db/queries.rs index 7bfbd01..0d44a84 100644 --- a/crates/hocket-core/src/db/queries.rs +++ b/crates/hocket-core/src/db/queries.rs @@ -1274,9 +1274,31 @@ impl From> for WhereClause { #[allow(dead_code)] fn _assert_tosql(_: &dyn ToSql) {} +/// The `play_history` row already recorded for the play of `track_id` +/// that started at `played_at` (a play's identity), if any. +pub fn recorded_play_in( + tx: &Connection, + server_id: &str, + track_id: &str, + played_at: f64, +) -> DbResult> { + Ok(tx + .query_row( + "SELECT id FROM play_history WHERE server_id = ?1 AND track_id = ?2 AND played_at = ?3", + params![server_id, track_id, played_at], + |r| r.get(0), + ) + .optional()?) +} + /// [`Db::record_play`] inside the caller's transaction: a `play_history` /// row (returning its id) and the track's `local_play_count` / /// `local_last_played` bump. The one place these rows are written. +/// +/// A play is recorded once: when this play (track and start) already has a +/// row (it came back to this device after a handoff and reached its +/// threshold here again), that row's id is returned and nothing is counted +/// twice; it is only marked `scrobbled` if this verdict says so. pub fn record_play_in( tx: &Connection, server_id: &str, @@ -1286,6 +1308,12 @@ pub fn record_play_in( scrobbled: bool, device_id: &str, ) -> DbResult { + if let Some(id) = recorded_play_in(tx, server_id, track_id, played_at)? { + if scrobbled { + tx.execute("UPDATE play_history SET scrobbled = 1 WHERE id = ?1", [id])?; + } + return Ok(id); + } tx.execute( "INSERT INTO play_history(server_id, track_id, played_at, played_ms, scrobbled, device_id) VALUES (?1, ?2, ?3, ?4, ?5, ?6)", params![server_id, track_id, played_at, played_ms, scrobbled as i64, device_id], diff --git a/crates/hocket-core/src/outbox/scrobbler.rs b/crates/hocket-core/src/outbox/scrobbler.rs index 3ea4e6a..2d03a57 100644 --- a/crates/hocket-core/src/outbox/scrobbler.rs +++ b/crates/hocket-core/src/outbox/scrobbler.rs @@ -336,6 +336,10 @@ impl ScrobbleRecorder { also: impl FnOnce(&rusqlite::Transaction) -> DbResult<()>, ) -> DbResult<()> { self.db.with_tx(|tx| { + // Asked again about a play already recorded here (it came back + // after a handoff): one row, and at most one submission. + let recorded = + crate::db::queries::recorded_play_in(tx, server_id, track_id, played_at)?.is_some(); let history_id = crate::db::queries::record_play_in( tx, server_id, @@ -345,7 +349,7 @@ impl ScrobbleRecorder { verdict == Verdict::ScrobbledElsewhere, &self.device_id, )?; - if verdict == Verdict::Submit { + if verdict == Verdict::Submit && !recorded { self.outbox.enqueue_in( tx, server_id, @@ -713,6 +717,81 @@ mod tests { assert_eq!((lpc, llp), (1, Some(4_000.0))); } + #[test] + fn a_play_asked_about_again_is_recorded_and_submitted_once() { + // The play came back to this device after a handoff and reached its + // threshold here again: same track, same start, more time played. + let db = Db::open_in_memory().unwrap(); + db.upsert_tracks( + &[crate::api::Track { + id: "t".into(), + server_id: "srv".into(), + title: "T".into(), + ..Default::default() + }], + &[], + 1, + ) + .unwrap(); + let clock = Arc::new(TestClock(Mutex::new(5_000.0))); + let outbox = Outbox::new(db.clone(), clock.clone()); + let rec = ScrobbleRecorder::new(db.clone(), outbox.clone(), "dev"); + let started_at = 1_790_294_961_606.5; + rec.record_verdict("srv", "t", started_at, 126_260, Verdict::Submit, |_| Ok(())) + .unwrap(); + rec.record_verdict("srv", "t", started_at, 132_388, Verdict::Submit, |_| Ok(())) + .unwrap(); + rec.record_verdict( + "srv", + "t", + started_at, + 140_000, + Verdict::ScrobbledElsewhere, + |_| Ok(()), + ) + .unwrap(); + let h = db.recently_played(5).unwrap(); + assert_eq!(h.len(), 1); + assert_eq!(h[0].played_ms, 126_260); + assert!(h[0].scrobbled); + let submissions = outbox + .pending() + .unwrap() + .into_iter() + .filter(|p| { + matches!( + p.mutation, + Mutation::Scrobble { + submission: true, + .. + } + ) + }) + .count(); + assert_eq!(submissions, 1); + let lpc: i64 = db + .with_conn(|c| { + Ok(c.query_row( + "SELECT local_play_count FROM tracks WHERE id='t'", + [], + |r| r.get(0), + )?) + }) + .unwrap(); + assert_eq!(lpc, 1); + // A later play of the same track is a play of its own. + rec.record_verdict( + "srv", + "t", + started_at + 300_000.0, + 126_260, + Verdict::Submit, + |_| Ok(()), + ) + .unwrap(); + assert_eq!(db.recently_played(5).unwrap().len(), 2); + } + mod props { use super::*; use proptest::prelude::*;