Merge pull request 'feat: M4a outbound ListenBrainz scrobble worker' (#26) from dev into main
This commit was merged in pull request #26.
This commit is contained in:
@@ -16,6 +16,8 @@ import (
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/library"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/logging"
|
||||
"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/subsonic"
|
||||
)
|
||||
@@ -73,6 +75,12 @@ func run() error {
|
||||
}()
|
||||
}
|
||||
|
||||
// Start the ListenBrainz scrobble worker. Per spec §M4a, runs every 30s
|
||||
// and drains up to 50 pending rows per tick. Per-user gating happens
|
||||
// inside the worker (rows from disabled users are skipped).
|
||||
scrobbleWorker := scrobble.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "scrobble"))
|
||||
go scrobbleWorker.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,403 @@
|
||||
# M4a — ListenBrainz outbound scrobble worker
|
||||
|
||||
**Status:** Spec draft, 2026-04-28
|
||||
**Tracking:** Fable #345
|
||||
**Milestone:** M4 — ListenBrainz scrobble + similarity + radio
|
||||
|
||||
## 1. Goal
|
||||
|
||||
Send the user's qualifying plays to ListenBrainz with batching and exponential
|
||||
retry. Reads from the existing M2 `play_events` stream; backed by a new
|
||||
`scrobble_queue` table that the worker drains every 30 seconds. Per-user token
|
||||
configuration via a new minimal `/settings` page in the SPA.
|
||||
|
||||
When this slice ships, a play that meets ListenBrainz's "≥240s OR ≥50%
|
||||
completion" threshold appears in the user's LB profile within ~30 seconds, and
|
||||
the system survives LB outages, bad networks, and process restarts without
|
||||
losing or duplicating scrobbles.
|
||||
|
||||
## 2. Non-goals (explicit)
|
||||
|
||||
- **`now-playing` endpoint** — real-time "user is currently listening" status.
|
||||
Requires push architecture; defer until a UI surface needs it.
|
||||
- **Listen import** — backfilling historical plays into LB. Useful for migration
|
||||
but not part of the core flow.
|
||||
- **MusicBrainz ID resolution at scan time** — payload uses MBIDs only when
|
||||
already populated in `tracks.mbid`/`albums.mbid`/`artists.mbid`. Better
|
||||
scanner-side MBID coverage is a future improvement.
|
||||
- **Encrypted-at-rest token storage** — chose plaintext for v1 (industry norm
|
||||
for self-hosted scrobble apps).
|
||||
- **Scrobble status indicator in PlayerBar** — latency-sensitive UX; bundle
|
||||
with the future now-playing push.
|
||||
- **Multi-instance queue safety** — single worker assumes single Minstrel
|
||||
process. ROW LOCK FOR UPDATE SKIP LOCKED is a future change if/when Minstrel
|
||||
goes multi-instance.
|
||||
- **CLI admin token rotation** — operators can `UPDATE users …` for
|
||||
break-glass; CLI tooling lands with M6 packaging.
|
||||
- **Proactive rate limiting** — we honor LB's `Retry-After` headers but don't
|
||||
throttle ourselves. Human-scale play rate is well below LB's limits.
|
||||
- **Multiple scrobble services** (Last.fm, Maloja, etc.) — LB-only by design.
|
||||
The `internal/scrobble/listenbrainz/` package structure leaves room for
|
||||
siblings later if demand emerges.
|
||||
|
||||
## 3. Architecture overview
|
||||
|
||||
```
|
||||
┌──────────────────────────┐
|
||||
play_event closed ──► │ scrobble.MaybeEnqueue │ (called from
|
||||
(M2 RecordPlayEnded) │ - apply LB threshold │ Writer hooks)
|
||||
│ - INSERT scrobble_queue │
|
||||
└────────────┬─────────────┘
|
||||
│
|
||||
▼
|
||||
┌────────────────────────┐
|
||||
│ scrobble_queue table │ status: pending | failed
|
||||
│ (work list, not log) │
|
||||
└────────────┬───────────┘
|
||||
│
|
||||
▼
|
||||
tick 30s ─────► ┌────────────────────────────────────┐
|
||||
(goroutine) │ scrobble.Worker │
|
||||
│ - SELECT pending, next_attempt<= now │
|
||||
│ - batch POST to LB submit-listens │
|
||||
│ - on 2xx: DELETE row + │
|
||||
│ UPDATE play_events.scrobbled_at │
|
||||
│ - on 5xx/network: increment │
|
||||
│ attempts, schedule next │
|
||||
│ - on 4xx: mark failed, log │
|
||||
│ - on 429 Retry-After: respect │
|
||||
│ header │
|
||||
└────────────────────────────────────┘
|
||||
│
|
||||
▼
|
||||
ListenBrainz https://api.listenbrainz.org
|
||||
```
|
||||
|
||||
### 3.1 New Go packages
|
||||
|
||||
- **`internal/scrobble/listenbrainz/`** — pure HTTP client. One method:
|
||||
`SubmitListens(ctx, token, listens)`. No DB, no worker plumbing. Tested
|
||||
against `httptest.Server`.
|
||||
- **`internal/scrobble/threshold.go`** — `Qualifies(durationPlayedMs,
|
||||
completionRatio) bool`. Pure function.
|
||||
- **`internal/scrobble/queue.go`** — `MaybeEnqueue(ctx, q, playEventID)`.
|
||||
Reads user config + threshold, inserts row.
|
||||
- **`internal/scrobble/worker.go`** — the goroutine. Started from
|
||||
`cmd/minstrel/main.go` alongside `playevents.Writer`.
|
||||
|
||||
### 3.2 Hooked into existing code
|
||||
|
||||
- **`playevents.Writer.RecordPlayEnded`** — after the close transaction
|
||||
commits, call `scrobble.MaybeEnqueue(ctx, q, playEvent.ID)` as a
|
||||
best-effort post-commit step. Errors are logged and swallowed (matches
|
||||
the M3 `CaptureContextualLikeIfPlaying` pattern: scrobble delivery is
|
||||
enrichment, not a precondition for the canonical play history).
|
||||
- **`playevents.Writer.RecordSyntheticCompletedPlay`** — same hook (Subsonic
|
||||
`submission=true` plays from `/rest/scrobble`).
|
||||
- **`internal/api/me.go`** — two new handlers: `handleGetListenBrainz` and
|
||||
`handlePutListenBrainz`. Wired in `Mount(...)` under `/api/me/listenbrainz`.
|
||||
- **`cmd/minstrel/main.go`** — start the worker:
|
||||
```go
|
||||
worker := scrobble.NewWorker(pool, listenbrainz.NewClient(), logger)
|
||||
go worker.Run(ctx)
|
||||
```
|
||||
|
||||
## 4. Database schema
|
||||
|
||||
New migration `0008_scrobble.up.sql`:
|
||||
|
||||
```sql
|
||||
ALTER TABLE users
|
||||
ADD COLUMN listenbrainz_token TEXT NULL,
|
||||
ADD COLUMN listenbrainz_enabled BOOLEAN NOT NULL DEFAULT FALSE;
|
||||
|
||||
CREATE TABLE scrobble_queue (
|
||||
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
play_event_id uuid NOT NULL REFERENCES play_events(id) ON DELETE CASCADE,
|
||||
status TEXT NOT NULL CHECK (status IN ('pending', 'failed')),
|
||||
attempts INTEGER NOT NULL DEFAULT 0,
|
||||
next_attempt_at timestamptz NOT NULL DEFAULT now(),
|
||||
last_error TEXT NULL,
|
||||
enqueued_at timestamptz NOT NULL DEFAULT now(),
|
||||
UNIQUE (play_event_id)
|
||||
);
|
||||
|
||||
CREATE INDEX scrobble_queue_pending_idx
|
||||
ON scrobble_queue (next_attempt_at)
|
||||
WHERE status = 'pending';
|
||||
```
|
||||
|
||||
Down migration drops the table and the user columns.
|
||||
|
||||
**Reused (no schema changes):**
|
||||
- `play_events.scrobbled_at` — already exists from M2 migration 0005,
|
||||
currently always NULL. Worker stamps it on successful submit.
|
||||
- `play_events.duration_played_ms`, `completion_ratio` — already populated
|
||||
by M2; used for the LB eligibility threshold.
|
||||
|
||||
**Lifecycle:**
|
||||
- Successful submit → `DELETE FROM scrobble_queue WHERE id = $1` AND
|
||||
`UPDATE play_events SET scrobbled_at = now() WHERE id = $1`. Canonical
|
||||
record stays in `play_events`; queue is purely a work list.
|
||||
- `UNIQUE(play_event_id)` makes `MaybeEnqueue` idempotent: re-running the
|
||||
close path on a play_event with an existing queue row is a no-op via
|
||||
`ON CONFLICT DO NOTHING`.
|
||||
- No `'sent'` status enum value — the simpler binary `pending`/`failed`
|
||||
state machine is sufficient because successful rows are deleted.
|
||||
|
||||
## 5. Backend components
|
||||
|
||||
### 5.1 ListenBrainz client (`internal/scrobble/listenbrainz/client.go`)
|
||||
|
||||
```go
|
||||
type Listen struct {
|
||||
ListenedAt int64
|
||||
Track Track
|
||||
}
|
||||
|
||||
type Track struct {
|
||||
ArtistName string
|
||||
TrackName string
|
||||
ReleaseName string
|
||||
DurationMs int
|
||||
RecordingMBID string
|
||||
ArtistMBIDs []string
|
||||
ReleaseMBID string
|
||||
}
|
||||
|
||||
type Client struct {
|
||||
BaseURL string // default https://api.listenbrainz.org
|
||||
HTTP *http.Client
|
||||
}
|
||||
|
||||
// Errors are typed so the worker can branch on response semantics.
|
||||
var (
|
||||
ErrAuth = errors.New("listenbrainz: auth rejected") // 401
|
||||
ErrPermanent = errors.New("listenbrainz: permanent error") // other 4xx
|
||||
ErrTransient = errors.New("listenbrainz: transient error") // 5xx, network
|
||||
)
|
||||
|
||||
type RetryAfterError struct{ Wait time.Duration }
|
||||
func (e *RetryAfterError) Error() string { ... }
|
||||
|
||||
func (c *Client) SubmitListens(ctx context.Context, token string, listens []Listen) error
|
||||
```
|
||||
|
||||
Sets `Authorization: Token <token>`. POSTs to `/1/submit-listens` with
|
||||
`payload_type=import` for batches >1, `single` for batches of 1 (LB convention).
|
||||
On 429, parses the `Retry-After` header and returns `*RetryAfterError`.
|
||||
|
||||
### 5.2 Threshold (`internal/scrobble/threshold.go`)
|
||||
|
||||
```go
|
||||
// Qualifies returns true if a closed play_event meets ListenBrainz's
|
||||
// recommended scrobble threshold.
|
||||
func Qualifies(durationPlayedMs int, completionRatio float64) bool {
|
||||
return durationPlayedMs >= 240_000 || completionRatio >= 0.5
|
||||
}
|
||||
```
|
||||
|
||||
### 5.3 Enqueue (`internal/scrobble/queue.go`)
|
||||
|
||||
```go
|
||||
// MaybeEnqueue inserts a scrobble_queue row for the given play_event if:
|
||||
// 1. The user has listenbrainz_enabled = true with a non-empty token.
|
||||
// 2. The play_event passes Qualifies().
|
||||
// Idempotent via UNIQUE(play_event_id) — re-runs are no-ops.
|
||||
// Returns nil even when the row is skipped (no-op is not an error).
|
||||
// Best-effort: callers (the playevents.Writer hooks) should log returned
|
||||
// errors but not propagate them, since scrobble delivery is enrichment
|
||||
// and must not block the canonical play history.
|
||||
func MaybeEnqueue(ctx context.Context, q *dbq.Queries, playEventID pgtype.UUID) error
|
||||
```
|
||||
|
||||
### 5.4 Worker (`internal/scrobble/worker.go`)
|
||||
|
||||
```go
|
||||
type Worker struct {
|
||||
pool *pgxpool.Pool
|
||||
client *listenbrainz.Client
|
||||
logger *slog.Logger
|
||||
tick time.Duration // 30s in production, injectable for tests
|
||||
now func() time.Time // injectable for tests
|
||||
batch int // up to 50 rows per tick
|
||||
}
|
||||
|
||||
func NewWorker(pool *pgxpool.Pool, client *listenbrainz.Client, logger *slog.Logger) *Worker
|
||||
|
||||
// Run blocks until ctx is cancelled.
|
||||
func (w *Worker) Run(ctx context.Context)
|
||||
|
||||
// tickOnce drains up to `batch` pending rows. Exposed for tests.
|
||||
func (w *Worker) tickOnce(ctx context.Context) error
|
||||
```
|
||||
|
||||
Backoff schedule (constants in `worker.go`):
|
||||
|
||||
```go
|
||||
var backoffSchedule = []time.Duration{
|
||||
1 * time.Minute,
|
||||
5 * time.Minute,
|
||||
30 * time.Minute,
|
||||
2 * time.Hour,
|
||||
6 * time.Hour,
|
||||
}
|
||||
const maxAttempts = 5
|
||||
|
||||
// backoffDelay maps a failure count (the value of the `attempts` column
|
||||
// AFTER the failure has been recorded) to the delay before the next
|
||||
// attempt. attempts=1 (first failure just happened) → backoffSchedule[0]
|
||||
// = 1m. attempts=5 → backoffSchedule[4] = 6h. attempts >= 6 → give up.
|
||||
// Returns (_, false) when attempts > maxAttempts.
|
||||
func backoffDelay(attempts int) (time.Duration, bool)
|
||||
```
|
||||
|
||||
Per-row outcome handling:
|
||||
- 2xx → `DELETE FROM scrobble_queue WHERE id = $1` + `UPDATE play_events SET
|
||||
scrobbled_at = now() WHERE id = $play_event_id`.
|
||||
- `*RetryAfterError` (429) → `UPDATE … SET next_attempt_at = now() + Wait`
|
||||
WITHOUT incrementing `attempts` (server told us to wait, not that we failed).
|
||||
- `ErrTransient` (5xx, network) → increment `attempts`, look up
|
||||
`backoffDelay(attempts)`. If `false`, mark `failed`. Otherwise schedule next.
|
||||
- `ErrPermanent` (4xx) → mark `failed` immediately, no retry.
|
||||
- `ErrAuth` (401) → mark `failed` immediately AND set
|
||||
`users.listenbrainz_enabled = FALSE` (defensive: bad token shouldn't keep
|
||||
retrying or feed future rows).
|
||||
|
||||
### 5.5 API endpoints (`internal/api/me.go`)
|
||||
|
||||
```
|
||||
GET /api/me/listenbrainz
|
||||
→ 200 { enabled: boolean, token_set: boolean, last_scrobbled_at: string|null }
|
||||
|
||||
PUT /api/me/listenbrainz
|
||||
body: { token?: string, enabled?: boolean }
|
||||
- token: any string sets the token. Empty string clears it AND forces enabled=false.
|
||||
- enabled: requires a non-empty stored token. 400 if attempting enabled=true with no token.
|
||||
→ 200 { enabled, token_set, last_scrobbled_at }
|
||||
```
|
||||
|
||||
The token is **write-only** — `GET` never returns the actual value, only
|
||||
`token_set: bool`. `last_scrobbled_at` is computed via `MAX(scrobbled_at)
|
||||
FROM play_events WHERE user_id = $1`.
|
||||
|
||||
## 6. Frontend (minimal `/settings`)
|
||||
|
||||
New page `web/src/routes/settings/+page.svelte` with a single "ListenBrainz"
|
||||
section. Form:
|
||||
|
||||
- **Token field**: password input with Save button when unset; masked
|
||||
"••••••••• (set)" + Clear button when set. Save calls `PUT /api/me/listenbrainz
|
||||
{ token }`. Clear calls `PUT { token: "" }`.
|
||||
- **Enabled checkbox**: "Send my plays to ListenBrainz". Disabled when no
|
||||
token. Toggle calls `PUT { enabled: !current }`.
|
||||
- **Last scrobbled at**: localized timestamp of `last_scrobbled_at` from the
|
||||
GET response.
|
||||
- **Disclaimer**: "Tokens are stored unencrypted in this server's database —
|
||||
treat as sensitive."
|
||||
|
||||
Shell nav (`web/src/lib/components/Shell.svelte`) gains `{ href: '/settings',
|
||||
label: 'Settings' }` after `/playlists`.
|
||||
|
||||
Helpers in `web/src/lib/api/client.ts` — confirm `apiPut` exists; add it if
|
||||
not (mirrors the existing `apiGet` shape).
|
||||
|
||||
## 7. Test plan
|
||||
|
||||
### 7.1 Pure unit tests
|
||||
|
||||
- **`threshold_test.go`**: boundary cases (exactly 240s, exactly 50%, just
|
||||
under both, both true).
|
||||
- **`worker_test.go`**: `backoffDelay(1) == 1m`, `backoffDelay(2) == 5m`,
|
||||
`backoffDelay(3) == 30m`, `backoffDelay(4) == 2h`, `backoffDelay(5) == 6h`,
|
||||
`backoffDelay(6)` returns `_, false`.
|
||||
- **`listenbrainz/client_test.go`** (uses `httptest.Server`):
|
||||
- 2xx → nil
|
||||
- 401 → `ErrAuth`
|
||||
- 400, 403 → `ErrPermanent`
|
||||
- 500, 503 → `ErrTransient`
|
||||
- 429 with `Retry-After: 60` → `*RetryAfterError{Wait: 60s}`
|
||||
- Network failure mid-request → `ErrTransient`
|
||||
- Body shape (JSON) and `Authorization: Token <token>` header asserted
|
||||
|
||||
### 7.2 Integration (live test DB)
|
||||
|
||||
- **`queue_test.go`**:
|
||||
- Idempotent: `MaybeEnqueue` twice for same play_event = one row
|
||||
- User with `listenbrainz_enabled=false` → no row
|
||||
- User with empty token → no row
|
||||
- Play_event below threshold → no row
|
||||
- Play_event passing threshold → row inserted (status=pending, attempts=0,
|
||||
next_attempt_at ≈ now())
|
||||
|
||||
- **`worker_integration_test.go`** (mocked LB via httptest):
|
||||
- 1 pending row + 200 → row deleted, `play_events.scrobbled_at` populated
|
||||
- 503 once → row remains, `attempts=1`, `next_attempt_at ≈ now() + 1m`
|
||||
- 503 five times → row marked `failed`, `last_error` populated
|
||||
- 401 → row marked `failed` immediately, `users.listenbrainz_enabled` set to
|
||||
`FALSE`
|
||||
- 429 with `Retry-After: 300` → row remains, `next_attempt_at ≈ now() + 5m`,
|
||||
`attempts` NOT incremented
|
||||
- Batch of 10 pending → single LB POST, 10 listens in payload
|
||||
|
||||
- **`internal/api/me_test.go`** extensions:
|
||||
- GET when no token → `{enabled:false, token_set:false, last_scrobbled_at:null}`
|
||||
- PUT `{token: "abc"}` → DB updated, GET shows `token_set:true`
|
||||
- PUT `{token: ""}` → token cleared, `enabled` forced to false
|
||||
- PUT `{enabled: true}` while no token → 400
|
||||
- `last_scrobbled_at` populated from `MAX(play_events.scrobbled_at)`
|
||||
|
||||
### 7.3 Frontend (`web/src/routes/settings/settings.test.ts`)
|
||||
|
||||
- Token-not-set state: input + Save button render
|
||||
- Token-set state: masked + Clear button render
|
||||
- Save mutation: PUT body has the typed token
|
||||
- Enabled checkbox: disabled when no token
|
||||
- Last-scrobbled-at: localized timestamp renders when present
|
||||
- Mocked `apiGet`/`apiPut`; no real network
|
||||
|
||||
### 7.4 Coverage target
|
||||
|
||||
`internal/scrobble/...` ≥ 80%. M3 combined coverage stays well above the 70%
|
||||
floor with these additions.
|
||||
|
||||
### 7.5 Manual verification post-merge
|
||||
|
||||
1. Generate a token at https://listenbrainz.org/profile/
|
||||
2. /settings → paste token → Save → toggle enabled
|
||||
3. Play a track for ≥240s OR finish it
|
||||
4. Within 30s, check listenbrainz.org/<your-username> → listen appears
|
||||
5. /settings → "Last scrobbled" timestamp populates
|
||||
|
||||
## 8. Backwards compatibility
|
||||
|
||||
- New migration; no changes to existing schema beyond two nullable columns
|
||||
on `users` and the new `scrobble_queue` table.
|
||||
- `/api/me/listenbrainz` is new; no existing endpoints change.
|
||||
- `MaybeEnqueue` is a new hook in `playevents.Writer`; if all users have
|
||||
`listenbrainz_enabled=false` (the default after migration), the hook is a
|
||||
no-op and behavior is unchanged.
|
||||
- The new `/settings` page is additive; the rest of the SPA is unchanged.
|
||||
- `playevents.Writer.RecordPlayEnded` and `RecordSyntheticCompletedPlay`
|
||||
signatures are unchanged. Only their bodies gain the enqueue call.
|
||||
|
||||
## 9. Decisions ledger
|
||||
|
||||
| # | Decision | Rationale |
|
||||
|---|---|---|
|
||||
| 1 | Per-user token (vs global) | Forward-compat to multi-user; cost is 2 nullable columns. |
|
||||
| 2 | Plaintext token storage (vs encrypted) | Industry norm for self-hosted scrobble apps; key-management is a separate scope. |
|
||||
| 3 | Threshold ≥240s OR ≥50% (separate from skip threshold) | Matches LB recommendation; skip threshold serves a different purpose. |
|
||||
| 4 | Pull-based 30s timer (vs push) | Survives restarts/outages; failure handling natural; latency invisible at single-user scale. |
|
||||
| 5 | Minimal `/settings` page in M4a (vs API-only) | User needs somewhere to put the token; scaffold for M6 to extend. |
|
||||
| 6 | Patient backoff: 1m → 5m → 30m → 2h → 6h, 5 attempts (~9h window) | Survives overnight LB outages without unbounded queue growth. |
|
||||
| 7 | Delete sent rows; keep failed rows | Canonical record in `play_events.scrobbled_at`; queue is a work list. |
|
||||
|
||||
## 10. Sub-plan progression (M4)
|
||||
|
||||
- **M4a (this) — outbound scrobble worker**.
|
||||
- M4b — inbound similarity ingest (`track_similarity` table + LB similarity
|
||||
fetcher; weekly cache).
|
||||
- M4c — radio similarity-driven candidate pool + queue-refresh-at-80%
|
||||
(closes M4).
|
||||
@@ -29,6 +29,8 @@ func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playev
|
||||
authed.Use(auth.RequireUser(pool))
|
||||
authed.Post("/auth/logout", h.handleLogout)
|
||||
authed.Get("/me", h.handleGetMe)
|
||||
authed.Get("/me/listenbrainz", h.handleGetListenBrainz)
|
||||
authed.Put("/me/listenbrainz", h.handlePutListenBrainz)
|
||||
|
||||
authed.Get("/artists", h.handleListArtists)
|
||||
authed.Get("/artists/{id}", h.handleGetArtist)
|
||||
|
||||
@@ -19,7 +19,7 @@ import (
|
||||
func truncateLibrary(t *testing.T, pool *pgxpool.Pool) {
|
||||
t.Helper()
|
||||
if _, err := pool.Exec(context.Background(),
|
||||
"TRUNCATE tracks, albums, artists RESTART IDENTITY CASCADE"); err != nil {
|
||||
"TRUNCATE tracks, albums, artists, scrobble_queue RESTART IDENTITY CASCADE"); err != nil {
|
||||
t.Fatalf("truncate: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,124 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/auth"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||
)
|
||||
|
||||
// LBStatus is the GET response body and the PUT response body. Token is
|
||||
// write-only — `token_set` is the only signal back to the client about
|
||||
// whether a token exists.
|
||||
type LBStatus struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
TokenSet bool `json:"token_set"`
|
||||
LastScrobbledAt *string `json:"last_scrobbled_at"`
|
||||
}
|
||||
|
||||
func (h *handlers) handleGetListenBrainz(w http.ResponseWriter, r *http.Request) {
|
||||
user, ok := auth.UserFromContext(r.Context())
|
||||
if !ok {
|
||||
writeErr(w, http.StatusUnauthorized, "unauthorized", "authentication required")
|
||||
return
|
||||
}
|
||||
resp, err := h.buildLBStatus(r.Context(), user.ID)
|
||||
if err != nil {
|
||||
h.logger.Error("api: get listenbrainz", "err", err)
|
||||
writeErr(w, http.StatusInternalServerError, "server_error", "lookup failed")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, resp)
|
||||
}
|
||||
|
||||
type putLBBody struct {
|
||||
Token *string `json:"token,omitempty"`
|
||||
Enabled *bool `json:"enabled,omitempty"`
|
||||
}
|
||||
|
||||
func (h *handlers) handlePutListenBrainz(w http.ResponseWriter, r *http.Request) {
|
||||
user, ok := auth.UserFromContext(r.Context())
|
||||
if !ok {
|
||||
writeErr(w, http.StatusUnauthorized, "unauthorized", "authentication required")
|
||||
return
|
||||
}
|
||||
var body putLBBody
|
||||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||||
writeErr(w, http.StatusBadRequest, "bad_request", "invalid json")
|
||||
return
|
||||
}
|
||||
|
||||
q := dbq.New(h.pool)
|
||||
if body.Token != nil {
|
||||
t := *body.Token
|
||||
var tokenPtr *string
|
||||
if t != "" {
|
||||
tokenPtr = &t
|
||||
}
|
||||
if err := q.SetListenBrainzToken(r.Context(), dbq.SetListenBrainzTokenParams{
|
||||
ID: user.ID, ListenbrainzToken: tokenPtr,
|
||||
}); err != nil {
|
||||
h.logger.Error("api: set listenbrainz token", "err", err)
|
||||
writeErr(w, http.StatusInternalServerError, "server_error", "update failed")
|
||||
return
|
||||
}
|
||||
}
|
||||
if body.Enabled != nil && *body.Enabled {
|
||||
// Enabling requires a non-empty token. Re-read state since the
|
||||
// preceding token update may have just set/cleared it.
|
||||
cfg, err := q.GetListenBrainzConfig(r.Context(), user.ID)
|
||||
if err != nil {
|
||||
h.logger.Error("api: get listenbrainz config (pre-enable)", "err", err)
|
||||
writeErr(w, http.StatusInternalServerError, "server_error", "lookup failed")
|
||||
return
|
||||
}
|
||||
if cfg.ListenbrainzToken == nil || *cfg.ListenbrainzToken == "" {
|
||||
writeErr(w, http.StatusBadRequest, "bad_request", "cannot enable without a token")
|
||||
return
|
||||
}
|
||||
}
|
||||
if body.Enabled != nil {
|
||||
if err := q.SetListenBrainzEnabled(r.Context(), dbq.SetListenBrainzEnabledParams{
|
||||
ID: user.ID, ListenbrainzEnabled: *body.Enabled,
|
||||
}); err != nil {
|
||||
h.logger.Error("api: set listenbrainz enabled", "err", err)
|
||||
writeErr(w, http.StatusInternalServerError, "server_error", "update failed")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
resp, err := h.buildLBStatus(r.Context(), user.ID)
|
||||
if err != nil {
|
||||
h.logger.Error("api: put listenbrainz refresh status", "err", err)
|
||||
writeErr(w, http.StatusInternalServerError, "server_error", "lookup failed")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, resp)
|
||||
}
|
||||
|
||||
func (h *handlers) buildLBStatus(ctx context.Context, userID pgtype.UUID) (LBStatus, error) {
|
||||
q := dbq.New(h.pool)
|
||||
cfg, err := q.GetListenBrainzConfig(ctx, userID)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return LBStatus{}, nil
|
||||
}
|
||||
return LBStatus{}, err
|
||||
}
|
||||
out := LBStatus{
|
||||
Enabled: cfg.ListenbrainzEnabled,
|
||||
TokenSet: cfg.ListenbrainzToken != nil && *cfg.ListenbrainzToken != "",
|
||||
}
|
||||
if cfg.LastScrobbledAt.Valid {
|
||||
s := cfg.LastScrobbledAt.Time.UTC().Format(time.RFC3339)
|
||||
out.LastScrobbledAt = &s
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/auth"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||
)
|
||||
|
||||
type lbStatusJSON struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
TokenSet bool `json:"token_set"`
|
||||
LastScrobbledAt *string `json:"last_scrobbled_at"`
|
||||
}
|
||||
|
||||
func callGetLB(h *handlers, user dbq.User) *httptest.ResponseRecorder {
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/me/listenbrainz", nil)
|
||||
req = req.WithContext(context.WithValue(req.Context(), auth.UserCtxKeyForTest(), user))
|
||||
w := httptest.NewRecorder()
|
||||
h.handleGetListenBrainz(w, req)
|
||||
return w
|
||||
}
|
||||
|
||||
func callPutLB(h *handlers, user dbq.User, body any) *httptest.ResponseRecorder {
|
||||
buf, _ := json.Marshal(body)
|
||||
req := httptest.NewRequest(http.MethodPut, "/api/me/listenbrainz", bytes.NewReader(buf))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req = req.WithContext(context.WithValue(req.Context(), auth.UserCtxKeyForTest(), user))
|
||||
w := httptest.NewRecorder()
|
||||
h.handlePutListenBrainz(w, req)
|
||||
return w
|
||||
}
|
||||
|
||||
func TestHandleGetListenBrainz_NoTokenInitial(t *testing.T) {
|
||||
h, pool := testHandlers(t)
|
||||
truncateLibrary(t, pool)
|
||||
u := seedUser(t, pool, "alice", "x", false)
|
||||
w := callGetLB(h, u)
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d body = %s", w.Code, w.Body.String())
|
||||
}
|
||||
var resp lbStatusJSON
|
||||
_ = json.Unmarshal(w.Body.Bytes(), &resp)
|
||||
if resp.Enabled || resp.TokenSet || resp.LastScrobbledAt != nil {
|
||||
t.Errorf("initial state wrong: %+v", resp)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlePutListenBrainz_SetTokenThenGet(t *testing.T) {
|
||||
h, pool := testHandlers(t)
|
||||
truncateLibrary(t, pool)
|
||||
u := seedUser(t, pool, "alice", "x", false)
|
||||
|
||||
put := callPutLB(h, u, map[string]any{"token": "abc"})
|
||||
if put.Code != http.StatusOK {
|
||||
t.Fatalf("PUT status = %d body = %s", put.Code, put.Body.String())
|
||||
}
|
||||
|
||||
get := callGetLB(h, u)
|
||||
var resp lbStatusJSON
|
||||
_ = json.Unmarshal(get.Body.Bytes(), &resp)
|
||||
if !resp.TokenSet {
|
||||
t.Error("TokenSet = false after PUT, want true")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlePutListenBrainz_ClearTokenForcesEnabledFalse(t *testing.T) {
|
||||
h, pool := testHandlers(t)
|
||||
truncateLibrary(t, pool)
|
||||
u := seedUser(t, pool, "alice", "x", false)
|
||||
|
||||
_ = callPutLB(h, u, map[string]any{"token": "abc"})
|
||||
_ = callPutLB(h, u, map[string]any{"enabled": true})
|
||||
clear := callPutLB(h, u, map[string]any{"token": ""})
|
||||
if clear.Code != http.StatusOK {
|
||||
t.Fatalf("clear status = %d body = %s", clear.Code, clear.Body.String())
|
||||
}
|
||||
|
||||
get := callGetLB(h, u)
|
||||
var resp lbStatusJSON
|
||||
_ = json.Unmarshal(get.Body.Bytes(), &resp)
|
||||
if resp.TokenSet {
|
||||
t.Error("TokenSet = true after clearing")
|
||||
}
|
||||
if resp.Enabled {
|
||||
t.Error("Enabled = true after clearing token, want false (force-cleared)")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlePutListenBrainz_EnableWithoutToken_400(t *testing.T) {
|
||||
h, pool := testHandlers(t)
|
||||
truncateLibrary(t, pool)
|
||||
u := seedUser(t, pool, "alice", "x", false)
|
||||
resp := callPutLB(h, u, map[string]any{"enabled": true})
|
||||
if resp.Code != http.StatusBadRequest {
|
||||
t.Errorf("status = %d, want 400 (cannot enable without token)", resp.Code)
|
||||
}
|
||||
}
|
||||
@@ -82,6 +82,17 @@ type PlaySession struct {
|
||||
ClientID *string
|
||||
}
|
||||
|
||||
type ScrobbleQueue struct {
|
||||
ID pgtype.UUID
|
||||
UserID pgtype.UUID
|
||||
PlayEventID pgtype.UUID
|
||||
Status string
|
||||
Attempts int32
|
||||
NextAttemptAt pgtype.Timestamptz
|
||||
LastError *string
|
||||
EnqueuedAt pgtype.Timestamptz
|
||||
}
|
||||
|
||||
type Session struct {
|
||||
ID pgtype.UUID
|
||||
UserID pgtype.UUID
|
||||
@@ -119,11 +130,13 @@ type Track struct {
|
||||
}
|
||||
|
||||
type User struct {
|
||||
ID pgtype.UUID
|
||||
Username string
|
||||
PasswordHash string
|
||||
ApiToken string
|
||||
IsAdmin bool
|
||||
CreatedAt pgtype.Timestamptz
|
||||
SubsonicPassword *string
|
||||
ID pgtype.UUID
|
||||
Username string
|
||||
PasswordHash string
|
||||
ApiToken string
|
||||
IsAdmin bool
|
||||
CreatedAt pgtype.Timestamptz
|
||||
SubsonicPassword *string
|
||||
ListenbrainzToken *string
|
||||
ListenbrainzEnabled bool
|
||||
}
|
||||
|
||||
@@ -0,0 +1,193 @@
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.27.0
|
||||
// source: scrobble.sql
|
||||
|
||||
package dbq
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
const enqueueScrobble = `-- name: EnqueueScrobble :execrows
|
||||
INSERT INTO scrobble_queue (user_id, play_event_id, status)
|
||||
VALUES ($1, $2, 'pending')
|
||||
ON CONFLICT (play_event_id) DO NOTHING
|
||||
`
|
||||
|
||||
type EnqueueScrobbleParams struct {
|
||||
UserID pgtype.UUID
|
||||
PlayEventID pgtype.UUID
|
||||
}
|
||||
|
||||
// Idempotent: if a row exists for this play_event_id, ON CONFLICT does nothing
|
||||
// and the affected-rows count is 0. Caller (MaybeEnqueue) doesn't treat 0 as
|
||||
// an error.
|
||||
func (q *Queries) EnqueueScrobble(ctx context.Context, arg EnqueueScrobbleParams) (int64, error) {
|
||||
result, err := q.db.Exec(ctx, enqueueScrobble, arg.UserID, arg.PlayEventID)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return result.RowsAffected(), nil
|
||||
}
|
||||
|
||||
const listPendingScrobbles = `-- name: ListPendingScrobbles :many
|
||||
SELECT
|
||||
sq.id AS queue_id,
|
||||
sq.user_id AS user_id,
|
||||
sq.play_event_id AS play_event_id,
|
||||
sq.attempts AS attempts,
|
||||
u.listenbrainz_token AS lb_token,
|
||||
u.listenbrainz_enabled AS lb_enabled,
|
||||
pe.started_at AS started_at,
|
||||
t.title AS track_title,
|
||||
t.duration_ms AS track_duration_ms,
|
||||
t.mbid AS track_mbid,
|
||||
al.title AS album_title,
|
||||
al.mbid AS album_mbid,
|
||||
ar.name AS artist_name,
|
||||
ar.mbid AS artist_mbid
|
||||
FROM scrobble_queue sq
|
||||
JOIN users u ON u.id = sq.user_id
|
||||
JOIN play_events pe ON pe.id = sq.play_event_id
|
||||
JOIN tracks t ON t.id = pe.track_id
|
||||
JOIN albums al ON al.id = t.album_id
|
||||
JOIN artists ar ON ar.id = t.artist_id
|
||||
WHERE sq.status = 'pending'
|
||||
AND sq.next_attempt_at <= now()
|
||||
ORDER BY sq.next_attempt_at
|
||||
LIMIT $1
|
||||
`
|
||||
|
||||
type ListPendingScrobblesRow struct {
|
||||
QueueID pgtype.UUID
|
||||
UserID pgtype.UUID
|
||||
PlayEventID pgtype.UUID
|
||||
Attempts int32
|
||||
LbToken *string
|
||||
LbEnabled bool
|
||||
StartedAt pgtype.Timestamptz
|
||||
TrackTitle string
|
||||
TrackDurationMs int32
|
||||
TrackMbid *string
|
||||
AlbumTitle string
|
||||
AlbumMbid *string
|
||||
ArtistName string
|
||||
ArtistMbid *string
|
||||
}
|
||||
|
||||
// Worker pulls up to N pending rows whose next_attempt_at has passed.
|
||||
// Joins play_events + tracks/albums/artists + users so the worker can
|
||||
// assemble the LB Listen payload in one round-trip.
|
||||
func (q *Queries) ListPendingScrobbles(ctx context.Context, limit int32) ([]ListPendingScrobblesRow, error) {
|
||||
rows, err := q.db.Query(ctx, listPendingScrobbles, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var items []ListPendingScrobblesRow
|
||||
for rows.Next() {
|
||||
var i ListPendingScrobblesRow
|
||||
if err := rows.Scan(
|
||||
&i.QueueID,
|
||||
&i.UserID,
|
||||
&i.PlayEventID,
|
||||
&i.Attempts,
|
||||
&i.LbToken,
|
||||
&i.LbEnabled,
|
||||
&i.StartedAt,
|
||||
&i.TrackTitle,
|
||||
&i.TrackDurationMs,
|
||||
&i.TrackMbid,
|
||||
&i.AlbumTitle,
|
||||
&i.AlbumMbid,
|
||||
&i.ArtistName,
|
||||
&i.ArtistMbid,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const markScrobbleFailed = `-- name: MarkScrobbleFailed :exec
|
||||
UPDATE scrobble_queue
|
||||
SET status = 'failed',
|
||||
last_error = $2
|
||||
WHERE id = $1
|
||||
`
|
||||
|
||||
type MarkScrobbleFailedParams struct {
|
||||
ID pgtype.UUID
|
||||
LastError *string
|
||||
}
|
||||
|
||||
// After a permanent failure (4xx, exhausted retries, auth) — mark failed
|
||||
// and keep the row for diagnostics.
|
||||
func (q *Queries) MarkScrobbleFailed(ctx context.Context, arg MarkScrobbleFailedParams) error {
|
||||
_, err := q.db.Exec(ctx, markScrobbleFailed, arg.ID, arg.LastError)
|
||||
return err
|
||||
}
|
||||
|
||||
const markScrobbleRetryAfter = `-- name: MarkScrobbleRetryAfter :exec
|
||||
UPDATE scrobble_queue
|
||||
SET next_attempt_at = $2
|
||||
WHERE id = $1
|
||||
`
|
||||
|
||||
type MarkScrobbleRetryAfterParams struct {
|
||||
ID pgtype.UUID
|
||||
NextAttemptAt pgtype.Timestamptz
|
||||
}
|
||||
|
||||
// After a 429: schedule next attempt per LB's Retry-After header, but do NOT
|
||||
// increment attempts (the server told us to wait, not that we failed).
|
||||
func (q *Queries) MarkScrobbleRetryAfter(ctx context.Context, arg MarkScrobbleRetryAfterParams) error {
|
||||
_, err := q.db.Exec(ctx, markScrobbleRetryAfter, arg.ID, arg.NextAttemptAt)
|
||||
return err
|
||||
}
|
||||
|
||||
const markScrobbleSent = `-- name: MarkScrobbleSent :exec
|
||||
WITH del AS (
|
||||
DELETE FROM scrobble_queue
|
||||
WHERE scrobble_queue.id = $1
|
||||
RETURNING play_event_id
|
||||
)
|
||||
UPDATE play_events
|
||||
SET scrobbled_at = now()
|
||||
WHERE play_events.id = (SELECT play_event_id FROM del)
|
||||
`
|
||||
|
||||
// Successful submit. Delete the queue row AND stamp play_events.scrobbled_at
|
||||
// in one statement via a CTE so partial failure is impossible.
|
||||
func (q *Queries) MarkScrobbleSent(ctx context.Context, id pgtype.UUID) error {
|
||||
_, err := q.db.Exec(ctx, markScrobbleSent, id)
|
||||
return err
|
||||
}
|
||||
|
||||
const rescheduleScrobble = `-- name: RescheduleScrobble :exec
|
||||
UPDATE scrobble_queue
|
||||
SET attempts = attempts + 1,
|
||||
next_attempt_at = $2,
|
||||
last_error = $3
|
||||
WHERE id = $1
|
||||
`
|
||||
|
||||
type RescheduleScrobbleParams struct {
|
||||
ID pgtype.UUID
|
||||
NextAttemptAt pgtype.Timestamptz
|
||||
LastError *string
|
||||
}
|
||||
|
||||
// After a transient failure: increment attempts, schedule next attempt,
|
||||
// record the error.
|
||||
func (q *Queries) RescheduleScrobble(ctx context.Context, arg RescheduleScrobbleParams) error {
|
||||
_, err := q.db.Exec(ctx, rescheduleScrobble, arg.ID, arg.NextAttemptAt, arg.LastError)
|
||||
return err
|
||||
}
|
||||
@@ -25,7 +25,7 @@ func (q *Queries) CountUsers(ctx context.Context) (int64, error) {
|
||||
const createUser = `-- name: CreateUser :one
|
||||
INSERT INTO users (username, password_hash, api_token, is_admin)
|
||||
VALUES ($1, $2, $3, $4)
|
||||
RETURNING id, username, password_hash, api_token, is_admin, created_at, subsonic_password
|
||||
RETURNING id, username, password_hash, api_token, is_admin, created_at, subsonic_password, listenbrainz_token, listenbrainz_enabled
|
||||
`
|
||||
|
||||
type CreateUserParams struct {
|
||||
@@ -51,12 +51,40 @@ func (q *Queries) CreateUser(ctx context.Context, arg CreateUserParams) (User, e
|
||||
&i.IsAdmin,
|
||||
&i.CreatedAt,
|
||||
&i.SubsonicPassword,
|
||||
&i.ListenbrainzToken,
|
||||
&i.ListenbrainzEnabled,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const getListenBrainzConfig = `-- name: GetListenBrainzConfig :one
|
||||
SELECT
|
||||
u.listenbrainz_token,
|
||||
u.listenbrainz_enabled,
|
||||
(SELECT MAX(pe.scrobbled_at)::timestamptz
|
||||
FROM play_events pe
|
||||
WHERE pe.user_id = u.id) AS last_scrobbled_at
|
||||
FROM users u
|
||||
WHERE u.id = $1
|
||||
`
|
||||
|
||||
type GetListenBrainzConfigRow struct {
|
||||
ListenbrainzToken *string
|
||||
ListenbrainzEnabled bool
|
||||
LastScrobbledAt pgtype.Timestamptz
|
||||
}
|
||||
|
||||
// Returns the user's LB token + enabled flag and the most recent
|
||||
// play_events.scrobbled_at for last-scrobbled-at status.
|
||||
func (q *Queries) GetListenBrainzConfig(ctx context.Context, id pgtype.UUID) (GetListenBrainzConfigRow, error) {
|
||||
row := q.db.QueryRow(ctx, getListenBrainzConfig, id)
|
||||
var i GetListenBrainzConfigRow
|
||||
err := row.Scan(&i.ListenbrainzToken, &i.ListenbrainzEnabled, &i.LastScrobbledAt)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const getUserByAPIToken = `-- name: GetUserByAPIToken :one
|
||||
SELECT id, username, password_hash, api_token, is_admin, created_at, subsonic_password FROM users WHERE api_token = $1
|
||||
SELECT id, username, password_hash, api_token, is_admin, created_at, subsonic_password, listenbrainz_token, listenbrainz_enabled FROM users WHERE api_token = $1
|
||||
`
|
||||
|
||||
func (q *Queries) GetUserByAPIToken(ctx context.Context, apiToken string) (User, error) {
|
||||
@@ -70,12 +98,14 @@ func (q *Queries) GetUserByAPIToken(ctx context.Context, apiToken string) (User,
|
||||
&i.IsAdmin,
|
||||
&i.CreatedAt,
|
||||
&i.SubsonicPassword,
|
||||
&i.ListenbrainzToken,
|
||||
&i.ListenbrainzEnabled,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const getUserByID = `-- name: GetUserByID :one
|
||||
SELECT id, username, password_hash, api_token, is_admin, created_at, subsonic_password FROM users WHERE id = $1
|
||||
SELECT id, username, password_hash, api_token, is_admin, created_at, subsonic_password, listenbrainz_token, listenbrainz_enabled FROM users WHERE id = $1
|
||||
`
|
||||
|
||||
func (q *Queries) GetUserByID(ctx context.Context, id pgtype.UUID) (User, error) {
|
||||
@@ -89,12 +119,14 @@ func (q *Queries) GetUserByID(ctx context.Context, id pgtype.UUID) (User, error)
|
||||
&i.IsAdmin,
|
||||
&i.CreatedAt,
|
||||
&i.SubsonicPassword,
|
||||
&i.ListenbrainzToken,
|
||||
&i.ListenbrainzEnabled,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const getUserByUsername = `-- name: GetUserByUsername :one
|
||||
SELECT id, username, password_hash, api_token, is_admin, created_at, subsonic_password FROM users WHERE username = $1
|
||||
SELECT id, username, password_hash, api_token, is_admin, created_at, subsonic_password, listenbrainz_token, listenbrainz_enabled FROM users WHERE username = $1
|
||||
`
|
||||
|
||||
func (q *Queries) GetUserByUsername(ctx context.Context, username string) (User, error) {
|
||||
@@ -108,10 +140,45 @@ func (q *Queries) GetUserByUsername(ctx context.Context, username string) (User,
|
||||
&i.IsAdmin,
|
||||
&i.CreatedAt,
|
||||
&i.SubsonicPassword,
|
||||
&i.ListenbrainzToken,
|
||||
&i.ListenbrainzEnabled,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const setListenBrainzEnabled = `-- name: SetListenBrainzEnabled :exec
|
||||
UPDATE users
|
||||
SET listenbrainz_enabled = $2
|
||||
WHERE id = $1
|
||||
`
|
||||
|
||||
type SetListenBrainzEnabledParams struct {
|
||||
ID pgtype.UUID
|
||||
ListenbrainzEnabled bool
|
||||
}
|
||||
|
||||
func (q *Queries) SetListenBrainzEnabled(ctx context.Context, arg SetListenBrainzEnabledParams) error {
|
||||
_, err := q.db.Exec(ctx, setListenBrainzEnabled, arg.ID, arg.ListenbrainzEnabled)
|
||||
return err
|
||||
}
|
||||
|
||||
const setListenBrainzToken = `-- name: SetListenBrainzToken :exec
|
||||
UPDATE users
|
||||
SET listenbrainz_token = $2,
|
||||
listenbrainz_enabled = CASE WHEN COALESCE($2, '') = '' THEN FALSE ELSE listenbrainz_enabled END
|
||||
WHERE id = $1
|
||||
`
|
||||
|
||||
type SetListenBrainzTokenParams struct {
|
||||
ID pgtype.UUID
|
||||
ListenbrainzToken *string
|
||||
}
|
||||
|
||||
func (q *Queries) SetListenBrainzToken(ctx context.Context, arg SetListenBrainzTokenParams) error {
|
||||
_, err := q.db.Exec(ctx, setListenBrainzToken, arg.ID, arg.ListenbrainzToken)
|
||||
return err
|
||||
}
|
||||
|
||||
const setSubsonicPassword = `-- name: SetSubsonicPassword :exec
|
||||
UPDATE users SET subsonic_password = $2 WHERE id = $1
|
||||
`
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
DROP TABLE IF EXISTS scrobble_queue;
|
||||
ALTER TABLE users DROP COLUMN IF EXISTS listenbrainz_enabled;
|
||||
ALTER TABLE users DROP COLUMN IF EXISTS listenbrainz_token;
|
||||
@@ -0,0 +1,26 @@
|
||||
-- M4a: outbound ListenBrainz scrobble worker.
|
||||
-- Per-user LB config (plaintext token; users rotate via /settings if leaked).
|
||||
-- scrobble_queue is a work list, not a log: successfully-sent rows are
|
||||
-- DELETEd by the worker (the canonical record stays in play_events.scrobbled_at
|
||||
-- via M2's column). Only `pending` and `failed` states exist.
|
||||
|
||||
ALTER TABLE users
|
||||
ADD COLUMN listenbrainz_token TEXT NULL,
|
||||
ADD COLUMN listenbrainz_enabled BOOLEAN NOT NULL DEFAULT FALSE;
|
||||
|
||||
CREATE TABLE scrobble_queue (
|
||||
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
play_event_id uuid NOT NULL REFERENCES play_events(id) ON DELETE CASCADE,
|
||||
status TEXT NOT NULL CHECK (status IN ('pending', 'failed')),
|
||||
attempts INTEGER NOT NULL DEFAULT 0,
|
||||
next_attempt_at timestamptz NOT NULL DEFAULT now(),
|
||||
last_error TEXT NULL,
|
||||
enqueued_at timestamptz NOT NULL DEFAULT now(),
|
||||
UNIQUE (play_event_id)
|
||||
);
|
||||
|
||||
-- Hot path for the worker's per-tick query.
|
||||
CREATE INDEX scrobble_queue_pending_idx
|
||||
ON scrobble_queue (next_attempt_at)
|
||||
WHERE status = 'pending';
|
||||
@@ -0,0 +1,73 @@
|
||||
-- name: EnqueueScrobble :execrows
|
||||
-- Idempotent: if a row exists for this play_event_id, ON CONFLICT does nothing
|
||||
-- and the affected-rows count is 0. Caller (MaybeEnqueue) doesn't treat 0 as
|
||||
-- an error.
|
||||
INSERT INTO scrobble_queue (user_id, play_event_id, status)
|
||||
VALUES ($1, $2, 'pending')
|
||||
ON CONFLICT (play_event_id) DO NOTHING;
|
||||
|
||||
-- name: ListPendingScrobbles :many
|
||||
-- Worker pulls up to N pending rows whose next_attempt_at has passed.
|
||||
-- Joins play_events + tracks/albums/artists + users so the worker can
|
||||
-- assemble the LB Listen payload in one round-trip.
|
||||
SELECT
|
||||
sq.id AS queue_id,
|
||||
sq.user_id AS user_id,
|
||||
sq.play_event_id AS play_event_id,
|
||||
sq.attempts AS attempts,
|
||||
u.listenbrainz_token AS lb_token,
|
||||
u.listenbrainz_enabled AS lb_enabled,
|
||||
pe.started_at AS started_at,
|
||||
t.title AS track_title,
|
||||
t.duration_ms AS track_duration_ms,
|
||||
t.mbid AS track_mbid,
|
||||
al.title AS album_title,
|
||||
al.mbid AS album_mbid,
|
||||
ar.name AS artist_name,
|
||||
ar.mbid AS artist_mbid
|
||||
FROM scrobble_queue sq
|
||||
JOIN users u ON u.id = sq.user_id
|
||||
JOIN play_events pe ON pe.id = sq.play_event_id
|
||||
JOIN tracks t ON t.id = pe.track_id
|
||||
JOIN albums al ON al.id = t.album_id
|
||||
JOIN artists ar ON ar.id = t.artist_id
|
||||
WHERE sq.status = 'pending'
|
||||
AND sq.next_attempt_at <= now()
|
||||
ORDER BY sq.next_attempt_at
|
||||
LIMIT $1;
|
||||
|
||||
-- name: MarkScrobbleSent :exec
|
||||
-- Successful submit. Delete the queue row AND stamp play_events.scrobbled_at
|
||||
-- in one statement via a CTE so partial failure is impossible.
|
||||
WITH del AS (
|
||||
DELETE FROM scrobble_queue
|
||||
WHERE scrobble_queue.id = $1
|
||||
RETURNING play_event_id
|
||||
)
|
||||
UPDATE play_events
|
||||
SET scrobbled_at = now()
|
||||
WHERE play_events.id = (SELECT play_event_id FROM del);
|
||||
|
||||
-- name: RescheduleScrobble :exec
|
||||
-- After a transient failure: increment attempts, schedule next attempt,
|
||||
-- record the error.
|
||||
UPDATE scrobble_queue
|
||||
SET attempts = attempts + 1,
|
||||
next_attempt_at = $2,
|
||||
last_error = $3
|
||||
WHERE id = $1;
|
||||
|
||||
-- name: MarkScrobbleRetryAfter :exec
|
||||
-- After a 429: schedule next attempt per LB's Retry-After header, but do NOT
|
||||
-- increment attempts (the server told us to wait, not that we failed).
|
||||
UPDATE scrobble_queue
|
||||
SET next_attempt_at = $2
|
||||
WHERE id = $1;
|
||||
|
||||
-- name: MarkScrobbleFailed :exec
|
||||
-- After a permanent failure (4xx, exhausted retries, auth) — mark failed
|
||||
-- and keep the row for diagnostics.
|
||||
UPDATE scrobble_queue
|
||||
SET status = 'failed',
|
||||
last_error = $2
|
||||
WHERE id = $1;
|
||||
@@ -19,3 +19,26 @@ UPDATE users SET subsonic_password = $2 WHERE id = $1;
|
||||
|
||||
-- name: GetUserByID :one
|
||||
SELECT * FROM users WHERE id = $1;
|
||||
|
||||
-- name: SetListenBrainzToken :exec
|
||||
UPDATE users
|
||||
SET listenbrainz_token = $2,
|
||||
listenbrainz_enabled = CASE WHEN COALESCE($2, '') = '' THEN FALSE ELSE listenbrainz_enabled END
|
||||
WHERE id = $1;
|
||||
|
||||
-- name: SetListenBrainzEnabled :exec
|
||||
UPDATE users
|
||||
SET listenbrainz_enabled = $2
|
||||
WHERE id = $1;
|
||||
|
||||
-- name: GetListenBrainzConfig :one
|
||||
-- Returns the user's LB token + enabled flag and the most recent
|
||||
-- play_events.scrobbled_at for last-scrobbled-at status.
|
||||
SELECT
|
||||
u.listenbrainz_token,
|
||||
u.listenbrainz_enabled,
|
||||
(SELECT MAX(pe.scrobbled_at)::timestamptz
|
||||
FROM play_events pe
|
||||
WHERE pe.user_id = u.id) AS last_scrobbled_at
|
||||
FROM users u
|
||||
WHERE u.id = $1;
|
||||
|
||||
@@ -21,6 +21,7 @@ import (
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/playsessions"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/recommendation"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/scrobble"
|
||||
)
|
||||
|
||||
type Writer struct {
|
||||
@@ -109,7 +110,7 @@ func (w *Writer) RecordPlayEnded(
|
||||
durationPlayedMs int32,
|
||||
at time.Time,
|
||||
) error {
|
||||
return pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error {
|
||||
if err := pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error {
|
||||
q := dbq.New(tx)
|
||||
ev, err := q.GetPlayEventByID(ctx, playEventID)
|
||||
if err != nil {
|
||||
@@ -147,7 +148,16 @@ func (w *Writer) RecordPlayEnded(
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
// Post-commit best-effort scrobble enqueue. Errors logged + swallowed:
|
||||
// scrobble delivery is enrichment, not a precondition for canonical play
|
||||
// history.
|
||||
if err := scrobble.MaybeEnqueue(ctx, dbq.New(w.pool), playEventID); err != nil {
|
||||
w.logger.Warn("playevents: scrobble enqueue failed", "play_event_id", playEventID, "err", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RecordPlaySkipped is the user-initiated-skip variant: was_skipped is forced
|
||||
@@ -203,7 +213,8 @@ func (w *Writer) RecordSyntheticCompletedPlay(
|
||||
clientID string,
|
||||
at time.Time,
|
||||
) error {
|
||||
return pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error {
|
||||
var playEventID pgtype.UUID
|
||||
if err := pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error {
|
||||
q := dbq.New(tx)
|
||||
track, err := q.GetTrackByID(ctx, trackID)
|
||||
if err != nil {
|
||||
@@ -230,6 +241,7 @@ func (w *Writer) RecordSyntheticCompletedPlay(
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
playEventID = ev.ID
|
||||
if err := w.captureSessionVector(ctx, q, ev.ID, sessionID, at); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -243,7 +255,16 @@ func (w *Writer) RecordSyntheticCompletedPlay(
|
||||
WasSkipped: false,
|
||||
})
|
||||
return err
|
||||
})
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
// Post-commit best-effort scrobble enqueue. Errors logged + swallowed:
|
||||
// scrobble delivery is enrichment, not a precondition for canonical play
|
||||
// history.
|
||||
if err := scrobble.MaybeEnqueue(ctx, dbq.New(w.pool), playEventID); err != nil {
|
||||
w.logger.Warn("playevents: scrobble enqueue failed", "play_event_id", playEventID, "err", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// captureSessionVector queries the user's prior plays in the given session
|
||||
|
||||
@@ -34,7 +34,7 @@ func testPool(t *testing.T) *pgxpool.Pool {
|
||||
}
|
||||
t.Cleanup(pool.Close)
|
||||
if _, err := pool.Exec(context.Background(),
|
||||
"TRUNCATE play_events, skip_events, play_sessions, sessions, users, tracks, albums, artists RESTART IDENTITY CASCADE"); err != nil {
|
||||
"TRUNCATE 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
|
||||
@@ -317,3 +317,60 @@ func TestRecordPlayStarted_VectorScopedToSession(t *testing.T) {
|
||||
t.Errorf("recent_track_ids has %d entries from prior session, want 0", len(recentIDs))
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecordPlayEnded_QualifyingPlay_EnqueuesScrobble(t *testing.T) {
|
||||
f := newFixture(t, 300_000) // 300s track
|
||||
ctx := context.Background()
|
||||
|
||||
// Enable LB for the user.
|
||||
tk := "tk"
|
||||
if err := f.q.SetListenBrainzToken(ctx, dbq.SetListenBrainzTokenParams{ID: f.user, ListenbrainzToken: &tk}); err != nil {
|
||||
t.Fatalf("token: %v", err)
|
||||
}
|
||||
if err := f.q.SetListenBrainzEnabled(ctx, dbq.SetListenBrainzEnabledParams{ID: f.user, ListenbrainzEnabled: true}); err != nil {
|
||||
t.Fatalf("enable: %v", err)
|
||||
}
|
||||
|
||||
res, err := f.w.RecordPlayStarted(ctx, f.user, f.track, "test", time.Now().Add(-1*time.Minute))
|
||||
if err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
// 250s played out of 300s = ~83% — well above LB threshold.
|
||||
if err := f.w.RecordPlayEnded(ctx, res.PlayEventID, 250_000, time.Now()); err != nil {
|
||||
t.Fatalf("end: %v", err)
|
||||
}
|
||||
|
||||
var n int
|
||||
if err := f.pool.QueryRow(ctx,
|
||||
`SELECT count(*) FROM scrobble_queue WHERE play_event_id = $1`, res.PlayEventID).Scan(&n); err != nil {
|
||||
t.Fatalf("count: %v", err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Errorf("scrobble_queue rows = %d, want 1", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecordPlayEnded_SubThreshold_NoEnqueue(t *testing.T) {
|
||||
f := newFixture(t, 300_000)
|
||||
ctx := context.Background()
|
||||
|
||||
tk := "tk"
|
||||
_ = f.q.SetListenBrainzToken(ctx, dbq.SetListenBrainzTokenParams{ID: f.user, ListenbrainzToken: &tk})
|
||||
_ = f.q.SetListenBrainzEnabled(ctx, dbq.SetListenBrainzEnabledParams{ID: f.user, ListenbrainzEnabled: true})
|
||||
|
||||
res, err := f.w.RecordPlayStarted(ctx, f.user, f.track, "test", time.Now().Add(-1*time.Minute))
|
||||
if err != nil {
|
||||
t.Fatalf("start: %v", err)
|
||||
}
|
||||
// 60s played, 20% — well below threshold.
|
||||
if err := f.w.RecordPlayEnded(ctx, res.PlayEventID, 60_000, time.Now()); err != nil {
|
||||
t.Fatalf("end: %v", err)
|
||||
}
|
||||
|
||||
var n int
|
||||
_ = f.pool.QueryRow(ctx,
|
||||
`SELECT count(*) FROM scrobble_queue WHERE play_event_id = $1`, res.PlayEventID).Scan(&n)
|
||||
if n != 0 {
|
||||
t.Errorf("expected 0 queue rows for sub-threshold play, got %d", n)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
// Package listenbrainz is the pure HTTP client for the ListenBrainz
|
||||
// submit-listens endpoint. No DB, no worker plumbing — just types and one
|
||||
// method. Errors are typed (ErrAuth, ErrPermanent, ErrTransient,
|
||||
// *RetryAfterError) so callers can branch on response semantics.
|
||||
package listenbrainz
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
const defaultBaseURL = "https://api.listenbrainz.org"
|
||||
|
||||
// Listen is one row sent to ListenBrainz.
|
||||
type Listen struct {
|
||||
ListenedAt int64 // unix seconds
|
||||
Track Track
|
||||
}
|
||||
|
||||
// Track is the per-listen metadata. Optional fields are omitted from the
|
||||
// payload when zero-valued.
|
||||
type Track struct {
|
||||
ArtistName string
|
||||
TrackName string
|
||||
ReleaseName string
|
||||
DurationMs int
|
||||
RecordingMBID string
|
||||
ArtistMBIDs []string
|
||||
ReleaseMBID string
|
||||
}
|
||||
|
||||
// Client posts to the LB submit-listens endpoint.
|
||||
type Client struct {
|
||||
BaseURL string
|
||||
HTTP *http.Client
|
||||
}
|
||||
|
||||
// NewClient returns a default-configured client (LB production base URL,
|
||||
// 30s timeout).
|
||||
func NewClient() *Client {
|
||||
return &Client{
|
||||
BaseURL: defaultBaseURL,
|
||||
HTTP: &http.Client{Timeout: 30 * time.Second},
|
||||
}
|
||||
}
|
||||
|
||||
// Sentinel errors. The worker branches on these to decide retry vs fail.
|
||||
var (
|
||||
ErrAuth = errors.New("listenbrainz: auth rejected")
|
||||
ErrPermanent = errors.New("listenbrainz: permanent error")
|
||||
ErrTransient = errors.New("listenbrainz: transient error")
|
||||
)
|
||||
|
||||
// RetryAfterError carries the LB-supplied wait duration for 429 responses.
|
||||
type RetryAfterError struct{ Wait time.Duration }
|
||||
|
||||
func (e *RetryAfterError) Error() string {
|
||||
return fmt.Sprintf("listenbrainz: retry after %v", e.Wait)
|
||||
}
|
||||
|
||||
// SubmitListens posts the given listens. payload_type is "single" for one
|
||||
// listen, "import" for batches.
|
||||
func (c *Client) SubmitListens(ctx context.Context, token string, listens []Listen) error {
|
||||
if len(listens) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
listenType := "import"
|
||||
if len(listens) == 1 {
|
||||
listenType = "single"
|
||||
}
|
||||
|
||||
body, err := json.Marshal(buildPayload(listenType, listens))
|
||||
if err != nil {
|
||||
return fmt.Errorf("listenbrainz: marshal: %w", err)
|
||||
}
|
||||
|
||||
base := c.BaseURL
|
||||
if base == "" {
|
||||
base = defaultBaseURL
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, base+"/1/submit-listens", bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return fmt.Errorf("listenbrainz: build request: %w", err)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Authorization", "Token "+token)
|
||||
|
||||
httpClient := c.HTTP
|
||||
if httpClient == nil {
|
||||
httpClient = http.DefaultClient
|
||||
}
|
||||
resp, err := httpClient.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: %v", ErrTransient, err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
switch {
|
||||
case resp.StatusCode >= 200 && resp.StatusCode < 300:
|
||||
return nil
|
||||
case resp.StatusCode == http.StatusUnauthorized:
|
||||
return ErrAuth
|
||||
case resp.StatusCode == http.StatusTooManyRequests:
|
||||
wait := parseRetryAfter(resp.Header.Get("Retry-After"))
|
||||
return &RetryAfterError{Wait: wait}
|
||||
case resp.StatusCode >= 500:
|
||||
return fmt.Errorf("%w: status %d", ErrTransient, resp.StatusCode)
|
||||
default:
|
||||
return fmt.Errorf("%w: status %d", ErrPermanent, resp.StatusCode)
|
||||
}
|
||||
}
|
||||
|
||||
func parseRetryAfter(header string) time.Duration {
|
||||
if header == "" {
|
||||
return 60 * time.Second // sensible default
|
||||
}
|
||||
if secs, err := strconv.Atoi(header); err == nil {
|
||||
return time.Duration(secs) * time.Second
|
||||
}
|
||||
if t, err := http.ParseTime(header); err == nil {
|
||||
d := time.Until(t)
|
||||
if d < 0 {
|
||||
return 60 * time.Second
|
||||
}
|
||||
return d
|
||||
}
|
||||
return 60 * time.Second
|
||||
}
|
||||
|
||||
// payload mirrors the LB submit-listens body. Optional MBID fields use
|
||||
// `omitempty` so empty strings/slices don't appear in the JSON.
|
||||
type payload struct {
|
||||
ListenType string `json:"listen_type"`
|
||||
Payload []payloadEntry `json:"payload"`
|
||||
}
|
||||
|
||||
type payloadEntry struct {
|
||||
ListenedAt int64 `json:"listened_at"`
|
||||
TrackMetadata trackMetadata `json:"track_metadata"`
|
||||
}
|
||||
|
||||
type trackMetadata struct {
|
||||
ArtistName string `json:"artist_name"`
|
||||
TrackName string `json:"track_name"`
|
||||
ReleaseName string `json:"release_name,omitempty"`
|
||||
AdditionalInfo additionalInfo `json:"additional_info,omitempty"`
|
||||
}
|
||||
|
||||
type additionalInfo struct {
|
||||
DurationMs int `json:"duration_ms,omitempty"`
|
||||
RecordingMBID string `json:"recording_mbid,omitempty"`
|
||||
ArtistMBIDs []string `json:"artist_mbids,omitempty"`
|
||||
ReleaseMBID string `json:"release_mbid,omitempty"`
|
||||
}
|
||||
|
||||
func buildPayload(listenType string, listens []Listen) payload {
|
||||
entries := make([]payloadEntry, 0, len(listens))
|
||||
for _, l := range listens {
|
||||
entries = append(entries, payloadEntry{
|
||||
ListenedAt: l.ListenedAt,
|
||||
TrackMetadata: trackMetadata{
|
||||
ArtistName: l.Track.ArtistName,
|
||||
TrackName: l.Track.TrackName,
|
||||
ReleaseName: l.Track.ReleaseName,
|
||||
AdditionalInfo: additionalInfo{
|
||||
DurationMs: l.Track.DurationMs,
|
||||
RecordingMBID: l.Track.RecordingMBID,
|
||||
ArtistMBIDs: l.Track.ArtistMBIDs,
|
||||
ReleaseMBID: l.Track.ReleaseMBID,
|
||||
},
|
||||
},
|
||||
})
|
||||
}
|
||||
return payload{ListenType: listenType, Payload: entries}
|
||||
}
|
||||
@@ -0,0 +1,164 @@
|
||||
package listenbrainz
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func newTestClient(handler http.HandlerFunc) (*Client, *httptest.Server) {
|
||||
srv := httptest.NewServer(handler)
|
||||
return &Client{BaseURL: srv.URL, HTTP: srv.Client()}, srv
|
||||
}
|
||||
|
||||
func TestClient_SubmitListens_Success(t *testing.T) {
|
||||
var seenPath, seenAuth, seenBody string
|
||||
c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) {
|
||||
seenPath = r.URL.Path
|
||||
seenAuth = r.Header.Get("Authorization")
|
||||
buf := make([]byte, 4096)
|
||||
n, _ := r.Body.Read(buf)
|
||||
seenBody = string(buf[:n])
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write([]byte(`{"status":"ok"}`))
|
||||
})
|
||||
defer srv.Close()
|
||||
|
||||
listens := []Listen{{
|
||||
ListenedAt: 1700000000,
|
||||
Track: Track{
|
||||
ArtistName: "Miles Davis",
|
||||
TrackName: "So What",
|
||||
ReleaseName: "Kind of Blue",
|
||||
DurationMs: 545000,
|
||||
},
|
||||
}}
|
||||
if err := c.SubmitListens(context.Background(), "tk", listens); err != nil {
|
||||
t.Fatalf("err: %v", err)
|
||||
}
|
||||
if seenPath != "/1/submit-listens" {
|
||||
t.Errorf("path = %q, want /1/submit-listens", seenPath)
|
||||
}
|
||||
if seenAuth != "Token tk" {
|
||||
t.Errorf("auth = %q, want %q", seenAuth, "Token tk")
|
||||
}
|
||||
if !strings.Contains(seenBody, `"track_name":"So What"`) {
|
||||
t.Errorf("body missing track_name: %s", seenBody)
|
||||
}
|
||||
if !strings.Contains(seenBody, `"artist_name":"Miles Davis"`) {
|
||||
t.Errorf("body missing artist_name: %s", seenBody)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_SubmitListens_401_ReturnsErrAuth(t *testing.T) {
|
||||
c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
})
|
||||
defer srv.Close()
|
||||
err := c.SubmitListens(context.Background(), "bad", []Listen{{}})
|
||||
if !errors.Is(err, ErrAuth) {
|
||||
t.Errorf("err = %v, want ErrAuth", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_SubmitListens_400_ReturnsErrPermanent(t *testing.T) {
|
||||
c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
})
|
||||
defer srv.Close()
|
||||
err := c.SubmitListens(context.Background(), "tk", []Listen{{}})
|
||||
if !errors.Is(err, ErrPermanent) {
|
||||
t.Errorf("err = %v, want ErrPermanent", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_SubmitListens_503_ReturnsErrTransient(t *testing.T) {
|
||||
c, srv := newTestClient(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
})
|
||||
defer srv.Close()
|
||||
err := c.SubmitListens(context.Background(), "tk", []Listen{{}})
|
||||
if !errors.Is(err, ErrTransient) {
|
||||
t.Errorf("err = %v, want ErrTransient", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_SubmitListens_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.SubmitListens(context.Background(), "tk", []Listen{{}})
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_SubmitListens_NetworkFailure_ReturnsErrTransient(t *testing.T) {
|
||||
c := &Client{BaseURL: "http://127.0.0.1:1", HTTP: &http.Client{Timeout: 100 * time.Millisecond}}
|
||||
err := c.SubmitListens(context.Background(), "tk", []Listen{{}})
|
||||
if !errors.Is(err, ErrTransient) {
|
||||
t.Errorf("err = %v, want ErrTransient", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_SubmitListens_PayloadTypeImportForBatch(t *testing.T) {
|
||||
var seenBody string
|
||||
c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) {
|
||||
buf := make([]byte, 4096)
|
||||
n, _ := r.Body.Read(buf)
|
||||
seenBody = string(buf[:n])
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
defer srv.Close()
|
||||
_ = c.SubmitListens(context.Background(), "tk", []Listen{
|
||||
{ListenedAt: 1, Track: Track{ArtistName: "A", TrackName: "X"}},
|
||||
{ListenedAt: 2, Track: Track{ArtistName: "B", TrackName: "Y"}},
|
||||
})
|
||||
if !strings.Contains(seenBody, `"listen_type":"import"`) {
|
||||
t.Errorf("batch body missing listen_type=import: %s", seenBody)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_SubmitListens_PayloadTypeSingleForOne(t *testing.T) {
|
||||
var seenBody string
|
||||
c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) {
|
||||
buf := make([]byte, 4096)
|
||||
n, _ := r.Body.Read(buf)
|
||||
seenBody = string(buf[:n])
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
defer srv.Close()
|
||||
_ = c.SubmitListens(context.Background(), "tk", []Listen{
|
||||
{ListenedAt: 1, Track: Track{ArtistName: "A", TrackName: "X"}},
|
||||
})
|
||||
if !strings.Contains(seenBody, `"listen_type":"single"`) {
|
||||
t.Errorf("single body missing listen_type=single: %s", seenBody)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_SubmitListens_OmitsEmptyMBIDs(t *testing.T) {
|
||||
var seenBody string
|
||||
c, srv := newTestClient(func(w http.ResponseWriter, r *http.Request) {
|
||||
buf := make([]byte, 4096)
|
||||
n, _ := r.Body.Read(buf)
|
||||
seenBody = string(buf[:n])
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
defer srv.Close()
|
||||
_ = c.SubmitListens(context.Background(), "tk", []Listen{
|
||||
{ListenedAt: 1, Track: Track{ArtistName: "A", TrackName: "X"}},
|
||||
})
|
||||
if strings.Contains(seenBody, "recording_mbid") {
|
||||
t.Errorf("body should omit empty recording_mbid: %s", seenBody)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
package scrobble
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||
)
|
||||
|
||||
// MaybeEnqueue inserts a scrobble_queue row for the given play_event if:
|
||||
// 1. The user has listenbrainz_enabled = true with a non-empty token.
|
||||
// 2. The play_event passes Qualifies().
|
||||
//
|
||||
// Idempotent via UNIQUE(play_event_id) — re-runs are no-ops.
|
||||
//
|
||||
// Best-effort: callers (the playevents.Writer hooks) should log returned
|
||||
// errors but not propagate them, since scrobble delivery is enrichment
|
||||
// and must not block the canonical play history.
|
||||
func MaybeEnqueue(ctx context.Context, q *dbq.Queries, playEventID pgtype.UUID) error {
|
||||
pe, err := q.GetPlayEventByID(ctx, playEventID)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
// Only closed events qualify (DurationPlayedMs/CompletionRatio populated).
|
||||
if pe.DurationPlayedMs == nil || pe.CompletionRatio == nil {
|
||||
return nil
|
||||
}
|
||||
if !Qualifies(int(*pe.DurationPlayedMs), *pe.CompletionRatio) {
|
||||
return nil
|
||||
}
|
||||
cfg, err := q.GetListenBrainzConfig(ctx, pe.UserID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !cfg.ListenbrainzEnabled || cfg.ListenbrainzToken == nil || *cfg.ListenbrainzToken == "" {
|
||||
return nil
|
||||
}
|
||||
_, err = q.EnqueueScrobble(ctx, dbq.EnqueueScrobbleParams{
|
||||
UserID: pe.UserID,
|
||||
PlayEventID: pe.ID,
|
||||
})
|
||||
return err
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
package scrobble
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"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"
|
||||
)
|
||||
|
||||
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 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 setup struct {
|
||||
pool *pgxpool.Pool
|
||||
q *dbq.Queries
|
||||
user pgtype.UUID
|
||||
playEvent pgtype.UUID
|
||||
}
|
||||
|
||||
type seedOpts struct {
|
||||
lbToken string
|
||||
lbEnabled bool
|
||||
durationMs int32
|
||||
ratio float64
|
||||
}
|
||||
|
||||
// seed creates: user (with optional LB token + enabled), one artist/album/track,
|
||||
// one closed play_event with the given duration_played_ms and completion_ratio.
|
||||
func seed(t *testing.T, opts seedOpts) setup {
|
||||
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)
|
||||
}
|
||||
if opts.lbToken != "" {
|
||||
if err := q.SetListenBrainzToken(ctx, dbq.SetListenBrainzTokenParams{
|
||||
ID: u.ID, ListenbrainzToken: &opts.lbToken,
|
||||
}); err != nil {
|
||||
t.Fatalf("token: %v", err)
|
||||
}
|
||||
}
|
||||
if opts.lbEnabled {
|
||||
if err := q.SetListenBrainzEnabled(ctx, dbq.SetListenBrainzEnabledParams{
|
||||
ID: u.ID, ListenbrainzEnabled: true,
|
||||
}); err != nil {
|
||||
t.Fatalf("enabled: %v", err)
|
||||
}
|
||||
}
|
||||
a, _ := q.UpsertArtist(ctx, dbq.UpsertArtistParams{Name: "X", SortName: "X"})
|
||||
al, _ := q.UpsertAlbum(ctx, dbq.UpsertAlbumParams{Title: "X", SortTitle: "X", ArtistID: a.ID})
|
||||
tr, _ := q.UpsertTrack(ctx, dbq.UpsertTrackParams{
|
||||
Title: "Y", AlbumID: al.ID, ArtistID: a.ID, FilePath: "/tmp/y.flac", DurationMs: 300_000,
|
||||
})
|
||||
var sessionID pgtype.UUID
|
||||
if err := pool.QueryRow(ctx,
|
||||
`INSERT INTO play_sessions (user_id, started_at, last_event_at, client_id)
|
||||
VALUES ($1, now() - interval '5 minutes', now(), 'test') RETURNING id`,
|
||||
u.ID).Scan(&sessionID); err != nil {
|
||||
t.Fatalf("session: %v", err)
|
||||
}
|
||||
var peID pgtype.UUID
|
||||
if err := pool.QueryRow(ctx,
|
||||
`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(), $4, $5, false) RETURNING id`,
|
||||
u.ID, tr.ID, sessionID, opts.durationMs, opts.ratio).Scan(&peID); err != nil {
|
||||
t.Fatalf("play_event: %v", err)
|
||||
}
|
||||
return setup{pool: pool, q: q, user: u.ID, playEvent: peID}
|
||||
}
|
||||
|
||||
func countQueueRows(t *testing.T, s setup) int {
|
||||
t.Helper()
|
||||
var n int
|
||||
if err := s.pool.QueryRow(context.Background(),
|
||||
`SELECT count(*) FROM scrobble_queue WHERE play_event_id = $1`, s.playEvent).Scan(&n); err != nil {
|
||||
t.Fatalf("count: %v", err)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func TestMaybeEnqueue_QualifyingPlayWithEnabledUser_InsertsRow(t *testing.T) {
|
||||
s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
|
||||
if err := MaybeEnqueue(context.Background(), s.q, s.playEvent); err != nil {
|
||||
t.Fatalf("MaybeEnqueue: %v", err)
|
||||
}
|
||||
if got := countQueueRows(t, s); got != 1 {
|
||||
t.Errorf("rows = %d, want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaybeEnqueue_SubThreshold_NoRow(t *testing.T) {
|
||||
s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 100_000, ratio: 0.33})
|
||||
if err := MaybeEnqueue(context.Background(), s.q, s.playEvent); err != nil {
|
||||
t.Fatalf("MaybeEnqueue: %v", err)
|
||||
}
|
||||
if got := countQueueRows(t, s); got != 0 {
|
||||
t.Errorf("rows = %d, want 0 (sub-threshold)", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaybeEnqueue_DisabledUser_NoRow(t *testing.T) {
|
||||
s := seed(t, seedOpts{lbToken: "tk", lbEnabled: false, durationMs: 250_000, ratio: 0.83})
|
||||
if err := MaybeEnqueue(context.Background(), s.q, s.playEvent); err != nil {
|
||||
t.Fatalf("MaybeEnqueue: %v", err)
|
||||
}
|
||||
if got := countQueueRows(t, s); got != 0 {
|
||||
t.Errorf("rows = %d, want 0 (disabled)", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaybeEnqueue_NoToken_NoRow(t *testing.T) {
|
||||
s := seed(t, seedOpts{lbToken: "", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
|
||||
if err := MaybeEnqueue(context.Background(), s.q, s.playEvent); err != nil {
|
||||
t.Fatalf("MaybeEnqueue: %v", err)
|
||||
}
|
||||
if got := countQueueRows(t, s); got != 0 {
|
||||
t.Errorf("rows = %d, want 0 (no token)", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaybeEnqueue_Idempotent_TwoCallsOneRow(t *testing.T) {
|
||||
s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
|
||||
for i := 0; i < 2; i++ {
|
||||
if err := MaybeEnqueue(context.Background(), s.q, s.playEvent); err != nil {
|
||||
t.Fatalf("MaybeEnqueue #%d: %v", i, err)
|
||||
}
|
||||
}
|
||||
if got := countQueueRows(t, s); got != 1 {
|
||||
t.Errorf("rows = %d, want 1 (idempotent)", got)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
// Package scrobble owns the outbound-scrobble pipeline: threshold detection,
|
||||
// the queue write path, and the worker goroutine. The HTTP client is in
|
||||
// internal/scrobble/listenbrainz; this package consumes it.
|
||||
package scrobble
|
||||
|
||||
// Qualifies returns true if a closed play_event meets ListenBrainz's
|
||||
// recommended scrobble threshold: ≥240 seconds played OR ≥50% of the
|
||||
// track length played.
|
||||
//
|
||||
// The two thresholds are deliberately separate from M2's skip-detection
|
||||
// thresholds (which serve a different purpose — internal engine signal
|
||||
// for the scoring formula's skip penalty).
|
||||
func Qualifies(durationPlayedMs int, completionRatio float64) bool {
|
||||
return durationPlayedMs >= 240_000 || completionRatio >= 0.5
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
package scrobble
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestQualifies_NeverPlayed(t *testing.T) {
|
||||
if Qualifies(0, 0) {
|
||||
t.Error("0/0 should not qualify")
|
||||
}
|
||||
}
|
||||
|
||||
func TestQualifies_AtLeast240Seconds(t *testing.T) {
|
||||
if !Qualifies(240_000, 0.0) {
|
||||
t.Error("240_000 ms played should qualify regardless of ratio")
|
||||
}
|
||||
}
|
||||
|
||||
func TestQualifies_AtLeastHalfRatio(t *testing.T) {
|
||||
if !Qualifies(0, 0.5) {
|
||||
t.Error("ratio=0.5 should qualify regardless of duration")
|
||||
}
|
||||
}
|
||||
|
||||
func TestQualifies_BelowBoth(t *testing.T) {
|
||||
if Qualifies(120_000, 0.49) {
|
||||
t.Error("120s + 49% should not qualify")
|
||||
}
|
||||
}
|
||||
|
||||
func TestQualifies_BoundaryJustUnder(t *testing.T) {
|
||||
if Qualifies(239_999, 0.4999) {
|
||||
t.Error("just-under both thresholds should not qualify")
|
||||
}
|
||||
}
|
||||
|
||||
func TestQualifies_BoundaryExactly(t *testing.T) {
|
||||
if !Qualifies(240_000, 0.4999) {
|
||||
t.Error("exactly 240s should qualify")
|
||||
}
|
||||
if !Qualifies(239_999, 0.5) {
|
||||
t.Error("exactly 50% should qualify")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,227 @@
|
||||
package scrobble
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"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"
|
||||
)
|
||||
|
||||
// backoffSchedule maps the failure count (1-indexed) to the delay before
|
||||
// the next attempt. Per spec §M4a: 1m → 5m → 30m → 2h → 6h, then give up.
|
||||
var backoffSchedule = []time.Duration{
|
||||
1 * time.Minute,
|
||||
5 * time.Minute,
|
||||
30 * time.Minute,
|
||||
2 * time.Hour,
|
||||
6 * time.Hour,
|
||||
}
|
||||
|
||||
const maxAttempts = 5
|
||||
|
||||
// backoffDelay returns the delay before the next attempt given the value
|
||||
// of the `attempts` column AFTER the failure has been recorded (1-indexed:
|
||||
// attempts=1 means the first failure just happened, the next retry should
|
||||
// wait 1 minute). Returns (_, false) when attempts > maxAttempts (the
|
||||
// caller should mark the row failed).
|
||||
func backoffDelay(attempts int) (time.Duration, bool) {
|
||||
if attempts < 1 || attempts > maxAttempts {
|
||||
return 0, false
|
||||
}
|
||||
return backoffSchedule[attempts-1], true
|
||||
}
|
||||
|
||||
// Worker drains scrobble_queue rows and POSTs them to ListenBrainz.
|
||||
type Worker struct {
|
||||
pool *pgxpool.Pool
|
||||
client *listenbrainz.Client
|
||||
logger *slog.Logger
|
||||
tick time.Duration
|
||||
batch int32
|
||||
}
|
||||
|
||||
// NewWorker constructs a worker with production defaults: 30s tick, 50-row
|
||||
// batch.
|
||||
func NewWorker(pool *pgxpool.Pool, client *listenbrainz.Client, logger *slog.Logger) *Worker {
|
||||
return &Worker{pool: pool, client: client, logger: logger, tick: 30 * time.Second, batch: 50}
|
||||
}
|
||||
|
||||
// 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("scrobble: tick failed", "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// tickOnce drains up to w.batch pending rows. Returns the first error
|
||||
// encountered loading rows; per-row errors are logged and don't abort
|
||||
// the batch.
|
||||
func (w *Worker) tickOnce(ctx context.Context) error {
|
||||
q := dbq.New(w.pool)
|
||||
rows, err := q.ListPendingScrobbles(ctx, w.batch)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(rows) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Group rows by user (each user has their own token + may produce its
|
||||
// own batch). Map keyed by user UUID string.
|
||||
type userBatch struct {
|
||||
token string
|
||||
queueIDs []pgtype.UUID
|
||||
listens []listenbrainz.Listen
|
||||
userID pgtype.UUID
|
||||
}
|
||||
batches := map[string]*userBatch{}
|
||||
for _, r := range rows {
|
||||
if r.LbToken == nil || *r.LbToken == "" || !r.LbEnabled {
|
||||
// Defensive: row got into queue but user since disabled. Skip.
|
||||
continue
|
||||
}
|
||||
key := r.UserID.String()
|
||||
ub := batches[key]
|
||||
if ub == nil {
|
||||
ub = &userBatch{token: *r.LbToken, userID: r.UserID}
|
||||
batches[key] = ub
|
||||
}
|
||||
ub.queueIDs = append(ub.queueIDs, r.QueueID)
|
||||
l := listenbrainz.Listen{
|
||||
ListenedAt: r.StartedAt.Time.Unix(),
|
||||
Track: listenbrainz.Track{
|
||||
ArtistName: r.ArtistName,
|
||||
TrackName: r.TrackTitle,
|
||||
ReleaseName: r.AlbumTitle,
|
||||
DurationMs: int(r.TrackDurationMs),
|
||||
},
|
||||
}
|
||||
if r.TrackMbid != nil {
|
||||
l.Track.RecordingMBID = *r.TrackMbid
|
||||
}
|
||||
if r.AlbumMbid != nil {
|
||||
l.Track.ReleaseMBID = *r.AlbumMbid
|
||||
}
|
||||
if r.ArtistMbid != nil {
|
||||
l.Track.ArtistMBIDs = []string{*r.ArtistMbid}
|
||||
}
|
||||
ub.listens = append(ub.listens, l)
|
||||
}
|
||||
|
||||
for _, ub := range batches {
|
||||
w.processBatch(ctx, q, ub.userID, ub.token, ub.queueIDs, ub.listens)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// processBatch submits one user's pending batch and updates queue rows
|
||||
// according to the response.
|
||||
func (w *Worker) processBatch(
|
||||
ctx context.Context,
|
||||
q *dbq.Queries,
|
||||
userID pgtype.UUID,
|
||||
token string,
|
||||
queueIDs []pgtype.UUID,
|
||||
listens []listenbrainz.Listen,
|
||||
) {
|
||||
err := w.client.SubmitListens(ctx, token, listens)
|
||||
now := time.Now().UTC()
|
||||
|
||||
if err == nil {
|
||||
for _, id := range queueIDs {
|
||||
if uerr := q.MarkScrobbleSent(ctx, id); uerr != nil {
|
||||
w.logger.Error("scrobble: MarkScrobbleSent", "queue_id", id, "err", uerr)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// 429: schedule retry per Retry-After, do NOT increment attempts.
|
||||
var ra *listenbrainz.RetryAfterError
|
||||
if errors.As(err, &ra) {
|
||||
next := pgtype.Timestamptz{Time: now.Add(ra.Wait), Valid: true}
|
||||
for _, id := range queueIDs {
|
||||
if uerr := q.MarkScrobbleRetryAfter(ctx, dbq.MarkScrobbleRetryAfterParams{
|
||||
ID: id, NextAttemptAt: next,
|
||||
}); uerr != nil {
|
||||
w.logger.Error("scrobble: MarkScrobbleRetryAfter", "err", uerr)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// 401: mark failed AND disable user (defensive — bad token shouldn't keep
|
||||
// retrying or feed future rows).
|
||||
if errors.Is(err, listenbrainz.ErrAuth) {
|
||||
w.logger.Warn("scrobble: auth rejected; disabling user listenbrainz", "user_id", userID)
|
||||
if uerr := q.SetListenBrainzEnabled(ctx, dbq.SetListenBrainzEnabledParams{
|
||||
ID: userID, ListenbrainzEnabled: false,
|
||||
}); uerr != nil {
|
||||
w.logger.Error("scrobble: SetListenBrainzEnabled", "err", uerr)
|
||||
}
|
||||
errStr := err.Error()
|
||||
for _, id := range queueIDs {
|
||||
if uerr := q.MarkScrobbleFailed(ctx, dbq.MarkScrobbleFailedParams{
|
||||
ID: id, LastError: &errStr,
|
||||
}); uerr != nil {
|
||||
w.logger.Error("scrobble: MarkScrobbleFailed", "err", uerr)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// 4xx: permanent. Mark failed, no retry.
|
||||
if errors.Is(err, listenbrainz.ErrPermanent) {
|
||||
errStr := err.Error()
|
||||
for _, id := range queueIDs {
|
||||
if uerr := q.MarkScrobbleFailed(ctx, dbq.MarkScrobbleFailedParams{
|
||||
ID: id, LastError: &errStr,
|
||||
}); uerr != nil {
|
||||
w.logger.Error("scrobble: MarkScrobbleFailed", "err", uerr)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// Transient (5xx, network): per-row, look up backoffDelay(currentAttempts+1).
|
||||
// If OK → RescheduleScrobble (which increments). If NOT OK → MarkScrobbleFailed
|
||||
// (don't increment further; current attempts already represents the cap).
|
||||
errStr := err.Error()
|
||||
for _, id := range queueIDs {
|
||||
var attempts int32
|
||||
if rerr := w.pool.QueryRow(ctx,
|
||||
`SELECT attempts FROM scrobble_queue WHERE id = $1`, id).Scan(&attempts); rerr != nil {
|
||||
w.logger.Error("scrobble: read attempts", "err", rerr)
|
||||
continue
|
||||
}
|
||||
if delay, ok := backoffDelay(int(attempts) + 1); ok {
|
||||
nextAt := pgtype.Timestamptz{Time: now.Add(delay), Valid: true}
|
||||
if uerr := q.RescheduleScrobble(ctx, dbq.RescheduleScrobbleParams{
|
||||
ID: id, NextAttemptAt: nextAt, LastError: &errStr,
|
||||
}); uerr != nil {
|
||||
w.logger.Error("scrobble: RescheduleScrobble", "err", uerr)
|
||||
}
|
||||
} else {
|
||||
if uerr := q.MarkScrobbleFailed(ctx, dbq.MarkScrobbleFailedParams{
|
||||
ID: id, LastError: &errStr,
|
||||
}); uerr != nil {
|
||||
w.logger.Error("scrobble: MarkScrobbleFailed", "err", uerr)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,224 @@
|
||||
package scrobble
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz"
|
||||
)
|
||||
|
||||
// helperEnqueue inserts a queue row directly so we don't go through the
|
||||
// threshold path. Reuses the `setup` struct from queue_test.go.
|
||||
func helperEnqueue(t *testing.T, s setup) {
|
||||
t.Helper()
|
||||
if _, err := s.pool.Exec(context.Background(),
|
||||
`INSERT INTO scrobble_queue (user_id, play_event_id, status) VALUES ($1, $2, 'pending')`,
|
||||
s.user, s.playEvent); err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func helperPendingCount(t *testing.T, s setup) int {
|
||||
t.Helper()
|
||||
var n int
|
||||
if err := s.pool.QueryRow(context.Background(),
|
||||
`SELECT count(*) FROM scrobble_queue WHERE status = 'pending' AND play_event_id = $1`, s.playEvent).Scan(&n); err != nil {
|
||||
t.Fatalf("pending: %v", err)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func helperFailedCount(t *testing.T, s setup) int {
|
||||
t.Helper()
|
||||
var n int
|
||||
if err := s.pool.QueryRow(context.Background(),
|
||||
`SELECT count(*) FROM scrobble_queue WHERE status = 'failed' AND play_event_id = $1`, s.playEvent).Scan(&n); err != nil {
|
||||
t.Fatalf("failed: %v", err)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func helperPlayEventScrobbledAt(t *testing.T, s setup) bool {
|
||||
t.Helper()
|
||||
var v *time.Time
|
||||
if err := s.pool.QueryRow(context.Background(),
|
||||
`SELECT scrobbled_at FROM play_events WHERE id = $1`, s.playEvent).Scan(&v); err != nil {
|
||||
t.Fatalf("scrobbled_at: %v", err)
|
||||
}
|
||||
return v != nil
|
||||
}
|
||||
|
||||
func helperLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) }
|
||||
|
||||
func newTestWorker(s setup, lb *listenbrainz.Client) *Worker {
|
||||
return NewWorker(s.pool, lb, helperLogger())
|
||||
}
|
||||
|
||||
func TestWorker_Success_DeletesQueueRowAndStampsScrobbledAt(t *testing.T) {
|
||||
s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
|
||||
helperEnqueue(t, s)
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer srv.Close()
|
||||
lb := &listenbrainz.Client{BaseURL: srv.URL, HTTP: srv.Client()}
|
||||
w := newTestWorker(s, lb)
|
||||
|
||||
if err := w.tickOnce(context.Background()); err != nil {
|
||||
t.Fatalf("tickOnce: %v", err)
|
||||
}
|
||||
if helperPendingCount(t, s) != 0 {
|
||||
t.Error("queue row should be deleted on success")
|
||||
}
|
||||
if helperFailedCount(t, s) != 0 {
|
||||
t.Error("no failed row expected on success")
|
||||
}
|
||||
if !helperPlayEventScrobbledAt(t, s) {
|
||||
t.Error("play_events.scrobbled_at should be populated on success")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorker_503Once_RowRemainsAttempts1(t *testing.T) {
|
||||
s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
|
||||
helperEnqueue(t, s)
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
}))
|
||||
defer srv.Close()
|
||||
lb := &listenbrainz.Client{BaseURL: srv.URL, HTTP: srv.Client()}
|
||||
w := newTestWorker(s, lb)
|
||||
_ = w.tickOnce(context.Background())
|
||||
|
||||
var attempts int
|
||||
if err := s.pool.QueryRow(context.Background(),
|
||||
`SELECT attempts FROM scrobble_queue WHERE play_event_id = $1`, s.playEvent).Scan(&attempts); err != nil {
|
||||
t.Fatalf("attempts: %v", err)
|
||||
}
|
||||
if attempts != 1 {
|
||||
t.Errorf("attempts = %d, want 1 after 503", attempts)
|
||||
}
|
||||
if helperPendingCount(t, s) != 1 {
|
||||
t.Error("row should remain pending after transient failure")
|
||||
}
|
||||
}
|
||||
|
||||
// TestWorker_503SixTimes_MarksFailed: 6 ticks of 503. Tick 1-5 reschedule
|
||||
// (attempts goes 1,2,3,4,5; backoffDelay returns OK for those). Tick 6
|
||||
// hits backoffDelay(6) = false and marks failed (attempts stays at 5).
|
||||
func TestWorker_503SixTimes_MarksFailed(t *testing.T) {
|
||||
s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
|
||||
helperEnqueue(t, s)
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
}))
|
||||
defer srv.Close()
|
||||
lb := &listenbrainz.Client{BaseURL: srv.URL, HTTP: srv.Client()}
|
||||
w := newTestWorker(s, lb)
|
||||
for i := 0; i < 6; i++ {
|
||||
if _, err := s.pool.Exec(context.Background(),
|
||||
`UPDATE scrobble_queue SET next_attempt_at = now() WHERE play_event_id = $1`, s.playEvent); err != nil {
|
||||
t.Fatalf("reset: %v", err)
|
||||
}
|
||||
_ = w.tickOnce(context.Background())
|
||||
}
|
||||
if helperFailedCount(t, s) != 1 {
|
||||
t.Errorf("expected row marked failed after 6 attempts; pending=%d failed=%d",
|
||||
helperPendingCount(t, s), helperFailedCount(t, s))
|
||||
}
|
||||
var attempts int
|
||||
var lastErr *string
|
||||
_ = s.pool.QueryRow(context.Background(),
|
||||
`SELECT attempts, last_error FROM scrobble_queue WHERE play_event_id = $1`, s.playEvent).Scan(&attempts, &lastErr)
|
||||
if attempts != 5 {
|
||||
t.Errorf("attempts = %d, want 5 (max retries reached)", attempts)
|
||||
}
|
||||
if lastErr == nil || *lastErr == "" {
|
||||
t.Error("last_error should be populated")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorker_401_MarksFailedAndDisablesUser(t *testing.T) {
|
||||
s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
|
||||
helperEnqueue(t, s)
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
}))
|
||||
defer srv.Close()
|
||||
lb := &listenbrainz.Client{BaseURL: srv.URL, HTTP: srv.Client()}
|
||||
w := newTestWorker(s, lb)
|
||||
_ = w.tickOnce(context.Background())
|
||||
|
||||
if helperFailedCount(t, s) != 1 {
|
||||
t.Error("401 should mark failed immediately")
|
||||
}
|
||||
cfg, _ := s.q.GetListenBrainzConfig(context.Background(), s.user)
|
||||
if cfg.ListenbrainzEnabled {
|
||||
t.Error("user listenbrainz_enabled should be set to false on auth failure")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorker_429_RespectsRetryAfterWithoutIncrementingAttempts(t *testing.T) {
|
||||
s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
|
||||
helperEnqueue(t, s)
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Retry-After", "300")
|
||||
w.WriteHeader(http.StatusTooManyRequests)
|
||||
}))
|
||||
defer srv.Close()
|
||||
lb := &listenbrainz.Client{BaseURL: srv.URL, HTTP: srv.Client()}
|
||||
w := newTestWorker(s, lb)
|
||||
_ = w.tickOnce(context.Background())
|
||||
|
||||
var attempts int
|
||||
var nextAttemptAt time.Time
|
||||
_ = s.pool.QueryRow(context.Background(),
|
||||
`SELECT attempts, next_attempt_at FROM scrobble_queue WHERE play_event_id = $1`, s.playEvent).
|
||||
Scan(&attempts, &nextAttemptAt)
|
||||
if attempts != 0 {
|
||||
t.Errorf("attempts = %d, want 0 (429 should not increment)", attempts)
|
||||
}
|
||||
if d := time.Until(nextAttemptAt); d < 250*time.Second || d > 350*time.Second {
|
||||
t.Errorf("next_attempt_at = +%v, want ~+5m (Retry-After 300)", d)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWorker_BatchOfTen_OnePost(t *testing.T) {
|
||||
s := seed(t, seedOpts{lbToken: "tk", lbEnabled: true, durationMs: 250_000, ratio: 0.83})
|
||||
// Enqueue 10 distinct play_events to test batching. We synthesize new
|
||||
// play_events by copying the seeded one; each gets its own queue row.
|
||||
for i := 0; i < 10; i++ {
|
||||
var newPE pgtype.UUID
|
||||
if err := s.pool.QueryRow(context.Background(),
|
||||
`INSERT INTO play_events (user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped)
|
||||
SELECT user_id, track_id, session_id, now(), now(), duration_played_ms, completion_ratio, was_skipped FROM play_events WHERE id = $1
|
||||
RETURNING id`, s.playEvent).Scan(&newPE); err != nil {
|
||||
t.Fatalf("dup play_event: %v", err)
|
||||
}
|
||||
if _, err := s.pool.Exec(context.Background(),
|
||||
`INSERT INTO scrobble_queue (user_id, play_event_id, status) VALUES ($1, $2, 'pending')`,
|
||||
s.user, newPE); err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
}
|
||||
var posts atomic.Int32
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
posts.Add(1)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer srv.Close()
|
||||
lb := &listenbrainz.Client{BaseURL: srv.URL, HTTP: srv.Client()}
|
||||
w := newTestWorker(s, lb)
|
||||
_ = w.tickOnce(context.Background())
|
||||
|
||||
if posts.Load() != 1 {
|
||||
t.Errorf("LB POSTs = %d, want 1 (batch)", posts.Load())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
package scrobble
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestBackoffDelay(t *testing.T) {
|
||||
cases := []struct {
|
||||
attempts int
|
||||
want time.Duration
|
||||
ok bool
|
||||
}{
|
||||
{1, 1 * time.Minute, true},
|
||||
{2, 5 * time.Minute, true},
|
||||
{3, 30 * time.Minute, true},
|
||||
{4, 2 * time.Hour, true},
|
||||
{5, 6 * time.Hour, true},
|
||||
{6, 0, false},
|
||||
{99, 0, false},
|
||||
}
|
||||
for _, c := range cases {
|
||||
got, ok := backoffDelay(c.attempts)
|
||||
if ok != c.ok {
|
||||
t.Errorf("backoffDelay(%d) ok = %v, want %v", c.attempts, ok, c.ok)
|
||||
}
|
||||
if got != c.want {
|
||||
t.Errorf("backoffDelay(%d) = %v, want %v", c.attempts, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -45,6 +45,8 @@ export const api = {
|
||||
get: <T>(path: string): Promise<T> => apiFetch(path) as Promise<T>,
|
||||
post: <T>(path: string, body: unknown): Promise<T> =>
|
||||
apiFetch(path, { method: 'POST', body: JSON.stringify(body) }) as Promise<T>,
|
||||
put: <T>(path: string, body: unknown): Promise<T> =>
|
||||
apiFetch(path, { method: 'PUT', body: JSON.stringify(body) }) as Promise<T>,
|
||||
del: (path: string): Promise<null> =>
|
||||
apiFetch(path, { method: 'DELETE' }) as Promise<null>
|
||||
};
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
import { createQuery, createMutation } from '@tanstack/svelte-query';
|
||||
import type { QueryClient } from '@tanstack/svelte-query';
|
||||
import { api } from './client';
|
||||
|
||||
export type LBStatus = {
|
||||
enabled: boolean;
|
||||
token_set: boolean;
|
||||
last_scrobbled_at: string | null;
|
||||
};
|
||||
|
||||
export function getListenBrainzStatus(): Promise<LBStatus> {
|
||||
return api.get<LBStatus>('/api/me/listenbrainz');
|
||||
}
|
||||
|
||||
export function setListenBrainzToken(token: string): Promise<LBStatus> {
|
||||
return api.put<LBStatus>('/api/me/listenbrainz', { token });
|
||||
}
|
||||
|
||||
export function setListenBrainzEnabled(enabled: boolean): Promise<LBStatus> {
|
||||
return api.put<LBStatus>('/api/me/listenbrainz', { enabled });
|
||||
}
|
||||
|
||||
export const LB_QUERY_KEY = ['settings', 'listenbrainz'] as const;
|
||||
|
||||
export function createLBStatusQuery() {
|
||||
return createQuery({
|
||||
queryKey: LB_QUERY_KEY,
|
||||
queryFn: getListenBrainzStatus
|
||||
});
|
||||
}
|
||||
|
||||
export function createTokenMutation(queryClient: QueryClient) {
|
||||
return createMutation<LBStatus, Error, string>({
|
||||
mutationFn: (token: string) => setListenBrainzToken(token),
|
||||
onSuccess: () => queryClient.invalidateQueries({ queryKey: LB_QUERY_KEY })
|
||||
});
|
||||
}
|
||||
|
||||
export function createEnabledMutation(queryClient: QueryClient) {
|
||||
return createMutation<LBStatus, Error, boolean>({
|
||||
mutationFn: (enabled: boolean) => setListenBrainzEnabled(enabled),
|
||||
onSuccess: () => queryClient.invalidateQueries({ queryKey: LB_QUERY_KEY })
|
||||
});
|
||||
}
|
||||
@@ -25,7 +25,8 @@
|
||||
{ href: '/', label: 'Library' },
|
||||
{ href: '/library/liked', label: 'Liked' },
|
||||
{ href: '/search', label: 'Search' },
|
||||
{ href: '/playlists', label: 'Playlists' }
|
||||
{ href: '/playlists', label: 'Playlists' },
|
||||
{ href: '/settings', label: 'Settings' }
|
||||
];
|
||||
</script>
|
||||
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
<script lang="ts">
|
||||
import { useQueryClient } from '@tanstack/svelte-query';
|
||||
import type { CreateQueryResult, CreateMutationResult } from '@tanstack/svelte-query';
|
||||
import {
|
||||
createLBStatusQuery,
|
||||
createTokenMutation,
|
||||
createEnabledMutation,
|
||||
type LBStatus
|
||||
} from '$lib/api/listenbrainz';
|
||||
|
||||
const queryClient = useQueryClient();
|
||||
|
||||
const status = createLBStatusQuery() as CreateQueryResult<LBStatus>;
|
||||
const tokenMutation = createTokenMutation(queryClient);
|
||||
const enabledMutation = createEnabledMutation(queryClient);
|
||||
|
||||
let tokenInput = $state('');
|
||||
|
||||
function saveToken() {
|
||||
const t = tokenInput.trim();
|
||||
if (!t) return;
|
||||
$tokenMutation.mutate(t, { onSuccess: () => { tokenInput = ''; } });
|
||||
}
|
||||
|
||||
function clearToken() {
|
||||
$tokenMutation.mutate('');
|
||||
}
|
||||
|
||||
function toggleEnabled() {
|
||||
if (!$status.data) return;
|
||||
$enabledMutation.mutate(!$status.data.enabled);
|
||||
}
|
||||
</script>
|
||||
|
||||
<div class="space-y-6">
|
||||
<h1 class="text-2xl font-semibold">Settings</h1>
|
||||
|
||||
<section class="space-y-4 rounded border border-border bg-surface p-4">
|
||||
<h2 class="text-lg font-semibold">ListenBrainz</h2>
|
||||
|
||||
{#if $status.isPending}
|
||||
<p class="text-text-secondary">Loading…</p>
|
||||
{:else if $status.isError}
|
||||
<p class="text-text-secondary">Couldn't load status.</p>
|
||||
{:else if $status.data}
|
||||
<div class="space-y-3">
|
||||
<label class="block text-sm">
|
||||
<span class="block text-text-secondary mb-1">User token</span>
|
||||
{#if $status.data.token_set}
|
||||
<div class="flex items-center gap-2">
|
||||
<span class="text-text-primary">••••••••••• (set)</span>
|
||||
<button
|
||||
type="button"
|
||||
onclick={clearToken}
|
||||
class="text-sm text-accent hover:underline"
|
||||
disabled={$tokenMutation.isPending}
|
||||
>clear</button>
|
||||
</div>
|
||||
{:else}
|
||||
<div class="flex gap-2">
|
||||
<input
|
||||
type="password"
|
||||
bind:value={tokenInput}
|
||||
placeholder="paste your LB token"
|
||||
class="flex-1 rounded border border-border bg-background px-2 py-1 text-sm"
|
||||
/>
|
||||
<button
|
||||
type="button"
|
||||
onclick={saveToken}
|
||||
disabled={!tokenInput.trim() || $tokenMutation.isPending}
|
||||
class="rounded bg-accent px-3 py-1 text-sm text-background disabled:opacity-50"
|
||||
>Save</button>
|
||||
</div>
|
||||
{/if}
|
||||
</label>
|
||||
|
||||
<label class="flex items-center gap-2 text-sm">
|
||||
<input
|
||||
type="checkbox"
|
||||
checked={$status.data.enabled}
|
||||
disabled={!$status.data.token_set || $enabledMutation.isPending}
|
||||
onchange={toggleEnabled}
|
||||
/>
|
||||
<span>Send my plays to ListenBrainz</span>
|
||||
</label>
|
||||
|
||||
{#if $status.data.last_scrobbled_at}
|
||||
<p class="text-sm text-text-secondary">
|
||||
Last scrobbled: {new Date($status.data.last_scrobbled_at).toLocaleString()}
|
||||
</p>
|
||||
{/if}
|
||||
|
||||
<p class="text-xs text-text-secondary">
|
||||
Get a token at <a href="https://listenbrainz.org/profile/" target="_blank" rel="noopener" class="text-accent hover:underline">listenbrainz.org/profile</a>.
|
||||
Tokens are stored unencrypted in this server's database — treat as sensitive.
|
||||
</p>
|
||||
</div>
|
||||
{/if}
|
||||
</section>
|
||||
</div>
|
||||
@@ -0,0 +1,102 @@
|
||||
import { afterEach, describe, expect, test, vi } from 'vitest';
|
||||
import { render, screen, fireEvent, waitFor } from '@testing-library/svelte';
|
||||
import { readable, writable } from 'svelte/store';
|
||||
import type { LBStatus } from '$lib/api/listenbrainz';
|
||||
|
||||
vi.mock('$lib/api/listenbrainz', () => ({
|
||||
createLBStatusQuery: vi.fn(),
|
||||
createTokenMutation: vi.fn(),
|
||||
createEnabledMutation: vi.fn(),
|
||||
setListenBrainzToken: vi.fn(),
|
||||
setListenBrainzEnabled: vi.fn()
|
||||
}));
|
||||
|
||||
vi.mock('@tanstack/svelte-query', async (orig) => {
|
||||
const actual = (await orig()) as Record<string, unknown>;
|
||||
return {
|
||||
...actual,
|
||||
useQueryClient: () => ({ invalidateQueries: vi.fn() })
|
||||
};
|
||||
});
|
||||
|
||||
import SettingsPage from './+page.svelte';
|
||||
import {
|
||||
createLBStatusQuery,
|
||||
createTokenMutation,
|
||||
createEnabledMutation,
|
||||
setListenBrainzToken,
|
||||
setListenBrainzEnabled
|
||||
} from '$lib/api/listenbrainz';
|
||||
|
||||
function mockStatusStore(data: LBStatus) {
|
||||
return readable({ data, isPending: false, isError: false });
|
||||
}
|
||||
|
||||
function mockMutationStore(mutateFn: (vars: unknown) => void = vi.fn()) {
|
||||
return readable({ isPending: false, mutate: mutateFn, mutateAsync: vi.fn() });
|
||||
}
|
||||
|
||||
afterEach(() => vi.clearAllMocks());
|
||||
|
||||
describe('Settings page — ListenBrainz', () => {
|
||||
test('token-not-set state renders input + Save button', async () => {
|
||||
(createLBStatusQuery as ReturnType<typeof vi.fn>).mockReturnValue(
|
||||
mockStatusStore({ enabled: false, token_set: false, last_scrobbled_at: null })
|
||||
);
|
||||
(createTokenMutation as ReturnType<typeof vi.fn>).mockReturnValue(mockMutationStore());
|
||||
(createEnabledMutation as ReturnType<typeof vi.fn>).mockReturnValue(mockMutationStore());
|
||||
render(SettingsPage);
|
||||
await waitFor(() =>
|
||||
expect(screen.getByPlaceholderText(/paste your lb token/i)).toBeInTheDocument()
|
||||
);
|
||||
expect(screen.getByRole('button', { name: /save/i })).toBeInTheDocument();
|
||||
});
|
||||
|
||||
test('token-set state renders masked + Clear button', async () => {
|
||||
(createLBStatusQuery as ReturnType<typeof vi.fn>).mockReturnValue(
|
||||
mockStatusStore({ enabled: false, token_set: true, last_scrobbled_at: null })
|
||||
);
|
||||
(createTokenMutation as ReturnType<typeof vi.fn>).mockReturnValue(mockMutationStore());
|
||||
(createEnabledMutation as ReturnType<typeof vi.fn>).mockReturnValue(mockMutationStore());
|
||||
render(SettingsPage);
|
||||
await waitFor(() => expect(screen.getByText(/\(set\)/)).toBeInTheDocument());
|
||||
expect(screen.getByText('clear')).toBeInTheDocument();
|
||||
});
|
||||
|
||||
test('Save button calls setListenBrainzToken with typed value', async () => {
|
||||
(createLBStatusQuery as ReturnType<typeof vi.fn>).mockReturnValue(
|
||||
mockStatusStore({ enabled: false, token_set: false, last_scrobbled_at: null })
|
||||
);
|
||||
const mutateFn = vi.fn();
|
||||
(createTokenMutation as ReturnType<typeof vi.fn>).mockReturnValue(mockMutationStore(mutateFn));
|
||||
(createEnabledMutation as ReturnType<typeof vi.fn>).mockReturnValue(mockMutationStore());
|
||||
render(SettingsPage);
|
||||
const input = await screen.findByPlaceholderText(/paste your lb token/i);
|
||||
await fireEvent.input(input, { target: { value: 'mytoken' } });
|
||||
await fireEvent.click(screen.getByRole('button', { name: /save/i }));
|
||||
await waitFor(() => expect(mutateFn).toHaveBeenCalledWith('mytoken', expect.anything()));
|
||||
});
|
||||
|
||||
test('Enabled checkbox is disabled when no token', async () => {
|
||||
(createLBStatusQuery as ReturnType<typeof vi.fn>).mockReturnValue(
|
||||
mockStatusStore({ enabled: false, token_set: false, last_scrobbled_at: null })
|
||||
);
|
||||
(createTokenMutation as ReturnType<typeof vi.fn>).mockReturnValue(mockMutationStore());
|
||||
(createEnabledMutation as ReturnType<typeof vi.fn>).mockReturnValue(mockMutationStore());
|
||||
render(SettingsPage);
|
||||
const checkbox = await screen.findByRole('checkbox');
|
||||
expect(checkbox).toBeDisabled();
|
||||
});
|
||||
|
||||
test('Last-scrobbled-at renders when present', async () => {
|
||||
(createLBStatusQuery as ReturnType<typeof vi.fn>).mockReturnValue(
|
||||
mockStatusStore({ enabled: true, token_set: true, last_scrobbled_at: '2026-04-28T03:00:00Z' })
|
||||
);
|
||||
(createTokenMutation as ReturnType<typeof vi.fn>).mockReturnValue(mockMutationStore());
|
||||
(createEnabledMutation as ReturnType<typeof vi.fn>).mockReturnValue(mockMutationStore());
|
||||
render(SettingsPage);
|
||||
await waitFor(() =>
|
||||
expect(screen.getByText(/last scrobbled:/i)).toBeInTheDocument()
|
||||
);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user