diff --git a/persistence/artwork_queue_repository.go b/persistence/artwork_queue_repository.go index 4b166aa57..fdf88db24 100644 --- a/persistence/artwork_queue_repository.go +++ b/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) diff --git a/persistence/artwork_queue_repository_test.go b/persistence/artwork_queue_repository_test.go new file mode 100644 index 000000000..73665b07c --- /dev/null +++ b/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()) + }) +})