minstrel/internal/playevents/writer.go

// 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. 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). Mirrors the
// `playlists_kind_variant_consistent` CHECK in migration
// 0028_discovery_mix_variants.up.sql — every variant in that CHECK
// list that ships as a refreshable system mix needs to be here so
// the rotation reporter sees Android + web plays from that surface.
//
// Drift #563: this map drifted behind the migrations. It had only
// for_you + discover but migrations 0021 + 0028 added 6 more
// variants. Plays from Rediscover / Deep Cuts / Songs Like X /
// New for You / On This Day / First Listens didn't advance the
// per-user rotation, so "unplayed first" ordering staled on those
// mixes — same tracks kept surfacing.
//
// Future variant additions should update this map AND the CHECK in
// the matching migration in the same change; the
// `db_check_constraint_for_new_variants` standing rule covers the
// CHECK half.
var systemPlaylistSources = map[string]bool{
	"for_you":           true,
	"discover":          true,
	"deep_cuts":         true,
	"rediscover":        true,
	"new_for_you":       true,
	"on_this_day":       true,
	"first_listens":     true,
	"songs_like_artist": true,
}

// lookupSystemPickKind resolves a system-playlist play's pick_kind
// (taste/fresh, Discover's bucket, or a rule-#131 tier) from the user's
// live snapshot of that variant at ingestion time. Attribution must be
// frozen now — snapshots rebuild daily, so it can't be reconstructed at
// metrics-read time (#1249, generalized in #1270). Source strings
// double as system_variant values (see systemPlaylistSources). Returns
// nil (unattributed) when the track isn't in the current snapshot —
// plays predating the feature, or an offline replay arriving after the
// track rotated out — and when the variant doesn't stamp yet. Real DB
// errors propagate and fail the caller's transaction.
func lookupSystemPickKind(
	ctx context.Context,
	q *dbq.Queries,
	userID, trackID pgtype.UUID,
	variant string,
) (*string, error) {
	kind, err := q.GetSystemPickKindForTrack(ctx, dbq.GetSystemPickKindForTrackParams{
		UserID:        userID,
		SystemVariant: &variant,
		TrackID:       trackID,
	})
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return kind, nil
}

// 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 {
		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
		}
		var sourcePtr *string
		if source != "" {
			sourcePtr = &source
		}
		var pickKind *string
		if systemPlaylistSources[source] {
			pickKind, err = lookupSystemPickKind(ctx, q, userID, trackID, source)
			if err != nil {
				return err
			}
		}
		ev, err := q.InsertPlayEvent(ctx, dbq.InsertPlayEventParams{
			UserID:    userID,
			TrackID:   trackID,
			SessionID: sessionID,
			StartedAt: pgtype.Timestamptz{Time: at, Valid: true},
			ClientID:  clientIDPtr,
			Source:    sourcePtr,
			PickKind:  pickKind,
		})
		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
		}

		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
}

// 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
		}
		// Idempotency for the client's durable close-by-id replay: if the
		// row is already closed, re-applying the close is harmless (the
		// UPDATE is last-write-wins) but a second skip_events insert must
		// NOT fire — a lost-response retry would otherwise double the
		// skip signal. See the 2026-06-11 contract audit.
		alreadyEnded := ev.EndedAt.Valid
		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 && !alreadyEnded {
			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
}

// RecordOfflinePlay writes a complete start+end play in one call from
// a caller-supplied timestamp + duration. This is the replay path for
// the Flutter offline mutation queue (#426 part B): a play that
// happened with no/flaky connectivity is captured locally as a
// completed unit and replayed here once back online, preserving the
// original `at`.
//
// Generalizes RecordSyntheticCompletedPlay: instead of assuming full
// completion it applies the spec §6 skip rule from the real
// durationPlayed vs track length (same AND-of-thresholds as
// RecordPlayEnded), and threads `source` so #415 rotation advances
// for system-playlist plays just like the live path.
//
// durationPlayedMs is clamped to [0, track duration] to absorb
// client clock / position drift.
func (w *Writer) RecordOfflinePlay(
	ctx context.Context,
	userID, trackID pgtype.UUID,
	clientID, source string,
	at time.Time,
	durationPlayedMs int32,
) 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
		}
		// Idempotency: the offline queue is at-least-once, so a response lost
		// after commit makes the client re-fire the same play. The replayed
		// payload carries the original `at`, so an exact (user, track,
		// started_at) match means we already recorded it — no-op instead of a
		// duplicate history row. See the 2026-06-11 contract audit.
		var dupID pgtype.UUID
		err = tx.QueryRow(ctx,
			`SELECT id FROM play_events
			 WHERE user_id = $1 AND track_id = $2 AND started_at = $3
			 LIMIT 1`,
			userID, trackID, pgtype.Timestamptz{Time: at, Valid: true},
		).Scan(&dupID)
		if err == nil {
			return nil
		}
		if !errors.Is(err, pgx.ErrNoRows) {
			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
		}
		var sourcePtr *string
		if source != "" {
			sourcePtr = &source
		}
		var pickKind *string
		if systemPlaylistSources[source] {
			pickKind, err = lookupSystemPickKind(ctx, q, userID, trackID, source)
			if err != nil {
				return err
			}
		}
		ev, err := q.InsertPlayEvent(ctx, dbq.InsertPlayEventParams{
			UserID:    userID,
			TrackID:   trackID,
			SessionID: sessionID,
			StartedAt: pgtype.Timestamptz{Time: at, Valid: true},
			ClientID:  clientIDPtr,
			Source:    sourcePtr,
			PickKind:  pickKind,
		})
		if err != nil {
			return err
		}
		playEventID = ev.ID
		if err := w.captureSessionVector(ctx, q, ev.ID, sessionID, at); err != nil {
			return err
		}

		played := durationPlayedMs
		if played < 0 {
			played = 0
		}
		if track.DurationMs > 0 && played > track.DurationMs {
			played = track.DurationMs
		}
		ratio := 0.0
		if track.DurationMs > 0 {
			ratio = float64(played) / float64(track.DurationMs)
		}
		isSkip := ratio < w.skipMaxCompletionRatio && played < w.skipMaxDurationPlayedMs
		ratioPtr := ratio
		durPtr := played
		if _, err := q.UpdatePlayEventEnded(ctx, dbq.UpdatePlayEventEndedParams{
			ID:               ev.ID,
			EndedAt:          pgtype.Timestamptz{Time: at.Add(time.Duration(played) * time.Millisecond), Valid: true},
			DurationPlayedMs: &durPtr,
			CompletionRatio:  &ratioPtr,
			WasSkipped:       isSkip,
		}); err != nil {
			return err
		}
		if isSkip {
			if _, err := q.InsertSkipEvent(ctx, dbq.InsertSkipEventParams{
				UserID:     userID,
				TrackID:    trackID,
				SessionID:  sessionID,
				SkippedAt:  pgtype.Timestamptz{Time: at.Add(time.Duration(played) * time.Millisecond), Valid: true},
				PositionMs: played,
			}); 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
	}); err != nil {
		return err
	}
	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 and duration_played_ms = min(at - started_at,
// track duration) — our best estimate of how long the orphan played when
// the client never sent play_ended (e.g. backgrounded mid-track). The same
// skip rule as RecordPlayEnded is then applied to that estimate: an orphan
// that sat open past the track's length reads as ratio ~1 → a real play, so
// a fully-listened track that lost its close still lands in history instead
// of being force-hidden. Deliberately writes no skip_events row — an
// auto-close is an ambiguous signal, not a deliberate skip, and must not
// feed the skip-ratio / recommendation signal.
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)
	}
	isSkip := ratio < w.skipMaxCompletionRatio && elapsedMs < w.skipMaxDurationPlayedMs
	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:       isSkip,
	}); err != nil {
		return err
	}
	w.logger.Info("playevents: auto-closed prior open row",
		"play_event_id", prior.ID, "elapsed_ms", elapsedMs, "was_skipped", isSkip)
	return nil
}