package scrobble import ( "context" "errors" "log/slog" "time" "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/scrobble/listenbrainz" ) // backoffSchedule maps the failure count (1-indexed) to the delay before // the next attempt. Per spec §M4a: 1m → 5m → 30m → 2h → 6h, then give up. var backoffSchedule = []time.Duration{ 1 * time.Minute, 5 * time.Minute, 30 * time.Minute, 2 * time.Hour, 6 * time.Hour, } const maxAttempts = 5 // backoffDelay returns the delay before the next attempt given the value // of the `attempts` column AFTER the failure has been recorded (1-indexed: // attempts=1 means the first failure just happened, the next retry should // wait 1 minute). Returns (_, false) when attempts > maxAttempts (the // caller should mark the row failed). func backoffDelay(attempts int) (time.Duration, bool) { if attempts < 1 || attempts > maxAttempts { return 0, false } return backoffSchedule[attempts-1], true } // Worker drains scrobble_queue rows and POSTs them to ListenBrainz. type Worker struct { pool *pgxpool.Pool client *listenbrainz.Client logger *slog.Logger tick time.Duration batch int32 } // NewWorker constructs a worker with production defaults: 30s tick, 50-row // batch. func NewWorker(pool *pgxpool.Pool, client *listenbrainz.Client, logger *slog.Logger) *Worker { return &Worker{pool: pool, client: client, logger: logger, tick: 30 * time.Second, batch: 50} } // Run blocks until ctx is cancelled, ticking every w.tick. func (w *Worker) Run(ctx context.Context) { t := time.NewTicker(w.tick) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: if err := w.tickOnce(ctx); err != nil { w.logger.Error("scrobble: tick failed", "err", err) } } } } // tickOnce drains up to w.batch pending rows. Returns the first error // encountered loading rows; per-row errors are logged and don't abort // the batch. func (w *Worker) tickOnce(ctx context.Context) error { q := dbq.New(w.pool) rows, err := q.ListPendingScrobbles(ctx, w.batch) if err != nil { return err } if len(rows) == 0 { return nil } // Group rows by user (each user has their own token + may produce its // own batch). Map keyed by user UUID string. type userBatch struct { token string queueIDs []pgtype.UUID listens []listenbrainz.Listen userID pgtype.UUID } batches := map[string]*userBatch{} for _, r := range rows { if r.LbToken == nil || *r.LbToken == "" || !r.LbEnabled { // Defensive: row got into queue but user since disabled. Skip. continue } key := r.UserID.String() ub := batches[key] if ub == nil { ub = &userBatch{token: *r.LbToken, userID: r.UserID} batches[key] = ub } ub.queueIDs = append(ub.queueIDs, r.QueueID) l := listenbrainz.Listen{ ListenedAt: r.StartedAt.Time.Unix(), Track: listenbrainz.Track{ ArtistName: r.ArtistName, TrackName: r.TrackTitle, ReleaseName: r.AlbumTitle, DurationMs: int(r.TrackDurationMs), }, } if r.TrackMbid != nil { l.Track.RecordingMBID = *r.TrackMbid } if r.AlbumMbid != nil { l.Track.ReleaseMBID = *r.AlbumMbid } if r.ArtistMbid != nil { l.Track.ArtistMBIDs = []string{*r.ArtistMbid} } ub.listens = append(ub.listens, l) } for _, ub := range batches { w.processBatch(ctx, q, ub.userID, ub.token, ub.queueIDs, ub.listens) } return nil } // processBatch submits one user's pending batch and updates queue rows // according to the response. func (w *Worker) processBatch( ctx context.Context, q *dbq.Queries, userID pgtype.UUID, token string, queueIDs []pgtype.UUID, listens []listenbrainz.Listen, ) { err := w.client.SubmitListens(ctx, token, listens) now := time.Now().UTC() if err == nil { for _, id := range queueIDs { if uerr := q.MarkScrobbleSent(ctx, id); uerr != nil { w.logger.Error("scrobble: MarkScrobbleSent", "queue_id", id, "err", uerr) } } return } // 429: schedule retry per Retry-After, do NOT increment attempts. var ra *listenbrainz.RetryAfterError if errors.As(err, &ra) { next := pgtype.Timestamptz{Time: now.Add(ra.Wait), Valid: true} for _, id := range queueIDs { if uerr := q.MarkScrobbleRetryAfter(ctx, dbq.MarkScrobbleRetryAfterParams{ ID: id, NextAttemptAt: next, }); uerr != nil { w.logger.Error("scrobble: MarkScrobbleRetryAfter", "err", uerr) } } return } // 401: mark failed AND disable user (defensive — bad token shouldn't keep // retrying or feed future rows). if errors.Is(err, listenbrainz.ErrAuth) { w.logger.Warn("scrobble: auth rejected; disabling user listenbrainz", "user_id", userID) if uerr := q.SetListenBrainzEnabled(ctx, dbq.SetListenBrainzEnabledParams{ ID: userID, ListenbrainzEnabled: false, }); uerr != nil { w.logger.Error("scrobble: SetListenBrainzEnabled", "err", uerr) } errStr := err.Error() for _, id := range queueIDs { if uerr := q.MarkScrobbleFailed(ctx, dbq.MarkScrobbleFailedParams{ ID: id, LastError: &errStr, }); uerr != nil { w.logger.Error("scrobble: MarkScrobbleFailed", "err", uerr) } } return } // 4xx: permanent. Mark failed, no retry. if errors.Is(err, listenbrainz.ErrPermanent) { errStr := err.Error() for _, id := range queueIDs { if uerr := q.MarkScrobbleFailed(ctx, dbq.MarkScrobbleFailedParams{ ID: id, LastError: &errStr, }); uerr != nil { w.logger.Error("scrobble: MarkScrobbleFailed", "err", uerr) } } return } // Transient (5xx, network): per-row, look up backoffDelay(currentAttempts+1). // If OK → RescheduleScrobble (which increments). If NOT OK → MarkScrobbleFailed // (don't increment further; current attempts already represents the cap). errStr := err.Error() for _, id := range queueIDs { var attempts int32 if rerr := w.pool.QueryRow(ctx, `SELECT attempts FROM scrobble_queue WHERE id = $1`, id).Scan(&attempts); rerr != nil { w.logger.Error("scrobble: read attempts", "err", rerr) continue } if delay, ok := backoffDelay(int(attempts) + 1); ok { nextAt := pgtype.Timestamptz{Time: now.Add(delay), Valid: true} if uerr := q.RescheduleScrobble(ctx, dbq.RescheduleScrobbleParams{ ID: id, NextAttemptAt: nextAt, LastError: &errStr, }); uerr != nil { w.logger.Error("scrobble: RescheduleScrobble", "err", uerr) } } else { if uerr := q.MarkScrobbleFailed(ctx, dbq.MarkScrobbleFailedParams{ ID: id, LastError: &errStr, }); uerr != nil { w.logger.Error("scrobble: MarkScrobbleFailed", "err", uerr) } } } }