feat(playlists): #415 stage 1 — rotation-state schema + play ingest

First of three stages for system-playlist sample-history dedup
(Fable #415, server-side rotation per the operator's choice).

Schema (migration 0027):
- play_events.source (nullable text): which surface a play came
  from. 'for_you' / 'discover' feed rotation; NULL for library /
  user-playlist / radio / Subsonic.
- system_playlist_rotation_state(user_id, playlist_kind,
  played_track_ids uuid[], rotation_started_at, updated_at): the
  per-(user,kind) set of already-heard tracks this rotation.

Ingest:
- New RecordPlayStartedWithSource on the writer; RecordPlayStarted
  is now a thin source="" wrapper so the frozen Subsonic shim is
  untouched (no signature ripple).
- When source is a known system kind, the same txn appends the
  track to rotation state (AppendRotationPlayed keeps the array a
  set via the conflict CASE).
- /api/events play_started accepts an optional "source".

No serve-behavior change yet — Stage 2 makes shuffle prefer the
unplayed tail + resets on exhaustion; Stage 3 wires the clients to
send source and consume the rotation-aware order.

Tests: rotation appends + dedupes for a system source; source-less
play writes no rotation row. (Existing RecordPlayStarted tests are
unchanged — same wrapper signature, identical behavior at source="".)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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))

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
+    user_id, track_id, session_id, started_at, client_id, source
-) VALUES ($1, $2, $3, $4, $5)
+) 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
+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
 }

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

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
+}

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;

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)
+);

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
+    user_id, track_id, session_id, started_at, client_id, source
-) VALUES ($1, $2, $3, $4, $5)
+) VALUES ($1, $2, $3, $4, $5, $6)
 RETURNING *;
 
 -- name: UpdatePlayEventEnded :one

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;

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

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")
+	}
+}