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