navidrome/persistence/artwork_queue_repository.go

package persistence

import (
	"cmp"
	"context"
	"fmt"
	"slices"
	"strings"
	"time"

	. "github.com/Masterminds/squirrel"
	"github.com/navidrome/navidrome/model"
	"github.com/navidrome/navidrome/utils/slice"
	"github.com/pocketbase/dbx"
)

// Keeps each multi-row insert under SQLite's bind-variable limit (at most 7 vars per row).
const enqueueChunkSize = 100

// Every insert writes these, in this order; the INSERT..SELECT forms must project them to match.
// DequeueBatch also selects exactly these, to leave the drain's rows free of the trace it never reads.
var enqueueColumns = []string{"item_kind", "item_id", "image_type", "priority", "attempts", "retry_at", "enqueued_at"}

type artworkQueueRepository struct {
	sqlRepository
}

func NewArtworkQueueRepository(ctx context.Context, db dbx.Builder) model.ArtworkQueueRepository {
	r := &artworkQueueRepository{}
	r.ctx = ctx
	r.db = db
	r.tableName = "artwork_queue"
	return r
}

func (r *artworkQueueRepository) Get(kind model.Kind, id, imageType string) (*model.ArtworkQueueItem, error) {
	var res model.ArtworkQueueItem
	err := r.queryOne(Select("*").From(r.tableName).
		Where(Eq{"item_kind": kind.Prefix(), "item_id": id, "image_type": imageType}), &res)
	if err != nil {
		return nil, err
	}
	return &res, nil
}

// Enqueue starts a fresh lifecycle: it resets enqueued_at (so a fresh request does not inherit an old
// row's spent retry budget) and clears trace (so explain does not show a prior failure at attempts 0).
func (r *artworkQueueRepository) Enqueue(items ...model.ArtworkQueueItem) error {
	return r.enqueue(`ON CONFLICT (item_kind, item_id, image_type) DO UPDATE SET
		priority = MAX(priority, excluded.priority), retry_at = excluded.retry_at,
		attempts = 0, enqueued_at = excluded.enqueued_at, trace = '[]'`, items)
}

func (r *artworkQueueRepository) EnqueuePreservingBackoff(items ...model.ArtworkQueueItem) error {
	return r.enqueue(`ON CONFLICT (item_kind, item_id, image_type) DO UPDATE SET
		priority = MAX(priority, excluded.priority)`, items)
}

func (r *artworkQueueRepository) EnqueueStaleAbsent(kind model.Kind, attemptedBefore time.Time, limit int) (int64, error) {
	now := time.Now()
	return r.insertIfNotQueued("", `SELECT item_kind, item_id, image_type, ?, 0, ?, ?
		FROM `+itemArtworkTable+` WHERE item_kind = ? AND hash = '' AND attempted_at < ?
		ORDER BY attempted_at LIMIT ?`,
		model.ArtworkPriorityRecheck, now, now, kind.Prefix(), attemptedBefore, limit)
}

func (r *artworkQueueRepository) EnqueueAllMissing(kind model.Kind, priority int) (int64, error) {
	entityTable, ok := artworkOwnerTables[kind]
	if !ok {
		return 0, fmt.Errorf("artwork queue: no entity table for kind %q", kind.Prefix())
	}
	now := time.Now()
	return r.insertIfNotQueued("", `SELECT ?, id, ?, ?, 0, ?, ?
		FROM `+entityTable+`
		WHERE id NOT IN (SELECT item_id FROM `+itemArtworkTable+` WHERE item_kind = ?)`,
		kind.Prefix(), model.ImageTypePrimary, priority, now, now, kind.Prefix())
}

func (r *artworkQueueRepository) EnqueueIfMissing(items ...model.ArtworkQueueItem) error {
	now := time.Now()
	for chunk := range slices.Chunk(items, enqueueChunkSize) {
		rows := make([]string, 0, len(chunk))
		args := make([]any, 0, len(chunk)*4+2)
		for _, it := range chunk {
			rows = append(rows, "(?,?,?,?)")
			args = append(args, it.ItemKind, it.ItemID, cmp.Or(it.ImageType, model.ImageTypePrimary), it.Priority)
		}
		args = append(args, now, now)
		_, err := r.insertIfNotQueued(
			`WITH new_items(item_kind, item_id, image_type, priority) AS (VALUES `+strings.Join(rows, ",")+`) `,
			`SELECT n.item_kind, n.item_id, n.image_type, n.priority, 0, ?, ?
			FROM new_items n
			WHERE NOT EXISTS (
				SELECT 1 FROM `+itemArtworkTable+` ia
				WHERE ia.item_kind = n.item_kind AND ia.item_id = n.item_id AND ia.image_type = n.image_type)`,
			args...)
		if err != nil {
			return err
		}
	}
	return nil
}

// DO NOTHING is deliberate: a recheck must not bump the priority or retry_at of an already-queued item.
const skipIfQueued = ` ON CONFLICT (item_kind, item_id, image_type) DO NOTHING`

// insertIfNotQueued inserts the rows selected by the given SQL, optionally prefixed by a CTE.
func (r *artworkQueueRepository) insertIfNotQueued(with, sql string, args ...any) (int64, error) {
	return r.executeSQL(Expr(with+`INSERT INTO `+r.tableName+
		` (`+strings.Join(enqueueColumns, ", ")+`) `+sql+skipIfQueued, args...))
}

// artworkSourceFilter selects item_artwork rows of a kind; no sources means every source, "" the absent state.
func artworkSourceFilter(kind model.Kind, sources []string) Sqlizer {
	f := And{Eq{"item_kind": kind.Prefix()}}
	if len(sources) > 0 {
		f = append(f, Eq{"source": sources})
	}
	return f
}

func (r *artworkQueueRepository) CountBySource(kind model.Kind, sources []string) (int64, error) {
	var res struct{ Count int64 }
	err := r.queryOne(Select("count(*) as count").From(itemArtworkTable).
		Where(artworkSourceFilter(kind, sources)), &res)
	return res.Count, err
}

func (r *artworkQueueRepository) SourcesInUse(kind model.Kind) ([]string, error) {
	var res []struct{ Source string }
	err := r.queryAll(Select("distinct source").From(itemArtworkTable).
		Where(Eq{"item_kind": kind.Prefix()}), &res)
	if err != nil {
		return nil, err
	}
	return slice.Map(res, func(s struct{ Source string }) string { return s.Source }), nil
}

// EnqueueBySource deliberately leaves item_artwork alone: clearing state in bulk would blank the
// library's artwork until every item is resolved again.
func (r *artworkQueueRepository) EnqueueBySource(kind model.Kind, sources []string, priority int) (int64, error) {
	now := time.Now()
	sel := Select("item_kind", "item_id", "image_type").
		Column(Expr("?", priority)).Column("0").Column(Expr("?", now)).Column(Expr("?", now)).
		From(itemArtworkTable).Where(artworkSourceFilter(kind, sources))
	return r.executeSQL(Insert(r.tableName).Columns(enqueueColumns...).Select(sel).Suffix(skipIfQueued))
}

func (r *artworkQueueRepository) enqueue(conflict string, items []model.ArtworkQueueItem) error {
	now := time.Now()
	for chunk := range slices.Chunk(items, enqueueChunkSize) {
		ins := Insert(r.tableName).Columns(enqueueColumns...)
		for _, it := range chunk {
			ins = ins.Values(it.ItemKind, it.ItemID, cmp.Or(it.ImageType, model.ImageTypePrimary), it.Priority, 0, now, now)
		}
		ins = ins.Suffix(conflict)
		if _, err := r.executeSQL(ins); err != nil {
			return err
		}
	}
	return nil
}

func (r *artworkQueueRepository) DequeueBatch(n int, kinds ...string) ([]model.ArtworkQueueItem, error) {
	sel := Select(enqueueColumns...).From(r.tableName).
		Where(LtOrEq{"retry_at": time.Now()}).
		OrderBy("priority DESC", "enqueued_at ASC").
		Limit(uint64(n))
	if len(kinds) > 0 {
		sel = sel.Where(Eq{"item_kind": kinds})
	}
	var res []model.ArtworkQueueItem
	err := r.queryAll(sel, &res)
	return res, err
}

func (r *artworkQueueRepository) MarkFailedIfUnchanged(kind, id, imageType string, seenRetryAt, retryAt time.Time, trace string) error {
	upd := Update(r.tableName).
		Set("attempts", Expr("attempts + 1")).
		Set("retry_at", retryAt).
		Set("trace", trace).
		Where(Eq{"item_kind": kind, "item_id": id, "image_type": imageType, "retry_at": seenRetryAt})
	_, err := r.executeSQL(upd)
	return err
}

func (r *artworkQueueRepository) DeleteIfUnchanged(kind, id, imageType string, retryAt time.Time) error {
	return r.delete(Eq{"item_kind": kind, "item_id": id, "image_type": imageType, "retry_at": retryAt})
}

func (r *artworkQueueRepository) PurgeDangling() (int64, error) {
	return purgeDangling(r.sqlRepository)
}

// artworkQueueFilter returns no conditions for an empty filter, so an unfiltered DELETE keeps
// SQLite's truncate path. It ignores retry_at: a backing-off row is pending work too.
func artworkQueueFilter(kinds []model.Kind, priorities []int) And {
	var f And
	if len(kinds) > 0 {
		f = append(f, Eq{"item_kind": model.KindPrefixes(kinds)})
	}
	if len(priorities) > 0 {
		f = append(f, Eq{"priority": priorities})
	}
	return f
}

// CountQueued shares its filter with PurgeQueued, so a preview cannot count rows the delete misses.
func (r *artworkQueueRepository) CountQueued(kinds []model.Kind, priorities []int) ([]model.ArtworkQueueStat, error) {
	sel := Select("item_kind", "priority", "count(*) as count").From(r.tableName).
		GroupBy("item_kind", "priority").OrderBy("item_kind", "priority desc")
	if f := artworkQueueFilter(kinds, priorities); len(f) > 0 {
		sel = sel.Where(f)
	}
	var res []model.ArtworkQueueStat
	err := r.queryAll(sel, &res)
	return res, err
}

func (r *artworkQueueRepository) PurgeQueued(kinds []model.Kind, priorities []int) (int64, error) {
	del := Delete(r.tableName)
	if f := artworkQueueFilter(kinds, priorities); len(f) > 0 {
		del = del.Where(f)
	}
	return r.executeSQL(del)
}

func (r *artworkQueueRepository) Count() (int64, error) {
	var res struct{ Count int64 }
	err := r.queryOne(Select("count(*) as count").From(r.tableName), &res)
	return res.Count, err
}

// CountAbsent matches EnqueueStaleAbsent on hash, so the stale count is the pool a recheck drains from.
func (r *artworkQueueRepository) CountAbsent(kind model.Kind, attemptedBefore time.Time) (model.ArtworkAbsentStat, error) {
	var res model.ArtworkAbsentStat
	err := r.queryOne(Select("count(*) as total").
		Column(Expr("coalesce(sum(attempted_at < ?), 0) as stale", attemptedBefore)).
		From(itemArtworkTable).Where(Eq{"item_kind": kind.Prefix(), "hash": ""}), &res)
	return res, err
}

var _ model.ArtworkQueueRepository = (*artworkQueueRepository)(nil)