mirror of
https://github.com/navidrome/navidrome.git
synced 2026-08-01 07:21:17 +00:00
feat(artwork): add acquisition processor
Resolves one queue item end to end: hash/dedup, decode + 128px thumbnail blurhash, place bytes (store vs source file), and persist found/absent/ failed state for the worker (Task 4) to act on.
This commit is contained in:
Deluan
parent
608db503a7
commit
e0655dc882
@ -15,9 +15,17 @@ import (
|
||||
"github.com/navidrome/navidrome/tests"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
"go.uber.org/goleak"
|
||||
)
|
||||
|
||||
func TestArtwork(t *testing.T) {
|
||||
// Only run goleak checks when the GOLEAK env var is set
|
||||
if os.Getenv("GOLEAK") != "" {
|
||||
defer goleak.VerifyNone(t,
|
||||
goleak.IgnoreTopFunction("github.com/onsi/ginkgo/v2/internal/interrupt_handler.(*InterruptHandler).registerForInterrupts.func2"),
|
||||
)
|
||||
}
|
||||
|
||||
tests.Init(t, false)
|
||||
log.SetLevel(log.LevelFatal)
|
||||
RegisterFailHandler(Fail)
|
||||
|
||||
210
core/artwork/processor.go
Normal file
210
core/artwork/processor.go
Normal file
@ -0,0 +1,210 @@
|
||||
package artwork
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"image"
|
||||
"image/draw"
|
||||
"io"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/core/artwork/blurhash"
|
||||
"github.com/navidrome/navidrome/core/external"
|
||||
"github.com/navidrome/navidrome/core/ffmpeg"
|
||||
"github.com/navidrome/navidrome/log"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
xdraw "golang.org/x/image/draw"
|
||||
)
|
||||
|
||||
// outcome tells the worker what to do with the queue row: found/absent
|
||||
// delete it, failed reschedules it via MarkFailed.
|
||||
type outcome int
|
||||
|
||||
const (
|
||||
outcomeFound outcome = iota
|
||||
outcomeAbsent
|
||||
outcomeFailed
|
||||
)
|
||||
|
||||
// thumbnailSize is the max dimension fed to blurhash, matching Jellyfin's own input cap.
|
||||
const thumbnailSize = 128
|
||||
|
||||
// workerDeps are the collaborators processItem needs; extGate is nil outside
|
||||
// tests, in which case resolveItem falls back to a plain passthrough.
|
||||
type workerDeps struct {
|
||||
ds model.DataStore
|
||||
store *ImageStore
|
||||
prov external.Provider
|
||||
ffmpeg ffmpeg.FFmpeg
|
||||
extGate func(func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error)
|
||||
}
|
||||
|
||||
// processItem resolves one queue item end to end: find an image, hash/decode/
|
||||
// blurhash it, place its bytes, and persist the resulting state.
|
||||
func processItem(ctx context.Context, deps *workerDeps, item model.ArtworkQueueItem) outcome {
|
||||
repo := deps.ds.Artwork(ctx)
|
||||
|
||||
res, err := resolveItem(ctx, deps.ds, deps.prov, deps.ffmpeg, item, deps.extGate)
|
||||
if err != nil {
|
||||
log.Warn(ctx, "artwork: could not resolve item", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed
|
||||
}
|
||||
if res.reader == nil {
|
||||
if res.extError {
|
||||
// An external source errored/timed out: never settle on absent, keep serving old state.
|
||||
return outcomeFailed
|
||||
}
|
||||
return writeAbsent(ctx, repo, item)
|
||||
}
|
||||
defer res.reader.Close()
|
||||
|
||||
data, err := io.ReadAll(res.reader)
|
||||
if err != nil {
|
||||
log.Warn(ctx, "artwork: failed to read resolved image", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed
|
||||
}
|
||||
log.Debug(ctx, "artwork: read resolved image", "kind", item.ItemKind, "id", item.ItemID, "source", res.source, "bytes", len(data))
|
||||
|
||||
hash, err := HashImage(bytes.NewReader(data))
|
||||
if err != nil {
|
||||
log.Warn(ctx, "artwork: failed to hash image", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed
|
||||
}
|
||||
|
||||
art, err := repo.GetImage(hash)
|
||||
switch {
|
||||
case err == nil:
|
||||
// Dedup hit: identical bytes already known, reuse dims/mime/blurhash.
|
||||
case errors.Is(err, model.ErrNotFound):
|
||||
art, err = decodeArtwork(ctx, hash, data)
|
||||
if err != nil {
|
||||
log.Warn(ctx, "artwork: failed to decode resolved image", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed
|
||||
}
|
||||
default:
|
||||
log.Warn(ctx, "artwork: failed to look up image hash", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed
|
||||
}
|
||||
art.SizeBytes = int64(len(data))
|
||||
|
||||
if err := placeBytes(deps.store, art, res, data); err != nil {
|
||||
log.Warn(ctx, "artwork: failed to write image store", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed
|
||||
}
|
||||
if err := repo.PutImage(art); err != nil {
|
||||
log.Warn(ctx, "artwork: failed to persist artwork image", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed
|
||||
}
|
||||
if err := repo.PutItemArtwork(&model.ItemArtwork{
|
||||
ItemKind: item.ItemKind,
|
||||
ItemID: item.ItemID,
|
||||
ImageType: item.ImageType,
|
||||
Hash: hash,
|
||||
Source: res.source,
|
||||
AttemptedAt: time.Now(),
|
||||
}); err != nil {
|
||||
log.Warn(ctx, "artwork: failed to persist item artwork state", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed
|
||||
}
|
||||
return outcomeFound
|
||||
}
|
||||
|
||||
// writeAbsent records a known-absent state: every local/external source answered definitively "no".
|
||||
func writeAbsent(ctx context.Context, repo model.ArtworkRepository, item model.ArtworkQueueItem) outcome {
|
||||
err := repo.PutItemArtwork(&model.ItemArtwork{
|
||||
ItemKind: item.ItemKind,
|
||||
ItemID: item.ItemID,
|
||||
ImageType: item.ImageType,
|
||||
AttemptedAt: time.Now(),
|
||||
})
|
||||
if err != nil {
|
||||
log.Warn(ctx, "artwork: failed to persist absent state", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed
|
||||
}
|
||||
return outcomeAbsent
|
||||
}
|
||||
|
||||
// decodeArtwork builds a new Artwork row from raw bytes: dimensions, mime and a
|
||||
// blurhash computed from a downscaled thumbnail.
|
||||
func decodeArtwork(ctx context.Context, hash string, data []byte) (*model.Artwork, error) {
|
||||
cfg, format, err := image.DecodeConfig(bytes.NewReader(data))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("decode image config: %w", err)
|
||||
}
|
||||
img, _, err := image.Decode(bytes.NewReader(data))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("decode image: %w", err)
|
||||
}
|
||||
|
||||
thumb := makeThumbnail(img, thumbnailSize)
|
||||
xComp, yComp := blurhash.Components(thumb.Bounds().Dx(), thumb.Bounds().Dy())
|
||||
bh, err := blurhash.Encode(thumb, xComp, yComp)
|
||||
if err != nil {
|
||||
log.Warn(ctx, "artwork: blurhash encoding failed", "hash", hash, err)
|
||||
bh = ""
|
||||
}
|
||||
|
||||
return &model.Artwork{
|
||||
Hash: hash,
|
||||
Mime: mimeForFormat(format),
|
||||
Width: cfg.Width,
|
||||
Height: cfg.Height,
|
||||
BlurHash: bh,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// makeThumbnail downscales img to fit within maxSize on its longest side,
|
||||
// reusing reader_resized.go's fast-scale-type + CatmullRom approach. Images
|
||||
// already within bounds are returned as-is (no upscaling).
|
||||
func makeThumbnail(img image.Image, maxSize int) image.Image {
|
||||
b := img.Bounds()
|
||||
w, h := b.Dx(), b.Dy()
|
||||
if w <= maxSize && h <= maxSize {
|
||||
return toFastScaleType(img)
|
||||
}
|
||||
scale := float64(maxSize) / float64(max(w, h))
|
||||
dst := image.NewRGBA(image.Rect(0, 0, max(1, int(float64(w)*scale)), max(1, int(float64(h)*scale))))
|
||||
xdraw.CatmullRom.Scale(dst, dst.Bounds(), toFastScaleType(img), b, draw.Src, nil)
|
||||
return dst
|
||||
}
|
||||
|
||||
// isFileBacked reports whether a resolution's bytes already live in a library/upload
|
||||
// file, so the acquisition must not duplicate them into the content-addressed store.
|
||||
func isFileBacked(source string) bool {
|
||||
return source == "folder" || source == "upload"
|
||||
}
|
||||
|
||||
// placeBytes fills in art's SourcePath/RefMtime and, for sources with no library file
|
||||
// backing them, writes the bytes into the store (embedded keeps its audio file provenance).
|
||||
func placeBytes(store *ImageStore, art *model.Artwork, res resolution, data []byte) error {
|
||||
if isFileBacked(res.source) {
|
||||
art.SourcePath = res.sourcePath
|
||||
art.RefMtime = res.refMtime
|
||||
return nil
|
||||
}
|
||||
art.SourcePath = ""
|
||||
art.RefMtime = 0
|
||||
if res.source == "embedded" {
|
||||
art.SourcePath = res.sourcePath
|
||||
art.RefMtime = res.refMtime
|
||||
}
|
||||
return store.Write(art.Hash, art.Mime, bytes.NewReader(data))
|
||||
}
|
||||
|
||||
// mimeForFormat maps an image.Decode format name to its MIME type; extForMime
|
||||
// in image_store.go performs the inverse for content-addressed file paths.
|
||||
func mimeForFormat(format string) string {
|
||||
switch format {
|
||||
case "jpeg":
|
||||
return "image/jpeg"
|
||||
case "png":
|
||||
return "image/png"
|
||||
case "gif":
|
||||
return "image/gif"
|
||||
case "webp":
|
||||
return "image/webp"
|
||||
}
|
||||
return "application/octet-stream"
|
||||
}
|
||||
206
core/artwork/processor_test.go
Normal file
206
core/artwork/processor_test.go
Normal file
@ -0,0 +1,206 @@
|
||||
package artwork
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/conf/configtest"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
"github.com/navidrome/navidrome/tests"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
var _ = Describe("processItem", func() {
|
||||
var (
|
||||
ctx context.Context
|
||||
ds *tests.MockDataStore
|
||||
folderRepo *fakeFolderRepo
|
||||
libRepo *tests.MockLibraryRepo
|
||||
ffm *tests.MockFFmpeg
|
||||
prov *fakeExternalProvider
|
||||
store *ImageStore
|
||||
artRepo *tests.MockArtworkRepo
|
||||
repoRoot string
|
||||
deps *workerDeps
|
||||
)
|
||||
|
||||
BeforeEach(func() {
|
||||
DeferCleanup(configtest.SetupConfig())
|
||||
ctx = context.Background()
|
||||
var err error
|
||||
repoRoot, err = os.Getwd()
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
folderRepo = &fakeFolderRepo{}
|
||||
libRepo = &tests.MockLibraryRepo{}
|
||||
libRepo.SetData(model.Libraries{{ID: 0, Path: testFileLibPath(repoRoot)}})
|
||||
ffm = tests.NewMockFFmpeg("")
|
||||
prov = &fakeExternalProvider{}
|
||||
artRepo = tests.CreateMockArtworkRepo()
|
||||
ds = &tests.MockDataStore{
|
||||
MockedFolder: folderRepo,
|
||||
MockedLibrary: libRepo,
|
||||
MockedArtwork: artRepo,
|
||||
}
|
||||
ds.MockedAlbum = tests.CreateMockAlbumRepo()
|
||||
store = NewImageStore(GinkgoT().TempDir())
|
||||
deps = &workerDeps{ds: ds, store: store, prov: prov, ffmpeg: ffm}
|
||||
|
||||
conf.Server.CoverArtPriority = "cover.jpg, embedded"
|
||||
})
|
||||
|
||||
It("found-folder: persists state from a folder image, writes no store file, keeps sourcePath/refMtime", func() {
|
||||
folderRepo.result = []model.Folder{{
|
||||
Path: "tests/fixtures/artist/an-album",
|
||||
ImageFiles: []string{"cover.jpg"},
|
||||
}}
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "al1", Name: "Album", FolderIDs: []string{"f1"}},
|
||||
})
|
||||
|
||||
out := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al1"})
|
||||
Expect(out).To(Equal(outcomeFound))
|
||||
|
||||
ia, err := artRepo.GetItemArtwork("al", "al1", model.ImageTypePrimary)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(ia.Hash).ToNot(BeEmpty())
|
||||
Expect(ia.Source).To(Equal("folder"))
|
||||
|
||||
art, err := artRepo.GetImage(ia.Hash)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(filepath.ToSlash(art.SourcePath)).To(HaveSuffix("tests/fixtures/artist/an-album/cover.jpg"))
|
||||
Expect(art.RefMtime).To(BeNumerically(">", 0))
|
||||
|
||||
_, err = store.Open(ia.Hash, art.Mime)
|
||||
Expect(os.IsNotExist(err)).To(BeTrue(), "folder-backed art must not be duplicated into the store")
|
||||
})
|
||||
|
||||
It("found-embedded: writes a store file and computes a non-empty blurhash from a real fixture", func() {
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "al2", Name: "Album", EmbedArtPath: "tests/fixtures/artist/an-album/test.mp3", FolderIDs: []string{"f1"}},
|
||||
})
|
||||
folderRepo.result = nil
|
||||
|
||||
out := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al2"})
|
||||
Expect(out).To(Equal(outcomeFound))
|
||||
|
||||
ia, err := artRepo.GetItemArtwork("al", "al2", model.ImageTypePrimary)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(ia.Source).To(Equal("embedded"))
|
||||
|
||||
art, err := artRepo.GetImage(ia.Hash)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(art.BlurHash).ToNot(BeEmpty())
|
||||
Expect(filepath.ToSlash(art.SourcePath)).To(HaveSuffix("tests/fixtures/artist/an-album/test.mp3"))
|
||||
|
||||
rc, err := store.Open(ia.Hash, art.Mime)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
rc.Close()
|
||||
})
|
||||
|
||||
It("absent: no local source and no external error persists a known-absent state", func() {
|
||||
folderRepo.result = nil
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "al3", Name: "Album"},
|
||||
})
|
||||
|
||||
out := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al3"})
|
||||
Expect(out).To(Equal(outcomeAbsent))
|
||||
|
||||
ia, err := artRepo.GetItemArtwork("al", "al3", model.ImageTypePrimary)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(ia.Hash).To(BeEmpty())
|
||||
Expect(ia.Source).To(BeEmpty())
|
||||
Expect(ia.AttemptedAt).To(BeTemporally("~", time.Now(), time.Second))
|
||||
})
|
||||
|
||||
It("failed-on-extError: leaves the item's state untouched", func() {
|
||||
conf.Server.CoverArtPriority = "external"
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "al4", Name: "Album"},
|
||||
})
|
||||
prov.albumImage = func(context.Context, string) (*url.URL, error) {
|
||||
return nil, errors.New("agent timed out")
|
||||
}
|
||||
|
||||
out := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al4"})
|
||||
Expect(out).To(Equal(outcomeFailed))
|
||||
|
||||
_, err := artRepo.GetItemArtwork("al", "al4", model.ImageTypePrimary)
|
||||
Expect(err).To(MatchError(model.ErrNotFound))
|
||||
})
|
||||
|
||||
It("dedup: a second item with identical bytes skips decode and reuses the artwork row", func() {
|
||||
folderRepo.result = []model.Folder{{
|
||||
Path: "tests/fixtures/artist/an-album",
|
||||
ImageFiles: []string{"cover.jpg"},
|
||||
}}
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "al5", Name: "Album A", FolderIDs: []string{"f1"}},
|
||||
{ID: "al6", Name: "Album B", FolderIDs: []string{"f1"}},
|
||||
})
|
||||
|
||||
out1 := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al5"})
|
||||
Expect(out1).To(Equal(outcomeFound))
|
||||
ia1, err := artRepo.GetItemArtwork("al", "al5", model.ImageTypePrimary)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
// Poison the stored blurhash: if the second item re-decodes instead of
|
||||
// deduping on hash, this sentinel gets overwritten by a real computed value.
|
||||
poisoned := artRepo.Data[ia1.Hash]
|
||||
poisoned.BlurHash = "SENTINEL"
|
||||
artRepo.Data[ia1.Hash] = poisoned
|
||||
|
||||
out2 := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al6"})
|
||||
Expect(out2).To(Equal(outcomeFound))
|
||||
ia2, err := artRepo.GetItemArtwork("al", "al6", model.ImageTypePrimary)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(ia2.Hash).To(Equal(ia1.Hash))
|
||||
|
||||
reused, err := artRepo.GetImage(ia1.Hash)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(reused.BlurHash).To(Equal("SENTINEL"))
|
||||
})
|
||||
|
||||
It("decode failure on found bytes: fails without writing state", func() {
|
||||
tmpDir := GinkgoT().TempDir()
|
||||
conf.Server.DataFolder = conf.NewDir(tmpDir)
|
||||
Expect(os.MkdirAll(filepath.Join(tmpDir, "artwork", "radio"), 0755)).To(Succeed())
|
||||
imgPath := filepath.Join(tmpDir, "artwork", "radio", "ra1_test.jpg")
|
||||
Expect(os.WriteFile(imgPath, []byte("not actually an image"), 0600)).To(Succeed())
|
||||
|
||||
radioRepo := tests.CreateMockedRadioRepo()
|
||||
radioRepo.Data = map[string]*model.Radio{"ra1": {ID: "ra1", Name: "Radio", UploadedImage: "ra1_test.jpg"}}
|
||||
ds.MockedRadio = radioRepo
|
||||
|
||||
out := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "ra", ItemID: "ra1"})
|
||||
Expect(out).To(Equal(outcomeFailed))
|
||||
|
||||
_, err := artRepo.GetItemArtwork("ra", "ra1", model.ImageTypePrimary)
|
||||
Expect(err).To(MatchError(model.ErrNotFound))
|
||||
})
|
||||
|
||||
It("store write failure: fails without writing state", func() {
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "al7", Name: "Album", EmbedArtPath: "tests/fixtures/artist/an-album/test.mp3", FolderIDs: []string{"f1"}},
|
||||
})
|
||||
folderRepo.result = nil
|
||||
|
||||
// A store root that is a plain file makes every MkdirAll under it fail.
|
||||
blockedRoot := filepath.Join(GinkgoT().TempDir(), "not-a-dir")
|
||||
Expect(os.WriteFile(blockedRoot, []byte("x"), 0600)).To(Succeed())
|
||||
deps.store = NewImageStore(blockedRoot)
|
||||
|
||||
out := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al7"})
|
||||
Expect(out).To(Equal(outcomeFailed))
|
||||
|
||||
_, err := artRepo.GetItemArtwork("al", "al7", model.ImageTypePrimary)
|
||||
Expect(err).To(MatchError(model.ErrNotFound))
|
||||
})
|
||||
})
|
||||
Loading…
x
Reference in New Issue
Block a user