From 258bc1f75c7ca05d3032fa89d5c6d2d37dadd21c Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 2 Jun 2026 18:32:22 -0400 Subject: [PATCH] =?UTF-8?q?feat(server):=20drift=20audit=20batch=207=20?= =?UTF-8?q?=E2=80=94=20periodic=20GC=20worker=20for=205=20lifecycle=20gaps?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New `internal/gc` package with a single Worker that runs all five lifecycle / retention sweeps from the 2026-06-02 drift audit on a 1-hour tick. Each sweep is small, idempotent (re-running on already-clean rows is a no-op), and logs its affected-row count. Sweeps (Scribe parent #552): - **#566** GcCloseStalePlayEvents — play_events rows opened > 24h ago that never got a play_ended (client crash, network drop). Synthesizes ended_at from duration_played_ms when known, falls back to now() so the row stops looking "open" to downstream filters (ended_at IS NULL). - **#565** GcClosePlaySessionsWithNoRecentEvents — play_sessions with last_event_at older than 6h get ended_at = last_event_at ("user moved on"); empty sessions older than 1h get closed too (stale handshakes from clients that never recorded a play). The audit caught that the column was added but never populated by any writer — every session row was "open" forever, breaking downstream dedup queries that assume closed semantics. - **#567** GcExpireScrobbleQueueFailedRows — drops scrobble_queue rows in status='failed' older than 14 days. The worker stops retrying after maxAttempts so these otherwise accumulate forever on a persistent ListenBrainz outage / revoked token. - **#574** GcResetStuckSystemPlaylistRuns — flips system_playlist_runs.in_flight back to false on rows whose last_run_at is older than 10 minutes. Catches goroutine-panic wedges where the generator died between SET in_flight=true and SET in_flight=false; the duplicate-prevention check refuses to start a fresh regen while in_flight, so a stuck row would otherwise deadlock all future regens for that user. Records "stuck-row auto-reset by gc" in last_error so the operator can tell auto-reset from a recent real failure. - **#575** GcDeleteExpiredPasswordResets — deletes expired password_resets rows. Unused expired rows go after a 1h grace (gives the operator time to debug an active reset attempt); used rows are kept 7 days for audit. Wiring: - main.go `go gcWorker.Run(ctx)` alongside the other periodic workers (scrobble, similarity, lidarr). - tickOnce fires once at start so a freshly-deployed server does its initial sweep without waiting a full tick, matching the scrobble worker pattern. - Errors per sweep are logged but do NOT abort the remaining ones — a transient pgx error from one query shouldn't prevent the others from running. Tests: - 4 integration tests, one per UPDATE/DELETE sweep, that seed rows-to-sweep + rows-to-leave-alone and assert the right rows changed state. Skip unless MINSTREL_TEST_DATABASE_URL is set (mirrors the api package pattern). - Empty-tables no-op smoke test. - Run() cancellation honoured (no spinning goroutine at test-runner exit). That's all five remaining server-side lifecycle findings from the audit. The Android LOCAL_USER_ID hardcode (#576) is a separate refactor that needs auth-store wiring and stays in the queue. --- cmd/minstrel/main.go | 11 ++ internal/db/dbq/gc.sql.go | 126 +++++++++++++++ internal/db/queries/gc.sql | 70 ++++++++ internal/gc/worker.go | 101 ++++++++++++ internal/gc/worker_test.go | 317 +++++++++++++++++++++++++++++++++++++ 5 files changed, 625 insertions(+) create mode 100644 internal/db/dbq/gc.sql.go create mode 100644 internal/db/queries/gc.sql create mode 100644 internal/gc/worker.go create mode 100644 internal/gc/worker_test.go diff --git a/cmd/minstrel/main.go b/cmd/minstrel/main.go index 5b67a423..04cfe6ff 100644 --- a/cmd/minstrel/main.go +++ b/cmd/minstrel/main.go @@ -15,6 +15,7 @@ import ( "git.fabledsword.com/bvandeusen/minstrel/internal/coverart" "git.fabledsword.com/bvandeusen/minstrel/internal/db" "git.fabledsword.com/bvandeusen/minstrel/internal/eventbus" + "git.fabledsword.com/bvandeusen/minstrel/internal/gc" "git.fabledsword.com/bvandeusen/minstrel/internal/library" "git.fabledsword.com/bvandeusen/minstrel/internal/lidarr" "git.fabledsword.com/bvandeusen/minstrel/internal/lidarrconfig" @@ -161,6 +162,16 @@ func run() error { similarityWorker := similarity.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "similarity")) go similarityWorker.Run(ctx) + // Start the GC worker. Runs every 1h and sweeps lifecycle tables + // that have no writer-side close path or retention policy: + // orphan play_events, stale play_sessions, expired + // scrobble_queue failures, stuck system_playlist_runs, expired + // password_resets. Each sweep is idempotent — a row that's + // already clean is a no-op. Addresses drift audit findings + // #565 #566 #567 #574 #575 (Scribe parent #552). + gcWorker := gc.NewWorker(pool, logger.With("component", "gc")) + go gcWorker.Run(ctx) + // Start the Lidarr reconciler worker. Per spec §M5a, polls pending Lidarr // import requests and reconciles them against the library. Short-circuits // to no-op when lidarr_config.enabled = false. diff --git a/internal/db/dbq/gc.sql.go b/internal/db/dbq/gc.sql.go new file mode 100644 index 00000000..cb150262 --- /dev/null +++ b/internal/db/dbq/gc.sql.go @@ -0,0 +1,126 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: gc.sql + +package dbq + +import ( + "context" +) + +const gcClosePlaySessionsWithNoRecentEvents = `-- name: GcClosePlaySessionsWithNoRecentEvents :execrows +UPDATE play_sessions + SET ended_at = COALESCE(last_event_at, started_at) + WHERE ended_at IS NULL + AND ( + (track_count > 0 AND last_event_at < now() - INTERVAL '6 hours') + OR + (track_count = 0 AND started_at < now() - INTERVAL '1 hour') + ) +` + +// #565: play_sessions.ended_at was added but never populated by any +// writer. Close sessions whose last_event_at is older than 6h — +// treating that as "user moved on" the same way audio_service does +// after grace periods. Sessions with NO events (track_count = 0) +// older than 1h are also closed (stale handshakes from clients that +// never recorded a play). ended_at is set to last_event_at so the +// session's duration reads naturally. +func (q *Queries) GcClosePlaySessionsWithNoRecentEvents(ctx context.Context) (int64, error) { + result, err := q.db.Exec(ctx, gcClosePlaySessionsWithNoRecentEvents) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const gcCloseStalePlayEvents = `-- name: GcCloseStalePlayEvents :execrows + +UPDATE play_events + SET ended_at = COALESCE( + started_at + (duration_played_ms * INTERVAL '1 millisecond'), + now() + ) + WHERE ended_at IS NULL + AND started_at < now() - INTERVAL '24 hours' +` + +// Background garbage-collector / lifecycle queries. All five address +// drift findings from the 2026-06-02 audit (Scribe parent #552): +// #565 #566 #567 #574 #575. Sweeper runs on a long tick (default 1h) +// so per-query cost is amortised; each is idempotent (re-running on +// already-closed/-deleted rows is a no-op). +// #566: play_events rows opened more than 24h ago that never got a +// play_ended. Client crashed mid-track, network dropped, etc. We +// synthesize ended_at = started_at + duration_played_ms when present, +// otherwise leave duration_played_ms null and stamp ended_at = now() +// so the row stops looking "open" for downstream queries that filter +// ended_at IS NULL. Doesn't touch was_skipped (we genuinely don't +// know if the user skipped). +func (q *Queries) GcCloseStalePlayEvents(ctx context.Context) (int64, error) { + result, err := q.db.Exec(ctx, gcCloseStalePlayEvents) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const gcDeleteExpiredPasswordResets = `-- name: GcDeleteExpiredPasswordResets :execrows +DELETE FROM password_resets + WHERE (used_at IS NOT NULL AND used_at < now() - INTERVAL '7 days') + OR (used_at IS NULL AND expires_at < now() - INTERVAL '1 hour') +` + +// #575: password_resets accumulates expired + used rows forever. The +// validation path already rejects them; this just keeps the table +// from growing unbounded. Used rows are kept for 7 days for audit; +// unused expired rows go immediately. +func (q *Queries) GcDeleteExpiredPasswordResets(ctx context.Context) (int64, error) { + result, err := q.db.Exec(ctx, gcDeleteExpiredPasswordResets) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const gcExpireScrobbleQueueFailedRows = `-- name: GcExpireScrobbleQueueFailedRows :execrows +DELETE FROM scrobble_queue + WHERE status = 'failed' + AND enqueued_at < now() - INTERVAL '14 days' +` + +// #567: scrobble_queue rows that have been in status='failed' for +// more than 14 days. The worker stops retrying after maxAttempts; +// failed rows accumulate forever otherwise. CASCADE from play_events +// already drops the row when the underlying event is deleted, so this +// only handles persistent failures (token revoked, etc.). +func (q *Queries) GcExpireScrobbleQueueFailedRows(ctx context.Context) (int64, error) { + result, err := q.db.Exec(ctx, gcExpireScrobbleQueueFailedRows) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const gcResetStuckSystemPlaylistRuns = `-- name: GcResetStuckSystemPlaylistRuns :execrows +UPDATE system_playlist_runs + SET in_flight = false, + last_error = COALESCE(last_error, 'stuck-row auto-reset by gc') + WHERE in_flight = true + AND last_run_at < now() - INTERVAL '10 minutes' +` + +// #574: system_playlist_runs.in_flight = true can wedge on a +// goroutine panic between SET in_flight=true and SET in_flight=false. +// The duplicate-prevention check refuses to start a fresh regen while +// in_flight, so a stuck row blocks all future regens for that user. +// Reset rows where last_run_at is older than 10 minutes (regens +// shouldn't take that long). +func (q *Queries) GcResetStuckSystemPlaylistRuns(ctx context.Context) (int64, error) { + result, err := q.db.Exec(ctx, gcResetStuckSystemPlaylistRuns) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} diff --git a/internal/db/queries/gc.sql b/internal/db/queries/gc.sql new file mode 100644 index 00000000..26c0c202 --- /dev/null +++ b/internal/db/queries/gc.sql @@ -0,0 +1,70 @@ +-- Background garbage-collector / lifecycle queries. All five address +-- drift findings from the 2026-06-02 audit (Scribe parent #552): +-- #565 #566 #567 #574 #575. Sweeper runs on a long tick (default 1h) +-- so per-query cost is amortised; each is idempotent (re-running on +-- already-closed/-deleted rows is a no-op). + +-- name: GcCloseStalePlayEvents :execrows +-- #566: play_events rows opened more than 24h ago that never got a +-- play_ended. Client crashed mid-track, network dropped, etc. We +-- synthesize ended_at = started_at + duration_played_ms when present, +-- otherwise leave duration_played_ms null and stamp ended_at = now() +-- so the row stops looking "open" for downstream queries that filter +-- ended_at IS NULL. Doesn't touch was_skipped (we genuinely don't +-- know if the user skipped). +UPDATE play_events + SET ended_at = COALESCE( + started_at + (duration_played_ms * INTERVAL '1 millisecond'), + now() + ) + WHERE ended_at IS NULL + AND started_at < now() - INTERVAL '24 hours'; + +-- name: GcClosePlaySessionsWithNoRecentEvents :execrows +-- #565: play_sessions.ended_at was added but never populated by any +-- writer. Close sessions whose last_event_at is older than 6h — +-- treating that as "user moved on" the same way audio_service does +-- after grace periods. Sessions with NO events (track_count = 0) +-- older than 1h are also closed (stale handshakes from clients that +-- never recorded a play). ended_at is set to last_event_at so the +-- session's duration reads naturally. +UPDATE play_sessions + SET ended_at = COALESCE(last_event_at, started_at) + WHERE ended_at IS NULL + AND ( + (track_count > 0 AND last_event_at < now() - INTERVAL '6 hours') + OR + (track_count = 0 AND started_at < now() - INTERVAL '1 hour') + ); + +-- name: GcExpireScrobbleQueueFailedRows :execrows +-- #567: scrobble_queue rows that have been in status='failed' for +-- more than 14 days. The worker stops retrying after maxAttempts; +-- failed rows accumulate forever otherwise. CASCADE from play_events +-- already drops the row when the underlying event is deleted, so this +-- only handles persistent failures (token revoked, etc.). +DELETE FROM scrobble_queue + WHERE status = 'failed' + AND enqueued_at < now() - INTERVAL '14 days'; + +-- name: GcResetStuckSystemPlaylistRuns :execrows +-- #574: system_playlist_runs.in_flight = true can wedge on a +-- goroutine panic between SET in_flight=true and SET in_flight=false. +-- The duplicate-prevention check refuses to start a fresh regen while +-- in_flight, so a stuck row blocks all future regens for that user. +-- Reset rows where last_run_at is older than 10 minutes (regens +-- shouldn't take that long). +UPDATE system_playlist_runs + SET in_flight = false, + last_error = COALESCE(last_error, 'stuck-row auto-reset by gc') + WHERE in_flight = true + AND last_run_at < now() - INTERVAL '10 minutes'; + +-- name: GcDeleteExpiredPasswordResets :execrows +-- #575: password_resets accumulates expired + used rows forever. The +-- validation path already rejects them; this just keeps the table +-- from growing unbounded. Used rows are kept for 7 days for audit; +-- unused expired rows go immediately. +DELETE FROM password_resets + WHERE (used_at IS NOT NULL AND used_at < now() - INTERVAL '7 days') + OR (used_at IS NULL AND expires_at < now() - INTERVAL '1 hour'); diff --git a/internal/gc/worker.go b/internal/gc/worker.go new file mode 100644 index 00000000..22b42269 --- /dev/null +++ b/internal/gc/worker.go @@ -0,0 +1,101 @@ +// Package gc runs periodic garbage-collection / lifecycle sweeps +// against tables that have NO writer-side close path or NO retention +// policy. Each sweep addresses a drift finding from the 2026-06-02 +// audit (Scribe parent #552) and is idempotent — re-running it on +// already-clean rows is a no-op. +// +// One Worker handles all sweeps so a single long-tick goroutine +// amortises the per-tick fixed cost. Each individual sweep is small +// (single UPDATE / DELETE with a time-bounded WHERE) and emits a +// log line with the affected-row count so the sweep cadence is +// visible in the application log without an explicit metrics layer. +// +// Sweeps: +// - GcCloseStalePlayEvents (#566) +// - GcClosePlaySessionsWithNoRecentEvents (#565) +// - GcExpireScrobbleQueueFailedRows (#567) +// - GcResetStuckSystemPlaylistRuns (#574) +// - GcDeleteExpiredPasswordResets (#575) +package gc + +import ( + "context" + "log/slog" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" +) + +// defaultTick is the production sweep cadence. 1 hour is generous +// since each sweep's WHERE clause uses a multi-hour staleness +// threshold; the worst-case delay between a row becoming sweepable +// and the worker noticing is bounded by tick + threshold. +const defaultTick = 1 * time.Hour + +// Worker holds the pool + logger + tick interval. Construct with +// [NewWorker]; pass the returned Worker to a goroutine that calls +// [Worker.Run] with a context that's cancelled on shutdown. +type Worker struct { + pool *pgxpool.Pool + logger *slog.Logger + tick time.Duration +} + +// NewWorker builds a Worker with the production tick (1h). Tests can +// reach into the Worker after construction to override `tick` for +// faster iteration. +func NewWorker(pool *pgxpool.Pool, logger *slog.Logger) *Worker { + return &Worker{pool: pool, logger: logger, tick: defaultTick} +} + +// Run blocks until ctx is cancelled, running every sweep on each +// tick. Sweeps fire in fixed order; an error in one does NOT abort +// the rest (the panic-vs-just-failed distinction matters here — a +// pgx transient error from one query shouldn't prevent the others +// from running). +func (w *Worker) Run(ctx context.Context) { + // Fire once at start so a freshly-deployed server doesn't wait a + // full tick before doing the initial sweep. Matches the scrobble + // + similarity workers' "sweep then tick" pattern. + w.tickOnce(ctx) + t := time.NewTicker(w.tick) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + w.tickOnce(ctx) + } + } +} + +// tickOnce runs each sweep once, logging the affected-row count. +// Errors are logged per-sweep but do NOT abort the remaining ones — +// each sweep is independent. +func (w *Worker) tickOnce(ctx context.Context) { + q := dbq.New(w.pool) + w.runSweep(ctx, "close_stale_play_events", q.GcCloseStalePlayEvents) + w.runSweep(ctx, "close_play_sessions", q.GcClosePlaySessionsWithNoRecentEvents) + w.runSweep(ctx, "expire_scrobble_failed", q.GcExpireScrobbleQueueFailedRows) + w.runSweep(ctx, "reset_stuck_system_runs", q.GcResetStuckSystemPlaylistRuns) + w.runSweep(ctx, "delete_expired_password_resets", q.GcDeleteExpiredPasswordResets) +} + +// runSweep is a small adapter so each sweep call site is a one-liner +// in tickOnce. Logs at info on rows>0 and debug on rows=0 to keep +// the normal-case (nothing-to-do) noise out of operator logs. +func (w *Worker) runSweep(ctx context.Context, name string, fn func(context.Context) (int64, error)) { + rows, err := fn(ctx) + if err != nil { + w.logger.Error("gc sweep failed", "sweep", name, "err", err) + return + } + if rows > 0 { + w.logger.Info("gc sweep", "sweep", name, "rows_affected", rows) + } else { + w.logger.Debug("gc sweep", "sweep", name, "rows_affected", 0) + } +} diff --git a/internal/gc/worker_test.go b/internal/gc/worker_test.go new file mode 100644 index 00000000..eb7ef9ee --- /dev/null +++ b/internal/gc/worker_test.go @@ -0,0 +1,317 @@ +package gc + +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" + "git.fabledsword.com/bvandeusen/minstrel/internal/dbtest" +) + +// testWorker constructs a Worker against MINSTREL_TEST_DATABASE_URL. +// Mirrors the api package's testHandlers pattern — skip when not in +// integration mode, migrate + reset, return the pool for the caller +// to seed. +func testWorker(t *testing.T) (*Worker, *pgxpool.Pool) { + t.Helper() + if testing.Short() { + t.Skip("skipping gc 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) + dbtest.ResetDB(t, pool) + return NewWorker(pool, logger), pool +} + +// seedUser creates a minimal user row for tests that need play_events +// / play_sessions / scrobble_queue rows. Username is prefixed so +// dbtest.ResetDB cleans up between runs. +func seedUser(t *testing.T, pool *pgxpool.Pool, name string) pgtype.UUID { + t.Helper() + u, err := dbq.New(pool).CreateUser(context.Background(), dbq.CreateUserParams{ + Username: dbtest.TestUserPrefix + name, + PasswordHash: "test-hash", + ApiToken: "test-token-" + name, + IsAdmin: false, + }) + if err != nil { + t.Fatalf("CreateUser: %v", err) + } + return u.ID +} + +func TestGcCloseStalePlayEvents_ClosesOnly24hOldRows(t *testing.T) { + w, pool := testWorker(t) + ctx := context.Background() + userID := seedUser(t, pool, "alice") + + // Need a track + session to satisfy FKs on play_events. + var trackID pgtype.UUID + if err := pool.QueryRow(ctx, ` + INSERT INTO artists (name) VALUES ('A') RETURNING id + `).Scan(new(pgtype.UUID)); err != nil { + t.Fatalf("seed artist row: %v", err) + } + // Get the just-inserted artist id. + var artistID pgtype.UUID + if err := pool.QueryRow(ctx, `SELECT id FROM artists WHERE name = 'A'`).Scan(&artistID); err != nil { + t.Fatalf("read artist: %v", err) + } + var albumID pgtype.UUID + if err := pool.QueryRow(ctx, ` + INSERT INTO albums (artist_id, title, sort_title) VALUES ($1, 'X', 'X') RETURNING id + `, artistID).Scan(&albumID); err != nil { + t.Fatalf("seed album: %v", err) + } + if err := pool.QueryRow(ctx, ` + INSERT INTO tracks (album_id, artist_id, title, file_path, duration_ms) + VALUES ($1, $2, 'T', '/x.mp3', 180000) RETURNING id + `, albumID, artistID).Scan(&trackID); err != nil { + t.Fatalf("seed track: %v", err) + } + var sessionID pgtype.UUID + if err := pool.QueryRow(ctx, ` + INSERT INTO play_sessions (user_id, started_at, last_event_at) + VALUES ($1, now() - interval '2 hours', now()) RETURNING id + `, userID).Scan(&sessionID); err != nil { + t.Fatalf("seed session: %v", err) + } + + // Stale row (25h old, no ended_at) — should be closed. + if _, err := pool.Exec(ctx, ` + INSERT INTO play_events (user_id, track_id, session_id, started_at) + VALUES ($1, $2, $3, now() - interval '25 hours') + `, userID, trackID, sessionID); err != nil { + t.Fatalf("seed stale event: %v", err) + } + // Fresh row (1h old, no ended_at) — should be left alone. + if _, err := pool.Exec(ctx, ` + INSERT INTO play_events (user_id, track_id, session_id, started_at) + VALUES ($1, $2, $3, now() - interval '1 hour') + `, userID, trackID, sessionID); err != nil { + t.Fatalf("seed fresh event: %v", err) + } + + w.tickOnce(ctx) + + var closedStale, openFresh bool + if err := pool.QueryRow(ctx, ` + SELECT EXISTS(SELECT 1 FROM play_events + WHERE started_at < now() - interval '24 hours' + AND ended_at IS NOT NULL) + `).Scan(&closedStale); err != nil { + t.Fatalf("check stale: %v", err) + } + if err := pool.QueryRow(ctx, ` + SELECT EXISTS(SELECT 1 FROM play_events + WHERE started_at > now() - interval '2 hours' + AND ended_at IS NULL) + `).Scan(&openFresh); err != nil { + t.Fatalf("check fresh: %v", err) + } + if !closedStale { + t.Errorf("stale play_events row not closed") + } + if !openFresh { + t.Errorf("fresh play_events row was closed (should be left alone)") + } +} + +func TestGcClosePlaySessions_ClosesIdleAndEmptyStarvedSessions(t *testing.T) { + w, pool := testWorker(t) + ctx := context.Background() + userID := seedUser(t, pool, "bob") + + // Idle session — has events, last_event_at 8h ago. + if _, err := pool.Exec(ctx, ` + INSERT INTO play_sessions (user_id, started_at, last_event_at, track_count) + VALUES ($1, now() - interval '8 hours', now() - interval '8 hours', 5) + `, userID); err != nil { + t.Fatalf("seed idle session: %v", err) + } + // Empty-starved session — no events, started 2h ago. + if _, err := pool.Exec(ctx, ` + INSERT INTO play_sessions (user_id, started_at, last_event_at, track_count) + VALUES ($1, now() - interval '2 hours', now() - interval '2 hours', 0) + `, userID); err != nil { + t.Fatalf("seed empty session: %v", err) + } + // Active session — recent last_event_at, has events. + if _, err := pool.Exec(ctx, ` + INSERT INTO play_sessions (user_id, started_at, last_event_at, track_count) + VALUES ($1, now() - interval '30 minutes', now() - interval '5 minutes', 3) + `, userID); err != nil { + t.Fatalf("seed active session: %v", err) + } + + w.tickOnce(ctx) + + var closedCount, openCount int + if err := pool.QueryRow(ctx, ` + SELECT count(*) FROM play_sessions + WHERE user_id = $1 AND ended_at IS NOT NULL + `, userID).Scan(&closedCount); err != nil { + t.Fatalf("count closed: %v", err) + } + if err := pool.QueryRow(ctx, ` + SELECT count(*) FROM play_sessions + WHERE user_id = $1 AND ended_at IS NULL + `, userID).Scan(&openCount); err != nil { + t.Fatalf("count open: %v", err) + } + if closedCount != 2 { + t.Errorf("closed sessions = %d, want 2 (idle + empty-starved)", closedCount) + } + if openCount != 1 { + t.Errorf("open sessions = %d, want 1 (active)", openCount) + } +} + +func TestGcResetStuckSystemPlaylistRuns(t *testing.T) { + w, pool := testWorker(t) + ctx := context.Background() + stuckUser := seedUser(t, pool, "stuck") + activeUser := seedUser(t, pool, "active") + + // Stuck row: in_flight, last_run_at 30 min ago. + if _, err := pool.Exec(ctx, ` + INSERT INTO system_playlist_runs (user_id, last_run_at, in_flight) + VALUES ($1, now() - interval '30 minutes', true) + `, stuckUser); err != nil { + t.Fatalf("seed stuck run: %v", err) + } + // Active row: in_flight, last_run_at 2 min ago — still legitimate. + if _, err := pool.Exec(ctx, ` + INSERT INTO system_playlist_runs (user_id, last_run_at, in_flight) + VALUES ($1, now() - interval '2 minutes', true) + `, activeUser); err != nil { + t.Fatalf("seed active run: %v", err) + } + + w.tickOnce(ctx) + + var stuckFlight, activeFlight bool + if err := pool.QueryRow(ctx, ` + SELECT in_flight FROM system_playlist_runs WHERE user_id = $1 + `, stuckUser).Scan(&stuckFlight); err != nil { + t.Fatalf("read stuck row: %v", err) + } + if err := pool.QueryRow(ctx, ` + SELECT in_flight FROM system_playlist_runs WHERE user_id = $1 + `, activeUser).Scan(&activeFlight); err != nil { + t.Fatalf("read active row: %v", err) + } + if stuckFlight { + t.Errorf("stuck row still in_flight after sweep") + } + if !activeFlight { + t.Errorf("active row was reset (should be left alone — only 2 min old)") + } +} + +func TestGcDeleteExpiredPasswordResets(t *testing.T) { + w, pool := testWorker(t) + ctx := context.Background() + userID := seedUser(t, pool, "pwd") + + // Expired-unused (> 1h past expires_at) — should delete. + if _, err := pool.Exec(ctx, ` + INSERT INTO password_resets (token, user_id, expires_at) + VALUES ('expired-old', $1, now() - interval '2 hours') + `, userID); err != nil { + t.Fatalf("seed expired: %v", err) + } + // Used (> 7 days past used_at) — should delete. + if _, err := pool.Exec(ctx, ` + INSERT INTO password_resets (token, user_id, expires_at, used_at) + VALUES ('used-old', $1, now() - interval '8 days', now() - interval '8 days') + `, userID); err != nil { + t.Fatalf("seed used-old: %v", err) + } + // Active (expires in future) — should survive. + if _, err := pool.Exec(ctx, ` + INSERT INTO password_resets (token, user_id, expires_at) + VALUES ('active', $1, now() + interval '1 hour') + `, userID); err != nil { + t.Fatalf("seed active: %v", err) + } + // Recently used (< 7 days) — should survive for audit. + if _, err := pool.Exec(ctx, ` + INSERT INTO password_resets (token, user_id, expires_at, used_at) + VALUES ('used-recent', $1, now() - interval '1 hour', now() - interval '1 day') + `, userID); err != nil { + t.Fatalf("seed used-recent: %v", err) + } + + w.tickOnce(ctx) + + var remaining []string + rows, err := pool.Query(ctx, `SELECT token FROM password_resets WHERE user_id = $1`, userID) + if err != nil { + t.Fatalf("list remaining: %v", err) + } + defer rows.Close() + for rows.Next() { + var tok string + if err := rows.Scan(&tok); err != nil { + t.Fatalf("scan: %v", err) + } + remaining = append(remaining, tok) + } + wantSet := map[string]bool{"active": true, "used-recent": true} + if len(remaining) != len(wantSet) { + t.Errorf("remaining tokens = %v, want %v", remaining, []string{"active", "used-recent"}) + } + for _, tok := range remaining { + if !wantSet[tok] { + t.Errorf("unexpected surviving token %q", tok) + } + } +} + +// Verifies tickOnce doesn't blow up when the tables are completely +// empty — sweeps just no-op. Regression guard against an EXEC vs +// QUERY-row-count mismatch failing on zero rows. +func TestGcTickOnce_NoOpOnEmptyTables(t *testing.T) { + w, _ := testWorker(t) + w.tickOnce(context.Background()) +} + +// Sanity check on the Run() loop's cancel behaviour — we don't want +// to leave a goroutine spinning at test-runner exit. 10ms tick with +// an immediate cancel should return promptly. +func TestGcRun_HonoursContextCancel(t *testing.T) { + w, _ := testWorker(t) + w.tick = 10 * time.Millisecond + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + w.Run(ctx) + close(done) + }() + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Run did not return after cancel") + } +}