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

chore: push image processing to a dedicated thread

Browse source
This commit is contained in:
Gabe Farrell 2025-06-11 23:42:29 -04:00
parent a35d1391a6
commit aba2b76def
9 changed files with 168 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

4
engine/engine.go
View file

@ -101,6 +101,8 @@ func Run(
EnableDeezer: !cfg.DeezerDisabled(),
})
ip := catalog.NewImageProcessor(1)
userCount, _ := store.CountUsers(ctx)
if userCount < 1 {
l.Debug().Msg("Creating default user...")
@ -145,7 +147,7 @@ func Run(
mux.Use(chimiddleware.Recoverer)
mux.Use(chimiddleware.RealIP)
// call router binds on mux
bindRoutes(mux, &ready, store, mbzC)
bindRoutes(mux, &ready, store, mbzC, ip)
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) http.HandlerFunc {
func ImageHandler(store db.DB, ip *catalog.ImageProcessor) 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) http.HandlerFunc {
imgid, err := uuid.Parse(filename)
if err != nil {
serveDefaultImage(w, r, imageSize)
serveDefaultImage(w, r, imageSize, ip)
return
}
@ -51,7 +51,7 @@ func ImageHandler(store db.DB) 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)
serveDefaultImage(w, r, imageSize, ip)
return
} else if err != nil {
// non-not found error for full file
@ -80,7 +80,7 @@ func ImageHandler(store db.DB) http.HandlerFunc {
return
}
err = catalog.CompressAndSaveImage(r.Context(), imgid.String(), imageSize, bytes.NewReader(imageBuf))
err = ip.EnqueueCompressAndSave(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) http.HandlerFunc {
}
}
func serveDefaultImage(w http.ResponseWriter, r *http.Request, size catalog.ImageSize) {
func serveDefaultImage(w http.ResponseWriter, r *http.Request, size catalog.ImageSize, ip *catalog.ImageProcessor) {
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 = catalog.CompressAndSaveImage(r.Context(), "default_img", size, file)
err = ip.EnqueueCompressAndSave(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) http.HandlerFunc {
func ReplaceImageHandler(store db.DB, ip *catalog.ImageProcessor) 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) http.HandlerFunc {
dlSize = catalog.ImageSizeLarge
}
l.Debug().Msg("Downloading album image from source...")
err = catalog.DownloadAndCacheImage(ctx, id, fileUrl, dlSize)
err = ip.EnqueueDownloadAndCache(ctx, id, fileUrl, dlSize)
if err != nil {
l.Err(err).Msg("Failed to cache image")
}
@ -120,7 +120,7 @@ func ReplaceImageHandler(store db.DB) http.HandlerFunc {
dlSize = catalog.ImageSizeLarge
}
err = catalog.CompressAndSaveImage(ctx, id.String(), dlSize, file)
err = ip.EnqueueCompressAndSave(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,6 +10,7 @@ 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"
@ -24,10 +25,11 @@ 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))
Get("/images/{size}/{filename}", handlers.ImageHandler(db, ip))
r.Route("/apis/web/v1", func(r chi.Router) {
r.Use(middleware.AllowedHosts)
@ -65,7 +67,7 @@ func bindRoutes(
r.Group(func(r chi.Router) {
r.Use(middleware.ValidateSession(db))
r.Post("/replace-image", handlers.ReplaceImageHandler(db))
r.Post("/replace-image", handlers.ReplaceImageHandler(db, ip))
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,6 +23,7 @@ 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) {
@ -133,7 +134,7 @@ func createOrUpdateAlbumWithMbzReleaseID(ctx context.Context, d db.DB, opts Asso
}
imgid = uuid.New()
l.Debug().Msg("Downloading album image from source...")
err = DownloadAndCacheImage(ctx, imgid, imgUrl, size)
err = opts.IP.EnqueueDownloadAndCache(ctx, imgid, imgUrl, size)
if err != nil {
l.Err(err).Msg("Failed to cache image")
}
@ -216,7 +217,7 @@ func matchAlbumByTitle(ctx context.Context, d db.DB, opts AssociateAlbumOpts) (*
}
imgid = uuid.New()
l.Debug().Msg("Downloading album image from source...")
err = DownloadAndCacheImage(ctx, imgid, imgUrl, size)
err = opts.IP.EnqueueDownloadAndCache(ctx, imgid, imgUrl, size)
if err != nil {
l.Err(err).Msg("Failed to cache image")
}

21
internal/catalog/associate_artists.go
View file

@ -22,6 +22,7 @@ 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) {
@ -40,7 +41,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)
nameMatches, err := matchArtistsByNames(ctx, opts.ArtistNames, result, d, opts)
if err != nil {
return nil, err
}
@ -50,7 +51,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)
fallbackMatches, err := matchArtistsByNames(ctx, allArtists, nil, d, opts)
if err != nil {
return nil, err
}
@ -67,7 +68,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)
return matchArtistsByNames(ctx, opts.ArtistNames, result, d, opts)
}
a, err := d.GetArtist(ctx, db.GetArtistOpts{
MusicBrainzID: id,
@ -85,20 +86,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.Mbzc)
a, err = resolveAliasOrCreateArtist(ctx, id, opts.ArtistNames, d, opts)
if err != nil {
l.Warn().Msg("MusicBrainz unreachable, falling back to artist name matching")
return matchArtistsByNames(ctx, opts.ArtistNames, result, d)
return matchArtistsByNames(ctx, opts.ArtistNames, result, d, opts)
// return nil, err
}
result = append(result, a)
}
return result, nil
}
func resolveAliasOrCreateArtist(ctx context.Context, mbzID uuid.UUID, names []string, d db.DB, mbz mbz.MusicBrainzCaller) (*models.Artist, error) {
func resolveAliasOrCreateArtist(ctx context.Context, mbzID uuid.UUID, names []string, d db.DB, opts AssociateArtistsOpts) (*models.Artist, error) {
l := logger.FromContext(ctx)
aliases, err := mbz.GetArtistPrimaryAliases(ctx, mbzID)
aliases, err := opts.Mbzc.GetArtistPrimaryAliases(ctx, mbzID)
if err != nil {
return nil, err
}
@ -145,7 +146,7 @@ func resolveAliasOrCreateArtist(ctx context.Context, mbzID uuid.UUID, names []st
}
imgid = uuid.New()
l.Debug().Msg("Downloading artist image from source...")
err = DownloadAndCacheImage(ctx, imgid, imgUrl, size)
err = opts.IP.EnqueueDownloadAndCache(ctx, imgid, imgUrl, size)
if err != nil {
l.Err(err).Msg("Failed to cache image")
}
@ -167,7 +168,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) ([]*models.Artist, error) {
func matchArtistsByNames(ctx context.Context, names []string, existing []*models.Artist, d db.DB, opts AssociateArtistsOpts) ([]*models.Artist, error) {
l := logger.FromContext(ctx)
var result []*models.Artist
@ -198,7 +199,7 @@ func matchArtistsByNames(ctx context.Context, names []string, existing []*models
}
imgid = uuid.New()
l.Debug().Msg("Downloading artist image from source...")
err = DownloadAndCacheImage(ctx, imgid, imgUrl, size)
err = opts.IP.EnqueueDownloadAndCache(ctx, imgid, imgUrl, size)
if err != nil {
l.Err(err).Msg("Failed to cache image")
}

3
internal/catalog/catalog.go
View file

@ -47,6 +47,7 @@ type SubmitListenOpts struct {
Time time.Time
UserID int32
Client string
IP *ImageProcessor
}
const (
@ -69,6 +70,7 @@ 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")
@ -90,6 +92,7 @@ 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,6 +3,7 @@ package catalog
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"net/http"
@ -10,6 +11,8 @@ import (
"path"
"path/filepath"
"strings"
"sync"
"time"
"github.com/gabehf/koito/internal/cfg"
"github.com/gabehf/koito/internal/db"
@ -30,6 +33,93 @@ 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":
@ -46,7 +136,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":
@ -88,9 +178,7 @@ func ValidateImageURL(url string) error {
return nil
}
// DownloadAndCacheImage downloads an image from the given URL, then calls CompressAndSaveImage.
func DownloadAndCacheImage(ctx context.Context, id uuid.UUID, url string, size ImageSize) error {
func (ip *ImageProcessor) downloadCompressAndSave(ctx context.Context, id string, url string, size ImageSize) error {
l := logger.FromContext(ctx)
err := ValidateImageURL(url)
if err != nil {
@ -99,7 +187,7 @@ func DownloadAndCacheImage(ctx context.Context, id uuid.UUID, url string, size I
l.Debug().Msgf("Downloading image for ID %s", id)
resp, err := http.Get(url)
if err != nil {
return fmt.Errorf("failed to download image: %w", err)
return err
}
defer resp.Body.Close()
@ -107,28 +195,28 @@ func DownloadAndCacheImage(ctx context.Context, id uuid.UUID, url string, size I
return fmt.Errorf("failed to download image, status code: %d", resp.StatusCode)
}
return CompressAndSaveImage(ctx, id.String(), size, resp.Body)
return ip.compressAndSave(ctx, id, size, resp.Body)
}
// 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 {
func (ip *ImageProcessor) compressAndSave(ctx context.Context, filename string, size ImageSize, body io.Reader) error {
l := logger.FromContext(ctx)
if size == ImageSizeFull {
return saveImage(filename, size, body)
l.Debug().Msg("Full size image desired, skipping compression")
return ip.saveImage(filename, size, body)
}
l.Debug().Msg("Creating resized image")
compressed, err := compressImage(size, body)
compressed, err := ip.compressImage(size, body)
if err != nil {
return err
}
return saveImage(filename, size, compressed)
return ip.saveImage(filename, size, compressed)
}
// SaveImage saves an image to the image_cache/{size} folder
func saveImage(filename string, size ImageSize, data io.Reader) error {
func (ip *ImageProcessor) saveImage(filename string, size ImageSize, data io.Reader) error {
configDir := cfg.ConfigDir()
cacheDir := filepath.Join(configDir, ImageCacheDir)
@ -155,12 +243,12 @@ func saveImage(filename string, size ImageSize, data io.Reader) error {
return nil
}
func compressImage(size ImageSize, data io.Reader) (io.Reader, error) {
func (ip *ImageProcessor) 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,11 +2,13 @@ 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"
@ -17,6 +19,8 @@ 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)
@ -29,46 +33,59 @@ func TestImageLifecycle(t *testing.T) {
imgID := uuid.New()
err = catalog.DownloadAndCacheImage(context.Background(), imgID, server.URL, catalog.ImageSizeFull)
err = ip.EnqueueDownloadAndCache(context.Background(), imgID, server.URL, catalog.ImageSizeFull)
require.NoError(t, err)
err = catalog.DownloadAndCacheImage(context.Background(), imgID, server.URL, catalog.ImageSizeMedium)
err = ip.EnqueueDownloadAndCache(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())
_, err = os.Stat(imagePath)
assert.NoError(t, err)
assert.NoError(t, waitForFile(imagePath, 1*time.Second))
imagePath = filepath.Join(cfg.ConfigDir(), catalog.ImageCacheDir, "medium", imgID.String())
_, err = os.Stat(imagePath)
assert.NoError(t, err)
assert.NoError(t, waitForFile(imagePath, 1*time.Second))
assert.NoError(t, catalog.DeleteImage(imgID))
// ensure delete works
imagePath = filepath.Join(cfg.ConfigDir(), catalog.ImageCacheDir, "full", imgID.String())
_, err = os.Stat(imagePath)
assert.Error(t, err)
assert.Error(t, waitForFile(imagePath, 1*time.Second))
imagePath = filepath.Join(cfg.ConfigDir(), catalog.ImageCacheDir, "medium", imgID.String())
_, err = os.Stat(imagePath)
assert.Error(t, err)
assert.Error(t, waitForFile(imagePath, 1*time.Second))
// re-download for prune
err = catalog.DownloadAndCacheImage(context.Background(), imgID, server.URL, catalog.ImageSizeFull)
err = ip.EnqueueDownloadAndCache(context.Background(), imgID, server.URL, catalog.ImageSizeFull)
require.NoError(t, err)
err = catalog.DownloadAndCacheImage(context.Background(), imgID, server.URL, catalog.ImageSizeMedium)
err = ip.EnqueueDownloadAndCache(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())
_, err = os.Stat(imagePath)
assert.Error(t, err)
assert.Error(t, waitForFile(imagePath, 1*time.Second))
imagePath = filepath.Join(cfg.ConfigDir(), catalog.ImageCacheDir, "medium", imgID.String())
_, err = os.Stat(imagePath)
assert.Error(t, err)
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)
}
}