diff --git a/internal/playevents/writer.go b/internal/playevents/writer.go index 14350f48..00bff645 100644 --- a/internal/playevents/writer.go +++ b/internal/playevents/writer.go @@ -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 diff --git a/internal/playevents/writer_test.go b/internal/playevents/writer_test.go index 5786e826..ac3876ec 100644 --- a/internal/playevents/writer_test.go +++ b/internal/playevents/writer_test.go @@ -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) + } +}