diff --git a/core/artwork/prune.go b/core/artwork/prune.go index 11d9ce34f..ebbb0bbc4 100644 --- a/core/artwork/prune.go +++ b/core/artwork/prune.go @@ -2,7 +2,6 @@ package artwork import ( "context" - "errors" "time" "github.com/navidrome/navidrome/log" @@ -34,9 +33,17 @@ func Prune(ctx context.Context, ds model.DataStore, store *ImageStore) error { log.Info(ctx, "Prune: removed orphan artwork", "count", len(orphans)) } + hashes, err := repo.GetAllHashes() + if err != nil { + return err + } + known := make(map[string]struct{}, len(hashes)) + for _, h := range hashes { + known[h] = struct{}{} + } removed, err := store.Sweep(func(hash string) bool { - _, err := repo.GetImage(hash) - return !errors.Is(err, model.ErrNotFound) + _, ok := known[hash] + return ok }) if err != nil { return err diff --git a/core/artwork/prune_test.go b/core/artwork/prune_test.go index 68211879d..2c8751674 100644 --- a/core/artwork/prune_test.go +++ b/core/artwork/prune_test.go @@ -17,7 +17,7 @@ type flakyGetArtworkRepo struct { *tests.MockArtworkRepo } -func (f *flakyGetArtworkRepo) GetImage(string) (*model.Artwork, error) { +func (f *flakyGetArtworkRepo) GetAllHashes() ([]string, error) { return nil, errors.New("db locked") } @@ -74,7 +74,7 @@ var _ = Describe("Prune", func() { h, _ := HashImage(bytes.NewReader(data)) Expect(store.Write(h, "image/jpeg", bytes.NewReader(data))).To(Succeed()) - Expect(Prune(context.Background(), ds, store)).To(Succeed()) + Expect(Prune(context.Background(), ds, store)).ToNot(Succeed()) rc, err := store.Open(h, "image/jpeg") Expect(err).ToNot(HaveOccurred()) diff --git a/model/artwork.go b/model/artwork.go index 6507c24d7..986a5c65d 100644 --- a/model/artwork.go +++ b/model/artwork.go @@ -35,9 +35,11 @@ type ItemArtworkInfo struct { ItemID string Hash string BlurHash string - Absent bool } +// Absent reports a known-absent artwork state (resolved, no image). +func (i ItemArtworkInfo) Absent() bool { return i.Hash == "" } + type ArtworkQueueItem struct { ItemKind string `structs:"item_kind"` ItemID string `structs:"item_id"` @@ -70,8 +72,8 @@ type ArtworkRepository interface { DeleteForItem(kind, id string) error // GetInfoForItems hydrates a page: one batched query, item_artwork joined to artwork. GetInfoForItems(kind string, ids []string) (map[string]ItemArtworkInfo, error) - // EnqueueStaleAbsent inserts queue rows (priority Recheck) for absent states older than cutoff. - EnqueueStaleAbsent(kind string, attemptedBefore time.Time) (int64, error) + // GetAllHashes returns every stored artwork hash, for sweep membership checks. + GetAllHashes() ([]string, error) } type ArtworkQueueRepository interface { @@ -83,4 +85,6 @@ type ArtworkQueueRepository interface { MarkFailed(kind, id, imageType string, retryAt time.Time) error Delete(kind, id, imageType string) error Count() (int64, error) + // EnqueueStaleAbsent inserts queue rows (priority Recheck) for absent states older than cutoff. + EnqueueStaleAbsent(kind string, attemptedBefore time.Time) (int64, error) } diff --git a/persistence/artwork_queue_repository.go b/persistence/artwork_queue_repository.go index fdf88db24..ced5e9913 100644 --- a/persistence/artwork_queue_repository.go +++ b/persistence/artwork_queue_repository.go @@ -2,6 +2,7 @@ package persistence import ( "context" + "slices" "time" . "github.com/Masterminds/squirrel" @@ -9,6 +10,9 @@ import ( "github.com/pocketbase/dbx" ) +// enqueueChunkSize keeps each multi-row insert under SQLite's bind-variable limit (7 cols -> 700 vars). +const enqueueChunkSize = 100 + type artworkQueueRepository struct { sqlRepository } @@ -23,14 +27,15 @@ 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 + for chunk := range slices.Chunk(items, enqueueChunkSize) { + ins := Insert(r.tableName).Columns("item_kind", "item_id", "image_type", "priority", "attempts", "retry_at", "enqueued_at") + for _, it := range chunk { + if it.ImageType == "" { + it.ImageType = model.ImageTypePrimary + } + ins = ins.Values(it.ItemKind, it.ItemID, it.ImageType, it.Priority, 0, now, now) } - 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 + ins = ins.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 @@ -66,15 +71,20 @@ func (r *artworkQueueRepository) Delete(kind, id, imageType string) error { } func (r *artworkQueueRepository) Count() (int64, error) { - 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 res struct{ Count int64 } + err := r.queryOne(Select("count(*) as count").From(r.tableName), &res) + return res.Count, err +} + +func (r *artworkQueueRepository) EnqueueStaleAbsent(kind string, attemptedBefore time.Time) (int64, error) { + now := time.Now() + // DO NOTHING is deliberate: rechecks must not bump priority/retry_at of already-queued items. + ins := Expr(`INSERT INTO `+r.tableName+` (item_kind, item_id, image_type, priority, attempts, retry_at, enqueued_at) + SELECT item_kind, item_id, image_type, ?, 0, ?, ? + FROM `+itemArtworkTable+` WHERE item_kind = ? AND hash = '' AND attempted_at < ? + ON CONFLICT (item_kind, item_id, image_type) DO NOTHING`, + model.ArtworkPriorityRecheck, now, now, kind, attemptedBefore) + return r.executeSQL(ins) } var _ model.ArtworkQueueRepository = (*artworkQueueRepository)(nil) diff --git a/persistence/artwork_queue_repository_test.go b/persistence/artwork_queue_repository_test.go index 73665b07c..2b6f0d8c5 100644 --- a/persistence/artwork_queue_repository_test.go +++ b/persistence/artwork_queue_repository_test.go @@ -62,4 +62,22 @@ var _ = Describe("ArtworkQueueRepository", func() { n, _ = repo.Count() Expect(n).To(BeZero()) }) + + 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("ar", 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)) + }) }) diff --git a/persistence/artwork_repository.go b/persistence/artwork_repository.go index 4556bc588..c87836ca9 100644 --- a/persistence/artwork_repository.go +++ b/persistence/artwork_repository.go @@ -10,8 +10,18 @@ import ( "github.com/pocketbase/dbx" ) +const ( + itemArtworkTable = "item_artwork" + artworkBatchSize = 200 +) + +type itemArtworkSQL struct { + sqlRepository +} + type artworkRepository struct { sqlRepository + items itemArtworkSQL } func NewArtworkRepository(ctx context.Context, db dbx.Builder) model.ArtworkRepository { @@ -19,6 +29,9 @@ func NewArtworkRepository(ctx context.Context, db dbx.Builder) model.ArtworkRepo r.ctx = ctx r.db = db r.tableName = "artwork" + r.items.ctx = ctx + r.items.db = db + r.items.tableName = itemArtworkTable return r } @@ -35,20 +48,20 @@ func (r *artworkRepository) PutImage(a *model.Artwork) error { if a.CreatedAt.IsZero() { a.CreatedAt = time.Now() } - ins := Insert(r.tableName).SetMap(map[string]any{ - "hash": a.Hash, "mime": a.Mime, "width": a.Width, "height": a.Height, - "size_bytes": a.SizeBytes, "blur_hash": a.BlurHash, - "source_path": a.SourcePath, "ref_mtime": a.RefMtime, "created_at": a.CreatedAt, - }).Suffix(`ON CONFLICT (hash) DO UPDATE SET mime=excluded.mime, width=excluded.width, + values, err := toSQLArgs(*a) + if err != nil { + return err + } + ins := Insert(r.tableName).SetMap(values).Suffix(`ON CONFLICT (hash) DO UPDATE SET mime=excluded.mime, width=excluded.width, height=excluded.height, size_bytes=excluded.size_bytes, blur_hash=excluded.blur_hash, source_path=excluded.source_path, ref_mtime=excluded.ref_mtime`) - _, err := r.executeSQL(ins) + _, err = r.executeSQL(ins) return err } func (r *artworkRepository) GetImages(hashes []string) (map[string]model.Artwork, error) { res := map[string]model.Artwork{} - for chunk := range slices.Chunk(hashes, 200) { + for chunk := range slices.Chunk(hashes, artworkBatchSize) { sel := Select("*").From(r.tableName).Where(Eq{"hash": chunk}) var all []model.Artwork if err := r.queryAll(sel, &all); err != nil { @@ -61,11 +74,18 @@ func (r *artworkRepository) GetImages(hashes []string) (map[string]model.Artwork return res, nil } +func (r *artworkRepository) GetAllHashes() ([]string, error) { + sel := Select("hash").From(r.tableName) + var hashes []string + err := r.queryAllSlice(sel, &hashes) + return hashes, err +} + func (r *artworkRepository) GetOrphanHashes(createdBefore time.Time) ([]string, error) { sel := Select("hash").From(r.tableName). Where(And{ Lt{"created_at": createdBefore}, - Expr("hash NOT IN (SELECT hash FROM item_artwork WHERE hash <> '')"), + Expr("hash NOT IN (SELECT hash FROM " + itemArtworkTable + " WHERE hash <> '')"), }) var hashes []string err := r.queryAllSlice(sel, &hashes) @@ -73,7 +93,7 @@ func (r *artworkRepository) GetOrphanHashes(createdBefore time.Time) ([]string, } func (r *artworkRepository) DeleteImages(hashes ...string) error { - for chunk := range slices.Chunk(hashes, 200) { + for chunk := range slices.Chunk(hashes, artworkBatchSize) { if err := r.delete(Eq{"hash": chunk}); err != nil { return err } @@ -82,10 +102,10 @@ func (r *artworkRepository) DeleteImages(hashes ...string) error { } func (r *artworkRepository) GetItemArtwork(kind, id, imageType string) (*model.ItemArtwork, error) { - sel := Select("*").From("item_artwork"). + sel := Select("*").From(itemArtworkTable). Where(Eq{"item_kind": kind, "item_id": id, "image_type": imageType}) var res model.ItemArtwork - if err := r.queryOne(sel, &res); err != nil { + if err := r.items.queryOne(sel, &res); err != nil { return nil, err } return &res, nil @@ -96,28 +116,26 @@ func (r *artworkRepository) PutItemArtwork(ia *model.ItemArtwork) error { ia.ImageType = model.ImageTypePrimary } ia.UpdatedAt = time.Now() - ins := Insert("item_artwork").SetMap(map[string]any{ - "item_kind": ia.ItemKind, "item_id": ia.ItemID, "image_type": ia.ImageType, - "hash": ia.Hash, "source": ia.Source, - "attempted_at": ia.AttemptedAt, "updated_at": ia.UpdatedAt, - }).Suffix(`ON CONFLICT (item_kind, item_id, image_type) DO UPDATE SET + values, err := toSQLArgs(*ia) + if err != nil { + return err + } + ins := Insert(itemArtworkTable).SetMap(values).Suffix(`ON CONFLICT (item_kind, item_id, image_type) DO UPDATE SET hash=excluded.hash, source=excluded.source, attempted_at=excluded.attempted_at, updated_at=excluded.updated_at`) - _, err := r.executeSQL(ins) + _, err = r.items.executeSQL(ins) return err } func (r *artworkRepository) DeleteForItem(kind, id string) error { - del := Delete("item_artwork").Where(Eq{"item_kind": kind, "item_id": id}) - _, err := r.executeSQL(del) - return err + return r.items.delete(Eq{"item_kind": kind, "item_id": id}) } func (r *artworkRepository) GetInfoForItems(kind string, ids []string) (map[string]model.ItemArtworkInfo, error) { res := map[string]model.ItemArtworkInfo{} - for chunk := range slices.Chunk(ids, 200) { + for chunk := range slices.Chunk(ids, artworkBatchSize) { sel := Select("ia.item_id", "ia.hash", "COALESCE(a.blur_hash, '') as blur_hash"). - From("item_artwork ia"). + From(itemArtworkTable + " ia"). LeftJoin("artwork a ON a.hash = ia.hash"). Where(And{ Eq{"ia.item_kind": kind}, @@ -129,26 +147,16 @@ func (r *artworkRepository) GetInfoForItems(kind string, ids []string) (map[stri Hash string BlurHash string } - if err := r.queryAll(sel, &rows); err != nil { + if err := r.items.queryAll(sel, &rows); err != nil { return nil, err } for _, row := range rows { res[row.ItemID] = model.ItemArtworkInfo{ - ItemID: row.ItemID, Hash: row.Hash, BlurHash: row.BlurHash, Absent: row.Hash == "", + ItemID: row.ItemID, Hash: row.Hash, BlurHash: row.BlurHash, } } } return res, nil } -func (r *artworkRepository) EnqueueStaleAbsent(kind string, attemptedBefore time.Time) (int64, error) { - now := time.Now() - ins := Expr(`INSERT INTO artwork_queue (item_kind, item_id, image_type, priority, attempts, retry_at, enqueued_at) - SELECT item_kind, item_id, image_type, ?, 0, ?, ? - FROM item_artwork WHERE item_kind = ? AND hash = '' AND attempted_at < ? - ON CONFLICT (item_kind, item_id, image_type) DO NOTHING`, - model.ArtworkPriorityRecheck, now, now, kind, attemptedBefore) - return r.executeSQL(ins) -} - var _ model.ArtworkRepository = (*artworkRepository)(nil) diff --git a/persistence/artwork_repository_test.go b/persistence/artwork_repository_test.go index a77fd7038..0fdb122b6 100644 --- a/persistence/artwork_repository_test.go +++ b/persistence/artwork_repository_test.go @@ -27,144 +27,129 @@ var _ = Describe("ArtworkRepository", func() { repo = NewArtworkRepository(context.Background(), GetDBXBuilder()) }) - It("stores and retrieves an artwork by hash", func() { - a := &model.Artwork{Hash: "abc123", Mime: "image/jpeg", Width: 500, Height: 500, SizeBytes: 1234, BlurHash: "LKO2?U%2Tw=w"} - Expect(repo.PutImage(a)).To(Succeed()) + Context("image identity", func() { + It("stores and retrieves an artwork by hash", func() { + a := &model.Artwork{Hash: "abc123", Mime: "image/jpeg", Width: 500, Height: 500, SizeBytes: 1234, BlurHash: "LKO2?U%2Tw=w"} + Expect(repo.PutImage(a)).To(Succeed()) - got, err := repo.GetImage("abc123") - Expect(err).ToNot(HaveOccurred()) - Expect(got.Mime).To(Equal("image/jpeg")) - Expect(got.BlurHash).To(Equal("LKO2?U%2Tw=w")) - Expect(got.CreatedAt).ToNot(BeZero()) + got, err := repo.GetImage("abc123") + Expect(err).ToNot(HaveOccurred()) + Expect(got.Mime).To(Equal("image/jpeg")) + Expect(got.BlurHash).To(Equal("LKO2?U%2Tw=w")) + Expect(got.CreatedAt).ToNot(BeZero()) + }) + + It("is idempotent on Put (upsert by hash)", func() { + a := &model.Artwork{Hash: "dup1", Mime: "image/png"} + Expect(repo.PutImage(a)).To(Succeed()) + a.BlurHash = "XYZ" + Expect(repo.PutImage(a)).To(Succeed()) + got, _ := repo.GetImage("dup1") + Expect(got.BlurHash).To(Equal("XYZ")) + }) + + It("returns ErrNotFound for a missing hash", func() { + _, err := repo.GetImage("nope") + Expect(err).To(MatchError(model.ErrNotFound)) + }) + + It("fetches a batch", func() { + Expect(repo.PutImage(&model.Artwork{Hash: "b1", Mime: "image/jpeg"})).To(Succeed()) + Expect(repo.PutImage(&model.Artwork{Hash: "b2", Mime: "image/png"})).To(Succeed()) + got, err := repo.GetImages([]string{"b1", "b2", "missing"}) + Expect(err).ToNot(HaveOccurred()) + Expect(got).To(HaveLen(2)) + Expect(got["b2"].Mime).To(Equal("image/png")) + }) + + It("returns every stored hash", func() { + Expect(repo.PutImage(&model.Artwork{Hash: "all1", Mime: "image/jpeg"})).To(Succeed()) + Expect(repo.PutImage(&model.Artwork{Hash: "all2", Mime: "image/png"})).To(Succeed()) + hashes, err := repo.GetAllHashes() + Expect(err).ToNot(HaveOccurred()) + Expect(hashes).To(ConsistOf("all1", "all2")) + }) + + It("finds orphans older than cutoff, honoring item_artwork references", func() { + Expect(repo.PutImage(&model.Artwork{Hash: "orph1", Mime: "image/jpeg"})).To(Succeed()) + Expect(repo.PutImage(&model.Artwork{Hash: "ref1", Mime: "image/jpeg"})).To(Succeed()) + Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "a1", ImageType: model.ImageTypePrimary, Hash: "ref1", Source: "folder"})).To(Succeed()) + + orphans, err := repo.GetOrphanHashes(time.Now().Add(time.Minute)) + Expect(err).ToNot(HaveOccurred()) + Expect(orphans).To(ContainElement("orph1")) + Expect(orphans).ToNot(ContainElement("ref1")) + + orphans, err = repo.GetOrphanHashes(time.Now().Add(-time.Hour)) + Expect(err).ToNot(HaveOccurred()) + Expect(orphans).To(BeEmpty()) + }) + + It("deletes by hashes", func() { + Expect(repo.PutImage(&model.Artwork{Hash: "d1", Mime: "image/jpeg"})).To(Succeed()) + Expect(repo.DeleteImages("d1")).To(Succeed()) + _, err := repo.GetImage("d1") + Expect(err).To(MatchError(model.ErrNotFound)) + }) + + It("fetches a batch larger than the SQL variable limit", func() { + hashes := make([]string, 0, 250) + for i := range 250 { + h := fmt.Sprintf("big%03d", i) + Expect(repo.PutImage(&model.Artwork{Hash: h, Mime: "image/jpeg"})).To(Succeed()) + hashes = append(hashes, h) + } + hashes = append(hashes, "absent1", "absent2") + + got, err := repo.GetImages(hashes) + Expect(err).ToNot(HaveOccurred()) + Expect(got).To(HaveLen(250)) + }) }) - It("is idempotent on Put (upsert by hash)", func() { - a := &model.Artwork{Hash: "dup1", Mime: "image/png"} - Expect(repo.PutImage(a)).To(Succeed()) - a.BlurHash = "XYZ" - Expect(repo.PutImage(a)).To(Succeed()) - got, _ := repo.GetImage("dup1") - Expect(got.BlurHash).To(Equal("XYZ")) - }) + Context("item state", func() { + It("upserts and reads state", func() { + ia := &model.ItemArtwork{ItemKind: "al", ItemID: "al1", ImageType: model.ImageTypePrimary, + Hash: "h1", Source: "folder", AttemptedAt: time.Now()} + Expect(repo.PutItemArtwork(ia)).To(Succeed()) + ia.Source = "embedded" + Expect(repo.PutItemArtwork(ia)).To(Succeed()) - It("returns ErrNotFound for a missing hash", func() { - _, err := repo.GetImage("nope") - Expect(err).To(MatchError(model.ErrNotFound)) - }) + got, err := repo.GetItemArtwork("al", "al1", model.ImageTypePrimary) + Expect(err).ToNot(HaveOccurred()) + Expect(got.Source).To(Equal("embedded")) + Expect(got.UpdatedAt).ToNot(BeZero()) + }) - It("fetches a batch", func() { - Expect(repo.PutImage(&model.Artwork{Hash: "b1", Mime: "image/jpeg"})).To(Succeed()) - Expect(repo.PutImage(&model.Artwork{Hash: "b2", Mime: "image/png"})).To(Succeed()) - got, err := repo.GetImages([]string{"b1", "b2", "missing"}) - Expect(err).ToNot(HaveOccurred()) - Expect(got).To(HaveLen(2)) - Expect(got["b2"].Mime).To(Equal("image/png")) - }) + It("represents known-absent as empty hash", func() { + Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "ar", ItemID: "ar1", + ImageType: model.ImageTypePrimary, Hash: "", AttemptedAt: time.Now()})).To(Succeed()) + got, err := repo.GetItemArtwork("ar", "ar1", model.ImageTypePrimary) + Expect(err).ToNot(HaveOccurred()) + Expect(got.Hash).To(BeEmpty()) + }) - It("finds orphans older than cutoff, honoring item_artwork references", func() { - Expect(repo.PutImage(&model.Artwork{Hash: "orph1", Mime: "image/jpeg"})).To(Succeed()) - Expect(repo.PutImage(&model.Artwork{Hash: "ref1", Mime: "image/jpeg"})).To(Succeed()) - Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "a1", ImageType: model.ImageTypePrimary, Hash: "ref1", Source: "folder"})).To(Succeed()) + It("hydrates a page in one batch, including blurhash and absence", func() { + Expect(repo.PutImage(&model.Artwork{Hash: "h9", Mime: "image/jpeg", BlurHash: "BH9"})).To(Succeed()) + Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "x1", ImageType: model.ImageTypePrimary, Hash: "h9", Source: "folder"})).To(Succeed()) + Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "x2", ImageType: model.ImageTypePrimary, Hash: "", Source: ""})).To(Succeed()) - orphans, err := repo.GetOrphanHashes(time.Now().Add(time.Minute)) - Expect(err).ToNot(HaveOccurred()) - Expect(orphans).To(ContainElement("orph1")) - Expect(orphans).ToNot(ContainElement("ref1")) + info, err := repo.GetInfoForItems("al", []string{"x1", "x2", "x3"}) + Expect(err).ToNot(HaveOccurred()) + Expect(info).To(HaveLen(2)) + Expect(info["x1"].Hash).To(Equal("h9")) + Expect(info["x1"].BlurHash).To(Equal("BH9")) + Expect(info["x1"].Absent()).To(BeFalse()) + Expect(info["x2"].Absent()).To(BeTrue()) + _, unresolved := info["x3"] + Expect(unresolved).To(BeFalse()) + }) - orphans, err = repo.GetOrphanHashes(time.Now().Add(-time.Hour)) - Expect(err).ToNot(HaveOccurred()) - Expect(orphans).To(BeEmpty()) - }) - - It("deletes by hashes", func() { - Expect(repo.PutImage(&model.Artwork{Hash: "d1", Mime: "image/jpeg"})).To(Succeed()) - Expect(repo.DeleteImages("d1")).To(Succeed()) - _, err := repo.GetImage("d1") - Expect(err).To(MatchError(model.ErrNotFound)) - }) - - It("fetches a batch larger than the SQL variable limit", func() { - hashes := make([]string, 0, 250) - for i := 0; i < 250; i++ { - h := fmt.Sprintf("big%03d", i) - Expect(repo.PutImage(&model.Artwork{Hash: h, Mime: "image/jpeg"})).To(Succeed()) - hashes = append(hashes, h) - } - hashes = append(hashes, "absent1", "absent2") - - got, err := repo.GetImages(hashes) - Expect(err).ToNot(HaveOccurred()) - Expect(got).To(HaveLen(250)) - }) -}) - -var _ = Describe("ArtworkRepository item state", func() { - var repo model.ArtworkRepository - - BeforeEach(func() { - clearArtworkTables() - repo = NewArtworkRepository(context.Background(), GetDBXBuilder()) - }) - - It("upserts and reads state", func() { - ia := &model.ItemArtwork{ItemKind: "al", ItemID: "al1", ImageType: model.ImageTypePrimary, - Hash: "h1", Source: "folder", AttemptedAt: time.Now()} - Expect(repo.PutItemArtwork(ia)).To(Succeed()) - ia.Source = "embedded" - Expect(repo.PutItemArtwork(ia)).To(Succeed()) - - got, err := repo.GetItemArtwork("al", "al1", model.ImageTypePrimary) - Expect(err).ToNot(HaveOccurred()) - Expect(got.Source).To(Equal("embedded")) - Expect(got.UpdatedAt).ToNot(BeZero()) - }) - - It("represents known-absent as empty hash", func() { - Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "ar", ItemID: "ar1", - ImageType: model.ImageTypePrimary, Hash: "", AttemptedAt: time.Now()})).To(Succeed()) - got, err := repo.GetItemArtwork("ar", "ar1", model.ImageTypePrimary) - Expect(err).ToNot(HaveOccurred()) - Expect(got.Hash).To(BeEmpty()) - }) - - It("hydrates a page in one batch, including blurhash and absence", func() { - Expect(repo.PutImage(&model.Artwork{Hash: "h9", Mime: "image/jpeg", BlurHash: "BH9"})).To(Succeed()) - Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "x1", ImageType: model.ImageTypePrimary, Hash: "h9", Source: "folder"})).To(Succeed()) - Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "x2", ImageType: model.ImageTypePrimary, Hash: "", Source: ""})).To(Succeed()) - - info, err := repo.GetInfoForItems("al", []string{"x1", "x2", "x3"}) - Expect(err).ToNot(HaveOccurred()) - Expect(info).To(HaveLen(2)) - Expect(info["x1"].Hash).To(Equal("h9")) - Expect(info["x1"].BlurHash).To(Equal("BH9")) - Expect(info["x1"].Absent).To(BeFalse()) - Expect(info["x2"].Absent).To(BeTrue()) - _, unresolved := info["x3"] - Expect(unresolved).To(BeFalse()) - }) - - It("deletes all rows for an item", func() { - Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "pl", ItemID: "p1", ImageType: model.ImageTypePrimary, Hash: "h1"})).To(Succeed()) - Expect(repo.DeleteForItem("pl", "p1")).To(Succeed()) - _, err := repo.GetItemArtwork("pl", "p1", model.ImageTypePrimary) - Expect(err).To(MatchError(model.ErrNotFound)) - }) - - It("enqueues stale absent states for recheck", func() { - old := time.Now().Add(-48 * time.Hour) - Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "ar", ItemID: "stale1", ImageType: model.ImageTypePrimary, Hash: "", AttemptedAt: old})).To(Succeed()) - Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "ar", ItemID: "fresh1", ImageType: model.ImageTypePrimary, Hash: "", AttemptedAt: time.Now()})).To(Succeed()) - Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "ar", ItemID: "found1", ImageType: model.ImageTypePrimary, Hash: "hX", AttemptedAt: old})).To(Succeed()) - - n, err := repo.EnqueueStaleAbsent("ar", time.Now().Add(-24*time.Hour)) - Expect(err).ToNot(HaveOccurred()) - Expect(n).To(Equal(int64(1))) - - qRepo := NewArtworkQueueRepository(context.Background(), GetDBXBuilder()) - items, err := qRepo.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("deletes all rows for an item", func() { + Expect(repo.PutItemArtwork(&model.ItemArtwork{ItemKind: "pl", ItemID: "p1", ImageType: model.ImageTypePrimary, Hash: "h1"})).To(Succeed()) + Expect(repo.DeleteForItem("pl", "p1")).To(Succeed()) + _, err := repo.GetItemArtwork("pl", "p1", model.ImageTypePrimary) + Expect(err).To(MatchError(model.ErrNotFound)) + }) }) }) diff --git a/tests/mock_artwork_queue_repo.go b/tests/mock_artwork_queue_repo.go index b08a1297c..dcec7486d 100644 --- a/tests/mock_artwork_queue_repo.go +++ b/tests/mock_artwork_queue_repo.go @@ -9,7 +9,7 @@ import ( type MockArtworkQueueRepo struct { model.ArtworkQueueRepository - Data map[string]model.ArtworkQueueItem // key: kind + "|" + id + "|" + imageType + Data map[string]model.ArtworkQueueItem // keyed by iaKey(kind, id, imageType) Err error } @@ -21,16 +21,23 @@ func (m *MockArtworkQueueRepo) Enqueue(items ...model.ArtworkQueueItem) error { if m.Err != nil { return m.Err } + now := time.Now() for _, it := range items { - k := iaKey(it.ItemKind, it.ItemID, it.ImageType) - if prev, ok := m.Data[k]; ok && prev.Priority > it.Priority { - it.Priority = prev.Priority - } - if it.EnqueuedAt.IsZero() { - it.EnqueuedAt = time.Now() + if it.ImageType == "" { + it.ImageType = model.ImageTypePrimary } if it.RetryAt.IsZero() { - it.RetryAt = time.Now() + it.RetryAt = now + } + k := iaKey(it.ItemKind, it.ItemID, it.ImageType) + if prev, ok := m.Data[k]; ok { + prev.Priority = max(prev.Priority, it.Priority) + prev.RetryAt = it.RetryAt + m.Data[k] = prev + continue + } + if it.EnqueuedAt.IsZero() { + it.EnqueuedAt = now } m.Data[k] = it } @@ -89,3 +96,7 @@ func (m *MockArtworkQueueRepo) Count() (int64, error) { } return int64(len(m.Data)), nil } + +func (m *MockArtworkQueueRepo) EnqueueStaleAbsent(kind string, attemptedBefore time.Time) (int64, error) { + return 0, m.Err +} diff --git a/tests/mock_artwork_repo.go b/tests/mock_artwork_repo.go index 3989499ce..8ac4820e6 100644 --- a/tests/mock_artwork_repo.go +++ b/tests/mock_artwork_repo.go @@ -9,7 +9,7 @@ import ( type MockArtworkRepo struct { model.ArtworkRepository Data map[string]model.Artwork - ItemData map[string]model.ItemArtwork // key: kind + "|" + id + "|" + imageType + ItemData map[string]model.ItemArtwork // keyed by iaKey(kind, id, imageType) OrphanHashes []string Err error } @@ -58,6 +58,17 @@ func (m *MockArtworkRepo) GetOrphanHashes(createdBefore time.Time) ([]string, er return m.OrphanHashes, m.Err } +func (m *MockArtworkRepo) GetAllHashes() ([]string, error) { + if m.Err != nil { + return nil, m.Err + } + hashes := make([]string, 0, len(m.Data)) + for h := range m.Data { + hashes = append(hashes, h) + } + return hashes, nil +} + func (m *MockArtworkRepo) DeleteImages(hashes ...string) error { if m.Err != nil { return m.Err @@ -82,6 +93,9 @@ func (m *MockArtworkRepo) PutItemArtwork(ia *model.ItemArtwork) error { if m.Err != nil { return m.Err } + if ia.ImageType == "" { + ia.ImageType = model.ImageTypePrimary + } ia.UpdatedAt = time.Now() m.ItemData[iaKey(ia.ItemKind, ia.ItemID, ia.ImageType)] = *ia return nil @@ -106,12 +120,12 @@ func (m *MockArtworkRepo) GetInfoForItems(kind string, ids []string) (map[string res := map[string]model.ItemArtworkInfo{} for _, id := range ids { if ia, ok := m.ItemData[iaKey(kind, id, model.ImageTypePrimary)]; ok { - res[id] = model.ItemArtworkInfo{ItemID: id, Hash: ia.Hash, Absent: ia.Hash == ""} + info := model.ItemArtworkInfo{ItemID: id, Hash: ia.Hash} + if a, ok := m.Data[ia.Hash]; ok { + info.BlurHash = a.BlurHash + } + res[id] = info } } return res, nil } - -func (m *MockArtworkRepo) EnqueueStaleAbsent(kind string, attemptedBefore time.Time) (int64, error) { - return 0, m.Err -}