From a0e21f21b6e306e4752bda8c756afb9c4790e6ff Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 28 Apr 2026 08:42:38 -0400 Subject: [PATCH] feat(db): add ListenBrainz user-config and scrobble_queue queries --- internal/db/dbq/models.go | 27 +++-- internal/db/dbq/scrobble.sql.go | 193 +++++++++++++++++++++++++++++++ internal/db/dbq/users.sql.go | 75 +++++++++++- internal/db/queries/scrobble.sql | 73 ++++++++++++ internal/db/queries/users.sql | 23 ++++ 5 files changed, 380 insertions(+), 11 deletions(-) create mode 100644 internal/db/dbq/scrobble.sql.go create mode 100644 internal/db/queries/scrobble.sql diff --git a/internal/db/dbq/models.go b/internal/db/dbq/models.go index 04475635..e7b7e229 100644 --- a/internal/db/dbq/models.go +++ b/internal/db/dbq/models.go @@ -82,6 +82,17 @@ type PlaySession struct { ClientID *string } +type ScrobbleQueue struct { + ID pgtype.UUID + UserID pgtype.UUID + PlayEventID pgtype.UUID + Status string + Attempts int32 + NextAttemptAt pgtype.Timestamptz + LastError *string + EnqueuedAt pgtype.Timestamptz +} + type Session struct { ID pgtype.UUID UserID pgtype.UUID @@ -119,11 +130,13 @@ type Track struct { } type User struct { - ID pgtype.UUID - Username string - PasswordHash string - ApiToken string - IsAdmin bool - CreatedAt pgtype.Timestamptz - SubsonicPassword *string + ID pgtype.UUID + Username string + PasswordHash string + ApiToken string + IsAdmin bool + CreatedAt pgtype.Timestamptz + SubsonicPassword *string + ListenbrainzToken *string + ListenbrainzEnabled bool } diff --git a/internal/db/dbq/scrobble.sql.go b/internal/db/dbq/scrobble.sql.go new file mode 100644 index 00000000..8c8da74c --- /dev/null +++ b/internal/db/dbq/scrobble.sql.go @@ -0,0 +1,193 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.27.0 +// source: scrobble.sql + +package dbq + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const enqueueScrobble = `-- name: EnqueueScrobble :execrows +INSERT INTO scrobble_queue (user_id, play_event_id, status) +VALUES ($1, $2, 'pending') +ON CONFLICT (play_event_id) DO NOTHING +` + +type EnqueueScrobbleParams struct { + UserID pgtype.UUID + PlayEventID pgtype.UUID +} + +// Idempotent: if a row exists for this play_event_id, ON CONFLICT does nothing +// and the affected-rows count is 0. Caller (MaybeEnqueue) doesn't treat 0 as +// an error. +func (q *Queries) EnqueueScrobble(ctx context.Context, arg EnqueueScrobbleParams) (int64, error) { + result, err := q.db.Exec(ctx, enqueueScrobble, arg.UserID, arg.PlayEventID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const listPendingScrobbles = `-- name: ListPendingScrobbles :many +SELECT + sq.id AS queue_id, + sq.user_id AS user_id, + sq.play_event_id AS play_event_id, + sq.attempts AS attempts, + u.listenbrainz_token AS lb_token, + u.listenbrainz_enabled AS lb_enabled, + pe.started_at AS started_at, + t.title AS track_title, + t.duration_ms AS track_duration_ms, + t.mbid AS track_mbid, + al.title AS album_title, + al.mbid AS album_mbid, + ar.name AS artist_name, + ar.mbid AS artist_mbid +FROM scrobble_queue sq +JOIN users u ON u.id = sq.user_id +JOIN play_events pe ON pe.id = sq.play_event_id +JOIN tracks t ON t.id = pe.track_id +JOIN albums al ON al.id = t.album_id +JOIN artists ar ON ar.id = t.artist_id +WHERE sq.status = 'pending' + AND sq.next_attempt_at <= now() +ORDER BY sq.next_attempt_at +LIMIT $1 +` + +type ListPendingScrobblesRow struct { + QueueID pgtype.UUID + UserID pgtype.UUID + PlayEventID pgtype.UUID + Attempts int32 + LbToken *string + LbEnabled bool + StartedAt pgtype.Timestamptz + TrackTitle string + TrackDurationMs int32 + TrackMbid *string + AlbumTitle string + AlbumMbid *string + ArtistName string + ArtistMbid *string +} + +// Worker pulls up to N pending rows whose next_attempt_at has passed. +// Joins play_events + tracks/albums/artists + users so the worker can +// assemble the LB Listen payload in one round-trip. +func (q *Queries) ListPendingScrobbles(ctx context.Context, limit int32) ([]ListPendingScrobblesRow, error) { + rows, err := q.db.Query(ctx, listPendingScrobbles, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListPendingScrobblesRow + for rows.Next() { + var i ListPendingScrobblesRow + if err := rows.Scan( + &i.QueueID, + &i.UserID, + &i.PlayEventID, + &i.Attempts, + &i.LbToken, + &i.LbEnabled, + &i.StartedAt, + &i.TrackTitle, + &i.TrackDurationMs, + &i.TrackMbid, + &i.AlbumTitle, + &i.AlbumMbid, + &i.ArtistName, + &i.ArtistMbid, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const markScrobbleFailed = `-- name: MarkScrobbleFailed :exec +UPDATE scrobble_queue +SET status = 'failed', + last_error = $2 +WHERE id = $1 +` + +type MarkScrobbleFailedParams struct { + ID pgtype.UUID + LastError *string +} + +// After a permanent failure (4xx, exhausted retries, auth) — mark failed +// and keep the row for diagnostics. +func (q *Queries) MarkScrobbleFailed(ctx context.Context, arg MarkScrobbleFailedParams) error { + _, err := q.db.Exec(ctx, markScrobbleFailed, arg.ID, arg.LastError) + return err +} + +const markScrobbleRetryAfter = `-- name: MarkScrobbleRetryAfter :exec +UPDATE scrobble_queue +SET next_attempt_at = $2 +WHERE id = $1 +` + +type MarkScrobbleRetryAfterParams struct { + ID pgtype.UUID + NextAttemptAt pgtype.Timestamptz +} + +// After a 429: schedule next attempt per LB's Retry-After header, but do NOT +// increment attempts (the server told us to wait, not that we failed). +func (q *Queries) MarkScrobbleRetryAfter(ctx context.Context, arg MarkScrobbleRetryAfterParams) error { + _, err := q.db.Exec(ctx, markScrobbleRetryAfter, arg.ID, arg.NextAttemptAt) + return err +} + +const markScrobbleSent = `-- name: MarkScrobbleSent :exec +WITH del AS ( + DELETE FROM scrobble_queue + WHERE scrobble_queue.id = $1 + RETURNING play_event_id +) +UPDATE play_events +SET scrobbled_at = now() +WHERE play_events.id = (SELECT play_event_id FROM del) +` + +// Successful submit. Delete the queue row AND stamp play_events.scrobbled_at +// in one statement via a CTE so partial failure is impossible. +func (q *Queries) MarkScrobbleSent(ctx context.Context, id pgtype.UUID) error { + _, err := q.db.Exec(ctx, markScrobbleSent, id) + return err +} + +const rescheduleScrobble = `-- name: RescheduleScrobble :exec +UPDATE scrobble_queue +SET attempts = attempts + 1, + next_attempt_at = $2, + last_error = $3 +WHERE id = $1 +` + +type RescheduleScrobbleParams struct { + ID pgtype.UUID + NextAttemptAt pgtype.Timestamptz + LastError *string +} + +// After a transient failure: increment attempts, schedule next attempt, +// record the error. +func (q *Queries) RescheduleScrobble(ctx context.Context, arg RescheduleScrobbleParams) error { + _, err := q.db.Exec(ctx, rescheduleScrobble, arg.ID, arg.NextAttemptAt, arg.LastError) + return err +} diff --git a/internal/db/dbq/users.sql.go b/internal/db/dbq/users.sql.go index 2e7dccbf..a5bfef95 100644 --- a/internal/db/dbq/users.sql.go +++ b/internal/db/dbq/users.sql.go @@ -25,7 +25,7 @@ func (q *Queries) CountUsers(ctx context.Context) (int64, error) { const createUser = `-- name: CreateUser :one INSERT INTO users (username, password_hash, api_token, is_admin) VALUES ($1, $2, $3, $4) -RETURNING id, username, password_hash, api_token, is_admin, created_at, subsonic_password +RETURNING id, username, password_hash, api_token, is_admin, created_at, subsonic_password, listenbrainz_token, listenbrainz_enabled ` type CreateUserParams struct { @@ -51,12 +51,40 @@ func (q *Queries) CreateUser(ctx context.Context, arg CreateUserParams) (User, e &i.IsAdmin, &i.CreatedAt, &i.SubsonicPassword, + &i.ListenbrainzToken, + &i.ListenbrainzEnabled, ) return i, err } +const getListenBrainzConfig = `-- name: GetListenBrainzConfig :one +SELECT + u.listenbrainz_token, + u.listenbrainz_enabled, + (SELECT MAX(pe.scrobbled_at) + FROM play_events pe + WHERE pe.user_id = u.id) AS last_scrobbled_at +FROM users u +WHERE u.id = $1 +` + +type GetListenBrainzConfigRow struct { + ListenbrainzToken *string + ListenbrainzEnabled bool + LastScrobbledAt interface{} +} + +// Returns the user's LB token + enabled flag and the most recent +// play_events.scrobbled_at for last-scrobbled-at status. +func (q *Queries) GetListenBrainzConfig(ctx context.Context, id pgtype.UUID) (GetListenBrainzConfigRow, error) { + row := q.db.QueryRow(ctx, getListenBrainzConfig, id) + var i GetListenBrainzConfigRow + err := row.Scan(&i.ListenbrainzToken, &i.ListenbrainzEnabled, &i.LastScrobbledAt) + return i, err +} + const getUserByAPIToken = `-- name: GetUserByAPIToken :one -SELECT id, username, password_hash, api_token, is_admin, created_at, subsonic_password FROM users WHERE api_token = $1 +SELECT id, username, password_hash, api_token, is_admin, created_at, subsonic_password, listenbrainz_token, listenbrainz_enabled FROM users WHERE api_token = $1 ` func (q *Queries) GetUserByAPIToken(ctx context.Context, apiToken string) (User, error) { @@ -70,12 +98,14 @@ func (q *Queries) GetUserByAPIToken(ctx context.Context, apiToken string) (User, &i.IsAdmin, &i.CreatedAt, &i.SubsonicPassword, + &i.ListenbrainzToken, + &i.ListenbrainzEnabled, ) return i, err } const getUserByID = `-- name: GetUserByID :one -SELECT id, username, password_hash, api_token, is_admin, created_at, subsonic_password FROM users WHERE id = $1 +SELECT id, username, password_hash, api_token, is_admin, created_at, subsonic_password, listenbrainz_token, listenbrainz_enabled FROM users WHERE id = $1 ` func (q *Queries) GetUserByID(ctx context.Context, id pgtype.UUID) (User, error) { @@ -89,12 +119,14 @@ func (q *Queries) GetUserByID(ctx context.Context, id pgtype.UUID) (User, error) &i.IsAdmin, &i.CreatedAt, &i.SubsonicPassword, + &i.ListenbrainzToken, + &i.ListenbrainzEnabled, ) return i, err } const getUserByUsername = `-- name: GetUserByUsername :one -SELECT id, username, password_hash, api_token, is_admin, created_at, subsonic_password FROM users WHERE username = $1 +SELECT id, username, password_hash, api_token, is_admin, created_at, subsonic_password, listenbrainz_token, listenbrainz_enabled FROM users WHERE username = $1 ` func (q *Queries) GetUserByUsername(ctx context.Context, username string) (User, error) { @@ -108,10 +140,45 @@ func (q *Queries) GetUserByUsername(ctx context.Context, username string) (User, &i.IsAdmin, &i.CreatedAt, &i.SubsonicPassword, + &i.ListenbrainzToken, + &i.ListenbrainzEnabled, ) return i, err } +const setListenBrainzEnabled = `-- name: SetListenBrainzEnabled :exec +UPDATE users +SET listenbrainz_enabled = $2 +WHERE id = $1 +` + +type SetListenBrainzEnabledParams struct { + ID pgtype.UUID + ListenbrainzEnabled bool +} + +func (q *Queries) SetListenBrainzEnabled(ctx context.Context, arg SetListenBrainzEnabledParams) error { + _, err := q.db.Exec(ctx, setListenBrainzEnabled, arg.ID, arg.ListenbrainzEnabled) + return err +} + +const setListenBrainzToken = `-- name: SetListenBrainzToken :exec +UPDATE users +SET listenbrainz_token = $2, + listenbrainz_enabled = CASE WHEN COALESCE($2, '') = '' THEN FALSE ELSE listenbrainz_enabled END +WHERE id = $1 +` + +type SetListenBrainzTokenParams struct { + ID pgtype.UUID + ListenbrainzToken *string +} + +func (q *Queries) SetListenBrainzToken(ctx context.Context, arg SetListenBrainzTokenParams) error { + _, err := q.db.Exec(ctx, setListenBrainzToken, arg.ID, arg.ListenbrainzToken) + return err +} + const setSubsonicPassword = `-- name: SetSubsonicPassword :exec UPDATE users SET subsonic_password = $2 WHERE id = $1 ` diff --git a/internal/db/queries/scrobble.sql b/internal/db/queries/scrobble.sql new file mode 100644 index 00000000..bc71b0d9 --- /dev/null +++ b/internal/db/queries/scrobble.sql @@ -0,0 +1,73 @@ +-- name: EnqueueScrobble :execrows +-- Idempotent: if a row exists for this play_event_id, ON CONFLICT does nothing +-- and the affected-rows count is 0. Caller (MaybeEnqueue) doesn't treat 0 as +-- an error. +INSERT INTO scrobble_queue (user_id, play_event_id, status) +VALUES ($1, $2, 'pending') +ON CONFLICT (play_event_id) DO NOTHING; + +-- name: ListPendingScrobbles :many +-- Worker pulls up to N pending rows whose next_attempt_at has passed. +-- Joins play_events + tracks/albums/artists + users so the worker can +-- assemble the LB Listen payload in one round-trip. +SELECT + sq.id AS queue_id, + sq.user_id AS user_id, + sq.play_event_id AS play_event_id, + sq.attempts AS attempts, + u.listenbrainz_token AS lb_token, + u.listenbrainz_enabled AS lb_enabled, + pe.started_at AS started_at, + t.title AS track_title, + t.duration_ms AS track_duration_ms, + t.mbid AS track_mbid, + al.title AS album_title, + al.mbid AS album_mbid, + ar.name AS artist_name, + ar.mbid AS artist_mbid +FROM scrobble_queue sq +JOIN users u ON u.id = sq.user_id +JOIN play_events pe ON pe.id = sq.play_event_id +JOIN tracks t ON t.id = pe.track_id +JOIN albums al ON al.id = t.album_id +JOIN artists ar ON ar.id = t.artist_id +WHERE sq.status = 'pending' + AND sq.next_attempt_at <= now() +ORDER BY sq.next_attempt_at +LIMIT $1; + +-- name: MarkScrobbleSent :exec +-- Successful submit. Delete the queue row AND stamp play_events.scrobbled_at +-- in one statement via a CTE so partial failure is impossible. +WITH del AS ( + DELETE FROM scrobble_queue + WHERE scrobble_queue.id = $1 + RETURNING play_event_id +) +UPDATE play_events +SET scrobbled_at = now() +WHERE play_events.id = (SELECT play_event_id FROM del); + +-- name: RescheduleScrobble :exec +-- After a transient failure: increment attempts, schedule next attempt, +-- record the error. +UPDATE scrobble_queue +SET attempts = attempts + 1, + next_attempt_at = $2, + last_error = $3 +WHERE id = $1; + +-- name: MarkScrobbleRetryAfter :exec +-- After a 429: schedule next attempt per LB's Retry-After header, but do NOT +-- increment attempts (the server told us to wait, not that we failed). +UPDATE scrobble_queue +SET next_attempt_at = $2 +WHERE id = $1; + +-- name: MarkScrobbleFailed :exec +-- After a permanent failure (4xx, exhausted retries, auth) — mark failed +-- and keep the row for diagnostics. +UPDATE scrobble_queue +SET status = 'failed', + last_error = $2 +WHERE id = $1; diff --git a/internal/db/queries/users.sql b/internal/db/queries/users.sql index 4710ed75..f1109a0f 100644 --- a/internal/db/queries/users.sql +++ b/internal/db/queries/users.sql @@ -19,3 +19,26 @@ UPDATE users SET subsonic_password = $2 WHERE id = $1; -- name: GetUserByID :one SELECT * FROM users WHERE id = $1; + +-- name: SetListenBrainzToken :exec +UPDATE users +SET listenbrainz_token = $2, + listenbrainz_enabled = CASE WHEN COALESCE($2, '') = '' THEN FALSE ELSE listenbrainz_enabled END +WHERE id = $1; + +-- name: SetListenBrainzEnabled :exec +UPDATE users +SET listenbrainz_enabled = $2 +WHERE id = $1; + +-- name: GetListenBrainzConfig :one +-- Returns the user's LB token + enabled flag and the most recent +-- play_events.scrobbled_at for last-scrobbled-at status. +SELECT + u.listenbrainz_token, + u.listenbrainz_enabled, + (SELECT MAX(pe.scrobbled_at) + FROM play_events pe + WHERE pe.user_id = u.id) AS last_scrobbled_at +FROM users u +WHERE u.id = $1;