gabehf/Koito
1
0
Fork 0
mirror of https://github.com/gabehf/Koito.git synced 2026-03-07 21:48:18 -08:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: go back to synchronous image processing

Browse source
This commit is contained in:
Gabe Farrell 2025-06-12 00:30:01 -04:00
parent aba2b76def
commit 1a5a6acc95
10 changed files with 58 additions and 168 deletions
Download patch file Download diff file Expand all files Collapse all files

4
engine/engine.go
View file

@ -101,8 +101,6 @@ func Run(
EnableDeezer: !cfg.DeezerDisabled(),
})
ip := catalog.NewImageProcessor(1)
userCount, _ := store.CountUsers(ctx)
if userCount < 1 {
l.Debug().Msg("Creating default user...")
@ -147,7 +145,7 @@ func Run(
mux.Use(chimiddleware.Recoverer)
mux.Use(chimiddleware.RealIP)
// call router binds on mux
bindRoutes(mux, &ready, store, mbzC, ip)
bindRoutes(mux, &ready, store, mbzC)
httpServer := &http.Server{
Addr: cfg.ListenAddr(),

12
engine/handlers/image_handler.go
View file

@ -17,7 +17,7 @@ import (
"github.com/google/uuid"
)
func ImageHandler(store db.DB, ip *catalog.ImageProcessor) http.HandlerFunc {
func ImageHandler(store db.DB) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
l := logger.FromContext(r.Context())
size := chi.URLParam(r, "size")
@ -31,7 +31,7 @@ func ImageHandler(store db.DB, ip *catalog.ImageProcessor) http.HandlerFunc {
imgid, err := uuid.Parse(filename)
if err != nil {
serveDefaultImage(w, r, imageSize, ip)
serveDefaultImage(w, r, imageSize)
return
}
@ -51,7 +51,7 @@ func ImageHandler(store db.DB, ip *catalog.ImageProcessor) http.HandlerFunc {
if _, err = os.Stat(fullSizePath); os.IsNotExist(err) {
if _, err = os.Stat(largeSizePath); os.IsNotExist(err) {
l.Warn().Msgf("Could not find requested image %s. If this image is tied to an album or artist, it should be replaced", imgid.String())
serveDefaultImage(w, r, imageSize, ip)
serveDefaultImage(w, r, imageSize)
return
} else if err != nil {
// non-not found error for full file
@ -80,7 +80,7 @@ func ImageHandler(store db.DB, ip *catalog.ImageProcessor) http.HandlerFunc {
return
}
err = ip.EnqueueCompressAndSave(r.Context(), imgid.String(), imageSize, bytes.NewReader(imageBuf))
err = catalog.CompressAndSaveImage(r.Context(), imgid.String(), imageSize, bytes.NewReader(imageBuf))
if err != nil {
l.Err(err).Msg("Failed to save compressed image to cache")
}
@ -96,7 +96,7 @@ func ImageHandler(store db.DB, ip *catalog.ImageProcessor) http.HandlerFunc {
}
}
func serveDefaultImage(w http.ResponseWriter, r *http.Request, size catalog.ImageSize, ip *catalog.ImageProcessor) {
func serveDefaultImage(w http.ResponseWriter, r *http.Request, size catalog.ImageSize) {
var lock sync.Mutex
l := logger.FromContext(r.Context())
defaultImagePath := filepath.Join(cfg.ConfigDir(), catalog.ImageCacheDir, string(size), "default_img")
@ -127,7 +127,7 @@ func serveDefaultImage(w http.ResponseWriter, r *http.Request, size catalog.Imag
w.WriteHeader(http.StatusInternalServerError)
return
}
err = ip.EnqueueCompressAndSave(r.Context(), "default_img", size, file)
err = catalog.CompressAndSaveImage(r.Context(), "default_img", size, file)
if err != nil {
l.Err(err).Msg("Error when caching default img at desired size")
w.WriteHeader(http.StatusInternalServerError)

6
engine/handlers/replace_image.go
View file

@ -20,7 +20,7 @@ type ReplaceImageResponse struct {
Message string `json:"message,omitempty"`
}
func ReplaceImageHandler(store db.DB, ip *catalog.ImageProcessor) http.HandlerFunc {
func ReplaceImageHandler(store db.DB) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
l := logger.FromContext(ctx)
@ -80,7 +80,7 @@ func ReplaceImageHandler(store db.DB, ip *catalog.ImageProcessor) http.HandlerFu
dlSize = catalog.ImageSizeLarge
}
l.Debug().Msg("Downloading album image from source...")
err = ip.EnqueueDownloadAndCache(ctx, id, fileUrl, dlSize)
err = catalog.DownloadAndCacheImage(ctx, id, fileUrl, dlSize)
if err != nil {
l.Err(err).Msg("Failed to cache image")
}
@ -120,7 +120,7 @@ func ReplaceImageHandler(store db.DB, ip *catalog.ImageProcessor) http.HandlerFu
dlSize = catalog.ImageSizeLarge
}
err = ip.EnqueueCompressAndSave(ctx, id.String(), dlSize, file)
err = catalog.CompressAndSaveImage(ctx, id.String(), dlSize, file)
if err != nil {
utils.WriteError(w, "Could not save file", http.StatusInternalServerError)
return

6
engine/routes.go
View file

@ -10,7 +10,6 @@ import (
"github.com/gabehf/koito/engine/handlers"
"github.com/gabehf/koito/engine/middleware"
"github.com/gabehf/koito/internal/catalog"
"github.com/gabehf/koito/internal/cfg"
"github.com/gabehf/koito/internal/db"
mbz "github.com/gabehf/koito/internal/mbz"
@ -25,11 +24,10 @@ func bindRoutes(
ready *atomic.Bool,
db db.DB,
mbz mbz.MusicBrainzCaller,
ip *catalog.ImageProcessor,
) {
r.With(chimiddleware.RequestSize(5<<20)).
With(middleware.AllowedHosts).
Get("/images/{size}/{filename}", handlers.ImageHandler(db, ip))
Get("/images/{size}/{filename}", handlers.ImageHandler(db))
r.Route("/apis/web/v1", func(r chi.Router) {
r.Use(middleware.AllowedHosts)
@ -67,7 +65,7 @@ func bindRoutes(
r.Group(func(r chi.Router) {
r.Use(middleware.ValidateSession(db))
r.Post("/replace-image", handlers.ReplaceImageHandler(db, ip))
r.Post("/replace-image", handlers.ReplaceImageHandler(db))
r.Post("/merge/tracks", handlers.MergeTracksHandler(db))
r.Post("/merge/albums", handlers.MergeReleaseGroupsHandler(db))
r.Post("/merge/artists", handlers.MergeArtistsHandler(db))

5
internal/catalog/associate_album.go
View file

@ -23,7 +23,6 @@ type AssociateAlbumOpts struct {
ReleaseName string
TrackName string // required
Mbzc mbz.MusicBrainzCaller
IP *ImageProcessor
}
func AssociateAlbum(ctx context.Context, d db.DB, opts AssociateAlbumOpts) (*models.Album, error) {
@ -134,7 +133,7 @@ func createOrUpdateAlbumWithMbzReleaseID(ctx context.Context, d db.DB, opts Asso
}
imgid = uuid.New()
l.Debug().Msg("Downloading album image from source...")
err = opts.IP.EnqueueDownloadAndCache(ctx, imgid, imgUrl, size)
err = DownloadAndCacheImage(ctx, imgid, imgUrl, size)
if err != nil {
l.Err(err).Msg("Failed to cache image")
}
@ -217,7 +216,7 @@ func matchAlbumByTitle(ctx context.Context, d db.DB, opts AssociateAlbumOpts) (*
}
imgid = uuid.New()
l.Debug().Msg("Downloading album image from source...")
err = opts.IP.EnqueueDownloadAndCache(ctx, imgid, imgUrl, size)
err = DownloadAndCacheImage(ctx, imgid, imgUrl, size)
if err != nil {
l.Err(err).Msg("Failed to cache image")
}

21
internal/catalog/associate_artists.go
View file

@ -22,7 +22,6 @@ type AssociateArtistsOpts struct {
ArtistName string
TrackTitle string
Mbzc mbz.MusicBrainzCaller
IP *ImageProcessor
}
func AssociateArtists(ctx context.Context, d db.DB, opts AssociateArtistsOpts) ([]*models.Artist, error) {
@ -41,7 +40,7 @@ func AssociateArtists(ctx context.Context, d db.DB, opts AssociateArtistsOpts) (
if len(opts.ArtistNames) > len(result) {
l.Debug().Msg("Associating artists by list of artist names")
nameMatches, err := matchArtistsByNames(ctx, opts.ArtistNames, result, d, opts)
nameMatches, err := matchArtistsByNames(ctx, opts.ArtistNames, result, d)
if err != nil {
return nil, err
}
@ -51,7 +50,7 @@ func AssociateArtists(ctx context.Context, d db.DB, opts AssociateArtistsOpts) (
if len(result) < 1 {
allArtists := slices.Concat(opts.ArtistNames, ParseArtists(opts.ArtistName, opts.TrackTitle))
l.Debug().Msgf("Associating artists by artist name(s) %v and track title '%s'", allArtists, opts.TrackTitle)
fallbackMatches, err := matchArtistsByNames(ctx, allArtists, nil, d, opts)
fallbackMatches, err := matchArtistsByNames(ctx, allArtists, nil, d)
if err != nil {
return nil, err
}
@ -68,7 +67,7 @@ func matchArtistsByMBID(ctx context.Context, d db.DB, opts AssociateArtistsOpts)
for _, id := range opts.ArtistMbzIDs {
if id == uuid.Nil {
l.Warn().Msg("Provided artist has uuid.Nil MusicBrainzID")
return matchArtistsByNames(ctx, opts.ArtistNames, result, d, opts)
return matchArtistsByNames(ctx, opts.ArtistNames, result, d)
}
a, err := d.GetArtist(ctx, db.GetArtistOpts{
MusicBrainzID: id,
@ -86,20 +85,20 @@ func matchArtistsByMBID(ctx context.Context, d db.DB, opts AssociateArtistsOpts)
if len(opts.ArtistNames) < 1 {
opts.ArtistNames = slices.Concat(opts.ArtistNames, ParseArtists(opts.ArtistName, opts.TrackTitle))
}
a, err = resolveAliasOrCreateArtist(ctx, id, opts.ArtistNames, d, opts)
a, err = resolveAliasOrCreateArtist(ctx, id, opts.ArtistNames, d, opts.Mbzc)
if err != nil {
l.Warn().Msg("MusicBrainz unreachable, falling back to artist name matching")
return matchArtistsByNames(ctx, opts.ArtistNames, result, d, opts)
return matchArtistsByNames(ctx, opts.ArtistNames, result, d)
// return nil, err
}
result = append(result, a)
}
return result, nil
}
func resolveAliasOrCreateArtist(ctx context.Context, mbzID uuid.UUID, names []string, d db.DB, opts AssociateArtistsOpts) (*models.Artist, error) {
func resolveAliasOrCreateArtist(ctx context.Context, mbzID uuid.UUID, names []string, d db.DB, mbz mbz.MusicBrainzCaller) (*models.Artist, error) {
l := logger.FromContext(ctx)
aliases, err := opts.Mbzc.GetArtistPrimaryAliases(ctx, mbzID)
aliases, err := mbz.GetArtistPrimaryAliases(ctx, mbzID)
if err != nil {
return nil, err
}
@ -146,7 +145,7 @@ func resolveAliasOrCreateArtist(ctx context.Context, mbzID uuid.UUID, names []st
}
imgid = uuid.New()
l.Debug().Msg("Downloading artist image from source...")
err = opts.IP.EnqueueDownloadAndCache(ctx, imgid, imgUrl, size)
err = DownloadAndCacheImage(ctx, imgid, imgUrl, size)
if err != nil {
l.Err(err).Msg("Failed to cache image")
}
@ -168,7 +167,7 @@ func resolveAliasOrCreateArtist(ctx context.Context, mbzID uuid.UUID, names []st
return u, nil
}
func matchArtistsByNames(ctx context.Context, names []string, existing []*models.Artist, d db.DB, opts AssociateArtistsOpts) ([]*models.Artist, error) {
func matchArtistsByNames(ctx context.Context, names []string, existing []*models.Artist, d db.DB) ([]*models.Artist, error) {
l := logger.FromContext(ctx)
var result []*models.Artist
@ -199,7 +198,7 @@ func matchArtistsByNames(ctx context.Context, names []string, existing []*models
}
imgid = uuid.New()
l.Debug().Msg("Downloading artist image from source...")
err = opts.IP.EnqueueDownloadAndCache(ctx, imgid, imgUrl, size)
err = DownloadAndCacheImage(ctx, imgid, imgUrl, size)
if err != nil {
l.Err(err).Msg("Failed to cache image")
}

3
internal/catalog/catalog.go
View file

@ -47,7 +47,6 @@ type SubmitListenOpts struct {
Time time.Time
UserID int32
Client string
IP *ImageProcessor
}
const (
@ -70,7 +69,6 @@ func SubmitListen(ctx context.Context, store db.DB, opts SubmitListenOpts) error
ArtistName: opts.Artist,
Mbzc: opts.MbzCaller,
TrackTitle: opts.TrackTitle,
IP: opts.IP,
})
if err != nil {
l.Error().Err(err).Msg("Failed to associate artists to listen")
@ -92,7 +90,6 @@ func SubmitListen(ctx context.Context, store db.DB, opts SubmitListenOpts) error
TrackName: opts.TrackTitle,
Mbzc: opts.MbzCaller,
Artists: artists,
IP: opts.IP,
})
if err != nil {
l.Error().Err(err).Msg("Failed to associate release group to listen")

116
internal/catalog/images.go
View file

@ -3,7 +3,6 @@ package catalog
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"net/http"
@ -11,8 +10,6 @@ import (
"path"
"path/filepath"
"strings"
"sync"
"time"
"github.com/gabehf/koito/internal/cfg"
"github.com/gabehf/koito/internal/db"
@ -33,93 +30,6 @@ const (
ImageCacheDir = "image_cache"
)
type imageJob struct {
ctx context.Context
id string
size ImageSize
url string // optional
reader io.Reader // optional
}
// ImageProcessor manages a single goroutine to process image jobs sequentially
type ImageProcessor struct {
jobs chan imageJob
wg sync.WaitGroup
closing chan struct{}
}
// NewImageProcessor creates an ImageProcessor and starts the worker goroutine
func NewImageProcessor(buffer int) *ImageProcessor {
ip := &ImageProcessor{
jobs: make(chan imageJob, buffer),
closing: make(chan struct{}),
}
ip.wg.Add(1)
go ip.worker()
return ip
}
func (ip *ImageProcessor) worker() {
for {
select {
case job := <-ip.jobs:
var err error
if job.reader != nil {
err = ip.compressAndSave(job.ctx, job.id, job.size, job.reader)
} else {
err = ip.downloadCompressAndSave(job.ctx, job.id, job.url, job.size)
}
if err != nil {
logger.FromContext(job.ctx).Err(err).Msg("Image processing failed")
}
case <-ip.closing:
return
}
}
}
func (ip *ImageProcessor) EnqueueDownloadAndCache(ctx context.Context, id uuid.UUID, url string, size ImageSize) error {
return ip.enqueueJob(imageJob{ctx: ctx, id: id.String(), size: size, url: url})
}
func (ip *ImageProcessor) EnqueueCompressAndSave(ctx context.Context, id string, size ImageSize, reader io.Reader) error {
return ip.enqueueJob(imageJob{ctx: ctx, id: id, size: size, reader: reader})
}
func (ip *ImageProcessor) WaitForIdle(timeout time.Duration) error {
timer := time.NewTimer(timeout)
defer timer.Stop()
for {
if len(ip.jobs) == 0 {
return nil
}
select {
case <-time.After(10 * time.Millisecond):
case <-timer.C:
return errors.New("image processor did not become idle in time")
}
}
}
func (ip *ImageProcessor) enqueueJob(job imageJob) error {
select {
case ip.jobs <- job:
return nil
case <-job.ctx.Done():
return job.ctx.Err()
case <-ip.closing:
return errors.New("image processor closed")
}
}
// Close stops the worker and waits for any ongoing processing to finish
func (ip *ImageProcessor) Close() {
close(ip.closing)
ip.wg.Wait()
close(ip.jobs)
}
func ParseImageSize(size string) (ImageSize, error) {
switch strings.ToLower(size) {
case "small":
@ -136,7 +46,7 @@ func ParseImageSize(size string) (ImageSize, error) {
return "", fmt.Errorf("unknown image size: %s", size)
}
}
func getImageSize(size ImageSize) int {
func GetImageSize(size ImageSize) int {
var px int
switch size {
case "small":
@ -178,7 +88,9 @@ func ValidateImageURL(url string) error {
return nil
}
func (ip *ImageProcessor) downloadCompressAndSave(ctx context.Context, id string, url string, size ImageSize) error {
// DownloadAndCacheImage downloads an image from the given URL, then calls CompressAndSaveImage.
func DownloadAndCacheImage(ctx context.Context, id uuid.UUID, url string, size ImageSize) error {
l := logger.FromContext(ctx)
err := ValidateImageURL(url)
if err != nil {
@ -187,7 +99,7 @@ func (ip *ImageProcessor) downloadCompressAndSave(ctx context.Context, id string
l.Debug().Msgf("Downloading image for ID %s", id)
resp, err := http.Get(url)
if err != nil {
return err
return fmt.Errorf("failed to download image: %w", err)
}
defer resp.Body.Close()
@ -195,28 +107,28 @@ func (ip *ImageProcessor) downloadCompressAndSave(ctx context.Context, id string
return fmt.Errorf("failed to download image, status code: %d", resp.StatusCode)
}
return ip.compressAndSave(ctx, id, size, resp.Body)
return CompressAndSaveImage(ctx, id.String(), size, resp.Body)
}
func (ip *ImageProcessor) compressAndSave(ctx context.Context, filename string, size ImageSize, body io.Reader) error {
// Compresses an image to the specified size, then saves it to the correct cache folder.
func CompressAndSaveImage(ctx context.Context, filename string, size ImageSize, body io.Reader) error {
l := logger.FromContext(ctx)
if size == ImageSizeFull {
l.Debug().Msg("Full size image desired, skipping compression")
return ip.saveImage(filename, size, body)
return saveImage(filename, size, body)
}
l.Debug().Msg("Creating resized image")
compressed, err := ip.compressImage(size, body)
compressed, err := compressImage(size, body)
if err != nil {
return err
}
return ip.saveImage(filename, size, compressed)
return saveImage(filename, size, compressed)
}
// SaveImage saves an image to the image_cache/{size} folder
func (ip *ImageProcessor) saveImage(filename string, size ImageSize, data io.Reader) error {
func saveImage(filename string, size ImageSize, data io.Reader) error {
configDir := cfg.ConfigDir()
cacheDir := filepath.Join(configDir, ImageCacheDir)
@ -243,12 +155,12 @@ func (ip *ImageProcessor) saveImage(filename string, size ImageSize, data io.Rea
return nil
}
func (ip *ImageProcessor) compressImage(size ImageSize, data io.Reader) (io.Reader, error) {
func compressImage(size ImageSize, data io.Reader) (io.Reader, error) {
imgBytes, err := io.ReadAll(data)
if err != nil {
return nil, err
}
px := getImageSize(size)
px := GetImageSize(size)
// Resize with bimg
imgBytes, err = bimg.NewImage(imgBytes).Process(bimg.Options{
Width: px,

49
internal/catalog/images_test.go
View file

@ -2,13 +2,11 @@ package catalog_test
import (
"context"
"fmt"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"testing"
"time"
"github.com/gabehf/koito/internal/catalog"
"github.com/gabehf/koito/internal/cfg"
@ -19,8 +17,6 @@ import (
func TestImageLifecycle(t *testing.T) {
ip := catalog.NewImageProcessor(1)
// serve yuu.jpg as test image
imageBytes, err := os.ReadFile(filepath.Join("static", "yuu.jpg"))
require.NoError(t, err)
@ -33,59 +29,46 @@ func TestImageLifecycle(t *testing.T) {
imgID := uuid.New()
err = ip.EnqueueDownloadAndCache(context.Background(), imgID, server.URL, catalog.ImageSizeFull)
err = catalog.DownloadAndCacheImage(context.Background(), imgID, server.URL, catalog.ImageSizeFull)
require.NoError(t, err)
err = ip.EnqueueDownloadAndCache(context.Background(), imgID, server.URL, catalog.ImageSizeMedium)
err = catalog.DownloadAndCacheImage(context.Background(), imgID, server.URL, catalog.ImageSizeMedium)
require.NoError(t, err)
ip.WaitForIdle(5 * time.Second)
// ensure download is correct
imagePath := filepath.Join(cfg.ConfigDir(), catalog.ImageCacheDir, "full", imgID.String())
assert.NoError(t, waitForFile(imagePath, 1*time.Second))
_, err = os.Stat(imagePath)
assert.NoError(t, err)
imagePath = filepath.Join(cfg.ConfigDir(), catalog.ImageCacheDir, "medium", imgID.String())
assert.NoError(t, waitForFile(imagePath, 1*time.Second))
_, err = os.Stat(imagePath)
assert.NoError(t, err)
assert.NoError(t, catalog.DeleteImage(imgID))
// ensure delete works
imagePath = filepath.Join(cfg.ConfigDir(), catalog.ImageCacheDir, "full", imgID.String())
assert.Error(t, waitForFile(imagePath, 1*time.Second))
_, err = os.Stat(imagePath)
assert.Error(t, err)
imagePath = filepath.Join(cfg.ConfigDir(), catalog.ImageCacheDir, "medium", imgID.String())
assert.Error(t, waitForFile(imagePath, 1*time.Second))
_, err = os.Stat(imagePath)
assert.Error(t, err)
// re-download for prune
err = ip.EnqueueDownloadAndCache(context.Background(), imgID, server.URL, catalog.ImageSizeFull)
err = catalog.DownloadAndCacheImage(context.Background(), imgID, server.URL, catalog.ImageSizeFull)
require.NoError(t, err)
err = ip.EnqueueDownloadAndCache(context.Background(), imgID, server.URL, catalog.ImageSizeMedium)
err = catalog.DownloadAndCacheImage(context.Background(), imgID, server.URL, catalog.ImageSizeMedium)
require.NoError(t, err)
ip.WaitForIdle(5 * time.Second)
assert.NoError(t, catalog.PruneOrphanedImages(context.Background(), store))
// ensure prune works
imagePath = filepath.Join(cfg.ConfigDir(), catalog.ImageCacheDir, "full", imgID.String())
assert.Error(t, waitForFile(imagePath, 1*time.Second))
_, err = os.Stat(imagePath)
assert.Error(t, err)
imagePath = filepath.Join(cfg.ConfigDir(), catalog.ImageCacheDir, "medium", imgID.String())
assert.Error(t, waitForFile(imagePath, 1*time.Second))
}
func waitForFile(path string, timeout time.Duration) error {
deadline := time.Now().Add(timeout)
for {
if _, err := os.Stat(path); err == nil {
return nil
} else if !os.IsNotExist(err) {
return err
}
if time.Now().After(deadline) {
return fmt.Errorf("timed out waiting for %s", path)
}
time.Sleep(20 * time.Millisecond)
}
_, err = os.Stat(imagePath)
assert.Error(t, err)
}

4
internal/images/imagesrc.go
View file

@ -51,6 +51,10 @@ func Initialize(opts ImageSourceOpts) {
})
}
func Shutdown() {
imgsrc.deezerC.Shutdown()
}
func GetArtistImage(ctx context.Context, opts ArtistImageOpts) (string, error) {
l := logger.FromContext(ctx)
if imgsrc.deezerC != nil {