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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216/** * Fill discogs_releases for every album with no Discogs link yet. * * Walks albums where discogs_release_id is null, oldest first, and asks the * Discogs service for each artist+album. Matches are stored and the album is * linked; misses are recorded in discogs_searches so a later run skips them. * * Pacing matters: the service is capped at Discogs' 60 requests per minute per * token, and one enrichment costs a search plus a release fetch (plus a master * for reissues). So this asks for at most BACKFILL_ALBUMS_PER_MINUTE albums a * minute rather than racing the service's queue and collecting 429s, which * also leaves quota for live scrobbles sharing the same token. When the * service cannot answer at all, that album is retried after a cooldown instead * of being recorded as a miss. * * Usage (also wired as `bun backfill:discogs`): * tsx ./src/scripts/backfill-discogs.ts [--cursor <offset>] * * --cursor is a zero-based offset into currently unlinked albums (default 0). * Example: --cursor 15829 skips 15,829 pending albums. Matches remove albums * from this list, so offsets from an earlier run may refer to different albums. * * Env: * DISCOGS_ENRICHMENT_ENABLED "true" — enable enrichment (default false) * BACKFILL_ALBUMS_PER_MINUTE albums asked about per minute (default 18) * BACKFILL_LIMIT stop after this many albums (default: all) * BACKFILL_DRY_RUN "1" — report what would be asked, write nothing */
import chalk from "chalk";import { consola } from "consola";import { ctx } from "context";import { and, asc, count, isNull, sql } from "drizzle-orm";import { parseArgs } from "node:util";import { enrichAlbumWithDiscogs } from "lib/discogsEnrichment";import { env } from "lib/env";import tables from "schema";import { retryPgConnection } from "lib/pgConnectionRecovery";
const ALBUMS_PER_MINUTE = Number(process.env.BACKFILL_ALBUMS_PER_MINUTE ?? 18);const LIMIT = process.env.BACKFILL_LIMIT ? Number(process.env.BACKFILL_LIMIT) : Number.POSITIVE_INFINITY;const DRY_RUN = process.env.BACKFILL_DRY_RUN === "1";
const SPACING_MS = Math.ceil(60_000 / Math.max(ALBUMS_PER_MINUTE, 1));const PAGE_SIZE = 200;const MAX_ATTEMPTS = 3;const UNAVAILABLE_COOLDOWN_MS = 60_000;
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, Math.max(ms, 0)));
const formatEta = (seconds: number) => { if (!Number.isFinite(seconds)) return "unknown"; const m = Math.floor(seconds / 60); const s = Math.round(seconds % 60); return m > 0 ? `${m}m ${s}s` : `${s}s`;};
const retryDb = <T>(label: string, run: () => PromiseLike<T>) => retryPgConnection(run, { onRetry: (attempt, delayMs) => consola.warn( `${label}: database connection lost, retry ${attempt}/5 in ${delayMs / 1000}s`, ), });
// Compare in Postgres to preserve timestamp precision, including microseconds.// The ID breaks ties for albums created at the same timestamp.const pendingAfter = (cursor?: string) => and( isNull(tables.albums.discogsReleaseId), cursor ? sql`(${tables.albums.createdAt}, ${tables.albums.id}) > (SELECT xata_createdat, xata_id FROM albums WHERE xata_id = ${cursor})` : undefined, );
async function main() { const { values } = parseArgs({ args: process.argv.slice(2), options: { cursor: { type: "string" } }, strict: true, allowPositionals: false, }); const rawOffset = values.cursor ?? "0"; const offset = Number(rawOffset); if (!/^\d+$/.test(rawOffset) || !Number.isSafeInteger(offset)) { throw new Error("--cursor must be a non-negative integer offset"); } if (!env.DISCOGS_ENRICHMENT_ENABLED) { consola.info( "Discogs enrichment is disabled. Set DISCOGS_ENRICHMENT_ENABLED=true to run the backfill.", ); process.exit(0); }
const pending = await retryDb("Count pending albums", () => ctx.db .select({ count: count() }) .from(tables.albums) .where(isNull(tables.albums.discogsReleaseId)) .then((rows) => rows[0]?.count ?? 0), );
const target = Math.min(Math.max(0, pending - offset), LIMIT); consola.info( `${pending} album(s) without a Discogs link · starting offset ${offset} · asking ${ALBUMS_PER_MINUTE}/min · ETA ${formatEta((target * SPACING_MS) / 1000)}${DRY_RUN ? " (dry run)" : ""}`, );
let processed = 0; let matched = 0; let missed = 0; let failed = 0; let skipped = 0; // Keyset cursor: a matched album drops out of the filter, but a miss stays // in it, so paging has to move past what has been asked about. let cursor: string | undefined; const startedAt = Date.now();
while (processed < target) { const albums = await retryDb("Read album page", () => ctx.db .select() .from(tables.albums) .where(pendingAfter(cursor)) .orderBy(asc(tables.albums.createdAt), asc(tables.albums.id)) .limit(PAGE_SIZE) // Apply the numeric offset once; matches shrink the pending set. .offset(cursor === undefined ? offset : 0), );
if (albums.length === 0) { break; }
for (const album of albums) { if (processed >= target) break;
const label = chalk.cyan(`${album.artist} - ${album.title}`); if (DRY_RUN) { consola.info(`[${offset + processed + 1}/${offset + target}] would ask for ${label}`); cursor = album.id; processed++; continue; }
// Retrying the entire enrichment replays its upserts and transactional // replacements safely; a failed transaction itself must never be reused. const enrich = () => retryDb(`Discogs album ${album.id}`, () => enrichAlbumWithDiscogs(ctx, album.artist, album.title, album.id), ); let result = await enrich(); for ( let attempt = 1; result.status === "unavailable" && attempt < MAX_ATTEMPTS; attempt++ ) { consola.warn( `discogs unavailable, retrying ${label} in ${formatEta(UNAVAILABLE_COOLDOWN_MS / 1000)}`, ); await sleep(UNAVAILABLE_COOLDOWN_MS); result = await enrich(); }
cursor = album.id; processed++; switch (result.status) { case "matched": matched++; consola.success( `[${offset + processed}/${offset + target}] ${label} → ${chalk.green(result.release.discogsId)}`, ); break; case "missed": missed++; consola.info(`[${offset + processed}/${offset + target}] ${label} → no match`); break; case "skipped": skipped++; consola.info(`[${offset + processed}/${offset + target}] ${label} → incomplete album`); break; default: failed++; consola.error( `[${offset + processed}/${offset + target}] ${label} → discogs unavailable`, ); break; }
const elapsed = (Date.now() - startedAt) / 1000; const rate = processed / elapsed; if (processed % 25 === 0) { consola.info( `progress: ${processed}/${target} · ${matched} matched · ${missed} without a match · ${failed} unavailable · ${(rate * 60).toFixed(1)}/min · ETA ${formatEta((target - processed) / Math.max(rate, 1e-9))}`, ); }
await sleep(SPACING_MS); } }
const elapsed = (Date.now() - startedAt) / 1000; consola.success( `done in ${formatEta(elapsed)}: ${processed} album(s) · ${matched} matched · ${missed} without a match · ${skipped} incomplete · ${failed} unavailable${DRY_RUN ? " (dry run — nothing written)" : ""}`, ); process.exit(0);}
main().catch((e) => { consola.error(e); process.exit(1);});