From 86cb8e5cbf0f01cb10da1450f55f7e0cfbed3427 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 28 Apr 2026 19:46:23 -0400 Subject: [PATCH 1/9] docs(spec): add M4b ListenBrainz inbound similarity ingest design Second M4 sub-plan (Fable #346). Periodic worker pulls track-track and artist-artist similarity edges from LB's public /explore/* endpoints, filters to the local library, stores top-20 per source track in two new tables (track_similarity, artist_similarity). Hourly tick, batch=5, weekly re-fetch cap per row, passive retry via timer. No auth (public endpoints). Discovery within library handled by LB's collaborative- filtering response naturally surfacing unplayed library tracks; spec notes M4c will add a serendipity floor + lazy fetch + sparse-fallback. Co-Authored-By: Claude Opus 4.7 --- .../specs/2026-04-28-m4b-similarity-design.md | 338 ++++++++++++++++++ 1 file changed, 338 insertions(+) create mode 100644 docs/superpowers/specs/2026-04-28-m4b-similarity-design.md diff --git a/docs/superpowers/specs/2026-04-28-m4b-similarity-design.md b/docs/superpowers/specs/2026-04-28-m4b-similarity-design.md new file mode 100644 index 00000000..3add5b22 --- /dev/null +++ b/docs/superpowers/specs/2026-04-28-m4b-similarity-design.md @@ -0,0 +1,338 @@ +# M4b — ListenBrainz inbound similarity ingest + +**Status:** Spec draft, 2026-04-28 +**Tracking:** Fable #346 +**Milestone:** M4 — ListenBrainz scrobble + similarity + radio +**Builds on:** M4a (outbound scrobble worker — shipped as PR #26) + +## 1. Goal + +A periodic background worker that pulls track-track and artist-artist similarity +edges from ListenBrainz's `/explore/similar-recordings/{mbid}` and +`/explore/similar-artists/{mbid}` endpoints and stores them in two new tables +(`track_similarity`, `artist_similarity`). Refreshes each row at most once per +7 days; bounded scope to "tracks the user has played" so cost stays +proportional to actual usage. + +When this slice ships, M4c can build candidate pools for radio from a +similarity graph rather than the user's whole library. + +## 2. Non-goals (explicit) + +- **Lazy fetch on radio request** — M4c. If a user clicks a never-played track + as seed and `track_similarity` is empty for it, M4c can synchronously call + `SimilarRecordings` then. +- **Score normalization to [0, 1]** — store raw LB scores + (`DOUBLE PRECISION`); M4c normalizes at query time if its scoring formula + needs it. +- **Symmetric edges** (storing both `(A, B)` and `(B, A)`) — store one-way as + LB returns. M4c queries `WHERE track_a_id = $seed`. +- **`musicbrainz_tag` and `user_cooccurrence` source values** — schema reserves + them via the `source` enum, but M4b only writes `'listenbrainz'`. +- **Suggested-additions / Lidarr integration** (LB-returned MBIDs not in the + library) — M5. Spec line 225 covers it. +- **Configurable LB algorithm parameter** — hardcoded for v1. +- **Per-user similarity overrides** — `track_similarity` has no `user_id`; + data is global per-instance. +- **Force-refresh HTTP endpoint** — operators can `UPDATE … SET fetched_at = + '1970-01-01' WHERE …` for break-glass. +- **Multi-instance worker safety** — single-process worker assumed (matches + M4a). +- **Frontend surface** — M4b is invisible until M4c uses the data. + +## 3. Architecture overview + +``` + ┌────────────────────────────────────┐ + │ similarity.Worker │ hourly tick + │ - SELECT distinct played tracks │ + │ with mbid where (no row OR │ + │ fetched_at < now() - 7d) │ + │ - LIMIT 5–10 per tick │ + └────────────┬───────────────────────┘ + │ + ▼ + ┌──────────────────────────────────────┐ + │ listenbrainz.Client │ + │ (extended from M4a) │ + │ GET /1/explore/similar-recordings/ │ + │ {mbid} │ + │ GET /1/explore/similar-artists/ │ + │ {artist_mbid} │ + │ No auth — public endpoints │ + └────────────┬─────────────────────────┘ + │ + ▼ + For each LB response: + - Filter to MBIDs we have in our local library + - Take top 20 (sorted by LB score) + - UPSERT into track_similarity / artist_similarity +``` + +### 3.1 New Go package + +**`internal/similarity/`** — the worker: + +- `Worker` struct (pool, client, logger, tick, batch, topK) +- `NewWorker(pool, client, logger) *Worker` — production defaults: 1h tick, + batch=5, topK=20 +- `(w *Worker) Run(ctx)` — blocks until ctx cancelled +- `(w *Worker) tickOnce(ctx) error` — drains one batch of tracks AND one + batch of artists; injectable for tests + +### 3.2 Existing-code extensions + +- **`internal/scrobble/listenbrainz/client.go`** — gains two methods on the + existing `Client` (which already houses `SubmitListens` from M4a): + - `SimilarRecordings(ctx, mbid, limit) ([]SimilarRecording, error)` + - `SimilarArtists(ctx, mbid, limit) ([]SimilarArtist, error)` + - Both return the same typed errors as `SubmitListens` (`ErrTransient`, + `ErrPermanent`, `*RetryAfterError`). 401 is defensive only — these are + public endpoints. + - The package stays at its current path. A future cleanup could move it + to `internal/listenbrainz/`, but that's a non-blocking refactor. + +- **`cmd/minstrel/main.go`** — start the worker alongside the M4a scrobble + worker: + ```go + similarityWorker := similarity.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "similarity")) + go similarityWorker.Run(ctx) + ``` + +### 3.3 Failure handling + +**Passive retry via timer.** Unlike M4a's durable scrobble queue: +- A failed `SimilarRecordings` call does NOT update `fetched_at`. The next + hourly tick selects the row again (it still satisfies the "needs fetch" + predicate) and retries. +- 429 with `Retry-After`: the worker logs the value and **aborts the current + tick** without updating `fetched_at` on any in-flight rows. The next + hourly tick (typically far longer than any LB-suggested back-off) picks + the work back up. Avoids mid-tick sleeps that would block the goroutine. +- ErrPermanent (4xx): logged as a warning + skipped. Permanent errors on + similarity reads typically mean the MBID isn't in LB's graph — there's no + remediation, but `fetched_at` stays old so we'll just retry forever (cheap + no-op since LB returns 4xx fast). Acceptable; could mark "permanently + empty" in a future iteration if telemetry shows it matters. +- ErrTransient (5xx, network): logged + skipped, retry next tick. + +No `scrobble_queue`-equivalent table needed; the work list IS the played-tracks +set + the `fetched_at` watermark. + +## 4. Database schema + +New migration `0009_similarity.up.sql`: + +```sql +CREATE TABLE track_similarity ( + track_a_id uuid NOT NULL REFERENCES tracks(id) ON DELETE CASCADE, + track_b_id uuid NOT NULL REFERENCES tracks(id) ON DELETE CASCADE, + score DOUBLE PRECISION NOT NULL, + source TEXT NOT NULL CHECK (source IN ('listenbrainz', 'musicbrainz_tag', 'user_cooccurrence')), + fetched_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (track_a_id, track_b_id, source), + CHECK (track_a_id <> track_b_id) +); + +CREATE INDEX track_similarity_a_score_idx + ON track_similarity (track_a_id, score DESC); + +CREATE TABLE artist_similarity ( + artist_a_id uuid NOT NULL REFERENCES artists(id) ON DELETE CASCADE, + artist_b_id uuid NOT NULL REFERENCES artists(id) ON DELETE CASCADE, + score DOUBLE PRECISION NOT NULL, + source TEXT NOT NULL CHECK (source IN ('listenbrainz', 'musicbrainz_tag', 'user_cooccurrence')), + fetched_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (artist_a_id, artist_b_id, source), + CHECK (artist_a_id <> artist_b_id) +); + +CREATE INDEX artist_similarity_a_score_idx + ON artist_similarity (artist_a_id, score DESC); +``` + +Down migration drops both tables. + +**Notes:** +- Primary key includes `source` so the schema can hold multiple parallel + similarity sources (per spec line 119). +- `(track_a_id, score DESC)` index matches the M4c hot-path query: "for seed + T, give me top-N similar tracks descending by score." +- `CHECK (a <> b)` prevents self-edges. +- `ON DELETE CASCADE` from both endpoints so deleting a track cleans up edges + on either side. + +## 5. New sqlc queries + +`internal/db/queries/similarity.sql`: + +```sql +-- name: ListPlayedTracksNeedingSimilarity :many +SELECT DISTINCT t.id, t.mbid +FROM tracks t +JOIN play_events pe ON pe.track_id = t.id +WHERE t.mbid IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM track_similarity ts + WHERE ts.track_a_id = t.id + AND ts.source = 'listenbrainz' + AND ts.fetched_at > now() - interval '7 days' + ) +ORDER BY t.id +LIMIT $1; + +-- name: ListPlayedArtistsNeedingSimilarity :many +SELECT DISTINCT ar.id, ar.mbid +FROM artists ar +JOIN tracks t ON t.artist_id = ar.id +JOIN play_events pe ON pe.track_id = t.id +WHERE ar.mbid IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM artist_similarity asim + WHERE asim.artist_a_id = ar.id + AND asim.source = 'listenbrainz' + AND asim.fetched_at > now() - interval '7 days' + ) +ORDER BY ar.id +LIMIT $1; + +-- name: GetTracksByMBIDs :many +SELECT id, mbid FROM tracks WHERE mbid = ANY($1::text[]); + +-- name: GetArtistsByMBIDs :many +SELECT id, mbid FROM artists WHERE mbid = ANY($1::text[]); + +-- name: UpsertTrackSimilarity :exec +INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at) +VALUES ($1, $2, $3, 'listenbrainz', now()) +ON CONFLICT (track_a_id, track_b_id, source) +DO UPDATE SET score = EXCLUDED.score, fetched_at = EXCLUDED.fetched_at; + +-- name: UpsertArtistSimilarity :exec +INSERT INTO artist_similarity (artist_a_id, artist_b_id, score, source, fetched_at) +VALUES ($1, $2, $3, 'listenbrainz', now()) +ON CONFLICT (artist_a_id, artist_b_id, source) +DO UPDATE SET score = EXCLUDED.score, fetched_at = EXCLUDED.fetched_at; +``` + +## 6. Worker algorithm + +`tickOnce(ctx)`: + +1. **Track pass:** + - `q.ListPlayedTracksNeedingSimilarity(batch=5)` → `[(track_id, mbid)…]` + - For each `(track_id, mbid)`: + - Call `c.SimilarRecordings(ctx, mbid, 100)` + - On 429 → log the `Retry-After` and **return from `tickOnce` early** + (don't update `fetched_at`; the next hourly tick will pick up the + work, which is virtually always longer than LB's `Retry-After`) + - On other error → log warn, skip (no `fetched_at` update; next tick + retries) + - On success: collect returned MBIDs → `q.GetTracksByMBIDs(returnedMBIDs)` + → take top 20 by score → for each `(local_id, score)` call + `q.UpsertTrackSimilarity(track_id, local_id, score)` +2. **Artist pass:** symmetric, using `ListPlayedArtistsNeedingSimilarity`, + `SimilarArtists`, `GetArtistsByMBIDs`, `UpsertArtistSimilarity`. + +**Constants:** +- Tick interval: 1 hour (production); injectable to ms-scale for tests. +- Batch size: 5 (production). 5 LB calls per pass × 2 passes = 10 LB calls/tick + → ~240/day, well under LB's documented rate limits (~100/5min unauth). +- Top-K: 20 per LB query. +- LB algorithm: hardcoded constant in the client (use LB's documented default + at implementation time). + +## 7. Test plan + +### 7.1 LB client unit tests (httptest) + +In `internal/scrobble/listenbrainz/client_test.go`, add 7 tests for each new +method: + +For `SimilarRecordings`: +- 200 + valid body → returns slice ordered by score +- 401 → `ErrAuth` (defensive — public endpoint shouldn't 401) +- 400 → `ErrPermanent` +- 503 → `ErrTransient` +- 429 with `Retry-After` → `*RetryAfterError` +- URL contains `algorithm=…` +- URL contains `limit=N` + +Same 7 tests for `SimilarArtists`. + +### 7.2 Worker integration tests (live DB + httptest) + +In `internal/similarity/worker_integration_test.go`: + +- `TickOnce_NoPlayedTracks_NoOp`: empty `play_events` → returns nil, no rows + in `track_similarity` +- `TickOnce_MapsLBResponseToLocalLibrary`: seed one played track with MBID; + LB returns 3 MBIDs (2 in library, 1 not); assert 2 rows inserted +- `TickOnce_TopKEnforced`: LB returns 50; assert + `count(*) WHERE track_a_id = $1` ≤ 20 +- `TickOnce_RespectsSevenDayCap`: row with `fetched_at = now()` → not + re-queried (the LB endpoint isn't called) +- `TickOnce_RefreshesStaleRow`: row with `fetched_at = now() - interval '8 + days'` → re-fetched, score updated, fetched_at bumps +- `TickOnce_429AbortsTick`: first response 429 with Retry-After → tickOnce + returns early; no `fetched_at` updates; subsequent tracks in the batch + are NOT processed (they're picked up on the next tick) +- `TickOnce_TransientErrorSkipsTrack`: 503 on one track → `fetched_at` + unchanged; other tracks in same batch process normally +- `TickOnce_FiltersInLibrary`: LB returns 5 MBIDs none of which are in the + library → 0 rows inserted (no error) +- `TickOnce_ArtistPassMirrors`: same coverage for the artist branch +- `TickOnce_NoMBIDOnTrack_Skipped`: track with `mbid IS NULL` → not selected + by `ListPlayedTracksNeedingSimilarity` + +### 7.3 Coverage target + +`internal/similarity` ≥ 75%. The worker has fewer error branches than M4a +because passive retry-via-timer eliminates the durable-queue state machine. + +The new `Similar*` methods in `internal/scrobble/listenbrainz` add ~14 tests +to that package; should keep its coverage ≥ 85%. + +### 7.4 Manual verification post-merge + +1. Restart server with M4b code. +2. After ≥1 hour, `psql -c "SELECT count(*) FROM track_similarity WHERE + source = 'listenbrainz'"` → non-zero (assuming user has any MBID-tagged + played tracks). +3. After ≥7 days, observe a `fetched_at` timestamp that's recent — + confirms re-fetch cadence. +4. If MBID coverage in the library is sparse, the table will be small. Not a + bug — M5 (Lidarr suggested-additions) is the eventual answer for + tracks-not-in-library; tagging coverage via Picard is a separate user-side + improvement. + +## 8. Backwards compatibility + +- New migration; no changes to existing schema. +- New package; no changes to consumer code (M4c will consume; M3's `Score()` + doesn't need the data until then). +- `MaybeEnqueue` and other M4a paths unchanged. +- Worker is new; production behavior identical to pre-M4b until the worker's + first tick fires (1 hour after restart). + +## 9. Decisions ledger + +| # | Decision | Rationale | +|---|---|---| +| 1 | Both `track_similarity` and `artist_similarity` in M4b | Symmetric pattern, M4c needs both, single-PR cost is small | +| 2 | Played-tracks-only input scope | Bounds work to user's actual interaction graph; LB's response naturally surfaces in-library tracks the user hasn't played, preserving discovery | +| 3 | Hourly tick, batch=5 | 240 LB calls/day fits well under LB rate limits; initial backfill in 24-48h | +| 4 | Top-K=20 per LB query, in-library only | Cuts long-tail noise; out-of-library tracks deferred to M5/Lidarr | +| 5 | Public endpoint, no auth | Similarity data is global to the instance; LB requires no token for `/explore/*` | +| 6 | Passive retry via timer (no durable queue) | Failure cost is "1 hour of staleness"; durable queue would be over-engineering vs. M4a's "lost scrobble" stakes | +| 7 | Hardcoded `algorithm` parameter | YAGNI; expose as YAML if telemetry warrants | + +## 10. Sub-plan progression (M4) + +- M4a (done) — outbound scrobble worker (PR #26). +- **M4b (this) — inbound LB similarity ingest.** +- M4c — radio similarity-driven candidate pool + queue refresh at 80% + (closes M4). M4c also picks up the discovery-mitigation work flagged in + brainstorm: serendipity floor (% random library picks), fallback to wider + pool when similarity-row count is sparse, lazy fetch on radio for + never-played seeds. From 9ec9caf667416ba5ec72c94fc3bbb24cdb5d45c4 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 28 Apr 2026 19:51:57 -0400 Subject: [PATCH 2/9] docs(plan): add M4b ListenBrainz inbound similarity ingest plan 8-task TDD-shaped plan covering: migration 0009 (track_similarity + artist_similarity tables with multi-source PK), sqlc queries (list- needing, get-by-mbids, upsert), LB client SimilarRecordings + Similar Artists methods (7 httptest tests each), Worker skeleton + tickOnce with track and artist passes (10 integration tests covering top-K=20, 7-day cap, in-library filter, 429-abort-tick, transient-skip, no-MBID skip), main.go boot wiring, final verification. --- .../plans/2026-04-28-m4b-similarity.md | 1551 +++++++++++++++++ 1 file changed, 1551 insertions(+) create mode 100644 docs/superpowers/plans/2026-04-28-m4b-similarity.md diff --git a/docs/superpowers/plans/2026-04-28-m4b-similarity.md b/docs/superpowers/plans/2026-04-28-m4b-similarity.md new file mode 100644 index 00000000..d39dbc5e --- /dev/null +++ b/docs/superpowers/plans/2026-04-28-m4b-similarity.md @@ -0,0 +1,1551 @@ +# M4b — ListenBrainz Inbound Similarity Ingest Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Background worker pulls track-track and artist-artist similarity edges from ListenBrainz's public `/explore/*` endpoints, filters to the local library (top-K=20 per source), stores in two new `track_similarity` + `artist_similarity` tables. Hourly tick, 7-day re-fetch cap per row, passive retry via timer. + +**Architecture:** New `internal/similarity/` package with a `Worker` goroutine paralleling M4a's scrobble worker. Existing `internal/scrobble/listenbrainz/Client` gets two new methods (`SimilarRecordings`, `SimilarArtists`). New migration adds two symmetric tables with multi-source primary keys + score-DESC indexes for M4c's hot-path query. No frontend changes. + +**Tech Stack:** Go 1.23 + sqlc + pgx/v5. No web changes. Runs alongside the M4a scrobble worker. + +**Reference:** design spec at `docs/superpowers/specs/2026-04-28-m4b-similarity-design.md`. + +--- + +## File Structure + +**New server files:** + +| File | Responsibility | +|---|---| +| `internal/db/migrations/0009_similarity.up.sql` + `.down.sql` | `track_similarity` + `artist_similarity` tables (multi-source PK, score-DESC indexes). | +| `internal/db/queries/similarity.sql` | New: `ListPlayedTracksNeedingSimilarity`, `ListPlayedArtistsNeedingSimilarity`, `GetTracksByMBIDs`, `GetArtistsByMBIDs`, `UpsertTrackSimilarity`, `UpsertArtistSimilarity`. | +| `internal/db/dbq/similarity.sql.go` | Generated bindings. | +| `internal/similarity/worker.go` | `Worker` struct, `NewWorker`, `Run` (hourly ticker), `tickOnce` (drains one batch of tracks + one batch of artists). | +| `internal/similarity/worker_test.go` | Pure unit tests (no DB). | +| `internal/similarity/worker_integration_test.go` | Live-DB + httptest LB end-to-end tests. | + +**Modified server files:** + +| File | Change | +|---|---| +| `internal/scrobble/listenbrainz/client.go` | Add `SimilarRecording` + `SimilarArtist` types, `SimilarRecordings(ctx, mbid, limit)` + `SimilarArtists(ctx, mbid, limit)` methods. Re-uses the existing `Client` struct, error types, and HTTP timeout pattern from M4a. | +| `internal/scrobble/listenbrainz/client_test.go` | 14 new httptest-driven tests (7 per method) covering success path, error mappings, query-param assertions. | +| `cmd/minstrel/main.go` | Construct `similarity.NewWorker(...)` after the M4a scrobble worker; `go similarityWorker.Run(ctx)`. | + +**No frontend changes.** M4b is invisible until M4c reads the data. + +--- + +## Task 1: Migration 0009 — schema + +**Files:** +- Create: `internal/db/migrations/0009_similarity.up.sql` +- Create: `internal/db/migrations/0009_similarity.down.sql` + +- [ ] **Step 1: Write the up migration** + +Create `internal/db/migrations/0009_similarity.up.sql`: + +```sql +-- M4b: ListenBrainz similarity ingest. Two symmetric tables — tracks and +-- artists — both keyed by (a_id, b_id, source) so the schema reserves room +-- for additional similarity sources (musicbrainz_tag, user_cooccurrence) +-- alongside listenbrainz. M4b only writes 'listenbrainz' rows. +-- +-- The (a_id, score DESC) index satisfies M4c's hot path: "for seed S, give +-- me top-N similar tracks/artists descending by score." + +CREATE TABLE track_similarity ( + track_a_id uuid NOT NULL REFERENCES tracks(id) ON DELETE CASCADE, + track_b_id uuid NOT NULL REFERENCES tracks(id) ON DELETE CASCADE, + score DOUBLE PRECISION NOT NULL, + source TEXT NOT NULL CHECK (source IN ('listenbrainz', 'musicbrainz_tag', 'user_cooccurrence')), + fetched_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (track_a_id, track_b_id, source), + CHECK (track_a_id <> track_b_id) +); + +CREATE INDEX track_similarity_a_score_idx + ON track_similarity (track_a_id, score DESC); + +CREATE TABLE artist_similarity ( + artist_a_id uuid NOT NULL REFERENCES artists(id) ON DELETE CASCADE, + artist_b_id uuid NOT NULL REFERENCES artists(id) ON DELETE CASCADE, + score DOUBLE PRECISION NOT NULL, + source TEXT NOT NULL CHECK (source IN ('listenbrainz', 'musicbrainz_tag', 'user_cooccurrence')), + fetched_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (artist_a_id, artist_b_id, source), + CHECK (artist_a_id <> artist_b_id) +); + +CREATE INDEX artist_similarity_a_score_idx + ON artist_similarity (artist_a_id, score DESC); +``` + +- [ ] **Step 2: Write the down migration** + +Create `internal/db/migrations/0009_similarity.down.sql`: + +```sql +DROP TABLE IF EXISTS artist_similarity; +DROP TABLE IF EXISTS track_similarity; +``` + +- [ ] **Step 3: Verify the migration applies cleanly** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' go test -race ./internal/db/` + +Expected: PASS. The migration test should now find version 9. + +- [ ] **Step 4: Commit** + +```bash +cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel +git add internal/db/migrations/0009_similarity.up.sql internal/db/migrations/0009_similarity.down.sql +git commit -m "feat(db): add migration 0009 for similarity (track_similarity + artist_similarity)" +``` + +--- + +## Task 2: sqlc queries + +**Files:** +- Create: `internal/db/queries/similarity.sql` +- Generated: `internal/db/dbq/similarity.sql.go` + +- [ ] **Step 1: Create the queries file** + +Create `internal/db/queries/similarity.sql`: + +```sql +-- name: ListPlayedTracksNeedingSimilarity :many +-- Tracks with at least one play, an MBID, and no fresh listenbrainz row +-- (no row at all, OR fetched_at older than 7 days). Used by the worker +-- to find work each tick. Bounded by $1. +SELECT DISTINCT t.id, t.mbid +FROM tracks t +JOIN play_events pe ON pe.track_id = t.id +WHERE t.mbid IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM track_similarity ts + WHERE ts.track_a_id = t.id + AND ts.source = 'listenbrainz' + AND ts.fetched_at > now() - interval '7 days' + ) +ORDER BY t.id +LIMIT $1; + +-- name: ListPlayedArtistsNeedingSimilarity :many +SELECT DISTINCT ar.id, ar.mbid +FROM artists ar +JOIN tracks t ON t.artist_id = ar.id +JOIN play_events pe ON pe.track_id = t.id +WHERE ar.mbid IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM artist_similarity asim + WHERE asim.artist_a_id = ar.id + AND asim.source = 'listenbrainz' + AND asim.fetched_at > now() - interval '7 days' + ) +ORDER BY ar.id +LIMIT $1; + +-- name: GetTracksByMBIDs :many +-- Bulk in-library lookup: maps a slice of MBIDs back to local track IDs. +-- Used by the worker to filter LB-returned MBIDs to those actually in the +-- user's library. +SELECT id, mbid FROM tracks WHERE mbid = ANY($1::text[]); + +-- name: GetArtistsByMBIDs :many +SELECT id, mbid FROM artists WHERE mbid = ANY($1::text[]); + +-- name: UpsertTrackSimilarity :exec +-- Refresh-in-place: if (a, b, source) already exists, update score and +-- fetched_at. Idempotent. +INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at) +VALUES ($1, $2, $3, 'listenbrainz', now()) +ON CONFLICT (track_a_id, track_b_id, source) +DO UPDATE SET score = EXCLUDED.score, fetched_at = EXCLUDED.fetched_at; + +-- name: UpsertArtistSimilarity :exec +INSERT INTO artist_similarity (artist_a_id, artist_b_id, score, source, fetched_at) +VALUES ($1, $2, $3, 'listenbrainz', now()) +ON CONFLICT (artist_a_id, artist_b_id, source) +DO UPDATE SET score = EXCLUDED.score, fetched_at = EXCLUDED.fetched_at; +``` + +- [ ] **Step 2: Run sqlc generate** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && make generate` + +Expected: dockerized sqlc 1.27.0 regenerates `internal/db/dbq/similarity.sql.go` with the 6 new methods. Other `*.sql.go` files may receive trivial version-comment updates from sqlc — stage them all. + +- [ ] **Step 3: Verify compile** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && go build ./...` + +Expected: clean. No callers yet — generated bindings just need to compile. + +- [ ] **Step 4: Commit** + +```bash +cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel +git add internal/db/queries/similarity.sql internal/db/dbq/ +git commit -m "feat(db): add similarity sqlc queries (list-needing, get-by-mbids, upsert)" +``` + +--- + +## Task 3: LB client `SimilarRecordings` method + +**Files:** +- Modify: `internal/scrobble/listenbrainz/client.go` (append types + method) +- Modify: `internal/scrobble/listenbrainz/client_test.go` (append tests) + +The existing `Client` already has `SubmitListens` from M4a with `BaseURL`, `HTTP`, and the typed errors `ErrAuth`/`ErrPermanent`/`ErrTransient`/`*RetryAfterError`. We add a sibling read-only method. + +LB algorithm parameter: hardcode `session_based_days_7500_session_30_contribution_5_threshold_15_limit_100_filter_True_skip_30` — this is the documented default at the time of writing. If LB changes its default at implementation time, swap to whatever's current per their `/explore/similar-recordings/` docs. + +- [ ] **Step 1: Write the failing tests** + +Append to `internal/scrobble/listenbrainz/client_test.go`: + +```go +func TestClient_SimilarRecordings_Success(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { + // Body shape from LB's API docs (simplified for tests). + _, _ = w.Write([]byte(`[ + {"recording_mbid": "11111111-1111-1111-1111-111111111111", "score": 0.95}, + {"recording_mbid": "22222222-2222-2222-2222-222222222222", "score": 0.80} + ]`)) + }) + defer srv.Close() + out, err := c.SimilarRecordings(context.Background(), "aaaa", 100) + if err != nil { + t.Fatalf("err: %v", err) + } + if len(out) != 2 { + t.Fatalf("got %d results, want 2", len(out)) + } + if out[0].MBID != "11111111-1111-1111-1111-111111111111" || out[0].Score != 0.95 { + t.Errorf("first result wrong: %+v", out[0]) + } +} + +func TestClient_SimilarRecordings_AlgorithmParamSet(t *testing.T) { + var seenURL string + c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { + seenURL = r.URL.String() + _, _ = w.Write([]byte(`[]`)) + }) + defer srv.Close() + _, _ = c.SimilarRecordings(context.Background(), "abc", 50) + if !strings.Contains(seenURL, "algorithm=") { + t.Errorf("URL missing algorithm param: %q", seenURL) + } +} + +func TestClient_SimilarRecordings_LimitParamSet(t *testing.T) { + var seenURL string + c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { + seenURL = r.URL.String() + _, _ = w.Write([]byte(`[]`)) + }) + defer srv.Close() + _, _ = c.SimilarRecordings(context.Background(), "abc", 50) + if !strings.Contains(seenURL, "count=50") && !strings.Contains(seenURL, "limit=50") { + t.Errorf("URL missing count/limit param: %q", seenURL) + } +} + +func TestClient_SimilarRecordings_401_ReturnsErrAuth(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusUnauthorized) + }) + defer srv.Close() + _, err := c.SimilarRecordings(context.Background(), "abc", 100) + if !errors.Is(err, ErrAuth) { + t.Errorf("err = %v, want ErrAuth", err) + } +} + +func TestClient_SimilarRecordings_400_ReturnsErrPermanent(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadRequest) + }) + defer srv.Close() + _, err := c.SimilarRecordings(context.Background(), "abc", 100) + if !errors.Is(err, ErrPermanent) { + t.Errorf("err = %v, want ErrPermanent", err) + } +} + +func TestClient_SimilarRecordings_503_ReturnsErrTransient(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + }) + defer srv.Close() + _, err := c.SimilarRecordings(context.Background(), "abc", 100) + if !errors.Is(err, ErrTransient) { + t.Errorf("err = %v, want ErrTransient", err) + } +} + +func TestClient_SimilarRecordings_429_ReturnsRetryAfter(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Retry-After", "120") + w.WriteHeader(http.StatusTooManyRequests) + }) + defer srv.Close() + _, err := c.SimilarRecordings(context.Background(), "abc", 100) + var ra *RetryAfterError + if !errors.As(err, &ra) { + t.Fatalf("err = %v, want *RetryAfterError", err) + } + if ra.Wait != 120*time.Second { + t.Errorf("Wait = %v, want 120s", ra.Wait) + } +} +``` + +Make sure `strings` is in the existing import block at the top of the file (it almost certainly already is from M4a — if not, add it). + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && go test ./internal/scrobble/listenbrainz/ -run SimilarRecordings -v` + +Expected: FAIL with "undefined: SimilarRecording" / "undefined: SimilarRecordings". + +- [ ] **Step 3: Implement the method** + +Append to `internal/scrobble/listenbrainz/client.go` (no new imports needed beyond what M4a already has — `bytes`, `context`, `encoding/json`, `errors`, `fmt`, `net/http`, `strconv`, `time`): + +```go +// LB algorithm parameter for /explore/similar-recordings/. Hardcoded for v1 +// — matches LB's documented default. If LB changes the default, swap here. +const lbSimilarRecordingsAlgorithm = "session_based_days_7500_session_30_contribution_5_threshold_15_limit_100_filter_True_skip_30" + +// LB algorithm parameter for /explore/similar-artists/. +const lbSimilarArtistsAlgorithm = "session_based_days_7500_session_30_contribution_5_threshold_15_limit_100_filter_True_skip_30" + +// SimilarRecording is one entry in the /explore/similar-recordings response. +type SimilarRecording struct { + MBID string `json:"recording_mbid"` + Score float64 `json:"score"` +} + +// SimilarRecordings fetches up to `limit` similar recordings for the given +// recording MBID. Public endpoint — no token required. Returns the same +// typed errors as SubmitListens. +func (c *Client) SimilarRecordings(ctx context.Context, mbid string, limit int) ([]SimilarRecording, error) { + base := c.BaseURL + if base == "" { + base = defaultBaseURL + } + url := fmt.Sprintf("%s/1/explore/similar-recordings/%s?algorithm=%s&count=%d", + base, mbid, lbSimilarRecordingsAlgorithm, limit) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("listenbrainz: build similar-recordings request: %w", err) + } + + httpClient := c.HTTP + if httpClient == nil { + httpClient = http.DefaultClient + } + resp, err := httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrTransient, err) + } + defer func() { _ = resp.Body.Close() }() + + switch { + case resp.StatusCode >= 200 && resp.StatusCode < 300: + var out []SimilarRecording + if derr := json.NewDecoder(resp.Body).Decode(&out); derr != nil { + return nil, fmt.Errorf("%w: decode similar-recordings: %v", ErrTransient, derr) + } + return out, nil + case resp.StatusCode == http.StatusUnauthorized: + return nil, ErrAuth + case resp.StatusCode == http.StatusTooManyRequests: + return nil, &RetryAfterError{Wait: parseRetryAfter(resp.Header.Get("Retry-After"))} + case resp.StatusCode >= 500: + return nil, fmt.Errorf("%w: status %d", ErrTransient, resp.StatusCode) + default: + return nil, fmt.Errorf("%w: status %d", ErrPermanent, resp.StatusCode) + } +} +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && go test ./internal/scrobble/listenbrainz/ -run SimilarRecordings -v` + +Expected: PASS for all 7 new tests. + +- [ ] **Step 5: Verify lint clean** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && golangci-lint run ./internal/scrobble/...` + +Expected: clean. + +- [ ] **Step 6: Commit** + +```bash +cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel +git add internal/scrobble/listenbrainz/client.go internal/scrobble/listenbrainz/client_test.go +git commit -m "feat(listenbrainz): add SimilarRecordings client method" +``` + +--- + +## Task 4: LB client `SimilarArtists` method + +Symmetric to Task 3. The artist endpoint has the same shape; the response key is `artist_mbid` instead of `recording_mbid`. + +**Files:** +- Modify: `internal/scrobble/listenbrainz/client.go` (append) +- Modify: `internal/scrobble/listenbrainz/client_test.go` (append) + +- [ ] **Step 1: Write the failing tests** + +Append to `internal/scrobble/listenbrainz/client_test.go`: + +```go +func TestClient_SimilarArtists_Success(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`[ + {"artist_mbid": "33333333-3333-3333-3333-333333333333", "score": 0.91}, + {"artist_mbid": "44444444-4444-4444-4444-444444444444", "score": 0.72} + ]`)) + }) + defer srv.Close() + out, err := c.SimilarArtists(context.Background(), "aaaa", 100) + if err != nil { + t.Fatalf("err: %v", err) + } + if len(out) != 2 { + t.Fatalf("got %d, want 2", len(out)) + } + if out[0].MBID != "33333333-3333-3333-3333-333333333333" || out[0].Score != 0.91 { + t.Errorf("first result wrong: %+v", out[0]) + } +} + +func TestClient_SimilarArtists_AlgorithmParamSet(t *testing.T) { + var seenURL string + c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { + seenURL = r.URL.String() + _, _ = w.Write([]byte(`[]`)) + }) + defer srv.Close() + _, _ = c.SimilarArtists(context.Background(), "abc", 50) + if !strings.Contains(seenURL, "algorithm=") { + t.Errorf("URL missing algorithm param: %q", seenURL) + } +} + +func TestClient_SimilarArtists_LimitParamSet(t *testing.T) { + var seenURL string + c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { + seenURL = r.URL.String() + _, _ = w.Write([]byte(`[]`)) + }) + defer srv.Close() + _, _ = c.SimilarArtists(context.Background(), "abc", 50) + if !strings.Contains(seenURL, "count=50") && !strings.Contains(seenURL, "limit=50") { + t.Errorf("URL missing count/limit param: %q", seenURL) + } +} + +func TestClient_SimilarArtists_401_ReturnsErrAuth(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusUnauthorized) + }) + defer srv.Close() + _, err := c.SimilarArtists(context.Background(), "abc", 100) + if !errors.Is(err, ErrAuth) { + t.Errorf("err = %v, want ErrAuth", err) + } +} + +func TestClient_SimilarArtists_400_ReturnsErrPermanent(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadRequest) + }) + defer srv.Close() + _, err := c.SimilarArtists(context.Background(), "abc", 100) + if !errors.Is(err, ErrPermanent) { + t.Errorf("err = %v, want ErrPermanent", err) + } +} + +func TestClient_SimilarArtists_503_ReturnsErrTransient(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + }) + defer srv.Close() + _, err := c.SimilarArtists(context.Background(), "abc", 100) + if !errors.Is(err, ErrTransient) { + t.Errorf("err = %v, want ErrTransient", err) + } +} + +func TestClient_SimilarArtists_429_ReturnsRetryAfter(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Retry-After", "60") + w.WriteHeader(http.StatusTooManyRequests) + }) + defer srv.Close() + _, err := c.SimilarArtists(context.Background(), "abc", 100) + var ra *RetryAfterError + if !errors.As(err, &ra) { + t.Fatalf("err = %v, want *RetryAfterError", err) + } + if ra.Wait != 60*time.Second { + t.Errorf("Wait = %v, want 60s", ra.Wait) + } +} +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && go test ./internal/scrobble/listenbrainz/ -run SimilarArtists -v` + +Expected: FAIL with "undefined: SimilarArtists". + +- [ ] **Step 3: Implement the method** + +Append to `internal/scrobble/listenbrainz/client.go`: + +```go +// SimilarArtist is one entry in the /explore/similar-artists response. +type SimilarArtist struct { + MBID string `json:"artist_mbid"` + Score float64 `json:"score"` +} + +// SimilarArtists fetches up to `limit` similar artists for the given +// artist MBID. Public endpoint — no token required. Returns the same +// typed errors as SubmitListens. +func (c *Client) SimilarArtists(ctx context.Context, mbid string, limit int) ([]SimilarArtist, error) { + base := c.BaseURL + if base == "" { + base = defaultBaseURL + } + url := fmt.Sprintf("%s/1/explore/similar-artists/%s?algorithm=%s&count=%d", + base, mbid, lbSimilarArtistsAlgorithm, limit) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("listenbrainz: build similar-artists request: %w", err) + } + + httpClient := c.HTTP + if httpClient == nil { + httpClient = http.DefaultClient + } + resp, err := httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrTransient, err) + } + defer func() { _ = resp.Body.Close() }() + + switch { + case resp.StatusCode >= 200 && resp.StatusCode < 300: + var out []SimilarArtist + if derr := json.NewDecoder(resp.Body).Decode(&out); derr != nil { + return nil, fmt.Errorf("%w: decode similar-artists: %v", ErrTransient, derr) + } + return out, nil + case resp.StatusCode == http.StatusUnauthorized: + return nil, ErrAuth + case resp.StatusCode == http.StatusTooManyRequests: + return nil, &RetryAfterError{Wait: parseRetryAfter(resp.Header.Get("Retry-After"))} + case resp.StatusCode >= 500: + return nil, fmt.Errorf("%w: status %d", ErrTransient, resp.StatusCode) + default: + return nil, fmt.Errorf("%w: status %d", ErrPermanent, resp.StatusCode) + } +} +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && go test ./internal/scrobble/listenbrainz/ -run SimilarArtists -v` + +Expected: PASS for all 7 new tests. + +- [ ] **Step 5: Verify lint clean** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && golangci-lint run ./internal/scrobble/...` + +Expected: clean. + +- [ ] **Step 6: Commit** + +```bash +cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel +git add internal/scrobble/listenbrainz/client.go internal/scrobble/listenbrainz/client_test.go +git commit -m "feat(listenbrainz): add SimilarArtists client method" +``` + +--- + +## Task 5: Worker skeleton + pure unit tests + +The Worker package gets skeleton types + `NewWorker` constructor first. Integration tests in Task 6 exercise the full pipeline. + +**Files:** +- Create: `internal/similarity/worker.go` +- Create: `internal/similarity/worker_test.go` + +- [ ] **Step 1: Write the failing tests** + +Create `internal/similarity/worker_test.go`: + +```go +package similarity + +import ( + "io" + "log/slog" + "testing" + "time" +) + +func TestNewWorker_DefaultsMatchSpec(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + w := NewWorker(nil, nil, logger) + if w.tick != 1*time.Hour { + t.Errorf("tick = %v, want 1h", w.tick) + } + if w.batch != 5 { + t.Errorf("batch = %d, want 5", w.batch) + } + if w.topK != 20 { + t.Errorf("topK = %d, want 20", w.topK) + } +} +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && go test ./internal/similarity/ -v` + +Expected: FAIL — package doesn't exist yet. + +- [ ] **Step 3: Write the worker skeleton** + +Create `internal/similarity/worker.go`: + +```go +// Package similarity owns the inbound ListenBrainz similarity ingest +// pipeline. A periodic worker queries LB's /explore/similar-recordings +// and /explore/similar-artists endpoints for tracks the user has played, +// filters returned MBIDs to the local library, and stores the top-K +// edges in track_similarity / artist_similarity for M4c's radio +// candidate-pool builder. +package similarity + +import ( + "context" + "log/slog" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz" +) + +// Worker drains played-tracks-and-artists needing similarity and POSTs +// the results into track_similarity / artist_similarity. Failures are +// passively retried via the timer (no durable queue table — losing one +// tick's worth of refresh attempts is "1 hour of staleness," fine). +type Worker struct { + pool *pgxpool.Pool + client *listenbrainz.Client + logger *slog.Logger + tick time.Duration + batch int32 + topK int +} + +// NewWorker constructs a worker with production defaults: 1h tick, +// batch=5, topK=20. +func NewWorker(pool *pgxpool.Pool, client *listenbrainz.Client, logger *slog.Logger) *Worker { + return &Worker{ + pool: pool, + client: client, + logger: logger, + tick: 1 * time.Hour, + batch: 5, + topK: 20, + } +} + +// Run blocks until ctx is cancelled, ticking every w.tick. +func (w *Worker) Run(ctx context.Context) { + t := time.NewTicker(w.tick) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + if err := w.tickOnce(ctx); err != nil { + w.logger.Error("similarity: tick failed", "err", err) + } + } + } +} + +// tickOnce drains one batch of tracks and one batch of artists. Stub — +// implementation lands in Task 6 along with the integration tests that +// drive it. +func (w *Worker) tickOnce(ctx context.Context) error { + return nil +} +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && go test ./internal/similarity/ -v` + +Expected: PASS for the constructor test. + +- [ ] **Step 5: Verify lint clean** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && golangci-lint run ./internal/similarity/...` + +Expected: clean. + +- [ ] **Step 6: Commit** + +```bash +cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel +git add internal/similarity/worker.go internal/similarity/worker_test.go +git commit -m "feat(similarity): add Worker skeleton with constructor + Run loop" +``` + +--- + +## Task 6: Worker tickOnce — full pipeline (integration) + +Full implementation of the worker drain path with comprehensive integration tests. + +**Files:** +- Modify: `internal/similarity/worker.go` (replace `tickOnce` stub with real impl) +- Create: `internal/similarity/worker_integration_test.go` + +The test file follows the M4a fixture pattern: a test pool helper that runs migrations, truncates known tables, and returns a `*pgxpool.Pool` + `*dbq.Queries`. Since the M4a `internal/scrobble` package already has a similar helper in `queue_test.go`, copy that shape. + +- [ ] **Step 1: Write the failing integration tests** + +Create `internal/similarity/worker_integration_test.go`: + +```go +package similarity + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db" + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + "git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz" +) + +func testPool(t *testing.T) (*pgxpool.Pool, *dbq.Queries) { + t.Helper() + if testing.Short() { + t.Skip("skipping in -short mode") + } + dsn := os.Getenv("MINSTREL_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("MINSTREL_TEST_DATABASE_URL not set") + } + if err := db.Migrate(dsn, slog.New(slog.NewTextHandler(io.Discard, nil))); err != nil { + t.Fatalf("migrate: %v", err) + } + pool, err := pgxpool.New(context.Background(), dsn) + if err != nil { + t.Fatalf("pool: %v", err) + } + t.Cleanup(pool.Close) + if _, err := pool.Exec(context.Background(), + "TRUNCATE artist_similarity, track_similarity, scrobble_queue, play_events, skip_events, play_sessions, sessions, users, tracks, albums, artists RESTART IDENTITY CASCADE"); err != nil { + t.Fatalf("truncate: %v", err) + } + return pool, dbq.New(pool) +} + +type fixture struct { + pool *pgxpool.Pool + q *dbq.Queries + user pgtype.UUID + artist dbq.Artist + album dbq.Album +} + +// newFixture creates a user, one artist (with MBID), and one album. +// Tests add tracks via seedTrack which optionally takes an MBID. +func newFixture(t *testing.T) fixture { + t.Helper() + pool, q := testPool(t) + ctx := context.Background() + u, err := q.CreateUser(ctx, dbq.CreateUserParams{ + Username: "tester", PasswordHash: "x", ApiToken: "x", IsAdmin: false, + }) + if err != nil { + t.Fatalf("user: %v", err) + } + artistMbid := "aaaaaaaa-1111-1111-1111-111111111111" + a, err := q.UpsertArtist(ctx, dbq.UpsertArtistParams{ + Name: "X", SortName: "X", Mbid: &artistMbid, + }) + if err != nil { + t.Fatalf("artist: %v", err) + } + al, err := q.UpsertAlbum(ctx, dbq.UpsertAlbumParams{ + Title: "X", SortTitle: "X", ArtistID: a.ID, + }) + if err != nil { + t.Fatalf("album: %v", err) + } + return fixture{pool: pool, q: q, user: u.ID, artist: a, album: al} +} + +func seedTrack(t *testing.T, f fixture, title string, mbid *string) dbq.Track { + t.Helper() + tr, err := f.q.UpsertTrack(context.Background(), dbq.UpsertTrackParams{ + Title: title, + AlbumID: f.album.ID, + ArtistID: f.artist.ID, + FilePath: "/tmp/" + title + ".flac", + DurationMs: 200_000, + Mbid: mbid, + }) + if err != nil { + t.Fatalf("track: %v", err) + } + return tr +} + +// markPlayed inserts a closed play_event so the track qualifies as "played" +// for the worker's selection query. +func markPlayed(t *testing.T, f fixture, trackID pgtype.UUID) { + t.Helper() + var sessionID pgtype.UUID + if err := f.pool.QueryRow(context.Background(), + `INSERT INTO play_sessions (user_id, started_at, last_event_at, client_id) + VALUES ($1, now() - interval '5 minutes', now(), 'test') RETURNING id`, + f.user).Scan(&sessionID); err != nil { + t.Fatalf("session: %v", err) + } + if _, err := f.pool.Exec(context.Background(), + `INSERT INTO play_events (user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped) + VALUES ($1, $2, $3, now() - interval '1 minute', now(), 250000, 0.83, false)`, + f.user, trackID, sessionID); err != nil { + t.Fatalf("play_event: %v", err) + } +} + +func newTestWorker(f fixture, lbBaseURL string) *Worker { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + return &Worker{ + pool: f.pool, + client: &listenbrainz.Client{BaseURL: lbBaseURL, HTTP: http.DefaultClient}, + logger: logger, + tick: 1 * time.Hour, + batch: 5, + topK: 20, + } +} + +// stubLB returns an httptest server that responds to similar-recordings and +// similar-artists with the given JSON payloads. Use empty string to simulate +// "endpoint returned []" (LB knows the MBID but has no neighbors). +func stubLB(recordingsBody, artistsBody string, status int) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if status != 0 && status != http.StatusOK { + w.WriteHeader(status) + return + } + if strings.Contains(r.URL.Path, "/similar-recordings/") { + _, _ = w.Write([]byte(recordingsBody)) + return + } + if strings.Contains(r.URL.Path, "/similar-artists/") { + _, _ = w.Write([]byte(artistsBody)) + return + } + w.WriteHeader(http.StatusNotFound) + })) +} + +func countTrackSim(t *testing.T, f fixture, a pgtype.UUID) int { + t.Helper() + var n int + if err := f.pool.QueryRow(context.Background(), + `SELECT count(*) FROM track_similarity WHERE track_a_id = $1`, a).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + return n +} + +func countArtistSim(t *testing.T, f fixture, a pgtype.UUID) int { + t.Helper() + var n int + if err := f.pool.QueryRow(context.Background(), + `SELECT count(*) FROM artist_similarity WHERE artist_a_id = $1`, a).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + return n +} + +func TestTickOnce_NoPlayedTracks_NoOp(t *testing.T) { + f := newFixture(t) + srv := stubLB(`[]`, `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + // no rows inserted anywhere + var n int + _ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM track_similarity`).Scan(&n) + if n != 0 { + t.Errorf("track_similarity rows = %d, want 0", n) + } + _ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM artist_similarity`).Scan(&n) + if n != 0 { + t.Errorf("artist_similarity rows = %d, want 0", n) + } +} + +func TestTickOnce_MapsLBResponseToLocalLibrary(t *testing.T) { + f := newFixture(t) + mbidA := "11111111-1111-1111-1111-111111111111" + mbidB := "22222222-2222-2222-2222-222222222222" + mbidC := "99999999-9999-9999-9999-999999999999" // NOT in library + trackA := seedTrack(t, f, "A", &mbidA) + trackB := seedTrack(t, f, "B", &mbidB) + markPlayed(t, f, trackA.ID) + // LB returns three results when queried for trackA — two in library, one not. + body := `[ + {"recording_mbid": "` + mbidB + `", "score": 0.9}, + {"recording_mbid": "` + mbidC + `", "score": 0.7} + ]` + _ = trackB + srv := stubLB(body, `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + if got := countTrackSim(t, f, trackA.ID); got != 1 { + t.Errorf("track_similarity rows = %d, want 1 (only mbidB is in-library)", got) + } +} + +func TestTickOnce_TopKEnforced(t *testing.T) { + f := newFixture(t) + mbidSeed := "11111111-1111-1111-1111-111111111111" + seed := seedTrack(t, f, "Seed", &mbidSeed) + markPlayed(t, f, seed.ID) + // Build 25 tracks with MBIDs in library, plus the LB response listing all 25. + type lbRow struct { + MBID string `json:"recording_mbid"` + Score float64 `json:"score"` + } + rows := make([]lbRow, 0, 25) + for i := 0; i < 25; i++ { + mbid := "20000000-0000-0000-0000-" + leftPad(i+1, 12) + _ = seedTrack(t, f, "T"+leftPad(i+1, 2), &mbid) + rows = append(rows, lbRow{MBID: mbid, Score: 1.0 - 0.01*float64(i)}) + } + body, _ := json.Marshal(rows) + srv := stubLB(string(body), `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + if got := countTrackSim(t, f, seed.ID); got > 20 { + t.Errorf("top-K not enforced: got %d, want ≤ 20", got) + } + if got := countTrackSim(t, f, seed.ID); got != 20 { + t.Errorf("got %d rows, want exactly 20 (all 25 in-library; top 20 by score)", got) + } +} + +func TestTickOnce_RespectsSevenDayCap(t *testing.T) { + f := newFixture(t) + mbid := "11111111-1111-1111-1111-111111111111" + trackA := seedTrack(t, f, "A", &mbid) + markPlayed(t, f, trackA.ID) + // Manually insert a fresh row so the worker should skip this track. + otherMbid := "55555555-5555-5555-5555-555555555555" + other := seedTrack(t, f, "Other", &otherMbid) + if _, err := f.pool.Exec(context.Background(), + `INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at) + VALUES ($1, $2, 0.5, 'listenbrainz', now())`, trackA.ID, other.ID); err != nil { + t.Fatalf("seed sim: %v", err) + } + // LB stub would return more rows if called; if the worker hits LB the + // test fails (different count after). + beforeCount := countTrackSim(t, f, trackA.ID) + srv := stubLB(`[{"recording_mbid":"`+otherMbid+`","score":0.99}]`, `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + _ = w.tickOnce(context.Background()) + if got := countTrackSim(t, f, trackA.ID); got != beforeCount { + t.Errorf("worker re-queried fresh track: before=%d after=%d", beforeCount, got) + } +} + +func TestTickOnce_RefreshesStaleRow(t *testing.T) { + f := newFixture(t) + mbid := "11111111-1111-1111-1111-111111111111" + trackA := seedTrack(t, f, "A", &mbid) + markPlayed(t, f, trackA.ID) + otherMbid := "55555555-5555-5555-5555-555555555555" + other := seedTrack(t, f, "Other", &otherMbid) + // Insert a stale row from 8 days ago with score 0.5. + if _, err := f.pool.Exec(context.Background(), + `INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at) + VALUES ($1, $2, 0.5, 'listenbrainz', now() - interval '8 days')`, trackA.ID, other.ID); err != nil { + t.Fatalf("seed sim: %v", err) + } + srv := stubLB(`[{"recording_mbid":"`+otherMbid+`","score":0.95}]`, `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + var score float64 + var fetchedAt time.Time + _ = f.pool.QueryRow(context.Background(), + `SELECT score, fetched_at FROM track_similarity WHERE track_a_id = $1 AND track_b_id = $2 AND source = 'listenbrainz'`, + trackA.ID, other.ID).Scan(&score, &fetchedAt) + if score != 0.95 { + t.Errorf("score = %v, want 0.95 (refreshed)", score) + } + if time.Since(fetchedAt) > time.Minute { + t.Errorf("fetched_at not bumped: %v ago", time.Since(fetchedAt)) + } +} + +func TestTickOnce_429AbortsTick(t *testing.T) { + f := newFixture(t) + mbid := "11111111-1111-1111-1111-111111111111" + trackA := seedTrack(t, f, "A", &mbid) + markPlayed(t, f, trackA.ID) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Retry-After", "60") + w.WriteHeader(http.StatusTooManyRequests) + })) + defer srv.Close() + w := newTestWorker(f, srv.URL) + _ = w.tickOnce(context.Background()) + // no rows updated; fetched_at should not be set anywhere + var n int + _ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM track_similarity WHERE track_a_id = $1`, trackA.ID).Scan(&n) + if n != 0 { + t.Errorf("rows on 429 = %d, want 0 (tick aborted before any inserts)", n) + } +} + +func TestTickOnce_TransientErrorSkipsTrack(t *testing.T) { + f := newFixture(t) + mbidA := "11111111-1111-1111-1111-111111111111" + mbidB := "22222222-2222-2222-2222-222222222222" + otherMbid := "55555555-5555-5555-5555-555555555555" + trackA := seedTrack(t, f, "A", &mbidA) + trackB := seedTrack(t, f, "B", &mbidB) + other := seedTrack(t, f, "Other", &otherMbid) + markPlayed(t, f, trackA.ID) + markPlayed(t, f, trackB.ID) + // Stub: 503 for the first MBID we see, success body for the second. + var seen atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.Contains(r.URL.Path, "/similar-recordings/") { + _, _ = w.Write([]byte(`[]`)) + return + } + n := seen.Add(1) + if n == 1 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + _, _ = w.Write([]byte(`[{"recording_mbid":"` + otherMbid + `","score":0.7}]`)) + })) + defer srv.Close() + w := newTestWorker(f, srv.URL) + _ = w.tickOnce(context.Background()) + _ = other // referenced via UPSERT + // Exactly one of trackA/trackB should have a row; the other (whichever + // hit 503) should be empty. + a := countTrackSim(t, f, trackA.ID) + b := countTrackSim(t, f, trackB.ID) + if a+b != 1 { + t.Errorf("expected exactly one track to succeed: a=%d b=%d", a, b) + } +} + +func TestTickOnce_FiltersInLibrary(t *testing.T) { + f := newFixture(t) + mbid := "11111111-1111-1111-1111-111111111111" + trackA := seedTrack(t, f, "A", &mbid) + markPlayed(t, f, trackA.ID) + // LB returns 3 MBIDs none of which are in our library. + body := `[ + {"recording_mbid":"99999999-9999-9999-9999-999999999991","score":0.9}, + {"recording_mbid":"99999999-9999-9999-9999-999999999992","score":0.8}, + {"recording_mbid":"99999999-9999-9999-9999-999999999993","score":0.7} + ]` + srv := stubLB(body, `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + if got := countTrackSim(t, f, trackA.ID); got != 0 { + t.Errorf("filtered out-of-library: got %d, want 0", got) + } +} + +func TestTickOnce_ArtistPassMirrors(t *testing.T) { + f := newFixture(t) + mbid := "11111111-1111-1111-1111-111111111111" + trackA := seedTrack(t, f, "A", &mbid) + markPlayed(t, f, trackA.ID) + // Seed a SECOND artist with MBID; LB returns it as similar to f.artist. + bMbid := "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb" + otherArtist, err := f.q.UpsertArtist(context.Background(), dbq.UpsertArtistParams{ + Name: "Other", SortName: "Other", Mbid: &bMbid, + }) + if err != nil { + t.Fatalf("artist: %v", err) + } + body := `[{"artist_mbid":"` + bMbid + `","score":0.85}]` + srv := stubLB(`[]`, body, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + _ = otherArtist + if got := countArtistSim(t, f, f.artist.ID); got != 1 { + t.Errorf("artist_similarity rows = %d, want 1", got) + } +} + +func TestTickOnce_NoMBIDOnTrack_Skipped(t *testing.T) { + f := newFixture(t) + trackNoMbid := seedTrack(t, f, "NoMbid", nil) + markPlayed(t, f, trackNoMbid.ID) + // LB stub would return rows if queried; if worker queries it, the test + // fails (any non-zero rows means we hit LB for a no-MBID track). + srv := stubLB(`[{"recording_mbid":"11111111-1111-1111-1111-111111111111","score":0.9}]`, `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + if got := countTrackSim(t, f, trackNoMbid.ID); got != 0 { + t.Errorf("no-MBID track produced rows: %d", got) + } +} + +// leftPad pads an integer with leading zeros to width n, used to build +// deterministic MBID-like strings in tests. +func leftPad(v, width int) string { + s := "" + for i := 0; i < width; i++ { + s = "0" + s + } + digits := "" + for v > 0 { + digits = string(rune('0'+v%10)) + digits + v /= 10 + } + if len(digits) >= width { + return digits + } + return s[:width-len(digits)] + digits +} +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' go test ./internal/similarity/ -v` + +Expected: tests run but most fail because `tickOnce` is a stub returning nil. + +- [ ] **Step 3: Implement tickOnce** + +Replace the `tickOnce` stub in `internal/similarity/worker.go` with the full implementation. Add the imports `errors`, `git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq`, `github.com/jackc/pgx/v5/pgtype` to the import block: + +```go +import ( + "context" + "errors" + "log/slog" + "sort" + "time" + + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + "git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz" +) +``` + +Replace `tickOnce` with: + +```go +// tickOnce drains one batch of tracks and one batch of artists. Per-row +// errors are logged and skipped (passive retry via timer). 429 aborts the +// entire tick (the next hourly tick is far longer than any LB Retry-After). +// +// Returns the first error that aborts the whole tick (currently only 429); +// nil otherwise. +func (w *Worker) tickOnce(ctx context.Context) error { + q := dbq.New(w.pool) + if err := w.tickTracks(ctx, q); err != nil { + return err + } + return w.tickArtists(ctx, q) +} + +func (w *Worker) tickTracks(ctx context.Context, q *dbq.Queries) error { + rows, err := q.ListPlayedTracksNeedingSimilarity(ctx, w.batch) + if err != nil { + return err + } + for _, r := range rows { + if r.Mbid == nil { + continue // defensive — query already filters NULL + } + results, err := w.client.SimilarRecordings(ctx, *r.Mbid, 100) + if err != nil { + var ra *listenbrainz.RetryAfterError + if errors.As(err, &ra) { + w.logger.Warn("similarity: 429 — aborting tick", "retry_after", ra.Wait) + return nil // abort this tick; passive retry on next + } + w.logger.Warn("similarity: similar-recordings failed", "track_id", r.ID, "err", err) + continue + } + w.upsertTrackSimilar(ctx, q, r.ID, results) + } + return nil +} + +func (w *Worker) tickArtists(ctx context.Context, q *dbq.Queries) error { + rows, err := q.ListPlayedArtistsNeedingSimilarity(ctx, w.batch) + if err != nil { + return err + } + for _, r := range rows { + if r.Mbid == nil { + continue + } + results, err := w.client.SimilarArtists(ctx, *r.Mbid, 100) + if err != nil { + var ra *listenbrainz.RetryAfterError + if errors.As(err, &ra) { + w.logger.Warn("similarity: 429 — aborting tick", "retry_after", ra.Wait) + return nil + } + w.logger.Warn("similarity: similar-artists failed", "artist_id", r.ID, "err", err) + continue + } + w.upsertArtistSimilar(ctx, q, r.ID, results) + } + return nil +} + +// upsertTrackSimilar filters returned MBIDs to those in our library, takes +// top-K by score, and upserts rows. +func (w *Worker) upsertTrackSimilar(ctx context.Context, q *dbq.Queries, trackAID pgtype.UUID, results []listenbrainz.SimilarRecording) { + if len(results) == 0 { + return + } + // Sort by score desc (LB returns sorted; defensive). + sort.Slice(results, func(i, j int) bool { return results[i].Score > results[j].Score }) + + // Bulk-resolve MBIDs to local track IDs. + mbids := make([]string, 0, len(results)) + for _, r := range results { + mbids = append(mbids, r.MBID) + } + rows, err := q.GetTracksByMBIDs(ctx, mbids) + if err != nil { + w.logger.Warn("similarity: GetTracksByMBIDs", "err", err) + return + } + idByMBID := make(map[string]pgtype.UUID, len(rows)) + for _, r := range rows { + if r.Mbid != nil { + idByMBID[*r.Mbid] = r.ID + } + } + + // Walk results in score order, take first K that map to a local track. + taken := 0 + for _, r := range results { + if taken >= w.topK { + break + } + localID, ok := idByMBID[r.MBID] + if !ok { + continue + } + if localID == trackAID { + continue // defensive — DB CHECK constraint also catches self-edges + } + if uerr := q.UpsertTrackSimilarity(ctx, dbq.UpsertTrackSimilarityParams{ + TrackAID: trackAID, TrackBID: localID, Score: r.Score, + }); uerr != nil { + w.logger.Warn("similarity: UpsertTrackSimilarity", "err", uerr) + continue + } + taken++ + } +} + +func (w *Worker) upsertArtistSimilar(ctx context.Context, q *dbq.Queries, artistAID pgtype.UUID, results []listenbrainz.SimilarArtist) { + if len(results) == 0 { + return + } + sort.Slice(results, func(i, j int) bool { return results[i].Score > results[j].Score }) + + mbids := make([]string, 0, len(results)) + for _, r := range results { + mbids = append(mbids, r.MBID) + } + rows, err := q.GetArtistsByMBIDs(ctx, mbids) + if err != nil { + w.logger.Warn("similarity: GetArtistsByMBIDs", "err", err) + return + } + idByMBID := make(map[string]pgtype.UUID, len(rows)) + for _, r := range rows { + if r.Mbid != nil { + idByMBID[*r.Mbid] = r.ID + } + } + + taken := 0 + for _, r := range results { + if taken >= w.topK { + break + } + localID, ok := idByMBID[r.MBID] + if !ok { + continue + } + if localID == artistAID { + continue + } + if uerr := q.UpsertArtistSimilarity(ctx, dbq.UpsertArtistSimilarityParams{ + ArtistAID: artistAID, ArtistBID: localID, Score: r.Score, + }); uerr != nil { + w.logger.Warn("similarity: UpsertArtistSimilarity", "err", uerr) + continue + } + taken++ + } +} +``` + +Note: the exact param-struct field names (`TrackAID`, `ArtistAID`, etc.) are emitted by sqlc — they may use different capitalization. After running `make generate`, inspect `internal/db/dbq/similarity.sql.go` and adjust the field references to match. Same for `r.Mbid` — sqlc may name the column type pointer field `Mbid` or `MBID`. Use whatever the generated code shows. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' go test ./internal/similarity/ -v` + +Expected: PASS for all 9 integration tests + 1 constructor test. + +- [ ] **Step 5: Verify lint clean** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && golangci-lint run ./internal/similarity/...` + +Expected: clean. + +- [ ] **Step 6: Commit** + +```bash +cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel +git add internal/similarity/worker.go internal/similarity/worker_integration_test.go +git commit -m "feat(similarity): implement Worker.tickOnce with track + artist passes" +``` + +--- + +## Task 7: Boot the worker in main.go + +**Files:** +- Modify: `cmd/minstrel/main.go` + +- [ ] **Step 1: Add the import** + +In `cmd/minstrel/main.go`, append to the existing import block: + +```go +"git.fabledsword.com/bvandeusen/minstrel/internal/similarity" +``` + +(The `internal/scrobble/listenbrainz` import already exists from M4a; we re-use the same client.) + +- [ ] **Step 2: Add the worker startup** + +In `cmd/minstrel/main.go`, find where the M4a scrobble worker is started (a `scrobbleWorker := scrobble.NewWorker(...)` followed by `go scrobbleWorker.Run(ctx)`). Append immediately after: + +```go +// Start the similarity ingest worker. Per spec §M4b, runs every 1h, drains +// up to 5 played tracks + 5 played artists per tick, weekly re-fetch cap +// per row. Public LB endpoints — no token required. +similarityWorker := similarity.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "similarity")) +go similarityWorker.Run(ctx) +``` + +- [ ] **Step 3: Verify compile** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && go build ./...` + +Expected: clean. + +- [ ] **Step 4: Verify the worker actually starts (smoke test)** + +Run a brief end-to-end check (re-uses the M4a smoke pattern): + +```bash +cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel +SMARTMUSIC_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' \ + SMARTMUSIC_LIBRARY_SCAN_PATHS='' \ + SMARTMUSIC_LIBRARY_SCAN_ON_STARTUP=false \ + SMARTMUSIC_SERVER_ADDRESS=:14533 \ + timeout 5 go run ./cmd/minstrel/ 2>&1 | head -20 +``` + +Expected: server starts, no panic. The similarity worker doesn't log per tick (only on errors), so the smoke is just "no crash." + +- [ ] **Step 5: Verify lint clean** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && golangci-lint run ./...` + +Expected: clean. + +- [ ] **Step 6: Commit** + +```bash +cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel +git add cmd/minstrel/main.go +git commit -m "feat(cmd): start similarity worker alongside HTTP server" +``` + +--- + +## Task 8: Final verification + branch finish + +**Files:** none (verification only) + +- [ ] **Step 1: Full Go test suite (-short -race)** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && go test -short -race ./...` + +Expected: PASS across all 16 packages (now including `internal/similarity`). + +- [ ] **Step 2: Full integration suite** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' go test -race -p 1 ./...` + +Expected: PASS apart from the pre-existing `TestScanner_Integration` flake (CI runs `-short` which skips it; not introduced by this PR). + +- [ ] **Step 3: Lint** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && golangci-lint run ./...` + +Expected: clean. + +- [ ] **Step 4: Coverage check on new package** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' go test -p 1 -coverprofile=/tmp/cover-m4b.out ./internal/similarity/... ./internal/scrobble/listenbrainz/... && go tool cover -func=/tmp/cover-m4b.out | tail -5` + +Expected: `internal/similarity` ≥ 75%, `internal/scrobble/listenbrainz` ≥ 85%. The new `Similar*` methods are added on top of M4a's coverage. + +- [ ] **Step 5: Web verification (no changes — sanity)** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel/web && npm run check && npm test -- --run && npm run build` + +Expected: svelte-check 0/0; 180 vitest tests pass; build succeeds. (M4b doesn't touch web.) + +- [ ] **Step 6: Docker smoke** + +Run: `cd /home/bvandeusen/Nextcloud/Projects/Minstrel/minstrel && docker build -t minstrel:m4b-smoke .` + +Expected: container builds. + +- [ ] **Step 7: Manual verification post-merge (optional, deferred)** + +Note: full validation requires a real LB-tagged library plus 1h+ uptime. The integration tests provide deterministic coverage. Manual verification at the user's discretion: + +1. `docker compose up --build -d minstrel` — pick up new code +2. Wait ≥1 hour +3. `psql -c "SELECT count(*) FROM track_similarity WHERE source = 'listenbrainz'"` → non-zero rows IF library has MBID-tagged played tracks + +- [ ] **Step 8: Update Fable task #346** + +After PR opens, mark `in_progress`. After PR merges, mark `done` with a summary of what shipped. + +- [ ] **Step 9: Finishing the branch** + +**REQUIRED SUB-SKILL:** Use `superpowers:finishing-a-development-branch` to verify tests, present completion options, and execute the user's choice. + +Per established cadence: this slice will land as a single-purpose PR. After merge, M4b closes; M4c (radio similarity-driven candidate pool + queue refresh, closes M4) is next. + +--- + +## Self-Review + +**Spec coverage:** +- §3.1 (new package `internal/similarity`) → Tasks 5, 6 ✓ +- §3.2 (LB client extensions) → Tasks 3, 4; boot wiring → Task 7 ✓ +- §3.3 (failure handling — passive retry, 429 abort tick) → Task 6 (worker logic + integration tests) ✓ +- §4 (database schema) → Task 1 ✓ +- §5 (sqlc queries) → Task 2 ✓ +- §6 (worker algorithm) → Task 6 ✓ +- §7.1 (LB client unit tests) → Tasks 3, 4 ✓ +- §7.2 (worker integration tests) → Task 6 ✓ +- §7.3 (coverage target) → Task 8 step 4 ✓ +- §7.4 (manual verification) → Task 8 step 7 ✓ +- §8 (backwards compat) → preserved by additive design (new tables, new package) ✓ + +No gaps. + +**Placeholder scan:** No "TBD"/"TODO" content. The "use whatever sqlc emits at make generate time" guidance in Task 6 is a known sqlc-specific instruction, not a placeholder. + +**Type consistency:** +- `Worker` struct fields (`pool`, `client`, `logger`, `tick`, `batch`, `topK`) — same in Task 5 (skeleton) and Task 6 (full impl). +- `tickOnce(ctx) error` — same signature throughout. +- `SimilarRecording`/`SimilarArtist` types defined in Tasks 3/4, consumed in Task 6. +- sqlc-generated method names referenced consistently: `ListPlayedTracksNeedingSimilarity`, `ListPlayedArtistsNeedingSimilarity`, `GetTracksByMBIDs`, `GetArtistsByMBIDs`, `UpsertTrackSimilarity`, `UpsertArtistSimilarity`. +- Constants `lbSimilarRecordingsAlgorithm` / `lbSimilarArtistsAlgorithm` defined in Task 3, both referenced from Tasks 3 and 4. From fc2d35d33c2baf9275b085f8efaeadc6752b47a0 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 28 Apr 2026 19:57:02 -0400 Subject: [PATCH 3/9] feat(db): add migration 0009 for similarity (track_similarity + artist_similarity) --- .../db/migrations/0009_similarity.down.sql | 2 ++ internal/db/migrations/0009_similarity.up.sql | 33 +++++++++++++++++++ 2 files changed, 35 insertions(+) create mode 100644 internal/db/migrations/0009_similarity.down.sql create mode 100644 internal/db/migrations/0009_similarity.up.sql diff --git a/internal/db/migrations/0009_similarity.down.sql b/internal/db/migrations/0009_similarity.down.sql new file mode 100644 index 00000000..b6718a77 --- /dev/null +++ b/internal/db/migrations/0009_similarity.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS artist_similarity; +DROP TABLE IF EXISTS track_similarity; diff --git a/internal/db/migrations/0009_similarity.up.sql b/internal/db/migrations/0009_similarity.up.sql new file mode 100644 index 00000000..8110a2bd --- /dev/null +++ b/internal/db/migrations/0009_similarity.up.sql @@ -0,0 +1,33 @@ +-- M4b: ListenBrainz similarity ingest. Two symmetric tables — tracks and +-- artists — both keyed by (a_id, b_id, source) so the schema reserves room +-- for additional similarity sources (musicbrainz_tag, user_cooccurrence) +-- alongside listenbrainz. M4b only writes 'listenbrainz' rows. +-- +-- The (a_id, score DESC) index satisfies M4c's hot path: "for seed S, give +-- me top-N similar tracks/artists descending by score." + +CREATE TABLE track_similarity ( + track_a_id uuid NOT NULL REFERENCES tracks(id) ON DELETE CASCADE, + track_b_id uuid NOT NULL REFERENCES tracks(id) ON DELETE CASCADE, + score DOUBLE PRECISION NOT NULL, + source TEXT NOT NULL CHECK (source IN ('listenbrainz', 'musicbrainz_tag', 'user_cooccurrence')), + fetched_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (track_a_id, track_b_id, source), + CHECK (track_a_id <> track_b_id) +); + +CREATE INDEX track_similarity_a_score_idx + ON track_similarity (track_a_id, score DESC); + +CREATE TABLE artist_similarity ( + artist_a_id uuid NOT NULL REFERENCES artists(id) ON DELETE CASCADE, + artist_b_id uuid NOT NULL REFERENCES artists(id) ON DELETE CASCADE, + score DOUBLE PRECISION NOT NULL, + source TEXT NOT NULL CHECK (source IN ('listenbrainz', 'musicbrainz_tag', 'user_cooccurrence')), + fetched_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (artist_a_id, artist_b_id, source), + CHECK (artist_a_id <> artist_b_id) +); + +CREATE INDEX artist_similarity_a_score_idx + ON artist_similarity (artist_a_id, score DESC); From d4e0e5fae31db1751bcc34ba0aab538f6055a978 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 28 Apr 2026 19:58:05 -0400 Subject: [PATCH 4/9] feat(db): add similarity sqlc queries (list-needing, get-by-mbids, upsert) --- internal/db/dbq/models.go | 16 +++ internal/db/dbq/similarity.sql.go | 191 +++++++++++++++++++++++++++++ internal/db/queries/similarity.sql | 50 ++++++++ 3 files changed, 257 insertions(+) create mode 100644 internal/db/dbq/similarity.sql.go create mode 100644 internal/db/queries/similarity.sql diff --git a/internal/db/dbq/models.go b/internal/db/dbq/models.go index e7b7e229..5ce88a32 100644 --- a/internal/db/dbq/models.go +++ b/internal/db/dbq/models.go @@ -29,6 +29,14 @@ type Artist struct { UpdatedAt pgtype.Timestamptz } +type ArtistSimilarity struct { + ArtistAID pgtype.UUID + ArtistBID pgtype.UUID + Score float64 + Source string + FetchedAt pgtype.Timestamptz +} + type ContextualLike struct { ID pgtype.UUID UserID pgtype.UUID @@ -129,6 +137,14 @@ type Track struct { UpdatedAt pgtype.Timestamptz } +type TrackSimilarity struct { + TrackAID pgtype.UUID + TrackBID pgtype.UUID + Score float64 + Source string + FetchedAt pgtype.Timestamptz +} + type User struct { ID pgtype.UUID Username string diff --git a/internal/db/dbq/similarity.sql.go b/internal/db/dbq/similarity.sql.go new file mode 100644 index 00000000..5a92cecf --- /dev/null +++ b/internal/db/dbq/similarity.sql.go @@ -0,0 +1,191 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.27.0 +// source: similarity.sql + +package dbq + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const getArtistsByMBIDs = `-- name: GetArtistsByMBIDs :many +SELECT id, mbid FROM artists WHERE mbid = ANY($1::text[]) +` + +type GetArtistsByMBIDsRow struct { + ID pgtype.UUID + Mbid *string +} + +func (q *Queries) GetArtistsByMBIDs(ctx context.Context, dollar_1 []string) ([]GetArtistsByMBIDsRow, error) { + rows, err := q.db.Query(ctx, getArtistsByMBIDs, dollar_1) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetArtistsByMBIDsRow + for rows.Next() { + var i GetArtistsByMBIDsRow + if err := rows.Scan(&i.ID, &i.Mbid); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getTracksByMBIDs = `-- name: GetTracksByMBIDs :many +SELECT id, mbid FROM tracks WHERE mbid = ANY($1::text[]) +` + +type GetTracksByMBIDsRow struct { + ID pgtype.UUID + Mbid *string +} + +// Bulk in-library lookup: maps a slice of MBIDs back to local track IDs. +func (q *Queries) GetTracksByMBIDs(ctx context.Context, dollar_1 []string) ([]GetTracksByMBIDsRow, error) { + rows, err := q.db.Query(ctx, getTracksByMBIDs, dollar_1) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetTracksByMBIDsRow + for rows.Next() { + var i GetTracksByMBIDsRow + if err := rows.Scan(&i.ID, &i.Mbid); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listPlayedArtistsNeedingSimilarity = `-- name: ListPlayedArtistsNeedingSimilarity :many +SELECT DISTINCT ar.id, ar.mbid +FROM artists ar +JOIN tracks t ON t.artist_id = ar.id +JOIN play_events pe ON pe.track_id = t.id +WHERE ar.mbid IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM artist_similarity asim + WHERE asim.artist_a_id = ar.id + AND asim.source = 'listenbrainz' + AND asim.fetched_at > now() - interval '7 days' + ) +ORDER BY ar.id +LIMIT $1 +` + +type ListPlayedArtistsNeedingSimilarityRow struct { + ID pgtype.UUID + Mbid *string +} + +func (q *Queries) ListPlayedArtistsNeedingSimilarity(ctx context.Context, limit int32) ([]ListPlayedArtistsNeedingSimilarityRow, error) { + rows, err := q.db.Query(ctx, listPlayedArtistsNeedingSimilarity, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListPlayedArtistsNeedingSimilarityRow + for rows.Next() { + var i ListPlayedArtistsNeedingSimilarityRow + if err := rows.Scan(&i.ID, &i.Mbid); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listPlayedTracksNeedingSimilarity = `-- name: ListPlayedTracksNeedingSimilarity :many +SELECT DISTINCT t.id, t.mbid +FROM tracks t +JOIN play_events pe ON pe.track_id = t.id +WHERE t.mbid IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM track_similarity ts + WHERE ts.track_a_id = t.id + AND ts.source = 'listenbrainz' + AND ts.fetched_at > now() - interval '7 days' + ) +ORDER BY t.id +LIMIT $1 +` + +type ListPlayedTracksNeedingSimilarityRow struct { + ID pgtype.UUID + Mbid *string +} + +// Tracks with at least one play, an MBID, and no fresh listenbrainz row +// (no row at all, OR fetched_at older than 7 days). Used by the worker +// to find work each tick. Bounded by $1. +func (q *Queries) ListPlayedTracksNeedingSimilarity(ctx context.Context, limit int32) ([]ListPlayedTracksNeedingSimilarityRow, error) { + rows, err := q.db.Query(ctx, listPlayedTracksNeedingSimilarity, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListPlayedTracksNeedingSimilarityRow + for rows.Next() { + var i ListPlayedTracksNeedingSimilarityRow + if err := rows.Scan(&i.ID, &i.Mbid); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const upsertArtistSimilarity = `-- name: UpsertArtistSimilarity :exec +INSERT INTO artist_similarity (artist_a_id, artist_b_id, score, source, fetched_at) +VALUES ($1, $2, $3, 'listenbrainz', now()) +ON CONFLICT (artist_a_id, artist_b_id, source) +DO UPDATE SET score = EXCLUDED.score, fetched_at = EXCLUDED.fetched_at +` + +type UpsertArtistSimilarityParams struct { + ArtistAID pgtype.UUID + ArtistBID pgtype.UUID + Score float64 +} + +func (q *Queries) UpsertArtistSimilarity(ctx context.Context, arg UpsertArtistSimilarityParams) error { + _, err := q.db.Exec(ctx, upsertArtistSimilarity, arg.ArtistAID, arg.ArtistBID, arg.Score) + return err +} + +const upsertTrackSimilarity = `-- name: UpsertTrackSimilarity :exec +INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at) +VALUES ($1, $2, $3, 'listenbrainz', now()) +ON CONFLICT (track_a_id, track_b_id, source) +DO UPDATE SET score = EXCLUDED.score, fetched_at = EXCLUDED.fetched_at +` + +type UpsertTrackSimilarityParams struct { + TrackAID pgtype.UUID + TrackBID pgtype.UUID + Score float64 +} + +func (q *Queries) UpsertTrackSimilarity(ctx context.Context, arg UpsertTrackSimilarityParams) error { + _, err := q.db.Exec(ctx, upsertTrackSimilarity, arg.TrackAID, arg.TrackBID, arg.Score) + return err +} diff --git a/internal/db/queries/similarity.sql b/internal/db/queries/similarity.sql new file mode 100644 index 00000000..15a76b58 --- /dev/null +++ b/internal/db/queries/similarity.sql @@ -0,0 +1,50 @@ +-- name: ListPlayedTracksNeedingSimilarity :many +-- Tracks with at least one play, an MBID, and no fresh listenbrainz row +-- (no row at all, OR fetched_at older than 7 days). Used by the worker +-- to find work each tick. Bounded by $1. +SELECT DISTINCT t.id, t.mbid +FROM tracks t +JOIN play_events pe ON pe.track_id = t.id +WHERE t.mbid IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM track_similarity ts + WHERE ts.track_a_id = t.id + AND ts.source = 'listenbrainz' + AND ts.fetched_at > now() - interval '7 days' + ) +ORDER BY t.id +LIMIT $1; + +-- name: ListPlayedArtistsNeedingSimilarity :many +SELECT DISTINCT ar.id, ar.mbid +FROM artists ar +JOIN tracks t ON t.artist_id = ar.id +JOIN play_events pe ON pe.track_id = t.id +WHERE ar.mbid IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM artist_similarity asim + WHERE asim.artist_a_id = ar.id + AND asim.source = 'listenbrainz' + AND asim.fetched_at > now() - interval '7 days' + ) +ORDER BY ar.id +LIMIT $1; + +-- name: GetTracksByMBIDs :many +-- Bulk in-library lookup: maps a slice of MBIDs back to local track IDs. +SELECT id, mbid FROM tracks WHERE mbid = ANY($1::text[]); + +-- name: GetArtistsByMBIDs :many +SELECT id, mbid FROM artists WHERE mbid = ANY($1::text[]); + +-- name: UpsertTrackSimilarity :exec +INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at) +VALUES ($1, $2, $3, 'listenbrainz', now()) +ON CONFLICT (track_a_id, track_b_id, source) +DO UPDATE SET score = EXCLUDED.score, fetched_at = EXCLUDED.fetched_at; + +-- name: UpsertArtistSimilarity :exec +INSERT INTO artist_similarity (artist_a_id, artist_b_id, score, source, fetched_at) +VALUES ($1, $2, $3, 'listenbrainz', now()) +ON CONFLICT (artist_a_id, artist_b_id, source) +DO UPDATE SET score = EXCLUDED.score, fetched_at = EXCLUDED.fetched_at; From 23d55a868bc78686cae9534af6b95eb381f462a8 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 28 Apr 2026 19:59:38 -0400 Subject: [PATCH 5/9] feat(listenbrainz): add SimilarRecordings client method --- internal/scrobble/listenbrainz/client.go | 53 +++++++++++ internal/scrobble/listenbrainz/client_test.go | 95 +++++++++++++++++++ 2 files changed, 148 insertions(+) diff --git a/internal/scrobble/listenbrainz/client.go b/internal/scrobble/listenbrainz/client.go index 949cae5b..e12dfce2 100644 --- a/internal/scrobble/listenbrainz/client.go +++ b/internal/scrobble/listenbrainz/client.go @@ -180,3 +180,56 @@ func buildPayload(listenType string, listens []Listen) payload { } return payload{ListenType: listenType, Payload: entries} } + +// LB algorithm parameter for /explore/similar-recordings/. Hardcoded for v1 +// — matches LB's documented default. +const lbSimilarRecordingsAlgorithm = "session_based_days_7500_session_30_contribution_5_threshold_15_limit_100_filter_True_skip_30" + +// SimilarRecording is one entry in the /explore/similar-recordings response. +type SimilarRecording struct { + MBID string `json:"recording_mbid"` + Score float64 `json:"score"` +} + +// SimilarRecordings fetches up to `limit` similar recordings for the given +// recording MBID. Public endpoint — no token required. Returns the same +// typed errors as SubmitListens. +func (c *Client) SimilarRecordings(ctx context.Context, mbid string, limit int) ([]SimilarRecording, error) { + base := c.BaseURL + if base == "" { + base = defaultBaseURL + } + url := fmt.Sprintf("%s/1/explore/similar-recordings/%s?algorithm=%s&count=%d", + base, mbid, lbSimilarRecordingsAlgorithm, limit) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("listenbrainz: build similar-recordings request: %w", err) + } + + httpClient := c.HTTP + if httpClient == nil { + httpClient = http.DefaultClient + } + resp, err := httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrTransient, err) + } + defer func() { _ = resp.Body.Close() }() + + switch { + case resp.StatusCode >= 200 && resp.StatusCode < 300: + var out []SimilarRecording + if derr := json.NewDecoder(resp.Body).Decode(&out); derr != nil { + return nil, fmt.Errorf("%w: decode similar-recordings: %v", ErrTransient, derr) + } + return out, nil + case resp.StatusCode == http.StatusUnauthorized: + return nil, ErrAuth + case resp.StatusCode == http.StatusTooManyRequests: + return nil, &RetryAfterError{Wait: parseRetryAfter(resp.Header.Get("Retry-After"))} + case resp.StatusCode >= 500: + return nil, fmt.Errorf("%w: status %d", ErrTransient, resp.StatusCode) + default: + return nil, fmt.Errorf("%w: status %d", ErrPermanent, resp.StatusCode) + } +} diff --git a/internal/scrobble/listenbrainz/client_test.go b/internal/scrobble/listenbrainz/client_test.go index 70129bbe..3be7963f 100644 --- a/internal/scrobble/listenbrainz/client_test.go +++ b/internal/scrobble/listenbrainz/client_test.go @@ -162,3 +162,98 @@ func TestClient_SubmitListens_OmitsEmptyMBIDs(t *testing.T) { t.Errorf("body should omit empty recording_mbid: %s", seenBody) } } + +func TestClient_SimilarRecordings_Success(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`[ + {"recording_mbid": "11111111-1111-1111-1111-111111111111", "score": 0.95}, + {"recording_mbid": "22222222-2222-2222-2222-222222222222", "score": 0.80} + ]`)) + }) + defer srv.Close() + out, err := c.SimilarRecordings(context.Background(), "aaaa", 100) + if err != nil { + t.Fatalf("err: %v", err) + } + if len(out) != 2 { + t.Fatalf("got %d results, want 2", len(out)) + } + if out[0].MBID != "11111111-1111-1111-1111-111111111111" || out[0].Score != 0.95 { + t.Errorf("first result wrong: %+v", out[0]) + } +} + +func TestClient_SimilarRecordings_AlgorithmParamSet(t *testing.T) { + var seenURL string + c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { + seenURL = r.URL.String() + _, _ = w.Write([]byte(`[]`)) + }) + defer srv.Close() + _, _ = c.SimilarRecordings(context.Background(), "abc", 50) + if !strings.Contains(seenURL, "algorithm=") { + t.Errorf("URL missing algorithm param: %q", seenURL) + } +} + +func TestClient_SimilarRecordings_LimitParamSet(t *testing.T) { + var seenURL string + c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { + seenURL = r.URL.String() + _, _ = w.Write([]byte(`[]`)) + }) + defer srv.Close() + _, _ = c.SimilarRecordings(context.Background(), "abc", 50) + if !strings.Contains(seenURL, "count=50") && !strings.Contains(seenURL, "limit=50") { + t.Errorf("URL missing count/limit param: %q", seenURL) + } +} + +func TestClient_SimilarRecordings_401_ReturnsErrAuth(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusUnauthorized) + }) + defer srv.Close() + _, err := c.SimilarRecordings(context.Background(), "abc", 100) + if !errors.Is(err, ErrAuth) { + t.Errorf("err = %v, want ErrAuth", err) + } +} + +func TestClient_SimilarRecordings_400_ReturnsErrPermanent(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadRequest) + }) + defer srv.Close() + _, err := c.SimilarRecordings(context.Background(), "abc", 100) + if !errors.Is(err, ErrPermanent) { + t.Errorf("err = %v, want ErrPermanent", err) + } +} + +func TestClient_SimilarRecordings_503_ReturnsErrTransient(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + }) + defer srv.Close() + _, err := c.SimilarRecordings(context.Background(), "abc", 100) + if !errors.Is(err, ErrTransient) { + t.Errorf("err = %v, want ErrTransient", err) + } +} + +func TestClient_SimilarRecordings_429_ReturnsRetryAfter(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Retry-After", "120") + w.WriteHeader(http.StatusTooManyRequests) + }) + defer srv.Close() + _, err := c.SimilarRecordings(context.Background(), "abc", 100) + var ra *RetryAfterError + if !errors.As(err, &ra) { + t.Fatalf("err = %v, want *RetryAfterError", err) + } + if ra.Wait != 120*time.Second { + t.Errorf("Wait = %v, want 120s", ra.Wait) + } +} From b0d1d6bb46cf3f15e45837c9f0115b2c4edea198 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 28 Apr 2026 20:01:17 -0400 Subject: [PATCH 6/9] feat(listenbrainz): add SimilarArtists client method --- internal/scrobble/listenbrainz/client.go | 53 +++++++++++ internal/scrobble/listenbrainz/client_test.go | 95 +++++++++++++++++++ 2 files changed, 148 insertions(+) diff --git a/internal/scrobble/listenbrainz/client.go b/internal/scrobble/listenbrainz/client.go index e12dfce2..dafac9ee 100644 --- a/internal/scrobble/listenbrainz/client.go +++ b/internal/scrobble/listenbrainz/client.go @@ -233,3 +233,56 @@ func (c *Client) SimilarRecordings(ctx context.Context, mbid string, limit int) return nil, fmt.Errorf("%w: status %d", ErrPermanent, resp.StatusCode) } } + +// LB algorithm parameter for /explore/similar-artists/. Hardcoded for v1 +// — matches LB's documented default. +const lbSimilarArtistsAlgorithm = "session_based_days_7500_session_30_contribution_5_threshold_15_limit_100_filter_True_skip_30" + +// SimilarArtist is one entry in the /explore/similar-artists response. +type SimilarArtist struct { + MBID string `json:"artist_mbid"` + Score float64 `json:"score"` +} + +// SimilarArtists fetches up to `limit` similar artists for the given +// artist MBID. Public endpoint — no token required. Returns the same +// typed errors as SubmitListens. +func (c *Client) SimilarArtists(ctx context.Context, mbid string, limit int) ([]SimilarArtist, error) { + base := c.BaseURL + if base == "" { + base = defaultBaseURL + } + url := fmt.Sprintf("%s/1/explore/similar-artists/%s?algorithm=%s&count=%d", + base, mbid, lbSimilarArtistsAlgorithm, limit) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("listenbrainz: build similar-artists request: %w", err) + } + + httpClient := c.HTTP + if httpClient == nil { + httpClient = http.DefaultClient + } + resp, err := httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrTransient, err) + } + defer func() { _ = resp.Body.Close() }() + + switch { + case resp.StatusCode >= 200 && resp.StatusCode < 300: + var out []SimilarArtist + if derr := json.NewDecoder(resp.Body).Decode(&out); derr != nil { + return nil, fmt.Errorf("%w: decode similar-artists: %v", ErrTransient, derr) + } + return out, nil + case resp.StatusCode == http.StatusUnauthorized: + return nil, ErrAuth + case resp.StatusCode == http.StatusTooManyRequests: + return nil, &RetryAfterError{Wait: parseRetryAfter(resp.Header.Get("Retry-After"))} + case resp.StatusCode >= 500: + return nil, fmt.Errorf("%w: status %d", ErrTransient, resp.StatusCode) + default: + return nil, fmt.Errorf("%w: status %d", ErrPermanent, resp.StatusCode) + } +} diff --git a/internal/scrobble/listenbrainz/client_test.go b/internal/scrobble/listenbrainz/client_test.go index 3be7963f..e04a83f2 100644 --- a/internal/scrobble/listenbrainz/client_test.go +++ b/internal/scrobble/listenbrainz/client_test.go @@ -257,3 +257,98 @@ func TestClient_SimilarRecordings_429_ReturnsRetryAfter(t *testing.T) { t.Errorf("Wait = %v, want 120s", ra.Wait) } } + +func TestClient_SimilarArtists_Success(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`[ + {"artist_mbid": "33333333-3333-3333-3333-333333333333", "score": 0.91}, + {"artist_mbid": "44444444-4444-4444-4444-444444444444", "score": 0.72} + ]`)) + }) + defer srv.Close() + out, err := c.SimilarArtists(context.Background(), "aaaa", 100) + if err != nil { + t.Fatalf("err: %v", err) + } + if len(out) != 2 { + t.Fatalf("got %d, want 2", len(out)) + } + if out[0].MBID != "33333333-3333-3333-3333-333333333333" || out[0].Score != 0.91 { + t.Errorf("first result wrong: %+v", out[0]) + } +} + +func TestClient_SimilarArtists_AlgorithmParamSet(t *testing.T) { + var seenURL string + c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { + seenURL = r.URL.String() + _, _ = w.Write([]byte(`[]`)) + }) + defer srv.Close() + _, _ = c.SimilarArtists(context.Background(), "abc", 50) + if !strings.Contains(seenURL, "algorithm=") { + t.Errorf("URL missing algorithm param: %q", seenURL) + } +} + +func TestClient_SimilarArtists_LimitParamSet(t *testing.T) { + var seenURL string + c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) { + seenURL = r.URL.String() + _, _ = w.Write([]byte(`[]`)) + }) + defer srv.Close() + _, _ = c.SimilarArtists(context.Background(), "abc", 50) + if !strings.Contains(seenURL, "count=50") && !strings.Contains(seenURL, "limit=50") { + t.Errorf("URL missing count/limit param: %q", seenURL) + } +} + +func TestClient_SimilarArtists_401_ReturnsErrAuth(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusUnauthorized) + }) + defer srv.Close() + _, err := c.SimilarArtists(context.Background(), "abc", 100) + if !errors.Is(err, ErrAuth) { + t.Errorf("err = %v, want ErrAuth", err) + } +} + +func TestClient_SimilarArtists_400_ReturnsErrPermanent(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadRequest) + }) + defer srv.Close() + _, err := c.SimilarArtists(context.Background(), "abc", 100) + if !errors.Is(err, ErrPermanent) { + t.Errorf("err = %v, want ErrPermanent", err) + } +} + +func TestClient_SimilarArtists_503_ReturnsErrTransient(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + }) + defer srv.Close() + _, err := c.SimilarArtists(context.Background(), "abc", 100) + if !errors.Is(err, ErrTransient) { + t.Errorf("err = %v, want ErrTransient", err) + } +} + +func TestClient_SimilarArtists_429_ReturnsRetryAfter(t *testing.T) { + c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Retry-After", "60") + w.WriteHeader(http.StatusTooManyRequests) + }) + defer srv.Close() + _, err := c.SimilarArtists(context.Background(), "abc", 100) + var ra *RetryAfterError + if !errors.As(err, &ra) { + t.Fatalf("err = %v, want *RetryAfterError", err) + } + if ra.Wait != 60*time.Second { + t.Errorf("Wait = %v, want 60s", ra.Wait) + } +} From 9adad094b0445bd6718c953074cab6ac170692a4 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 28 Apr 2026 20:02:15 -0400 Subject: [PATCH 7/9] feat(similarity): add Worker skeleton with constructor + Run loop --- internal/similarity/worker.go | 66 ++++++++++++++++++++++++++++++ internal/similarity/worker_test.go | 22 ++++++++++ 2 files changed, 88 insertions(+) create mode 100644 internal/similarity/worker.go create mode 100644 internal/similarity/worker_test.go diff --git a/internal/similarity/worker.go b/internal/similarity/worker.go new file mode 100644 index 00000000..9d3f1263 --- /dev/null +++ b/internal/similarity/worker.go @@ -0,0 +1,66 @@ +// Package similarity owns the inbound ListenBrainz similarity ingest +// pipeline. A periodic worker queries LB's /explore/similar-recordings +// and /explore/similar-artists endpoints for tracks the user has played, +// filters returned MBIDs to the local library, and stores the top-K +// edges in track_similarity / artist_similarity for M4c's radio +// candidate-pool builder. +package similarity + +import ( + "context" + "log/slog" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz" +) + +// Worker drains played-tracks-and-artists needing similarity and POSTs +// the results into track_similarity / artist_similarity. Failures are +// passively retried via the timer (no durable queue table — losing one +// tick's worth of refresh attempts is "1 hour of staleness," fine). +type Worker struct { + pool *pgxpool.Pool + client *listenbrainz.Client + logger *slog.Logger + tick time.Duration + batch int32 + topK int +} + +// NewWorker constructs a worker with production defaults: 1h tick, +// batch=5, topK=20. +func NewWorker(pool *pgxpool.Pool, client *listenbrainz.Client, logger *slog.Logger) *Worker { + return &Worker{ + pool: pool, + client: client, + logger: logger, + tick: 1 * time.Hour, + batch: 5, + topK: 20, + } +} + +// Run blocks until ctx is cancelled, ticking every w.tick. +func (w *Worker) Run(ctx context.Context) { + t := time.NewTicker(w.tick) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + if err := w.tickOnce(ctx); err != nil { + w.logger.Error("similarity: tick failed", "err", err) + } + } + } +} + +// tickOnce drains one batch of tracks and one batch of artists. Stub — +// implementation lands in Task 6 along with the integration tests that +// drive it. +func (w *Worker) tickOnce(_ context.Context) error { + return nil +} diff --git a/internal/similarity/worker_test.go b/internal/similarity/worker_test.go new file mode 100644 index 00000000..3dfa0994 --- /dev/null +++ b/internal/similarity/worker_test.go @@ -0,0 +1,22 @@ +package similarity + +import ( + "io" + "log/slog" + "testing" + "time" +) + +func TestNewWorker_DefaultsMatchSpec(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + w := NewWorker(nil, nil, logger) + if w.tick != 1*time.Hour { + t.Errorf("tick = %v, want 1h", w.tick) + } + if w.batch != 5 { + t.Errorf("batch = %d, want 5", w.batch) + } + if w.topK != 20 { + t.Errorf("topK = %d, want 20", w.topK) + } +} From 893adf04c931ddb0b86fc2879823999ef6fd02be Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 28 Apr 2026 20:06:47 -0400 Subject: [PATCH 8/9] feat(similarity): implement Worker.tickOnce with track + artist passes Replace stub with full tickOnce: drains played tracks/artists via ListPlayedTracksNeedingSimilarity / ListPlayedArtistsNeedingSimilarity, calls LB SimilarRecordings/SimilarArtists, filters to local library via bulk MBID lookup, enforces top-K=20 by score, and upserts similarity rows. 429 aborts the entire tick; other errors log-and-skip. Ten integration tests covering no-op, library filtering, top-K cap, 7-day freshness cap, stale refresh, 429 abort, transient skip, and artist pass. Co-Authored-By: Claude Sonnet 4.6 --- internal/similarity/worker.go | 154 ++++++- .../similarity/worker_integration_test.go | 398 ++++++++++++++++++ 2 files changed, 548 insertions(+), 4 deletions(-) create mode 100644 internal/similarity/worker_integration_test.go diff --git a/internal/similarity/worker.go b/internal/similarity/worker.go index 9d3f1263..6b017e3e 100644 --- a/internal/similarity/worker.go +++ b/internal/similarity/worker.go @@ -8,11 +8,15 @@ package similarity import ( "context" + "errors" "log/slog" + "sort" "time" + "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" "git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz" ) @@ -58,9 +62,151 @@ func (w *Worker) Run(ctx context.Context) { } } -// tickOnce drains one batch of tracks and one batch of artists. Stub — -// implementation lands in Task 6 along with the integration tests that -// drive it. -func (w *Worker) tickOnce(_ context.Context) error { +// tickOnce drains one batch of tracks and one batch of artists. Per-row +// errors are logged and skipped (passive retry via timer). 429 aborts the +// entire tick. +func (w *Worker) tickOnce(ctx context.Context) error { + q := dbq.New(w.pool) + if err := w.tickTracks(ctx, q); err != nil { + return err + } + return w.tickArtists(ctx, q) +} + +func (w *Worker) tickTracks(ctx context.Context, q *dbq.Queries) error { + rows, err := q.ListPlayedTracksNeedingSimilarity(ctx, w.batch) + if err != nil { + return err + } + for _, r := range rows { + if r.Mbid == nil { + continue // defensive — query already filters NULL + } + results, err := w.client.SimilarRecordings(ctx, *r.Mbid, 100) + if err != nil { + var ra *listenbrainz.RetryAfterError + if errors.As(err, &ra) { + w.logger.Warn("similarity: 429 — aborting tick", "retry_after", ra.Wait) + return nil + } + w.logger.Warn("similarity: similar-recordings failed", "track_id", r.ID, "err", err) + continue + } + w.upsertTrackSimilar(ctx, q, r.ID, results) + } return nil } + +func (w *Worker) tickArtists(ctx context.Context, q *dbq.Queries) error { + rows, err := q.ListPlayedArtistsNeedingSimilarity(ctx, w.batch) + if err != nil { + return err + } + for _, r := range rows { + if r.Mbid == nil { + continue + } + results, err := w.client.SimilarArtists(ctx, *r.Mbid, 100) + if err != nil { + var ra *listenbrainz.RetryAfterError + if errors.As(err, &ra) { + w.logger.Warn("similarity: 429 — aborting tick", "retry_after", ra.Wait) + return nil + } + w.logger.Warn("similarity: similar-artists failed", "artist_id", r.ID, "err", err) + continue + } + w.upsertArtistSimilar(ctx, q, r.ID, results) + } + return nil +} + +// upsertTrackSimilar filters returned MBIDs to those in our library, takes +// top-K by score, and upserts rows. +func (w *Worker) upsertTrackSimilar(ctx context.Context, q *dbq.Queries, trackAID pgtype.UUID, results []listenbrainz.SimilarRecording) { + if len(results) == 0 { + return + } + sort.Slice(results, func(i, j int) bool { return results[i].Score > results[j].Score }) + + mbids := make([]string, 0, len(results)) + for _, r := range results { + mbids = append(mbids, r.MBID) + } + rows, err := q.GetTracksByMBIDs(ctx, mbids) + if err != nil { + w.logger.Warn("similarity: GetTracksByMBIDs", "err", err) + return + } + idByMBID := make(map[string]pgtype.UUID, len(rows)) + for _, r := range rows { + if r.Mbid != nil { + idByMBID[*r.Mbid] = r.ID + } + } + + taken := 0 + for _, r := range results { + if taken >= w.topK { + break + } + localID, ok := idByMBID[r.MBID] + if !ok { + continue + } + if localID == trackAID { + continue // defensive — DB CHECK constraint also catches self-edges + } + if uerr := q.UpsertTrackSimilarity(ctx, dbq.UpsertTrackSimilarityParams{ + TrackAID: trackAID, TrackBID: localID, Score: r.Score, + }); uerr != nil { + w.logger.Warn("similarity: UpsertTrackSimilarity", "err", uerr) + continue + } + taken++ + } +} + +func (w *Worker) upsertArtistSimilar(ctx context.Context, q *dbq.Queries, artistAID pgtype.UUID, results []listenbrainz.SimilarArtist) { + if len(results) == 0 { + return + } + sort.Slice(results, func(i, j int) bool { return results[i].Score > results[j].Score }) + + mbids := make([]string, 0, len(results)) + for _, r := range results { + mbids = append(mbids, r.MBID) + } + rows, err := q.GetArtistsByMBIDs(ctx, mbids) + if err != nil { + w.logger.Warn("similarity: GetArtistsByMBIDs", "err", err) + return + } + idByMBID := make(map[string]pgtype.UUID, len(rows)) + for _, r := range rows { + if r.Mbid != nil { + idByMBID[*r.Mbid] = r.ID + } + } + + taken := 0 + for _, r := range results { + if taken >= w.topK { + break + } + localID, ok := idByMBID[r.MBID] + if !ok { + continue + } + if localID == artistAID { + continue + } + if uerr := q.UpsertArtistSimilarity(ctx, dbq.UpsertArtistSimilarityParams{ + ArtistAID: artistAID, ArtistBID: localID, Score: r.Score, + }); uerr != nil { + w.logger.Warn("similarity: UpsertArtistSimilarity", "err", uerr) + continue + } + taken++ + } +} diff --git a/internal/similarity/worker_integration_test.go b/internal/similarity/worker_integration_test.go new file mode 100644 index 00000000..137bfdb7 --- /dev/null +++ b/internal/similarity/worker_integration_test.go @@ -0,0 +1,398 @@ +package similarity + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db" + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + "git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz" +) + +func testPool(t *testing.T) (*pgxpool.Pool, *dbq.Queries) { + t.Helper() + if testing.Short() { + t.Skip("skipping in -short mode") + } + dsn := os.Getenv("MINSTREL_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("MINSTREL_TEST_DATABASE_URL not set") + } + if err := db.Migrate(dsn, slog.New(slog.NewTextHandler(io.Discard, nil))); err != nil { + t.Fatalf("migrate: %v", err) + } + pool, err := pgxpool.New(context.Background(), dsn) + if err != nil { + t.Fatalf("pool: %v", err) + } + t.Cleanup(pool.Close) + if _, err := pool.Exec(context.Background(), + "TRUNCATE artist_similarity, track_similarity, scrobble_queue, play_events, skip_events, play_sessions, sessions, users, tracks, albums, artists RESTART IDENTITY CASCADE"); err != nil { + t.Fatalf("truncate: %v", err) + } + return pool, dbq.New(pool) +} + +type fixture struct { + pool *pgxpool.Pool + q *dbq.Queries + user pgtype.UUID + artist dbq.Artist + album dbq.Album +} + +func newFixture(t *testing.T) fixture { + t.Helper() + pool, q := testPool(t) + ctx := context.Background() + u, err := q.CreateUser(ctx, dbq.CreateUserParams{ + Username: "tester", PasswordHash: "x", ApiToken: "x", IsAdmin: false, + }) + if err != nil { + t.Fatalf("user: %v", err) + } + artistMbid := "aaaaaaaa-1111-1111-1111-111111111111" + a, err := q.UpsertArtist(ctx, dbq.UpsertArtistParams{ + Name: "X", SortName: "X", Mbid: &artistMbid, + }) + if err != nil { + t.Fatalf("artist: %v", err) + } + al, err := q.UpsertAlbum(ctx, dbq.UpsertAlbumParams{ + Title: "X", SortTitle: "X", ArtistID: a.ID, + }) + if err != nil { + t.Fatalf("album: %v", err) + } + return fixture{pool: pool, q: q, user: u.ID, artist: a, album: al} +} + +func seedTrack(t *testing.T, f fixture, title string, mbid *string) dbq.Track { + t.Helper() + tr, err := f.q.UpsertTrack(context.Background(), dbq.UpsertTrackParams{ + Title: title, + AlbumID: f.album.ID, + ArtistID: f.artist.ID, + FilePath: "/tmp/" + title + ".flac", + DurationMs: 200_000, + Mbid: mbid, + }) + if err != nil { + t.Fatalf("track: %v", err) + } + return tr +} + +func markPlayed(t *testing.T, f fixture, trackID pgtype.UUID) { + t.Helper() + var sessionID pgtype.UUID + if err := f.pool.QueryRow(context.Background(), + `INSERT INTO play_sessions (user_id, started_at, last_event_at, client_id) + VALUES ($1, now() - interval '5 minutes', now(), 'test') RETURNING id`, + f.user).Scan(&sessionID); err != nil { + t.Fatalf("session: %v", err) + } + if _, err := f.pool.Exec(context.Background(), + `INSERT INTO play_events (user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped) + VALUES ($1, $2, $3, now() - interval '1 minute', now(), 250000, 0.83, false)`, + f.user, trackID, sessionID); err != nil { + t.Fatalf("play_event: %v", err) + } +} + +func newTestWorker(f fixture, lbBaseURL string) *Worker { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + return &Worker{ + pool: f.pool, + client: &listenbrainz.Client{BaseURL: lbBaseURL, HTTP: http.DefaultClient}, + logger: logger, + tick: 1 * time.Hour, + batch: 5, + topK: 20, + } +} + +// stubLB returns an httptest server that responds to similar-recordings and +// similar-artists with the given JSON payloads. +func stubLB(recordingsBody, artistsBody string, status int) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if status != 0 && status != http.StatusOK { + w.WriteHeader(status) + return + } + if strings.Contains(r.URL.Path, "/similar-recordings/") { + _, _ = w.Write([]byte(recordingsBody)) + return + } + if strings.Contains(r.URL.Path, "/similar-artists/") { + _, _ = w.Write([]byte(artistsBody)) + return + } + w.WriteHeader(http.StatusNotFound) + })) +} + +func countTrackSim(t *testing.T, f fixture, a pgtype.UUID) int { + t.Helper() + var n int + if err := f.pool.QueryRow(context.Background(), + `SELECT count(*) FROM track_similarity WHERE track_a_id = $1`, a).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + return n +} + +func countArtistSim(t *testing.T, f fixture, a pgtype.UUID) int { + t.Helper() + var n int + if err := f.pool.QueryRow(context.Background(), + `SELECT count(*) FROM artist_similarity WHERE artist_a_id = $1`, a).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + return n +} + +func TestTickOnce_NoPlayedTracks_NoOp(t *testing.T) { + f := newFixture(t) + srv := stubLB(`[]`, `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + var n int + _ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM track_similarity`).Scan(&n) + if n != 0 { + t.Errorf("track_similarity rows = %d, want 0", n) + } + _ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM artist_similarity`).Scan(&n) + if n != 0 { + t.Errorf("artist_similarity rows = %d, want 0", n) + } +} + +func TestTickOnce_MapsLBResponseToLocalLibrary(t *testing.T) { + f := newFixture(t) + mbidA := "11111111-1111-1111-1111-111111111111" + mbidB := "22222222-2222-2222-2222-222222222222" + mbidC := "99999999-9999-9999-9999-999999999999" // NOT in library + trackA := seedTrack(t, f, "A", &mbidA) + _ = seedTrack(t, f, "B", &mbidB) + markPlayed(t, f, trackA.ID) + body := `[ + {"recording_mbid": "` + mbidB + `", "score": 0.9}, + {"recording_mbid": "` + mbidC + `", "score": 0.7} + ]` + srv := stubLB(body, `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + if got := countTrackSim(t, f, trackA.ID); got != 1 { + t.Errorf("track_similarity rows = %d, want 1 (only mbidB is in-library)", got) + } +} + +func TestTickOnce_TopKEnforced(t *testing.T) { + f := newFixture(t) + mbidSeed := "11111111-1111-1111-1111-111111111111" + seed := seedTrack(t, f, "Seed", &mbidSeed) + markPlayed(t, f, seed.ID) + type lbRow struct { + MBID string `json:"recording_mbid"` + Score float64 `json:"score"` + } + rows := make([]lbRow, 0, 25) + for i := 0; i < 25; i++ { + mbid := fmt.Sprintf("20000000-0000-0000-0000-%012d", i+1) + _ = seedTrack(t, f, fmt.Sprintf("T%02d", i+1), &mbid) + rows = append(rows, lbRow{MBID: mbid, Score: 1.0 - 0.01*float64(i)}) + } + body, _ := json.Marshal(rows) + srv := stubLB(string(body), `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + if got := countTrackSim(t, f, seed.ID); got != 20 { + t.Errorf("got %d rows, want exactly 20 (all 25 in-library; top 20 by score)", got) + } +} + +func TestTickOnce_RespectsSevenDayCap(t *testing.T) { + f := newFixture(t) + mbid := "11111111-1111-1111-1111-111111111111" + trackA := seedTrack(t, f, "A", &mbid) + markPlayed(t, f, trackA.ID) + otherMbid := "55555555-5555-5555-5555-555555555555" + other := seedTrack(t, f, "Other", &otherMbid) + if _, err := f.pool.Exec(context.Background(), + `INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at) + VALUES ($1, $2, 0.5, 'listenbrainz', now())`, trackA.ID, other.ID); err != nil { + t.Fatalf("seed sim: %v", err) + } + beforeCount := countTrackSim(t, f, trackA.ID) + srv := stubLB(`[{"recording_mbid":"`+otherMbid+`","score":0.99}]`, `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + _ = w.tickOnce(context.Background()) + if got := countTrackSim(t, f, trackA.ID); got != beforeCount { + t.Errorf("worker re-queried fresh track: before=%d after=%d", beforeCount, got) + } +} + +func TestTickOnce_RefreshesStaleRow(t *testing.T) { + f := newFixture(t) + mbid := "11111111-1111-1111-1111-111111111111" + trackA := seedTrack(t, f, "A", &mbid) + markPlayed(t, f, trackA.ID) + otherMbid := "55555555-5555-5555-5555-555555555555" + other := seedTrack(t, f, "Other", &otherMbid) + if _, err := f.pool.Exec(context.Background(), + `INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at) + VALUES ($1, $2, 0.5, 'listenbrainz', now() - interval '8 days')`, trackA.ID, other.ID); err != nil { + t.Fatalf("seed sim: %v", err) + } + srv := stubLB(`[{"recording_mbid":"`+otherMbid+`","score":0.95}]`, `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + var score float64 + var fetchedAt time.Time + _ = f.pool.QueryRow(context.Background(), + `SELECT score, fetched_at FROM track_similarity WHERE track_a_id = $1 AND track_b_id = $2 AND source = 'listenbrainz'`, + trackA.ID, other.ID).Scan(&score, &fetchedAt) + if score != 0.95 { + t.Errorf("score = %v, want 0.95 (refreshed)", score) + } + if time.Since(fetchedAt) > time.Minute { + t.Errorf("fetched_at not bumped: %v ago", time.Since(fetchedAt)) + } +} + +func TestTickOnce_429AbortsTick(t *testing.T) { + f := newFixture(t) + mbid := "11111111-1111-1111-1111-111111111111" + trackA := seedTrack(t, f, "A", &mbid) + markPlayed(t, f, trackA.ID) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Retry-After", "60") + w.WriteHeader(http.StatusTooManyRequests) + })) + defer srv.Close() + w := newTestWorker(f, srv.URL) + _ = w.tickOnce(context.Background()) + if got := countTrackSim(t, f, trackA.ID); got != 0 { + t.Errorf("rows on 429 = %d, want 0 (tick aborted before any inserts)", got) + } +} + +func TestTickOnce_TransientErrorSkipsTrack(t *testing.T) { + f := newFixture(t) + mbidA := "11111111-1111-1111-1111-111111111111" + mbidB := "22222222-2222-2222-2222-222222222222" + otherMbid := "55555555-5555-5555-5555-555555555555" + trackA := seedTrack(t, f, "A", &mbidA) + trackB := seedTrack(t, f, "B", &mbidB) + _ = seedTrack(t, f, "Other", &otherMbid) + markPlayed(t, f, trackA.ID) + markPlayed(t, f, trackB.ID) + var seen atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.Contains(r.URL.Path, "/similar-recordings/") { + _, _ = w.Write([]byte(`[]`)) + return + } + n := seen.Add(1) + if n == 1 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + _, _ = w.Write([]byte(`[{"recording_mbid":"` + otherMbid + `","score":0.7}]`)) + })) + defer srv.Close() + w := newTestWorker(f, srv.URL) + _ = w.tickOnce(context.Background()) + a := countTrackSim(t, f, trackA.ID) + b := countTrackSim(t, f, trackB.ID) + if a+b != 1 { + t.Errorf("expected exactly one track to succeed: a=%d b=%d", a, b) + } +} + +func TestTickOnce_FiltersInLibrary(t *testing.T) { + f := newFixture(t) + mbid := "11111111-1111-1111-1111-111111111111" + trackA := seedTrack(t, f, "A", &mbid) + markPlayed(t, f, trackA.ID) + body := `[ + {"recording_mbid":"99999999-9999-9999-9999-999999999991","score":0.9}, + {"recording_mbid":"99999999-9999-9999-9999-999999999992","score":0.8}, + {"recording_mbid":"99999999-9999-9999-9999-999999999993","score":0.7} + ]` + srv := stubLB(body, `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + if got := countTrackSim(t, f, trackA.ID); got != 0 { + t.Errorf("filtered out-of-library: got %d, want 0", got) + } +} + +func TestTickOnce_ArtistPassMirrors(t *testing.T) { + f := newFixture(t) + mbid := "11111111-1111-1111-1111-111111111111" + trackA := seedTrack(t, f, "A", &mbid) + markPlayed(t, f, trackA.ID) + bMbid := "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb" + _, err := f.q.UpsertArtist(context.Background(), dbq.UpsertArtistParams{ + Name: "Other", SortName: "Other", Mbid: &bMbid, + }) + if err != nil { + t.Fatalf("artist: %v", err) + } + body := `[{"artist_mbid":"` + bMbid + `","score":0.85}]` + srv := stubLB(`[]`, body, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + if got := countArtistSim(t, f, f.artist.ID); got != 1 { + t.Errorf("artist_similarity rows = %d, want 1", got) + } +} + +func TestTickOnce_NoMBIDOnTrack_Skipped(t *testing.T) { + f := newFixture(t) + trackNoMbid := seedTrack(t, f, "NoMbid", nil) + markPlayed(t, f, trackNoMbid.ID) + srv := stubLB(`[{"recording_mbid":"11111111-1111-1111-1111-111111111111","score":0.9}]`, `[]`, http.StatusOK) + defer srv.Close() + w := newTestWorker(f, srv.URL) + if err := w.tickOnce(context.Background()); err != nil { + t.Fatalf("tickOnce: %v", err) + } + if got := countTrackSim(t, f, trackNoMbid.ID); got != 0 { + t.Errorf("no-MBID track produced rows: %d", got) + } +} From 358bfd94533ff42e10844a3e9d8f0ae3ac55f19b Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 28 Apr 2026 20:25:11 -0400 Subject: [PATCH 9/9] feat(cmd): start similarity worker alongside HTTP server --- cmd/minstrel/main.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/cmd/minstrel/main.go b/cmd/minstrel/main.go index a78835f3..9b000abe 100644 --- a/cmd/minstrel/main.go +++ b/cmd/minstrel/main.go @@ -19,6 +19,7 @@ import ( "git.fabledsword.com/bvandeusen/minstrel/internal/scrobble" "git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz" "git.fabledsword.com/bvandeusen/minstrel/internal/server" + "git.fabledsword.com/bvandeusen/minstrel/internal/similarity" "git.fabledsword.com/bvandeusen/minstrel/internal/subsonic" ) @@ -81,6 +82,12 @@ func run() error { scrobbleWorker := scrobble.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "scrobble")) go scrobbleWorker.Run(ctx) + // Start the similarity ingest worker. Per spec §M4b, runs every 1h, drains + // up to 5 played tracks + 5 played artists per tick, weekly re-fetch cap + // per row. Public LB endpoints — no token required. + similarityWorker := similarity.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "similarity")) + go similarityWorker.Run(ctx) + srv := server.New(logger, pool, scanner, subsonic.Config{ AllowPlaintextPassword: cfg.Subsonic.AllowPlaintextPassword, }, cfg.Events, cfg.Recommendation)