feat(playevents): post-commit MaybeEnqueue in RecordPlayEnded + Synthetic
This commit is contained in:
@@ -21,6 +21,7 @@ import (
|
|||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/playsessions"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/playsessions"
|
||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/recommendation"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/recommendation"
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/scrobble"
|
||||||
)
|
)
|
||||||
|
|
||||||
type Writer struct {
|
type Writer struct {
|
||||||
@@ -109,7 +110,7 @@ func (w *Writer) RecordPlayEnded(
|
|||||||
durationPlayedMs int32,
|
durationPlayedMs int32,
|
||||||
at time.Time,
|
at time.Time,
|
||||||
) error {
|
) 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)
|
q := dbq.New(tx)
|
||||||
ev, err := q.GetPlayEventByID(ctx, playEventID)
|
ev, err := q.GetPlayEventByID(ctx, playEventID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -147,7 +148,16 @@ func (w *Writer) RecordPlayEnded(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
return nil
|
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
|
// RecordPlaySkipped is the user-initiated-skip variant: was_skipped is forced
|
||||||
@@ -203,7 +213,8 @@ func (w *Writer) RecordSyntheticCompletedPlay(
|
|||||||
clientID string,
|
clientID string,
|
||||||
at time.Time,
|
at time.Time,
|
||||||
) error {
|
) 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)
|
q := dbq.New(tx)
|
||||||
track, err := q.GetTrackByID(ctx, trackID)
|
track, err := q.GetTrackByID(ctx, trackID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -230,6 +241,7 @@ func (w *Writer) RecordSyntheticCompletedPlay(
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
playEventID = ev.ID
|
||||||
if err := w.captureSessionVector(ctx, q, ev.ID, sessionID, at); err != nil {
|
if err := w.captureSessionVector(ctx, q, ev.ID, sessionID, at); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -243,7 +255,16 @@ func (w *Writer) RecordSyntheticCompletedPlay(
|
|||||||
WasSkipped: false,
|
WasSkipped: false,
|
||||||
})
|
})
|
||||||
return err
|
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
|
// 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)
|
t.Cleanup(pool.Close)
|
||||||
if _, err := pool.Exec(context.Background(),
|
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)
|
t.Fatalf("truncate: %v", err)
|
||||||
}
|
}
|
||||||
return pool
|
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))
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user