diff --git a/engine/engine.go b/engine/engine.go index a7133b9..ba9bfef 100644 --- a/engine/engine.go +++ b/engine/engine.go @@ -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(), diff --git a/engine/handlers/image_handler.go b/engine/handlers/image_handler.go index 0ce5b81..cbbc85d 100644 --- a/engine/handlers/image_handler.go +++ b/engine/handlers/image_handler.go @@ -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) diff --git a/engine/handlers/replace_image.go b/engine/handlers/replace_image.go index 8a33e2f..5713772 100644 --- a/engine/handlers/replace_image.go +++ b/engine/handlers/replace_image.go @@ -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 diff --git a/engine/routes.go b/engine/routes.go index 732fadd..b9a9c1b 100644 --- a/engine/routes.go +++ b/engine/routes.go @@ -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)) diff --git a/internal/catalog/associate_album.go b/internal/catalog/associate_album.go index af39152..1391870 100644 --- a/internal/catalog/associate_album.go +++ b/internal/catalog/associate_album.go @@ -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") } diff --git a/internal/catalog/associate_artists.go b/internal/catalog/associate_artists.go index 0014b3e..21393b6 100644 --- a/internal/catalog/associate_artists.go +++ b/internal/catalog/associate_artists.go @@ -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") } diff --git a/internal/catalog/catalog.go b/internal/catalog/catalog.go index 495104d..b209635 100644 --- a/internal/catalog/catalog.go +++ b/internal/catalog/catalog.go @@ -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") diff --git a/internal/catalog/images.go b/internal/catalog/images.go index d93ac46..2e61c3c 100644 --- a/internal/catalog/images.go +++ b/internal/catalog/images.go @@ -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, diff --git a/internal/catalog/images_test.go b/internal/catalog/images_test.go index e0077e9..256510e 100644 --- a/internal/catalog/images_test.go +++ b/internal/catalog/images_test.go @@ -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) + } }