// 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" "encoding/json" "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" "git.fabledsword.com/bvandeusen/minstrel/internal/recommendation" "git.fabledsword.com/bvandeusen/minstrel/internal/scrobble" ) 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 // 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 } // 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 { if err := 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 }); 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 // 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 { var playEventID pgtype.UUID if err := 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 } playEventID = ev.ID 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{ 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 }); 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 // (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 // 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 }