diff --git a/internal/scrobble/queue.go b/internal/scrobble/queue.go new file mode 100644 index 00000000..3e814be4 --- /dev/null +++ b/internal/scrobble/queue.go @@ -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 +} diff --git a/internal/scrobble/queue_test.go b/internal/scrobble/queue_test.go new file mode 100644 index 00000000..86cd3943 --- /dev/null +++ b/internal/scrobble/queue_test.go @@ -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) + } +}