diff --git a/internal/api/events.go b/internal/api/events.go index 6bdc898e..c7e49f16 100644 --- a/internal/api/events.go +++ b/internal/api/events.go @@ -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, diff --git a/internal/playevents/writer.go b/internal/playevents/writer.go index a218f16a..f523a3b4 100644 --- a/internal/playevents/writer.go +++ b/internal/playevents/writer.go @@ -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