feat(subsonic): wire /rest/scrobble into playevents.Writer

submission=false → play_started (replaces in-memory nowPlayingMap).
submission=true (default) → synthetic completed play. The nowPlayingMap
struct + factory + the nowPlaying field on mediaHandlers are deleted;
play_events WHERE ended_at IS NULL is now the source of truth for
'currently playing,' reachable from M3+ work.

api.Mount now takes the shared *playevents.Writer instead of cfg so
the same writer instance feeds both surfaces.

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.
+// RequireUser; everything else is gated by the middleware. The events writer
-func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, cfg config.EventsConfig) {
+// is shared with the Subsonic mount so /rest/scrobble feeds the same store.
-	w := playevents.NewWriter(
+func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playevents.Writer) {
-		pool, logger,
+	h := &handlers{pool: pool, logger: logger, events: events}
-		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)

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{
+	w := playevents.NewWriter(h.pool, slog.New(slog.NewTextHandler(io.Discard, nil)),
-		SessionTimeoutMinutes:   30,
+		30*time.Minute, 0.5, 30000)
-		SkipMaxCompletionRatio:  0.5,
+	Mount(r, h.pool, h.logger, w)
-		SkipMaxDurationPlayedMs: 30000,
-	})
 
 	paths := []string{
 		"/api/artists",

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()

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)
+	}
+}

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
+// getCoverArt, scrobble. Scrobble routes through playevents.Writer so
-// because M1 has no event ingestion yet (see M2).
+// Subsonic clients feed the same play_events table as the native /api/events.
 type mediaHandlers struct {
 	pool   *pgxpool.Pool
-	nowPlaying *nowPlayingMap
+	events *playevents.Writer
 }
 
-func newMediaHandlers(pool *pgxpool.Pool) *mediaHandlers {
+func newMediaHandlers(pool *pgxpool.Pool, events *playevents.Writer) *mediaHandlers {
-	return &mediaHandlers{pool: pool, nowPlaying: newNowPlayingMap()}
+	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
+// handleScrobble translates Subsonic scrobble calls into native play_events.
-// event ingestion, so submission=true is a no-op that still returns ok — the
+//
-// real play/skip/seek/complete events land in M2.
+// 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
 	}
+	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")
 
-	// Per spec, submission defaults to true. Only false means "this is now
+	// Subsonic default is submission=true (completed play).
-	// playing, don't record a play event." Anything else is an M2 no-op.
 	submission := strings.ToLower(params.Get("submission"))
-	if submission == "false" {
+	switch submission {
-		if user, ok := UserFromContext(r.Context()); ok {
+	case "true", "":
-			m.nowPlaying.Set(uuidToID(user.ID), trackID)
+		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
-}

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")
-	}
-}

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)