From 0c7e1b040886f171670e66352816cbcc0a1c11f2 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 25 Apr 2026 22:32:33 -0400 Subject: [PATCH] feat(playevents): add Writer with start/end/skipped/synthetic paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Owns the auto-close-prior step (caps each user at one open row), spec §6 skip classification rule on play_ended (AND of completion < 0.5 and duration_played < 30s), and skip_events row writes. Synthetic completed play wires the Subsonic /rest/scrobble?submission=true path into the same store as native events. --- internal/db/dbq/albums.sql.go | 2 +- internal/db/dbq/artists.sql.go | 2 +- internal/db/dbq/db.go | 2 +- internal/db/dbq/models.go | 36 +++- internal/db/dbq/sessions.sql.go | 2 +- internal/db/dbq/tracks.sql.go | 2 +- internal/db/dbq/users.sql.go | 2 +- internal/playevents/writer.go | 286 +++++++++++++++++++++++++++++ internal/playevents/writer_test.go | 236 ++++++++++++++++++++++++ 9 files changed, 563 insertions(+), 7 deletions(-) create mode 100644 internal/playevents/writer.go create mode 100644 internal/playevents/writer_test.go diff --git a/internal/db/dbq/albums.sql.go b/internal/db/dbq/albums.sql.go index f94956b6..acbdeb5b 100644 --- a/internal/db/dbq/albums.sql.go +++ b/internal/db/dbq/albums.sql.go @@ -1,6 +1,6 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.27.0 +// sqlc v1.31.1 // source: albums.sql package dbq diff --git a/internal/db/dbq/artists.sql.go b/internal/db/dbq/artists.sql.go index 259efbc4..cafeb6b5 100644 --- a/internal/db/dbq/artists.sql.go +++ b/internal/db/dbq/artists.sql.go @@ -1,6 +1,6 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.27.0 +// sqlc v1.31.1 // source: artists.sql package dbq diff --git a/internal/db/dbq/db.go b/internal/db/dbq/db.go index 8650b40f..1890abd0 100644 --- a/internal/db/dbq/db.go +++ b/internal/db/dbq/db.go @@ -1,6 +1,6 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.27.0 +// sqlc v1.31.1 package dbq diff --git a/internal/db/dbq/models.go b/internal/db/dbq/models.go index 57410eda..1783cd70 100644 --- a/internal/db/dbq/models.go +++ b/internal/db/dbq/models.go @@ -1,6 +1,6 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.27.0 +// sqlc v1.31.1 package dbq @@ -29,6 +29,31 @@ type Artist struct { UpdatedAt pgtype.Timestamptz } +type PlayEvent struct { + ID pgtype.UUID + UserID pgtype.UUID + TrackID pgtype.UUID + SessionID pgtype.UUID + StartedAt pgtype.Timestamptz + EndedAt pgtype.Timestamptz + DurationPlayedMs *int32 + CompletionRatio *float64 + WasSkipped bool + ClientID *string + SessionVectorAtPlay []byte + ScrobbledAt pgtype.Timestamptz +} + +type PlaySession struct { + ID pgtype.UUID + UserID pgtype.UUID + StartedAt pgtype.Timestamptz + EndedAt pgtype.Timestamptz + LastEventAt pgtype.Timestamptz + TrackCount int32 + ClientID *string +} + type Session struct { ID pgtype.UUID UserID pgtype.UUID @@ -38,6 +63,15 @@ type Session struct { LastSeenAt pgtype.Timestamptz } +type SkipEvent struct { + ID pgtype.UUID + UserID pgtype.UUID + TrackID pgtype.UUID + SessionID pgtype.UUID + SkippedAt pgtype.Timestamptz + PositionMs int32 +} + type Track struct { ID pgtype.UUID Title string diff --git a/internal/db/dbq/sessions.sql.go b/internal/db/dbq/sessions.sql.go index 3ac9741c..bc34671b 100644 --- a/internal/db/dbq/sessions.sql.go +++ b/internal/db/dbq/sessions.sql.go @@ -1,6 +1,6 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.27.0 +// sqlc v1.31.1 // source: sessions.sql package dbq diff --git a/internal/db/dbq/tracks.sql.go b/internal/db/dbq/tracks.sql.go index 5254bb50..d17447c9 100644 --- a/internal/db/dbq/tracks.sql.go +++ b/internal/db/dbq/tracks.sql.go @@ -1,6 +1,6 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.27.0 +// sqlc v1.31.1 // source: tracks.sql package dbq diff --git a/internal/db/dbq/users.sql.go b/internal/db/dbq/users.sql.go index 2e7dccbf..c3b5f3d9 100644 --- a/internal/db/dbq/users.sql.go +++ b/internal/db/dbq/users.sql.go @@ -1,6 +1,6 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.27.0 +// sqlc v1.31.1 // source: users.sql package dbq diff --git a/internal/playevents/writer.go b/internal/playevents/writer.go new file mode 100644 index 00000000..e8b73bb8 --- /dev/null +++ b/internal/playevents/writer.go @@ -0,0 +1,286 @@ +// Package playevents owns the four write paths into play_events / skip_events: +// RecordPlayStarted, RecordPlayEnded, RecordPlaySkipped, and +// RecordSyntheticCompletedPlay (used by the Subsonic /rest/scrobble shim). +// +// All paths run inside a single transaction. The auto-close-prior step on +// RecordPlayStarted bounds each user to at most one open play_event at any +// time (per spec data-flow section). +package playevents + +import ( + "context" + "errors" + "log/slog" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + "git.fabledsword.com/bvandeusen/minstrel/internal/playsessions" +) + +type Writer struct { + pool *pgxpool.Pool + logger *slog.Logger + sessionTimeout time.Duration + skipMaxCompletionRatio float64 + skipMaxDurationPlayedMs int32 +} + +// NewWriter constructs a writer. timeout is the play-session window; the two +// remaining args are the skip rule thresholds from spec §6. +func NewWriter( + pool *pgxpool.Pool, + logger *slog.Logger, + sessionTimeout time.Duration, + skipMaxCompletionRatio float64, + skipMaxDurationPlayedMs int, +) *Writer { + return &Writer{ + pool: pool, + logger: logger, + sessionTimeout: sessionTimeout, + skipMaxCompletionRatio: skipMaxCompletionRatio, + skipMaxDurationPlayedMs: int32(skipMaxDurationPlayedMs), + } +} + +type StartedResult struct { + PlayEventID pgtype.UUID + SessionID pgtype.UUID +} + +// RecordPlayStarted: in one transaction, auto-close any prior open row for +// the user, FindOrCreate the play_session, insert the new play_event, +// return its id and session id. +func (w *Writer) RecordPlayStarted( + ctx context.Context, + userID, trackID pgtype.UUID, + clientID string, + at time.Time, +) (StartedResult, error) { + var out StartedResult + err := pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error { + q := dbq.New(tx) + if err := w.autoClosePriorOpen(ctx, q, userID, at); err != nil { + return err + } + sessionID, err := playsessions.FindOrCreate(ctx, q, userID, at, clientID, w.sessionTimeout) + if err != nil { + return err + } + var clientIDPtr *string + if clientID != "" { + clientIDPtr = &clientID + } + ev, err := q.InsertPlayEvent(ctx, dbq.InsertPlayEventParams{ + UserID: userID, + TrackID: trackID, + SessionID: sessionID, + StartedAt: pgtype.Timestamptz{Time: at, Valid: true}, + ClientID: clientIDPtr, + }) + if err != nil { + return err + } + out.PlayEventID = ev.ID + out.SessionID = sessionID + return nil + }) + return out, err +} + +// RecordPlayEnded closes a play_event with duration + computed completion +// ratio. Applies the spec §6 skip rule (AND of completion < threshold AND +// duration_played < threshold) — both conditions must hold for the row to +// be marked was_skipped=true. If skipped, also writes a skip_events row. +func (w *Writer) RecordPlayEnded( + ctx context.Context, + playEventID pgtype.UUID, + durationPlayedMs int32, + at time.Time, +) error { + return pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error { + q := dbq.New(tx) + ev, err := q.GetPlayEventByID(ctx, playEventID) + if err != nil { + return err + } + track, err := q.GetTrackByID(ctx, ev.TrackID) + if err != nil { + return err + } + ratio := 0.0 + if track.DurationMs > 0 { + ratio = float64(durationPlayedMs) / float64(track.DurationMs) + } + isSkip := ratio < w.skipMaxCompletionRatio && durationPlayedMs < w.skipMaxDurationPlayedMs + ratioPtr := ratio + durPtr := durationPlayedMs + if _, err := q.UpdatePlayEventEnded(ctx, dbq.UpdatePlayEventEndedParams{ + ID: playEventID, + EndedAt: pgtype.Timestamptz{Time: at, Valid: true}, + DurationPlayedMs: &durPtr, + CompletionRatio: &ratioPtr, + WasSkipped: isSkip, + }); err != nil { + return err + } + if isSkip { + if _, err := q.InsertSkipEvent(ctx, dbq.InsertSkipEventParams{ + UserID: ev.UserID, + TrackID: ev.TrackID, + SessionID: ev.SessionID, + SkippedAt: pgtype.Timestamptz{Time: at, Valid: true}, + PositionMs: durationPlayedMs, + }); err != nil { + return err + } + } + return nil + }) +} + +// RecordPlaySkipped is the user-initiated-skip variant: was_skipped is forced +// true regardless of the rule, and a skip_events row is always written. +func (w *Writer) RecordPlaySkipped( + ctx context.Context, + playEventID pgtype.UUID, + positionMs int32, + at time.Time, +) error { + return pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error { + q := dbq.New(tx) + ev, err := q.GetPlayEventByID(ctx, playEventID) + if err != nil { + return err + } + track, err := q.GetTrackByID(ctx, ev.TrackID) + if err != nil { + return err + } + ratio := 0.0 + if track.DurationMs > 0 { + ratio = float64(positionMs) / float64(track.DurationMs) + } + ratioPtr := ratio + durPtr := positionMs + if _, err := q.UpdatePlayEventEnded(ctx, dbq.UpdatePlayEventEndedParams{ + ID: playEventID, + EndedAt: pgtype.Timestamptz{Time: at, Valid: true}, + DurationPlayedMs: &durPtr, + CompletionRatio: &ratioPtr, + WasSkipped: true, + }); err != nil { + return err + } + _, err = q.InsertSkipEvent(ctx, dbq.InsertSkipEventParams{ + UserID: ev.UserID, + TrackID: ev.TrackID, + SessionID: ev.SessionID, + SkippedAt: pgtype.Timestamptz{Time: at, Valid: true}, + PositionMs: positionMs, + }) + return err + }) +} + +// RecordSyntheticCompletedPlay writes a start+end pair for a Subsonic +// scrobble?submission=true call. ended_at = at + track.duration_ms, +// duration_played_ms = track.duration_ms, was_skipped = false. +func (w *Writer) RecordSyntheticCompletedPlay( + ctx context.Context, + userID, trackID pgtype.UUID, + clientID string, + at time.Time, +) error { + return pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error { + q := dbq.New(tx) + track, err := q.GetTrackByID(ctx, trackID) + if err != nil { + return err + } + if err := w.autoClosePriorOpen(ctx, q, userID, at); err != nil { + return err + } + sessionID, err := playsessions.FindOrCreate(ctx, q, userID, at, clientID, w.sessionTimeout) + if err != nil { + return err + } + var clientIDPtr *string + if clientID != "" { + clientIDPtr = &clientID + } + ev, err := q.InsertPlayEvent(ctx, dbq.InsertPlayEventParams{ + UserID: userID, + TrackID: trackID, + SessionID: sessionID, + StartedAt: pgtype.Timestamptz{Time: at, Valid: true}, + ClientID: clientIDPtr, + }) + if err != nil { + return err + } + ratio := 1.0 + dur := track.DurationMs + _, err = q.UpdatePlayEventEnded(ctx, dbq.UpdatePlayEventEndedParams{ + ID: ev.ID, + EndedAt: pgtype.Timestamptz{Time: at.Add(time.Duration(track.DurationMs) * time.Millisecond), Valid: true}, + DurationPlayedMs: &dur, + CompletionRatio: &ratio, + WasSkipped: false, + }) + return err + }) +} + +// 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 +// rows are flagged skipped regardless because we don't know what the user +// actually heard. +func (w *Writer) autoClosePriorOpen( + ctx context.Context, + q *dbq.Queries, + userID pgtype.UUID, + at time.Time, +) error { + prior, err := q.GetOpenPlayEventForUser(ctx, userID) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil + } + return err + } + track, err := q.GetTrackByID(ctx, prior.TrackID) + if err != nil { + return err + } + elapsedMs := int32(at.Sub(prior.StartedAt.Time) / time.Millisecond) + if elapsedMs < 0 { + elapsedMs = 0 + } + if elapsedMs > track.DurationMs && track.DurationMs > 0 { + elapsedMs = track.DurationMs + } + ratio := 0.0 + if track.DurationMs > 0 { + ratio = float64(elapsedMs) / float64(track.DurationMs) + } + ratioPtr := ratio + durPtr := elapsedMs + if _, err := q.UpdatePlayEventEnded(ctx, dbq.UpdatePlayEventEndedParams{ + ID: prior.ID, + EndedAt: pgtype.Timestamptz{Time: at, Valid: true}, + DurationPlayedMs: &durPtr, + CompletionRatio: &ratioPtr, + WasSkipped: true, + }); err != nil { + return err + } + w.logger.Info("playevents: auto-closed prior open row", + "play_event_id", prior.ID, "elapsed_ms", elapsedMs) + return nil +} diff --git a/internal/playevents/writer_test.go b/internal/playevents/writer_test.go new file mode 100644 index 00000000..4d300a5c --- /dev/null +++ b/internal/playevents/writer_test.go @@ -0,0 +1,236 @@ +package playevents + +import ( + "context" + "io" + "log/slog" + "os" + "testing" + "time" + + "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 { + t.Helper() + if testing.Short() { + t.Skip("skipping integration test 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 play_events, skip_events, play_sessions, sessions, users, tracks, albums, artists RESTART IDENTITY CASCADE"); err != nil { + t.Fatalf("truncate: %v", err) + } + return pool +} + +type fixture struct { + pool *pgxpool.Pool + q *dbq.Queries + w *Writer + user pgtype.UUID + track pgtype.UUID +} + +func newFixture(t *testing.T, durationMs int32) fixture { + t.Helper() + pool := testPool(t) + q := dbq.New(pool) + u, err := q.CreateUser(context.Background(), dbq.CreateUserParams{ + Username: "tester", PasswordHash: "x", ApiToken: "x", IsAdmin: false, + }) + if err != nil { + t.Fatalf("user: %v", err) + } + a, err := q.UpsertArtist(context.Background(), dbq.UpsertArtistParams{Name: "Artist", SortName: "Artist"}) + if err != nil { + t.Fatalf("artist: %v", err) + } + al, err := q.UpsertAlbum(context.Background(), dbq.UpsertAlbumParams{Title: "Album", SortTitle: "Album", ArtistID: a.ID}) + if err != nil { + t.Fatalf("album: %v", err) + } + tr, err := q.UpsertTrack(context.Background(), dbq.UpsertTrackParams{ + Title: "Track", AlbumID: al.ID, ArtistID: a.ID, + FilePath: "/tmp/track-fixture.flac", + DurationMs: durationMs, + }) + if err != nil { + t.Fatalf("track: %v", err) + } + return fixture{ + pool: pool, + q: q, + w: NewWriter(pool, slog.New(slog.NewTextHandler(io.Discard, nil)), 30*time.Minute, 0.5, 30000), + user: u.ID, + track: tr.ID, + } +} + +func TestRecordPlayStarted_InsertsRow(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) + } + if !res.PlayEventID.Valid { + t.Fatalf("play_event_id invalid") + } + got, err := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) + if err != nil { + t.Fatalf("get: %v", err) + } + if got.EndedAt.Valid { + t.Errorf("ended_at should be NULL right after start") + } + if got.WasSkipped { + t.Errorf("was_skipped should default false") + } +} + +func TestRecordPlayStarted_AutoClosesPriorOpenRow(t *testing.T) { + f := newFixture(t, 200_000) + t1 := time.Now().UTC() + first, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) + if err != nil { + t.Fatalf("first: %v", err) + } + t2 := t1.Add(45 * time.Second) + _, err = f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t2) + if err != nil { + t.Fatalf("second: %v", err) + } + prior, err := f.q.GetPlayEventByID(context.Background(), first.PlayEventID) + if err != nil { + t.Fatalf("get: %v", err) + } + if !prior.EndedAt.Valid { + t.Errorf("prior should be closed") + } + if !prior.WasSkipped { + t.Errorf("prior should be marked was_skipped=true (auto-close convention)") + } + if prior.DurationPlayedMs == nil || *prior.DurationPlayedMs != 45_000 { + t.Errorf("duration_played_ms = %v, want 45000", prior.DurationPlayedMs) + } +} + +func TestRecordPlayStarted_AutoCloseCapsAtTrackDuration(t *testing.T) { + f := newFixture(t, 60_000) // 60s track + t1 := time.Now().UTC() + first, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) + // Start a second play 5 minutes later — elapsed > track duration. + t2 := t1.Add(5 * time.Minute) + _, _ = f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t2) + prior, _ := f.q.GetPlayEventByID(context.Background(), first.PlayEventID) + if prior.DurationPlayedMs == nil || *prior.DurationPlayedMs != 60_000 { + t.Errorf("duration_played_ms = %v, want capped at 60000", prior.DurationPlayedMs) + } +} + +func TestRecordPlayEnded_AppliesSkipRule_BothFail_NotSkip(t *testing.T) { + f := newFixture(t, 200_000) + t1 := time.Now().UTC() + res, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) + // 15% completion (30_000ms), 30s duration -> 30s satisfies "duration >= 30s" + // so the rule's AND fails -> NOT a skip. + t2 := t1.Add(30 * time.Second) + if err := f.w.RecordPlayEnded(context.Background(), res.PlayEventID, 30_000, t2); err != nil { + t.Fatalf("RecordPlayEnded: %v", err) + } + got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) + if got.WasSkipped { + t.Errorf("was_skipped = true, want false (duration_played_ms == 30000 fails the AND)") + } +} + +func TestRecordPlayEnded_AppliesSkipRule_BothPass_IsSkip(t *testing.T) { + f := newFixture(t, 200_000) + t1 := time.Now().UTC() + res, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) + // 5% completion (10_000ms), 10s duration -> both conditions hold -> SKIP. + t2 := t1.Add(10 * time.Second) + if err := f.w.RecordPlayEnded(context.Background(), res.PlayEventID, 10_000, t2); err != nil { + t.Fatalf("RecordPlayEnded: %v", err) + } + got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) + if !got.WasSkipped { + t.Errorf("expected was_skipped=true") + } + rows, err := f.pool.Query(context.Background(), + "SELECT id FROM skip_events WHERE user_id=$1 AND track_id=$2", f.user, f.track) + if err != nil { + t.Fatalf("query skips: %v", err) + } + defer rows.Close() + count := 0 + for rows.Next() { + count++ + } + if count != 1 { + t.Errorf("skip_events count = %d, want 1", count) + } +} + +func TestRecordPlayEnded_DurationOver50Percent_NotSkip(t *testing.T) { + f := newFixture(t, 200_000) // 200s track + t1 := time.Now().UTC() + res, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) + // 50% completion (100_000ms), 5s duration. Completion fails the "< 0.5" + // check — so AND fails, NOT a skip. + t2 := t1.Add(5 * time.Second) + _ = f.w.RecordPlayEnded(context.Background(), res.PlayEventID, 100_000, t2) + got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) + if got.WasSkipped { + t.Errorf("expected was_skipped=false at completion >= 0.5") + } +} + +func TestRecordPlaySkipped_AlwaysSkipFlagAndSkipEventRow(t *testing.T) { + f := newFixture(t, 200_000) + t1 := time.Now().UTC() + res, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) + // User-initiated skip at 90% — overrides the rule, always treated as skip. + t2 := t1.Add(180 * time.Second) + if err := f.w.RecordPlaySkipped(context.Background(), res.PlayEventID, 180_000, t2); err != nil { + t.Fatalf("RecordPlaySkipped: %v", err) + } + got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) + if !got.WasSkipped { + t.Errorf("RecordPlaySkipped must always set was_skipped=true") + } +} + +func TestRecordSyntheticCompletedPlay_WritesBothRows(t *testing.T) { + f := newFixture(t, 200_000) + now := time.Now().UTC() + if err := f.w.RecordSyntheticCompletedPlay(context.Background(), f.user, f.track, "feishin", now); err != nil { + t.Fatalf("synthetic: %v", err) + } + rows, _ := f.pool.Query(context.Background(), + "SELECT id FROM play_events WHERE user_id=$1 AND ended_at IS NOT NULL", f.user) + defer rows.Close() + count := 0 + for rows.Next() { + count++ + } + if count != 1 { + t.Errorf("closed play_events count = %d, want 1", count) + } +}