ee43995e96
The worker ran every job through a single pg Client while BullMQ was configured with concurrency: 10. A Client is one connection with one protocol stream and no queueing: ten concurrent jobs interleave on it, and any BEGIN/COMMIT is shared by all of them, so an unrelated job's failure can roll back another's work and a rollback can discard a third's committed intent. Switched to a Pool, added a small withTransaction(pool, fn) helper that takes a dedicated connection per transaction, and threaded a Queryable interface through the services so they accept either a pool or a pooled client. Both reprocess_artists merge blocks — the artist merge and the duplicate-album merge — now run inside withTransaction; previously a failure partway through left artists merged and their tracks unmoved. integrity.service and cleanup.service get only the constructor type change here so this commit compiles; their own fixes follow in the next two commits. cleanup.service's BEGIN/COMMIT-on-a-Pool is therefore still wrong at this commit and is replaced wholesale by the hard-delete commit. REVIEW-2026-07-30.md finding 4 (and the concurrency note in finding 3). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
232 lines
8.0 KiB
TypeScript
232 lines
8.0 KiB
TypeScript
import type { Queryable } from './db.js';
|
|
import { MusicBrainzClient } from './integrations/musicbrainz.client.js';
|
|
import { normalizeForMatching } from './utils/fuzzy-match.js';
|
|
|
|
function generateSortName(name: string): string {
|
|
const match = name.match(/^(The|A|An)\s+(.+)$/i);
|
|
if (match) return `${match[2]}, ${match[1]}`;
|
|
return name;
|
|
}
|
|
|
|
export class MbSpineWriter {
|
|
constructor(private pgClient: Queryable) {}
|
|
|
|
/**
|
|
* Fetch full artist-credit for a recording MBID and write claims.
|
|
* - First credit → "credited_main_on"
|
|
* - Additional credits → "featured_on"
|
|
* - Updates tracks.recording_mbid
|
|
* - Resolves or stubs credited artists by MBID
|
|
*/
|
|
async writeRecordingClaims(
|
|
recordingMbid: string,
|
|
trackId: string,
|
|
mbClient: MusicBrainzClient
|
|
): Promise<number> {
|
|
const recording = await mbClient.getRecording(recordingMbid);
|
|
if (!recording || recording.artistCredit.length === 0) return 0;
|
|
|
|
await this.pgClient.query(
|
|
`UPDATE tracks SET recording_mbid = $1 WHERE id = $2 AND recording_mbid IS DISTINCT FROM $1`,
|
|
[recordingMbid, trackId]
|
|
);
|
|
|
|
let claimsWritten = 0;
|
|
|
|
for (let i = 0; i < recording.artistCredit.length; i++) {
|
|
const credit = recording.artistCredit[i];
|
|
|
|
const artistId = await this.resolveArtist(
|
|
credit.artistMbid,
|
|
credit.artistName,
|
|
credit.creditName,
|
|
);
|
|
if (!artistId) {
|
|
console.warn(
|
|
`[MbSpineWriter] Could not resolve artist for credit ${i} on recording ${recordingMbid}`
|
|
);
|
|
continue;
|
|
}
|
|
|
|
const predicate = i === 0 ? 'credited_main_on' : 'featured_on';
|
|
const raw = credit.joinphrase ? JSON.stringify({ joinphrase: credit.joinphrase }) : null;
|
|
|
|
await this.pgClient.query(
|
|
`INSERT INTO claims
|
|
(subject_type, subject_id, predicate, object_type, object_id, source, confidence, raw)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
|
ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id)
|
|
DO NOTHING`,
|
|
['track', trackId, predicate, 'artist', artistId, 'mb', 1.0, raw]
|
|
);
|
|
claimsWritten++;
|
|
}
|
|
|
|
return claimsWritten;
|
|
}
|
|
|
|
/**
|
|
* Fetch full artist-credit for a release-group MBID and write claims.
|
|
* - First credit → "credited_main_on_album"
|
|
* - Additional credits → "featured_on_album"
|
|
* - Resolves or stubs credited artists by MBID
|
|
*/
|
|
async writeAlbumClaims(
|
|
releaseGroupMbid: string,
|
|
albumId: string,
|
|
mbClient: MusicBrainzClient
|
|
): Promise<number> {
|
|
const releaseGroup = await mbClient.getReleaseGroup(releaseGroupMbid);
|
|
if (!releaseGroup || releaseGroup.artistCredit.length === 0) return 0;
|
|
|
|
let claimsWritten = 0;
|
|
|
|
for (let i = 0; i < releaseGroup.artistCredit.length; i++) {
|
|
const credit = releaseGroup.artistCredit[i];
|
|
|
|
const artistId = await this.resolveArtist(
|
|
credit.artistMbid,
|
|
credit.artistName,
|
|
credit.creditName,
|
|
);
|
|
if (!artistId) {
|
|
console.warn(
|
|
`[MbSpineWriter] Could not resolve artist for credit ${i} on release-group ${releaseGroupMbid}`
|
|
);
|
|
continue;
|
|
}
|
|
|
|
const predicate = i === 0 ? 'credited_main_on_album' : 'featured_on_album';
|
|
const raw = credit.joinphrase ? JSON.stringify({ joinphrase: credit.joinphrase }) : null;
|
|
|
|
await this.pgClient.query(
|
|
`INSERT INTO claims
|
|
(subject_type, subject_id, predicate, object_type, object_id, source, confidence, raw)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
|
ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id)
|
|
DO NOTHING`,
|
|
['artist', artistId, predicate, 'album', albumId, 'mb', 1.0, raw]
|
|
);
|
|
claimsWritten++;
|
|
}
|
|
|
|
return claimsWritten;
|
|
}
|
|
|
|
/**
|
|
* Fetch an artist's MB artist-relations and write structural claims:
|
|
* - "member of" → "member_of" (subject = this artist, object = target group)
|
|
* - "is alias of" / "is performance name of" / "is legal name of" → "alias_of"
|
|
*
|
|
* Direction handling: MB marks the relationship as forward when it points from
|
|
* the queried artist to the target. For "is performance name of" (forward)
|
|
* the queried artist is the alias of the target, so subject = queried artist,
|
|
* object = target. For backward relations we flip subject/object.
|
|
*
|
|
* Collab/vocal/instrument ARs are artist-artist contributions rather than
|
|
* identity/structure, and the spec maps them to track-level featured_on; we
|
|
* skip them here to avoid over-claiming artist-artist edges.
|
|
*/
|
|
async writeArtistRelationClaims(
|
|
artistMbid: string,
|
|
artistId: string,
|
|
mbClient: MusicBrainzClient
|
|
): Promise<number> {
|
|
const relations = await mbClient.getArtistRelations(artistMbid);
|
|
if (relations.length === 0) return 0;
|
|
|
|
const memberOfTypes = new Set(['member of', 'founder of', 'founder']);
|
|
const aliasOfTypes = new Set(['is alias of', 'is performance name of', 'is legal name of']);
|
|
|
|
let claimsWritten = 0;
|
|
|
|
for (const rel of relations) {
|
|
const isMemberOf = [...memberOfTypes].some(t => rel.type.toLowerCase() === t);
|
|
const aliasMatch = [...aliasOfTypes].find(t => rel.type.toLowerCase() === t);
|
|
if (!isMemberOf && !aliasMatch) continue;
|
|
|
|
const targetArtistId = await this.resolveArtist(
|
|
rel.targetMbid,
|
|
rel.targetName,
|
|
rel.targetName,
|
|
);
|
|
if (!targetArtistId) {
|
|
console.warn(
|
|
`[MbSpineWriter] Could not resolve target artist for relation "${rel.type}" on artist ${artistMbid}`
|
|
);
|
|
continue;
|
|
}
|
|
|
|
// Forward direction: subject = this artist, object = target.
|
|
// Backward direction: the relation is phrased from target → subject, so
|
|
// we flip subject/object to preserve the predicate's intended direction.
|
|
let subjectId = artistId;
|
|
let objectId = targetArtistId;
|
|
if (rel.direction === 'backward') {
|
|
subjectId = targetArtistId;
|
|
objectId = artistId;
|
|
}
|
|
|
|
const predicate = aliasMatch ? 'alias_of' : 'member_of';
|
|
const raw = JSON.stringify({
|
|
mb_relation_type: rel.type,
|
|
direction: rel.direction,
|
|
});
|
|
|
|
await this.pgClient.query(
|
|
`INSERT INTO claims
|
|
(subject_type, subject_id, predicate, object_type, object_id, source, confidence, raw)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
|
ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id)
|
|
DO NOTHING`,
|
|
['artist', subjectId, predicate, 'artist', objectId, 'mb', 1.0, raw]
|
|
);
|
|
claimsWritten++;
|
|
}
|
|
|
|
return claimsWritten;
|
|
}
|
|
|
|
/**
|
|
* Resolve an MB artist to a local artists.id. Priority:
|
|
* 1. Match by mbid (UUID column)
|
|
* 2. Match by normalized credit name
|
|
* 3. Create a stub row
|
|
*/
|
|
private async resolveArtist(
|
|
mbid: string,
|
|
artistName: string,
|
|
creditName: string,
|
|
): Promise<string | null> {
|
|
const existing = await this.pgClient.query<{ id: string }>(
|
|
`SELECT id FROM artists WHERE mbid = $1`,
|
|
[mbid]
|
|
);
|
|
if (existing.rows.length > 0) return existing.rows[0].id;
|
|
|
|
const normalized = normalizeForMatching(creditName);
|
|
const nameMatch = await this.pgClient.query<{ id: string }>(
|
|
`SELECT id FROM artists WHERE normalized_name = $1`,
|
|
[normalized]
|
|
);
|
|
if (nameMatch.rows.length > 0) {
|
|
await this.pgClient.query(
|
|
`UPDATE artists SET mbid = $1, updated_at = CURRENT_TIMESTAMP WHERE id = $2`,
|
|
[mbid, nameMatch.rows[0].id]
|
|
);
|
|
return nameMatch.rows[0].id;
|
|
}
|
|
|
|
const sortName = generateSortName(artistName);
|
|
const result = await this.pgClient.query<{ id: string }>(
|
|
`INSERT INTO artists (name, canonical_name, sort_name, mbid)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (mbid) WHERE mbid IS NOT NULL
|
|
DO UPDATE SET name = EXCLUDED.name, updated_at = CURRENT_TIMESTAMP
|
|
RETURNING id`,
|
|
[creditName, artistName, sortName, mbid]
|
|
);
|
|
return result.rows[0]?.id ?? null;
|
|
}
|
|
}
|