feat(playevents): #426B server — RecordOfflinePlay + /api/events play_offline

Server half of the offline-replay capture. New writer path RecordOfflinePlay: writes a complete start+end play in one txn from a caller-supplied `at` + duration_played_ms, applying the spec §6 skip rule (same AND-of-thresholds as RecordPlayEnded) and threading `source` so #415 rotation advances for system-playlist plays just like the live path. Generalizes RecordSyntheticCompletedPlay (which hard-codes full completion). duration clamped to [0, track len]. New /api/events type "play_offline" → handleEventPlayOffline: validates track + duration, reuses the existing req.At parse so the play lands on the original timeline, not replay time. Subsonic shim + live 3-call lifecycle untouched. Flutter half next: EventsApi.playOffline, a play.offline MutationQueue kind, and PlayEventsReporter capturing the completed play + enqueuing it when the live calls have no server id / fail. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-15 11:14:21 -04:00
parent 8652d7124e
commit 47aa178850
2 changed files with 157 additions and 0 deletions
@@ -306,6 +306,119 @@ func (w *Writer) RecordSyntheticCompletedPlay(
 	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
+		}
+		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
+		}
+		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
+		}
+		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