kami/muzick
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: extract reprocess_artists out of the worker's job switch
Typecheck / typecheck (backend) (push) Has been cancelled
Details
Typecheck / typecheck (workers) (push) Has been cancelled
Details
Typecheck / typecheck (backend) (pull_request) Has been cancelled
Details
Typecheck / typecheck (workers) (pull_request) Has been cancelled
Details

Browse Source 
The reprocess_artists case was 208 of index.ts's 515 lines — 40% of the file
and most of what this PR changed in it, buried inside a switch. index.ts is
now 313 lines and reads as what it is: wiring, the job switch, cron
registration, shutdown.

The move also collapses a real duplication. The artist merge and the
normalized_name dedup pass ran the same five statements in the same order
against different id pairs, so the withTransaction change had to be made
twice, identically. Both now call one mergeArtistInto(client, keepId,
loserId), which takes a Queryable so the caller owns the transaction, and the
ON DELETE CASCADE hazard is documented once instead of twice.

Verified as behaviour-preserving: the statement sequence diffs identical
against the previous commit, and workers typecheck is clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
kami
2026-07-31 00:32:03 +04:00
parent dee2b0ad57
commit bec77f4297
2 changed files with 207 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
workers/src/index.ts
+6 -208
View File
@@ -7,7 +7,7 @@ import { IntegrityService } from './integrity.service.js';
import { EnrichmentService } from './enrichment.service.js'; import { EnrichmentService } from './enrichment.service.js';
import { AudioFeaturesService } from './audio-features.service.js'; import { AudioFeaturesService } from './audio-features.service.js';
import { CleanupSweepService } from './cleanup.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 // Cron for the periodic integrity sweep (default: daily at 03:00). Configurable
// via INTEGRITY_SWEEP_CRON. MUSIC_DIR (consumed by IntegrityService) controls // via INTEGRITY_SWEEP_CRON. MUSIC_DIR (consumed by IntegrityService) controls
@@ -220,213 +220,11 @@ async function initWorker() {
break; break;
} }
case 'reprocess_artists': { case 'reprocess_artists': {
const payload = job.data as ReprocessArtistsJob; await reprocessArtists(job.data as ReprocessArtistsJob, {
const batchSize = payload.batchSize ?? 100; pgPool,
const offset = payload.offset ?? 0; queue,
console.log(`[ReprocessArtists] Starting artist reprocessing (batch=${batchSize}, offset=${offset})`); enrichmentService,
});
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!`);
}
break; break;
} }
default: default:
workers/src/reprocess-artists.service.ts
+201
View File
@@ -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<void> {
// 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<void> {
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!`);
}