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
@@ -65,6 +65,8 @@ func (h *handlers) handleEvents(w http.ResponseWriter, r *http.Request) {
 		h.handleEventPlayEnded(w, r, user, req, at)
 	case "play_skipped":
 		h.handleEventPlaySkipped(w, r, user, req, at)
+	case "play_offline":
+		h.handleEventPlayOffline(w, r, user, req, at, clientID)
 	default:
 		writeErr(w, apierror.BadRequest("bad_request", "unknown event type"))
 	}
@@ -104,6 +106,48 @@ func (h *handlers) handleEventPlayStarted(
 	})
 }

+// handleEventPlayOffline records a complete play that happened with
+// no/flaky connectivity, replayed from the Flutter offline mutation
+// queue (#426 part B). `at` is the original (client) play-start time
+// — parsed by handleEvents from req.At — so history lands on the
+// real timeline, not replay time. duration_played_ms drives the skip
+// rule; source advances #415 rotation for system-playlist plays.
+func (h *handlers) handleEventPlayOffline(
+	w http.ResponseWriter, r *http.Request,
+	user dbq.User, req eventRequest, at time.Time, clientID string,
+) {
+	trackID, ok := parseUUID(req.TrackID)
+	if !ok {
+		writeErr(w, apierror.BadRequest("bad_request", "invalid track_id"))
+		return
+	}
+	if req.DurationPlayedMs == nil || *req.DurationPlayedMs < 0 {
+		writeErr(w, apierror.BadRequest("bad_request", "duration_played_ms required and must be >= 0"))
+		return
+	}
+	if _, err := dbq.New(h.pool).GetTrackByID(r.Context(), trackID); err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			writeErr(w, &apierror.Error{Status: 404, Code: "not_found", Message: "track not found"})
+			return
+		}
+		h.logger.Error("api: events: offline lookup track", "err", err)
+		writeErr(w, apierror.InternalMsg("lookup failed", err))
+		return
+	}
+	source := ""
+	if req.Source != nil {
+		source = *req.Source
+	}
+	if err := h.events.RecordOfflinePlay(
+		r.Context(), user.ID, trackID, clientID, source, at, *req.DurationPlayedMs,
+	); err != nil {
+		h.logger.Error("api: events: play_offline", "err", err)
+		writeErr(w, apierror.InternalMsg("record failed", err))
+		return
+	}
+	writeJSON(w, http.StatusOK, okResponse{OK: true})
+}
+
 func (h *handlers) handleEventPlayEnded(
 	w http.ResponseWriter, r *http.Request,
 	user dbq.User, req eventRequest, at time.Time,
@@ -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