music: near-realtime now playing (currently-playing every 5 s while playing, 30 s idle backoff, Retry-After, recent plays ~10 s after a track change, holder-based lease) and a public SSE stream /api/listening/stream feeding the listening card live with client-side progress

src/lib/music/collector.server.ts

@@ -1,12 +1,14 @@
-// Spotify poller: one call does what a scrobbler does every minute.
-//  1. GET /me/player             → music.now_playing (row 1)
+// Spotify poller building blocks:
+//  1. GET /me/player/currently-playing → music.now_playing (row 1); the app polls this every
+//     5 s while music plays (see src/lib/server/music/poller.ts) and pushes changes live
 //  2. GET /me/player/recently-played?after=<cursor> → music.plays (Spotify lists plays of
-//     30 s or more, max 50 per call, so polling every minute never misses one)
+//     30 s or more, max 50 per call), every minute and shortly after each track change
 // The authorization (refresh token) lives in music.spotify_auth, written by
-// /api/spotify/callback. A lease column keeps two app containers (e.g. mid-deploy) or the
-// optional host timer (scripts/music-poll.ts) from polling at the same time.
+// /api/spotify/callback. A lease (lease_until + lease_holder) keeps two app containers (e.g.
+// mid-deploy) or the optional host timer (scripts/music-poll.ts) from polling at once.
 import { newCache, upsertTrack, insertPlay, type Db } from './store.server';
 import { parsePlayer, parseRecentlyPlayed } from './spotify-parse';
+import type { PlayerSample } from './now-playing';
 
 export type SpotifyCredentials = { clientId: string; clientSecret: string };
 type Fetch = typeof fetch;
@@ -25,7 +27,9 @@ export class SpotifyError extends Error {
 	constructor(
 		readonly status: number,
 		message: string,
-		readonly code?: string
+		readonly code?: string,
+		/** Seconds from a 429 Retry-After header. */
+		readonly retryAfter?: number
 	) {
 		super(message);
 	}
@@ -82,7 +86,9 @@ export async function spotifyGet(fetchFn: Fetch, access: string, path: string) {
 	if (!response.ok)
 		throw new SpotifyError(
 			response.status,
-			`Spotify GET /${path.split('?')[0]} → ${response.status}`
+			`Spotify GET /${path.split('?')[0]} → ${response.status}`,
+			undefined,
+			Number(response.headers.get('retry-after')) || undefined
 		);
 	const text = await response.text();
 	return text ? (JSON.parse(text) as unknown) : null;
@@ -93,88 +99,166 @@ export type PollResult =
 	| { status: 'ok'; playing: boolean; newPlays: number }
 	| { status: 'error'; error: string };
 
+export type SpotifyAuthRow = {
+	refresh_token: string;
+	access_token: string | null;
+	access_expires_at: string | Date | null;
+	recent_cursor: string | number | null;
+};
+
+/**
+ * Take or renew the single-poller lease. Null when Spotify is not connected (or needs
+ * re-auth) or another poller holds an unexpired lease.
+ */
+export async function acquireLease(
+	sql: Db,
+	holder: string,
+	seconds: number
+): Promise<SpotifyAuthRow | null> {
+	const [auth] = await sql`
+		update music.spotify_auth
+		set lease_until = now() + make_interval(secs => ${seconds}), lease_holder = ${holder}
+		where id = 1 and not needs_reauth
+		  and (lease_until is null or lease_until < now() or lease_holder = ${holder})
+		returning refresh_token, access_token, access_expires_at, recent_cursor`;
+	return (auth as SpotifyAuthRow | undefined) ?? null;
+}
+
+export async function releaseLease(sql: Db, holder: string) {
+	await sql`update music.spotify_auth set lease_until = null
+		where id = 1 and lease_holder = ${holder}`;
+}
+
+export async function recordPoll(sql: Db, error: string | null) {
+	await sql`update music.spotify_auth set last_poll_at = now(), last_error = ${error} where id = 1`;
+}
+
+/** A valid access token, refreshing (and storing) it when it expires within a minute. */
+export async function ensureAccess(
+	sql: Db,
+	creds: SpotifyCredentials,
+	auth: SpotifyAuthRow,
+	fetchFn: Fetch = fetch
+) {
+	const expires = auth.access_expires_at ? new Date(auth.access_expires_at).getTime() : 0;
+	if (auth.access_token && expires >= Date.now() + 60_000) return auth.access_token;
+	const fresh = await refreshAccessToken(fetchFn, creds, auth.refresh_token);
+	await sql`
+		update music.spotify_auth set access_token = ${fresh.access_token},
+			access_expires_at = now() + make_interval(secs => ${fresh.expires_in}),
+			refresh_token = coalesce(${fresh.refresh_token ?? null}, refresh_token),
+			scope = coalesce(${fresh.scope ?? null}, scope), updated_at = now()
+		where id = 1`;
+	return fresh.access_token;
+}
+
+export const CURRENTLY_PLAYING = 'me/player/currently-playing?additional_types=episode';
+export const PLAYER = 'me/player?additional_types=episode';
+
+/** Sample the player and store it in music.now_playing. */
+export async function pollPlayer(
+	sql: Db,
+	access: string,
+	cache = newCache(),
+	path = CURRENTLY_PLAYING,
+	fetchFn: Fetch = fetch
+): Promise<PlayerSample> {
+	const sampledAt = Date.now();
+	const player = parsePlayer(await spotifyGet(fetchFn, access, path));
+	const trackId = player?.track ? await upsertTrack(sql, player.track, true, cache) : null;
+	// currently-playing has no device: keep the last known one (from /me/player) then.
+	await sql`
+		insert into music.now_playing (id, track_id, is_playing, progress_ms, device_name, device_type, context_uri, updated_at)
+		values (1, ${trackId}, ${!!(player?.isPlaying && trackId)}, ${player?.progressMs ?? null},
+		        ${player?.deviceName ?? null}, ${player?.deviceType ?? null}, ${player?.contextUri ?? null}, now())
+		on conflict (id) do update set
+			track_id = coalesce(excluded.track_id, music.now_playing.track_id),
+			is_playing = excluded.is_playing, progress_ms = excluded.progress_ms,
+			device_name = coalesce(excluded.device_name, music.now_playing.device_name),
+			device_type = coalesce(excluded.device_type, music.now_playing.device_type),
+			context_uri = excluded.context_uri, updated_at = now()`;
+	return {
+		trackId,
+		track: player?.track ?? null,
+		isPlaying: !!(player?.isPlaying && trackId),
+		progressMs: player?.progressMs ?? null,
+		sampledAt
+	};
+}
+
+/** Store plays newer than the cursor; returns how many were new. */
+export async function pollRecent(
+	sql: Db,
+	access: string,
+	cache = newCache(),
+	fetchFn: Fetch = fetch
+): Promise<number> {
+	const [row] = await sql`select recent_cursor from music.spotify_auth where id = 1`;
+	const cursor: number | null = row?.recent_cursor == null ? null : Number(row.recent_cursor);
+	const recent = parseRecentlyPlayed(
+		await spotifyGet(
+			fetchFn,
+			access,
+			`me/player/recently-played?limit=50${cursor ? `&after=${cursor}` : ''}`
+		)
+	);
+	let newPlays = 0;
+	let newest = cursor ?? 0;
+	for (const play of recent) {
+		const at = play.playedAt.getTime();
+		if (cursor && at <= cursor) continue;
+		const id = await upsertTrack(sql, play.track, true, cache);
+		if (
+			await insertPlay(sql, {
+				trackId: id,
+				playedAt: play.playedAt,
+				source: 'spotify',
+				origin: 'api',
+				contextType: play.contextType,
+				contextUri: play.contextUri
+			})
+		)
+			newPlays++;
+		newest = Math.max(newest, at);
+	}
+	if (newest && newest !== cursor)
+		await sql`update music.spotify_auth set recent_cursor = ${newest} where id = 1`;
+	return newPlays;
+}
+
+/** Record a failed poll: revoked grants stop polling, a stale access token is dropped. */
+export async function handlePollError(sql: Db, e: unknown) {
+	if (e instanceof SpotifyError && e.code === 'invalid_grant')
+		// The refresh token was revoked: stop polling until /api/spotify/connect is used again.
+		await sql`update music.spotify_auth set needs_reauth = true where id = 1`;
+	else if (e instanceof SpotifyError && e.status === 401)
+		await sql`update music.spotify_auth set access_token = null where id = 1`;
+	return e instanceof Error ? e.message : String(e);
+}
+
+/** One full poll (player + recent plays) for the optional host timer, scripts/music-poll.ts. */
 export async function pollSpotify(
 	sql: Db,
 	creds: SpotifyCredentials,
-	fetchFn: Fetch = fetch
+	fetchFn: Fetch = fetch,
+	holder = 'host-timer'
 ): Promise<PollResult> {
-	const [auth] = await sql`
-		update music.spotify_auth set lease_until = now() + interval '55 seconds'
-		where id = 1 and not needs_reauth and (lease_until is null or lease_until < now())
-		returning refresh_token, access_token, access_expires_at, recent_cursor`;
+	const auth = await acquireLease(sql, holder, 55);
 	if (!auth) return { status: 'idle' };
 	let error: string | null = null;
 	let result: PollResult;
 	try {
-		let access: string | null = auth.access_token;
-		const expires = auth.access_expires_at ? new Date(auth.access_expires_at).getTime() : 0;
-		if (!access || expires < Date.now() + 60_000) {
-			const fresh = await refreshAccessToken(fetchFn, creds, auth.refresh_token);
-			access = fresh.access_token;
-			await sql`
-				update music.spotify_auth set access_token = ${access},
-					access_expires_at = now() + make_interval(secs => ${fresh.expires_in}),
-					refresh_token = coalesce(${fresh.refresh_token ?? null}, refresh_token),
-					scope = coalesce(${fresh.scope ?? null}, scope), updated_at = now()
-				where id = 1`;
-		}
+		const access = await ensureAccess(sql, creds, auth, fetchFn);
 		const cache = newCache();
-
-		const player = parsePlayer(
-			await spotifyGet(fetchFn, access!, 'me/player?additional_types=episode')
-		);
-		const trackId = player?.track ? await upsertTrack(sql, player.track, true, cache) : null;
-		await sql`
-			insert into music.now_playing (id, track_id, is_playing, progress_ms, device_name, device_type, context_uri, updated_at)
-			values (1, ${trackId}, ${!!(player?.isPlaying && trackId)}, ${player?.progressMs ?? null},
-			        ${player?.deviceName ?? null}, ${player?.deviceType ?? null}, ${player?.contextUri ?? null}, now())
-			on conflict (id) do update set
-				track_id = coalesce(excluded.track_id, music.now_playing.track_id),
-				is_playing = excluded.is_playing, progress_ms = excluded.progress_ms,
-				device_name = excluded.device_name, device_type = excluded.device_type,
-				context_uri = excluded.context_uri, updated_at = now()`;
-
-		const cursor: number | null = auth.recent_cursor === null ? null : Number(auth.recent_cursor);
-		const recent = parseRecentlyPlayed(
-			await spotifyGet(
-				fetchFn,
-				access!,
-				`me/player/recently-played?limit=50${cursor ? `&after=${cursor}` : ''}`
-			)
-		);
-		let newPlays = 0;
-		let newest = cursor ?? 0;
-		for (const play of recent) {
-			const at = play.playedAt.getTime();
-			if (cursor && at <= cursor) continue;
-			const id = await upsertTrack(sql, play.track, true, cache);
-			if (
-				await insertPlay(sql, {
-					trackId: id,
-					playedAt: play.playedAt,
-					source: 'spotify',
-					origin: 'api',
-					contextType: play.contextType,
-					contextUri: play.contextUri
-				})
-			)
-				newPlays++;
-			newest = Math.max(newest, at);
-		}
-		if (newest && newest !== cursor)
-			await sql`update music.spotify_auth set recent_cursor = ${newest} where id = 1`;
-		result = { status: 'ok', playing: !!player?.isPlaying, newPlays };
+		const sample = await pollPlayer(sql, access, cache, PLAYER, fetchFn);
+		const newPlays = await pollRecent(sql, access, cache, fetchFn);
+		result = { status: 'ok', playing: sample.isPlaying, newPlays };
 	} catch (e) {
-		error = e instanceof Error ? e.message : String(e);
+		error = await handlePollError(sql, e);
 		result = { status: 'error', error };
-		if (e instanceof SpotifyError && e.code === 'invalid_grant')
-			// The refresh token was revoked: stop polling until /api/spotify/connect is used again.
-			await sql`update music.spotify_auth set needs_reauth = true where id = 1`;
-		else if (e instanceof SpotifyError && e.status === 401)
-			await sql`update music.spotify_auth set access_token = null where id = 1`;
 	} finally {
-		await sql`update music.spotify_auth set lease_until = null, last_poll_at = now(),
-			last_error = ${error} where id = 1`;
+		await recordPoll(sql, error);
+		await releaseLease(sql, holder);
 	}
 	return result;
 }