feat(playevents): persist session_vector_at_play on every record path
RecordPlayStarted and RecordSyntheticCompletedPlay both capture the session vector inside their existing transactions. Single private helper queries the prior 5 tracks, builds the vector, UPDATEs the just-inserted play_event. Tests verify the seed flag boundary and session-scope isolation.
This commit is contained in:
@@ -9,6 +9,7 @@ package playevents
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"time"
|
||||
@@ -19,6 +20,7 @@ import (
|
||||
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/playsessions"
|
||||
"git.fabledsword.com/bvandeusen/minstrel/internal/recommendation"
|
||||
)
|
||||
|
||||
type Writer struct {
|
||||
@@ -87,6 +89,11 @@ func (w *Writer) RecordPlayStarted(
|
||||
}
|
||||
out.PlayEventID = ev.ID
|
||||
out.SessionID = sessionID
|
||||
|
||||
// Capture session vector for the just-inserted row.
|
||||
if err := w.captureSessionVector(ctx, q, ev.ID, sessionID, at); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return out, err
|
||||
@@ -223,6 +230,9 @@ func (w *Writer) RecordSyntheticCompletedPlay(
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := w.captureSessionVector(ctx, q, ev.ID, sessionID, at); err != nil {
|
||||
return err
|
||||
}
|
||||
ratio := 1.0
|
||||
dur := track.DurationMs
|
||||
_, err = q.UpdatePlayEventEnded(ctx, dbq.UpdatePlayEventEndedParams{
|
||||
@@ -236,6 +246,36 @@ func (w *Writer) RecordSyntheticCompletedPlay(
|
||||
})
|
||||
}
|
||||
|
||||
// captureSessionVector queries the user's prior plays in the given session
|
||||
// (before `at`), builds the session vector, and UPDATEs the just-inserted
|
||||
// play_event with it. Runs inside the caller's transaction (q is the
|
||||
// transaction-bound *dbq.Queries). Errors here propagate — vector capture
|
||||
// is best-effort, but DB errors should fail the transaction.
|
||||
func (w *Writer) captureSessionVector(
|
||||
ctx context.Context,
|
||||
q *dbq.Queries,
|
||||
playEventID, sessionID pgtype.UUID,
|
||||
at time.Time,
|
||||
) error {
|
||||
priorTracks, err := q.ListRecentSessionTracks(ctx, dbq.ListRecentSessionTracksParams{
|
||||
SessionID: sessionID,
|
||||
StartedAt: pgtype.Timestamptz{Time: at, Valid: true},
|
||||
Limit: 5,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
vec := recommendation.BuildSessionVector(priorTracks)
|
||||
vecJSON, err := json.Marshal(vec)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return q.UpdatePlayEventVector(ctx, dbq.UpdatePlayEventVectorParams{
|
||||
ID: playEventID,
|
||||
SessionVectorAtPlay: vecJSON,
|
||||
})
|
||||
}
|
||||
|
||||
// autoClosePriorOpen closes any open (ended_at IS NULL) play_event for the
|
||||
// user. Sets ended_at = at, duration_played_ms = min(at - started_at, track
|
||||
// duration), was_skipped = true. Skip rule is NOT applied — auto-closed
|
||||
|
||||
@@ -2,6 +2,7 @@ package playevents
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
@@ -234,3 +235,85 @@ func TestRecordSyntheticCompletedPlay_WritesBothRows(t *testing.T) {
|
||||
t.Errorf("closed play_events count = %d, want 1", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecordPlayStarted_PersistsSessionVector_Seed(t *testing.T) {
|
||||
f := newFixture(t, 200_000)
|
||||
now := time.Now().UTC()
|
||||
res, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", now)
|
||||
if err != nil {
|
||||
t.Fatalf("RecordPlayStarted: %v", err)
|
||||
}
|
||||
got, err := f.q.GetPlayEventByID(context.Background(), res.PlayEventID)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if got.SessionVectorAtPlay == nil {
|
||||
t.Fatalf("session_vector_at_play is NULL")
|
||||
}
|
||||
var vec map[string]any
|
||||
if err := json.Unmarshal(got.SessionVectorAtPlay, &vec); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if seed, _ := vec["seed"].(bool); !seed {
|
||||
t.Errorf("seed = %v, want true (first play in session)", vec["seed"])
|
||||
}
|
||||
if artists, _ := vec["artists"].([]any); len(artists) != 0 {
|
||||
t.Errorf("artists not empty: %v", artists)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecordPlayStarted_PersistsSessionVector_Populated(t *testing.T) {
|
||||
f := newFixture(t, 200_000)
|
||||
// Seed 4 prior plays with the same track (reusing the fixture's single
|
||||
// track is fine — vector dedup means we'll see 1 artist regardless).
|
||||
t0 := time.Now().UTC().Add(-1 * time.Hour)
|
||||
for i := 0; i < 4; i++ {
|
||||
at := t0.Add(time.Duration(i) * time.Minute)
|
||||
_, _ = f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", at)
|
||||
}
|
||||
// 5th play — should see vector with 4 prior tracks.
|
||||
at := t0.Add(10 * time.Minute)
|
||||
res, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", at)
|
||||
if err != nil {
|
||||
t.Fatalf("RecordPlayStarted: %v", err)
|
||||
}
|
||||
got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID)
|
||||
if got.SessionVectorAtPlay == nil {
|
||||
t.Fatal("session_vector_at_play is NULL")
|
||||
}
|
||||
var vec map[string]any
|
||||
_ = json.Unmarshal(got.SessionVectorAtPlay, &vec)
|
||||
if seed, _ := vec["seed"].(bool); seed {
|
||||
t.Errorf("seed = true after 4 prior tracks, want false")
|
||||
}
|
||||
recentIDs, _ := vec["recent_track_ids"].([]any)
|
||||
if len(recentIDs) != 4 {
|
||||
t.Errorf("recent_track_ids len = %d, want 4", len(recentIDs))
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecordPlayStarted_VectorScopedToSession(t *testing.T) {
|
||||
f := newFixture(t, 200_000)
|
||||
// Play 1: at t=0.
|
||||
t0 := time.Now().UTC().Add(-2 * time.Hour)
|
||||
_, _ = f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t0)
|
||||
// Play 2: t = t0 + 31 minutes (exceeds 30-min session timeout).
|
||||
t1 := t0.Add(31 * time.Minute)
|
||||
res, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1)
|
||||
if err != nil {
|
||||
t.Fatalf("RecordPlayStarted: %v", err)
|
||||
}
|
||||
got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID)
|
||||
if got.SessionVectorAtPlay == nil {
|
||||
t.Fatal("vector NULL")
|
||||
}
|
||||
var vec map[string]any
|
||||
_ = json.Unmarshal(got.SessionVectorAtPlay, &vec)
|
||||
if seed, _ := vec["seed"].(bool); !seed {
|
||||
t.Errorf("seed = false in fresh session, want true")
|
||||
}
|
||||
recentIDs, _ := vec["recent_track_ids"].([]any)
|
||||
if len(recentIDs) != 0 {
|
||||
t.Errorf("recent_track_ids has %d entries from prior session, want 0", len(recentIDs))
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user