diff --git a/internal/api/events.go b/internal/api/events.go index 2e95ee36..6bdc898e 100644 --- a/internal/api/events.go +++ b/internal/api/events.go @@ -20,6 +20,10 @@ type eventRequest struct { PositionMs *int32 `json:"position_ms"` At *string `json:"at"` ClientID *string `json:"client_id"` + // Source tags which surface a play_started came from. "for_you" + // / "discover" feed the system-playlist rotation dedup (#415); + // absent / "" for library, user-playlist, radio, Subsonic. + Source *string `json:"source"` } type playStartedResponse struct { @@ -84,7 +88,11 @@ func (h *handlers) handleEventPlayStarted( writeErr(w, apierror.InternalMsg("lookup failed", err)) return } - res, err := h.events.RecordPlayStarted(r.Context(), user.ID, trackID, clientID, at) + source := "" + if req.Source != nil { + source = *req.Source + } + res, err := h.events.RecordPlayStartedWithSource(r.Context(), user.ID, trackID, clientID, source, at) if err != nil { h.logger.Error("api: events: play_started", "err", err) writeErr(w, apierror.InternalMsg("record failed", err)) diff --git a/internal/db/dbq/events.sql.go b/internal/db/dbq/events.sql.go index 83a04dee..59f08d04 100644 --- a/internal/db/dbq/events.sql.go +++ b/internal/db/dbq/events.sql.go @@ -54,7 +54,7 @@ func (q *Queries) GetMostRecentPlaySessionForUser(ctx context.Context, userID pg } const getOpenPlayEventForUser = `-- name: GetOpenPlayEventForUser :one -SELECT id, user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped, client_id, session_vector_at_play, scrobbled_at FROM play_events +SELECT id, user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped, client_id, session_vector_at_play, scrobbled_at, source FROM play_events WHERE user_id = $1 AND ended_at IS NULL ORDER BY started_at DESC LIMIT 1 @@ -78,12 +78,13 @@ func (q *Queries) GetOpenPlayEventForUser(ctx context.Context, userID pgtype.UUI &i.ClientID, &i.SessionVectorAtPlay, &i.ScrobbledAt, + &i.Source, ) return i, err } const getPlayEventByID = `-- name: GetPlayEventByID :one -SELECT id, user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped, client_id, session_vector_at_play, scrobbled_at FROM play_events WHERE id = $1 +SELECT id, user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped, client_id, session_vector_at_play, scrobbled_at, source FROM play_events WHERE id = $1 ` func (q *Queries) GetPlayEventByID(ctx context.Context, id pgtype.UUID) (PlayEvent, error) { @@ -102,15 +103,16 @@ func (q *Queries) GetPlayEventByID(ctx context.Context, id pgtype.UUID) (PlayEve &i.ClientID, &i.SessionVectorAtPlay, &i.ScrobbledAt, + &i.Source, ) return i, err } const insertPlayEvent = `-- name: InsertPlayEvent :one INSERT INTO play_events ( - user_id, track_id, session_id, started_at, client_id -) VALUES ($1, $2, $3, $4, $5) -RETURNING id, user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped, client_id, session_vector_at_play, scrobbled_at + user_id, track_id, session_id, started_at, client_id, source +) VALUES ($1, $2, $3, $4, $5, $6) +RETURNING id, user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped, client_id, session_vector_at_play, scrobbled_at, source ` type InsertPlayEventParams struct { @@ -119,6 +121,7 @@ type InsertPlayEventParams struct { SessionID pgtype.UUID StartedAt pgtype.Timestamptz ClientID *string + Source *string } func (q *Queries) InsertPlayEvent(ctx context.Context, arg InsertPlayEventParams) (PlayEvent, error) { @@ -128,6 +131,7 @@ func (q *Queries) InsertPlayEvent(ctx context.Context, arg InsertPlayEventParams arg.SessionID, arg.StartedAt, arg.ClientID, + arg.Source, ) var i PlayEvent err := row.Scan( @@ -143,6 +147,7 @@ func (q *Queries) InsertPlayEvent(ctx context.Context, arg InsertPlayEventParams &i.ClientID, &i.SessionVectorAtPlay, &i.ScrobbledAt, + &i.Source, ) return i, err } @@ -286,7 +291,7 @@ SET ended_at = $2, completion_ratio = $4, was_skipped = $5 WHERE id = $1 -RETURNING id, user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped, client_id, session_vector_at_play, scrobbled_at +RETURNING id, user_id, track_id, session_id, started_at, ended_at, duration_played_ms, completion_ratio, was_skipped, client_id, session_vector_at_play, scrobbled_at, source ` type UpdatePlayEventEndedParams struct { @@ -322,6 +327,7 @@ func (q *Queries) UpdatePlayEventEnded(ctx context.Context, arg UpdatePlayEventE &i.ClientID, &i.SessionVectorAtPlay, &i.ScrobbledAt, + &i.Source, ) return i, err } diff --git a/internal/db/dbq/models.go b/internal/db/dbq/models.go index f99894a5..3b772075 100644 --- a/internal/db/dbq/models.go +++ b/internal/db/dbq/models.go @@ -369,6 +369,7 @@ type PlayEvent struct { ClientID *string SessionVectorAtPlay []byte ScrobbledAt pgtype.Timestamptz + Source *string } type PlaySession struct { @@ -475,6 +476,14 @@ type SmtpConfig struct { UpdatedAt pgtype.Timestamptz } +type SystemPlaylistRotationState struct { + UserID pgtype.UUID + PlaylistKind string + PlayedTrackIds []pgtype.UUID + RotationStartedAt pgtype.Timestamptz + UpdatedAt pgtype.Timestamptz +} + type SystemPlaylistRun struct { UserID pgtype.UUID LastRunAt pgtype.Timestamptz diff --git a/internal/db/dbq/system_playlist_rotation.sql.go b/internal/db/dbq/system_playlist_rotation.sql.go new file mode 100644 index 00000000..6aeff689 --- /dev/null +++ b/internal/db/dbq/system_playlist_rotation.sql.go @@ -0,0 +1,91 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: system_playlist_rotation.sql + +package dbq + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const appendRotationPlayed = `-- name: AppendRotationPlayed :exec + +INSERT INTO system_playlist_rotation_state ( + user_id, playlist_kind, played_track_ids, rotation_started_at, updated_at +) VALUES ( + $1, $2, + array_append('{}'::uuid[], $3::uuid), now(), now() +) +ON CONFLICT (user_id, playlist_kind) DO UPDATE +SET played_track_ids = + CASE WHEN $3::uuid = ANY(system_playlist_rotation_state.played_track_ids) + THEN system_playlist_rotation_state.played_track_ids + ELSE array_append(system_playlist_rotation_state.played_track_ids, $3::uuid) + END, + updated_at = now() +` + +type AppendRotationPlayedParams struct { + UserID pgtype.UUID + PlaylistKind string + TrackID pgtype.UUID +} + +// Rotation state for system-playlist sample-history dedup (#415). +// One row per (user, playlist_kind). played_track_ids accumulates +// tracks heard from that playlist in the current rotation; shuffle +// prefers ids NOT in this set, then resets when the snapshot is +// exhausted. +// Records that the given track was played from the (user, kind) +// system playlist. Creates the row on first play; on conflict +// appends only if not already present so the array stays a set. +func (q *Queries) AppendRotationPlayed(ctx context.Context, arg AppendRotationPlayedParams) error { + _, err := q.db.Exec(ctx, appendRotationPlayed, arg.UserID, arg.PlaylistKind, arg.TrackID) + return err +} + +const getRotationState = `-- name: GetRotationState :one +SELECT user_id, playlist_kind, played_track_ids, rotation_started_at, updated_at FROM system_playlist_rotation_state +WHERE user_id = $1 AND playlist_kind = $2 +` + +type GetRotationStateParams struct { + UserID pgtype.UUID + PlaylistKind string +} + +func (q *Queries) GetRotationState(ctx context.Context, arg GetRotationStateParams) (SystemPlaylistRotationState, error) { + row := q.db.QueryRow(ctx, getRotationState, arg.UserID, arg.PlaylistKind) + var i SystemPlaylistRotationState + err := row.Scan( + &i.UserID, + &i.PlaylistKind, + &i.PlayedTrackIds, + &i.RotationStartedAt, + &i.UpdatedAt, + ) + return i, err +} + +const resetRotationState = `-- name: ResetRotationState :exec +UPDATE system_playlist_rotation_state +SET played_track_ids = '{}', + rotation_started_at = now(), + updated_at = now() +WHERE user_id = $1 AND playlist_kind = $2 +` + +type ResetRotationStateParams struct { + UserID pgtype.UUID + PlaylistKind string +} + +// Clears the played set and advances the rotation window. Called +// when the playlist snapshot has been fully sampled (Stage 2). +func (q *Queries) ResetRotationState(ctx context.Context, arg ResetRotationStateParams) error { + _, err := q.db.Exec(ctx, resetRotationState, arg.UserID, arg.PlaylistKind) + return err +} diff --git a/internal/db/migrations/0027_system_playlist_rotation.down.sql b/internal/db/migrations/0027_system_playlist_rotation.down.sql new file mode 100644 index 00000000..c52a8f5f --- /dev/null +++ b/internal/db/migrations/0027_system_playlist_rotation.down.sql @@ -0,0 +1,5 @@ +-- 0027_system_playlist_rotation.down.sql +DROP TABLE IF EXISTS system_playlist_rotation_state; + +ALTER TABLE play_events + DROP COLUMN IF EXISTS source; diff --git a/internal/db/migrations/0027_system_playlist_rotation.up.sql b/internal/db/migrations/0027_system_playlist_rotation.up.sql new file mode 100644 index 00000000..3314a4ea --- /dev/null +++ b/internal/db/migrations/0027_system_playlist_rotation.up.sql @@ -0,0 +1,26 @@ +-- 0027_system_playlist_rotation.up.sql — sample-history dedup for +-- system playlists (Fable #415). Two parts: +-- +-- 1. play_events.source: which system playlist a play originated +-- from ('for_you' | 'discover' | future kinds), or NULL for any +-- play not launched from a system-playlist surface (library, +-- user playlist, radio, Subsonic). Lets the ingest path know a +-- play should count against that playlist's rotation. +-- +-- 2. system_playlist_rotation_state: per (user, playlist_kind) set +-- of track ids already played in the current rotation window. +-- Shuffle-on-play prefers the unplayed tail; when the snapshot +-- is exhausted the set resets and rotation_started_at advances. +-- Centralized server-side so phone / web / watch share one +-- rotation, mirroring how radio dedup already lives server-side. +ALTER TABLE play_events + ADD COLUMN source text; + +CREATE TABLE system_playlist_rotation_state ( + user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE, + playlist_kind text NOT NULL, + played_track_ids uuid[] NOT NULL DEFAULT '{}', + rotation_started_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (user_id, playlist_kind) +); diff --git a/internal/db/queries/events.sql b/internal/db/queries/events.sql index 60e42f4d..db0d4829 100644 --- a/internal/db/queries/events.sql +++ b/internal/db/queries/events.sql @@ -25,8 +25,8 @@ LIMIT 1; -- name: InsertPlayEvent :one INSERT INTO play_events ( - user_id, track_id, session_id, started_at, client_id -) VALUES ($1, $2, $3, $4, $5) + user_id, track_id, session_id, started_at, client_id, source +) VALUES ($1, $2, $3, $4, $5, $6) RETURNING *; -- name: UpdatePlayEventEnded :one diff --git a/internal/db/queries/system_playlist_rotation.sql b/internal/db/queries/system_playlist_rotation.sql new file mode 100644 index 00000000..3f95050e --- /dev/null +++ b/internal/db/queries/system_playlist_rotation.sql @@ -0,0 +1,36 @@ +-- Rotation state for system-playlist sample-history dedup (#415). +-- One row per (user, playlist_kind). played_track_ids accumulates +-- tracks heard from that playlist in the current rotation; shuffle +-- prefers ids NOT in this set, then resets when the snapshot is +-- exhausted. + +-- name: AppendRotationPlayed :exec +-- Records that the given track was played from the (user, kind) +-- system playlist. Creates the row on first play; on conflict +-- appends only if not already present so the array stays a set. +INSERT INTO system_playlist_rotation_state ( + user_id, playlist_kind, played_track_ids, rotation_started_at, updated_at +) VALUES ( + sqlc.arg(user_id), sqlc.arg(playlist_kind), + array_append('{}'::uuid[], sqlc.arg(track_id)::uuid), now(), now() +) +ON CONFLICT (user_id, playlist_kind) DO UPDATE +SET played_track_ids = + CASE WHEN sqlc.arg(track_id)::uuid = ANY(system_playlist_rotation_state.played_track_ids) + THEN system_playlist_rotation_state.played_track_ids + ELSE array_append(system_playlist_rotation_state.played_track_ids, sqlc.arg(track_id)::uuid) + END, + updated_at = now(); + +-- name: GetRotationState :one +SELECT * FROM system_playlist_rotation_state +WHERE user_id = $1 AND playlist_kind = $2; + +-- name: ResetRotationState :exec +-- Clears the played set and advances the rotation window. Called +-- when the playlist snapshot has been fully sampled (Stage 2). +UPDATE system_playlist_rotation_state +SET played_track_ids = '{}', + rotation_started_at = now(), + updated_at = now() +WHERE user_id = $1 AND playlist_kind = $2; diff --git a/internal/playevents/writer.go b/internal/playevents/writer.go index 00bff645..a218f16a 100644 --- a/internal/playevents/writer.go +++ b/internal/playevents/writer.go @@ -57,12 +57,36 @@ type StartedResult struct { // 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. +// return its id and session id. Source-agnostic — kept for the Subsonic +// shim (frozen API), which never originates from a system playlist. func (w *Writer) RecordPlayStarted( ctx context.Context, userID, trackID pgtype.UUID, clientID string, at time.Time, +) (StartedResult, error) { + return w.RecordPlayStartedWithSource(ctx, userID, trackID, clientID, "", at) +} + +// systemPlaylistSources are the play_events.source values that count +// against a system playlist's rotation (Fable #415). Add future +// system-playlist kinds here as they ship (deep_cuts, rediscover, …). +var systemPlaylistSources = map[string]bool{ + "for_you": true, + "discover": true, +} + +// RecordPlayStartedWithSource is RecordPlayStarted plus a `source` +// tag identifying which surface the play came from. When source is a +// known system-playlist kind the track is appended to that user's +// rotation set so shuffle-on-play can prefer the unplayed tail on a +// re-play within the day. Empty source → behaves exactly like the +// pre-#415 path (NULL column, no rotation write). +func (w *Writer) RecordPlayStartedWithSource( + ctx context.Context, + userID, trackID pgtype.UUID, + clientID, source string, + at time.Time, ) (StartedResult, error) { var out StartedResult err := pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error { @@ -78,12 +102,17 @@ func (w *Writer) RecordPlayStarted( if clientID != "" { clientIDPtr = &clientID } + var sourcePtr *string + if source != "" { + sourcePtr = &source + } ev, err := q.InsertPlayEvent(ctx, dbq.InsertPlayEventParams{ UserID: userID, TrackID: trackID, SessionID: sessionID, StartedAt: pgtype.Timestamptz{Time: at, Valid: true}, ClientID: clientIDPtr, + Source: sourcePtr, }) if err != nil { return err @@ -95,6 +124,16 @@ func (w *Writer) RecordPlayStarted( if err := w.captureSessionVector(ctx, q, ev.ID, sessionID, at); err != nil { return err } + + if systemPlaylistSources[source] { + if err := q.AppendRotationPlayed(ctx, dbq.AppendRotationPlayedParams{ + UserID: userID, + PlaylistKind: source, + TrackID: trackID, + }); err != nil { + return err + } + } return nil }) return out, err diff --git a/internal/playevents/writer_test.go b/internal/playevents/writer_test.go index 6f9d8233..bfb30dc7 100644 --- a/internal/playevents/writer_test.go +++ b/internal/playevents/writer_test.go @@ -372,3 +372,52 @@ func TestRecordPlayEnded_SubThreshold_NoEnqueue(t *testing.T) { t.Errorf("expected 0 queue rows for sub-threshold play, got %d", n) } } + +func TestRecordPlayStartedWithSource_AppendsRotation(t *testing.T) { + f := newFixture(t, 200_000) + now := time.Now().UTC() + if _, err := f.w.RecordPlayStartedWithSource( + context.Background(), f.user, f.track, "c", "for_you", now, + ); err != nil { + t.Fatalf("RecordPlayStartedWithSource: %v", err) + } + st, err := f.q.GetRotationState(context.Background(), dbq.GetRotationStateParams{ + UserID: f.user, PlaylistKind: "for_you", + }) + if err != nil { + t.Fatalf("GetRotationState: %v", err) + } + if len(st.PlayedTrackIds) != 1 || st.PlayedTrackIds[0] != f.track { + t.Errorf("played_track_ids = %v, want [%v]", st.PlayedTrackIds, f.track) + } + + // Re-playing the same track stays a set (no duplicate append). + if _, err := f.w.RecordPlayStartedWithSource( + context.Background(), f.user, f.track, "c", "for_you", now.Add(time.Minute), + ); err != nil { + t.Fatalf("RecordPlayStartedWithSource 2: %v", err) + } + st, err = f.q.GetRotationState(context.Background(), dbq.GetRotationStateParams{ + UserID: f.user, PlaylistKind: "for_you", + }) + if err != nil { + t.Fatalf("GetRotationState 2: %v", err) + } + if len(st.PlayedTrackIds) != 1 { + t.Errorf("played_track_ids should dedupe, got %v", st.PlayedTrackIds) + } +} + +func TestRecordPlayStarted_NoRotationWithoutSource(t *testing.T) { + f := newFixture(t, 200_000) + now := time.Now().UTC() + if _, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", now); err != nil { + t.Fatalf("RecordPlayStarted: %v", err) + } + _, err := f.q.GetRotationState(context.Background(), dbq.GetRotationStateParams{ + UserID: f.user, PlaylistKind: "for_you", + }) + if err == nil { + t.Errorf("expected no rotation row for a source-less play") + } +}