feat(db): add ListenBrainz user-config and scrobble_queue queries

2026-04-28 08:42:38 -04:00
parent efc2308348
commit a0e21f21b6
5 changed files with 380 additions and 11 deletions
@@ -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
 }
@@ -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
+}
@@ -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
 `
@@ -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;
@@ -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;