pipu-music/services/api/internal/genius/service.go

package genius

import (
	"context"
	"encoding/json"
	"errors"
	"log/slog"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"pipu/api/internal/api/gen"
)

type Service struct {
	pool       *pgxpool.Pool
	client     *Client
	configured bool
	log        *slog.Logger
	wake       chan struct{}
}

func NewService(pool *pgxpool.Pool, token string, interval time.Duration, log *slog.Logger) *Service {
	client := New(token)
	client.SetInterval(interval)
	return &Service{pool: pool, client: client, configured: strings.TrimSpace(token) != "", log: log, wake: make(chan struct{}, 1)}
}

func (s *Service) Wake() {
	select {
	case s.wake <- struct{}{}:
	default:
	}
}

// Run discovers new tracks, including imports, without making catalog requests wait for Genius.
func (s *Service) Run(ctx context.Context) {
	if !s.configured {
		return
	}
	timer := time.NewTicker(30 * time.Second)
	defer timer.Stop()
	for {
		if err := s.drain(ctx); err != nil && ctx.Err() == nil {
			s.log.Error("genius worker", "err", err)
		}
		select {
		case <-ctx.Done():
			return
		case <-timer.C:
		case <-s.wake:
		}
	}
}

func (s *Service) drain(ctx context.Context) error {
	_, err := s.pool.Exec(ctx, `INSERT INTO genius_tracks(track_id) SELECT id FROM tracks ON CONFLICT DO NOTHING`)
	if err != nil {
		return err
	}
	for ctx.Err() == nil {
		var id uuid.UUID
		// Atomic claim and persisted lease recover interrupted work after a restart.
		err := s.pool.QueryRow(ctx, `UPDATE genius_tracks SET status='processing', locked_until=now()+interval '10 minutes', attempts=attempts+1
   WHERE track_id=(SELECT track_id FROM genius_tracks WHERE status IN ('pending','processing','error','partial')
    AND next_attempt<=now() AND locked_until<=now()
    AND EXISTS(SELECT 1 FROM genius_state WHERE NOT paused AND cooldown_until<=now())
    ORDER BY next_attempt FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING track_id`).Scan(&id)
		if errors.Is(err, pgx.ErrNoRows) {
			return nil
		}
		if err != nil {
			return err
		}
		jobCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
		content, albumImage, err := s.enrich(jobCtx, id)
		cancel()
		if ctx.Err() != nil {
			return ctx.Err()
		}
		status, message, next := "ready", "", time.Now().Add(30*24*time.Hour)
		if content.SourceUrl == "" {
			status = "unmatched"
		}
		if err != nil {
			message = errorCode(err)
			status = "error"
			if content.SourceUrl != "" {
				status = "partial"
			}
			next = time.Now().Add(6 * time.Hour)
		}
		content.Status = status
		body, marshalErr := json.Marshal(content)
		if marshalErr != nil {
			return marshalErr
		}
		tx, txErr := s.pool.Begin(ctx)
		if txErr != nil {
			return txErr
		}
		saveErr := func() error {
			if _, e := tx.Exec(ctx, `UPDATE genius_tracks SET status=$2,content=$3,last_error=$4,next_attempt=$5,
    locked_until='1970-01-01',updated_at=now() WHERE track_id=$1`, id, status, body, message, next); e != nil {
				return e
			}
			if content.ArtistImageUrl != "" {
				if _, e := tx.Exec(ctx, `UPDATE artists SET genius_image_url=$2 WHERE id=(SELECT artist_id FROM track_artists
     WHERE track_id=$1 AND role='main' ORDER BY position LIMIT 1) AND image_path IS NULL AND genius_image_url IS NULL`, id, content.ArtistImageUrl); e != nil {
					return e
				}
			}
			if albumImage != "" {
				if _, e := tx.Exec(ctx, `UPDATE albums SET genius_cover_url=$2 WHERE id=(SELECT album_id FROM tracks WHERE id=$1)
     AND cover_path IS NULL AND genius_cover_url IS NULL`, id, albumImage); e != nil {
					return e
				}
			}
			if err != nil {
				pause := false
				cooldown := time.Now()
				var he *HTTPError
				if errors.As(err, &he) {
					pause = he.StatusCode == 401 || he.StatusCode == 403
					if he.StatusCode == 429 {
						cooldown = time.Now().Add(retryDelay(he.RetryAfter))
					}
				}
				if _, e := tx.Exec(ctx, `UPDATE genius_state SET last_error=$1, paused=paused OR $2,
     cooldown_until=GREATEST(cooldown_until,$3)`, message, pause, cooldown); e != nil {
					return e
				}
			}
			return tx.Commit(ctx)
		}()
		_ = tx.Rollback(ctx)
		if saveErr != nil {
			return saveErr
		}
	}
	return ctx.Err()
}

func emptyContent() gen.GeniusContent {
	return gen.GeniusContent{Status: "pending", Annotations: []gen.GeniusAnnotation{}}
}

func clip(s string, n int) string {
	s = strings.TrimSpace(s)
	if r := []rune(s); len(r) > n {
		return strings.TrimSpace(string(r[:n]))
	}
	return s
}

func (s *Service) enrich(ctx context.Context, id uuid.UUID) (gen.GeniusContent, string, error) {
	content := emptyContent()
	var title, artist, album string
	var old []byte
	err := s.pool.QueryRow(ctx, `SELECT t.title, COALESCE((SELECT a.name FROM track_artists ta JOIN artists a ON a.id=ta.artist_id
  WHERE ta.track_id=t.id AND ta.role='main' ORDER BY ta.position LIMIT 1),''), al.title, g.content
  FROM tracks t JOIN albums al ON al.id=t.album_id JOIN genius_tracks g ON g.track_id=t.id WHERE t.id=$1`, id).Scan(&title, &artist, &album, &old)
	if err != nil {
		return content, "", err
	}
	if err = json.Unmarshal(old, &content); err != nil {
		return content, "", err
	}
	if artist == "" || title == "" {
		return content, "", nil
	}
	songs, err := s.client.Search(ctx, artist+" "+title)
	if err != nil {
		return content, "", err
	}
	song, ok := matchSong(songs, artist, title)
	if !ok {
		return content, "", nil
	}
	content.SourceUrl = sourceURL(song.URL)
	content.SongImageUrl, content.ArtistImageUrl, _ = songArtwork(song)
	details, err := s.client.Song(ctx, song.ID)
	if err != nil {
		return content, "", err
	}
	if u := sourceURL(details.URL); u != "" {
		content.SourceUrl = u
	}
	songImage, artistImage, albumImage := songArtwork(details)
	if songImage != "" {
		content.SongImageUrl = songImage
	}
	if artistImage != "" {
		content.ArtistImageUrl = artistImage
	}
	if normalize(details.Album.Name) != normalize(album) {
		albumImage = ""
	}
	if text := clip(details.Description.Plain, 8000); text != "" {
		content.Description = text
	}
	if content.ArtistImageUrl == "" && details.Artist.ID > 0 {
		geniusArtist, artistErr := s.client.Artist(ctx, details.Artist.ID)
		if artistErr != nil {
			var he *HTTPError
			if !errors.As(artistErr, &he) || he.StatusCode != http.StatusNotFound {
				return content, albumImage, artistErr
			}
		} else {
			content.ArtistImageUrl = imageURL(geniusArtist.ImageURL)
		}
	}
	pageURL := song.URL
	if candidate := sourceURL(details.URL); strings.HasSuffix(urlPath(candidate), "-lyrics") {
		pageURL = candidate
	}
	annotations, err := s.client.Annotations(ctx, song.ID)
	if err != nil {
		return content, albumImage, err
	}
	content.Annotations = annotations
	lyrics, err := s.client.FetchLyrics(ctx, pageURL)
	if err != nil {
		return content, albumImage, err
	}
	content.Lyrics = lyrics.Text
	return content, albumImage, nil
}

func errorCode(err error) string {
	var he *HTTPError
	if errors.As(err, &he) {
		return "http_" + strconv.Itoa(he.StatusCode)
	}
	if errors.Is(err, ErrLyricsUnavailable) {
		return "lyrics_unavailable"
	}
	return "request_failed"
}

func retryDelay(raw string) time.Duration {
	if seconds, err := strconv.Atoi(raw); err == nil {
		return max(time.Minute, min(time.Duration(seconds)*time.Second, 24*time.Hour))
	}
	if date, err := http.ParseTime(raw); err == nil {
		return max(time.Minute, min(time.Until(date), 24*time.Hour))
	}
	return 15 * time.Minute
}

func (s *Service) Stats(ctx context.Context) (gen.GeniusStats, error) {
	out := gen.GeniusStats{Configured: s.configured}
	err := s.pool.QueryRow(ctx, `SELECT paused,last_error,cooldown_until FROM genius_state`).Scan(&out.Paused, &out.LastError, &out.CooldownUntil)
	if err != nil {
		return out, err
	}
	err = s.pool.QueryRow(ctx, `SELECT count(*),
  count(*) FILTER(WHERE g.track_id IS NULL OR g.status IN ('pending','processing')),
  count(*) FILTER(WHERE COALESCE(g.content->>'source_url','')<>''),
  count(*) FILTER(WHERE COALESCE(g.content->>'lyrics','')<>''),
  count(*) FILTER(WHERE jsonb_array_length(COALESCE(g.content->'annotations','[]'))>0),
  count(*) FILTER(WHERE COALESCE(g.content->>'song_image_url','')<>'' OR COALESCE(g.content->>'artist_image_url','')<>''),
  count(*) FILTER(WHERE g.status='unmatched'),
  count(*) FILTER(WHERE g.status IN ('error','partial')),
  COALESCE(bool_or(g.status='processing' AND g.locked_until>now()),false)
  FROM tracks t LEFT JOIN genius_tracks g ON g.track_id=t.id`).Scan(&out.Total, &out.Pending, &out.Matched, &out.Lyrics, &out.Annotations, &out.Images, &out.Unmatched, &out.Errors, &out.Running)
	return out, err
}