website/src/lib/music/collector.server.ts

// 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), every minute and shortly after each track change
// The authorization (refresh token) lives in music.spotify_auth, written by
// /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;

export const SPOTIFY_SCOPES = [
	'user-read-currently-playing',
	'user-read-recently-played',
	'user-read-playback-state',
	'user-top-read',
	// Map finder: liked songs and Stan's own playlists.
	'user-library-read',
	'playlist-read-private'
] as const;

export class SpotifyError extends Error {
	constructor(
		readonly status: number,
		message: string,
		readonly code?: string,
		/** Seconds from a 429 Retry-After header. */
		readonly retryAfter?: number
	) {
		super(message);
	}
}

type TokenResponse = {
	access_token: string;
	expires_in: number;
	refresh_token?: string;
	scope?: string;
};

async function token(
	fetchFn: Fetch,
	creds: SpotifyCredentials,
	body: Record<string, string>
): Promise<TokenResponse> {
	const response = await fetchFn('https://accounts.spotify.com/api/token', {
		method: 'POST',
		headers: {
			Authorization: `Basic ${btoa(`${creds.clientId}:${creds.clientSecret}`)}`,
			'Content-Type': 'application/x-www-form-urlencoded'
		},
		body: new URLSearchParams(body),
		signal: AbortSignal.timeout(15_000)
	});
	const json = (await response.json().catch(() => ({}))) as Record<string, unknown>;
	if (!response.ok || typeof json.access_token !== 'string')
		throw new SpotifyError(
			response.status,
			`Spotify token request failed (${response.status} ${String(json.error ?? '')})`,
			typeof json.error === 'string' ? json.error : undefined
		);
	return json as unknown as TokenResponse;
}

export const exchangeCode = (
	fetchFn: Fetch,
	creds: SpotifyCredentials,
	code: string,
	redirectUri: string
) => token(fetchFn, creds, { grant_type: 'authorization_code', code, redirect_uri: redirectUri });

export const refreshAccessToken = (fetchFn: Fetch, creds: SpotifyCredentials, refresh: string) =>
	token(fetchFn, creds, { grant_type: 'refresh_token', refresh_token: refresh });

/** GET a Web API path; null for 204 No Content. */
export async function spotifyGet(fetchFn: Fetch, access: string, path: string) {
	const response = await fetchFn(`https://api.spotify.com/v1/${path}`, {
		headers: { Authorization: `Bearer ${access}` },
		signal: AbortSignal.timeout(15_000)
	});
	if (response.status === 204) return null;
	if (!response.ok)
		throw new SpotifyError(
			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;
}

export type PollResult =
	| { status: 'idle' }
	| { 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,
	holder = 'host-timer'
): Promise<PollResult> {
	const auth = await acquireLease(sql, holder, 55);
	if (!auth) return { status: 'idle' };
	let error: string | null = null;
	let result: PollResult;
	try {
		const access = await ensureAccess(sql, creds, auth, fetchFn);
		const cache = newCache();
		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 = await handlePollError(sql, e);
		result = { status: 'error', error };
	} finally {
		await recordPoll(sql, error);
		await releaseLease(sql, holder);
	}
	return result;
}