Merge pull request 'Repair the 2026-07-30 review findings' (#1) from repair/review-2026-07-30 into master
Typecheck / typecheck (backend) (push) Has been cancelled
Typecheck / typecheck (workers) (push) Has been cancelled

Reviewed-on: #1
This commit was merged in pull request #1.
This commit is contained in:
2026-07-30 22:48:14 +02:00
29 changed files with 3404 additions and 976 deletions
-37
View File
@@ -1,37 +0,0 @@
# Muzick — audit journal
## Goal
Audit the muzick project at `/home/kami/apps/muzick/`, file Vikunja tasks for findings, and work on autonomous items.
## Progress
### 2026-07-14 — Initial audit
**Git**: initialized, initial state committed (737bf19). Fixes committed in 42474c6.
**Project structure**: 196 files — TypeScript/Fastify backend, React/Vite frontend, BullMQ workers, PostgreSQL + Redis + Typesense.
### Fixes applied
1. **AGENTS.md tech stack**: `python/fastapi``fastify/typescript` (was wrong)
2. **CORS**: Added `@fastify/cors` plugin to backend with env-based origin config
3. **`.env.example`**: Created with placeholder values (secrets were only in `.env` which is gitignored)
4. **`backend/src/index.ts`**: Removed dead code (empty file, `server.ts` is real entry point)
### Tasks filed
- See Homelab infra project. Key items:
- #109: pin Docker images (NEEDS FIX — minio:latest etc)
- #110: N+1 queries in generators (PERFORMANCE)
- #111: image proxy SSRF guard (SECURITY)
### New critical issues found
1. **No auth on any API**`x-user-id` header with hardcoded fallback UUID is the only identity
2. **Admin routes unprotected** — anyone can trigger scan/reindex/delete
3. **Postgres password "password"** hardcoded in docker-compose.yml
4. **Typesense API key "muzick-key"** hardcoded in docker-compose.yml
5. **SOCKS proxy IP** `192.168.1.104` exposed in .env and AGENTS.md
6. **No tests for any worker service** (1311-line enrichment.service.ts has 0 tests)
7. **Frontend never typechecked in CI**
8. **No input validation** on many routes (admin, library — `as any` casts)
### Remaining autonomous work items
- #109 — Pin Docker images in docker-compose.yml — CAN DO
+62
View File
@@ -0,0 +1,62 @@
# CLAUDE.md
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## What this is
muzick is a self-hosted music player + recommendation engine (deployed at `muzick.kvmx.ru:5174`). Three deployable units — `backend/` (Fastify API), `frontend/` (Vite React SPA), `workers/` (BullMQ job processor) — plus Postgres, Redis, and Typesense. All wired together by `docker-compose.yml`.
## Commands
Each unit is its own npm package; `cd` into it first.
```bash
# backend/ and workers/
npm run dev # tsx watch
npm run typecheck # tsc --noEmit (also runs as prebuild)
npm run build # tsc
# backend/ only
npm test # vitest run
npm run test:watch
npx vitest run src/services/generators.test.ts # single file
npx vitest run -t "comfortGenerator" # single test by name
# frontend/
npm run dev # vite
npm run build # vite build
npm run typecheck
# whole stack
docker-compose up -d --build
```
There is no lint step. `typecheck` is the gate; the `prebuild` hook fails the build on type errors.
## Architecture
**Split by process, sharing one Postgres database.** The backend serves the API and the frontend consumes it; the worker runs offline enrichment/analysis. They communicate only through Postgres (source of truth) and Redis (BullMQ queue). There is no shared code package — `workers/` and `backend/` each carry their own copy of things like `queue.ts` and pg clients.
- **backend/src/app.ts** — the real entry point (`server.ts` just calls `buildApp`). Registers all routes, connects pg/redis/typesense, and runs several `setInterval` background jobs directly in-process: claim-fusion materialized-view refresh (10s), belief decay (hourly), forgotten-profile derivation (nightly). Auth is a single `onRequest` hook keyed on `MUZICK_API_KEY` / `MUZICK_ADMIN_KEY` (both optional — no keys means open); `/api/admin/*` requires the admin key specifically, `/api/health` is always exempt.
- **backend/src/services/** — business logic. `db.service.ts` owns schema application (`ensureSchema`/`runMigrations` run on every boot — the docker init-mount only fires on a fresh volume, so migrations live here). `session-director.service.ts` and `generators.service.ts` implement the recommendation/vibe logic.
- **workers/src/index.ts** — one BullMQ `Worker` with a `switch` on job name (scan_library, metadata refresh, audio analysis, artist similarity, image enrichment…). Also registers cron repeatables (integrity sweep, dislike cleanup, stale-session reaper). External metadata clients live in `workers/src/integrations/`.
- **frontend/src/** — TanStack Router + TanStack Query. `services/api.ts` is the axios base; per-domain service files wrap endpoints. Zustand stores in `store/` hold playback/vibe/toast state. `components/ethos/` is the shared Ethos design-system UI.
### Domain concepts (from README + spec comments)
- **Rolling Vibe** — continuous stream interleaving owned library tracks with "probation" external discoveries. Managed by the session-director.
- **Belief / claim fusion** — enrichment produces `claims` from multiple sources fused into a materialized view; beliefs decay over time. The in-process timers in `app.ts` keep this fresh.
- **Dislike lifecycle** — multi-stage state machine; the worker's cleanup sweep advances it.
## Deployment gotchas (see AGENTS.md)
- **Typesense is pinned to 0.25.1** — do not bump casually, the API breaks across majors.
- **Worker uses `network_mode: host` + a SOCKS5 proxy** (`SOCKS_PROXY_URL`) for all external metadata calls (Last.fm, Discogs, MusicBrainz, etc.).
- **DB schema** is `backend/src/db/schema.sql`, dropped into `docker-entrypoint-initdb.d` — only applied on a fresh volume. Schema changes for existing volumes must go through `db.service.ts` migrations.
- Music dir is a **read-only** bind from `/mnt/hdd1/media/Music``/music`.
- Redis host inside compose is `infra-redis` (external `infra-net` network); the host-mode worker reaches it at `127.0.0.1:6379`.
- An older `muzick.service` systemd unit runs the pre-docker backend on port 5213 and may conflict — docker compose is the active deployment.
## Working docs
`PLANS.md`, `AUDIT.md`, `progress.md`, and dated `SESSION-*.md` files track in-flight work and are not part of the running system.
+454
View File
@@ -0,0 +1,454 @@
# muzick — engineering review, 2026-07-30
Scope: full-project review (backend, workers, frontend, schema, deployment, docs) against
live runtime state. Every finding marked **confirmed** was verified directly against the
code, the live database, or a throwaway container — not inferred.
Live context at time of review: containers up 8 days, 3,962 tracks, 13,910 claims,
843 beliefs, 68 plays in the preceding 7 days (last: 2026-07-29). Actively used.
---
## verdict
A genuinely good system that is quietly broken in ways its own dashboards cannot show.
The architecture is sound and proportionate to the problem. The failures are not design
failures — roughly six of the most interesting subsystems are dead or silently degrading
in production, and nothing surfaces that.
**Repair. Do not rewrite.**
## what the project is now
A single-user, actively-used self-hosted music player with an ambitious belief/claim
recommendation engine. 17.2k LOC across three deployable units. Reachable only from
LAN/VPN — `/etc/nginx/sites-available/muzick.kvmx.ru` has
`allow 10.42.0.0/24; allow 192.168.1.0/24; deny all` and listens on LAN/VPN interfaces
only, not `0.0.0.0`.
## what it should become
The same system, with its ambitious half either **working or removed**, and with a
fresh-volume rebuild that actually works. At present it is a recommendation engine whose
learning loop is partly disconnected, with no way to notice.
## classification
`fragile` + `misaligned`, plus `overbuilt` in one specific place (System D diversity
budgets, discovery walk).
Not stale, not abandoned, not better replaced.
---
## first assessment
| | |
|---|---|
| purpose | self-hosted music player + recommendation engine over a local library |
| intended users | single user (owner) |
| actual users | single user, actively — 68 plays in 7 days |
| critical workflows | browse/play library; Rolling Vibe session; enrichment; quarantine |
| current state | live and serving, with several subsystems silently inert |
| known failures | see findings 110 |
| maintenance burden | moderate; concentrated in 3 oversized files and absent verification |
| technical constraints | `/music` is a **read-only** bind; worker needs SOCKS5 for egress; Typesense pinned 0.25.1 |
| personal constraints | homelab, single operator, LAN/VPN-only exposure |
| what still works well | playback, library browsing, scanning, enrichment writes, migration registry, cron registration |
| what has become obsolete | discovery graph walk, System D budgets, `workers` shadow schema, dead frontend components |
---
## main findings
### 1. The project cannot be rebuilt from scratch — confirmed
**Problem.** A fresh Postgres volume produces a permanently empty library.
**Evidence.** Booted a throwaway `postgres:16-alpine` with the real
`backend/src/db/schema.sql` and ran the scanner's exact insert
(`workers/src/scanner.service.ts:241``INSERT INTO artists (name) VALUES ($1)`):
```
ERROR: null value in column "canonical_name" of relation "artists"
violates not-null constraint
```
`schema.sql:50` declares `canonical_name TEXT NOT NULL` with no default. The **live**
database has that column nullable — the volume predates the constraint. That is the only
reason anything currently works.
**Impact.** Highest severity. Every artist insert fails on a fresh volume, and
`processFile` swallows the error per-file (`scanner.service.ts:199`), so a scan reports
**success with 0 tracks detected**. There is no disaster recovery, and `docker-compose up -d`
— the documented setup path in the README — silently yields an empty library.
**Action.** Repair: insert `canonical_name` in the scanner, or give the column a default.
Then prove it with a scratch-volume rebuild.
### 2. Every admin action in the UI returns 403 — confirmed
**Problem.** The entire admin surface of the SPA has been non-functional since auth landed.
**Evidence.** Confirmed at three layers:
- `frontend/nginx.conf.template:16` injects only `Authorization: Bearer ${MUZICK_API_KEY}`
- `docker-compose.yml` passes only `MUZICK_API_KEY` to the frontend container
(verified inside it: `API=set ADMIN=`)
- `backend/src/app.ts:108` requires `token === adminKey` for `/api/admin/*`
- the two keys differ (35 vs 41 chars); `frontend/src/services/api.ts` is 5 lines and
sets no headers at all
**Impact.** All 10 admin call sites are dead: the 477-line Jobs page (polling 403s every
3s/5s forever, rendering a blank Overview with no error state) and every Settings library
action — Scan, Reindex, Reprocess artists, Re-enrich, Duplicates merge. Introduced by
commits `5ed8d9e` / `3bc9f2d` without updating the frontend path.
**Action.** Repair: inject the admin key for `location /api/admin/` in the nginx template.
Honest context — the outer proxy already forges credentials for anything reaching
`location /api`, so behind a LAN-only proxy this key split buys nothing while costing the
whole admin surface.
### 3. Belief decay has never once run — confirmed
**Problem.** The temporal dimension of the recommendation engine is entirely inert.
**Evidence.** `backend/src/services/db.service.ts:1788` builds a CTE with
`SELECT profile, CASE profile ...` and **no `FROM` clause**. Postgres rejects it:
`column "profile" does not exist` (SQLSTATE 42703), thrown hourly for 8+ days.
Rewritten as `WITH halflives(profile, halflife_sec) AS (VALUES ...)` and run against the
live DB inside a transaction: **`UPDATE 843`**, then `ROLLBACK`. No data was changed.
**Impact.** `obsession` (14-day half-life) and `contextual` (7-day) never fade, so old
fixations stay maximally weighted forever. Three of six spec'd profiles — `discovery`,
`contextual`, `forgotten` — have **zero rows**.
**Action.** Repair (fix verified). Warning: the first successful run applies ~23 days of
accrued decay at once, cutting `obsession` beliefs to ~0.32×. That is correct behaviour,
but it will visibly change recommendations.
### 4. Claim de-duplication is broken and is actively corrupting scores — confirmed
**Problem.** Re-enrichment inserts duplicate claims instead of reinforcing them, inflating
fusion weights.
**Evidence.** `schema.sql:343` declares `UNIQUE (..., source, user_id)`. Postgres treats
NULLs as distinct, and every objective claim has `user_id IS NULL` — so
`ON CONFLICT DO UPDATE` / `DO NOTHING` (`db.service.ts:1425`,
`workers/src/mb-spine-writer.ts:54`) **never fires**. Live data:
```
dup_groups: 236 | excess_rows: 2110 | worst single claim: 86 copies
```
`claim_fusion` is `SUM(trust * confidence * recency)`, so one edge can carry **86× its
intended weight**. ~15% of the 13,910 claims are re-enrichment duplicates.
**Impact.** The recommendation graph is measurably skewed toward whatever was re-enriched
most — the most likely cause of repetitive recommendations. Grows with every re-enrich.
Almost certainly the root cause behind commit `d497588` ("claim_fusion MV duplicate-key
failure").
**Action.** Repair: `NULLS NOT DISTINCT` (PG15+; running 16) or store the zero-UUID, plus a
one-time dedup migration. **Dedup must run before the constraint is added.**
### 5. SQL injection via column names — confirmed
**Problem.** Request-body keys are interpolated into SQL as column identifiers.
**Evidence.** `db.service.ts:1237`:
```ts
const setClause = fields.map((f, i) => `"${f}" = $${i + 2}`).join(', ');
```
`fields = Object.keys(data)`, and `data` is `request.body as any`
(`library.routes.ts:176`). No allowlist. Same shape in `updateArtist` / `updateAlbum`.
**Impact.** A crafted body key closes the quoted identifier and injects into the SET list;
also plain mass-assignment of `path`, `state`, `play_count`. Mitigated in practice only by
the LAN/VPN-only proxy — which *supplies the auth token automatically*, so any device on
the LAN can reach it from a browser.
**Action.** Repair: allowlist columns.
### 6. The dislike lifecycle is architecturally impossible as specified — confirmed
**Problem.** The spec's terminal state cannot be reached in this deployment.
**Evidence.** `docs/architecture/02-invariants-and-risks.md` §C makes hard file deletion the
final step. `/music` is mounted `:ro` in both services. `workers/src/cleanup.service.ts:72`
calls `unlink()``EROFS``continue`. Live state: **37 tracks stuck `WARNED` since
2026-07-03** (27 days), **296 EROFS errors in the last 24h**, and `feedback` contains zero
`deleted_permanent` rows.
It fails *closed* — no DB row is deleted — which is the good outcome. But it will never
progress, and it is by far the loudest log source.
Related, independent of the mount: `db.service.ts:909` inserts the `deleted_permanent`
audit row and *then* deletes the track, but `feedback.track_id` is `ON DELETE CASCADE`
(verified: `confdeltype = c`) — so that audit row destroys itself.
**Action.** Requires a decision — see *open decisions* below. Fix the cascade regardless.
### 7. Prev and repeat-all are permanently broken — confirmed
**Problem.** `next()` destroys the queue history it needs.
**Evidence.** `frontend/src/store/usePlaybackStore.ts:102``next()` does
`queue.slice(idx + 1)`, so the new track is always `queue[0]`. `prev()` requires
`idx > 0` (`:123`) → always resolves null. `repeat: 'all'` jumps to `queue[0]`, which is
the current last track.
**Impact.** After any auto-advance, Previous does nothing, ever. At the end of an album,
repeat-all loops the final track instead of restarting.
**Action.** Repair: track a `currentIndex` instead of trimming. One change fixes both.
Separately, the **uncommitted** `usePlaybackStore.ts` change in the working tree is a
correct fix for a real end-of-queue auto-resume loop — commit it.
### 8. MusicBrainz rate limiting is not enforced — confirmed
**Problem.** The 1 req/s throttle is a TOCTOU race and does not hold.
**Evidence.** `workers/src/integrations/http.ts:153-161` reads `lastRequestAt`, then
`await delay(...)`, then writes — no mutex, no queue, with `concurrency: 10`. Ten jobs read
the same timestamp, sleep the same duration, and fire in the same tick.
**Impact.** Up to ~10 req/s against MusicBrainz's 1 req/s policy, risking an IP block. And
because `musicbrainz.client.ts:280` catches `HttpError` and returns `null`, a rate-limited
MusicBrainz is indistinguishable from "no data for your library" while every job reports
success.
**Action.** Repair: serialize `throttle` per host with a promise chain.
### 9. Expensive work computed and discarded — confirmed
**Problem.** Two spec'd subsystems cost real time and produce nothing.
**Evidence.**
- `session-director.service.ts:242-270` `getBudgets` runs 7+ aggregate queries per replan.
`rankCandidates` (`:677`) never reads its `budgets` or `state` parameters, and
`detectAntiLoop` ignores its first three. System D budget enforcement is unimplemented.
- `discovery.service.ts:86` writes `source: 'graph_exploration'`, which is absent from
`source_trust` (verified — only 7 keys: `curated, mb, cover_art_archive, discogs,
lastfm, listener_behavior, tag`). `POST /api/discovery/walk` therefore **always** 500s on
an FK violation, after having already committed an orphan candidate row.
**Action.** Delete, or finish. The discovery walk has never worked.
### 10. Verification is largely theatre — confirmed
**Problem.** Nothing would have caught findings 1, 3, or 4.
**Evidence.**
- Frontend `npm run typecheck` **cannot run**`frontend/node_modules/typescript` is a
partial install (no `package.json`, dangling `.bin/tsc` symlink).
- `.github/workflows/typecheck.yml` matrixes only `[backend, workers]` and has **no test
job** — the 33 backend tests never run in CI.
- Those 33 tests pass in 440ms with zero database. All three files mock
`{ query: vi.fn() }` and largely assert that a substring appears in a SQL template.
Several cannot fail: `expect(results).toBeDefined()`, and assertions wrapped in
`if (results.length > 0)`.
- `frontend/Dockerfile` uses `npm install`, not `npm ci`, so the lockfile is ignored.
Backend `node_modules` has `@fastify/cors@11.3.0` installed against a
`^9.0.1` declaration — local and built images do not agree.
Every confirmed bug above lives in untested code.
---
## secondary findings
- **Integrity sweep has no sanity guard.** `workers/src/integrity.service.ts:188` marks
every unreachable track `MISSING` with no percentage threshold and no check that
`MUSIC_DIR` is mounted. If `/mnt/hdd1` is unmounted when the 03:00 sweep runs, the
**entire library** is flipped to `MISSING` with no automatic path back. It also only ever
checks an arbitrary 5,000 tracks (`:74``LIMIT` with no `ORDER BY`, no pagination).
- **Worker shares one pg connection across `concurrency: 10`.** `workers/src/index.ts:24`
creates a single `Client`; `cleanup.service.ts:82` issues bare `BEGIN`/`COMMIT`/`ROLLBACK`
on it. Another job's query can be enrolled in — and discarded by — that transaction. This
is the exact hazard `backend/src/app.ts:35-40` documents as its reason for using a `Pool`.
- **No BullMQ retries anywhere.** Zero `attempts`/`backoff` in either package (default is
1 attempt). Combined with `jobId: meta-<trackId>` and `removeOnFail: {age: 86400}`,
re-enqueueing a failed track within 24h is a **silent no-op**.
- **Image proxy SSRF.** `backend/src/routes/images.routes.ts:79` validates redirect hop 1,
then re-fetches *without* `redirect: 'manual'` — following up to 20 further hops
unvalidated. Directly contradicts the comment above it. Any open redirect on an
allowlisted host (`commons.wikimedia.org`, `*.musicbrainz.org`) becomes a full SSRF
primitive.
- **`split-collab-artists.ts` cascade-deletes tracks.** `:125-136` leaves colliding albums
on the doomed artist, then deletes it; `albums.artist_id` and `tracks.album_id` are both
`ON DELETE CASCADE`. Do not run as written. `dedup-artists-albums.ts` is the correct
implementation.
- **Play recording only happens inside a Vibe session.** `historyService.recordPlay` /
`recordSkip` / `feedback` have no callers in the frontend (verified by grep — only
`historyService.list()` is used). The sole write path is `v2.routes.ts:137` on a Vibe
`completed`. Live: `sum(play_count) = 1445` = exactly the `play_history` row count, and
only 535/3,962 tracks (13.5%) have any plays. Browsing/album playback contributes no
signal, yet Home's "Most Played" and Vibe's seed list both sort by `play_count`.
- **No `error` listener on the audio element** (`AudioEngine.tsx:89-93`), and
`play().catch(() => {})` swallows failures. A 404 stream leaves `isPlaying: true`, the
scrubber at 0:00, and no toast — playback hangs silently.
- **Nine bare `usePlaybackStore()` calls** with no selector re-render on every state change;
`setPosition` fires ~4×/s, so 50 `TrackRow`s re-render four times a second during playback.
- **`workers`' own `ensureSchema()`** (`enrichment.service.ts:78-156`) is 79 lines of shadow
schema that recreates a non-unique `idx_artists_mbid` on every boot — an index the backend
migration `20260709_artists_mbid_unique` deliberately dropped.
- **`runMigrations` has no advisory lock** (`db.service.ts:573`). Concurrent boots can both
run a pending migration; several bodies are not concurrency-safe.
- **No keyboard control of playback at all** — no Space, no arrow seek, no next/prev.
`TrackRow.tsx:59` is a `<div onClick>` with no `tabIndex`/`role`/`onKeyDown`, so no list
in the app is playable by keyboard.
---
## corrected claims
Recorded so these are not acted on at the wrong priority.
- **`dedup-albums` grouping by title alone** (`admin.routes.ts:41`, duplicated in
`enrichment.service.ts:1264` — omits `artist_id` from the `GROUP BY`) is real in code, but
**blast radius today is zero**: no two albums in the library share a lowercase title. A
landmine that fires on the next `reprocess_artists` if one appears — fix it, but it is not
active data loss.
- **`dislikes.grace_hours` being NULL** is a non-issue — the column has `DEFAULT 48`.
- **Mixed `timestamp` / `timestamptz`** (19 naive vs 12 aware; `tracks` has both) is latent,
not active — DB and backend are both `Etc/UTC`. Refactor-later.
- **"Tracks missing from disk"** — an initial check stat'd container paths from the host and
was wrong. Re-run inside the container: **60/60 present**. Invariant A holds.
- **`normalize_artist` truncation** is real (`AC/DC``AC`, `Felix Mendelssohn``Feli`)
but affects only 2 live rows and produces 0 collisions. Low priority.
---
## keep
The three-process split; Postgres as source of truth; the migration registry
(`schema_migrations` is recorded, ordered, individually atomic, append-only by convention —
genuinely well done); `upsertJobScheduler` cron registration (correctly idempotent across
restarts); graceful worker shutdown; the backend's `Pool`-not-`Client` reasoning; the Ethos
design system; `Artwork`'s gradient fallback; `useDislikeTrack` (the one flow with full
success/error/undo feedback); the `docs/architecture/` spec set — it is the reason these
bugs are identifiable *as* bugs.
## remove
- `discovery.service.walkGraphForDiscovery` — never worked
- System D budget computation — or wire it in
- `frontend/src/components/Inspector.tsx` — 169 lines, unreachable (`setInspector` is only
ever called by `closeInspector`)
- `frontend/src/components/PanelHeader.tsx` — 36 lines, zero importers, documenting a
refactor that was never applied
- `frontend/nginx.conf` — dead, unreferenced by the Dockerfile, and the *insecure* variant
- ~20 dead frontend service exports
- `workers`' `ensureSchema()` — shadow schema
- the stale `muzick.service` note in `AGENTS.md` and `CLAUDE.md` — verified: no such systemd
unit exists
## repair now, in order
1. `canonical_name` — restores rebuildability (finding 1)
2. nginx admin-key injection — restores the entire admin UI (finding 2)
3. `decayBeliefs` CTE — fix verified (finding 3)
4. claim dedup + `NULLS NOT DISTINCT` (finding 4)
5. column allowlist in the three `update*` methods (finding 5)
6. `currentIndex` in the playback store; commit the pending fix (finding 7)
7. serialize `throttle` per host (finding 8)
8. integrity-sweep abort threshold (secondary)
9. worker `Client``Pool` (secondary)
## refactor later
Decompose `enrichment.service.ts` (1,311 lines, six unrelated concerns); move the 328-line
inline `MIGRATIONS` array out of `db.service.ts`; add selectors to the nine bare
`usePlaybackStore()` call sites; unify `Genres.tsx` / `Discover.tsx` (~250 near-duplicate
lines); normalize timestamp types; advisory lock around `runMigrations`.
## rewrite only if
Nothing here justifies a rewrite. Every finding is a local repair in structurally sound
code.
The one thing worth reconsidering from first principles is **System A/B (claims → fusion →
beliefs)** — but only *after* findings 3 and 4 are fixed, because that subsystem has never
once been observed running correctly. Judge the design on working behaviour, not on the
current state.
## verification plan
- `docker run` a scratch Postgres with `schema.sql`, run a real scan, assert
`count(tracks) > 0` — turns finding 1 into a permanent regression test
- add `frontend` to the CI matrix; add a `npm test` job; reinstall frontend `node_modules`;
switch the frontend Dockerfile to `npm ci`
- backend integration tests against a real container for `withTransaction`,
`runMigrations`, `recordPlay`, and the auth hook
- post-fix assertions:
- `claims` duplicate groups → 0
- `max(last_decayed_at)` advances hourly
- zero EROFS errors in 24h
- Jobs page renders stats
## open decisions
**The dislike lifecycle.** The spec says delete the file; the mount is read-only. Pick one:
- **(a) Drop hard deletion.** Terminal state becomes `HIDDEN`; amend invariant §C; delete
`finalizeDeleted`. Safe, honest, and the library stays immutable. **Recommended** — the
library is a read-only bind for a reason.
- **(b) Make deletion real.** Requires an `rw` mount. `cleanup.service.ts` currently unlinks
*before* its DB transaction and has no dry-run and no per-sweep cap; that ordering must be
fixed first.
Either way, clear the 37 stuck `WARNED` rows — they generate ~300 errors/day.
## uncertainties
- No HTTP endpoint could be exercised: the review sandbox blocked egress (identical 503s
that never reached the backend, confirmed absent from its logs). The auth and SSRF
findings are established from configuration and code, not from a live request. The SSRF
one is unambiguous in code.
- `docker-compose.yml` sets **no resource limits**, though
`docs/architecture/02-invariants-and-risks.md` §B names cgroup limits as the mitigation
for Essentia starving the API. Unquantified in practice.
- Whether the `discovery` / `contextual` profiles are empty because decay never ran or
because nothing ever wrote them is not yet separable; re-evaluate after finding 3 is fixed.
---
## second-opinion review, 2026-07-30 (Claude)
Independently spot-checked the load-bearing findings against the code. **Verdict endorsed:
repair, do not rewrite.**
Verified directly:
- **Finding 1** — `schema.sql:50` declares `canonical_name TEXT NOT NULL` with no default;
`scanner.service.ts:241` inserts only `(name)`. Confirmed.
- **Finding 3** — the `decayBeliefs` CTE is `WITH halflives AS (SELECT profile, CASE ...)`
with no `FROM` clause; Postgres rejects it on every hourly run. Confirmed; the
`VALUES`-based rewrite is the right fix.
- **Finding 4** — `UNIQUE (..., source, user_id)` with nullable `user_id` means
`ON CONFLICT` never fires for objective claims. Confirmed; `NULLS NOT DISTINCT` (PG16)
is correct, and dedup must indeed precede the constraint.
- **Finding 5** — `updateTrack` interpolates `Object.keys(body)` into the SET clause.
Confirmed.
- **Finding 7** — `next()` does `queue.slice(idx + 1)`, so the current track is always
index 0; `prev()`'s `idx > 0` guard can never pass after an auto-advance, and repeat-all
restarts from the track that just ended. Confirmed. The uncommitted end-of-queue fix in
the working tree is correct and independent of this bug.
- **Finding 2** — `nginx.conf.template` injects only `Bearer ${MUZICK_API_KEY}` for all of
`/api`, no admin-key location block. Confirmed at the config layer.
Judgment calls also endorsed: the repair ordering (rebuildability → admin UI → learning
loop), the "corrected claims" discipline, and option (a) for the dislike lifecycle — the
`:ro` mount is intentional.
Two additions:
1. **Finding 3 rollout:** the first successful decay run applies ~23 days of accrued decay
at once (obsession beliefs → ~0.32×). Correct behaviour, but recommendations will shift
visibly — expected, not a regression.
2. **Finding 4 dedup:** the one-time migration should collapse each duplicate group keeping
`MAX(last_reinforced_at)` (and ideally max confidence), not merely delete extra rows,
or reinforcement recency is lost.
+513
View File
@@ -0,0 +1,513 @@
// ---------------------------------------------------------------------------
// Migrations registry
// Add new entries at the END. Never edit or remove existing entries.
// Convention for id: "YYYYMMDD_short_description"
// ---------------------------------------------------------------------------
/** One forward-only, individually atomic schema change, recorded in `schema_migrations`. */
export interface Migration {
id: string;
sql: string;
}
export const MIGRATIONS: Migration[] = [
{
id: '20260608_track_artists',
sql: `
CREATE TABLE IF NOT EXISTS track_artists (
track_id UUID REFERENCES tracks(id) ON DELETE CASCADE,
artist_id UUID REFERENCES artists(id) ON DELETE CASCADE,
role TEXT NOT NULL DEFAULT 'main',
PRIMARY KEY (track_id, artist_id, role)
);
CREATE INDEX IF NOT EXISTS idx_track_artists_artist ON track_artists(artist_id);
-- Backfill main artist from albums for tracks not yet in track_artists
INSERT INTO track_artists (track_id, artist_id, role)
SELECT t.id, al.artist_id, 'main'
FROM tracks t
JOIN albums al ON al.id = t.album_id
WHERE NOT EXISTS (
SELECT 1 FROM track_artists ta WHERE ta.track_id = t.id
);
`,
},
{
id: '20260608_clear_lastfm_placeholder_images',
sql: `
UPDATE artists SET image_path = NULL
WHERE image_path LIKE '%2a96cbd8b46e442fc41c2b86b821562f%';
`,
},
{
// normalize_artist() was extended (schema.sql) to also split collaboration
// separators ( ; & / ) — not just commas/feat. STORED generated columns are
// NOT recomputed when the function definition changes, so force a recompute
// by touching the base column of every dependent row. ensureSchema() (which
// installs the new function) runs before migrations, so the new definition
// is already active here.
id: '20260612_recompute_normalized_artist',
sql: `
UPDATE artists SET name = name;
UPDATE tracks SET artist = artist;
`,
},
{
// The name-based Wikimedia Commons image fallback (now removed from the
// enrichment chain) frequently attached the wrong photo. Clear those rows so
// they fall back to the placeholder / a better source. Verified Wikidata
// images (fetched via MBID, step 3) also live on wikimedia.org but only on
// artists that HAVE an mbid, so restricting to mbid IS NULL spares them.
id: '20260612_clear_namebased_wikimedia_images',
sql: `
UPDATE artists SET image_path = NULL
WHERE mbid IS NULL AND image_path LIKE '%wikimedia.org%';
`,
},
{
// claim_fusion materialised view + compatibility views for v2 graph.
// Depends on claims, source_trust tables which are created by schema.sql
// (run before migrations). The MV resolves truth at read time as a weighted
// vote across claims per the fusion formula in spec §A.4.
id: '20260707_claim_fusion',
sql: `
CREATE OR REPLACE VIEW claim_fusion AS
SELECT
c.subject_type,
c.subject_id,
c.predicate,
c.object_type,
c.object_id,
COALESCE(c.user_id, '00000000-0000-0000-0000-000000000000'::uuid) AS user_id,
SUM(
st.trust * c.confidence *
GREATEST(0.1, 1.0 - EXTRACT(DAY FROM NOW() - c.last_reinforced_at) / 180.0)
) AS fused_value,
COUNT(*) AS claim_count,
MAX(c.last_reinforced_at) AS last_reinforced_at
FROM claims c
JOIN source_trust st ON st.key = c.source
GROUP BY c.subject_type, c.subject_id, c.predicate, c.object_type, c.object_id, c.user_id;
-- Compatibility view: track → artist credits via fusion
CREATE OR REPLACE VIEW track_artists_v2 AS
SELECT t.id AS track_id,
a.id AS artist_id,
a.name AS artist_name,
CASE cf.predicate WHEN 'credited_main_on' THEN 'main' ELSE 'featured' END AS role,
cf.fused_value AS confidence
FROM tracks t
JOIN claim_fusion cf
ON cf.subject_type = 'track' AND cf.subject_id = t.id
AND cf.predicate IN ('credited_main_on', 'featured_on')
AND cf.object_type = 'artist'
JOIN artists a ON a.id = cf.object_id;
-- Compatibility view: album → artist credits via fusion
CREATE OR REPLACE VIEW album_artists_v2 AS
SELECT al.id AS album_id,
a.id AS artist_id,
a.name AS artist_name,
CASE cf.predicate WHEN 'credited_main_on_album' THEN 'main' ELSE 'featured' END AS role,
cf.fused_value AS confidence
FROM albums al
JOIN claim_fusion cf
ON cf.subject_type = 'album' AND cf.subject_id = al.id
AND cf.predicate IN ('credited_main_on_album', 'featured_on_album')
AND cf.object_type = 'artist'
JOIN artists a ON a.id = cf.object_id;
`,
},
{
// Backfill existing data into the claims graph:
// 1. track_artists → credited_main_on / featured_on claims (source=tag)
// 2. artist_similar → same_scene_as claims (source=lastfm, confidence=match)
// This makes the graph immediately usable without waiting for re-enrichment.
id: '20260707_backfill_claims',
sql: `
-- 1. Populate claims from track_artists (tag-derived)
INSERT INTO claims (subject_type, subject_id, predicate, object_type, object_id, source, confidence, evidence_at)
SELECT
'track' AS subject_type,
ta.track_id AS subject_id,
CASE WHEN ta.role = 'main' THEN 'credited_main_on' ELSE 'featured_on' END AS predicate,
'artist' AS object_type,
ta.artist_id AS object_id,
'tag' AS source,
1.0 AS confidence,
NOW() AS evidence_at
FROM track_artists ta
ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id) DO NOTHING;
-- 2. Populate claims from artist_similar (Last.fm-derived)
INSERT INTO claims (subject_type, subject_id, predicate, object_type, object_id, source, confidence, evidence_at)
SELECT
'artist' AS subject_type,
ar.id AS subject_id,
'same_scene_as' AS predicate,
'artist' AS object_type,
similar_ar.id AS object_id,
'lastfm' AS source,
LEAST(asim.match, 1.0) AS confidence,
COALESCE(asim.fetched_at, NOW()) AS evidence_at
FROM artist_similar asim
JOIN artists ar ON ar.id = asim.artist_id
-- Resolve similar_name to an artist row so object_id is a real entity
JOIN artists similar_ar ON similar_ar.normalized_name = normalize_artist(asim.similar_name)
ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id) DO NOTHING;
-- 3. Also write claims with source='lastfm' for similar_name that didn't
-- resolve to an artist row (store as 'artist' object_type with name in raw)
INSERT INTO claims (subject_type, subject_id, predicate, object_type, object_id, source, confidence, evidence_at, raw)
SELECT
'artist' AS subject_type,
ar.id AS subject_id,
'same_scene_as' AS predicate,
'artist_name' AS object_type,
gen_random_uuid() AS object_id,
'lastfm' AS source,
LEAST(asim.match, 1.0) AS confidence,
COALESCE(asim.fetched_at, NOW()) AS evidence_at,
jsonb_build_object('similar_name', asim.similar_name)
FROM artist_similar asim
JOIN artists ar ON ar.id = asim.artist_id
WHERE NOT EXISTS (
SELECT 1 FROM artists a WHERE a.normalized_name = normalize_artist(asim.similar_name)
)
ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id) DO NOTHING;
`,
},
{
id: '20260708_materialize_claim_fusion',
sql: `
-- Drop old view + dependent views
DROP VIEW IF EXISTS claim_fusion CASCADE;
DROP VIEW IF EXISTS track_artists_v2 CASCADE;
DROP VIEW IF EXISTS album_artists_v2 CASCADE;
-- Create materialized view (same query as old view)
CREATE MATERIALIZED VIEW IF NOT EXISTS claim_fusion AS
SELECT
c.subject_type,
c.subject_id,
c.predicate,
c.object_type,
c.object_id,
COALESCE(c.user_id, '00000000-0000-0000-0000-000000000000'::uuid) AS user_id,
SUM(
st.trust * c.confidence *
GREATEST(0.1, 1.0 - EXTRACT(DAY FROM NOW() - c.last_reinforced_at) / 180.0)
) AS fused_value,
COUNT(*) AS claim_count,
MAX(c.last_reinforced_at) AS last_reinforced_at
FROM claims c
JOIN source_trust st ON st.key = c.source
GROUP BY c.subject_type, c.subject_id, c.predicate, c.object_type, c.object_id, c.user_id;
-- Unique index on the MV
CREATE UNIQUE INDEX IF NOT EXISTS idx_claim_fusion_pk ON claim_fusion (subject_type, subject_id, predicate, object_type, object_id, COALESCE(user_id, '00000000-0000-0000-0000-000000000000'));
-- Recreate compatibility views (now reading from MV)
CREATE OR REPLACE VIEW track_artists_v2 AS
SELECT t.id AS track_id,
a.id AS artist_id,
a.name AS artist_name,
CASE cf.predicate WHEN 'credited_main_on' THEN 'main' ELSE 'featured' END AS role,
cf.fused_value AS confidence
FROM tracks t
JOIN claim_fusion cf
ON cf.subject_type = 'track' AND cf.subject_id = t.id
AND cf.predicate IN ('credited_main_on', 'featured_on')
AND cf.object_type = 'artist'
JOIN artists a ON a.id = cf.object_id;
CREATE OR REPLACE VIEW album_artists_v2 AS
SELECT al.id AS album_id,
a.id AS artist_id,
a.name AS artist_name,
CASE cf.predicate WHEN 'credited_main_on_album' THEN 'main' ELSE 'featured' END AS role,
cf.fused_value AS confidence
FROM albums al
JOIN claim_fusion cf
ON cf.subject_type = 'album' AND cf.subject_id = al.id
AND cf.predicate IN ('credited_main_on_album', 'featured_on_album')
AND cf.object_type = 'artist'
JOIN artists a ON a.id = cf.object_id;
-- Refresh function for the MV
CREATE OR REPLACE FUNCTION refresh_claim_fusion() RETURNS void AS $$
BEGIN
REFRESH MATERIALIZED VIEW CONCURRENTLY claim_fusion;
END;
$$ LANGUAGE plpgsql;
-- Trigger function that notifies on claims changes
CREATE OR REPLACE FUNCTION notify_claim_fusion_change() RETURNS trigger AS $$
BEGIN
NOTIFY claim_fusion_changed;
RETURN NULL;
END;
$$ LANGUAGE plpgsql;
-- Trigger on claims table
DROP TRIGGER IF EXISTS trg_claim_fusion_refresh ON claims;
CREATE TRIGGER trg_claim_fusion_refresh AFTER INSERT OR UPDATE OR DELETE ON claims FOR EACH STATEMENT EXECUTE FUNCTION notify_claim_fusion_change();
`,
},
{
id: '20260708_fix_claim_fusion_index',
sql: `
-- Drop the expression-based unique index that blocks CONCURRENTLY refresh.
-- The MV's user_id column is already COALESCE'd (non-null) from the SELECT,
-- so we can use the plain column name instead.
DROP INDEX IF EXISTS idx_claim_fusion_pk;
CREATE UNIQUE INDEX idx_claim_fusion_pk
ON claim_fusion (subject_type, subject_id, predicate, object_type, object_id, user_id);
`,
},
{
id: '20260709_artists_mbid_unique',
sql: `
-- Replace the non-unique partial index on artists.mbid with a unique one,
-- so ON CONFLICT (mbid) works in MbSpineWriter.resolveArtist(). The partial
-- predicate (WHERE mbid IS NOT NULL) allows multiple artists with no MBID.
DROP INDEX IF EXISTS idx_artists_mbid;
CREATE UNIQUE INDEX IF NOT EXISTS artists_mbid_unique
ON artists (mbid) WHERE mbid IS NOT NULL;
`,
},
{
id: '20260717_claim_fusion_group_by_coalesce',
sql: `
-- Fix duplicate-key failures in REFRESH MATERIALIZED VIEW CONCURRENTLY.
-- The MV SELECTs COALESCE(user_id, zero-uuid) but GROUPed BY the raw
-- user_id, so a global enrichment claim (user_id NULL) and a default-user
-- behavior claim (user_id = zero-uuid, e.g. listener_behavior same_scene_as)
-- for the same edge fell into separate groups yet collapsed to the same
-- output key → two rows violating idx_claim_fusion_pk. Group by the same
-- COALESCE'd expression so they fuse into one row.
DROP MATERIALIZED VIEW IF EXISTS claim_fusion CASCADE;
CREATE MATERIALIZED VIEW claim_fusion AS
SELECT
c.subject_type,
c.subject_id,
c.predicate,
c.object_type,
c.object_id,
COALESCE(c.user_id, '00000000-0000-0000-0000-000000000000'::uuid) AS user_id,
SUM(
st.trust * c.confidence *
GREATEST(0.1, 1.0 - EXTRACT(DAY FROM NOW() - c.last_reinforced_at) / 180.0)
) AS fused_value,
COUNT(*) AS claim_count,
MAX(c.last_reinforced_at) AS last_reinforced_at
FROM claims c
JOIN source_trust st ON st.key = c.source
GROUP BY c.subject_type, c.subject_id, c.predicate, c.object_type, c.object_id,
COALESCE(c.user_id, '00000000-0000-0000-0000-000000000000'::uuid);
CREATE UNIQUE INDEX idx_claim_fusion_pk
ON claim_fusion (subject_type, subject_id, predicate, object_type, object_id, user_id);
CREATE OR REPLACE VIEW track_artists_v2 AS
SELECT t.id AS track_id,
a.id AS artist_id,
a.name AS artist_name,
CASE cf.predicate WHEN 'credited_main_on' THEN 'main' ELSE 'featured' END AS role,
cf.fused_value AS confidence
FROM tracks t
JOIN claim_fusion cf
ON cf.subject_type = 'track' AND cf.subject_id = t.id
AND cf.predicate IN ('credited_main_on', 'featured_on')
AND cf.object_type = 'artist'
JOIN artists a ON a.id = cf.object_id;
CREATE OR REPLACE VIEW album_artists_v2 AS
SELECT al.id AS album_id,
a.id AS artist_id,
a.name AS artist_name,
CASE cf.predicate WHEN 'credited_main_on_album' THEN 'main' ELSE 'featured' END AS role,
cf.fused_value AS confidence
FROM albums al
JOIN claim_fusion cf
ON cf.subject_type = 'album' AND cf.subject_id = al.id
AND cf.predicate IN ('credited_main_on_album', 'featured_on_album')
AND cf.object_type = 'artist'
JOIN artists a ON a.id = cf.object_id;
`,
},
{
id: '20260730_claims_dedup_nulls_not_distinct',
sql: `
-- The claims uniqueness constraint was declared as a plain
-- UNIQUE (subject_type, subject_id, predicate, object_type, object_id,
-- source, user_id). Every objective claim (MB, Discogs, tags) has
-- user_id IS NULL, and Postgres treats NULLs as distinct, so none of the
-- ON CONFLICT clauses in upsertClaim() / MbSpineWriter ever fired:
-- re-enrichment inserted a fresh duplicate row every time instead of
-- reinforcing. claim_fusion is SUM(trust * confidence * recency), so a
-- duplicated edge carried N times its intended weight.
--
-- Two phases, in this order (the constraint cannot be added while
-- duplicates exist):
-- 1. collapse each duplicate group into its most recently reinforced
-- row, carrying forward MAX(last_reinforced_at) / MAX(evidence_at) /
-- MAX(confidence) so reinforcement recency is not lost;
-- 2. replace the constraint with a NULLS NOT DISTINCT version (PG15+).
-- Phase 1: dedup.
CREATE TEMP TABLE claims_dedup ON COMMIT DROP AS
SELECT
id,
ROW_NUMBER() OVER ordered AS rn,
-- These MUST use the unordered window: an ORDER BY in the window spec
-- makes the default frame "UNBOUNDED PRECEDING TO CURRENT ROW", turning
-- MAX() into a running maximum rather than a per-group one.
MAX(last_reinforced_at) OVER grp AS max_last_reinforced_at,
MAX(evidence_at) OVER grp AS max_evidence_at,
MAX(confidence) OVER grp AS max_confidence
FROM claims
WINDOW
grp AS (
PARTITION BY subject_type, subject_id, predicate, object_type, object_id,
source,
COALESCE(user_id, '00000000-0000-0000-0000-000000000000'::uuid)
),
ordered AS (
grp ORDER BY last_reinforced_at DESC, evidence_at DESC, id
);
-- Keeper of each group absorbs the group's best values.
UPDATE claims c
SET last_reinforced_at = d.max_last_reinforced_at,
evidence_at = d.max_evidence_at,
confidence = d.max_confidence
FROM claims_dedup d
WHERE c.id = d.id
AND d.rn = 1;
DELETE FROM claims c
USING claims_dedup d
WHERE c.id = d.id
AND d.rn > 1;
-- Phase 2: replace the constraint. The live DB has drifted from
-- schema.sql, so find the existing constraint by its definition rather
-- than assuming Postgres' auto-generated name.
DO $mig$
DECLARE
cname TEXT;
BEGIN
FOR cname IN
SELECT con.conname
FROM pg_constraint con
WHERE con.conrelid = 'claims'::regclass
AND con.contype = 'u'
AND pg_get_constraintdef(con.oid) LIKE '%subject_type%'
AND pg_get_constraintdef(con.oid) NOT LIKE '%NULLS NOT DISTINCT%'
LOOP
EXECUTE format('ALTER TABLE claims DROP CONSTRAINT %I', cname);
END LOOP;
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
WHERE conrelid = 'claims'::regclass AND conname = 'claims_edge_source_user_key'
) THEN
ALTER TABLE claims ADD CONSTRAINT claims_edge_source_user_key
UNIQUE NULLS NOT DISTINCT
(subject_type, subject_id, predicate, object_type, object_id, source, user_id);
END IF;
END $mig$;
`,
},
{
id: '20260730_feedback_track_id_set_null',
sql: `
-- feedback is an audit log, but feedback.track_id was
-- REFERENCES tracks(id) ON DELETE CASCADE. hardDeleteTrack() /
-- permanentlyDeleteTrack() insert a 'deleted_permanent' row and then
-- delete the track, so the audit row deleted itself — which is exactly
-- why the live feedback table contains zero 'deleted_permanent' rows.
-- Switch to ON DELETE SET NULL so audit rows outlive their track. Nothing
-- reads feedback.track_id expecting non-null (there are no SELECTs against
-- it at all; the only other reference is the dedup merge in
-- mergeTracks(), which rewrites track_id to the survivor).
ALTER TABLE feedback ALTER COLUMN track_id DROP NOT NULL;
DO $mig$
DECLARE
cname TEXT;
BEGIN
FOR cname IN
SELECT con.conname
FROM pg_constraint con
WHERE con.conrelid = 'feedback'::regclass
AND con.contype = 'f'
AND con.confrelid = 'tracks'::regclass
AND con.confdeltype <> 'n' -- 'n' = SET NULL; anything else is wrong
LOOP
EXECUTE format('ALTER TABLE feedback DROP CONSTRAINT %I', cname);
END LOOP;
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
WHERE conrelid = 'feedback'::regclass
AND contype = 'f'
AND confrelid = 'tracks'::regclass
) THEN
ALTER TABLE feedback
ADD CONSTRAINT feedback_track_id_fkey
FOREIGN KEY (track_id) REFERENCES tracks(id) ON DELETE SET NULL;
END IF;
END $mig$;
`,
},
{
// Sorts after 20260730_claims_dedup_nulls_not_distinct and
// 20260730_feedback_track_id_set_null (append-only registry; 'h' > 'f' > 'c').
id: '20260730_hard_delete_audit_trail',
sql: `
-- Real hard deletion of disliked files (invariant §C). Three pieces:
--
-- 1. Denormalised identity on the audit row. 20260730_feedback_track_id_set_null
-- made feedback.track_id ON DELETE SET NULL so the 'deleted_permanent'
-- audit row outlives the track — but the surviving row no longer says
-- WHICH track was destroyed. Under real (irreversible) deletion that row
-- is the only forensic record, so copy the identifying fields into it.
ALTER TABLE feedback ADD COLUMN IF NOT EXISTS track_path TEXT;
ALTER TABLE feedback ADD COLUMN IF NOT EXISTS track_title TEXT;
ALTER TABLE feedback ADD COLUMN IF NOT EXISTS track_artist TEXT;
-- 2. An explicit "entered HIDDEN" timestamp. The pre-deletion grace period
-- (7 days, MUZICK_HARD_DELETE_GRACE_DAYS) is measured from it. Until now
-- the only nearby column was disliked_at, which lines up with HIDDEN
-- entry solely because 'HIDDEN' happens to be the state DEFAULT and
-- restoreDislike() deletes the row outright. That is far too incidental a
-- basis for an irreversible clock, so record it directly.
-- Added without a DEFAULT first: a DEFAULT would backfill existing rows
-- with now(), resetting every in-flight dislike's clock.
ALTER TABLE dislikes ADD COLUMN IF NOT EXISTS hidden_at TIMESTAMP;
UPDATE dislikes SET hidden_at = disliked_at WHERE hidden_at IS NULL;
ALTER TABLE dislikes ALTER COLUMN hidden_at SET DEFAULT CURRENT_TIMESTAMP;
-- 3. A durable marker for files whose DB row is already gone but whose
-- bytes are not. The sweep commits the transaction FIRST and unlinks
-- afterwards, so this row is what makes the window crash-safe and an
-- unlink failure (EROFS, EACCES, EBUSY) recorded rather than swallowed.
-- Deliberately no FK to tracks: the track row no longer exists.
CREATE TABLE IF NOT EXISTS pending_file_deletions (
path TEXT PRIMARY KEY,
track_id UUID,
track_title TEXT,
track_artist TEXT,
requested_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
attempts INTEGER NOT NULL DEFAULT 0,
last_attempt_at TIMESTAMP,
last_error TEXT
);
CREATE INDEX IF NOT EXISTS idx_pending_file_deletions_requested
ON pending_file_deletions(requested_at);
`,
},
];
+44 -2
View File
@@ -211,8 +211,15 @@ CREATE TABLE IF NOT EXISTS track_genre (
CREATE TABLE IF NOT EXISTS dislikes (
track_id UUID PRIMARY KEY REFERENCES tracks(id) ON DELETE CASCADE,
disliked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-- When the track entered HIDDEN. The pre-deletion grace period
-- (MUZICK_HARD_DELETE_GRACE_DAYS, default 7 days) is measured from here —
-- separate from, and much longer than, grace_hours below. Recorded
-- explicitly rather than inferred from disliked_at, because an irreversible
-- clock must not rest on 'HIDDEN' merely being the state DEFAULT.
hidden_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
warned_at TIMESTAMP,
deleted_at TIMESTAMP,
-- HIDDEN -> WARNED delay (the ntfy warning), NOT the deletion delay.
grace_hours INTEGER DEFAULT 48,
state dislike_state DEFAULT 'HIDDEN'
);
@@ -257,13 +264,42 @@ CREATE INDEX IF NOT EXISTS idx_play_history_user_played_at ON play_history(user_
CREATE TABLE IF NOT EXISTS feedback (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
user_id UUID NOT NULL,
track_id UUID REFERENCES tracks(id) ON DELETE CASCADE,
-- SET NULL, not CASCADE: feedback is an audit log and must outlive the
-- track. With CASCADE the 'deleted_permanent' row written by
-- hardDeleteTrack() deletes itself as soon as the track row goes.
track_id UUID REFERENCES tracks(id) ON DELETE SET NULL,
action TEXT NOT NULL,
-- Denormalised track identity, written on 'deleted_permanent' rows only.
-- track_id goes NULL when the track is deleted, so without these the audit
-- row survives but no longer says what was destroyed.
track_path TEXT,
track_title TEXT,
track_artist TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_feedback_user_action ON feedback(user_id, action);
-- Files whose DB row has been deleted but whose bytes are still on disk.
-- The cleanup sweep commits its transaction BEFORE unlinking, so this table is
-- the durable record of that window: a row is inserted with the deletion and
-- removed only once the unlink succeeds. A row that lingers means the unlink
-- failed (EROFS/EACCES/...) or the worker died mid-sweep; the next sweep retries
-- it. No FK to tracks — the track is already gone.
CREATE TABLE IF NOT EXISTS pending_file_deletions (
path TEXT PRIMARY KEY,
track_id UUID,
track_title TEXT,
track_artist TEXT,
requested_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
attempts INTEGER NOT NULL DEFAULT 0,
last_attempt_at TIMESTAMP,
last_error TEXT
);
CREATE INDEX IF NOT EXISTS idx_pending_file_deletions_requested
ON pending_file_deletions(requested_at);
CREATE TABLE IF NOT EXISTS track_audio_features (
track_id UUID PRIMARY KEY REFERENCES tracks(id) ON DELETE CASCADE,
bpm REAL,
@@ -340,7 +376,13 @@ CREATE TABLE IF NOT EXISTS claims (
last_reinforced_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
raw JSONB,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE (subject_type, subject_id, predicate, object_type, object_id, source, user_id)
-- NULLS NOT DISTINCT (PG15+) is load-bearing: user_id is NULL for every
-- objective claim, and with default NULLS DISTINCT semantics the
-- ON CONFLICT clauses in upsertClaim() / MbSpineWriter never fire, so
-- re-enrichment inserts duplicates instead of reinforcing.
CONSTRAINT claims_edge_source_user_key
UNIQUE NULLS NOT DISTINCT
(subject_type, subject_id, predicate, object_type, object_id, source, user_id)
);
CREATE INDEX IF NOT EXISTS idx_claims_subject ON claims (subject_type, subject_id, predicate);
+168
View File
@@ -0,0 +1,168 @@
// Row shapes for the muzick schema, extracted verbatim from db.service.ts so
// that file is about behaviour and this one about data. Re-exported from
// db.service.ts, so an existing `import { Track } from '../services/db.service.js'`
// keeps working.
export interface Artist {
id: string;
name: string;
mbid?: string | null;
discogs_id?: string | null;
image_path?: string | null;
}
export interface Album {
id: string;
artist_id: string;
title: string;
year?: number | null;
artwork_id?: string | null;
}
export interface TrackArtist {
id: string;
name: string;
role: 'main' | 'featured';
}
export interface Track {
id: string;
path: string;
hash: string;
title: string;
artist: string;
album_id: string;
duration: number;
state: string;
play_count: number;
skip_count: number;
dislike_count: number;
last_played_at?: Date | null;
mtime?: number | null;
source_type: string;
artists?: TrackArtist[];
}
export const FEEDBACK_ACTIONS = ['promoted', 'disliked', 'skipped', 'deleted_permanent'] as const;
export type FeedbackAction = (typeof FEEDBACK_ACTIONS)[number];
export interface HistoryEntry extends Track {
history_id: string;
batch_id: string | null;
played_at: Date;
completed: boolean;
}
export interface Genre {
id: string;
name: string;
parent_id?: string | null;
track_count?: number;
}
export interface DislikeEntry {
track_id: string;
disliked_at: Date;
warned_at: Date | null;
deleted_at: Date | null;
grace_hours: number;
state: string; // 'HIDDEN' | 'WARNED' | 'DELETED'
track_title: string;
track_artist: string;
track_path: string;
}
export interface ArtistWithAlbums {
id: string;
name: string;
mbid?: string | null;
discogs_id?: string | null;
image_path?: string | null;
albums: Album[];
}
export interface AlbumWithTracks {
id: string;
artist_id: string;
title: string;
year?: number | null;
artwork_id?: string | null;
tracks: Track[];
}
// ---------------------------------------------------------------------------
// v2 Recommendation Engine types
// ---------------------------------------------------------------------------
export interface Claim {
id: string;
user_id: string | null;
subject_type: string;
subject_id: string;
predicate: string;
object_type: string;
object_id: string;
source: string;
confidence: number;
evidence_at: Date;
last_reinforced_at: Date;
raw: unknown | null;
created_at: Date;
}
export interface Evidence {
id: string;
user_id: string;
entity_type: string;
entity_id: string;
signal: string;
profile: string;
weight: number;
context: unknown | null;
created_at: Date;
}
export interface ListenerBelief {
user_id: string;
profile: string;
entity_type: string;
entity_id: string;
dimension: string;
value: number;
confidence: number;
evidence_count: number;
last_reinforced_at: Date;
last_decayed_at: Date;
}
export interface ClaimEdge {
subjectType: string;
subjectId: string;
predicate: string;
objectType: string;
objectId: string;
fusedValue: number;
}
export interface SessionState {
session_id: string;
user_id: string;
started_at: Date;
last_interaction: Date;
context: string | null;
state_vector: Record<string, unknown>;
}
export interface DiversityBudget {
user_id: string;
dimension: string;
budget_share: number;
horizon_min: number;
}
export interface RepetitionRule {
user_id: string;
dimension: string;
min_distance: number;
}
+42
View File
@@ -0,0 +1,42 @@
/**
* Per-table allowlists of columns the HTTP `PUT /artists|albums|tracks/:id`
* endpoints may write.
*
* These endpoints pass `request.body as any` straight through, and the update
* builders interpolate `Object.keys(body)` into the SET clause as quoted
* identifiers — so without an allowlist a crafted key both closes the quoted
* identifier (SQL injection) and mass-assigns columns that are not user-editable.
*
* Deliberately excluded:
* - `tracks.path` / `tracks.hash` / `tracks.mtime` — owned by the scanner; the
* path is the only link to the file on the read-only `/music` bind.
* - `tracks.state` / `quarantined_at` / `deleted_at` — owned by the dislike
* lifecycle and the integrity sweep, not by metadata editing.
* - `play_count` / `skip_count` / `dislike_count` / `last_played_at` — learning
* signal; forgeable counters would poison the recommendation engine.
* - `source_type` — decides library vs. recommendation semantics.
* - `id` / `created_at` / `updated_at` and every generated column
* (`normalized_name`, `normalized_artist` — Postgres rejects writes anyway).
*/
export const UPDATABLE_COLUMNS = {
artists: ['name', 'canonical_name', 'sort_name', 'mbid', 'discogs_id', 'image_path'],
albums: ['artist_id', 'title', 'year', 'release_date', 'artwork_id', 'mbid'],
tracks: ['title', 'artist', 'album_id', 'duration', 'release_date'],
} as const;
/**
* Filter a partial update payload down to the allowlisted, defined columns for
* `table`. Unknown keys are dropped silently rather than raising: the callers
* take `Partial<T>` and the routes do no error mapping, so a 500 on an extra
* key would be worse than a no-op. The subsequent "No fields to update" throw
* still surfaces a payload that was *entirely* rejected.
*/
export function allowedFields<T extends object>(
table: keyof typeof UPDATABLE_COLUMNS,
data: T
): (keyof T & string)[] {
const allowed = UPDATABLE_COLUMNS[table] as readonly string[];
return Object.keys(data).filter(
(k) => allowed.includes(k) && (data as any)[k] !== undefined
) as (keyof T & string)[];
}
+136 -580
View File
@@ -1,4 +1,7 @@
import { readFile, unlink } from 'fs/promises';
// No `unlink` here, deliberately: the backend mounts /music `:ro` and must never
// remove a library file. Deletions are queued into pending_file_deletions and
// carried out by the worker's gated cleanup sweep.
import { readFile } from 'fs/promises';
import { fileURLToPath } from 'url';
import { dirname, join } from 'path';
import { Pool, PoolClient } from 'pg';
@@ -7,504 +10,32 @@ import { SearchService } from './search.service.js';
/** Anything with a `.query()` — either the shared Pool or a checked-out client. */
type Queryable = Pool | PoolClient;
export interface Artist {
id: string;
name: string;
mbid?: string | null;
discogs_id?: string | null;
image_path?: string | null;
}
import { MIGRATIONS } from '../db/migrations.js';
import { allowedFields } from '../db/updatable-columns.js';
export interface Album {
id: string;
artist_id: string;
title: string;
year?: number | null;
artwork_id?: string | null;
}
export interface TrackArtist {
id: string;
name: string;
role: 'main' | 'featured';
}
export interface Track {
id: string;
path: string;
hash: string;
title: string;
artist: string;
album_id: string;
duration: number;
state: string;
play_count: number;
skip_count: number;
dislike_count: number;
last_played_at?: Date | null;
mtime?: number | null;
source_type: string;
artists?: TrackArtist[];
}
export const FEEDBACK_ACTIONS = ['promoted', 'disliked', 'skipped', 'deleted_permanent'] as const;
export type FeedbackAction = (typeof FEEDBACK_ACTIONS)[number];
export interface HistoryEntry extends Track {
history_id: string;
batch_id: string | null;
played_at: Date;
completed: boolean;
}
export interface Genre {
id: string;
name: string;
parent_id?: string | null;
track_count?: number;
}
export interface DislikeEntry {
track_id: string;
disliked_at: Date;
warned_at: Date | null;
deleted_at: Date | null;
grace_hours: number;
state: string; // 'HIDDEN' | 'WARNED' | 'DELETED'
track_title: string;
track_artist: string;
track_path: string;
}
export interface ArtistWithAlbums {
id: string;
name: string;
mbid?: string | null;
discogs_id?: string | null;
image_path?: string | null;
albums: Album[];
}
export interface AlbumWithTracks {
id: string;
artist_id: string;
title: string;
year?: number | null;
artwork_id?: string | null;
tracks: Track[];
}
// ---------------------------------------------------------------------------
// v2 Recommendation Engine types
// ---------------------------------------------------------------------------
export interface Claim {
id: string;
user_id: string | null;
subject_type: string;
subject_id: string;
predicate: string;
object_type: string;
object_id: string;
source: string;
confidence: number;
evidence_at: Date;
last_reinforced_at: Date;
raw: unknown | null;
created_at: Date;
}
export interface Evidence {
id: string;
user_id: string;
entity_type: string;
entity_id: string;
signal: string;
profile: string;
weight: number;
context: unknown | null;
created_at: Date;
}
export interface ListenerBelief {
user_id: string;
profile: string;
entity_type: string;
entity_id: string;
dimension: string;
value: number;
confidence: number;
evidence_count: number;
last_reinforced_at: Date;
last_decayed_at: Date;
}
export interface ClaimEdge {
subjectType: string;
subjectId: string;
predicate: string;
objectType: string;
objectId: string;
fusedValue: number;
}
export interface SessionState {
session_id: string;
user_id: string;
started_at: Date;
last_interaction: Date;
context: string | null;
state_vector: Record<string, unknown>;
}
export interface DiversityBudget {
user_id: string;
dimension: string;
budget_share: number;
horizon_min: number;
}
export interface RepetitionRule {
user_id: string;
dimension: string;
min_distance: number;
}
// ---------------------------------------------------------------------------
// Migrations registry
// Add new entries at the END. Never edit or remove existing entries.
// Convention for id: "YYYYMMDD_short_description"
// ---------------------------------------------------------------------------
const MIGRATIONS: { id: string; sql: string }[] = [
{
id: '20260608_track_artists',
sql: `
CREATE TABLE IF NOT EXISTS track_artists (
track_id UUID REFERENCES tracks(id) ON DELETE CASCADE,
artist_id UUID REFERENCES artists(id) ON DELETE CASCADE,
role TEXT NOT NULL DEFAULT 'main',
PRIMARY KEY (track_id, artist_id, role)
);
CREATE INDEX IF NOT EXISTS idx_track_artists_artist ON track_artists(artist_id);
-- Backfill main artist from albums for tracks not yet in track_artists
INSERT INTO track_artists (track_id, artist_id, role)
SELECT t.id, al.artist_id, 'main'
FROM tracks t
JOIN albums al ON al.id = t.album_id
WHERE NOT EXISTS (
SELECT 1 FROM track_artists ta WHERE ta.track_id = t.id
);
`,
},
{
id: '20260608_clear_lastfm_placeholder_images',
sql: `
UPDATE artists SET image_path = NULL
WHERE image_path LIKE '%2a96cbd8b46e442fc41c2b86b821562f%';
`,
},
{
// normalize_artist() was extended (schema.sql) to also split collaboration
// separators ( ; & / ) — not just commas/feat. STORED generated columns are
// NOT recomputed when the function definition changes, so force a recompute
// by touching the base column of every dependent row. ensureSchema() (which
// installs the new function) runs before migrations, so the new definition
// is already active here.
id: '20260612_recompute_normalized_artist',
sql: `
UPDATE artists SET name = name;
UPDATE tracks SET artist = artist;
`,
},
{
// The name-based Wikimedia Commons image fallback (now removed from the
// enrichment chain) frequently attached the wrong photo. Clear those rows so
// they fall back to the placeholder / a better source. Verified Wikidata
// images (fetched via MBID, step 3) also live on wikimedia.org but only on
// artists that HAVE an mbid, so restricting to mbid IS NULL spares them.
id: '20260612_clear_namebased_wikimedia_images',
sql: `
UPDATE artists SET image_path = NULL
WHERE mbid IS NULL AND image_path LIKE '%wikimedia.org%';
`,
},
{
// claim_fusion materialised view + compatibility views for v2 graph.
// Depends on claims, source_trust tables which are created by schema.sql
// (run before migrations). The MV resolves truth at read time as a weighted
// vote across claims per the fusion formula in spec §A.4.
id: '20260707_claim_fusion',
sql: `
CREATE OR REPLACE VIEW claim_fusion AS
SELECT
c.subject_type,
c.subject_id,
c.predicate,
c.object_type,
c.object_id,
COALESCE(c.user_id, '00000000-0000-0000-0000-000000000000'::uuid) AS user_id,
SUM(
st.trust * c.confidence *
GREATEST(0.1, 1.0 - EXTRACT(DAY FROM NOW() - c.last_reinforced_at) / 180.0)
) AS fused_value,
COUNT(*) AS claim_count,
MAX(c.last_reinforced_at) AS last_reinforced_at
FROM claims c
JOIN source_trust st ON st.key = c.source
GROUP BY c.subject_type, c.subject_id, c.predicate, c.object_type, c.object_id, c.user_id;
-- Compatibility view: track → artist credits via fusion
CREATE OR REPLACE VIEW track_artists_v2 AS
SELECT t.id AS track_id,
a.id AS artist_id,
a.name AS artist_name,
CASE cf.predicate WHEN 'credited_main_on' THEN 'main' ELSE 'featured' END AS role,
cf.fused_value AS confidence
FROM tracks t
JOIN claim_fusion cf
ON cf.subject_type = 'track' AND cf.subject_id = t.id
AND cf.predicate IN ('credited_main_on', 'featured_on')
AND cf.object_type = 'artist'
JOIN artists a ON a.id = cf.object_id;
-- Compatibility view: album → artist credits via fusion
CREATE OR REPLACE VIEW album_artists_v2 AS
SELECT al.id AS album_id,
a.id AS artist_id,
a.name AS artist_name,
CASE cf.predicate WHEN 'credited_main_on_album' THEN 'main' ELSE 'featured' END AS role,
cf.fused_value AS confidence
FROM albums al
JOIN claim_fusion cf
ON cf.subject_type = 'album' AND cf.subject_id = al.id
AND cf.predicate IN ('credited_main_on_album', 'featured_on_album')
AND cf.object_type = 'artist'
JOIN artists a ON a.id = cf.object_id;
`,
},
{
// Backfill existing data into the claims graph:
// 1. track_artists → credited_main_on / featured_on claims (source=tag)
// 2. artist_similar → same_scene_as claims (source=lastfm, confidence=match)
// This makes the graph immediately usable without waiting for re-enrichment.
id: '20260707_backfill_claims',
sql: `
-- 1. Populate claims from track_artists (tag-derived)
INSERT INTO claims (subject_type, subject_id, predicate, object_type, object_id, source, confidence, evidence_at)
SELECT
'track' AS subject_type,
ta.track_id AS subject_id,
CASE WHEN ta.role = 'main' THEN 'credited_main_on' ELSE 'featured_on' END AS predicate,
'artist' AS object_type,
ta.artist_id AS object_id,
'tag' AS source,
1.0 AS confidence,
NOW() AS evidence_at
FROM track_artists ta
ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id) DO NOTHING;
-- 2. Populate claims from artist_similar (Last.fm-derived)
INSERT INTO claims (subject_type, subject_id, predicate, object_type, object_id, source, confidence, evidence_at)
SELECT
'artist' AS subject_type,
ar.id AS subject_id,
'same_scene_as' AS predicate,
'artist' AS object_type,
similar_ar.id AS object_id,
'lastfm' AS source,
LEAST(asim.match, 1.0) AS confidence,
COALESCE(asim.fetched_at, NOW()) AS evidence_at
FROM artist_similar asim
JOIN artists ar ON ar.id = asim.artist_id
-- Resolve similar_name to an artist row so object_id is a real entity
JOIN artists similar_ar ON similar_ar.normalized_name = normalize_artist(asim.similar_name)
ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id) DO NOTHING;
-- 3. Also write claims with source='lastfm' for similar_name that didn't
-- resolve to an artist row (store as 'artist' object_type with name in raw)
INSERT INTO claims (subject_type, subject_id, predicate, object_type, object_id, source, confidence, evidence_at, raw)
SELECT
'artist' AS subject_type,
ar.id AS subject_id,
'same_scene_as' AS predicate,
'artist_name' AS object_type,
gen_random_uuid() AS object_id,
'lastfm' AS source,
LEAST(asim.match, 1.0) AS confidence,
COALESCE(asim.fetched_at, NOW()) AS evidence_at,
jsonb_build_object('similar_name', asim.similar_name)
FROM artist_similar asim
JOIN artists ar ON ar.id = asim.artist_id
WHERE NOT EXISTS (
SELECT 1 FROM artists a WHERE a.normalized_name = normalize_artist(asim.similar_name)
)
ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id) DO NOTHING;
`,
},
{
id: '20260708_materialize_claim_fusion',
sql: `
-- Drop old view + dependent views
DROP VIEW IF EXISTS claim_fusion CASCADE;
DROP VIEW IF EXISTS track_artists_v2 CASCADE;
DROP VIEW IF EXISTS album_artists_v2 CASCADE;
-- Create materialized view (same query as old view)
CREATE MATERIALIZED VIEW IF NOT EXISTS claim_fusion AS
SELECT
c.subject_type,
c.subject_id,
c.predicate,
c.object_type,
c.object_id,
COALESCE(c.user_id, '00000000-0000-0000-0000-000000000000'::uuid) AS user_id,
SUM(
st.trust * c.confidence *
GREATEST(0.1, 1.0 - EXTRACT(DAY FROM NOW() - c.last_reinforced_at) / 180.0)
) AS fused_value,
COUNT(*) AS claim_count,
MAX(c.last_reinforced_at) AS last_reinforced_at
FROM claims c
JOIN source_trust st ON st.key = c.source
GROUP BY c.subject_type, c.subject_id, c.predicate, c.object_type, c.object_id, c.user_id;
-- Unique index on the MV
CREATE UNIQUE INDEX IF NOT EXISTS idx_claim_fusion_pk ON claim_fusion (subject_type, subject_id, predicate, object_type, object_id, COALESCE(user_id, '00000000-0000-0000-0000-000000000000'));
-- Recreate compatibility views (now reading from MV)
CREATE OR REPLACE VIEW track_artists_v2 AS
SELECT t.id AS track_id,
a.id AS artist_id,
a.name AS artist_name,
CASE cf.predicate WHEN 'credited_main_on' THEN 'main' ELSE 'featured' END AS role,
cf.fused_value AS confidence
FROM tracks t
JOIN claim_fusion cf
ON cf.subject_type = 'track' AND cf.subject_id = t.id
AND cf.predicate IN ('credited_main_on', 'featured_on')
AND cf.object_type = 'artist'
JOIN artists a ON a.id = cf.object_id;
CREATE OR REPLACE VIEW album_artists_v2 AS
SELECT al.id AS album_id,
a.id AS artist_id,
a.name AS artist_name,
CASE cf.predicate WHEN 'credited_main_on_album' THEN 'main' ELSE 'featured' END AS role,
cf.fused_value AS confidence
FROM albums al
JOIN claim_fusion cf
ON cf.subject_type = 'album' AND cf.subject_id = al.id
AND cf.predicate IN ('credited_main_on_album', 'featured_on_album')
AND cf.object_type = 'artist'
JOIN artists a ON a.id = cf.object_id;
-- Refresh function for the MV
CREATE OR REPLACE FUNCTION refresh_claim_fusion() RETURNS void AS $$
BEGIN
REFRESH MATERIALIZED VIEW CONCURRENTLY claim_fusion;
END;
$$ LANGUAGE plpgsql;
-- Trigger function that notifies on claims changes
CREATE OR REPLACE FUNCTION notify_claim_fusion_change() RETURNS trigger AS $$
BEGIN
NOTIFY claim_fusion_changed;
RETURN NULL;
END;
$$ LANGUAGE plpgsql;
-- Trigger on claims table
DROP TRIGGER IF EXISTS trg_claim_fusion_refresh ON claims;
CREATE TRIGGER trg_claim_fusion_refresh AFTER INSERT OR UPDATE OR DELETE ON claims FOR EACH STATEMENT EXECUTE FUNCTION notify_claim_fusion_change();
`,
},
{
id: '20260708_fix_claim_fusion_index',
sql: `
-- Drop the expression-based unique index that blocks CONCURRENTLY refresh.
-- The MV's user_id column is already COALESCE'd (non-null) from the SELECT,
-- so we can use the plain column name instead.
DROP INDEX IF EXISTS idx_claim_fusion_pk;
CREATE UNIQUE INDEX idx_claim_fusion_pk
ON claim_fusion (subject_type, subject_id, predicate, object_type, object_id, user_id);
`,
},
{
id: '20260709_artists_mbid_unique',
sql: `
-- Replace the non-unique partial index on artists.mbid with a unique one,
-- so ON CONFLICT (mbid) works in MbSpineWriter.resolveArtist(). The partial
-- predicate (WHERE mbid IS NOT NULL) allows multiple artists with no MBID.
DROP INDEX IF EXISTS idx_artists_mbid;
CREATE UNIQUE INDEX IF NOT EXISTS artists_mbid_unique
ON artists (mbid) WHERE mbid IS NOT NULL;
`,
},
{
id: '20260717_claim_fusion_group_by_coalesce',
sql: `
-- Fix duplicate-key failures in REFRESH MATERIALIZED VIEW CONCURRENTLY.
-- The MV SELECTs COALESCE(user_id, zero-uuid) but GROUPed BY the raw
-- user_id, so a global enrichment claim (user_id NULL) and a default-user
-- behavior claim (user_id = zero-uuid, e.g. listener_behavior same_scene_as)
-- for the same edge fell into separate groups yet collapsed to the same
-- output key → two rows violating idx_claim_fusion_pk. Group by the same
-- COALESCE'd expression so they fuse into one row.
DROP MATERIALIZED VIEW IF EXISTS claim_fusion CASCADE;
CREATE MATERIALIZED VIEW claim_fusion AS
SELECT
c.subject_type,
c.subject_id,
c.predicate,
c.object_type,
c.object_id,
COALESCE(c.user_id, '00000000-0000-0000-0000-000000000000'::uuid) AS user_id,
SUM(
st.trust * c.confidence *
GREATEST(0.1, 1.0 - EXTRACT(DAY FROM NOW() - c.last_reinforced_at) / 180.0)
) AS fused_value,
COUNT(*) AS claim_count,
MAX(c.last_reinforced_at) AS last_reinforced_at
FROM claims c
JOIN source_trust st ON st.key = c.source
GROUP BY c.subject_type, c.subject_id, c.predicate, c.object_type, c.object_id,
COALESCE(c.user_id, '00000000-0000-0000-0000-000000000000'::uuid);
CREATE UNIQUE INDEX idx_claim_fusion_pk
ON claim_fusion (subject_type, subject_id, predicate, object_type, object_id, user_id);
CREATE OR REPLACE VIEW track_artists_v2 AS
SELECT t.id AS track_id,
a.id AS artist_id,
a.name AS artist_name,
CASE cf.predicate WHEN 'credited_main_on' THEN 'main' ELSE 'featured' END AS role,
cf.fused_value AS confidence
FROM tracks t
JOIN claim_fusion cf
ON cf.subject_type = 'track' AND cf.subject_id = t.id
AND cf.predicate IN ('credited_main_on', 'featured_on')
AND cf.object_type = 'artist'
JOIN artists a ON a.id = cf.object_id;
CREATE OR REPLACE VIEW album_artists_v2 AS
SELECT al.id AS album_id,
a.id AS artist_id,
a.name AS artist_name,
CASE cf.predicate WHEN 'credited_main_on_album' THEN 'main' ELSE 'featured' END AS role,
cf.fused_value AS confidence
FROM albums al
JOIN claim_fusion cf
ON cf.subject_type = 'album' AND cf.subject_id = al.id
AND cf.predicate IN ('credited_main_on_album', 'featured_on_album')
AND cf.object_type = 'artist'
JOIN artists a ON a.id = cf.object_id;
`,
},
];
// Row shapes live in ../db/types.ts; re-exported here so existing importers
// (routes, generators, session-director) need no change.
import type {
Artist,
Album,
TrackArtist,
Track,
FeedbackAction,
HistoryEntry,
Genre,
DislikeEntry,
ArtistWithAlbums,
AlbumWithTracks,
Claim,
Evidence,
ListenerBelief,
ClaimEdge,
SessionState,
DiversityBudget,
RepetitionRule,
} from '../db/types.js';
import { FEEDBACK_ACTIONS } from '../db/types.js';
export * from '../db/types.js';
export class DbService {
/** Exposed so route handlers (e.g. settings) can query the database directly. */
@@ -897,42 +428,13 @@ export class DbService {
return (res.rows[0] as DislikeEntry) || null;
}
/**
* Permanently delete a track: remove from DB (cascades to related tables)
* and delete the physical file. Used by the cleanup_sweep worker after the
* full lifecycle (grace period + 24h warning) has elapsed.
*/
async hardDeleteTrack(userId: string, trackId: string, filePath: string): Promise<void> {
const { unlink } = await import('fs/promises');
// Log the permanent deletion feedback event first (before the track is gone)
await this.pgClient.query(
"INSERT INTO feedback (user_id, track_id, action) VALUES ($1, $2, 'deleted_permanent')",
[userId, trackId]
);
// Write evidence: manual_deleted → negative profile (strongest negative
// signal, spec §B.3). entity_id has no FK to tracks, so the row survives.
await this.recordEvidence({
user_id: userId,
entity_type: 'track',
entity_id: trackId,
signal: 'manual_deleted',
profile: 'negative',
weight: -0.90,
});
// Delete DB record (ON DELETE CASCADE handles track_genre, play_history,
// feedback, track_audio_features, track_lyrics)
await this.pgClient.query('DELETE FROM tracks WHERE id = $1', [trackId]);
// Delete physical file from disk
try {
await unlink(filePath);
} catch {
// File may already be gone (e.g. missing). Log but don't abort.
}
}
// NOTE: hardDeleteTrack() used to live here as a second, unreferenced deletion
// path with the same insert-then-delete shape as permanentlyDeleteTrack(), but
// with the audit insert outside any transaction and the unlink after the DELETE
// and silently swallowed. It had no callers. Removed rather than fixed twice
// over: there is now exactly one deletion path in the backend
// (permanentlyDeleteTrack, below) and exactly one unlink site in the whole
// system (workers/src/cleanup.service.ts).
async recordPlay(userId: string, trackId: string, completed: boolean, batchId?: string): Promise<string> {
if (!completed) {
@@ -1147,15 +649,20 @@ export class DbService {
async createArtist(data: Artist): Promise<Artist> {
const res = await this.pgClient.query(
`INSERT INTO artists (name, mbid, discogs_id, image_path)
VALUES (normalize_artist($1), $2, $3, $4) RETURNING *`,
[data.name, data.mbid, data.discogs_id, data.image_path]
// `canonical_name` is NOT NULL with no default, so omitting it fails on a
// fresh volume (the live DB's column predates the constraint). It holds the
// DISPLAY name: the raw input, not normalize_artist()'s output, which
// truncates on `/` and a standalone `x` ("AC/DC" -> "AC"). Matches the
// convention in workers' scanner.service.resolveOrCreateArtist.
`INSERT INTO artists (name, canonical_name, mbid, discogs_id, image_path)
VALUES (normalize_artist($1), $2, $3, $4, $5) RETURNING *`,
[data.name, data.name?.trim() || data.name, data.mbid, data.discogs_id, data.image_path]
);
return res.rows[0];
}
async updateArtist(id: string, data: Partial<Artist>): Promise<Artist> {
const fields = Object.keys(data).filter(k => data[k as keyof Artist] !== undefined);
const fields = allowedFields('artists', data);
if (fields.length === 0) throw new Error('No fields to update');
// Normalize name if it's being updated (the normalized_name generated column
@@ -1188,7 +695,7 @@ export class DbService {
}
async updateAlbum(id: string, data: Partial<Album>): Promise<Album> {
const fields = Object.keys(data).filter(k => data[k as keyof Album] !== undefined);
const fields = allowedFields('albums', data);
if (fields.length === 0) throw new Error('No fields to update');
const setClause = fields.map((f, i) => `"${f}" = $${i + 2}`).join(', ');
@@ -1231,7 +738,7 @@ export class DbService {
}
async updateTrack(id: string, data: Partial<Track>): Promise<Track> {
const fields = Object.keys(data).filter(k => data[k as keyof Track] !== undefined);
const fields = allowedFields('tracks', data);
if (fields.length === 0) throw new Error('No fields to update');
const setClause = fields.map((f, i) => `"${f}" = $${i + 2}`).join(', ');
@@ -1249,26 +756,67 @@ export class DbService {
}
/**
* Permanently delete a disliked track: remove the file from disk, then cascade-
* delete the DB record (which also removes the dislikes row via ON DELETE CASCADE).
* Logs a 'deleted_permanent' feedback event. Atomic: DB deletion only happens after
* the file is gone (or the file was already missing).
* Permanently delete a disliked track, skipping the grace period. Cascade-
* deletes the DB record (which also removes the dislikes row), writes the
* 'deleted_permanent' audit row, and hands the file itself to the worker.
*
* This method does NOT unlink. The backend mounts /music `:ro` on purpose —
* nothing reachable from an HTTP request may write to the library — so the
* unlink is queued as a pending_file_deletions row and performed by the
* cleanup sweep in the worker, the one process with an rw mount, and only when
* MUZICK_ALLOW_HARD_DELETE is on. Until then the row is visible evidence of an
* orphaned file rather than a lost one.
*
* Ordering: everything durable commits in one transaction, and the
* irreversible filesystem act happens strictly afterwards. The previous
* implementation unlinked first, which on this deployment meant EROFS and a
* 500 with the track still present.
*
* `filePath` is accepted for call-site compatibility but the path is re-read
* from the row under the transaction, so the queued deletion can never target
* a stale or caller-supplied path.
*/
async permanentlyDeleteTrack(userId: string, trackId: string, filePath: string): Promise<void> {
try {
await unlink(filePath);
} catch (err: any) {
if (err.code !== 'ENOENT') throw err;
}
async permanentlyDeleteTrack(userId: string, trackId: string, _filePath?: string): Promise<void> {
await this.withTransaction(async (client) => {
await client.query(
"INSERT INTO feedback (user_id, track_id, action) VALUES ($1, $2, 'deleted_permanent')",
[userId, trackId]
const trackRes = await client.query(
'SELECT path, title, artist FROM tracks WHERE id = $1 FOR UPDATE',
[trackId]
);
// CASCADE deletes dislikes, play_history, feedback, etc.
const track = trackRes.rows[0] as
| { path: string; title: string | null; artist: string | null }
| undefined;
if (!track) return; // Already gone; nothing to audit or unlink.
// Denormalised identity: track_id is ON DELETE SET NULL, so without these
// the surviving audit row would not say which track was destroyed.
await client.query(
`INSERT INTO feedback (user_id, track_id, action, track_path, track_title, track_artist)
VALUES ($1, $2, 'deleted_permanent', $3, $4, $5)`,
[userId, trackId, track.path, track.title, track.artist]
);
await client.query(
`INSERT INTO pending_file_deletions (path, track_id, track_title, track_artist)
VALUES ($1, $2, $3, $4)
ON CONFLICT (path) DO UPDATE SET requested_at = NOW()`,
[track.path, trackId, track.title, track.artist]
);
// CASCADE deletes dislikes, play_history, track_genre, etc.
await client.query('DELETE FROM tracks WHERE id = $1', [trackId]);
});
// Strongest negative signal (spec §B.3). entity_id has no FK to tracks, so
// the row survives the deletion. Outside the transaction: a failure here
// must not resurrect an already-audited deletion.
await this.recordEvidence({
user_id: userId,
entity_type: 'track',
entity_id: trackId,
signal: 'manual_deleted',
profile: 'negative',
weight: -0.90,
});
}
/**
@@ -1352,29 +900,36 @@ export class DbService {
* remaining files from disk, then hard-delete the rest. All inside a
* transaction so a partial failure leaves no orphans.
*
* File deletion happens BEFORE the transaction (best-effort: we delete the
* file, then commit the DB changes; if the DB commit fails the file is
* already gone but the track stays orphaned in the DB — a subsequent
* integrity sweep will catch it and mark it MISSING).
* Like permanentlyDeleteTrack(), this no longer unlinks: the loser files are
* queued into pending_file_deletions inside the same transaction and removed
* later by the worker's gated cleanup sweep. Previously the unlink ran BEFORE
* the transaction and rethrew anything other than ENOENT, so on this
* deployment (/music mounted `:ro` in the backend) every dedup merge failed
* with EROFS before touching the DB at all.
*/
async mergeDuplicates(keepId: string, deleteIds: string[]): Promise<void> {
// 1. Fetch paths for the tracks being deleted (before they're gone).
const { rows: losers } = await this.pgClient.query<{ id: string; path: string }>(
`SELECT id, path FROM tracks WHERE id = ANY($1::uuid[])`,
// 1. Fetch paths + identity for the tracks being deleted (before they go).
const { rows: losers } = await this.pgClient.query<{
id: string;
path: string;
title: string | null;
artist: string | null;
}>(
`SELECT id, path, title, artist FROM tracks WHERE id = ANY($1::uuid[])`,
[deleteIds]
);
// 2. Delete files from disk (best-effort — tolerate missing files).
// 2. DB transaction: queue the file removals, re-parent history, delete rows.
await this.withTransaction(async (client) => {
for (const row of losers) {
try {
await unlink(row.path);
} catch (err: any) {
if (err.code !== 'ENOENT') throw err;
}
await client.query(
`INSERT INTO pending_file_deletions (path, track_id, track_title, track_artist)
VALUES ($1, $2, $3, $4)
ON CONFLICT (path) DO UPDATE SET requested_at = NOW()`,
[row.path, row.id, row.title, row.artist]
);
}
// 3. DB transaction: re-parent history + delete rows.
await this.withTransaction(async (client) => {
for (const id of deleteIds) {
await client.query(
`UPDATE play_history SET track_id = $1 WHERE track_id = $2`,
@@ -1785,16 +1340,16 @@ export class DbService {
*/
async decayBeliefs(): Promise<number> {
const res = await this.pgClient.query(`
WITH halflives AS (
SELECT profile,
CASE profile
WHEN 'longterm' THEN 365 * 86400
WHEN 'obsession' THEN 14 * 86400
WHEN 'discovery' THEN 30 * 86400
WHEN 'negative' THEN 180 * 86400
WHEN 'contextual' THEN 7 * 86400
ELSE 30 * 86400
END AS halflife_sec
-- NOTE: this MUST be a VALUES list, not "SELECT profile, CASE profile ...",
-- which has no FROM clause and is rejected by Postgres with 42703
-- (column "profile" does not exist) on every single run.
WITH halflives(profile, halflife_sec) AS (
VALUES
('longterm', 365 * 86400),
('obsession', 14 * 86400),
('discovery', 30 * 86400),
('negative', 180 * 86400),
('contextual', 7 * 86400)
)
UPDATE listener_beliefs lb
SET value = GREATEST(-1.0, LEAST(1.0, lb.value * POWER(0.5,
@@ -1840,3 +1395,4 @@ export class DbService {
return res.rowCount ?? 0;
}
}
+20 -1
View File
@@ -34,6 +34,9 @@ services:
MUZICK_ADMIN_KEY: ${MUZICK_ADMIN_KEY}
MUSIC_DIR: /music
volumes:
# READ-ONLY, deliberately. Nothing in the API request path may write to
# the library. Hard deletion of disliked files happens only in the worker,
# which is the sole service with an rw mount.
- /mnt/hdd1/media/Music:/music:ro
depends_on:
- db
@@ -49,6 +52,7 @@ services:
- "127.0.0.1:5174:80"
environment:
MUZICK_API_KEY: ${MUZICK_API_KEY}
MUZICK_ADMIN_KEY: ${MUZICK_ADMIN_KEY}
depends_on:
- backend
@@ -65,8 +69,23 @@ services:
DISCOGS_TOKEN: ${DISCOGS_TOKEN}
SOCKS_PROXY_URL: ${SOCKS_PROXY_URL}
MUSIC_DIR: /music
# Hard-deletion gates for the dislike lifecycle (invariant §C). All three
# default to the safe value inside cleanup.service.ts; they are listed here
# as documentation and are intentionally left unset.
#
# MASTER SWITCH — leave unset/false. While off, the cleanup sweep logs
# exactly which files it WOULD delete and changes nothing at all. Set to
# true by hand, only after reviewing a dry-run log:
# MUZICK_ALLOW_HARD_DELETE: "false"
# Blast-radius cap per sweep (default 5):
# MUZICK_HARD_DELETE_MAX_PER_SWEEP: "5"
# Days a track must sit in HIDDEN before its file is eligible (default 7):
# MUZICK_HARD_DELETE_GRACE_DAYS: "7"
volumes:
- /mnt/hdd1/media/Music:/music:ro
# READ-WRITE, and the only rw mount of the library in the stack. The worker
# is the sole process permitted to unlink a music file, and only via the
# gated cleanup sweep. The backend keeps `:ro`.
- /mnt/hdd1/media/Music:/music:rw
networks:
infra-net:
+25 -1
View File
@@ -14,7 +14,31 @@ This document outlines the critical rules that MUST be respected to maintain sys
### **C. Filesystem Safety (The "Irreversible Action" Rule)**
* **Invariant:** Hard deletion of a file from the filesystem is the final, irreversible step in the `PENDING_REMOVAL` lifecycle.
* **Mechanism:** A track only enters the `DELETE_FILE` state after it has been `warned` for at least 24 hours and the user has not explicitly triggered a `RESTORE`.
* **Invariant:** Exactly **one** process may unlink a library file, and exactly **one** code path may do it: the worker's cleanup sweep (`workers/src/cleanup.service.ts`). Nothing reachable from an HTTP request may write to the library.
#### Implemented behaviour
**Mount.** Only the `worker` service binds `/mnt/hdd1/media/Music` read-write. `backend` keeps `:ro`, so an API bug or a compromised request handler physically cannot remove a file. The backend's two deletion entry points (`permanentlyDeleteTrack`, the Quarantine immediate-delete endpoint; and `mergeDuplicates`, the admin dedup merge) therefore do **not** unlink — they commit the DB side and queue the path into `pending_file_deletions` for the worker to carry out under the gates below.
**Ordering.** The DB transaction commits *first*, the unlink happens *after*:
1. Select eligible candidates (all gates applied in SQL).
2. In one transaction: write the `deleted_permanent` audit row, insert a `pending_file_deletions` marker, `DELETE FROM tracks`. Commit.
3. `unlink()`. On success, clear the marker. On failure, log at **error** and increment `attempts` / record `last_error` on the marker.
A marker that outlives its sweep is the durable record of a file whose DB row is gone but whose bytes are not — from a failed unlink or a worker killed mid-sweep. The next sweep retries it. Failures are never swallowed.
**Audit row.** `feedback.track_id` is `ON DELETE SET NULL`, so the audit row survives its track but loses its identity. Deletion rows therefore carry denormalised `track_path`, `track_title`, `track_artist`. This row is the only forensic record that the file ever existed.
**Gates.** All three default to the safe value; none is set to an unsafe value anywhere in the repo.
| Env var | Default | Effect |
|---|---|---|
| `MUZICK_ALLOW_HARD_DELETE` | **off** | Master switch. While off the sweep runs its full selection logic and logs exactly which files it *would* delete, performs no unlink, and makes **no** DB write of any kind — not the audit row, not the terminal-state transition. The owner enables it by hand after reviewing a dry-run log. |
| `MUZICK_HARD_DELETE_MAX_PER_SWEEP` | `5` | Blast-radius cap. On hitting it the sweep stops, logs at error, and leaves the remainder for the next run, so a selection bug cannot empty the library in one pass. |
| `MUZICK_HARD_DELETE_GRACE_DAYS` | `7` | Minimum age since the track entered `HIDDEN`, measured from `dislikes.hidden_at` (an explicit column — an irreversible clock must not be inferred). Separate from, and much longer than, `dislikes.grace_hours` (default 48), which only governs `HIDDEN``WARNED`. |
* **Mechanism:** A track's file is deleted only when *all* of: 24h have passed since `warned_at`; `MUZICK_HARD_DELETE_GRACE_DAYS` have passed since `hidden_at`; the user has not triggered a `RESTORE` (which deletes the dislike row outright); the per-sweep cap has not been reached; and `MUZICK_ALLOW_HARD_DELETE` is on.
## 2. Known Technical Risks
+1 -1
View File
@@ -9,4 +9,4 @@ FROM nginx:stable-alpine
COPY --from=build /app/dist /usr/share/nginx/html
COPY nginx.conf.template /etc/nginx/templates/default.conf.template
EXPOSE 80
CMD ["/bin/sh", "-c", "envsubst '${MUZICK_API_KEY}' < /etc/nginx/templates/default.conf.template > /etc/nginx/conf.d/default.conf && nginx -g 'daemon off;'"]
CMD ["/bin/sh", "-c", "envsubst '${MUZICK_API_KEY} ${MUZICK_ADMIN_KEY}' < /etc/nginx/templates/default.conf.template > /etc/nginx/conf.d/default.conf && nginx -g 'daemon off;'"]
-18
View File
@@ -1,18 +0,0 @@
server {
listen 80;
location / {
root /usr/share/nginx/html;
index index.html index.htm;
try_files $uri $uri/ /index.html;
}
location /api {
proxy_pass http://backend:3000;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection 'upgrade';
proxy_set_header Host $host;
proxy_cache_bypass $http_upgrade;
}
}
+13
View File
@@ -7,6 +7,19 @@ server {
try_files $uri $uri/ /index.html;
}
# Admin endpoints need the admin key, not the regular API key. This prefix
# location is longer than "/api", and nginx picks the longest matching
# prefix location, so it wins for /api/admin/* while /api handles the rest.
location /api/admin/ {
proxy_pass http://backend:3000;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection 'upgrade';
proxy_set_header Host $host;
proxy_set_header Authorization "Bearer ${MUZICK_ADMIN_KEY}";
proxy_cache_bypass $http_upgrade;
}
location /api {
proxy_pass http://backend:3000;
proxy_http_version 1.1;
+920 -8
View File
File diff suppressed because it is too large Load Diff
+4 -2
View File
@@ -26,7 +26,7 @@ interface TrackRowProps {
showVibe?: boolean;
}
export function TrackRow({ track, queue, index, showActions = true, variant = 'default', showVibe = false }: TrackRowProps) {
export function TrackRow({ track, queue, showActions = true, variant = 'default', showVibe = false }: TrackRowProps) {
const { setQueue, playTrack, play, pause, currentTrack, isPlaying } = usePlaybackStore();
const dislikeTrack = useDislikeTrack();
const router = useRouter();
@@ -35,7 +35,9 @@ export function TrackRow({ track, queue, index, showActions = true, variant = 'd
const handlePlay = () => {
if (isCurrent) { isPlaying ? pause() : play(); return; }
setQueue(queue.slice(index));
// Queue the whole list and start at this track, so Previous can walk back
// into the tracks before it.
setQueue(queue);
playTrack(track);
};
+41 -4
View File
@@ -317,6 +317,17 @@ function FilterBar({
);
}
/** Human-readable one-liner for a failed admin request. */
function describeError(err: unknown): string {
const status = (err as { response?: { status?: number } })?.response?.status;
if (status === 401 || status === 403) {
return `Request rejected (HTTP ${status}) — the admin API key is missing or wrong.`;
}
if (status) return `Request failed with HTTP ${status}.`;
const message = err instanceof Error ? err.message : String(err);
return message || 'Unknown error.';
}
/* ─────────────────────────────────────────── Page ────────────────────────────────────────────────── */
export default function JobsPage() {
@@ -325,20 +336,26 @@ export default function JobsPage() {
const [filters, setFilters] = useState<Filters>(INITIAL_FILTERS);
const [expandedIds, setExpandedIds] = useState<Set<string>>(new Set());
const { data: stats, refetch: refetchStats } = useQuery({
const statsQ = useQuery({
queryKey: ['queueStats'],
queryFn: () => jobsService.getQueueStats(),
refetchInterval: autoRefresh ? 3000 : false,
// Stop polling once the endpoint is failing — otherwise a permission or
// outage error is retried silently every 3s forever.
refetchInterval: (query) => (autoRefresh && !query.state.error ? 3000 : false),
staleTime: 1000,
});
const { data: history, refetch: refetchHistory } = useQuery({
const historyQ = useQuery({
queryKey: ['jobHistory'],
queryFn: () => jobsService.getJobHistory(200),
refetchInterval: autoRefresh ? 5000 : false,
refetchInterval: (query) => (autoRefresh && !query.state.error ? 5000 : false),
staleTime: 2000,
});
const { data: stats, refetch: refetchStats } = statsQ;
const { data: history, refetch: refetchHistory } = historyQ;
const loadError = statsQ.error ?? historyQ.error;
useEffect(() => {
if (selectedTab === 'overview') refetchStats();
else refetchHistory();
@@ -415,7 +432,27 @@ export default function JobsPage() {
{/* ── Content ── */}
<div className="flex-1 overflow-y-auto p-4 space-y-4">
{/* ── Error banner ── */}
{loadError && (
<div className="flex items-start gap-2 rounded-md border border-red-500/40 bg-red-500/10 px-3 py-2 text-sm text-red-300">
<AlertCircle className="w-4 h-4 mt-0.5 shrink-0" />
<div className="space-y-1">
<p className="font-medium">Could not load job data</p>
<p className="text-xs opacity-80">{describeError(loadError)}</p>
<button
onClick={() => { refetchStats(); refetchHistory(); }}
className="text-xs underline hover:no-underline"
>
Retry
</button>
</div>
</div>
)}
{/* ── Overview tab ── */}
{selectedTab === 'overview' && !stats && !loadError && (
<div className="text-center py-16 text-muted text-sm">Loading queue stats</div>
)}
{selectedTab === 'overview' && stats && (
<div className="grid grid-cols-2 md:grid-cols-3 lg:grid-cols-6 gap-4">
<StatCard icon={Clock} label="Waiting" value={stats.waiting} color="#eab308" />
+113 -29
View File
@@ -3,9 +3,17 @@ import type { Track } from '../types';
export type RepeatMode = 'none' | 'all' | 'one';
/**
* How many already-played tracks to keep behind the cursor. Bounds queue growth
* from the Vibe prefetch loop while still leaving real history for prev().
*/
const MAX_HISTORY = 50;
interface PlaybackState {
currentTrack: Track | null;
queue: Track[];
/** Cursor into `queue` for the current track, or -1 when the current track is not queued. */
currentIndex: number;
isPlaying: boolean;
position: number;
duration: number;
@@ -29,9 +37,33 @@ interface PlaybackState {
cycleRepeat: () => void;
}
/**
* Move to `index` in `queue`, trimming stale history so the queue stays bounded.
* Returns the state patch (queue may be re-sliced, so the index is adjusted).
*/
function advanceTo(queue: Track[], index: number) {
let nextQueue = queue;
let nextIndex = index;
if (index > MAX_HISTORY) {
const drop = index - MAX_HISTORY;
nextQueue = queue.slice(drop);
nextIndex = index - drop;
}
const track = nextQueue[nextIndex];
return {
queue: nextQueue,
currentIndex: nextIndex,
currentTrack: track,
position: 0,
duration: track?.duration ?? 0,
isPlaying: true,
};
}
export const usePlaybackStore = create<PlaybackState>((set, get) => ({
currentTrack: null,
queue: [],
currentIndex: -1,
isPlaying: false,
position: 0,
duration: 0,
@@ -40,25 +72,42 @@ export const usePlaybackStore = create<PlaybackState>((set, get) => ({
repeat: 'none',
shufflePlayed: new Set<string>(),
setQueue: (queue) => set({ queue, shufflePlayed: new Set() }),
setQueue: (queue) =>
set((state) => ({
queue,
// Keep the cursor pointing at whatever is playing, if it is still queued.
currentIndex: state.currentTrack ? queue.findIndex((t) => t.id === state.currentTrack!.id) : -1,
shufflePlayed: new Set(),
})),
playTrack: (track) =>
set({
set((state) => ({
currentTrack: track,
currentIndex: state.queue.findIndex((t) => t.id === track.id),
isPlaying: true,
position: 0,
duration: track.duration ?? 0,
shufflePlayed: new Set(),
}),
})),
play: () => set({ isPlaying: true }),
pause: () => set({ isPlaying: false }),
next: () => {
const { queue, currentTrack, shuffle, repeat, shufflePlayed } = get();
if (queue.length === 0) return;
if (queue.length === 0) {
set({ isPlaying: false, position: 0 });
return;
}
const idx = currentTrack ? queue.findIndex((t) => t.id === currentTrack.id) : -1;
// Prefer the tracked cursor; fall back to a lookup if it has drifted.
const tracked = get().currentIndex;
const idx =
tracked >= 0 && queue[tracked]?.id === currentTrack?.id
? tracked
: currentTrack
? queue.findIndex((t) => t.id === currentTrack.id)
: -1;
// Repeat one: replay current track
if (repeat === 'one' && currentTrack) {
@@ -71,49 +120,84 @@ export const usePlaybackStore = create<PlaybackState>((set, get) => ({
// so repeat-all doesn't bounce between the same few tracks.
const played = new Set(shufflePlayed);
if (currentTrack) played.add(currentTrack.id);
let remaining = queue.filter((t) => !played.has(t.id));
const indices = queue.map((_, i) => i);
let remaining = indices.filter((i) => !played.has(queue[i].id));
if (remaining.length === 0) {
if (repeat !== 'all') return;
if (repeat !== 'all') {
set({ isPlaying: false, position: 0 });
return;
}
// Lap complete — start a fresh one.
played.clear();
if (currentTrack) played.add(currentTrack.id);
remaining = queue.filter((t) => t.id !== currentTrack?.id);
if (remaining.length === 0) return;
remaining = indices.filter((i) => queue[i].id !== currentTrack?.id);
if (remaining.length === 0) {
set({ isPlaying: false, position: 0 });
return;
}
const pick = remaining[Math.floor(Math.random() * remaining.length)];
played.add(pick.id);
set({ currentTrack: pick, shufflePlayed: played, position: 0, duration: pick.duration ?? 0, isPlaying: true });
}
const pickIdx = remaining[Math.floor(Math.random() * remaining.length)];
played.add(queue[pickIdx].id);
const pick = queue[pickIdx];
set({
currentTrack: pick,
currentIndex: pickIdx,
shufflePlayed: played,
position: 0,
duration: pick.duration ?? 0,
isPlaying: true,
});
return;
}
// Sequential
const nextTrack = idx >= 0 ? queue[idx + 1] : null;
if (nextTrack) {
// Trim played tracks from queue to prevent unbounded growth (Vibe prefetch leak)
const trimmed = queue.slice(idx + 1);
set({ queue: trimmed, currentTrack: nextTrack, position: 0, duration: nextTrack.duration ?? 0, isPlaying: true });
} else if (repeat === 'all') {
const first = queue[0];
if (first) {
set({ currentTrack: first, position: 0, duration: first.duration ?? 0, isPlaying: true });
}
// Sequential — keep the queue intact and move the cursor, so prev() has
// history and repeat-all restarts from the true first track.
if (idx < 0) {
// Nothing playing (or the current track left the queue) — start at the top.
set(advanceTo(queue, 0));
} else if (idx + 1 < queue.length) {
set(advanceTo(queue, idx + 1));
} else if (repeat === 'all' && queue.length > 0) {
set(advanceTo(queue, 0));
} else {
// End of queue with no looping — stop playback so the store, DOM audio
// element, and MediaSession all agree nothing is playing. Without this
// the browser's now-playing widget can auto-resume the finished track
// from its near-end position, causing a tight "play last second → end →
// auto-resume → play last second → …" loop that also degrades performance.
set({ isPlaying: false, position: 0 });
}
},
prev: () => {
const { queue, currentTrack } = get();
const { queue, currentTrack, currentIndex } = get();
if (queue.length === 0) return;
const idx = currentTrack ? queue.findIndex((t) => t.id === currentTrack.id) : -1;
const prevTrack = idx > 0 ? queue[idx - 1] : null;
if (prevTrack) {
set({ currentTrack: prevTrack, position: 0, duration: prevTrack.duration ?? 0, isPlaying: true });
const idx =
currentIndex >= 0 && queue[currentIndex]?.id === currentTrack?.id
? currentIndex
: currentTrack
? queue.findIndex((t) => t.id === currentTrack.id)
: -1;
if (idx > 0) {
const prevTrack = queue[idx - 1];
set({
currentTrack: prevTrack,
currentIndex: idx - 1,
position: 0,
duration: prevTrack.duration ?? 0,
isPlaying: true,
});
}
},
setPosition: (position) => set({ position }),
setDuration: (duration) => set({ duration }),
setVolume: (volume) => set({ volume }),
setCurrentTrack: (currentTrack) => set({ currentTrack }),
setCurrentTrack: (currentTrack) =>
set((state) => ({
currentTrack,
currentIndex: currentTrack ? state.queue.findIndex((t) => t.id === currentTrack.id) : -1,
})),
toggleShuffle: () => set((state) => ({ shuffle: !state.shuffle })),
cycleRepeat: () =>
+2 -2
View File
@@ -15,7 +15,7 @@
*/
import { spawn } from 'child_process';
import mm from 'music-metadata';
import { Client as PgClient } from 'pg';
import type { Queryable } from './db.js';
// Lazy WASM singleton — heavy to load (~2.4 MB), so we initialise once and
// reuse across all enrichment jobs within the same worker process.
@@ -76,7 +76,7 @@ export interface AudioFeatures {
}
export class AudioFeaturesService {
constructor(private pgClient: PgClient) {}
constructor(private pgClient: Queryable) {}
async ensureSchema(): Promise<void> {
await this.pgClient.query(
+262 -34
View File
@@ -1,10 +1,73 @@
import { Client as PgClient } from 'pg';
import type { Pool } from 'pg';
import { unlink } from 'fs/promises';
import { withTransaction } from './db.js';
const NTFY_URL = process.env.NTFY_URL || '';
const NTFY_TOPIC = process.env.NTFY_TOPIC || 'muzick';
const SYSTEM_USER_ID = '00000000-0000-0000-0000-000000000000';
/**
* Hard deletion is IRREVERSIBLE and destroys the owner's music files. Three
* independent gates stand in front of it, all env-configurable, all defaulting
* to the safe value:
*
* - MUZICK_ALLOW_HARD_DELETE (default false) — the master switch. While it is
* off the sweep runs its full selection logic and logs exactly what it WOULD
* delete, but performs no unlink and makes no transition to the terminal
* state. This is the default mode of operation; the owner enables it by hand
* after reading a dry-run log. It is not set to true anywhere in the repo.
* - MUZICK_HARD_DELETE_MAX_PER_SWEEP (default 5) — blast-radius cap. On hitting
* it the sweep stops, logs at error, and leaves the remainder for next time,
* so a selection bug cannot empty the library in one run.
* - MUZICK_HARD_DELETE_GRACE_DAYS (default 7) — minimum age since the track
* entered HIDDEN (dislikes.hidden_at). Separate from, and much longer than,
* dislikes.grace_hours (default 48), which only governs HIDDEN -> WARNED.
*
* Note that only the worker container mounts /music read-write; the backend
* keeps `:ro`, so nothing in the API request path can unlink a library file.
*/
function envFlag(name: string): boolean {
const raw = (process.env[name] || '').trim().toLowerCase();
return raw === '1' || raw === 'true' || raw === 'yes';
}
function envInt(name: string, fallback: number, min: number): number {
const raw = (process.env[name] || '').trim();
if (!raw) return fallback;
const n = Number(raw);
if (!Number.isFinite(n) || !Number.isInteger(n) || n < min) {
console.error(
`[Cleanup] Ignoring invalid ${name}=${JSON.stringify(raw)}; using ${fallback}`
);
return fallback;
}
return n;
}
export const HARD_DELETE_DEFAULTS = {
/** MUZICK_ALLOW_HARD_DELETE */
allow: false,
/** MUZICK_HARD_DELETE_MAX_PER_SWEEP */
maxPerSweep: 5,
/** MUZICK_HARD_DELETE_GRACE_DAYS */
graceDays: 7,
} as const;
interface HardDeleteConfig {
allow: boolean;
maxPerSweep: number;
graceDays: number;
}
export function readHardDeleteConfig(): HardDeleteConfig {
return {
// Default OFF. Anything other than an explicit truthy opt-in is a dry run.
allow: envFlag('MUZICK_ALLOW_HARD_DELETE'),
maxPerSweep: envInt('MUZICK_HARD_DELETE_MAX_PER_SWEEP', HARD_DELETE_DEFAULTS.maxPerSweep, 1),
graceDays: envInt('MUZICK_HARD_DELETE_GRACE_DAYS', HARD_DELETE_DEFAULTS.graceDays, 1),
};
}
async function sendNtfy(title: string, message: string): Promise<void> {
if (!NTFY_URL) return;
try {
@@ -18,20 +81,51 @@ async function sendNtfy(title: string, message: string): Promise<void> {
}
}
export class CleanupSweepService {
constructor(private pgClient: PgClient) {}
interface Candidate {
track_id: string;
track_path: string;
track_title: string | null;
track_artist: string | null;
}
async runSweep(): Promise<{ warned: number; deleted: number }> {
export interface SweepResult {
warned: number;
/** Files actually unlinked and DB rows actually deleted. Always 0 in dry-run. */
deleted: number;
/**
* Eligible candidates seen. Selection stops at `maxPerSweep + 1` rows, so this
* saturates at cap+1 — read `eligible > maxPerSweep` as "more remain".
*/
eligible: number;
/** True when MUZICK_ALLOW_HARD_DELETE was off, i.e. nothing was destroyed. */
dryRun: boolean;
/** Unlinks that failed and were left recorded in pending_file_deletions. */
failed: number;
/** Previously-failed/interrupted unlinks reaped at the start of this sweep. */
reaped: number;
}
export class CleanupSweepService {
// A Pool, not a Client: finalizeDeleted() runs a real transaction and must
// check out a dedicated connection for it. The worker runs jobs with
// `concurrency: 10`, so a BEGIN on a shared connection would enrol other
// jobs' queries in this transaction and discard them on ROLLBACK.
constructor(private pool: Pool) {}
async runSweep(): Promise<SweepResult> {
const config = readHardDeleteConfig();
const warned = await this.advanceToWarned();
const deleted = await this.finalizeDeleted();
return { warned, deleted };
const reaped = await this.reapPendingFileDeletions(config);
const finalized = await this.finalizeDeleted(config);
return { warned, reaped, ...finalized };
}
/**
* Find HIDDEN dislikes past their grace period → send ntfy warning, mark WARNED.
* Not gated: this transition destroys nothing.
*/
private async advanceToWarned(): Promise<number> {
const res = await this.pgClient.query<{ track_id: string; track_title: string; track_artist: string }>(
const res = await this.pool.query<{ track_id: string; track_title: string; track_artist: string }>(
`SELECT d.track_id, t.title AS track_title, t.artist AS track_artist
FROM dislikes d
JOIN tracks t ON t.id = d.track_id
@@ -40,7 +134,7 @@ export class CleanupSweepService {
);
for (const row of res.rows) {
await this.pgClient.query(
await this.pool.query(
`UPDATE dislikes SET warned_at = NOW(), state = 'WARNED' WHERE track_id = $1`,
[row.track_id]
);
@@ -55,44 +149,178 @@ export class CleanupSweepService {
}
/**
* Find WARNED dislikes where 24h has passed since warned_at → delete file + DB record.
* Retry files whose DB row is already gone but whose bytes survive — an unlink
* that failed, or a worker that died between COMMIT and unlink. Subject to the
* same gates as a fresh deletion.
*/
private async finalizeDeleted(): Promise<number> {
const res = await this.pgClient.query<{ track_id: string; track_path: string; track_title: string }>(
`SELECT d.track_id, t.path AS track_path, t.title AS track_title
private async reapPendingFileDeletions(config: HardDeleteConfig): Promise<number> {
const res = await this.pool.query<{ path: string; track_id: string | null; attempts: number }>(
`SELECT path, track_id, attempts
FROM pending_file_deletions
ORDER BY requested_at ASC
LIMIT $1`,
[config.maxPerSweep]
);
if (res.rows.length === 0) return 0;
if (!config.allow) {
for (const row of res.rows) {
console.warn(
`[Cleanup] DRY RUN (MUZICK_ALLOW_HARD_DELETE off): orphaned file left on disk ` +
`after ${row.attempts} attempt(s): ${row.path}`
);
}
return 0;
}
let reaped = 0;
for (const row of res.rows) {
if (await this.unlinkAndSettle(row.path, row.track_id)) reaped++;
}
return reaped;
}
/**
* Find WARNED dislikes eligible for hard deletion and, unless dry-run, destroy
* them: DB transaction commits FIRST, then the file is unlinked.
*
* The old implementation unlinked before the transaction and `continue`d on
* failure, which is backwards on both counts: the irreversible act happened
* before anything durable recorded it, and a failure left no trace beyond a log
* line. Now the commit records the intent (audit row + pending_file_deletions
* marker) and the unlink either clears the marker or annotates it for retry.
* A crash in that window leaves the marker, which the next sweep reaps.
*/
private async finalizeDeleted(
config: HardDeleteConfig
): Promise<{ deleted: number; eligible: number; dryRun: boolean; failed: number }> {
// Ask for one more than the cap purely to detect overflow.
const res = await this.pool.query<Candidate>(
`SELECT d.track_id, t.path AS track_path,
t.title AS track_title, t.artist AS track_artist
FROM dislikes d
JOIN tracks t ON t.id = d.track_id
WHERE d.state = 'WARNED'
AND NOW() > d.warned_at + INTERVAL '24 hours'`
AND NOW() > d.warned_at + INTERVAL '24 hours'
AND NOW() > COALESCE(d.hidden_at, d.disliked_at) + ($1 || ' days')::interval
ORDER BY d.warned_at ASC
LIMIT $2`,
[config.graceDays, config.maxPerSweep + 1]
);
const eligible = res.rows.length;
if (eligible === 0) {
return { deleted: 0, eligible: 0, dryRun: !config.allow, failed: 0 };
}
let batch = res.rows;
if (batch.length > config.maxPerSweep) {
batch = batch.slice(0, config.maxPerSweep);
console.error(
`[Cleanup] Per-sweep deletion cap reached (${config.maxPerSweep}, ` +
`MUZICK_HARD_DELETE_MAX_PER_SWEEP). Stopping after ${batch.length}; ` +
`remaining eligible tracks are left for the next sweep.`
);
}
if (!config.allow) {
// Dry run: log the selection and return. No unlink, no state transition,
// no audit row — this branch performs no writes of any kind.
console.warn(
`[Cleanup] DRY RUN: MUZICK_ALLOW_HARD_DELETE is off. ` +
`${batch.length} file(s) WOULD be permanently deleted ` +
`(grace=${config.graceDays}d, cap=${config.maxPerSweep}). Nothing was changed.`
);
for (const row of batch) {
console.warn(
`[Cleanup] DRY RUN would delete: ${row.track_path} ` +
`("${row.track_title ?? '?'}" by ${row.track_artist ?? '?'}, track ${row.track_id})`
);
}
return { deleted: 0, eligible, dryRun: true, failed: 0 };
}
let deleted = 0;
for (const row of res.rows) {
let failed = 0;
for (const row of batch) {
try {
await unlink(row.track_path);
} catch (err: any) {
if (err.code !== 'ENOENT') {
console.error(`[Cleanup] Failed to delete file ${row.track_path}:`, err);
// Step 1: commit the DB side. The audit row carries the denormalised
// path/title/artist because feedback.track_id becomes NULL the moment
// the track is deleted (ON DELETE SET NULL) — this row is the only
// forensic record that this file ever existed.
await withTransaction(this.pool, async (client) => {
await client.query(
`INSERT INTO feedback (user_id, track_id, action, track_path, track_title, track_artist)
VALUES ($1, $2, 'deleted_permanent', $3, $4, $5)`,
[SYSTEM_USER_ID, row.track_id, row.track_path, row.track_title, row.track_artist]
);
await client.query(
`INSERT INTO pending_file_deletions (path, track_id, track_title, track_artist)
VALUES ($1, $2, $3, $4)
ON CONFLICT (path) DO UPDATE SET requested_at = NOW()`,
[row.track_path, row.track_id, row.track_title, row.track_artist]
);
await client.query('DELETE FROM tracks WHERE id = $1', [row.track_id]);
});
} catch (err) {
// withTransaction already rolled back and released the connection.
// Nothing was committed and no file was touched: safe to skip.
console.error(`[Cleanup] DB deletion failed for ${row.track_id}, file untouched:`, err);
failed++;
continue;
}
// Step 2: the irreversible part, now that it is durably recorded.
if (await this.unlinkAndSettle(row.track_path, row.track_id)) {
deleted++;
console.log(`[Cleanup] Permanently deleted: ${row.track_title} (${row.track_id})`);
} else {
failed++;
}
}
return { deleted, eligible, dryRun: false, failed };
}
/**
* Unlink one file and settle its pending_file_deletions marker: clear it on
* success (or ENOENT — the file is gone either way, which is the desired end
* state), annotate it on any other error so the failure is durable and
* retryable instead of swallowed. Returns true if the file is gone.
*
* Callers must only reach this once the DB side is committed.
*/
private async unlinkAndSettle(path: string, trackId: string | null): Promise<boolean> {
try {
await unlink(path);
} catch (err: any) {
if (err?.code !== 'ENOENT') {
console.error(
`[Cleanup] Unlink FAILED for ${path} (track ${trackId ?? 'unknown'}); ` +
`DB row already deleted, file left on disk and recorded in ` +
`pending_file_deletions for retry:`,
err
);
try {
await this.pool.query(
`UPDATE pending_file_deletions
SET attempts = attempts + 1, last_attempt_at = NOW(), last_error = $2
WHERE path = $1`,
[path, String(err?.message || err?.code || err)]
);
} catch (markerErr) {
console.error(`[Cleanup] Could not record unlink failure for ${path}:`, markerErr);
}
return false;
}
}
try {
await this.pgClient.query('BEGIN');
await this.pgClient.query(
`INSERT INTO feedback (user_id, track_id, action) VALUES ($1, $2, 'deleted_permanent')`,
[SYSTEM_USER_ID, row.track_id]
);
await this.pgClient.query('DELETE FROM tracks WHERE id = $1', [row.track_id]);
await this.pgClient.query('COMMIT');
deleted++;
console.log(`[Cleanup] Permanently deleted: ${row.track_title} (${row.track_id})`);
} catch (err) {
await this.pgClient.query('ROLLBACK');
console.error(`[Cleanup] DB deletion failed for ${row.track_id}:`, err);
}
}
return deleted;
await this.pool.query('DELETE FROM pending_file_deletions WHERE path = $1', [path]);
} catch (markerErr) {
// The file is gone; a stale marker only causes a harmless ENOENT retry.
console.error(`[Cleanup] Could not clear deletion marker for ${path}:`, markerErr);
}
return true;
}
}
+57
View File
@@ -0,0 +1,57 @@
import type { Pool, PoolClient, QueryResult, QueryResultRow } from 'pg';
/**
* The subset of pg's API the worker services actually need: a `query()` that
* takes SQL plus positional params.
*
* Services are typed against this rather than against `Client` so they work
* unchanged whether they are handed a `Pool` (the worker process — see
* index.ts), a `PoolClient` checked out for a transaction, or a plain `Client`
* (the one-off maintenance scripts in ./scripts, which are single-threaded and
* own their connection).
*
* IMPORTANT: a `Queryable` gives NO transaction guarantees. When a `Pool` is
* behind it, consecutive `query()` calls may land on different connections, so
* bare `BEGIN`/`COMMIT` must never be issued through it — check out a dedicated
* client with `withTransaction()` instead.
*/
export interface Queryable {
query<R extends QueryResultRow = any>(
sql: string,
params?: any[]
): Promise<QueryResult<R>>;
}
/**
* Run `fn` inside a transaction on a dedicated pooled connection, committing on
* success and rolling back on any throw. The client is always released.
*
* This is the only correct way to run a transaction in the worker: the process
* runs BullMQ with `concurrency: 10`, so issuing `BEGIN` on a shared connection
* would enrol another job's unrelated queries in this transaction — and discard
* them on `ROLLBACK`. (The backend documents the same hazard as its reason for
* using a `Pool`; see backend/src/app.ts.)
*/
export async function withTransaction<T>(
pool: Pool,
fn: (client: PoolClient) => Promise<T>
): Promise<T> {
const client = await pool.connect();
try {
await client.query('BEGIN');
const result = await fn(client);
await client.query('COMMIT');
return result;
} catch (err) {
try {
await client.query('ROLLBACK');
} catch (rollbackErr) {
// A failed ROLLBACK means the connection is unusable; log and move on —
// release() below discards it rather than returning it to the pool.
console.error('[DB] ROLLBACK failed:', rollbackErr);
}
throw err;
} finally {
client.release();
}
}
+13 -8
View File
@@ -1,4 +1,4 @@
import { Client as PgClient } from 'pg';
import type { Queryable } from './db.js';
import {
MusicBrainzClient,
LastFmClient,
@@ -65,7 +65,7 @@ export class EnrichmentService {
private readonly deezer = new DeezerClient();
private readonly audioFeatures: AudioFeaturesService;
constructor(private pgClient: PgClient) {
constructor(private pgClient: Queryable) {
this.audioFeatures = new AudioFeaturesService(pgClient);
}
@@ -330,14 +330,17 @@ export class EnrichmentService {
const sortName = mbArtist.sortName || this.generateSortName(canonicalName);
const newArtist = await this.pgClient.query(
`INSERT INTO artists (name, canonical_name, sort_name, mbid, normalized_name)
VALUES ($1, $2, $3, $4, $5)
// normalized_name is a GENERATED ALWAYS column in schema.sql
// (normalize_artist(name)); writing to it explicitly errors with
// 428C9 on any database built from schema.sql. Let Postgres derive it.
`INSERT INTO artists (name, canonical_name, sort_name, mbid)
VALUES ($1, $2, $3, $4)
ON CONFLICT (mbid) DO UPDATE SET
canonical_name = EXCLUDED.canonical_name,
sort_name = EXCLUDED.sort_name,
name = EXCLUDED.name
RETURNING id`,
[rawArtistName, canonicalName, sortName, mbArtist.artistMbid, normalized]
[rawArtistName, canonicalName, sortName, mbArtist.artistMbid]
);
const artistId = newArtist.rows[0].id;
@@ -371,10 +374,12 @@ export class EnrichmentService {
const sortName = this.generateSortName(rawName);
const result = await this.pgClient.query(
`INSERT INTO artists (name, canonical_name, sort_name, normalized_name)
VALUES ($1, $2, $3, $4)
// normalized_name is GENERATED ALWAYS (normalize_artist(name)) in
// schema.sql — inserting it explicitly fails with 428C9. Derived by PG.
`INSERT INTO artists (name, canonical_name, sort_name)
VALUES ($1, $2, $3)
RETURNING id`,
[rawName, rawName, sortName, normalized]
[rawName, rawName, sortName]
);
const artistId = result.rows[0].id;
+45 -210
View File
@@ -1,12 +1,13 @@
import { Worker, Job } from 'bullmq';
import { connection, QUEUE_NAME, queue } from './queue.js';
import { MetadataRefreshJob, AudioAnalysisJob, LibraryScanJob, ArtistSimilarityJob, ArtistImageJob, AlbumCoverJob, ReprocessArtistsJob } from './types.js';
import { Client as PgClient } from 'pg';
import { Pool } from 'pg';
import { ScannerService } from './scanner.service.js';
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 { 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
@@ -21,23 +22,40 @@ const VIBE_REAP_CRON = process.env.VIBE_REAP_CRON || '0 * * * *';
// Worker concurrency - how many jobs to process in parallel
const WORKER_CONCURRENCY = parseInt(process.env.WORKER_CONCURRENCY || '10', 10);
const pgClient = new PgClient({
// A Pool, not a single Client. The worker processes jobs with
// `concurrency: 10` on one event loop, so a shared Client would multiplex every
// job's queries onto one connection: a `BEGIN` issued by one job (see
// cleanup.service.ts) would enrol other jobs' unrelated queries in that
// transaction and discard them on `ROLLBACK`. Pool gives each transaction its
// own connection. Same reasoning as backend/src/app.ts.
//
// Sized at least as large as the job concurrency so concurrent jobs never
// serialise waiting for a connection.
const pgPool = new Pool({
connectionString: process.env.DATABASE_URL,
max: Math.max(WORKER_CONCURRENCY + 2, 10),
});
pgPool.on('error', (err) => {
// Idle-client errors would otherwise be an unhandled 'error' event and crash
// the worker; the pool discards the client itself.
console.error('[DB] Idle pool client error:', err);
});
async function initWorker() {
await pgClient.connect();
// Fail fast if the database is unreachable at boot (Pool is lazy otherwise).
await pgPool.query('SELECT 1');
console.log('Worker connected to PostgreSQL');
const scannerService = new ScannerService(pgClient, queue);
const enrichmentService = new EnrichmentService(pgClient);
const audioFeaturesService = new AudioFeaturesService(pgClient);
const scannerService = new ScannerService(pgPool, queue);
const enrichmentService = new EnrichmentService(pgPool);
const audioFeaturesService = new AudioFeaturesService(pgPool);
await audioFeaturesService.ensureSchema();
// Self-provision the enrichment schema additions (artist_similar + reused
// columns) at startup, alongside the integrity table.
await enrichmentService.ensureSchema();
await new IntegrityService(pgClient, queue).ensureSchema();
await new IntegrityService(pgPool, queue).ensureSchema();
const worker = new Worker(QUEUE_NAME, async (job: Job<any>) => {
// console.log(`Processing job: ${job.name} (ID: ${job.id})`);
@@ -93,24 +111,32 @@ async function initWorker() {
// Periodic self-healing pass: detect corrupt/missing track metadata,
// auto-fix via rescan + SQL strip, flag the rest for manual review.
console.log('[Integrity] Starting integrity sweep');
const integrityService = new IntegrityService(pgClient, queue);
const integrityService = new IntegrityService(pgPool, queue);
const summary = await integrityService.runSweep();
if (summary.aborted) {
console.error(`[Integrity] Sweep ABORTED by safety guard: ${summary.abortReason}`);
} else {
console.log(
`[Integrity] Sweep complete: detected=${summary.detected} fixed=${summary.fixed} needsReview=${summary.needsReview}`
);
}
break;
}
case 'cleanup_sweep': {
console.log('[Cleanup] Starting dislike cleanup sweep');
const cleanupService = new CleanupSweepService(pgClient);
const cleanupService = new CleanupSweepService(pgPool);
const result = await cleanupService.runSweep();
console.log(`[Cleanup] Sweep complete: warned=${result.warned} deleted=${result.deleted}`);
console.log(
`[Cleanup] Sweep complete: warned=${result.warned} eligible=${result.eligible} ` +
`deleted=${result.deleted} failed=${result.failed} reaped=${result.reaped}` +
(result.dryRun ? ' (DRY RUN — MUZICK_ALLOW_HARD_DELETE off, nothing deleted)' : '')
);
break;
}
case 'vibe_reap': {
// Stale-session reaper (Invariant B). Transitions ACTIVE batches with no
// interaction for 24h to RESOLVED. Runs hourly; the SQL is idempotent.
const reaped = await pgClient.query(
const reaped = await pgPool.query(
`UPDATE recommendation_batch
SET status = 'RESOLVED'
WHERE status = 'ACTIVE'
@@ -152,7 +178,7 @@ async function initWorker() {
if (!err?.message?.includes('already exists')) throw err;
}
const tracksRes = await pgClient.query(
const tracksRes = await pgPool.query(
`SELECT t.id, t.title, t.artist, al.title AS album, t.duration, t.play_count, t.source_type
FROM tracks t
LEFT JOIN albums al ON al.id = t.album_id
@@ -161,7 +187,7 @@ async function initWorker() {
const tracks = tracksRes.rows;
const genreMap = new Map<string, string[]>();
const genreRes = await pgClient.query(
const genreRes = await pgPool.query(
`SELECT tg.track_id, g.name FROM track_genre tg JOIN genre g ON g.id = tg.genre_id`
);
for (const row of genreRes.rows) {
@@ -194,202 +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 pgClient.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++;
// 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 pgClient.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 pgClient.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 pgClient.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 pgClient.query(`UPDATE tracks SET album_id = $1 WHERE album_id = $2`, [keeper_id, loser_id]);
await pgClient.query(`DELETE FROM albums WHERE id = $1`, [loser_id]);
}
// Remaining (non-colliding) albums move over cleanly.
await pgClient.query(
`UPDATE albums SET artist_id = $1 WHERE artist_id = $2`,
[result.artistId, artist.id]
);
await pgClient.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 pgClient.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 },
await reprocessArtists(job.data as ReprocessArtistsJob, {
pgPool,
queue,
enrichmentService,
});
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 pgClient.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 {
// 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 pgClient.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 pgClient.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 pgClient.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 pgClient.query(`UPDATE tracks SET album_id = $1 WHERE album_id = $2`, [keeper_id, loser_id]);
await pgClient.query(`DELETE FROM albums WHERE id = $1`, [loser_id]);
}
await pgClient.query(
`UPDATE albums SET artist_id = $1 WHERE artist_id = $2`,
[keepId, mergeId]
);
await pgClient.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;
}
default:
@@ -437,7 +272,7 @@ async function initWorker() {
// deleting data/postgres, or first deploy). On subsequent restarts the library
// is already populated, so an unconditional scan would be wasteful.
const startupDir = process.env.MUSIC_DIR || '/music';
const trackCount = await pgClient.query('SELECT COUNT(*)::int AS n FROM tracks');
const trackCount = await pgPool.query('SELECT COUNT(*)::int AS n FROM tracks');
if (trackCount.rows[0].n === 0) {
await queue.add('scan_library', { directory: startupDir }, {
removeOnComplete: { age: 86400, count: 100 },
@@ -460,7 +295,7 @@ async function initWorker() {
try {
await worker.close();
await queue.close();
await pgClient.end();
await pgPool.end();
console.log('[Shutdown] Clean shutdown complete.');
} catch (err) {
console.error('[Shutdown] Error during shutdown:', err);
+52 -11
View File
@@ -134,9 +134,24 @@ export interface RequestJsonOptions {
timeoutMs?: number;
}
// Per-host timestamp of the last request start, used to enforce the rate limit.
// Module-scoped so all clients sharing a host coordinate automatically.
const lastRequestAt = new Map<string, number>();
// Per-host rate-limit state. Module-scoped so all clients sharing a host
// coordinate automatically.
//
// `tail` is the promise chain for a host: every throttle() call links onto the
// host's current tail, so slot acquisition is strictly serialized. Without this
// chain the old implementation was a TOCTOU race — under the worker's
// `concurrency: 10`, ten jobs read the same `lastRequestAt`, slept the same
// duration and fired in the same tick, i.e. ~10 req/s against MusicBrainz's
// 1 req/s policy. Serialization is deliberately PER HOST (not global) so a slow
// MusicBrainz queue cannot starve Last.fm, Discogs, cover art, etc.
interface HostLimiter {
/** Timestamp of the last granted request slot. */
lastRequestAt: number;
/** Tail of the serialization chain; resolves when the previous waiter is done. */
tail: Promise<void>;
}
const hostLimiters = new Map<string, HostLimiter>();
const delay = (ms: number): Promise<void> =>
new Promise((resolve) => setTimeout(resolve, ms));
@@ -149,16 +164,42 @@ function hostOf(url: string): string {
}
}
/** Block until at least `minIntervalMs` has elapsed since this host's last request. */
async function throttle(host: string, minIntervalMs: number): Promise<void> {
const now = Date.now();
const last = lastRequestAt.get(host);
if (last !== undefined) {
const wait = minIntervalMs - (now - last);
/**
* Acquire this host's next request slot, blocking until at least
* `minIntervalMs` has elapsed since the previously granted slot.
*
* Concurrent callers for the same host queue strictly in arrival order: each
* one appends to `limiter.tail` and only computes its wait once every earlier
* waiter has already claimed its timestamp, so N concurrent callers are spaced
* `minIntervalMs` apart rather than all firing at once.
*/
function throttle(host: string, minIntervalMs: number): Promise<void> {
let limiter = hostLimiters.get(host);
if (!limiter) {
limiter = { lastRequestAt: 0, tail: Promise.resolve() };
hostLimiters.set(host, limiter);
}
const lim = limiter;
// Link onto the tail. The critical section (read lastRequestAt → sleep →
// write lastRequestAt) runs only after the previous waiter finished it.
const slot = lim.tail.then(async () => {
const wait = minIntervalMs - (Date.now() - lim.lastRequestAt);
if (wait > 0) await delay(wait);
lim.lastRequestAt = Date.now();
});
// The next caller waits for this slot. Swallow rejections on the chain itself
// so one failure can never poison the queue for subsequent requests.
lim.tail = slot.catch(() => {});
return slot;
}
lastRequestAt.set(host, Date.now());
}
/**
* Test-only handle on the per-host throttle, so its serialization can be
* asserted without issuing real network requests. Not used in production code.
*/
export const __throttleForTest = throttle;
/** Parse a Retry-After header (delta-seconds or HTTP date) into ms, or null. */
function parseRetryAfter(value: string | null): number | null {
+36 -7
View File
@@ -206,6 +206,35 @@ interface MbReleaseGroupSearchResponse {
'release-groups'?: MbReleaseGroupSearch[];
}
/**
* Log a failed MusicBrainz call.
*
* Every public method swallows errors and returns null/[] so enrichment never
* breaks. That makes a *rate-limited* MusicBrainz indistinguishable from "no
* data exists for your library" — the job still reports success. So throttling
* (429) and MB's own 503 "your requests are exceeding the allowable rate limit"
* are escalated to console.error with an explicit, greppable message, while
* ordinary failures stay at warn.
*/
function logMbFailure(op: string, err: unknown): void {
const e = err as { status?: number; message?: string; body?: string };
const status = typeof e?.status === 'number' ? e.status : undefined;
const msg = e?.message ?? String(err);
const rateLimited =
status === 429 ||
(status === 503 && /rate limit/i.test(`${e?.body ?? ''} ${msg}`));
if (rateLimited) {
console.error(
`[MusicBrainz] RATE LIMITED (HTTP ${status}) on ${op} — results are ` +
`INCOMPLETE, not empty. Enrichment will report success with missing ` +
`data. Check the 1 req/s throttle and MUSICBRAINZ_CONTACT. ${msg}`
);
return;
}
console.warn(`[MusicBrainz] ${op} failed:`, msg);
}
export class MusicBrainzClient {
private readonly cfg: MusicBrainzConfig;
private readonly userAgent: string;
@@ -278,7 +307,7 @@ export class MusicBrainzClient {
score,
};
} catch (err) {
console.warn('[MusicBrainz] lookupRecording failed:', (err as Error).message);
logMbFailure('lookupRecording', err);
return null;
}
}
@@ -313,7 +342,7 @@ export class MusicBrainzClient {
disambiguation: best.disambiguation ?? null,
};
} catch (err) {
console.warn('[MusicBrainz] searchArtist failed:', (err as Error).message);
logMbFailure('searchArtist', err);
return null;
}
}
@@ -365,7 +394,7 @@ export class MusicBrainzClient {
score,
};
} catch (err) {
console.warn('[MusicBrainz] searchReleaseGroup failed:', (err as Error).message);
logMbFailure('searchReleaseGroup', err);
return null;
}
}
@@ -399,7 +428,7 @@ export class MusicBrainzClient {
.map(([name, count]) => ({ name, weight: count / maxCount }))
.sort((a, b) => b.weight - a.weight);
} catch (err) {
console.warn('[MusicBrainz] getArtistTags failed:', (err as Error).message);
logMbFailure('getArtistTags', err);
return [];
}
}
@@ -434,7 +463,7 @@ export class MusicBrainzClient {
artistCredit: credit,
};
} catch (err) {
console.warn('[MusicBrainz] getRecording failed:', (err as Error).message);
logMbFailure('getRecording', err);
return null;
}
}
@@ -469,7 +498,7 @@ export class MusicBrainzClient {
artistCredit: credit,
};
} catch (err) {
console.warn('[MusicBrainz] getReleaseGroup failed:', (err as Error).message);
logMbFailure('getReleaseGroup', err);
return null;
}
}
@@ -512,7 +541,7 @@ export class MusicBrainzClient {
}))
.filter(r => r.targetMbid !== '');
} catch (err) {
console.warn('[MusicBrainz] getArtistRelations failed:', (err as Error).message);
logMbFailure('getArtistRelations', err);
return [];
}
}
+155 -8
View File
@@ -1,10 +1,28 @@
import fs from 'fs/promises';
import { Client as PgClient } from 'pg';
import type { Queryable } from './db.js';
import { Queue } from 'bullmq';
import { ScannerService } from './scanner.service.js';
const MUSIC_DIR = process.env.MUSIC_DIR || '/mnt/hdd1/media/Music';
// Safety rails for the missing-file pass. `MISSING` is written by an automated
// 03:00 cron and there is no automatic path back, so an unmounted /mnt/hdd1
// would otherwise flip the ENTIRE library to MISSING in one sweep.
//
// INTEGRITY_MISSING_ABORT_PCT: if more than this share of the tracks actually
// checked are unreachable, the pass writes nothing and logs an error. A genuine
// bulk deletion is rare and is better handled by an explicit re-scan than by a
// silent cron; a vanished mount is common and catastrophic.
const MISSING_ABORT_PCT = Number(process.env.INTEGRITY_MISSING_ABORT_PCT ?? 10);
// Below this many checked tracks the percentage is statistically meaningless
// (1 of 3 missing is 33%), so the threshold is not applied.
const MISSING_ABORT_MIN_SAMPLE = Number(
process.env.INTEGRITY_MISSING_ABORT_MIN_SAMPLE ?? 20
);
// Tracks examined per sweep. The sweep is paginated by a persisted keyset
// cursor, so consecutive runs deterministically cover the whole library.
const MISSING_PAGE_SIZE = Number(process.env.INTEGRITY_MISSING_PAGE_SIZE ?? 5000);
// Detection regexes for the known (now-fixed) corruption bug. These are the
// exact patterns used by the original one-off repair script:
// title : trailing ' (Enriched)' (possibly stacked)
@@ -31,12 +49,16 @@ export interface SweepSummary {
detected: number;
fixed: number;
needsReview: number;
/** True when a guard stopped the sweep before it wrote anything. */
aborted?: boolean;
/** Human-readable reason when `aborted` is set. */
abortReason?: string;
}
export class IntegrityService {
private scanner: ScannerService;
constructor(private pgClient: PgClient, private queue: Queue) {
constructor(private pgClient: Queryable, private queue: Queue) {
this.scanner = new ScannerService(pgClient, queue);
}
@@ -57,6 +79,64 @@ export class IntegrityService {
UNIQUE(track_id, issue_type)
)`
);
// Persisted keyset cursor for the paginated missing-file pass, so the sweep
// resumes where the previous run stopped instead of re-checking the same
// arbitrary 5,000 rows forever.
await this.pgClient.query(
`CREATE TABLE IF NOT EXISTS integrity_sweep_state (
key TEXT PRIMARY KEY,
cursor_path TEXT,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)`
);
}
/**
* Verify MUSIC_DIR actually looks like the mounted library before any pass
* that treats "file not on disk" as authoritative.
*
* An unmounted bind leaves either a missing path or an EMPTY directory —
* both indistinguishable, from `fs.access` on individual tracks, from "every
* file was deleted". Requiring at least one entry catches the empty-mountpoint
* case, which is the one that would otherwise destroy the library's state.
*/
private async checkMusicDirLive(): Promise<{ live: boolean; reason?: string }> {
try {
const st = await fs.stat(MUSIC_DIR);
if (!st.isDirectory()) {
return { live: false, reason: `${MUSIC_DIR} exists but is not a directory` };
}
} catch (err) {
return { live: false, reason: `${MUSIC_DIR} is not accessible: ${(err as Error).message}` };
}
try {
const entries = await fs.readdir(MUSIC_DIR);
if (entries.length === 0) {
return {
live: false,
reason: `${MUSIC_DIR} is empty — the library bind mount is almost certainly not mounted`,
};
}
} catch (err) {
return { live: false, reason: `${MUSIC_DIR} is not readable: ${(err as Error).message}` };
}
return { live: true };
}
private async getSweepCursor(): Promise<string | null> {
const res = await this.pgClient.query<{ cursor_path: string | null }>(
`SELECT cursor_path FROM integrity_sweep_state WHERE key = 'missing_files'`
);
return res.rows[0]?.cursor_path ?? null;
}
private async setSweepCursor(cursor: string | null): Promise<void> {
await this.pgClient.query(
`INSERT INTO integrity_sweep_state (key, cursor_path, updated_at)
VALUES ('missing_files', $1, NOW())
ON CONFLICT (key) DO UPDATE SET cursor_path = EXCLUDED.cursor_path, updated_at = NOW()`,
[cursor]
);
}
/** Tracks whose title or artist still carry the corruption markers. */
@@ -70,14 +150,28 @@ export class IntegrityService {
return res.rows;
}
/** Tracks (excluding ones already flagged DELETED) whose file is gone from disk. */
async detectMissingFiles(limit = 5000): Promise<MissingRow[]> {
/**
* Tracks (excluding ones already flagged DELETED) whose file is gone from disk.
*
* Paginated by keyset on `path` (UNIQUE, hence a total order): each call takes
* the next `limit` rows after `afterPath`. Callers persist the returned
* `nextCursor` so successive sweeps walk the whole library and wrap around,
* instead of re-checking an arbitrary unordered `LIMIT 5000` forever.
*
* Returns the checked count too, so the caller can apply a ratio guard.
*/
async detectMissingFiles(
limit = MISSING_PAGE_SIZE,
afterPath: string | null = null
): Promise<{ missing: MissingRow[]; checked: number; nextCursor: string | null }> {
const res = await this.pgClient.query<MissingRow>(
`SELECT id, path, title, artist
FROM tracks
WHERE state <> 'DELETED'
AND ($2::text IS NULL OR path > $2)
ORDER BY path
LIMIT $1`,
[limit]
[limit, afterPath]
);
const missing: MissingRow[] = [];
@@ -88,7 +182,13 @@ export class IntegrityService {
missing.push(row);
}
}
return missing;
// Short page (or empty) means we reached the end: wrap to the start so the
// next sweep begins a fresh cycle.
const nextCursor =
res.rows.length < limit ? null : res.rows[res.rows.length - 1].path;
return { missing, checked: res.rows.length, nextCursor };
}
/** Upsert an issue row keyed on (track_id, issue_type), refreshing on re-detection. */
@@ -137,6 +237,22 @@ export class IntegrityService {
const summary: SweepSummary = { detected: 0, fixed: 0, needsReview: 0 };
// --- Liveness guard ------------------------------------------------------
// Both passes treat the filesystem as authoritative (PASS 1 re-scans it,
// the missing-file pass writes state='MISSING' from it). If the library
// mount is gone, every conclusion the sweep draws is wrong and there is no
// automatic way back, so refuse to run at all.
const live = await this.checkMusicDirLive();
if (!live.live) {
summary.aborted = true;
summary.abortReason = live.reason;
console.error(
`[Integrity] ABORTING sweep: music library not available — ${live.reason}. ` +
`No tracks were examined and nothing was written.`
);
return summary;
}
// --- Corrupted metadata --------------------------------------------------
const corrupt = await this.detectCorruptedMetadata();
summary.detected = corrupt.length;
@@ -185,7 +301,34 @@ export class IntegrityService {
}
// --- Missing files -------------------------------------------------------
const missing = await this.detectMissingFiles();
const cursor = await this.getSweepCursor();
const { missing, checked, nextCursor } = await this.detectMissingFiles(
MISSING_PAGE_SIZE,
cursor
);
// Ratio guard: a large fraction of the page being unreachable means the
// filesystem, not the library, changed (partially-mounted disk, permission
// loss, a moved music root). Write nothing and leave the cursor untouched so
// the same page is re-examined once the cause is fixed.
const missingPct = checked > 0 ? (missing.length / checked) * 100 : 0;
if (
checked >= MISSING_ABORT_MIN_SAMPLE &&
missingPct > MISSING_ABORT_PCT
) {
summary.aborted = true;
summary.abortReason =
`${missing.length}/${checked} checked tracks unreachable ` +
`(${missingPct.toFixed(1)}% > ${MISSING_ABORT_PCT}% threshold)`;
console.error(
`[Integrity] ABORTING missing-file pass: ${summary.abortReason}. ` +
`Nothing was marked MISSING and the sweep cursor was not advanced. ` +
`Verify ${MUSIC_DIR} is fully mounted, then re-run; raise ` +
`INTEGRITY_MISSING_ABORT_PCT only if this bulk removal is genuine.`
);
return summary;
}
for (const row of missing) {
summary.detected++;
summary.needsReview++;
@@ -203,8 +346,12 @@ export class IntegrityService {
);
}
await this.setSweepCursor(nextCursor);
console.log(
`[Integrity] Sweep summary: detected=${summary.detected} fixed=${summary.fixed} needsReview=${summary.needsReview}`
`[Integrity] Sweep summary: detected=${summary.detected} fixed=${summary.fixed} ` +
`needsReview=${summary.needsReview} checked=${checked} missing=${missing.length} ` +
`nextCursor=${nextCursor === null ? '<start>' : nextCursor}`
);
return summary;
}
+2 -2
View File
@@ -1,4 +1,4 @@
import { Client as PgClient } from 'pg';
import type { Queryable } from './db.js';
import { MusicBrainzClient } from './integrations/musicbrainz.client.js';
import { normalizeForMatching } from './utils/fuzzy-match.js';
@@ -9,7 +9,7 @@ function generateSortName(name: string): string {
}
export class MbSpineWriter {
constructor(private pgClient: PgClient) {}
constructor(private pgClient: Queryable) {}
/**
* Fetch full artist-credit for a recording MBID and write claims.
+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!`);
}
+12 -4
View File
@@ -3,7 +3,7 @@ import { createReadStream } from 'fs';
import { createHash } from 'crypto';
import path from 'path';
import mm from 'music-metadata';
import { Client as PgClient } from 'pg';
import type { Queryable } from './db.js';
import { Queue } from 'bullmq';
import { MetadataRefreshJob, ArtistSimilarityJob, ArtistImageJob, AlbumCoverJob } from './types.js';
import { splitArtistNames, parseArtists } from './utils/artist-names.js';
@@ -80,7 +80,7 @@ export class ScannerService {
private enqueuedArtists = new Set<string>();
private enqueuedAlbums = new Set<string>();
constructor(private pgClient: PgClient, private queue: Queue) {}
constructor(private pgClient: Queryable, private queue: Queue) {}
async scanDirectory(directory: string) {
console.log(`[Scanner] Starting scan in: ${directory}`);
@@ -237,9 +237,17 @@ export class ScannerService {
return { id: String(found.rows[0].id), name: String(found.rows[0].name) };
}
// `canonical_name` is NOT NULL in schema.sql with no default, so it MUST be
// supplied here — omitting it makes every artist insert fail on a fresh
// volume (and processFile swallows the error, so the scan silently yields an
// empty library). It holds the DISPLAY name: we store the raw tag name, not
// the normalize_artist() output, because that function truncates on `/` and
// a standalone `x` ("AC/DC" -> "AC", "Felix Mendelssohn" -> "Feli"). The
// enrichment path later overwrites canonical_name with the MusicBrainz name;
// until then the raw tag is the most faithful display value we have.
const inserted = await this.pgClient.query(
'INSERT INTO artists (name) VALUES ($1) RETURNING id, name',
[name]
'INSERT INTO artists (name, canonical_name) VALUES ($1, $2) RETURNING id, name',
[name, rawName.trim() || name]
);
return { id: String(inserted.rows[0].id), name: String(inserted.rows[0].name) };
}
@@ -29,9 +29,13 @@ async function main() {
const integrity = new IntegrityService(pgClient, queue);
const summary = await integrity.runSweep();
if (summary.aborted) {
console.error(`[Repair] Sweep ABORTED by safety guard: ${summary.abortReason}`);
} else {
console.log(
`[Repair] Done: detected=${summary.detected} fixed=${summary.fixed} needsReview=${summary.needsReview}`
);
}
await pgClient.end();
console.log('[Repair] Connection closed.');