diff --git a/internal/api/api.go b/internal/api/api.go index 0ad95e11..ebae8648 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -6,26 +6,19 @@ 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, 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} +// RequireUser; everything else is gated by the middleware. The events writer +// is shared with the Subsonic mount so /rest/scrobble feeds the same store. +func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playevents.Writer) { + h := &handlers{pool: pool, logger: logger, events: events} r.Route("/api", func(api chi.Router) { api.Post("/auth/login", h.handleLogin) diff --git a/internal/api/library_test.go b/internal/api/library_test.go index d8822c14..0c970582 100644 --- a/internal/api/library_test.go +++ b/internal/api/library_test.go @@ -2,13 +2,16 @@ package api import ( "encoding/json" + "io" + "log/slog" "net/http" "net/http/httptest" "testing" + "time" "github.com/go-chi/chi/v5" - "git.fabledsword.com/bvandeusen/minstrel/internal/config" + "git.fabledsword.com/bvandeusen/minstrel/internal/playevents" ) // newLibraryRouter builds a test-only chi router with the library handlers @@ -435,11 +438,9 @@ func TestHandleListArtists_EmptyDB(t *testing.T) { func TestRoutesRegisteredInMount(t *testing.T) { h, _ := testHandlers(t) r := chi.NewRouter() - Mount(r, h.pool, h.logger, config.EventsConfig{ - SessionTimeoutMinutes: 30, - SkipMaxCompletionRatio: 0.5, - SkipMaxDurationPlayedMs: 30000, - }) + w := playevents.NewWriter(h.pool, slog.New(slog.NewTextHandler(io.Discard, nil)), + 30*time.Minute, 0.5, 30000) + Mount(r, h.pool, h.logger, w) paths := []string{ "/api/artists", diff --git a/internal/server/server.go b/internal/server/server.go index 097b66d0..09a6c331 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -11,10 +11,13 @@ import ( "github.com/go-chi/chi/v5/middleware" "github.com/jackc/pgx/v5/pgxpool" + "time" + "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/playevents" "git.fabledsword.com/bvandeusen/minstrel/internal/subsonic" "git.fabledsword.com/bvandeusen/minstrel/web" ) @@ -45,14 +48,20 @@ func (s *Server) Router() http.Handler { r.Get("/healthz", s.handleHealthz) if s.Pool != nil { - api.Mount(r, s.Pool, s.Logger, s.EventsCfg) + writer := playevents.NewWriter( + s.Pool, s.Logger, + time.Duration(s.EventsCfg.SessionTimeoutMinutes)*time.Minute, + s.EventsCfg.SkipMaxCompletionRatio, + s.EventsCfg.SkipMaxDurationPlayedMs, + ) + api.Mount(r, s.Pool, s.Logger, writer) r.Route("/api/admin", func(admin chi.Router) { admin.Use(auth.RequireAdmin(s.Pool)) if s.Scanner != nil { admin.Post("/scan", s.handleAdminScan) } }) - subsonic.Mount(r, s.Pool, s.Logger, s.SubsonicCfg) + subsonic.Mount(r, s.Pool, s.Logger, s.SubsonicCfg, writer) } spa := web.Handler() diff --git a/internal/subsonic/scrobble_test.go b/internal/subsonic/scrobble_test.go new file mode 100644 index 00000000..2c2e1876 --- /dev/null +++ b/internal/subsonic/scrobble_test.go @@ -0,0 +1,119 @@ +package subsonic + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db" + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + "git.fabledsword.com/bvandeusen/minstrel/internal/playevents" +) + +func testScrobblePool(t *testing.T) (*pgxpool.Pool, dbq.User, dbq.Track) { + t.Helper() + if testing.Short() { + t.Skip("skipping in -short mode") + } + dsn := os.Getenv("MINSTREL_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("MINSTREL_TEST_DATABASE_URL not set") + } + if err := db.Migrate(dsn, slog.New(slog.NewTextHandler(io.Discard, nil))); err != nil { + t.Fatalf("migrate: %v", err) + } + pool, err := pgxpool.New(context.Background(), dsn) + if err != nil { + t.Fatalf("pool: %v", err) + } + t.Cleanup(pool.Close) + if _, err := pool.Exec(context.Background(), + "TRUNCATE play_events, skip_events, play_sessions, sessions, users, tracks, albums, artists RESTART IDENTITY CASCADE"); err != nil { + t.Fatalf("truncate: %v", err) + } + q := dbq.New(pool) + u, err := q.CreateUser(context.Background(), dbq.CreateUserParams{ + Username: "tester", PasswordHash: "x", ApiToken: "x", IsAdmin: false, + }) + if err != nil { + t.Fatalf("user: %v", err) + } + a, _ := q.UpsertArtist(context.Background(), dbq.UpsertArtistParams{Name: "X", SortName: "X"}) + al, _ := q.UpsertAlbum(context.Background(), dbq.UpsertAlbumParams{Title: "X", SortTitle: "X", ArtistID: a.ID}) + tr, _ := q.UpsertTrack(context.Background(), dbq.UpsertTrackParams{ + Title: "Y", AlbumID: al.ID, ArtistID: a.ID, FilePath: "/tmp/scrob.flac", DurationMs: 100_000, + }) + return pool, u, tr +} + +func newScrobbleHandlers(pool *pgxpool.Pool) *mediaHandlers { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + w := playevents.NewWriter(pool, logger, 30*time.Minute, 0.5, 30000) + return newMediaHandlers(pool, w) +} + +func TestHandleScrobble_SubmissionFalseInsertsOpenPlayEvent(t *testing.T) { + pool, user, track := testScrobblePool(t) + m := newScrobbleHandlers(pool) + + q := url.Values{} + q.Set("id", uuidToID(track.ID)) + q.Set("submission", "false") + req := httptest.NewRequest(http.MethodGet, "/rest/scrobble?"+q.Encode(), nil) + ctx := context.WithValue(req.Context(), userCtxKey, user) + w := httptest.NewRecorder() + m.handleScrobble(w, req.WithContext(ctx)) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d", w.Code) + } + rows, _ := pool.Query(context.Background(), + "SELECT id FROM play_events WHERE user_id=$1 AND ended_at IS NULL", user.ID) + defer rows.Close() + count := 0 + for rows.Next() { + count++ + } + if count != 1 { + t.Errorf("open play_events count = %d, want 1", count) + } +} + +func TestHandleScrobble_SubmissionTrueWritesCompletedPlay(t *testing.T) { + pool, user, track := testScrobblePool(t) + m := newScrobbleHandlers(pool) + + q := url.Values{} + q.Set("id", uuidToID(track.ID)) + q.Set("submission", "true") + req := httptest.NewRequest(http.MethodGet, "/rest/scrobble?"+q.Encode(), nil) + ctx := context.WithValue(req.Context(), userCtxKey, user) + w := httptest.NewRecorder() + m.handleScrobble(w, req.WithContext(ctx)) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d", w.Code) + } + rows, _ := pool.Query(context.Background(), + "SELECT ended_at IS NOT NULL FROM play_events WHERE user_id=$1", user.ID) + defer rows.Close() + closed := 0 + for rows.Next() { + var c bool + _ = rows.Scan(&c) + if c { + closed++ + } + } + if closed != 1 { + t.Errorf("closed play_events = %d, want 1", closed) + } +} diff --git a/internal/subsonic/stream.go b/internal/subsonic/stream.go index f15f6861..e05d73df 100644 --- a/internal/subsonic/stream.go +++ b/internal/subsonic/stream.go @@ -7,27 +7,27 @@ import ( "net/http" "os" "path/filepath" + "strconv" "strings" - "sync" "time" "github.com/jackc/pgx/v5" - "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + "git.fabledsword.com/bvandeusen/minstrel/internal/playevents" ) // mediaHandlers serves the bytes-on-the-wire endpoints: stream, download, -// getCoverArt, scrobble. They share a pool and an in-memory now-playing map -// because M1 has no event ingestion yet (see M2). +// getCoverArt, scrobble. Scrobble routes through playevents.Writer so +// Subsonic clients feed the same play_events table as the native /api/events. type mediaHandlers struct { - pool *pgxpool.Pool - nowPlaying *nowPlayingMap + pool *pgxpool.Pool + events *playevents.Writer } -func newMediaHandlers(pool *pgxpool.Pool) *mediaHandlers { - return &mediaHandlers{pool: pool, nowPlaying: newNowPlayingMap()} +func newMediaHandlers(pool *pgxpool.Pool, events *playevents.Writer) *mediaHandlers { + return &mediaHandlers{pool: pool, events: events} } // handleStream serves the raw track bytes. http.ServeContent gives us Range, @@ -196,9 +196,12 @@ func imageContentType(path string) string { } } -// handleScrobble records a now-playing hint when submission=false. M1 has no -// event ingestion, so submission=true is a no-op that still returns ok — the -// real play/skip/seek/complete events land in M2. +// handleScrobble translates Subsonic scrobble calls into native play_events. +// +// submission=false (now-playing): writes a play_started, leaves ended_at NULL. +// submission=true (completed): writes a synthetic completed play (start+end +// in one transaction). play_events WHERE ended_at IS NULL is the source of +// truth for "currently playing"; the prior in-memory nowPlayingMap is gone. func (m *mediaHandlers) handleScrobble(w http.ResponseWriter, r *http.Request) { params := r.URL.Query() idStr := params.Get("id") @@ -211,45 +214,38 @@ func (m *mediaHandlers) handleScrobble(w http.ResponseWriter, r *http.Request) { WriteFail(w, r, ErrDataNotFound, "Track not found") return } - - // Per spec, submission defaults to true. Only false means "this is now - // playing, don't record a play event." Anything else is an M2 no-op. - submission := strings.ToLower(params.Get("submission")) - if submission == "false" { - if user, ok := UserFromContext(r.Context()); ok { - m.nowPlaying.Set(uuidToID(user.ID), trackID) + user, ok := UserFromContext(r.Context()) + if !ok { + WriteFail(w, r, ErrGeneric, "Unauthenticated") + return + } + at := time.Now().UTC() + if t := params.Get("time"); t != "" { + // Subsonic spec uses ms-since-epoch; some clients send ISO 8601. + // Accept both for resilience. + if ms, err := strconv.ParseInt(t, 10, 64); err == nil { + at = time.UnixMilli(ms).UTC() + } else if parsed, err := time.Parse(time.RFC3339, t); err == nil { + at = parsed } } + clientID := params.Get("c") + // Subsonic default is submission=true (completed play). + submission := strings.ToLower(params.Get("submission")) + switch submission { + case "true", "": + if err := m.events.RecordSyntheticCompletedPlay(r.Context(), user.ID, trackID, clientID, at); err != nil { + WriteFail(w, r, ErrGeneric, "Could not record play") + return + } + case "false": + if _, err := m.events.RecordPlayStarted(r.Context(), user.ID, trackID, clientID, at); err != nil { + WriteFail(w, r, ErrGeneric, "Could not record now-playing") + return + } + default: + // Unknown submission values: ack and ignore. + } Write(w, r, struct{ Envelope }{Envelope: NewEnvelope("ok")}) } - -// nowPlayingMap is a tiny in-memory store keyed by user id. M1 doesn't expose -// a getNowPlaying endpoint yet; this structure exists so scrobble has a place -// to write and M2 has something to read. -type nowPlayingMap struct { - mu sync.RWMutex - m map[string]nowPlayingEntry -} - -type nowPlayingEntry struct { - TrackID pgtype.UUID - At time.Time -} - -func newNowPlayingMap() *nowPlayingMap { - return &nowPlayingMap{m: make(map[string]nowPlayingEntry)} -} - -func (n *nowPlayingMap) Set(userID string, trackID pgtype.UUID) { - n.mu.Lock() - defer n.mu.Unlock() - n.m[userID] = nowPlayingEntry{TrackID: trackID, At: time.Now()} -} - -func (n *nowPlayingMap) Get(userID string) (nowPlayingEntry, bool) { - n.mu.RLock() - defer n.mu.RUnlock() - e, ok := n.m[userID] - return e, ok -} diff --git a/internal/subsonic/stream_test.go b/internal/subsonic/stream_test.go index 67fddf24..9452913a 100644 --- a/internal/subsonic/stream_test.go +++ b/internal/subsonic/stream_test.go @@ -4,8 +4,6 @@ import ( "os" "path/filepath" "testing" - - "github.com/jackc/pgx/v5/pgtype" ) func TestFindSidecarCover(t *testing.T) { @@ -53,26 +51,3 @@ func TestImageContentType(t *testing.T) { } } } - -func TestNowPlayingMap(t *testing.T) { - m := newNowPlayingMap() - user := "user-1" - if _, ok := m.Get(user); ok { - t.Fatalf("unexpected entry on fresh map") - } - - var trackID pgtype.UUID - _ = trackID.Scan("6ba7b810-9dad-11d1-80b4-00c04fd430c8") - m.Set(user, trackID) - - entry, ok := m.Get(user) - if !ok { - t.Fatalf("Set did not persist") - } - if uuidToID(entry.TrackID) != "6ba7b810-9dad-11d1-80b4-00c04fd430c8" { - t.Errorf("trackID round-trip broke: %q", uuidToID(entry.TrackID)) - } - if entry.At.IsZero() { - t.Errorf("At timestamp not set") - } -} diff --git a/internal/subsonic/subsonic.go b/internal/subsonic/subsonic.go index 524a08a1..ed3f0069 100644 --- a/internal/subsonic/subsonic.go +++ b/internal/subsonic/subsonic.go @@ -11,15 +11,17 @@ import ( "github.com/go-chi/chi/v5" "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/playevents" ) // Mount attaches Subsonic handlers at /rest on r. Endpoints are exposed at // both /rest/foo and /rest/foo.view because client conventions vary. Both // GET and POST are accepted; Subsonic's auth params live in the query string // either way. -func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, cfg Config) { +func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, cfg Config, events *playevents.Writer) { b := &browseHandlers{pool: pool} - m := newMediaHandlers(pool) + m := newMediaHandlers(pool, events) r.Route("/rest", func(sub chi.Router) { sub.Use(Middleware(pool, cfg)) register(sub, "/ping", handlePing)