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>
This commit is contained in:
@@ -65,6 +65,8 @@ func (h *handlers) handleEvents(w http.ResponseWriter, r *http.Request) {
|
|||||||
h.handleEventPlayEnded(w, r, user, req, at)
|
h.handleEventPlayEnded(w, r, user, req, at)
|
||||||
case "play_skipped":
|
case "play_skipped":
|
||||||
h.handleEventPlaySkipped(w, r, user, req, at)
|
h.handleEventPlaySkipped(w, r, user, req, at)
|
||||||
|
case "play_offline":
|
||||||
|
h.handleEventPlayOffline(w, r, user, req, at, clientID)
|
||||||
default:
|
default:
|
||||||
writeErr(w, apierror.BadRequest("bad_request", "unknown event type"))
|
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(
|
func (h *handlers) handleEventPlayEnded(
|
||||||
w http.ResponseWriter, r *http.Request,
|
w http.ResponseWriter, r *http.Request,
|
||||||
user dbq.User, req eventRequest, at time.Time,
|
user dbq.User, req eventRequest, at time.Time,
|
||||||
|
|||||||
@@ -306,6 +306,119 @@ func (w *Writer) RecordSyntheticCompletedPlay(
|
|||||||
return nil
|
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
|
// captureSessionVector queries the user's prior plays in the given session
|
||||||
// (before `at`), builds the session vector, and UPDATEs the just-inserted
|
// (before `at`), builds the session vector, and UPDATEs the just-inserted
|
||||||
// play_event with it. Runs inside the caller's transaction (q is the
|
// play_event with it. Runs inside the caller's transaction (q is the
|
||||||
|
|||||||
Reference in New Issue
Block a user