feat(scrobble): add MaybeEnqueue with idempotent insert + threshold gate
Bridges closed play_events to the scrobble_queue: loads the event, applies the Qualifies() threshold, checks user LB config (enabled + token), and inserts via ON CONFLICT DO NOTHING for idempotency. Best-effort — callers (playevents.Writer hooks, Task 8) will log and swallow errors. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user