Merge pull request 'feat: M4b inbound ListenBrainz similarity ingest' (#27) from dev into main
This commit was merged in pull request #27.
This commit is contained in:
@@ -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)
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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.
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
DROP TABLE IF EXISTS artist_similarity;
|
||||
DROP TABLE IF EXISTS track_similarity;
|
||||
@@ -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);
|
||||
@@ -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;
|
||||
@@ -180,3 +180,109 @@ 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)
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -162,3 +162,193 @@ 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)
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,212 @@
|
||||
// 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"
|
||||
"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"
|
||||
)
|
||||
|
||||
// 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. 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++
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user