[D3] Update enrichment consumer to upsert album_metadata (linked rows)

[jakebromberg]

Problem

After D1+D2 land, both enrichment writers must be updated to UPSERT into album_metadata keyed by album_id instead of writing flowsheet's 10 metadata columns. Free-form rows (album_id IS NULL) keep writing the inline columns until linkage resolves.

Sequencing decision (2026-05-22)

D3 patches both writer sites in one PR:

• apps/backend/services/metadata/enrichment.service.ts — the in-process fireAndForgetMetadataForRow .then arm (line 95-135). This is what runs in prod today.
• apps/enrichment-worker/enrich.ts — the worker's finalizeRow (line 121-164). Merged via BS#892 PR-2 (feat(enrichment-worker): wire claim → LML → finalize + SSE broadcast (BS#892 PR-2) #1010, merged 2026-05-23) and PR-3 (acd77e7) but not yet deployed to prod.

Rationale: PR-3 shipped the Dockerfile + healthcheck, so the worker is deployable. Whether it goes live before or after D3 is a coordination question we don't want to depend on. Patching both writers in the same PR makes D3 self-contained: whoever deploys the worker first inherits correct album_metadata semantics; until then the in-process path is correct on its own.

BS#894 (C5 — drop in-process fire-and-forget) is the eventual cleanup: it deletes the in-process callsites once the worker is verified in prod. D3 makes the in-process arm correct in the meantime; C5 doesn't depend on D3 and D3 doesn't depend on C5.

Scope

1. In-process writer (enrichment.service.ts)

The .then arm at line 95 today writes 10 metadata cols + stamps metadata_attempt_at with the idempotency guard WHERE metadata_attempt_at IS NULL. D3:

.then(async (metadata) => {
  if (!metadata) return;

  if (input.albumId !== undefined) {
    await db.insert(album_metadata).values({
      album_id: input.albumId,
      artwork_url, discogs_url, /* …other 8 fields… */,
      updated_at: sql`NOW()`,
    }).onConflictDoUpdate({
      target: album_metadata.album_id,
      set: { artwork_url, discogs_url, /* …other 8 fields… */, updated_at: sql`NOW()` },
      // Idempotency / race guard: only overwrite when the incoming write
      // is newer. Prevents the drift-repair backfill from clobbering a
      // fresh runtime enrichment, and lets the worker + in-process path
      // converge under the dual-writer window before C5 ships.
      setWhere: sql`${album_metadata.updated_at} < NOW()`,
    });
    await db.update(flowsheet)
      .set({ metadata_attempt_at: sql`NOW()` })
      .where(sql`"id" = ${input.flowsheetId} AND "metadata_attempt_at" IS NULL`);
  } else {
    // Free-form / unlinked rows: write the 10 inline columns as today.
    await db.update(flowsheet).set({
      artwork_url, discogs_url, /* …other 8 fields… */,
      metadata_attempt_at: sql`NOW()`,
    }).where(sql`"id" = ${input.flowsheetId} AND "metadata_attempt_at" IS NULL`);
  }
})
.catch(/* unchanged — leaves metadata_attempt_at NULL for drift-repair retry */);

EnrichmentInput.albumId is already passed by all callers (flowsheet.controller.ts:99-107, internal.route.ts:165). No new plumbing.

2. Worker writer (apps/enrichment-worker/enrich.ts)

finalizeRow today writes 10 cols on match / 3 search URLs on no-match, flipping metadata_status 'enriching' → terminal, guarded by WHERE metadata_status = 'enriching'. D3:

• On match + linked: UPSERT album_metadata (full 10-field payload), then UPDATE flowsheet metadata_status='enriched_match' only (no metadata cols).
• On match + unlinked: UPDATE flowsheet 10 cols + status (today's behavior).
• On no-match + linked: UPSERT album_metadata (search URLs only, 7 other fields untouched per existing semantics), then UPDATE flowsheet status only.
• On no-match + unlinked: UPDATE flowsheet 3 search URL cols + status (today's behavior).

The race detector (.returning({ id }) empty → _raced outcome) stays on the flowsheet UPDATE — the worker's signal for "did this consumer finalize the row" is unchanged. The album_metadata UPSERT has its own race guard (setWhere: album_metadata.updated_at < NOW()) and is allowed to noop.

3. CDC subscriber widening (apps/enrichment-worker/cdc-subscriber.ts)

EnrichmentCandidate (line 24-31) does not carry album_id today. D3 widens it (and filterForEnrichment) to read data.album_id from the CDC payload — CDC already includes it. Same widening needed for EnrichRow in enrich.ts:43-48.

metadata_status lifecycle arms (post-BS#891)

The enum has 5 states (pending | enriching | enriched_match | enriched_no_match | failed_no_retry). The worker covers:

Outcome	metadata_status	album_metadata write
LML match	enriched_match	UPSERT (linked); inline UPDATE (unlinked)
LML success, no Discogs match	enriched_no_match	UPSERT search-URL fallbacks only (linked); inline UPDATE (unlinked)
LML threw (transient)	unchanged (enriching)	no write; C6 sweep reverts past enriching_since + 60s
Retry budget exceeded	failed_no_retry	no write; manual triage
Mid-LML-call	enriching (claim)	no write yet

The in-process path doesn't claim rows or count retries — it has only two arms (.then writes, .catch leaves metadata_attempt_at NULL for drift-repair retry). It does not set metadata_status today and D3 doesn't change that; the dual-signal split (metadata_attempt_at for in-process, metadata_status for worker) is pre-existing and goes away when C5 deletes the in-process callsites.

Coexistence with the drift-repair backfill

The flowsheet-metadata-backfill cron is currently stopped (BS#995; 2026-05-21 LML monopolization incident). When it resumes, both runtime writers and the cron will UPSERT album_metadata for the same album_id. The setWhere: ${album_metadata.updated_at} < NOW() guard ensures last-write-wins by timestamp across all three writers.

Acceptance

• Linked rows from both writers: UPSERT album_metadata, no flowsheet metadata-column writes.
• Unlinked rows from both writers: 10 inline columns written, signal column stamped (metadata_attempt_at in-process, metadata_status in worker).
• album_metadata.updated_at set in both INSERT and UPDATE branches of the UPSERT.
• EnrichmentCandidate + EnrichRow widened to carry album_id; filterForEnrichment extracts it from CDC data.
• Worker's _raced outcome unchanged (race guard stays on flowsheet UPDATE).
• Unit tests: tests/unit/apps/enrichment-worker/cdc-subscriber.test.ts + enrich.test.ts updated for the new shape; in-process tests in tests/unit/services/metadata/enrichment.service.test.ts updated for linked vs. unlinked branches.
• Integration test (tests/integration/): a dj-site addEntry for an already-album_metadata-populated linked album produces zero UPDATE flowsheet SET artwork_url = ... statements via either writer. Assert via DB snapshot or write-spy.
• After D4 (column drop): unlinked rows can't enrich at all — intentional; they need linkage first.

Related

• Parent: [Epic D] album_metadata extraction #878 (Epic D)
• Depends on: [D1] Schema: new album_metadata table + read-path join (absorbs #532) #897 ✅ (schema), [D2] Migrate existing flowsheet metadata into album_metadata #898 (historical data), BS#891 (metadata_status schema live; companion CASE backfill still un-run per docs/flowsheet-metadata-status-backfill.md)
• Coordinates with: BS#894 (C5 — drops the in-process callsites once worker is verified in prod) and BS#895 (C6 — stranded-claim sweep retune)
• Coexists with: BS#995 — backfill cron pacing/circuit-breaker; race guard above handles concurrent writes
• BS#892 PR-2 (#1010) + PR-3 (acd77e7) shipped the worker code D3 now updates