music: backup play log from now-playing samples (origin nowplaying; 30 s or half the track, ignoring pauses and seeks); recently-played takes those rows over with its exact timestamp

src/lib/music/collector.server.ts

@@ -186,6 +186,35 @@ export async function pollPlayer(
 	};
 }
 
+/**
+ * How far apart a backup play (end seen in now-playing) and Spotify's own played_at for the
+ * same play may be: up to 90 s, but under half the track so a repeat is never merged.
+ */
+export const backupWindowS = (durationMs: number | null) =>
+	Math.round(Math.min(90, Math.max(20, (durationMs ?? 180_000) / 2_000)));
+
+/** Store a backup play (origin 'nowplaying') unless Spotify already has that play. */
+export async function recordBackupPlay(
+	sql: Db,
+	play: { trackId: number; endedAt: number; listenedMs: number; durationMs: number | null }
+) {
+	const at = new Date(play.endedAt).toISOString();
+	const w = backupWindowS(play.durationMs);
+	const rows = await sql`
+		insert into music.plays (track_id, played_at, source, origin, ms_played)
+		select ${play.trackId}, ${at}::timestamptz, 'spotify', 'nowplaying',
+		       ${Math.round(play.durationMs ? Math.min(play.listenedMs, play.durationMs) : play.listenedMs)}
+		where not exists (
+			select 1 from music.plays p join music.tracks t on t.id = p.track_id
+			where p.source = 'spotify'
+			  and p.played_at between ${at}::timestamptz - make_interval(secs => ${w})
+			                      and ${at}::timestamptz + make_interval(secs => ${w})
+			  and t.match_key = (select match_key from music.tracks where id = ${play.trackId}))
+		on conflict do nothing
+		returning id`;
+	return rows.length > 0;
+}
+
 /** Store plays newer than the cursor; returns how many were new. */
 export async function pollRecent(
 	sql: Db,
@@ -208,6 +237,29 @@ export async function pollRecent(
 		const at = play.playedAt.getTime();
 		if (cursor && at <= cursor) continue;
 		const id = await upsertTrack(sql, play.track, true, cache);
+		// The backup log may already have this play: take over that row with Spotify's
+		// exact timestamp and context instead of adding a second one.
+		const w = backupWindowS(play.track.durationMs ?? null);
+		const iso = play.playedAt.toISOString();
+		const claimed = await sql`
+			update music.plays set played_at = ${iso}::timestamptz, origin = 'api',
+				context_type = ${play.contextType ?? null}, context_uri = ${play.contextUri ?? null}
+			where id = (
+				select p.id from music.plays p join music.tracks t on t.id = p.track_id
+				where p.source = 'spotify' and p.origin = 'nowplaying'
+				  and p.played_at between ${iso}::timestamptz - make_interval(secs => ${w})
+				                      and ${iso}::timestamptz + make_interval(secs => ${w})
+				  and t.match_key = (select match_key from music.tracks where id = ${id})
+				order by abs(extract(epoch from p.played_at - ${iso}::timestamptz)) limit 1)
+			  and not exists (select 1 from music.plays x
+				where x.source = 'spotify' and x.track_id = music.plays.track_id
+				  and x.played_at = ${iso}::timestamptz)
+			returning id`;
+		if (claimed.length) {
+			newPlays++;
+			newest = Math.max(newest, at);
+			continue;
+		}
 		if (
 			await insertPlay(sql, {
 				trackId: id,

src/lib/music/now-playing.test.ts

@@ -6,6 +6,9 @@ import {
 	nextDelay,
 	sseEvent,
 	toLive,
+	newPlayTracker,
+	trackPlay,
+	type PlayTracker,
 	type PlayerSample
 } from './now-playing';
 
@@ -78,3 +81,79 @@ describe('now playing', () => {
 		expect(nextDelay(false, 50)).toBe(30_000);
 	});
 });
+
+describe('trackPlay (backup play log)', () => {
+	const at = (trackId: number, progressMs: number, sampledAt: number, isPlaying = true) =>
+		({
+			trackId,
+			track: { ...track(`t${trackId}`), durationMs: trackId === 9 ? 40_000 : 200_000 },
+			isPlaying,
+			progressMs,
+			sampledAt
+		}) as unknown as PlayerSample;
+	const run = (samples: PlayerSample[]) => {
+		let state: PlayTracker = newPlayTracker();
+		const out = [];
+		for (const s of samples) {
+			const r = trackPlay(state, s);
+			state = r.state;
+			if (r.finished) out.push(r.finished);
+		}
+		return out;
+	};
+	const playFor = (id: number, fromMs: number, toMs: number, t0: number) => {
+		const out = [];
+		for (let p = fromMs; p <= toMs; p += 5_000) out.push(at(id, p, t0 + (p - fromMs)));
+		return out;
+	};
+
+	it('records the previous track once on a track change', () => {
+		const plays = run([...playFor(1, 0, 60_000, 0), at(2, 2_000, 65_000)]);
+		expect(plays).toHaveLength(1);
+		expect(plays[0]).toMatchObject({ trackId: 1, endedAt: 60_000 });
+		expect(plays[0].listenedMs).toBe(60_000);
+	});
+
+	it('skips tracks played under 30 s', () => {
+		expect(run([...playFor(1, 0, 20_000, 0), at(2, 1_000, 25_000)])).toHaveLength(0);
+	});
+
+	it('counts half of a short track', () => {
+		expect(run([...playFor(9, 0, 20_000, 0), at(2, 1_000, 25_000)])).toHaveLength(1);
+	});
+
+	it('ignores pauses and forward seeks', () => {
+		const plays = run([
+			at(1, 0, 0),
+			at(1, 5_000, 5_000),
+			at(1, 5_000, 600_000, false), // paused ten minutes
+			at(1, 10_000, 605_000),
+			at(1, 150_000, 610_000), // seek forward
+			at(1, 155_000, 615_000),
+			at(1, 160_000, 620_000),
+			at(1, 165_000, 625_000),
+			at(2, 0, 630_000)
+		]);
+		// 5 + 0 (paused) + 5 + 6 (seek, capped at wall + 1 s) + 5 + 5 + 5
+		expect(plays).toHaveLength(1);
+		expect(plays[0].listenedMs).toBe(31_000);
+		expect(plays[0].endedAt).toBe(625_000);
+	});
+
+	it('records a repeat of the same track', () => {
+		const plays = run([...playFor(1, 0, 195_000, 0), ...playFor(1, 0, 40_000, 200_000)]);
+		expect(plays).toHaveLength(1);
+		expect(plays[0].listenedMs).toBe(195_000);
+	});
+
+	it('records the last track when playback stops', () => {
+		const stop = {
+			trackId: null,
+			track: null,
+			isPlaying: false,
+			progressMs: null,
+			sampledAt: 70_000
+		};
+		expect(run([...playFor(1, 0, 60_000, 0), stop])).toHaveLength(1);
+	});
+});

src/lib/music/now-playing.ts

@@ -79,3 +79,72 @@ export function nextDelay(isPlaying: boolean, idleTicks: number) {
 	if (isPlaying) return 5_000;
 	return Math.min(30_000, 10_000 + idleTicks * 5_000);
 }
+
+// Backup play log, independent of recently-played (which Spotify can rate-limit for hours):
+// follow the polled track's progress and, when the track changes, starts again or playback
+// stops, emit the previous one if it played for 30 s or more, or at least half its length.
+// Listening time only counts progress that advanced no faster than the wall clock, so
+// pauses and forward seeks add nothing and backward seeks are not counted twice.
+export const MIN_PLAY_MS = 30_000;
+const RESTART_MS = 10_000;
+
+export type TrackedPlay = {
+	key: string;
+	trackId: number;
+	durationMs: number | null;
+	listenedMs: number;
+	/** Last time (server ms) it was seen playing: the play's end, like Spotify's played_at. */
+	endedAt: number;
+};
+
+export type PlayTracker = { current: TrackedPlay | null; last: PlayerSample | null };
+export const newPlayTracker = (): PlayTracker => ({ current: null, last: null });
+
+export const countsAsPlay = (p: TrackedPlay) =>
+	p.listenedMs >= MIN_PLAY_MS || (!!p.durationMs && p.listenedMs >= p.durationMs / 2);
+
+const sampleKey = (s: PlayerSample) =>
+	s.trackId == null ? null : (s.track?.spotifyId ?? `id:${s.trackId}`);
+
+/** Feed one player sample; `finished` is the previous track when it qualifies as a play. */
+export function trackPlay(
+	state: PlayTracker,
+	next: PlayerSample
+): { state: PlayTracker; finished: TrackedPlay | null } {
+	const prev = state.last;
+	const key = sampleKey(next);
+	const wall = prev ? Math.max(0, next.sampledAt - prev.sampledAt) : 0;
+	const start = (): TrackedPlay | null =>
+		key == null || next.trackId == null
+			? null
+			: {
+					key,
+					trackId: next.trackId,
+					durationMs: next.track?.durationMs ?? null,
+					// Started since the previous sample: count what it played in between.
+					listenedMs: prev && next.isPlaying ? Math.min(next.progressMs ?? 0, wall) : 0,
+					endedAt: next.sampledAt
+				};
+	let ended: TrackedPlay | null = null;
+	let current = state.current;
+	if (!current || current.key !== key) {
+		ended = current;
+		current = start();
+	} else if (prev?.progressMs != null && next.progressMs != null) {
+		const diff = next.progressMs - prev.progressMs;
+		if (diff < -SEEK_TOLERANCE_MS && next.progressMs < RESTART_MS) {
+			ended = current; // the same track started again (repeat)
+			current = start();
+		} else if (prev.isPlaying || next.isPlaying) {
+			current = {
+				...current,
+				listenedMs: current.listenedMs + Math.min(Math.max(0, diff), wall + 1_000)
+			};
+		}
+	}
+	if (current && next.isPlaying) current = { ...current, endedAt: next.sampledAt };
+	return {
+		state: { current, last: next },
+		finished: ended && countsAsPlay(ended) ? ended : null
+	};
+}

src/lib/music/store.server.ts

@@ -7,7 +7,8 @@ import type { ParsedAlbum, ParsedArtist, ParsedTrack } from './spotify-parse';
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
 export type Db = Sql<any>;
 export type PlaySource = 'spotify' | 'lastfm' | 'youtube' | 'local';
-export type PlayOrigin = 'api' | 'export' | 'import' | 'manual';
+/** 'nowplaying': the poller's backup log from now-playing samples (see trackPlay). */
+export type PlayOrigin = 'api' | 'export' | 'import' | 'manual' | 'nowplaying';
 
 /** Per-run cache so a poll or import does not look up the same artist/album/track twice. */
 export type StoreCache = Map<string, number>;

src/lib/server/db/schema/music.ts

@@ -110,8 +110,9 @@ export const musicPlays = musicSchema.table(
 		 * usually when the track started. See the music.plays_deduped view. */
 		playedAt: timestamptz('played_at').notNull(),
 		source: text('source').notNull().default('spotify'),
-		/** How the row got here: 'api' (poller), 'export' (Spotify streaming history),
-		 * 'import' (Last.fm and other backfills), 'manual'. */
+		/** How the row got here: 'api' (poller, recently-played), 'nowplaying' (poller backup
+		 * log from now-playing samples, until recently-played confirms it), 'export' (Spotify
+		 * streaming history), 'import' (Last.fm and other backfills), 'manual'. */
 		origin: text('origin').notNull().default('api'),
 		/** Known for imports; null for API plays (Spotify only reports plays of 30 s or more). */
 		msPlayed: integer('ms_played'),
@@ -126,7 +127,10 @@ export const musicPlays = musicSchema.table(
 		index('plays_source_played_idx').on(t.source, t.playedAt),
 		index('plays_track_played_idx').on(t.trackId, t.playedAt),
 		check('plays_source_check', sql`${t.source} in ('spotify', 'lastfm', 'youtube', 'local')`),
-		check('plays_origin_check', sql`${t.origin} in ('api', 'export', 'import', 'manual')`)
+		check(
+			'plays_origin_check',
+			sql`${t.origin} in ('api', 'export', 'import', 'manual', 'nowplaying')`
+		)
 	]
 );
 

src/lib/server/music/poller.ts

@@ -6,11 +6,19 @@ import {
 	handlePollError,
 	pollPlayer,
 	pollRecent,
+	recordBackupPlay,
 	recordPoll,
 	SpotifyError
 } from '#lib/music/collector.server.js';
 import { newCache } from '#lib/music/store.server.js';
-import { detectChange, nextDelay, toLive, type PlayerSample } from '#lib/music/now-playing.js';
+import {
+	detectChange,
+	newPlayTracker,
+	nextDelay,
+	toLive,
+	trackPlay,
+	type PlayerSample
+} from '#lib/music/now-playing.js';
 import { currentLive, liveSubscribers, publishLive } from './live';
 import { getLiveFromDb } from './stats';
 
@@ -25,7 +33,9 @@ import { getLiveFromDb } from './stats';
 // every minute, plus ~10 s after a track change so a finished track lands quickly.
 // Spotify rate-limits endpoints separately (recently-played can be blocked for hours while
 // currently-playing still answers), so a 429 on the play log only pauses the play log, for
-// the full Retry-After, and never slows the now-playing card down.
+// the full Retry-After, and never slows the now-playing card down. Meanwhile the backup log
+// (trackPlay → origin 'nowplaying') records finished tracks from the now-playing samples;
+// recently-played later takes those rows over with its exact timestamp.
 const LEASE_SECONDS = 20; // renewed every tick; another container takes over 20 s after we stop
 const RECENT_MS = 60_000;
 const CATCH_UP_MS = 10_000;
@@ -48,6 +58,7 @@ let catchUps = 0;
 let recentBlockedUntil = 0;
 let recentError: string | null = null;
 let cache = newCache();
+let tracker = newPlayTracker();
 
 export const spotifyCredentials = () => ({
 	clientId: SPOTIFY_CLIENT_ID,
@@ -81,6 +92,7 @@ export async function tick(): Promise<number> {
 	const auth = await acquireLease(client, HOLDER, LEASE_SECONDS);
 	if (!auth) {
 		last = null;
+		tracker = newPlayTracker();
 		await followDatabase().catch(() => {});
 		return 10_000;
 	}
@@ -97,6 +109,10 @@ export async function tick(): Promise<number> {
 			catchUps = 2;
 		}
 		last = sample;
+		// Backup play log: works even while recently-played is rate-limited.
+		const tracked = trackPlay(tracker, sample);
+		tracker = tracked.state;
+		if (tracked.finished) await recordBackupPlay(client, tracked.finished);
 		idleTicks = sample.isPlaying ? 0 : idleTicks + 1;
 		if (Date.now() >= nextRecentAt && Date.now() >= recentBlockedUntil) {
 			try {