diff --git a/workers/src/index.ts b/workers/src/index.ts index 1fa917e..b0cd93e 100644 --- a/workers/src/index.ts +++ b/workers/src/index.ts @@ -7,7 +7,7 @@ import { IntegrityService } from './integrity.service.js'; import { EnrichmentService } from './enrichment.service.js'; import { AudioFeaturesService } from './audio-features.service.js'; import { CleanupSweepService } from './cleanup.service.js'; -import { withTransaction } from './db.js'; +import { reprocessArtists } from './reprocess-artists.service.js'; // Cron for the periodic integrity sweep (default: daily at 03:00). Configurable // via INTEGRITY_SWEEP_CRON. MUSIC_DIR (consumed by IntegrityService) controls @@ -220,213 +220,11 @@ async function initWorker() { break; } case 'reprocess_artists': { - const payload = job.data as ReprocessArtistsJob; - const batchSize = payload.batchSize ?? 100; - const offset = payload.offset ?? 0; - console.log(`[ReprocessArtists] Starting artist reprocessing (batch=${batchSize}, offset=${offset})`); - - const artistsRes = await pgPool.query( - `SELECT id, name, canonical_name, mbid FROM artists ORDER BY id LIMIT $1 OFFSET $2`, - [batchSize, offset] - ); - - let processed = 0; - let updated = 0; - let merged = 0; - - for (const artist of artistsRes.rows) { - processed++; - try { - const result = await enrichmentService.resolveArtistIdentity(artist.name); - // Merge if resolved to a different artist (duplicate detected) - if (result.artistId !== artist.id) { - merged++; - // Whole merge is one transaction on one dedicated connection: the - // intermediate states (track links moved but albums not yet, or - // vice versa) must never be visible, and a failure part-way must - // not leave an artist half-merged. ON DELETE CASCADE makes a - // partial merge destructive. - await withTransaction(pgPool, async (client) => { - // Move track links to the keeper, skipping any (track, role) the - // keeper already has, then drop the loser's — a plain UPDATE would - // violate track_artists_pkey when both are on the same track. - await client.query( - `INSERT INTO track_artists (track_id, artist_id, role) - SELECT track_id, $1, role FROM track_artists WHERE artist_id = $2 - ON CONFLICT (track_id, artist_id, role) DO NOTHING`, - [result.artistId, artist.id] - ); - await client.query( - `DELETE FROM track_artists WHERE artist_id = $1`, - [artist.id] - ); - // Fold albums into the keeper. Move tracks of any same-title album - // to the keeper's matching album first (UNIQUE(artist_id,title) and - // ON DELETE CASCADE mean a blind UPDATE could collide or, worse, - // cascade-delete tracks when the loser artist is removed). - const dupAlbums = await client.query( - `SELECT l.id AS loser_id, k.id AS keeper_id - FROM albums l JOIN albums k - ON k.artist_id = $1 AND lower(k.title) = lower(l.title) - WHERE l.artist_id = $2`, - [result.artistId, artist.id] - ); - for (const { loser_id, keeper_id } of dupAlbums.rows) { - await client.query(`UPDATE tracks SET album_id = $1 WHERE album_id = $2`, [keeper_id, loser_id]); - await client.query(`DELETE FROM albums WHERE id = $1`, [loser_id]); - } - // Remaining (non-colliding) albums move over cleanly. - await client.query( - `UPDATE albums SET artist_id = $1 WHERE artist_id = $2`, - [result.artistId, artist.id] - ); - await client.query( - `DELETE FROM artists WHERE id = $1`, - [artist.id] - ); - }); - console.log(`[ReprocessArtists] Merged "${artist.name}" (${artist.id}) -> "${result.canonicalName}" (${result.artistId})`); - } else { - // Update the existing artist record with new canonical_name and/or mbid - const updates: string[] = []; - const params: any[] = []; - let paramIdx = 1; - - if (result.canonicalName && artist.canonical_name !== result.canonicalName) { - updates.push(`canonical_name = $${paramIdx++}`); - params.push(result.canonicalName); - } - if (result.mbid && artist.mbid !== result.mbid) { - updates.push(`mbid = $${paramIdx++}`); - params.push(result.mbid); - } - if (result.sortName && artist.sort_name !== result.sortName) { - updates.push(`sort_name = $${paramIdx++}`); - params.push(result.sortName); - } - if (updates.length > 0) { - updates.push(`updated_at = CURRENT_TIMESTAMP`); - params.push(artist.id); - await pgPool.query( - `UPDATE artists SET ${updates.join(', ')} WHERE id = $${paramIdx}`, - params - ); - updated++; - console.log(`[ReprocessArtists] Updated "${artist.name}" (${artist.id}): ${updates.join(', ')}`); - } - } - - // Refresh the artist image via the dedicated job rather than inline, - // so the reprocess batches aren't blocked on image HTTP. Deduped by - // jobId across the run. - await queue.add( - 'artist_image', - { artistId: result.artistId } satisfies ArtistImageJob, - { - jobId: `artist-image-${result.artistId}`, - removeOnComplete: { age: 86400, count: 5000 }, - removeOnFail: { age: 86400, count: 5000 }, - } - ); - } catch (err) { - console.error(`[ReprocessArtists] Failed for artist ${artist.id} (${artist.name}):`, err); - } - } - - console.log(`[ReprocessArtists] Batch complete: processed=${processed}, updated=${updated}, merged=${merged}`); - - // If we processed a full batch, enqueue the next one - if (artistsRes.rows.length === batchSize) { - await queue.add('reprocess_artists', { batchSize, offset: offset + batchSize }, { - removeOnComplete: { age: 86400, count: 100 }, - removeOnFail: { age: 86400, count: 100 }, - }); - console.log(`[ReprocessArtists] Enqueued next batch at offset ${offset + batchSize}`); - } else { - // Final batch - run deduplication pass to merge artists with same normalized_name - console.log(`[ReprocessArtists] All batches complete, running deduplication pass...`); - - const dupRes = await pgPool.query( - `SELECT normalized_name, array_agg(id ORDER BY - CASE WHEN mbid IS NOT NULL THEN 0 ELSE 1 END, - CASE WHEN canonical_name IS NOT NULL THEN 0 ELSE 1 END, - id - ) as ids - FROM artists - GROUP BY normalized_name - HAVING COUNT(*) > 1` - ); - - let dedupMerged = 0; - for (const row of dupRes.rows) { - const ids = row.ids; - const keepId = ids[0]; // First one (prefers MBID, then canonical_name, then lowest id) - const mergeIds = ids.slice(1); - - for (const mergeId of mergeIds) { - if (mergeId === keepId) continue; - try { - // One transaction per merge, on a dedicated connection: a - // partial merge is destructive (ON DELETE CASCADE). - await withTransaction(pgPool, async (client) => { - // First, handle track_artists conflicts: if both artists are on same track, - // keep the 'main' role, or merge roles. Use ON CONFLICT DO NOTHING to skip duplicates. - await client.query( - `INSERT INTO track_artists (track_id, artist_id, role) - SELECT track_id, $1, role FROM track_artists WHERE artist_id = $2 - ON CONFLICT (track_id, artist_id, role) DO NOTHING`, - [keepId, mergeId] - ); - // Then delete the old track_artists entries - await client.query( - `DELETE FROM track_artists WHERE artist_id = $1`, - [mergeId] - ); - - // Fold same-title albums (move tracks) before reassigning the rest, - // to avoid UNIQUE(artist_id,title) collisions / cascade deletes. - const dupAlbums = await client.query( - `SELECT l.id AS loser_id, k.id AS keeper_id - FROM albums l JOIN albums k - ON k.artist_id = $1 AND lower(k.title) = lower(l.title) - WHERE l.artist_id = $2`, - [keepId, mergeId] - ); - for (const { loser_id, keeper_id } of dupAlbums.rows) { - await client.query(`UPDATE tracks SET album_id = $1 WHERE album_id = $2`, [keeper_id, loser_id]); - await client.query(`DELETE FROM albums WHERE id = $1`, [loser_id]); - } - await client.query( - `UPDATE albums SET artist_id = $1 WHERE artist_id = $2`, - [keepId, mergeId] - ); - await client.query( - `DELETE FROM artists WHERE id = $1`, - [mergeId] - ); - }); - dedupMerged++; - console.log(`[ReprocessArtists] Dedup merged ${mergeId} -> ${keepId} (normalized: ${row.normalized_name})`); - } catch (err) { - console.error(`[ReprocessArtists] Dedup failed for ${mergeId}:`, err); - } - } - } - - console.log(`[ReprocessArtists] Deduplication complete: merged=${dedupMerged}`); - - // Album dedup pass: merge duplicate album rows (same title or same - // MBID) that accumulated before the albumartist scanner fix. Run - // after artist dedup so album artist_id refs are already resolved. - try { - const albumMerged = await enrichmentService.dedupAlbums(); - console.log(`[ReprocessArtists] Album dedup: merged=${albumMerged}`); - } catch (err) { - console.error('[ReprocessArtists] Album dedup failed:', err); - } - - console.log(`[ReprocessArtists] All artists processed!`); - } + await reprocessArtists(job.data as ReprocessArtistsJob, { + pgPool, + queue, + enrichmentService, + }); break; } default: diff --git a/workers/src/reprocess-artists.service.ts b/workers/src/reprocess-artists.service.ts new file mode 100644 index 0000000..af90ea0 --- /dev/null +++ b/workers/src/reprocess-artists.service.ts @@ -0,0 +1,201 @@ +import type { Pool } from 'pg'; +import type { Queue } from 'bullmq'; +import type { ArtistImageJob, ReprocessArtistsJob } from './types.js'; +import type { EnrichmentService } from './enrichment.service.js'; +import { withTransaction, type Queryable } from './db.js'; + +/** + * Fold `loserId` into `keepId`: track links, then albums, then the artist row. + * + * Order matters and the whole thing must be one transaction. `artists` is + * referenced with ON DELETE CASCADE, so a partial merge is destructive: dropping + * the loser before its albums have moved cascade-deletes those albums and their + * tracks. Callers are responsible for the transaction (see `withTransaction`) — + * this takes a `Queryable` so it can run on the dedicated client. + */ +export async function mergeArtistInto( + client: Queryable, + keepId: string, + loserId: string +): Promise { + // Move track links to the keeper, skipping any (track, role) the keeper + // already has, then drop the loser's — a plain UPDATE would violate + // track_artists_pkey when both artists are on the same track. + await client.query( + `INSERT INTO track_artists (track_id, artist_id, role) + SELECT track_id, $1, role FROM track_artists WHERE artist_id = $2 + ON CONFLICT (track_id, artist_id, role) DO NOTHING`, + [keepId, loserId] + ); + await client.query(`DELETE FROM track_artists WHERE artist_id = $1`, [loserId]); + + // Fold albums into the keeper. Move tracks of any same-title album to the + // keeper's matching album first: UNIQUE(artist_id, title) plus ON DELETE + // CASCADE mean a blind UPDATE could collide or, worse, cascade-delete tracks + // when the loser artist is removed. + const dupAlbums = await client.query<{ loser_id: string; keeper_id: string }>( + `SELECT l.id AS loser_id, k.id AS keeper_id + FROM albums l JOIN albums k + ON k.artist_id = $1 AND lower(k.title) = lower(l.title) + WHERE l.artist_id = $2`, + [keepId, loserId] + ); + for (const { loser_id, keeper_id } of dupAlbums.rows) { + await client.query(`UPDATE tracks SET album_id = $1 WHERE album_id = $2`, [keeper_id, loser_id]); + await client.query(`DELETE FROM albums WHERE id = $1`, [loser_id]); + } + // Remaining (non-colliding) albums move over cleanly. + await client.query(`UPDATE albums SET artist_id = $1 WHERE artist_id = $2`, [keepId, loserId]); + + await client.query(`DELETE FROM artists WHERE id = $1`, [loserId]); +} + +interface Deps { + pgPool: Pool; + queue: Queue; + enrichmentService: EnrichmentService; +} + +/** + * Re-resolve artist identities one batch at a time, self-enqueueing the next + * batch until the table is exhausted; the final batch then runs the + * normalized_name dedup pass and the album dedup pass. + */ +export async function reprocessArtists( + payload: ReprocessArtistsJob, + { pgPool, queue, enrichmentService }: Deps +): Promise { + const batchSize = payload.batchSize ?? 100; + const offset = payload.offset ?? 0; + console.log(`[ReprocessArtists] Starting artist reprocessing (batch=${batchSize}, offset=${offset})`); + + const artistsRes = await pgPool.query( + `SELECT id, name, canonical_name, mbid FROM artists ORDER BY id LIMIT $1 OFFSET $2`, + [batchSize, offset] + ); + + let processed = 0; + let updated = 0; + let merged = 0; + + for (const artist of artistsRes.rows) { + processed++; + try { + const result = await enrichmentService.resolveArtistIdentity(artist.name); + + // Merge if resolved to a different artist (duplicate detected) + if (result.artistId !== artist.id) { + merged++; + // Whole merge is one transaction on one dedicated connection: the + // intermediate states (track links moved but albums not yet, or vice + // versa) must never be visible, and a failure part-way must not leave + // an artist half-merged. + await withTransaction(pgPool, (client) => mergeArtistInto(client, result.artistId, artist.id)); + console.log(`[ReprocessArtists] Merged "${artist.name}" (${artist.id}) -> "${result.canonicalName}" (${result.artistId})`); + } else { + // Update the existing artist record with new canonical_name and/or mbid + const updates: string[] = []; + const params: any[] = []; + let paramIdx = 1; + + if (result.canonicalName && artist.canonical_name !== result.canonicalName) { + updates.push(`canonical_name = $${paramIdx++}`); + params.push(result.canonicalName); + } + if (result.mbid && artist.mbid !== result.mbid) { + updates.push(`mbid = $${paramIdx++}`); + params.push(result.mbid); + } + if (result.sortName && artist.sort_name !== result.sortName) { + updates.push(`sort_name = $${paramIdx++}`); + params.push(result.sortName); + } + if (updates.length > 0) { + updates.push(`updated_at = CURRENT_TIMESTAMP`); + params.push(artist.id); + await pgPool.query( + `UPDATE artists SET ${updates.join(', ')} WHERE id = $${paramIdx}`, + params + ); + updated++; + console.log(`[ReprocessArtists] Updated "${artist.name}" (${artist.id}): ${updates.join(', ')}`); + } + } + + // Refresh the artist image via the dedicated job rather than inline, + // so the reprocess batches aren't blocked on image HTTP. Deduped by + // jobId across the run. + await queue.add( + 'artist_image', + { artistId: result.artistId } satisfies ArtistImageJob, + { + jobId: `artist-image-${result.artistId}`, + removeOnComplete: { age: 86400, count: 5000 }, + removeOnFail: { age: 86400, count: 5000 }, + } + ); + } catch (err) { + console.error(`[ReprocessArtists] Failed for artist ${artist.id} (${artist.name}):`, err); + } + } + + console.log(`[ReprocessArtists] Batch complete: processed=${processed}, updated=${updated}, merged=${merged}`); + + // If we processed a full batch, enqueue the next one + if (artistsRes.rows.length === batchSize) { + await queue.add('reprocess_artists', { batchSize, offset: offset + batchSize }, { + removeOnComplete: { age: 86400, count: 100 }, + removeOnFail: { age: 86400, count: 100 }, + }); + console.log(`[ReprocessArtists] Enqueued next batch at offset ${offset + batchSize}`); + return; + } + + // Final batch - run deduplication pass to merge artists with same normalized_name + console.log(`[ReprocessArtists] All batches complete, running deduplication pass...`); + + const dupRes = await pgPool.query( + `SELECT normalized_name, array_agg(id ORDER BY + CASE WHEN mbid IS NOT NULL THEN 0 ELSE 1 END, + CASE WHEN canonical_name IS NOT NULL THEN 0 ELSE 1 END, + id + ) as ids + FROM artists + GROUP BY normalized_name + HAVING COUNT(*) > 1` + ); + + let dedupMerged = 0; + for (const row of dupRes.rows) { + const ids = row.ids; + const keepId = ids[0]; // First one (prefers MBID, then canonical_name, then lowest id) + const mergeIds = ids.slice(1); + + for (const mergeId of mergeIds) { + if (mergeId === keepId) continue; + try { + // One transaction per merge, on a dedicated connection: a partial merge + // is destructive (ON DELETE CASCADE). + await withTransaction(pgPool, (client) => mergeArtistInto(client, keepId, mergeId)); + dedupMerged++; + console.log(`[ReprocessArtists] Dedup merged ${mergeId} -> ${keepId} (normalized: ${row.normalized_name})`); + } catch (err) { + console.error(`[ReprocessArtists] Dedup failed for ${mergeId}:`, err); + } + } + } + + console.log(`[ReprocessArtists] Deduplication complete: merged=${dedupMerged}`); + + // Album dedup pass: merge duplicate album rows (same title or same MBID) that + // accumulated before the albumartist scanner fix. Run after artist dedup so + // album artist_id refs are already resolved. + try { + const albumMerged = await enrichmentService.dedupAlbums(); + console.log(`[ReprocessArtists] Album dedup: merged=${albumMerged}`); + } catch (err) { + console.error('[ReprocessArtists] Album dedup failed:', err); + } + + console.log(`[ReprocessArtists] All artists processed!`); +}