diff --git a/cmd/minstrel/main.go b/cmd/minstrel/main.go index df3398a9..0c912bd5 100644 --- a/cmd/minstrel/main.go +++ b/cmd/minstrel/main.go @@ -75,7 +75,7 @@ func run() error { srv := server.New(logger, pool, scanner, subsonic.Config{ AllowPlaintextPassword: cfg.Subsonic.AllowPlaintextPassword, - }) + }, cfg.Events) httpServer := &http.Server{ Addr: cfg.Server.Address, Handler: srv.Router(), diff --git a/internal/api/api.go b/internal/api/api.go index 8c18e6ee..0ad95e11 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -6,17 +6,26 @@ package api import ( "log/slog" + "time" "github.com/go-chi/chi/v5" "github.com/jackc/pgx/v5/pgxpool" "git.fabledsword.com/bvandeusen/minstrel/internal/auth" + "git.fabledsword.com/bvandeusen/minstrel/internal/config" + "git.fabledsword.com/bvandeusen/minstrel/internal/playevents" ) // Mount attaches /api/* handlers to r. Public endpoints (login) are outside // RequireUser; everything else is gated by the middleware. -func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger) { - h := &handlers{pool: pool, logger: logger} +func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, cfg config.EventsConfig) { + w := playevents.NewWriter( + pool, logger, + time.Duration(cfg.SessionTimeoutMinutes)*time.Minute, + cfg.SkipMaxCompletionRatio, + cfg.SkipMaxDurationPlayedMs, + ) + h := &handlers{pool: pool, logger: logger, events: w} r.Route("/api", func(api chi.Router) { api.Post("/auth/login", h.handleLogin) @@ -33,6 +42,7 @@ func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger) { authed.Get("/tracks/{id}/stream", h.handleGetStream) authed.Get("/search", h.handleSearch) authed.Get("/radio", h.handleRadio) + authed.Post("/events", h.handleEvents) }) }) } @@ -40,4 +50,5 @@ func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger) { type handlers struct { pool *pgxpool.Pool logger *slog.Logger + events *playevents.Writer } diff --git a/internal/api/auth_test.go b/internal/api/auth_test.go index f0b8cd39..d91a21e6 100644 --- a/internal/api/auth_test.go +++ b/internal/api/auth_test.go @@ -12,12 +12,15 @@ import ( "strings" "testing" + "time" + "github.com/jackc/pgx/v5/pgxpool" "golang.org/x/crypto/bcrypt" "git.fabledsword.com/bvandeusen/minstrel/internal/auth" "git.fabledsword.com/bvandeusen/minstrel/internal/db" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + "git.fabledsword.com/bvandeusen/minstrel/internal/playevents" ) // testHandlers spins up a handlers instance against MINSTREL_TEST_DATABASE_URL. @@ -42,10 +45,11 @@ func testHandlers(t *testing.T) (*handlers, *pgxpool.Pool) { } t.Cleanup(pool.Close) if _, err := pool.Exec(context.Background(), - "TRUNCATE sessions, users RESTART IDENTITY CASCADE"); err != nil { + "TRUNCATE play_events, skip_events, play_sessions, sessions, users RESTART IDENTITY CASCADE"); err != nil { t.Fatalf("truncate: %v", err) } - return &handlers{pool: pool, logger: logger}, pool + w := playevents.NewWriter(pool, logger, 30*time.Minute, 0.5, 30000) + return &handlers{pool: pool, logger: logger, events: w}, pool } func seedUser(t *testing.T, pool *pgxpool.Pool, username, password string, isAdmin bool) dbq.User { diff --git a/internal/api/events.go b/internal/api/events.go new file mode 100644 index 00000000..af32a3ce --- /dev/null +++ b/internal/api/events.go @@ -0,0 +1,168 @@ +package api + +import ( + "encoding/json" + "errors" + "net/http" + "time" + + "github.com/jackc/pgx/v5" + + "git.fabledsword.com/bvandeusen/minstrel/internal/auth" + "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"` +} + +type playStartedResponse struct { + PlayEventID string `json:"play_event_id"` + SessionID string `json:"session_id"` +} + +type okResponse struct { + OK bool `json:"ok"` +} + +func (h *handlers) handleEvents(w http.ResponseWriter, r *http.Request) { + user, ok := auth.UserFromContext(r.Context()) + if !ok { + writeErr(w, http.StatusUnauthorized, "unauthorized", "authentication required") + return + } + var req eventRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeErr(w, http.StatusBadRequest, "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, http.StatusBadRequest, "bad_request", "invalid `at` timestamp") + return + } + at = parsed + } + 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) + default: + writeErr(w, http.StatusBadRequest, "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, http.StatusBadRequest, "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, http.StatusNotFound, "not_found", "track not found") + return + } + h.logger.Error("api: events: lookup track", "err", err) + writeErr(w, http.StatusInternalServerError, "server_error", "lookup failed") + return + } + res, err := h.events.RecordPlayStarted(r.Context(), user.ID, trackID, clientID, at) + if err != nil { + h.logger.Error("api: events: play_started", "err", err) + writeErr(w, http.StatusInternalServerError, "server_error", "record failed") + return + } + writeJSON(w, http.StatusOK, playStartedResponse{ + PlayEventID: uuidToString(res.PlayEventID), + SessionID: uuidToString(res.SessionID), + }) +} + +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, http.StatusBadRequest, "bad_request", "invalid play_event_id") + return + } + if req.DurationPlayedMs == nil || *req.DurationPlayedMs < 0 { + writeErr(w, http.StatusBadRequest, "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, http.StatusNotFound, "not_found", "play_event not found") + return + } + h.logger.Error("api: events: lookup play_event", "err", err) + writeErr(w, http.StatusInternalServerError, "server_error", "lookup failed") + return + } + if ev.UserID != user.ID { + writeErr(w, http.StatusForbidden, "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, http.StatusInternalServerError, "server_error", "record failed") + 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, http.StatusBadRequest, "bad_request", "invalid play_event_id") + return + } + if req.PositionMs == nil || *req.PositionMs < 0 { + writeErr(w, http.StatusBadRequest, "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, http.StatusNotFound, "not_found", "play_event not found") + return + } + h.logger.Error("api: events: lookup play_event", "err", err) + writeErr(w, http.StatusInternalServerError, "server_error", "lookup failed") + return + } + if ev.UserID != user.ID { + writeErr(w, http.StatusForbidden, "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, http.StatusInternalServerError, "server_error", "record failed") + return + } + writeJSON(w, http.StatusOK, okResponse{OK: true}) +} diff --git a/internal/api/events_test.go b/internal/api/events_test.go new file mode 100644 index 00000000..2a5340fc --- /dev/null +++ b/internal/api/events_test.go @@ -0,0 +1,137 @@ +package api + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/jackc/pgx/v5/pgtype" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" +) + +// callEvents invokes h.handleEvents directly, injecting `user` via the same +// context key RequireUser populates in production. Mirrors the pattern in +// me_test.go (userCtxKeyForTest comes from auth_test.go). +func callEvents(h *handlers, user dbq.User, body []byte) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodPost, "/api/events", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req = req.WithContext(context.WithValue(req.Context(), userCtxKeyForTest(), user)) + w := httptest.NewRecorder() + h.handleEvents(w, req) + return w +} + +func TestHandleEvents_PlayStartedReturnsIDs(t *testing.T) { + h, pool := testHandlers(t) + truncateLibrary(t, pool) + user := seedUser(t, pool, "alice", "x", false) + artist := seedArtist(t, pool, "Beatles") + album := seedAlbum(t, pool, artist.ID, "Abbey Road", 1969) + track := seedTrack(t, pool, album.ID, artist.ID, "Something", 3, 183_000) + + body, _ := json.Marshal(map[string]any{ + "type": "play_started", + "track_id": uuidToString(track.ID), + }) + w := callEvents(h, user, body) + if w.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", w.Code, w.Body.String()) + } + var resp struct { + PlayEventID string `json:"play_event_id"` + SessionID string `json:"session_id"` + } + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode: %v", err) + } + if resp.PlayEventID == "" || resp.SessionID == "" { + t.Errorf("ids empty: %+v", resp) + } +} + +func TestHandleEvents_PlayEndedClosesRow(t *testing.T) { + h, pool := testHandlers(t) + truncateLibrary(t, pool) + user := seedUser(t, pool, "alice", "x", false) + artist := seedArtist(t, pool, "Beatles") + album := seedAlbum(t, pool, artist.ID, "Abbey Road", 1969) + track := seedTrack(t, pool, album.ID, artist.ID, "Something", 3, 183_000) + + startedID := postPlayStartedHelper(t, h, user, track.ID) + + body, _ := json.Marshal(map[string]any{ + "type": "play_ended", + "play_event_id": startedID, + "duration_played_ms": 180_000, + }) + w := callEvents(h, user, body) + if w.Code != http.StatusOK { + t.Fatalf("ended status = %d body=%s", w.Code, w.Body.String()) + } +} + +func TestHandleEvents_MissingTypeIs400(t *testing.T) { + h, pool := testHandlers(t) + user := seedUser(t, pool, "alice", "x", false) + body, _ := json.Marshal(map[string]any{"track_id": "00000000-0000-0000-0000-000000000000"}) + w := callEvents(h, user, body) + if w.Code != http.StatusBadRequest { + t.Errorf("status = %d", w.Code) + } +} + +func TestHandleEvents_PlayStartedWithUnknownTrackIs404(t *testing.T) { + h, pool := testHandlers(t) + user := seedUser(t, pool, "alice", "x", false) + body, _ := json.Marshal(map[string]any{ + "type": "play_started", + "track_id": "00000000-0000-0000-0000-000000000000", + }) + w := callEvents(h, user, body) + if w.Code != http.StatusNotFound { + t.Errorf("status = %d", w.Code) + } +} + +func TestHandleEvents_PlayEndedOnOtherUsersRowIs403(t *testing.T) { + h, pool := testHandlers(t) + truncateLibrary(t, pool) + alice := seedUser(t, pool, "alice", "x", false) + bob := seedUser(t, pool, "bob", "x", false) + artist := seedArtist(t, pool, "X") + album := seedAlbum(t, pool, artist.ID, "X", 1990) + track := seedTrack(t, pool, album.ID, artist.ID, "X", 1, 100_000) + + startedID := postPlayStartedHelper(t, h, alice, track.ID) + + body, _ := json.Marshal(map[string]any{ + "type": "play_ended", + "play_event_id": startedID, + "duration_played_ms": 50_000, + }) + w := callEvents(h, bob, body) + if w.Code != http.StatusForbidden { + t.Errorf("status = %d, want 403", w.Code) + } +} + +func postPlayStartedHelper(t *testing.T, h *handlers, user dbq.User, trackID pgtype.UUID) string { + t.Helper() + body, _ := json.Marshal(map[string]any{ + "type": "play_started", + "track_id": uuidToString(trackID), + }) + w := callEvents(h, user, body) + if w.Code != http.StatusOK { + t.Fatalf("play_started status = %d body=%s", w.Code, w.Body.String()) + } + var resp struct { + PlayEventID string `json:"play_event_id"` + } + _ = json.Unmarshal(w.Body.Bytes(), &resp) + return resp.PlayEventID +} diff --git a/internal/api/library_test.go b/internal/api/library_test.go index d0aff688..d8822c14 100644 --- a/internal/api/library_test.go +++ b/internal/api/library_test.go @@ -7,6 +7,8 @@ import ( "testing" "github.com/go-chi/chi/v5" + + "git.fabledsword.com/bvandeusen/minstrel/internal/config" ) // newLibraryRouter builds a test-only chi router with the library handlers @@ -433,7 +435,11 @@ func TestHandleListArtists_EmptyDB(t *testing.T) { func TestRoutesRegisteredInMount(t *testing.T) { h, _ := testHandlers(t) r := chi.NewRouter() - Mount(r, h.pool, h.logger) + Mount(r, h.pool, h.logger, config.EventsConfig{ + SessionTimeoutMinutes: 30, + SkipMaxCompletionRatio: 0.5, + SkipMaxDurationPlayedMs: 30000, + }) paths := []string{ "/api/artists", diff --git a/internal/server/server.go b/internal/server/server.go index 2515a019..097b66d0 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -13,6 +13,7 @@ import ( "git.fabledsword.com/bvandeusen/minstrel/internal/api" "git.fabledsword.com/bvandeusen/minstrel/internal/auth" + "git.fabledsword.com/bvandeusen/minstrel/internal/config" "git.fabledsword.com/bvandeusen/minstrel/internal/library" "git.fabledsword.com/bvandeusen/minstrel/internal/subsonic" "git.fabledsword.com/bvandeusen/minstrel/web" @@ -29,10 +30,11 @@ type Server struct { Pool *pgxpool.Pool Scanner ScanTrigger SubsonicCfg subsonic.Config + EventsCfg config.EventsConfig } -func New(logger *slog.Logger, pool *pgxpool.Pool, scanner ScanTrigger, subCfg subsonic.Config) *Server { - return &Server{Logger: logger, Pool: pool, Scanner: scanner, SubsonicCfg: subCfg} +func New(logger *slog.Logger, pool *pgxpool.Pool, scanner ScanTrigger, subCfg subsonic.Config, eventsCfg config.EventsConfig) *Server { + return &Server{Logger: logger, Pool: pool, Scanner: scanner, SubsonicCfg: subCfg, EventsCfg: eventsCfg} } func (s *Server) Router() http.Handler { @@ -43,7 +45,7 @@ func (s *Server) Router() http.Handler { r.Get("/healthz", s.handleHealthz) if s.Pool != nil { - api.Mount(r, s.Pool, s.Logger) + api.Mount(r, s.Pool, s.Logger, s.EventsCfg) r.Route("/api/admin", func(admin chi.Router) { admin.Use(auth.RequireAdmin(s.Pool)) if s.Scanner != nil { diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 9c207b0d..66358ca4 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -9,11 +9,12 @@ import ( "strings" "testing" + "git.fabledsword.com/bvandeusen/minstrel/internal/config" "git.fabledsword.com/bvandeusen/minstrel/internal/subsonic" ) func TestHealthz(t *testing.T) { - s := New(slog.New(slog.NewTextHandler(io.Discard, nil)), nil, nil, subsonic.Config{}) + s := New(slog.New(slog.NewTextHandler(io.Discard, nil)), nil, nil, subsonic.Config{}, config.EventsConfig{}) ts := httptest.NewServer(s.Router()) defer ts.Close() @@ -36,7 +37,7 @@ func TestHealthz(t *testing.T) { } func TestRouter_ServesSPAAtRoot(t *testing.T) { - s := New(slog.New(slog.NewTextHandler(io.Discard, nil)), nil, nil, subsonic.Config{}) + s := New(slog.New(slog.NewTextHandler(io.Discard, nil)), nil, nil, subsonic.Config{}, config.EventsConfig{}) ts := httptest.NewServer(s.Router()) defer ts.Close() @@ -55,7 +56,7 @@ func TestRouter_ServesSPAAtRoot(t *testing.T) { } func TestRouter_DeepLinkFallbackReturnsSPA(t *testing.T) { - s := New(slog.New(slog.NewTextHandler(io.Discard, nil)), nil, nil, subsonic.Config{}) + s := New(slog.New(slog.NewTextHandler(io.Discard, nil)), nil, nil, subsonic.Config{}, config.EventsConfig{}) ts := httptest.NewServer(s.Router()) defer ts.Close() @@ -74,7 +75,7 @@ func TestRouter_DeepLinkFallbackReturnsSPA(t *testing.T) { } func TestRouter_APIPathNotSwallowedBySPA(t *testing.T) { - s := New(slog.New(slog.NewTextHandler(io.Discard, nil)), nil, nil, subsonic.Config{}) + s := New(slog.New(slog.NewTextHandler(io.Discard, nil)), nil, nil, subsonic.Config{}, config.EventsConfig{}) ts := httptest.NewServer(s.Router()) defer ts.Close() @@ -93,7 +94,7 @@ func TestRouter_APIPathNotSwallowedBySPA(t *testing.T) { } func TestRouter_RestPathNotSwallowedBySPA(t *testing.T) { - s := New(slog.New(slog.NewTextHandler(io.Discard, nil)), nil, nil, subsonic.Config{}) + s := New(slog.New(slog.NewTextHandler(io.Discard, nil)), nil, nil, subsonic.Config{}, config.EventsConfig{}) ts := httptest.NewServer(s.Router()) defer ts.Close()