feat(artwork): implement artwork_queue repository

persistence/artwork_queue_repository.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"time"
 
+	. "github.com/Masterminds/squirrel"
 	"github.com/navidrome/navidrome/model"
 	"github.com/pocketbase/dbx"
 )
@@ -21,23 +22,59 @@ func NewArtworkQueueRepository(ctx context.Context, db dbx.Builder) model.Artwor
 }
 
 func (r *artworkQueueRepository) Enqueue(items ...model.ArtworkQueueItem) error {
+	now := time.Now()
+	for _, it := range items {
+		if it.ImageType == "" {
+			it.ImageType = model.ImageTypePrimary
+		}
+		ins := Insert(r.tableName).SetMap(map[string]any{
+			"item_kind": it.ItemKind, "item_id": it.ItemID, "image_type": it.ImageType,
+			"priority": it.Priority, "attempts": 0, "retry_at": now, "enqueued_at": now,
+		}).Suffix(`ON CONFLICT (item_kind, item_id, image_type) DO UPDATE SET
+			priority = MAX(priority, excluded.priority), retry_at = excluded.retry_at`)
+		if _, err := r.executeSQL(ins); err != nil {
+			return err
+		}
+	}
 	return nil
 }
 
 func (r *artworkQueueRepository) DequeueBatch(n int) ([]model.ArtworkQueueItem, error) {
-	return nil, nil
+	sel := Select("*").From(r.tableName).
+		Where(LtOrEq{"retry_at": time.Now()}).
+		OrderBy("priority DESC", "enqueued_at ASC").
+		Limit(uint64(n))
+	var res []model.ArtworkQueueItem
+	err := r.queryAll(sel, &res)
+	return res, err
 }
 
 func (r *artworkQueueRepository) MarkFailed(kind, id, imageType string, retryAt time.Time) error {
-	return nil
+	upd := Update(r.tableName).
+		Set("attempts", Expr("attempts + 1")).
+		Set("retry_at", retryAt).
+		Where(Eq{"item_kind": kind, "item_id": id, "image_type": imageType})
+	c, err := r.executeSQL(upd)
+	if err == nil && c == 0 {
+		return model.ErrNotFound
+	}
+	return err
 }
 
 func (r *artworkQueueRepository) Delete(kind, id, imageType string) error {
-	return nil
+	return r.delete(Eq{"item_kind": kind, "item_id": id, "image_type": imageType})
 }
 
 func (r *artworkQueueRepository) Count() (int64, error) {
-	return 0, nil
+	sel := Select("count(*)").From(r.tableName)
+	var counts []int64
+	if err := r.queryAllSlice(sel, &counts); err != nil {
+		return 0, err
+	}
+	if len(counts) == 0 {
+		return 0, nil
+	}
+	return counts[0], nil
 }
 
 var _ model.ArtworkQueueRepository = (*artworkQueueRepository)(nil)

persistence/artwork_queue_repository_test.go

@@ -0,0 +1,65 @@
+package persistence
+
+import (
+	"context"
+	"time"
+
+	"github.com/navidrome/navidrome/model"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("ArtworkQueueRepository", func() {
+	var repo model.ArtworkQueueRepository
+
+	item := func(kind, id string, prio int) model.ArtworkQueueItem {
+		return model.ArtworkQueueItem{ItemKind: kind, ItemID: id,
+			ImageType: model.ImageTypePrimary, Priority: prio}
+	}
+
+	BeforeEach(func() {
+		clearArtworkTables()
+		repo = NewArtworkQueueRepository(context.Background(), GetDBXBuilder())
+	})
+
+	It("enqueues and dequeues by priority then FIFO", func() {
+		Expect(repo.Enqueue(item("al", "low", model.ArtworkPriorityBackfill))).To(Succeed())
+		Expect(repo.Enqueue(item("ar", "high", model.ArtworkPriorityBump))).To(Succeed())
+
+		got, err := repo.DequeueBatch(10)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(got).To(HaveLen(2))
+		Expect(got[0].ItemID).To(Equal("high"))
+	})
+
+	It("keeps the higher priority on duplicate enqueue", func() {
+		Expect(repo.Enqueue(item("al", "a1", model.ArtworkPriorityBump))).To(Succeed())
+		Expect(repo.Enqueue(item("al", "a1", model.ArtworkPriorityBackfill))).To(Succeed())
+		got, _ := repo.DequeueBatch(10)
+		Expect(got).To(HaveLen(1))
+		Expect(got[0].Priority).To(Equal(model.ArtworkPriorityBump))
+	})
+
+	It("hides failed items until retry_at", func() {
+		Expect(repo.Enqueue(item("al", "f1", model.ArtworkPriorityScan))).To(Succeed())
+		Expect(repo.MarkFailed("al", "f1", model.ImageTypePrimary, time.Now().Add(time.Hour))).To(Succeed())
+
+		got, err := repo.DequeueBatch(10)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(got).To(BeEmpty())
+
+		Expect(repo.MarkFailed("al", "f1", model.ImageTypePrimary, time.Now().Add(-time.Minute))).To(Succeed())
+		got, _ = repo.DequeueBatch(10)
+		Expect(got).To(HaveLen(1))
+		Expect(got[0].Attempts).To(Equal(2))
+	})
+
+	It("deletes on completion and counts", func() {
+		Expect(repo.Enqueue(item("al", "c1", 0))).To(Succeed())
+		n, _ := repo.Count()
+		Expect(n).To(Equal(int64(1)))
+		Expect(repo.Delete("al", "c1", model.ImageTypePrimary)).To(Succeed())
+		n, _ = repo.Count()
+		Expect(n).To(BeZero())
+	})
+})