docs(spec): add M4a outbound scrobble worker design

First M4 sub-plan (Fable #345). Outbound scrobble worker with batching + exponential-backoff retry. Per-user token, plaintext storage, pull-based 30s timer worker, patient backoff (1m → 5m → 30m → 2h → 6h, 5 attempts). New scrobble_queue table; sent rows deleted (canonical record stays in play_events.scrobbled_at). Minimal /settings page in M4a as scaffold for M6's full Settings sub-plan to extend. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-04-28 08:21:43 -04:00
parent 4f67dd6b0a
commit 791c6c7fd6
1 changed files with 403 additions and 0 deletions
@@ -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).