Something went wrong. Try again.
A decentralized music tracking and discovery platform built on AT Protocol 🎵 rocksky.app
spotify atproto lastfm musicbrainz scrobbling listenbrainz
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431import { matchTrack, matchingRequest } from "lib/matching";import { consola } from "consola";import type { Context } from "context";import { JSONCodec } from "nats";import { withFallbackAlbumArt } from "lib";import { createAgent } from "lib/agent";import type * as Status from "lexicon/types/app/rocksky/actor/status";import type { TrackView } from "lexicon/types/app/rocksky/actor/defs";import { remainingPlaybackMs } from "./playback";
const jc = JSONCodec();
interface SongChangedPayload { did: string; track: { name?: string; id?: string; external_urls?: { spotify?: string }; external_ids?: { isrc?: string }; // Spotify: structured artists + album object artists?: { id: string; name: string }[]; album?: { id: string; name: string; cover?: string } | string; duration_ms?: number; progress_ms?: number; // ListenBrainz / websocket: flat strings artist?: string; album_name?: string; albumCoverUrl?: string; recording_mb_id?: string; source?: string; // Track position within its album, when the source provides it (either // naming convention). Often absent or 0, in which case Deezer fills it. track_number?: number; trackNumber?: number; };}
interface SongStoppedPayload { did: string;}
async function resolveTrackInfo( ctx: Context, raw: SongChangedPayload["track"],): Promise<{ recordingMbId: string | undefined; albumArt: string | undefined; trackNumber: number | undefined;}> { const name = raw.name ?? ""; const artist = raw.artists?.[0]?.name ?? raw.artist ?? ""; const album = (typeof raw.album === "object" ? raw.album?.name : raw.album) ?? raw.album_name ?? "";
// 1. Already provided in the payload (e.g. from ListenBrainz) let recordingMbId: string | undefined = raw.recording_mb_id; let albumArt: string | undefined; // Only accept a positive track number from the payload; 0 / missing means // "unknown" and should be resolved from the DB or Deezer below. const payloadTrackNumber = raw.track_number ?? raw.trackNumber; let trackNumber: number | undefined = payloadTrackNumber && payloadTrackNumber > 0 ? payloadTrackNumber : undefined;
if (!name || !artist) return { recordingMbId, albumArt, trackNumber };
try { const candidate = await matchTrack( ctx, matchingRequest({ title: name, artist, artists: raw.artists, album, duration: raw.duration_ms, mbId: recordingMbId, isrc: raw.external_ids?.isrc, spotifyLink: raw.external_urls?.spotify ?? (raw.source === "spotify" || raw.artists ? raw.id : undefined), }), ); if (candidate) { recordingMbId ??= candidate.identifiers.musicbrainzRecordingId ?? undefined; albumArt = candidate.albumArt ?? undefined; trackNumber ??= candidate.trackNumber ?? undefined; } } catch (error) { consola.warn( "[status] Matching unavailable; preserving source metadata", error instanceof Error ? error.message : error, ); }
return { recordingMbId, albumArt, trackNumber };}
function normalizeTrack( raw: SongChangedPayload["track"], recordingMbId: string | undefined, dbAlbumArt?: string, trackNumber?: number,): TrackView { // Spotify emits: name, artists[], album.{name,cover}, duration_ms // ListenBrainz/websocket emits: name, artist, album (string), duration_ms, albumCoverUrl const name = raw.name ?? ""; const artist = raw.artists?.[0]?.name ?? raw.artist ?? ""; const album = (typeof raw.album === "object" ? raw.album?.name : raw.album) ?? raw.album_name; const albumCoverUrl = withFallbackAlbumArt( (typeof raw.album === "object" ? raw.album?.cover : undefined) ?? raw.albumCoverUrl ?? dbAlbumArt, ); const durationMs = raw.duration_ms ?? 0; const source = raw.source ?? (raw.artists ? "spotify" : "listenbrainz");
return { name, artist, album, albumCoverUrl, durationMs, source, recordingMbId, // Only emit a track number when it's a valid position (>0); the lexicon // requires minimum 1, so never publish 0. ...(trackNumber && trackNumber > 0 ? { trackNumber } : {}), };}
async function getSwapCid( agent: Awaited<ReturnType<typeof createAgent>>,): Promise<string | undefined> { try { const res = await agent!.com.atproto.repo.getRecord({ repo: agent!.assertDid, collection: "app.rocksky.actor.status", rkey: "self", }); return res.data.cid; } catch (err: any) { const status = err?.response?.status ?? err?.status; if (status === 400 || status === 404) return undefined; throw err; }}
function rateLimitResetMsg(err: any): string { const headers = err?.headers ?? err?.response?.headers ?? err?.error?.headers; const reset = headers?.["ratelimit-reset"] ?? headers?.["x-ratelimit-reset"]; const retryAfter = headers?.["retry-after"]; if (reset) { const resetDate = new Date(Number(reset) * 1000); return ` — rate limit resets at ${resetDate.toISOString()} (in ${Math.max(0, Math.round((resetDate.getTime() - Date.now()) / 1000))}s)`; } if (retryAfter) { return ` — retry after ${retryAfter}s`; } return "";}
// Last status successfully written to the PDS per DID.// Tracks key ("name:artist") and source so Navidrome can override a Rockbox// entry for the same track without being deduped, plus when it was written and// how long the track runs, so a stop arriving mid-track can be deferred.const lastPushedStatus = new Map< string, { key: string; source: string; startedAt: number; durationMs: number } | null>();
// Deletes deferred until the current track would have finished, keyed by DID.const pendingDeletes = new Map<string, NodeJS.Timeout>();
function cancelPendingDelete(did: string): void { const timer = pendingDeletes.get(did); if (timer) { clearTimeout(timer); pendingDeletes.delete(did); }}
const remainingMs = (did: string): number => remainingPlaybackMs(lastPushedStatus.get(did), Date.now());
// Unix ms timestamp until which PDS writes are suppressed for a given DID.// Set when a 429 is received; cleared automatically once the time passes.const rateLimitedUntil = new Map<string, number>();
function isRateLimited(did: string): boolean { const until = rateLimitedUntil.get(did); if (!until) return false; if (Date.now() >= until) { rateLimitedUntil.delete(did); return false; } return true;}
function applyRateLimitBackoff(did: string, err: any): void { const headers = err?.headers ?? err?.response?.headers ?? err?.error?.headers; const reset = headers?.["ratelimit-reset"] ?? headers?.["x-ratelimit-reset"]; const retryAfter = headers?.["retry-after"]; let until: number; if (reset) { until = Number(reset) * 1000; } else if (retryAfter) { until = Date.now() + Number(retryAfter) * 1000; } else { until = Date.now() + 60_000; // conservative 1-minute fallback } rateLimitedUntil.set(did, until);}
const STOP_GRACE_MS = 5_000;
async function clearStatus(ctx: Context, did: string): Promise<void> { const agent = await createAgent(ctx.oauthClient, did); if (!agent) { consola.warn(`[status] No agent for ${did}, skipping song.stopped`); return; }
await agent.com.atproto.repo.deleteRecord({ repo: agent.assertDid, collection: "app.rocksky.actor.status", rkey: "self", });
lastPushedStatus.set(did, null); consola.info(`[status] Cleared status for ${did}`);}
export function onSongChanged(ctx: Context) { const sub = ctx.nc.subscribe("rocksky.song.changed"); (async () => { for await (const m of sub) { let did = "(unknown)"; try { const payload = jc.decode(m.data) as SongChangedPayload; did = payload.did; const { track: rawTrack } = payload;
if (isRateLimited(did)) { consola.debug( `[status] song.changed skipped for ${did} — PDS rate limited until ${new Date(rateLimitedUntil.get(did)!).toISOString()}`, ); continue; }
const trackKey = `${rawTrack.name ?? ""}:${rawTrack.artists?.[0]?.name ?? rawTrack.artist ?? ""}`; const source = rawTrack.source ?? (rawTrack.artists ? "spotify" : "listenbrainz"); const last = lastPushedStatus.get(did); // Skip only when the same track was already written by the same source. // A different source (e.g. Navidrome after Rockbox) is always allowed through // so the active player can take ownership of the status record. if (last && last.key === trackKey && last.source === source) { consola.info( `[status] skip unchanged status for ${did}: ${rawTrack.name} (source: ${source})`, ); continue; }
const [agent, { recordingMbId, albumArt, trackNumber }] = await Promise.all([ createAgent(ctx.oauthClient, did), resolveTrackInfo(ctx, rawTrack), ]);
if (!agent) { consola.warn(`[status] No agent for ${did}, skipping song.changed`); continue; }
const track = normalizeTrack( rawTrack, recordingMbId, albumArt, trackNumber, ); const startedAt = new Date().toISOString(); const expiresAt = track.durationMs ? new Date(Date.now() + track.durationMs).toISOString() : undefined;
const record: Status.Record = { $type: "app.rocksky.actor.status", track, startedAt, expiresAt, };
const swapRecord = await getSwapCid(agent); await agent.com.atproto.repo.putRecord({ repo: agent.assertDid, collection: "app.rocksky.actor.status", rkey: "self", record, swapRecord, validate: false, });
lastPushedStatus.set(did, { key: trackKey, source, startedAt: Date.parse(startedAt), durationMs: track.durationMs, }); // A new track owns the record now, so a stop deferred for the previous // one must not fire and delete it. cancelPendingDelete(did);
// When Navidrome is the source, clear ws_lastsong so a stale status=0 // from Rockbox cannot fire song.stopped and delete this record. // Do NOT update lastsong here — Rockbox heartbeats use lastsong to dedup // themselves, and changing it would make Rockbox's last-played track look // "new", causing it to fire song.changed and overwrite this record. if (source === "navidrome") { await ctx.redis.del(`ws_lastsong:${did}`); }
// Auto-expire the in-memory dedup after the track duration so that a // replay of the same song after ATProto record expiry (expiresAt) will // re-write the record rather than being silently deduplicated. if (track.durationMs > 0) { const keySnapshot = trackKey; const sourceSnapshot = source; setTimeout(() => { const current = lastPushedStatus.get(did); if ( current && current.key === keySnapshot && current.source === sourceSnapshot ) { lastPushedStatus.delete(did); } }, track.durationMs + 10_000); }
consola.info( `[status] Updated status for ${did}: ${track.artist} – ${track.name}${recordingMbId ? ` (${recordingMbId})` : ""}`, ); } catch (err: any) { const status = err?.status ?? err?.response?.status ?? err?.error?.status; const message = err?.message ?? err?.error?.message ?? String(err); if (status === 429) { applyRateLimitBackoff(did, err); consola.warn( `[status] song.changed rate limited for ${did}${rateLimitResetMsg(err)}`, ); } else { consola.error( `[status] Error handling song.changed for ${did} — HTTP ${status ?? "?"}: ${message}`, ); } } } })();}
export function onSongStopped(ctx: Context) { const sub = ctx.nc.subscribe("rocksky.song.stopped"); (async () => { for await (const m of sub) { let did = "(unknown)"; try { ({ did } = jc.decode(m.data) as SongStoppedPayload);
if (isRateLimited(did)) { consola.debug( `[status] song.stopped skipped for ${did} — PDS rate limited until ${new Date(rateLimitedUntil.get(did)!).toISOString()}`, ); continue; }
if (lastPushedStatus.get(did) === null) { consola.debug(`[status] skip already-stopped status for ${did}`); continue; }
const remaining = remainingMs(did); if (remaining > 0) { cancelPendingDelete(did); const snapshot = lastPushedStatus.get(did); pendingDeletes.set( did, setTimeout(() => { pendingDeletes.delete(did); // Only clear the status this stop was actually about. if (lastPushedStatus.get(did) !== snapshot) return; clearStatus(ctx, did).catch((err) => consola.error( `[status] deferred clear failed for ${did}: ${err?.message ?? err}`, ), ); }, remaining + STOP_GRACE_MS), ); consola.info( `[status] deferred clear for ${did} — ${Math.round(remaining / 1000)}s of the track left`, ); continue; }
await clearStatus(ctx, did); } catch (err: any) { const status = err?.status ?? err?.response?.status ?? err?.error?.status; if (status === 400 || status === 404) continue; // already gone, not an error const message = err?.message ?? err?.error?.message ?? String(err); if (status === 429) { applyRateLimitBackoff(did, err); consola.warn( `[status] song.stopped rate limited for ${did}${rateLimitResetMsg(err)}`, ); } else { consola.error( `[status] Error handling song.stopped for ${did} — HTTP ${status ?? "?"}: ${message}`, ); } } } })();}