minstrel/internal/api/events.go

package api

import (
	"encoding/json"
	"errors"
	"net/http"
	"time"

	"github.com/jackc/pgx/v5"

	"git.fabledsword.com/bvandeusen/minstrel/internal/apierror"
	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
)

type eventRequest struct {
	Type             string  `json:"type"`
	TrackID          string  `json:"track_id"`
	PlayEventID      string  `json:"play_event_id"`
	DurationPlayedMs *int32  `json:"duration_played_ms"`
	PositionMs       *int32  `json:"position_ms"`
	At               *string `json:"at"`
	ClientID         *string `json:"client_id"`
	// Source tags which surface a play_started came from. "for_you"
	// / "discover" feed the system-playlist rotation dedup (#415);
	// absent / "" for library, user-playlist, radio, Subsonic.
	Source *string `json:"source"`
}

type playStartedResponse struct {
	PlayEventID string `json:"play_event_id"`
	SessionID   string `json:"session_id"`
}

// eventFutureSkew is the tolerated client-clock drift into the future
// before a timestamp is treated as bogus and replaced with now.
const eventFutureSkew = 5 * time.Minute

// clampEventTime bounds a client-supplied event timestamp to sanity:
// no earlier than the account's creation (offline replays can
// legitimately be days old, but no play can predate the user) and no
// later than now + a small skew allowance. Unbounded client clocks
// previously let a skewed device write arbitrarily old plays, which
// poisoned Rediscover's "not played in 6 months" ordering (#1246).
func clampEventTime(at, userCreatedAt, now time.Time) time.Time {
	if at.Before(userCreatedAt) {
		return userCreatedAt
	}
	if at.After(now.Add(eventFutureSkew)) {
		return now
	}
	return at
}

type okResponse struct {
	OK bool `json:"ok"`
}

func (h *handlers) handleEvents(w http.ResponseWriter, r *http.Request) {
	user, ok := requireUser(w, r)
	if !ok {
		return
	}
	var req eventRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeErr(w, apierror.BadRequest("bad_request", "invalid JSON body"))
		return
	}
	at := time.Now().UTC()
	if req.At != nil && *req.At != "" {
		parsed, err := time.Parse(time.RFC3339, *req.At)
		if err != nil {
			writeErr(w, apierror.BadRequest("bad_request", "invalid `at` timestamp"))
			return
		}
		at = clampEventTime(parsed.UTC(), user.CreatedAt.Time, time.Now().UTC())
	}
	clientID := ""
	if req.ClientID != nil {
		clientID = *req.ClientID
	}
	switch req.Type {
	case "play_started":
		h.handleEventPlayStarted(w, r, user, req, at, clientID)
	case "play_ended":
		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"))
	}
}

func (h *handlers) handleEventPlayStarted(
	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 _, 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: lookup track", "err", err)
		writeErr(w, apierror.InternalMsg("lookup failed", err))
		return
	}
	source := ""
	if req.Source != nil {
		source = *req.Source
	}
	res, err := h.events.RecordPlayStartedWithSource(r.Context(), user.ID, trackID, clientID, source, at)
	if err != nil {
		h.logger.Error("api: events: play_started", "err", err)
		writeErr(w, apierror.InternalMsg("record failed", err))
		return
	}
	writeJSON(w, http.StatusOK, playStartedResponse{
		PlayEventID: uuidToString(res.PlayEventID),
		SessionID:   uuidToString(res.SessionID),
	})
}

// 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,
) {
	playEventID, ok := parseUUID(req.PlayEventID)
	if !ok {
		writeErr(w, apierror.BadRequest("bad_request", "invalid play_event_id"))
		return
	}
	if req.DurationPlayedMs == nil || *req.DurationPlayedMs < 0 {
		writeErr(w, apierror.BadRequest("bad_request", "duration_played_ms required and must be >= 0"))
		return
	}
	ev, err := dbq.New(h.pool).GetPlayEventByID(r.Context(), playEventID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			writeErr(w, &apierror.Error{Status: 404, Code: "not_found", Message: "play_event not found"})
			return
		}
		h.logger.Error("api: events: lookup play_event", "err", err)
		writeErr(w, apierror.InternalMsg("lookup failed", err))
		return
	}
	if ev.UserID != user.ID {
		writeErr(w, apierror.Forbidden("forbidden", "play_event belongs to a different user"))
		return
	}
	if err := h.events.RecordPlayEnded(r.Context(), playEventID, *req.DurationPlayedMs, at); err != nil {
		h.logger.Error("api: events: play_ended", "err", err)
		writeErr(w, apierror.InternalMsg("record failed", err))
		return
	}
	writeJSON(w, http.StatusOK, okResponse{OK: true})
}

func (h *handlers) handleEventPlaySkipped(
	w http.ResponseWriter, r *http.Request,
	user dbq.User, req eventRequest, at time.Time,
) {
	playEventID, ok := parseUUID(req.PlayEventID)
	if !ok {
		writeErr(w, apierror.BadRequest("bad_request", "invalid play_event_id"))
		return
	}
	if req.PositionMs == nil || *req.PositionMs < 0 {
		writeErr(w, apierror.BadRequest("bad_request", "position_ms required and must be >= 0"))
		return
	}
	ev, err := dbq.New(h.pool).GetPlayEventByID(r.Context(), playEventID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			writeErr(w, &apierror.Error{Status: 404, Code: "not_found", Message: "play_event not found"})
			return
		}
		h.logger.Error("api: events: lookup play_event", "err", err)
		writeErr(w, apierror.InternalMsg("lookup failed", err))
		return
	}
	if ev.UserID != user.ID {
		writeErr(w, apierror.Forbidden("forbidden", "play_event belongs to a different user"))
		return
	}
	if err := h.events.RecordPlaySkipped(r.Context(), playEventID, *req.PositionMs, at); err != nil {
		h.logger.Error("api: events: play_skipped", "err", err)
		writeErr(w, apierror.InternalMsg("record failed", err))
		return
	}
	writeJSON(w, http.StatusOK, okResponse{OK: true})
}