feat(playevents): add Writer with start/end/skipped/synthetic paths

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.
This commit is contained in:
2026-04-25 22:32:33 -04:00
parent 73d3962fff
commit 0c7e1b0408
9 changed files with 563 additions and 7 deletions
+1 -1
View File
@@ -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
+1 -1
View File
@@ -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
+1 -1
View File
@@ -1,6 +1,6 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.27.0
// sqlc v1.31.1
package dbq
+35 -1
View File
@@ -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
+1 -1
View File
@@ -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
+1 -1
View File
@@ -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
+1 -1
View File
@@ -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
+286
View File
@@ -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
}
+236
View File
@@ -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)
}
}