bvandeusen/minstrel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 38 Wiki Activity

feat(api): add POST /api/events handler

Browse Source 
Discriminated-union JSON over three event types: play_started,
play_ended, play_skipped. Auth via existing RequireUser. play_started
returns play_event_id + session_id; the other two return { ok: true }.
Validates ownership: play_ended/play_skipped on another user's row
returns 403. Mount signature now takes EventsConfig and constructs
the playevents.Writer; server.New propagates it through.
This commit is contained in:
Bryan Van Deusen
2026-04-26 00:12:06 -04:00
parent 0c7e1b0408
commit f4e73b81b4
8 changed files with 343 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/minstrel/main.go
+1 -1
View File
@@ -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(),
internal/api/api.go
+13 -2
View File
@@ -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
}
internal/api/auth_test.go
+6 -2
View File
@@ -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 {
internal/api/events.go
+168
View File
@@ -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})
}
internal/api/events_test.go
+137
View File
@@ -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
}
internal/api/library_test.go
+7 -1
View File
@@ -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",
internal/server/server.go
+5 -3
View File
@@ -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 {
internal/server/server_test.go
+6 -5
View File
@@ -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()