package persistence import ( "context" "time" "github.com/Masterminds/squirrel" "github.com/navidrome/navidrome/model" "github.com/navidrome/navidrome/utils/slice" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" "github.com/pocketbase/dbx" ) 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} } // Writes the post-failure state directly: the production path, MarkFailedIfUnchanged, needs // a retry_at only a dequeue can hand it. backOff := func(kind, id string, retryAt time.Time) { GinkgoHelper() r := repo.(*artworkQueueRepository) _, err := r.executeSQL(squirrel.Update(r.tableName). Set("attempts", squirrel.Expr("attempts + 1")). Set("retry_at", retryAt). Where(squirrel.Eq{"item_kind": kind, "item_id": id, "image_type": model.ImageTypePrimary})) Expect(err).ToNot(HaveOccurred()) } remove := func(kind, id string) { GinkgoHelper() r := repo.(*artworkQueueRepository) Expect(r.delete(squirrel.Eq{"item_kind": kind, "item_id": id, "image_type": model.ImageTypePrimary})).To(Succeed()) } BeforeEach(func() { clearArtworkTables() DeferCleanup(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("EnqueuePreservingBackoff raises priority without resetting a backing-off row's retry_at", func() { Expect(repo.Enqueue(item("al", "b1", model.ArtworkPriorityScan))).To(Succeed()) backOff("al", "b1", time.Now().Add(time.Hour)) Expect(repo.DequeueBatch(10)).To(BeEmpty()) Expect(repo.EnqueuePreservingBackoff(item("al", "b1", model.ArtworkPriorityBump))).To(Succeed()) Expect(repo.DequeueBatch(10)).To(BeEmpty(), "bump must not reset retry_at") // Enqueue (scan/manual), by contrast, resets retry_at and makes it eligible now. Expect(repo.Enqueue(item("al", "b1", model.ArtworkPriorityScan))).To(Succeed()) got, _ := repo.DequeueBatch(10) Expect(got).To(HaveLen(1)) Expect(got[0].Priority).To(Equal(model.ArtworkPriorityBump), "bump's higher priority is preserved") }) It("EnqueuePreservingBackoff inserts a brand-new row eligible immediately", func() { Expect(repo.EnqueuePreservingBackoff(item("ar", "n1", model.ArtworkPriorityBump))).To(Succeed()) got, _ := repo.DequeueBatch(10) Expect(got).To(HaveLen(1)) Expect(got[0].ItemID).To(Equal("n1")) }) It("hides failed items until retry_at", func() { Expect(repo.Enqueue(item("al", "f1", model.ArtworkPriorityScan))).To(Succeed()) backOff("al", "f1", time.Now().Add(time.Hour)) got, err := repo.DequeueBatch(10) Expect(err).ToNot(HaveOccurred()) Expect(got).To(BeEmpty()) backOff("al", "f1", time.Now().Add(-time.Minute)) got, _ = repo.DequeueBatch(10) Expect(got).To(HaveLen(1)) Expect(got[0].Attempts).To(Equal(2)) }) It("MarkFailedIfUnchanged applies backoff only while retry_at is unchanged", func() { Expect(repo.Enqueue(item("al", "m1", model.ArtworkPriorityScan))).To(Succeed()) // Anchor retry_at in the past so it can never collide with the re-enqueue's now. backOff("al", "m1", time.Now().Add(-time.Hour)) got, err := repo.DequeueBatch(10) Expect(err).ToNot(HaveOccurred()) Expect(got).To(HaveLen(1)) original := got[0].RetryAt // A concurrent scan re-enqueues, resetting retry_at to now. Expect(repo.Enqueue(item("al", "m1", model.ArtworkPriorityScan))).To(Succeed()) future := time.Now().Add(48 * time.Hour) Expect(repo.MarkFailedIfUnchanged("al", "m1", model.ImageTypePrimary, original, future)).To(Succeed()) got, _ = repo.DequeueBatch(10) Expect(got).To(HaveLen(1), "the fresh re-enqueue stays immediately eligible") Expect(got[0].Attempts).To(BeZero(), "re-enqueue clears attempts, and the stale failure must not bump them") current := got[0].RetryAt Expect(repo.MarkFailedIfUnchanged("al", "m1", model.ImageTypePrimary, current, future)).To(Succeed()) got, _ = repo.DequeueBatch(10) Expect(got).To(BeEmpty(), "backed-off row is hidden until the future retry_at") all, _ := repo.Count() Expect(all).To(Equal(int64(1))) }) It("Enqueue restarts the retry budget an existing row had spent", func() { Expect(repo.Enqueue(item("al", "e1", model.ArtworkPriorityScan))).To(Succeed()) backOff("al", "e1", time.Now().Add(-time.Hour)) stale := time.Now().Add(-48 * time.Hour) _, err := GetDBXBuilder().NewQuery("UPDATE artwork_queue SET enqueued_at = {:t} WHERE item_id = 'e1'"). Bind(dbx.Params{"t": stale}).Execute() Expect(err).ToNot(HaveOccurred()) Expect(repo.Enqueue(item("al", "e1", model.ArtworkPriorityScan))).To(Succeed()) got, err := repo.DequeueBatch(10) Expect(err).ToNot(HaveOccurred()) Expect(got).To(HaveLen(1)) Expect(got[0].EnqueuedAt).To(BeTemporally("~", time.Now(), time.Minute), "a re-request must not inherit a spent 12h window and give up on its first attempt") Expect(got[0].Attempts).To(BeZero()) }) 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))) remove("al", "c1") n, _ = repo.Count() Expect(n).To(BeZero()) }) It("DeleteIfUnchanged deletes only while retry_at is unchanged", func() { Expect(repo.Enqueue(item("al", "d1", model.ArtworkPriorityScan))).To(Succeed()) // Anchor retry_at in the past so it can never collide with the re-enqueue's now. backOff("al", "d1", time.Now().Add(-time.Hour)) got, err := repo.DequeueBatch(10) Expect(err).ToNot(HaveOccurred()) Expect(got).To(HaveLen(1)) original := got[0].RetryAt // A concurrent scan re-enqueues, resetting retry_at to now. Expect(repo.Enqueue(item("al", "d1", model.ArtworkPriorityScan))).To(Succeed()) // Deleting with the stale retry_at is a no-op: the re-enqueued row survives. Expect(repo.DeleteIfUnchanged("al", "d1", model.ImageTypePrimary, original)).To(Succeed()) n, _ := repo.Count() Expect(n).To(Equal(int64(1))) got, _ = repo.DequeueBatch(10) Expect(got).To(HaveLen(1)) Expect(repo.DeleteIfUnchanged("al", "d1", model.ImageTypePrimary, got[0].RetryAt)).To(Succeed()) n, _ = repo.Count() Expect(n).To(BeZero()) }) It("purges queue rows whose entity no longer exists, per kind", func() { Expect(repo.Enqueue( item("al", albumSgtPeppers.ID, model.ArtworkPriorityScan), item("al", "no-such-album", model.ArtworkPriorityScan), item("ar", artistKraftwerk.ID, model.ArtworkPriorityScan), item("ar", "no-such-artist", model.ArtworkPriorityScan), item("pl", plsBest.ID, model.ArtworkPriorityScan), item("pl", "no-such-playlist", model.ArtworkPriorityScan), item("ra", radioWithHomePage.ID, model.ArtworkPriorityScan), item("ra", "no-such-radio", model.ArtworkPriorityScan), item("mf", songDayInALife.ID, model.ArtworkPriorityScan), item("mf", "no-such-mediafile", model.ArtworkPriorityScan), )).To(Succeed()) purged, err := repo.PurgeDangling() Expect(err).ToNot(HaveOccurred()) Expect(purged).To(Equal(int64(5))) got, _ := repo.DequeueBatch(100) ids := slice.Map(got, func(it model.ArtworkQueueItem) string { return it.ItemID }) Expect(ids).To(ConsistOf(albumSgtPeppers.ID, artistKraftwerk.ID, plsBest.ID, radioWithHomePage.ID, songDayInALife.ID)) }) It("enqueues stale absent states for recheck", func() { awRepo := NewArtworkRepository(context.Background(), GetDBXBuilder()) old := time.Now().Add(-48 * time.Hour) Expect(awRepo.PutItemArtwork(&model.ItemArtwork{ItemKind: "ar", ItemID: "stale1", ImageType: model.ImageTypePrimary, Hash: "", AttemptedAt: old})).To(Succeed()) Expect(awRepo.PutItemArtwork(&model.ItemArtwork{ItemKind: "ar", ItemID: "fresh1", ImageType: model.ImageTypePrimary, Hash: "", AttemptedAt: time.Now()})).To(Succeed()) Expect(awRepo.PutItemArtwork(&model.ItemArtwork{ItemKind: "ar", ItemID: "found1", ImageType: model.ImageTypePrimary, Hash: "hX", AttemptedAt: old})).To(Succeed()) n, err := repo.EnqueueStaleAbsent(model.KindArtistArtwork, time.Now().Add(-24*time.Hour)) Expect(err).ToNot(HaveOccurred()) Expect(n).To(Equal(int64(1))) items, err := repo.DequeueBatch(10) Expect(err).ToNot(HaveOccurred()) Expect(items).To(HaveLen(1)) Expect(items[0].ItemID).To(Equal("stale1")) Expect(items[0].Priority).To(Equal(model.ArtworkPriorityRecheck)) }) It("enqueues entities that have no item_artwork row at all", func() { awRepo := NewArtworkRepository(context.Background(), GetDBXBuilder()) Expect(awRepo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: albumSgtPeppers.ID, ImageType: model.ImageTypePrimary, Hash: "hX", AttemptedAt: time.Now()})).To(Succeed()) Expect(awRepo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: albumAbbeyRoad.ID, ImageType: model.ImageTypePrimary, Hash: "", AttemptedAt: time.Now()})).To(Succeed()) n, err := repo.EnqueueAllMissing(model.KindAlbumArtwork, model.ArtworkPriorityRecheck) Expect(err).ToNot(HaveOccurred()) Expect(n).To(BeNumerically(">=", 1)) got, err := repo.DequeueBatch(1000) Expect(err).ToNot(HaveOccurred()) ids := make([]string, 0, len(got)) for _, it := range got { Expect(it.ItemKind).To(Equal("al")) Expect(it.Priority).To(Equal(model.ArtworkPriorityRecheck)) ids = append(ids, it.ItemID) } Expect(ids).To(ContainElement(albumRadioactivity.ID), "an album with no row must be enqueued") Expect(ids).ToNot(ContainElement(albumSgtPeppers.ID), "a resolved album must not be re-enqueued") Expect(ids).ToNot(ContainElement(albumAbbeyRoad.ID), "an absent-state album must not be enqueued as missing") }) It("EnqueueIfMissing skips items that already have an item_artwork row", func() { awRepo := NewArtworkRepository(context.Background(), GetDBXBuilder()) Expect(awRepo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "resolved", ImageType: model.ImageTypePrimary, Hash: "hX", AttemptedAt: time.Now()})).To(Succeed()) Expect(awRepo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "absent", ImageType: model.ImageTypePrimary, Hash: "", AttemptedAt: time.Now()})).To(Succeed()) Expect(repo.EnqueueIfMissing( item("al", "resolved", model.ArtworkPriorityScan), item("al", "absent", model.ArtworkPriorityScan), item("al", "brandnew", model.ArtworkPriorityScan), )).To(Succeed()) got, err := repo.DequeueBatch(100) Expect(err).ToNot(HaveOccurred()) ids := slice.Map(got, func(it model.ArtworkQueueItem) string { return it.ItemID }) Expect(ids).To(ConsistOf("brandnew"), "only an item with no state row may be enqueued") }) It("EnqueueIfMissing leaves an already-queued row untouched", func() { Expect(repo.Enqueue(item("al", "queued", model.ArtworkPriorityBump))).To(Succeed()) Expect(repo.EnqueueIfMissing(item("al", "queued", model.ArtworkPriorityScan))).To(Succeed()) got, _ := repo.DequeueBatch(100) Expect(got).To(HaveLen(1)) Expect(got[0].Priority).To(Equal(model.ArtworkPriorityBump), "the existing priority must survive") }) It("does not disturb an already-queued entity when enqueueing missing rows", func() { Expect(repo.Enqueue(item("al", albumRadioactivity.ID, model.ArtworkPriorityBump))).To(Succeed()) _, err := repo.EnqueueAllMissing(model.KindAlbumArtwork, model.ArtworkPriorityRecheck) Expect(err).ToNot(HaveOccurred()) got, _ := repo.DequeueBatch(1000) var count int for _, it := range got { if it.ItemID == albumRadioactivity.ID { count++ Expect(it.Priority).To(Equal(model.ArtworkPriorityBump), "existing bump priority must survive") } } Expect(count).To(Equal(1), "the already-queued row must not be duplicated") }) })