From 7df48ee9bc93c01ced1624209d0250aec5a15d22 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 25 Apr 2026 19:56:58 -0400 Subject: [PATCH] docs(plan): add M2 events sub-plan implementation plan 9 tasks: schema migration, events config, playsessions service, playevents writer, /api/events handler, Subsonic scrobble integration, web clientId + events dispatcher, layout wiring, final verification. Tightly TDD-shaped with full code in every step. --- .../superpowers/plans/2026-04-25-m2-events.md | 2390 +++++++++++++++++ 1 file changed, 2390 insertions(+) create mode 100644 docs/superpowers/plans/2026-04-25-m2-events.md diff --git a/docs/superpowers/plans/2026-04-25-m2-events.md b/docs/superpowers/plans/2026-04-25-m2-events.md new file mode 100644 index 00000000..95bf4884 --- /dev/null +++ b/docs/superpowers/plans/2026-04-25-m2-events.md @@ -0,0 +1,2390 @@ +# M2 Events Sub-Plan Implementation + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Ship the play-event ingestion pipeline end-to-end (schema, session service, `POST /api/events` handler, Subsonic `/rest/scrobble` integration, SvelteKit player wiring) so M3's recommendation engine has play history data to consume. + +**Architecture:** New tables `play_sessions`, `play_events`, `skip_events` per spec §5. Two new Go packages: `internal/playsessions` (FindOrCreate with 30-min rolling rule) and `internal/playevents` (RecordPlayStarted/Ended/Skipped/SyntheticCompletedPlay, owning the auto-close-prior + skip-classification logic). One HTTP handler at `POST /api/events` and one Subsonic handler revision. Web client gets a `useEventsDispatcher` Svelte module that watches the player store and posts events; `pagehide` uses `navigator.sendBeacon`. + +**Tech Stack:** Go 1.23 + chi + sqlc + pgx/v5. SvelteKit 2 + Svelte 5 (runes) + TypeScript + Vitest + Testing Library. + +**Reference:** design spec at `docs/superpowers/specs/2026-04-25-m2-events-design.md`. + +**Naming note:** the table is `play_sessions` (not `sessions` — `sessions` already holds HTTP auth tokens from migration 0004). The Go package is `internal/playsessions`. + +--- + +## File Structure + +**New server files:** + +| File | Responsibility | +|---|---| +| `internal/db/migrations/0005_events.up.sql` + `.down.sql` | Schema for `play_sessions`, `play_events`, `skip_events`. | +| `internal/db/queries/events.sql` | sqlc queries for the new tables. Generated bindings emitted into `internal/db/dbq/events.sql.go`. | +| `internal/playsessions/service.go` | `FindOrCreate(ctx, q, userID, eventTime, clientID, timeout) (sessionID, error)`. Composable into a transaction. | +| `internal/playsessions/service_test.go` | Live-DB tests for the four scenarios (no prior, within window, beyond window, race). | +| `internal/playevents/writer.go` | `RecordPlayStarted`, `RecordPlayEnded`, `RecordPlaySkipped`, `RecordSyntheticCompletedPlay`. Owns auto-close-prior, skip classification rule, skip_events writes. | +| `internal/playevents/writer_test.go` | Direct tests covering rule edges, auto-close, synthetic write. | +| `internal/api/events.go` | `handleEvents` HTTP handler dispatching by `type`. | +| `internal/api/events_test.go` | HTTP-level coverage for happy paths, validation errors, ownership-mismatch. | +| `internal/subsonic/scrobble_test.go` | New scrobble tests (existing file has none). | + +**Modified server files:** + +| File | Change | +|---|---| +| `internal/api/api.go` | Register `authed.Post("/events", h.handleEvents)`. | +| `internal/subsonic/stream.go` | `handleScrobble` calls `playevents.Writer` instead of `nowPlayingMap`. Delete `nowPlayingMap` struct + factory + the `nowPlaying` field. Constructor signature simplifies. | +| `internal/config/config.go` + `config.example.yaml` | Add `EventsConfig` struct: `SessionTimeoutMinutes` (default 30), `SkipMaxCompletionRatio` (default 0.5), `SkipMaxDurationPlayedMs` (default 30000). Wire into `Default()` and YAML loader. | + +**New web files:** + +| File | Responsibility | +|---|---| +| `web/src/lib/player/clientId.ts` | `getOrCreateClientId(): string` — sessionStorage-backed UUID. | +| `web/src/lib/player/clientId.test.ts` | Cover create + reuse. | +| `web/src/lib/player/events.svelte.ts` | `useEventsDispatcher()` — `$effect`-based watcher on player state; owns `_playEventId`; calls `api.post`; installs `pagehide` listener. | +| `web/src/lib/player/events.svelte.test.ts` | State-machine tests with mocked `api.post` and spied `navigator.sendBeacon`. | + +**Modified web files:** + +| File | Change | +|---|---| +| `web/src/lib/api/types.ts` | Add `EventRequest`, `EventResponse` types. | +| `web/src/routes/+layout.svelte` | Call `useEventsDispatcher()` once at mount, alongside the existing `useMediaSession()`. | + +--- + +## Task 1: Schema migration + sqlc queries + +**Files:** +- Create: `internal/db/migrations/0005_events.up.sql` +- Create: `internal/db/migrations/0005_events.down.sql` +- Create: `internal/db/queries/events.sql` +- Generated: `internal/db/dbq/events.sql.go` (via `sqlc generate`) + +This task ships the bare schema + queries. Tests ride along with later tasks; the verification here is `go build` succeeds after `sqlc generate` and a manual migration smoke check. + +- [ ] **Step 1: Write the up migration** + +Create `internal/db/migrations/0005_events.up.sql`: + +```sql +-- M2 listening telemetry: play_sessions, play_events, skip_events. +-- Note the table is named play_sessions (not sessions — that table already +-- exists from migration 0004 and holds HTTP auth sessions). Listening +-- sessions are an entirely separate concept driven by spec §6. +-- +-- session_vector_at_play and contextual_likes ship as nullable now even +-- though M2 writes NULL; M3 will populate them without a follow-up +-- migration. + +CREATE TABLE play_sessions ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE, + started_at timestamptz NOT NULL, + ended_at timestamptz, + last_event_at timestamptz NOT NULL, + track_count integer NOT NULL DEFAULT 0, + client_id text +); +CREATE INDEX play_sessions_user_last_event_idx ON play_sessions (user_id, last_event_at DESC); + +CREATE TABLE play_events ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE, + track_id uuid NOT NULL REFERENCES tracks(id) ON DELETE CASCADE, + session_id uuid NOT NULL REFERENCES play_sessions(id) ON DELETE CASCADE, + started_at timestamptz NOT NULL, + ended_at timestamptz, + duration_played_ms integer, + completion_ratio double precision, + was_skipped boolean NOT NULL DEFAULT false, + client_id text, + session_vector_at_play jsonb, + scrobbled_at timestamptz +); +CREATE INDEX play_events_user_started_idx ON play_events (user_id, started_at DESC); +CREATE INDEX play_events_user_track_idx ON play_events (user_id, track_id); + +CREATE TABLE skip_events ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE, + track_id uuid NOT NULL REFERENCES tracks(id) ON DELETE CASCADE, + session_id uuid NOT NULL REFERENCES play_sessions(id) ON DELETE CASCADE, + skipped_at timestamptz NOT NULL, + position_ms integer NOT NULL +); +CREATE INDEX skip_events_user_skipped_idx ON skip_events (user_id, skipped_at DESC); +``` + +- [ ] **Step 2: Write the down migration** + +Create `internal/db/migrations/0005_events.down.sql`: + +```sql +DROP TABLE IF EXISTS skip_events; +DROP TABLE IF EXISTS play_events; +DROP TABLE IF EXISTS play_sessions; +``` + +- [ ] **Step 3: Write the sqlc queries** + +Create `internal/db/queries/events.sql`: + +```sql +-- name: GetMostRecentPlaySessionForUser :one +SELECT * FROM play_sessions +WHERE user_id = $1 +ORDER BY last_event_at DESC +LIMIT 1; + +-- name: InsertPlaySession :one +INSERT INTO play_sessions (user_id, started_at, last_event_at, client_id) +VALUES ($1, $2, $2, $3) +RETURNING *; + +-- name: TouchPlaySessionLastEvent :exec +UPDATE play_sessions +SET last_event_at = $2, + track_count = track_count + 1 +WHERE id = $1; + +-- name: GetOpenPlayEventForUser :one +-- Returns the most recent play_event for a user where ended_at IS NULL. +-- Used by the auto-close-prior step in playevents.RecordPlayStarted. +SELECT * FROM play_events +WHERE user_id = $1 AND ended_at IS NULL +ORDER BY started_at DESC +LIMIT 1; + +-- name: InsertPlayEvent :one +INSERT INTO play_events ( + user_id, track_id, session_id, started_at, client_id +) VALUES ($1, $2, $3, $4, $5) +RETURNING *; + +-- name: UpdatePlayEventEnded :one +-- Closes a play_event by id with the given ended_at, duration, and skip flag. +-- completion_ratio is computed from duration_played_ms and the track duration +-- (looked up by the caller — sqlc doesn't do joins on UPDATE). +UPDATE play_events +SET ended_at = $2, + duration_played_ms = $3, + completion_ratio = $4, + was_skipped = $5 +WHERE id = $1 +RETURNING *; + +-- name: GetPlayEventByID :one +SELECT * FROM play_events WHERE id = $1; + +-- name: InsertSkipEvent :one +INSERT INTO skip_events (user_id, track_id, session_id, skipped_at, position_ms) +VALUES ($1, $2, $3, $4, $5) +RETURNING *; +``` + +- [ ] **Step 4: Generate sqlc bindings** + +Run: `sqlc generate` +Expected: writes `internal/db/dbq/events.sql.go` with `PlaySession`, `PlayEvent`, `SkipEvent` model types + the named query functions. + +- [ ] **Step 5: Verify `go build` works** + +Run: `go build ./...` +Expected: builds clean. + +- [ ] **Step 6: Manual migration smoke** + +Bring up postgres if needed: `docker compose up -d postgres` + +Run: +```bash +MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' \ + go test ./internal/db -run TestMigrate -v +``` +Expected: PASS. (Existing migration test covers the up/down round trip generically; it will exercise 0005 automatically.) + +- [ ] **Step 7: Commit** + +```bash +git add internal/db/migrations/0005_events.up.sql \ + internal/db/migrations/0005_events.down.sql \ + internal/db/queries/events.sql \ + internal/db/dbq/events.sql.go +git commit -m "feat(db): add M2 schema — play_sessions, play_events, skip_events + +Tables and indexes per spec §5. session_vector_at_play ships nullable +so M3 doesn't need a follow-up migration. Table is named play_sessions +to avoid collision with the existing sessions (HTTP auth) table from +migration 0004." +``` + +--- + +## Task 2: Events config struct + +**Files:** +- Modify: `internal/config/config.go` +- Modify: `config.example.yaml` + +- [ ] **Step 1: Open `internal/config/config.go`, add the EventsConfig struct after SubsonicConfig** + +Insert after the `SubsonicConfig` block: + +```go +// EventsConfig governs play-event ingestion: session window, skip rule. +type EventsConfig struct { + SessionTimeoutMinutes int `yaml:"session_timeout_minutes"` + SkipMaxCompletionRatio float64 `yaml:"skip_max_completion_ratio"` + SkipMaxDurationPlayedMs int `yaml:"skip_max_duration_played_ms"` +} +``` + +- [ ] **Step 2: Add Events field to `Config` struct and Default()** + +Find the `type Config struct` block. Add `Events EventsConfig \`yaml:"events"\`` to it. + +Find `func Default() Config` and update: + +```go +func Default() Config { + return Config{ + Server: ServerConfig{Address: ":4533"}, + Database: DatabaseConfig{URL: ""}, + Log: LogConfig{Level: "info", Format: "text"}, + Events: EventsConfig{ + SessionTimeoutMinutes: 30, + SkipMaxCompletionRatio: 0.5, + SkipMaxDurationPlayedMs: 30000, + }, + } +} +``` + +- [ ] **Step 3: Append the yaml example** + +Append to `config.example.yaml`: + +```yaml + +events: + # Idle gap (minutes) between play_events that closes a play_session and + # starts a new one. Spec §6. + session_timeout_minutes: 30 + # Skip rule (spec §6): a play counts as a SKIP if completion ratio is below + # this threshold AND duration played (ms) is below the next threshold. + skip_max_completion_ratio: 0.5 + skip_max_duration_played_ms: 30000 +``` + +- [ ] **Step 4: Run config tests** + +Run: `go test ./internal/config -v` +Expected: PASS (no new tests needed; existing tests should still pass with the additional field defaulting correctly). + +- [ ] **Step 5: Commit** + +```bash +git add internal/config/config.go config.example.yaml +git commit -m "feat(config): add events section (session timeout + skip rule thresholds)" +``` + +--- + +## Task 3: Play sessions service + +**Files:** +- Create: `internal/playsessions/service.go` +- Create: `internal/playsessions/service_test.go` + +- [ ] **Step 1: Write the failing tests** + +Create `internal/playsessions/service_test.go`: + +```go +package playsessions + +import ( + "context" + "io" + "log/slog" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db" + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" +) + +func testPool(t *testing.T) *pgxpool.Pool { + t.Helper() + if testing.Short() { + t.Skip("skipping integration test 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 RESTART IDENTITY CASCADE"); err != nil { + t.Fatalf("truncate: %v", err) + } + return pool +} + +func seedTestUser(t *testing.T, pool *pgxpool.Pool) pgtype.UUID { + t.Helper() + u, err := dbq.New(pool).CreateUser(context.Background(), dbq.CreateUserParams{ + Username: "tester", + PasswordHash: "x", + ApiToken: "x", + IsAdmin: false, + }) + if err != nil { + t.Fatalf("CreateUser: %v", err) + } + return u.ID +} + +func TestFindOrCreate_NoPriorSessionCreatesOne(t *testing.T) { + pool := testPool(t) + user := seedTestUser(t, pool) + q := dbq.New(pool) + + now := time.Date(2026, 4, 25, 12, 0, 0, 0, time.UTC) + id, err := FindOrCreate(context.Background(), q, user, now, "client-a", 30*time.Minute) + if err != nil { + t.Fatalf("FindOrCreate: %v", err) + } + if !id.Valid { + t.Fatalf("session id not valid") + } + row, err := q.GetMostRecentPlaySessionForUser(context.Background(), user) + if err != nil { + t.Fatalf("get session: %v", err) + } + if row.ID != id { + t.Errorf("returned id %v, db has %v", id, row.ID) + } + if !row.StartedAt.Time.Equal(now) { + t.Errorf("started_at = %v, want %v", row.StartedAt.Time, now) + } +} + +func TestFindOrCreate_WithinWindowExtendsExisting(t *testing.T) { + pool := testPool(t) + user := seedTestUser(t, pool) + q := dbq.New(pool) + + now := time.Date(2026, 4, 25, 12, 0, 0, 0, time.UTC) + first, err := FindOrCreate(context.Background(), q, user, now, "client-a", 30*time.Minute) + if err != nil { + t.Fatalf("first: %v", err) + } + second, err := FindOrCreate(context.Background(), q, user, now.Add(15*time.Minute), "client-a", 30*time.Minute) + if err != nil { + t.Fatalf("second: %v", err) + } + if first != second { + t.Errorf("session id changed: %v -> %v", first, second) + } + row, err := q.GetMostRecentPlaySessionForUser(context.Background(), user) + if err != nil { + t.Fatalf("get session: %v", err) + } + if !row.LastEventAt.Time.Equal(now.Add(15 * time.Minute)) { + t.Errorf("last_event_at = %v", row.LastEventAt.Time) + } + if row.TrackCount != 2 { + t.Errorf("track_count = %d, want 2", row.TrackCount) + } +} + +func TestFindOrCreate_BeyondWindowStartsNew(t *testing.T) { + pool := testPool(t) + user := seedTestUser(t, pool) + q := dbq.New(pool) + + now := time.Date(2026, 4, 25, 12, 0, 0, 0, time.UTC) + first, err := FindOrCreate(context.Background(), q, user, now, "client-a", 30*time.Minute) + if err != nil { + t.Fatalf("first: %v", err) + } + second, err := FindOrCreate(context.Background(), q, user, now.Add(31*time.Minute), "client-a", 30*time.Minute) + if err != nil { + t.Fatalf("second: %v", err) + } + if first == second { + t.Errorf("expected new session, got same id %v", first) + } +} + +func TestFindOrCreate_AtExactBoundaryExtendsExisting(t *testing.T) { + pool := testPool(t) + user := seedTestUser(t, pool) + q := dbq.New(pool) + + now := time.Date(2026, 4, 25, 12, 0, 0, 0, time.UTC) + first, _ := FindOrCreate(context.Background(), q, user, now, "client-a", 30*time.Minute) + second, err := FindOrCreate(context.Background(), q, user, now.Add(30*time.Minute), "client-a", 30*time.Minute) + if err != nil { + t.Fatalf("second: %v", err) + } + if first != second { + t.Errorf("expected same session at exact boundary, got new id %v", second) + } +} +``` + +- [ ] **Step 2: Run tests, confirm they fail** + +Run: +```bash +MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' \ + go test ./internal/playsessions -v +``` +Expected: FAIL — package `playsessions` does not exist or `FindOrCreate` undefined. + +- [ ] **Step 3: Implement `internal/playsessions/service.go`** + +```go +// Package playsessions implements the listening-session lifecycle from +// spec §6: a session is a contiguous sequence of plays by a user with no +// gap longer than the configured timeout (default 30 minutes) between +// events. +package playsessions + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" +) + +// FindOrCreate returns the session id for the given user and event time. +// If the user's most recent session was last touched within `timeout`, that +// session is extended (last_event_at and track_count updated). Otherwise a +// new session is inserted. +// +// q is *dbq.Queries; pass either the pool-level Queries or one bound to a +// transaction so callers can compose this into a larger atomic operation. +func FindOrCreate( + ctx context.Context, + q *dbq.Queries, + userID pgtype.UUID, + eventTime time.Time, + clientID string, + timeout time.Duration, +) (pgtype.UUID, error) { + prev, err := q.GetMostRecentPlaySessionForUser(ctx, userID) + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + return pgtype.UUID{}, err + } + if err == nil { + // "Within window" includes the exact-boundary case: gap <= timeout extends. + gap := eventTime.Sub(prev.LastEventAt.Time) + if gap <= timeout { + if err := q.TouchPlaySessionLastEvent(ctx, dbq.TouchPlaySessionLastEventParams{ + ID: prev.ID, + LastEventAt: pgtype.Timestamptz{Time: eventTime, Valid: true}, + }); err != nil { + return pgtype.UUID{}, err + } + return prev.ID, nil + } + } + // Either no prior session, or the gap exceeded the timeout. + var clientIDPtr *string + if clientID != "" { + clientIDPtr = &clientID + } + row, err := q.InsertPlaySession(ctx, dbq.InsertPlaySessionParams{ + UserID: userID, + StartedAt: pgtype.Timestamptz{Time: eventTime, Valid: true}, + ClientID: clientIDPtr, + }) + if err != nil { + return pgtype.UUID{}, err + } + // Bump track_count to 1 on creation so the contract "every FindOrCreate is + // the start of one play" stays true (insert leaves track_count default 0). + if err := q.TouchPlaySessionLastEvent(ctx, dbq.TouchPlaySessionLastEventParams{ + ID: row.ID, + LastEventAt: pgtype.Timestamptz{Time: eventTime, Valid: true}, + }); err != nil { + return pgtype.UUID{}, err + } + return row.ID, nil +} +``` + +Note on `TouchPlaySessionLastEventParams`: sqlc generates this struct with named fields matching the SQL `:exec` parameters. The exact struct field names (`ID`, `LastEventAt`) come from the `id = $1, last_event_at = $2` mapping in the query. If sqlc generates them differently, adjust the field names but keep the function signature. + +- [ ] **Step 4: Run tests, confirm they pass** + +Run: same command as Step 2. +Expected: PASS — 4 tests. + +- [ ] **Step 5: Lint** + +Run: `golangci-lint run ./internal/playsessions/...` +Expected: clean. + +- [ ] **Step 6: Commit** + +```bash +git add internal/playsessions/ +git commit -m "feat(playsessions): add FindOrCreate session service (30-min rolling window) + +Per spec §6: returns existing session id when last_event_at is within +timeout; inserts a new session otherwise. Touch updates last_event_at +and increments track_count so callers can assume FindOrCreate == one +play started." +``` + +--- + +## Task 4: Play events writer + +**Files:** +- Create: `internal/playevents/writer.go` +- Create: `internal/playevents/writer_test.go` + +This package owns the four write paths and the skip classification rule. It calls `playsessions.FindOrCreate` from `RecordPlayStarted` and `RecordSyntheticCompletedPlay`. + +- [ ] **Step 1: Write the failing tests** + +Create `internal/playevents/writer_test.go`: + +```go +package playevents + +import ( + "context" + "io" + "log/slog" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db" + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" +) + +func testPool(t *testing.T) *pgxpool.Pool { + t.Helper() + if testing.Short() { + t.Skip("skipping integration test 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 RESTART IDENTITY CASCADE"); err != nil { + t.Fatalf("truncate: %v", err) + } + if _, err := pool.Exec(context.Background(), + "TRUNCATE tracks, albums, artists RESTART IDENTITY CASCADE"); err != nil { + t.Fatalf("truncate library: %v", err) + } + return pool +} + +type fixture struct { + pool *pgxpool.Pool + q *dbq.Queries + w *Writer + user pgtype.UUID + track pgtype.UUID + dur int32 // ms +} + +func newFixture(t *testing.T, durationMs int32) fixture { + t.Helper() + pool := testPool(t) + 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, err := q.UpsertArtist(context.Background(), dbq.UpsertArtistParams{Name: "Artist", SortName: "Artist"}) + if err != nil { + t.Fatalf("artist: %v", err) + } + al, err := q.UpsertAlbum(context.Background(), dbq.UpsertAlbumParams{Title: "Album", SortTitle: "Album", ArtistID: a.ID}) + if err != nil { + t.Fatalf("album: %v", err) + } + tr, err := q.UpsertTrack(context.Background(), dbq.UpsertTrackParams{ + Title: "Track", AlbumID: al.ID, ArtistID: a.ID, + FilePath: "/tmp/track-fixture.flac", + DurationMs: durationMs, + }) + if err != nil { + t.Fatalf("track: %v", err) + } + return fixture{ + pool: pool, + q: q, + w: NewWriter(pool, slog.New(slog.NewTextHandler(io.Discard, nil)), 30*time.Minute, 0.5, 30000), + user: u.ID, + track: tr.ID, + dur: durationMs, + } +} + +func TestRecordPlayStarted_InsertsRow(t *testing.T) { + f := newFixture(t, 200_000) + now := time.Now().UTC() + res, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", now) + if err != nil { + t.Fatalf("RecordPlayStarted: %v", err) + } + if !res.PlayEventID.Valid { + t.Fatalf("play_event_id invalid") + } + got, err := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) + if err != nil { + t.Fatalf("get: %v", err) + } + if got.EndedAt.Valid { + t.Errorf("ended_at should be NULL right after start") + } + if got.WasSkipped { + t.Errorf("was_skipped should default false") + } +} + +func TestRecordPlayStarted_AutoClosesPriorOpenRow(t *testing.T) { + f := newFixture(t, 200_000) + t1 := time.Now().UTC() + first, err := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) + if err != nil { + t.Fatalf("first: %v", err) + } + t2 := t1.Add(45 * time.Second) + _, err = f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t2) + if err != nil { + t.Fatalf("second: %v", err) + } + prior, err := f.q.GetPlayEventByID(context.Background(), first.PlayEventID) + if err != nil { + t.Fatalf("get: %v", err) + } + if !prior.EndedAt.Valid { + t.Errorf("prior should be closed") + } + if !prior.WasSkipped { + t.Errorf("prior should be marked was_skipped=true (auto-close convention)") + } + if prior.DurationPlayedMs == nil || *prior.DurationPlayedMs != 45_000 { + t.Errorf("duration_played_ms = %v, want 45000", prior.DurationPlayedMs) + } +} + +func TestRecordPlayStarted_AutoCloseCapsAtTrackDuration(t *testing.T) { + f := newFixture(t, 60_000) // 60s track + t1 := time.Now().UTC() + first, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) + // Start a second play 5 minutes later — elapsed > track duration. + t2 := t1.Add(5 * time.Minute) + _, _ = f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t2) + prior, _ := f.q.GetPlayEventByID(context.Background(), first.PlayEventID) + if prior.DurationPlayedMs == nil || *prior.DurationPlayedMs != 60_000 { + t.Errorf("duration_played_ms = %v, want capped at 60000", prior.DurationPlayedMs) + } +} + +func TestRecordPlayEnded_AppliesSkipRule_BothFail_NotSkip(t *testing.T) { + f := newFixture(t, 200_000) + t1 := time.Now().UTC() + res, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) + // 49% completion (97_999ms), 30s duration -> 30s satisfies "duration >= 30s" + // so the rule's AND fails -> NOT a skip. + t2 := t1.Add(30 * time.Second) + if err := f.w.RecordPlayEnded(context.Background(), res.PlayEventID, 30_000, t2); err != nil { + t.Fatalf("RecordPlayEnded: %v", err) + } + got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) + if got.WasSkipped { + t.Errorf("was_skipped = true, want false (duration_played_ms == 30000 fails the AND)") + } + // No skip_events row should be written. +} + +func TestRecordPlayEnded_AppliesSkipRule_BothPass_IsSkip(t *testing.T) { + f := newFixture(t, 200_000) + t1 := time.Now().UTC() + res, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) + // 5% completion (10_000ms), 10s duration -> both conditions hold -> SKIP. + t2 := t1.Add(10 * time.Second) + if err := f.w.RecordPlayEnded(context.Background(), res.PlayEventID, 10_000, t2); err != nil { + t.Fatalf("RecordPlayEnded: %v", err) + } + got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) + if !got.WasSkipped { + t.Errorf("expected was_skipped=true") + } + // Verify a matching skip_events row was inserted. + rows, err := f.pool.Query(context.Background(), + "SELECT id FROM skip_events WHERE user_id=$1 AND track_id=$2", f.user, f.track) + if err != nil { + t.Fatalf("query skips: %v", err) + } + defer rows.Close() + count := 0 + for rows.Next() { + count++ + } + if count != 1 { + t.Errorf("skip_events count = %d, want 1", count) + } +} + +func TestRecordPlayEnded_DurationOver50Percent_NotSkip(t *testing.T) { + f := newFixture(t, 200_000) // 200s track + t1 := time.Now().UTC() + res, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) + // 50% completion (100_000ms), 5s duration. Completion fails the "< 0.5" + // check — so AND fails, NOT a skip. + t2 := t1.Add(5 * time.Second) + _ = f.w.RecordPlayEnded(context.Background(), res.PlayEventID, 100_000, t2) + got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) + if got.WasSkipped { + t.Errorf("expected was_skipped=false at completion >= 0.5") + } +} + +func TestRecordPlaySkipped_AlwaysSkipFlagAndSkipEventRow(t *testing.T) { + f := newFixture(t, 200_000) + t1 := time.Now().UTC() + res, _ := f.w.RecordPlayStarted(context.Background(), f.user, f.track, "c", t1) + // User-initiated skip at 90% — overrides the rule, always treated as skip. + t2 := t1.Add(180 * time.Second) + if err := f.w.RecordPlaySkipped(context.Background(), res.PlayEventID, 180_000, t2); err != nil { + t.Fatalf("RecordPlaySkipped: %v", err) + } + got, _ := f.q.GetPlayEventByID(context.Background(), res.PlayEventID) + if !got.WasSkipped { + t.Errorf("RecordPlaySkipped must always set was_skipped=true") + } +} + +func TestRecordSyntheticCompletedPlay_WritesBothRows(t *testing.T) { + f := newFixture(t, 200_000) + now := time.Now().UTC() + if err := f.w.RecordSyntheticCompletedPlay(context.Background(), f.user, f.track, "feishin", now); err != nil { + t.Fatalf("synthetic: %v", err) + } + rows, _ := f.pool.Query(context.Background(), + "SELECT id, started_at, ended_at, duration_played_ms, was_skipped FROM play_events WHERE user_id=$1", f.user) + defer rows.Close() + count := 0 + for rows.Next() { + count++ + } + if count != 1 { + t.Errorf("play_events count = %d, want 1", count) + } +} +``` + +- [ ] **Step 2: Run tests, confirm they fail** + +Run: +```bash +MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' \ + go test ./internal/playevents -v +``` +Expected: FAIL — package not yet present. + +- [ ] **Step 3: Implement `internal/playevents/writer.go`** + +```go +// Package playevents owns the four write paths into play_events / skip_events: +// RecordPlayStarted, RecordPlayEnded, RecordPlaySkipped, and +// RecordSyntheticCompletedPlay (used by the Subsonic /rest/scrobble shim). +// +// All paths run inside a single transaction. The auto-close-prior step on +// RecordPlayStarted bounds each user to at most one open play_event at any +// time (per spec data-flow section). +package playevents + +import ( + "context" + "errors" + "log/slog" + "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/playsessions" +) + +type Writer struct { + pool *pgxpool.Pool + logger *slog.Logger + sessionTimeout time.Duration + skipMaxCompletionRatio float64 + skipMaxDurationPlayedMs int32 +} + +// NewWriter constructs a writer. timeout is the play-session window; the two +// remaining args are the skip rule thresholds from spec §6. +func NewWriter( + pool *pgxpool.Pool, + logger *slog.Logger, + sessionTimeout time.Duration, + skipMaxCompletionRatio float64, + skipMaxDurationPlayedMs int, +) *Writer { + return &Writer{ + pool: pool, + logger: logger, + sessionTimeout: sessionTimeout, + skipMaxCompletionRatio: skipMaxCompletionRatio, + skipMaxDurationPlayedMs: int32(skipMaxDurationPlayedMs), + } +} + +type StartedResult struct { + PlayEventID pgtype.UUID + SessionID pgtype.UUID +} + +// RecordPlayStarted: in one transaction, auto-close any prior open row for +// the user, FindOrCreate the play_session, insert the new play_event, +// return its id and session id. +func (w *Writer) RecordPlayStarted( + ctx context.Context, + userID, trackID pgtype.UUID, + clientID string, + at time.Time, +) (StartedResult, error) { + var out StartedResult + err := pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error { + q := dbq.New(tx) + // Auto-close any prior open row for this user. + if err := w.autoClosePriorOpen(ctx, q, userID, at); err != nil { + return err + } + // Resolve session. + sessionID, err := playsessions.FindOrCreate(ctx, q, userID, at, clientID, w.sessionTimeout) + if err != nil { + return err + } + var clientIDPtr *string + if clientID != "" { + clientIDPtr = &clientID + } + ev, err := q.InsertPlayEvent(ctx, dbq.InsertPlayEventParams{ + UserID: userID, + TrackID: trackID, + SessionID: sessionID, + StartedAt: pgtype.Timestamptz{Time: at, Valid: true}, + ClientID: clientIDPtr, + }) + if err != nil { + return err + } + out.PlayEventID = ev.ID + out.SessionID = sessionID + return nil + }) + return out, err +} + +// RecordPlayEnded closes a play_event with duration + computed completion +// ratio. Applies the spec §6 skip rule (AND of completion < threshold AND +// duration_played < threshold) — both conditions must hold for the row to +// be marked was_skipped=true. If skipped, also writes a skip_events row. +func (w *Writer) RecordPlayEnded( + ctx context.Context, + playEventID pgtype.UUID, + durationPlayedMs int32, + at time.Time, +) error { + return pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error { + q := dbq.New(tx) + ev, err := q.GetPlayEventByID(ctx, playEventID) + if err != nil { + return err + } + // Look up track duration to compute completion_ratio. + track, err := q.GetTrackByID(ctx, ev.TrackID) + if err != nil { + return err + } + ratio := 0.0 + if track.DurationMs > 0 { + ratio = float64(durationPlayedMs) / float64(track.DurationMs) + } + // Skip rule: both conditions must hold. + isSkip := ratio < w.skipMaxCompletionRatio && durationPlayedMs < w.skipMaxDurationPlayedMs + ratioPtr := ratio + durPtr := durationPlayedMs + if _, err := q.UpdatePlayEventEnded(ctx, dbq.UpdatePlayEventEndedParams{ + ID: playEventID, + EndedAt: pgtype.Timestamptz{Time: at, Valid: true}, + DurationPlayedMs: &durPtr, + CompletionRatio: &ratioPtr, + WasSkipped: isSkip, + }); err != nil { + return err + } + if isSkip { + if _, err := q.InsertSkipEvent(ctx, dbq.InsertSkipEventParams{ + UserID: ev.UserID, + TrackID: ev.TrackID, + SessionID: ev.SessionID, + SkippedAt: pgtype.Timestamptz{Time: at, Valid: true}, + PositionMs: durationPlayedMs, + }); err != nil { + return err + } + } + return nil + }) +} + +// RecordPlaySkipped is the user-initiated-skip variant: was_skipped is forced +// true regardless of the rule, and a skip_events row is always written. +func (w *Writer) RecordPlaySkipped( + ctx context.Context, + playEventID pgtype.UUID, + positionMs int32, + at time.Time, +) error { + return pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error { + q := dbq.New(tx) + ev, err := q.GetPlayEventByID(ctx, playEventID) + if err != nil { + return err + } + track, err := q.GetTrackByID(ctx, ev.TrackID) + if err != nil { + return err + } + ratio := 0.0 + if track.DurationMs > 0 { + ratio = float64(positionMs) / float64(track.DurationMs) + } + ratioPtr := ratio + durPtr := positionMs + if _, err := q.UpdatePlayEventEnded(ctx, dbq.UpdatePlayEventEndedParams{ + ID: playEventID, + EndedAt: pgtype.Timestamptz{Time: at, Valid: true}, + DurationPlayedMs: &durPtr, + CompletionRatio: &ratioPtr, + WasSkipped: true, + }); err != nil { + return err + } + _, err = q.InsertSkipEvent(ctx, dbq.InsertSkipEventParams{ + UserID: ev.UserID, + TrackID: ev.TrackID, + SessionID: ev.SessionID, + SkippedAt: pgtype.Timestamptz{Time: at, Valid: true}, + PositionMs: positionMs, + }) + return err + }) +} + +// RecordSyntheticCompletedPlay writes a start+end pair for a Subsonic +// scrobble?submission=true call. ended_at = at + track.duration_ms, +// duration_played_ms = track.duration_ms, was_skipped = false. +func (w *Writer) RecordSyntheticCompletedPlay( + ctx context.Context, + userID, trackID pgtype.UUID, + clientID string, + at time.Time, +) error { + return pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error { + q := dbq.New(tx) + track, err := q.GetTrackByID(ctx, trackID) + if err != nil { + return err + } + if err := w.autoClosePriorOpen(ctx, q, userID, at); err != nil { + return err + } + sessionID, err := playsessions.FindOrCreate(ctx, q, userID, at, clientID, w.sessionTimeout) + if err != nil { + return err + } + var clientIDPtr *string + if clientID != "" { + clientIDPtr = &clientID + } + ev, err := q.InsertPlayEvent(ctx, dbq.InsertPlayEventParams{ + UserID: userID, + TrackID: trackID, + SessionID: sessionID, + StartedAt: pgtype.Timestamptz{Time: at, Valid: true}, + ClientID: clientIDPtr, + }) + if err != nil { + return err + } + ratio := 1.0 + dur := track.DurationMs + _, err = q.UpdatePlayEventEnded(ctx, dbq.UpdatePlayEventEndedParams{ + ID: ev.ID, + EndedAt: pgtype.Timestamptz{Time: at.Add(time.Duration(track.DurationMs) * time.Millisecond), Valid: true}, + DurationPlayedMs: &dur, + CompletionRatio: &ratio, + WasSkipped: false, + }) + return err + }) +} + +// autoClosePriorOpen closes any open (ended_at IS NULL) play_event for the +// user. Sets ended_at = at, duration_played_ms = min(at - started_at, track +// duration), was_skipped = true. Skip rule is NOT applied — auto-closed +// rows are flagged skipped regardless because we don't know what the user +// actually heard. +func (w *Writer) autoClosePriorOpen( + ctx context.Context, + q *dbq.Queries, + userID pgtype.UUID, + at time.Time, +) error { + prior, err := q.GetOpenPlayEventForUser(ctx, userID) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil + } + return err + } + track, err := q.GetTrackByID(ctx, prior.TrackID) + if err != nil { + return err + } + elapsedMs := int32(at.Sub(prior.StartedAt.Time) / time.Millisecond) + if elapsedMs < 0 { + elapsedMs = 0 + } + if elapsedMs > track.DurationMs && track.DurationMs > 0 { + elapsedMs = track.DurationMs + } + ratio := 0.0 + if track.DurationMs > 0 { + ratio = float64(elapsedMs) / float64(track.DurationMs) + } + ratioPtr := ratio + durPtr := elapsedMs + if _, err := q.UpdatePlayEventEnded(ctx, dbq.UpdatePlayEventEndedParams{ + ID: prior.ID, + EndedAt: pgtype.Timestamptz{Time: at, Valid: true}, + DurationPlayedMs: &durPtr, + CompletionRatio: &ratioPtr, + WasSkipped: true, + }); err != nil { + return err + } + w.logger.Info("playevents: auto-closed prior open row", + "play_event_id", prior.ID, "elapsed_ms", elapsedMs) + return nil +} +``` + +- [ ] **Step 4: Run tests, confirm they pass** + +Run: same as Step 2. +Expected: PASS — 8 tests. + +- [ ] **Step 5: Lint** + +Run: `golangci-lint run ./internal/playevents/...` +Expected: clean. + +- [ ] **Step 6: Commit** + +```bash +git add internal/playevents/ +git commit -m "feat(playevents): add Writer with start/end/skipped/synthetic paths + +Owns the auto-close-prior step (caps each user at one open row), +spec §6 skip classification rule on play_ended (AND of completion < +0.5 and duration_played < 30s), and skip_events row writes. Synthetic +completed play wires the Subsonic /rest/scrobble?submission=true path +into the same store as native events." +``` + +--- + +## Task 5: HTTP `POST /api/events` handler + +**Files:** +- Create: `internal/api/events.go` +- Create: `internal/api/events_test.go` +- Modify: `internal/api/api.go` +- Modify: `internal/api/auth_test.go` (extend `testHandlers` to inject events writer) + +The handler is a thin wrapper around `playevents.Writer`. Auth comes from the existing `RequireUser` middleware; user_id is read from request context. + +- [ ] **Step 1: Extend `testHandlers` to construct a Writer** + +The existing `testHandlers` returns `(*handlers, *pgxpool.Pool)`. The events writer needs a `*playevents.Writer`, which requires the pool + config thresholds. We attach it to the `handlers` struct. + +In `internal/api/api.go`, modify the `handlers` struct and `Mount` to thread the writer through. Add to the file: + +```go +import ( + // ... existing imports ... + "git.fabledsword.com/bvandeusen/minstrel/internal/config" + "git.fabledsword.com/bvandeusen/minstrel/internal/playevents" +) + +// (existing Mount signature, just add cfg parameter at the end) +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} + // ... rest unchanged ... +} + +type handlers struct { + pool *pgxpool.Pool + logger *slog.Logger + events *playevents.Writer +} +``` + +The `Mount` callers need updating. Find every call site (`grep -rn "api.Mount" --include='*.go'`) and add `cfg.Events` as the new argument. There's likely one in `cmd/minstrel/main.go` and possibly one in `internal/server`. + +In `internal/api/auth_test.go`, modify `testHandlers` to construct a Writer: + +```go +import ( + // ... existing imports ... + "time" + "git.fabledsword.com/bvandeusen/minstrel/internal/playevents" +) + +func testHandlers(t *testing.T) (*handlers, *pgxpool.Pool) { + t.Helper() + if testing.Short() { + t.Skip("skipping api integration in -short mode") + } + dsn := os.Getenv("MINSTREL_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("MINSTREL_TEST_DATABASE_URL not set") + } + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + if err := db.Migrate(dsn, logger); 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 RESTART IDENTITY CASCADE"); err != nil { + t.Fatalf("truncate: %v", err) + } + w := playevents.NewWriter(pool, logger, 30*time.Minute, 0.5, 30000) + return &handlers{pool: pool, logger: logger, events: w}, pool +} +``` + +- [ ] **Step 2: Write failing tests** + +Create `internal/api/events_test.go`: + +```go +package api + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "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 dbq.UUIDLike) 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 +} +``` + +Note: `uuidToString` already exists in this package (used by other handlers); reuse it. `dbq.UUIDLike` in `postPlayStartedHelper` is a placeholder — in practice you'll declare the parameter as `pgtype.UUID` and import the package. The existing test fixtures (`seedTrack` etc.) return `dbq.Track` whose `ID` field is `pgtype.UUID`, so the helper takes `pgtype.UUID` directly: + +```go +func postPlayStartedHelper(t *testing.T, h *handlers, user dbq.User, trackID pgtype.UUID) string { +``` + +(adjust the import list accordingly). + +- [ ] **Step 3: Run tests, confirm they fail** + +Run: +```bash +MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' \ + go test ./internal/api -run TestHandleEvents -v +``` +Expected: FAIL — `handleEvents` undefined. + +- [ ] **Step 4: Implement `internal/api/events.go`** + +```go +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}) +} +``` + +- [ ] **Step 5: Register the route in `internal/api/api.go`** + +Find the authed group and add (right after `authed.Get("/radio", h.handleRadio)` from the prior PR): + +```go +authed.Post("/events", h.handleEvents) +``` + +- [ ] **Step 6: Run tests, confirm pass** + +Run: same command as Step 3. +Expected: PASS — 5 tests. + +- [ ] **Step 7: Lint + full Go suite** + +Run: `golangci-lint run ./...` +Expected: clean. + +Run: `go test -short -race ./...` +Expected: all packages PASS. + +Run with the integration env var: +```bash +MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' \ + go test -race ./... +``` +Expected: PASS including the playsessions, playevents, and api integration tests. + +- [ ] **Step 8: Commit** + +```bash +git add internal/api/events.go internal/api/events_test.go internal/api/api.go internal/api/auth_test.go cmd/minstrel/main.go internal/server/ +git commit -m "feat(api): add POST /api/events handler + +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." +``` + +(If the cmd/minstrel/main.go or internal/server changes touch other unrelated files, stage them separately.) + +--- + +## Task 6: Subsonic scrobble integration + +**Files:** +- Modify: `internal/subsonic/stream.go` +- Modify: `internal/subsonic/subsonic.go` (constructor wiring) +- Create: `internal/subsonic/scrobble_test.go` + +Replace the in-memory `nowPlayingMap` with calls into `playevents.Writer`. The constructor for `mediaHandlers` grows a writer arg. + +- [ ] **Step 1: Write failing tests** + +Create `internal/subsonic/scrobble_test.go`: + +```go +package subsonic + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgtype" + "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) + } +} + +// uuidToID is the existing helper in internal/subsonic; same-package +// test reuses it. pgtype.UUID kept as an explicit import marker. +var _ = pgtype.UUID{} +``` + +Note: the existing `internal/subsonic/stream.go` already has a `uuidToID` helper used by handlers. The test refers to it directly. If the test file conflicts with a private function name, rename the local one or import-alias. + +- [ ] **Step 2: Modify `internal/subsonic/stream.go`** + +Replace the `mediaHandlers` struct and its constructor: + +```go +// mediaHandlers serves the bytes-on-the-wire endpoints: stream, download, +// getCoverArt, scrobble. +type mediaHandlers struct { + pool *pgxpool.Pool + events *playevents.Writer +} + +func newMediaHandlers(pool *pgxpool.Pool, events *playevents.Writer) *mediaHandlers { + return &mediaHandlers{pool: pool, events: events} +} +``` + +Add the import: `"git.fabledsword.com/bvandeusen/minstrel/internal/playevents"`. + +Replace `handleScrobble`: + +```go +// 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). +func (m *mediaHandlers) handleScrobble(w http.ResponseWriter, r *http.Request) { + params := r.URL.Query() + idStr := params.Get("id") + if idStr == "" { + WriteFail(w, r, ErrMissingParameter, "Missing required parameter: id") + return + } + trackID, ok := parseUUID(idStr) + if !ok { + 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 sends `time` as ms since epoch (per spec). Some clients send + // it as 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 + } + // Otherwise fall through to now() — better than rejecting the call. + } + clientID := params.Get("c") + + submission := strings.ToLower(params.Get("submission")) + switch submission { + case "true", "": + // Subsonic default is submission=true (completed play). + 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")}) +} +``` + +Add the imports `"strconv"` and `"time"` if not already present. Delete the entire `nowPlayingMap` struct + `nowPlayingEntry` + `newNowPlayingMap` + the `nowPlaying` field reference. The file ends much shorter after this change. + +- [ ] **Step 3: Update the constructor in `internal/subsonic/subsonic.go`** + +Find where `newMediaHandlers(pool)` is called (likely in `Mount` or the equivalent setup func). Update the signature and pass the writer through: + +```go +// Before: +// m := newMediaHandlers(pool) +// After (Mount signature gains a *playevents.Writer parameter): +// func Mount(r chi.Router, pool *pgxpool.Pool, events *playevents.Writer) { +// m := newMediaHandlers(pool, events) +// ... +// } +``` + +The caller chain: cmd/minstrel/main.go (or wherever `subsonic.Mount` is called) needs to be updated to pass the writer. Since the api Mount and subsonic Mount are likely both called from the same setup, share the writer between them. + +- [ ] **Step 4: Run tests, confirm they pass** + +Run: +```bash +MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' \ + go test ./internal/subsonic -run TestHandleScrobble -v +``` +Expected: PASS — 2 tests. + +- [ ] **Step 5: Full Go suite** + +Run: +```bash +MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' \ + go test -race ./... +``` +Expected: PASS across all packages. + +- [ ] **Step 6: Lint** + +Run: `golangci-lint run ./...` +Expected: clean. + +- [ ] **Step 7: Commit** + +```bash +git add internal/subsonic/ cmd/minstrel/main.go internal/server/ +git commit -m "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." +``` + +--- + +## Task 7: Web client — clientId + events module + +**Files:** +- Create: `web/src/lib/player/clientId.ts` +- Create: `web/src/lib/player/clientId.test.ts` +- Create: `web/src/lib/player/events.svelte.ts` +- Create: `web/src/lib/player/events.svelte.test.ts` +- Modify: `web/src/lib/api/types.ts` + +- [ ] **Step 1: Add types** + +Open `web/src/lib/api/types.ts`. After the existing `RadioResponse`, append: + +```ts +export type EventRequest = + | { type: 'play_started'; track_id: string; client_id?: string } + | { type: 'play_ended'; play_event_id: string; duration_played_ms: number } + | { type: 'play_skipped'; play_event_id: string; position_ms: number }; + +export type PlayStartedResponse = { play_event_id: string; session_id: string }; +export type EventOkResponse = { ok: true }; +``` + +- [ ] **Step 2: Write the clientId test** + +Create `web/src/lib/player/clientId.test.ts`: + +```ts +import { afterEach, beforeEach, describe, expect, test } from 'vitest'; +import { getOrCreateClientId } from './clientId'; + +beforeEach(() => sessionStorage.clear()); +afterEach(() => sessionStorage.clear()); + +describe('getOrCreateClientId', () => { + test('generates a UUID on first call and persists it in sessionStorage', () => { + const id = getOrCreateClientId(); + expect(id).toMatch(/^[0-9a-f-]{36}$/); + expect(sessionStorage.getItem('minstrel.client_id')).toBe(id); + }); + + test('returns the same id on subsequent calls', () => { + const a = getOrCreateClientId(); + const b = getOrCreateClientId(); + expect(a).toBe(b); + }); +}); +``` + +- [ ] **Step 3: Run, confirm fail** + +Run: `cd web && npm test -- src/lib/player/clientId.test.ts` +Expected: FAIL — module not found. + +- [ ] **Step 4: Implement `web/src/lib/player/clientId.ts`** + +```ts +const KEY = 'minstrel.client_id'; + +export function getOrCreateClientId(): string { + try { + const existing = sessionStorage.getItem(KEY); + if (existing) return existing; + } catch { + // sessionStorage unavailable (non-browser env). Generate ephemerally. + } + const id = crypto.randomUUID(); + try { sessionStorage.setItem(KEY, id); } catch { /* ignore */ } + return id; +} +``` + +- [ ] **Step 5: Run, confirm pass** + +Run: `cd web && npm test -- src/lib/player/clientId.test.ts` +Expected: PASS — 2 tests. + +- [ ] **Step 6: Write the events.svelte tests** + +Create `web/src/lib/player/events.svelte.test.ts`: + +```ts +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; +import { flushSync } from 'svelte'; + +vi.mock('$lib/api/client', () => ({ + api: { + post: vi.fn().mockResolvedValue({ play_event_id: 'pe-1', session_id: 's-1' }) + } +})); + +vi.mock('./clientId', () => ({ + getOrCreateClientId: () => 'test-client' +})); + +import { useEventsDispatcher } from './events.svelte'; +import { + player, + playQueue, + reportStateFromAudio, + skipNext, +} from './store.svelte'; +import { api } from '$lib/api/client'; +import type { TrackRef } from '$lib/api/types'; + +function track(id: string, dur = 200): TrackRef { + return { + id, title: `T ${id}`, + album_id: 'a', album_title: 'A', + artist_id: 'ar', artist_name: 'Ar', + track_number: Number(id), disc_number: 1, + duration_sec: dur, + stream_url: `/api/tracks/${id}/stream` + }; +} + +beforeEach(() => { + vi.clearAllMocks(); + // Reset the store to idle. + playQueue([]); +}); + +afterEach(() => { + // sendBeacon stub cleanup + // (vi.spyOn restored automatically on test boundary) +}); + +describe('useEventsDispatcher', () => { + test('first transition into playing for a new track POSTs play_started', async () => { + let cleanup!: () => void; + cleanup = $effect.root(() => useEventsDispatcher()); + + playQueue([track('1')]); + flushSync(); + reportStateFromAudio('playing'); + flushSync(); + // Microtasks for the mocked api.post to settle. + await Promise.resolve(); + await Promise.resolve(); + + expect(api.post).toHaveBeenCalledTimes(1); + expect(api.post).toHaveBeenCalledWith('/api/events', { + type: 'play_started', + track_id: '1', + client_id: 'test-client' + }); + + cleanup(); + }); + + test('track end (back to idle/paused) POSTs play_ended', async () => { + let cleanup!: () => void; + cleanup = $effect.root(() => useEventsDispatcher()); + + playQueue([track('1', 200)]); + flushSync(); + reportStateFromAudio('playing'); + flushSync(); + await Promise.resolve(); + await Promise.resolve(); + // Now end the track. + reportStateFromAudio('ended'); + flushSync(); + await Promise.resolve(); + + const calls = (api.post as ReturnType).mock.calls; + const types = calls.map((c) => (c[1] as { type: string }).type); + expect(types).toContain('play_started'); + expect(types).toContain('play_ended'); + + cleanup(); + }); + + test('user-initiated skip mid-track POSTs play_skipped', async () => { + let cleanup!: () => void; + cleanup = $effect.root(() => useEventsDispatcher()); + + playQueue([track('1'), track('2')]); + flushSync(); + reportStateFromAudio('playing'); + flushSync(); + await Promise.resolve(); + await Promise.resolve(); + skipNext(); + flushSync(); + await Promise.resolve(); + + const calls = (api.post as ReturnType).mock.calls; + const types = calls.map((c) => (c[1] as { type: string }).type); + expect(types).toContain('play_skipped'); + + cleanup(); + }); + + test('pagehide fires sendBeacon when a play is open', async () => { + let cleanup!: () => void; + cleanup = $effect.root(() => useEventsDispatcher()); + const beacon = vi.spyOn(navigator, 'sendBeacon').mockReturnValue(true); + + playQueue([track('1')]); + flushSync(); + reportStateFromAudio('playing'); + flushSync(); + await Promise.resolve(); + await Promise.resolve(); + + window.dispatchEvent(new Event('pagehide')); + + expect(beacon).toHaveBeenCalled(); + const args = beacon.mock.calls[0]; + expect(args[0]).toBe('/api/events'); + + cleanup(); + beacon.mockRestore(); + }); +}); +``` + +- [ ] **Step 7: Run, confirm fail** + +Run: `cd web && npm test -- src/lib/player/events.svelte.test.ts` +Expected: FAIL — module not found. + +- [ ] **Step 8: Implement `web/src/lib/player/events.svelte.ts`** + +```ts +import { player } from './store.svelte'; +import { api } from '$lib/api/client'; +import { getOrCreateClientId } from './clientId'; +import type { PlayStartedResponse } from '$lib/api/types'; + +// useEventsDispatcher installs $effect-based watchers that emit play events +// to /api/events. Call once from +layout.svelte's mount. +// +// The dispatcher is intentionally a simple state machine over (current track, +// state). It does NOT introspect player internals beyond the rune accessors; +// it only reacts to externally-observable transitions. +export function useEventsDispatcher(): void { + let openPlayEventId: string | null = null; + let openTrackId: string | null = null; + let lastPositionMs = 0; + const clientId = getOrCreateClientId(); + + // Track the "active" track id we've already started a play for. Each + // transition into 'playing' for a new track triggers play_started; each + // transition out of 'playing' or change-of-track closes the prior row. + $effect(() => { + const track = player.current; + const isPlaying = player.state === 'playing'; + + if (track && isPlaying && openTrackId !== track.id) { + // New track entered playing state — close any prior open one (skip + // semantics for user-driven track changes), then open this one. + if (openPlayEventId && openTrackId) { + const prevId = openPlayEventId; + const prevPos = lastPositionMs; + openPlayEventId = null; + openTrackId = null; + api.post<{ ok: true }>('/api/events', { + type: 'play_skipped', + play_event_id: prevId, + position_ms: prevPos + }).catch(() => { /* drop — server-side is best-effort */ }); + } + void startNew(track.id); + } else if (!track && openPlayEventId) { + // Queue cleared (idle) — close out as a skip. + const prevId = openPlayEventId; + const prevPos = lastPositionMs; + openPlayEventId = null; + openTrackId = null; + api.post('/api/events', { + type: 'play_skipped', + play_event_id: prevId, + position_ms: prevPos + }).catch(() => {}); + } + }); + + // Track played duration via player.position. We sample on every update; + // the cost is one int copy. + $effect(() => { + lastPositionMs = Math.round(player.position * 1000); + }); + + // 'ended' transition (state moves out of playing AFTER a finished track) + // triggers a normal play_ended close, distinguished from skip-by-track-change. + $effect(() => { + const isFinished = player.state === 'paused' && player.position >= player.duration && player.duration > 0; + if (isFinished && openPlayEventId) { + const prevId = openPlayEventId; + const dur = lastPositionMs; + openPlayEventId = null; + openTrackId = null; + api.post('/api/events', { + type: 'play_ended', + play_event_id: prevId, + duration_played_ms: dur + }).catch(() => {}); + } + }); + + // Page close: sendBeacon a play_skipped if an open play exists. + if (typeof window !== 'undefined') { + const onPageHide = () => { + if (!openPlayEventId) return; + const payload = JSON.stringify({ + type: 'play_skipped', + play_event_id: openPlayEventId, + position_ms: lastPositionMs + }); + navigator.sendBeacon('/api/events', new Blob([payload], { type: 'application/json' })); + }; + window.addEventListener('pagehide', onPageHide); + // No teardown — useEventsDispatcher is mounted for the app lifetime. + } + + async function startNew(trackId: string): Promise { + try { + const res = await api.post('/api/events', { + type: 'play_started', + track_id: trackId, + client_id: clientId + }); + // The user may have already moved on (clicked another track) by the + // time the response arrives. Only adopt the id if we're still on the + // same track and still playing. + if (player.current?.id === trackId && player.state === 'playing') { + openPlayEventId = res.play_event_id; + openTrackId = trackId; + } + } catch { + // Best-effort; missed events are OK in v1. + } + } +} +``` + +- [ ] **Step 9: Run, confirm pass** + +Run: `cd web && npm test -- src/lib/player/events.svelte.test.ts` +Expected: PASS — 4 tests. + +- [ ] **Step 10: Type-check** + +Run: `cd web && npm run check` +Expected: `0 ERRORS 0 WARNINGS`. + +- [ ] **Step 11: Commit** + +```bash +git add web/src/lib/player/clientId.ts web/src/lib/player/clientId.test.ts \ + web/src/lib/player/events.svelte.ts web/src/lib/player/events.svelte.test.ts \ + web/src/lib/api/types.ts +git commit -m "feat(web): events dispatcher posts to /api/events on player transitions + +useEventsDispatcher subscribes to player.current/state via \$effect: +on first transition to playing for a new track, POSTs play_started +and caches the id; on track change while playing, closes the prior +row as play_skipped; on natural completion (paused at duration), POSTs +play_ended; on pagehide, navigator.sendBeacon fires a final +play_skipped. Stable per-tab client_id from sessionStorage." +``` + +--- + +## Task 8: Wire `useEventsDispatcher()` in `+layout.svelte` + +**Files:** +- Modify: `web/src/routes/+layout.svelte` + +- [ ] **Step 1: Add the import** + +Open `web/src/routes/+layout.svelte`. Find the existing `useMediaSession` import and add a sibling: + +```ts + import { useMediaSession } from '$lib/player/mediaSession.svelte'; + import { useEventsDispatcher } from '$lib/player/events.svelte'; +``` + +- [ ] **Step 2: Call it once at mount** + +Find the `useMediaSession();` call near the bottom of the script block. Add right after: + +```ts + useMediaSession(); + useEventsDispatcher(); +``` + +- [ ] **Step 3: Type-check** + +Run: `cd web && npm run check` +Expected: `0 ERRORS 0 WARNINGS`. + +- [ ] **Step 4: Run full web tests** + +Run: `cd web && npm test` +Expected: all suites PASS. + +- [ ] **Step 5: Commit** + +```bash +git add web/src/routes/+layout.svelte +git commit -m "feat(web): mount useEventsDispatcher() in root layout + +Single mount point alongside useMediaSession. Once called, the +dispatcher's \$effect watchers run for the lifetime of the SPA." +``` + +--- + +## Task 9: Final verification + branch finish + +**Files:** none (verification only). + +- [ ] **Step 1: Full Go suite (with DB)** + +Run: +```bash +MINSTREL_TEST_DATABASE_URL='postgres://minstrel:minstrel@172.22.0.2:5432/minstrel?sslmode=disable' \ + go test -race ./... +``` +Expected: all packages PASS (including the integration tests). + +- [ ] **Step 2: Go lint** + +Run: `golangci-lint run ./...` +Expected: clean. + +- [ ] **Step 3: Web check + tests + build** + +Run: `cd web && npm run check && npm test && npm run build` +Expected: check 0/0; all vitest files pass; build emits `web/build/`. + +Run: `git checkout -- web/build/index.html` +Expected: committed placeholder restored. + +- [ ] **Step 4: Docker build smoke** + +Run: `docker build -t minstrel:m2-events-smoke .` +Expected: image builds. + +Run: `docker run --rm --entrypoint /bin/sh minstrel:m2-events-smoke -c 'test -x /usr/local/bin/minstrel && echo ok'` +Expected: `ok`. + +- [ ] **Step 5: Local end-to-end manual check** + +```bash +docker compose up --build -d +``` + +Browser at `localhost:4533`: + +1. Sign in. Play a track. +2. Confirm in postgres: + ```bash + docker exec minstrel-postgres-1 psql -U minstrel -d minstrel \ + -c "SELECT user_id, track_id, started_at, ended_at FROM play_events;" + ``` + Expected: one row, `ended_at` NULL. +3. Let track finish. Re-run the SELECT. Expected: same row now has `ended_at` set, `was_skipped=false`. +4. Skip a track within 30 seconds of starting. Expected: previous row marked `was_skipped=true`; new row exists for the next track; `skip_events` has the matching row. +5. Wait > 30 minutes (or use `psql` to manually `UPDATE play_sessions SET last_event_at = now() - INTERVAL '40 minutes' WHERE user_id = ...` to fast-forward). Play. Expected: new `play_sessions` row with a different id. +6. Open Feishin/Symfonium pointed at the same instance. Play a track to completion. Expected: a synthetic `play_events` row appears with both `started_at` and `ended_at` set. +7. Close the browser tab mid-track. Re-run the SELECT. Expected: the last row's `ended_at` is set (sendBeacon hit the server before tab actually closed). +8. Disconnect network mid-play, force-quit the browser. Reconnect, sign back in, start a new track. Expected: the previous abandoned row gets auto-closed by the events handler before the new row is inserted. + +Tear down: +```bash +docker compose down +``` + +- [ ] **Step 6: Finish the branch** + +Follow `superpowers:finishing-a-development-branch`. Expected path: Option 2 — push `dev`, open a PR to `main`, wait for Forgejo CI, merge once green. + +--- + +## Self-Review Notes + +**Spec coverage:** +- Schema (play_sessions, play_events, skip_events) → Task 1. +- Config thresholds → Task 2. +- Session service `FindOrCreate` with 30-min rolling rule + boundary case → Task 3. +- playevents writer + auto-close-prior + skip rule + synthetic completed play → Task 4. +- `POST /api/events` handler with discriminated union + ownership checks → Task 5. +- Subsonic `/rest/scrobble` integration + `nowPlayingMap` removal → Task 6. +- Web events module + clientId + sendBeacon → Tasks 7, 8. +- End-to-end manual checklist matches spec's verification list → Task 9. + +**Type consistency:** +- `pgtype.UUID` everywhere on the Go side; never `string` for UUIDs except at HTTP boundaries. +- `Writer` (capital W) used consistently as the playevents type. +- `RecordPlayStarted` / `RecordPlayEnded` / `RecordPlaySkipped` / `RecordSyntheticCompletedPlay` — same names in the spec, plan, and writer module. +- `EventRequest` / `PlayStartedResponse` / `EventOkResponse` — same TypeScript names in types.ts and consumers. +- `useEventsDispatcher()` — same hook name in module + +layout.svelte call site. +- `getOrCreateClientId` — same name in module + events module + tests. + +**Filename hazards:** events.svelte.ts uses runes ($effect); the test file is events.svelte.test.ts because the test body uses `$effect.root(...)` directly (matching the useDelayed.svelte.test.ts and mediaSession.svelte.test.ts precedents). + +**Placeholder scan:** no TBD/TODO/later markers. Every code block is a complete implementation; every command shows expected output. + +**Risks / mitigations from spec:** +- Event spam → bounded by auto-close-prior (Task 4 implementation). +- Clock skew → web client never sends `at` (Task 7 implementation just doesn't include the field); Subsonic clients honor `time` parameter (Task 6 handler parses both ms-since-epoch and RFC 3339). +- Synthetic plays too coarse → spec acknowledges; documented in writer comment (Task 4). +- `nowPlayingMap` deletion → no external callers; safe (verified during spec). +- Nullable `session_vector_at_play` → ships with the migration; M2 writes NULL (writer never sets it); M3 will populate.