Spanglish/photo-api/internal/worker/worker.go

// Package worker turns queued photos_photos rows into ready ones by
// generating the thumb/preview variants. The rows themselves are the queue
// (status/attempts/next_attempt_at) — no Redis, no jobs table. Claims are
// optimistic UPDATEs so they are safe on both Postgres and SQLite.
package worker

import (
	"context"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"strings"
	"time"

	"git.azzamo.net/Michilis/Spanglish/photo-api/internal/imaging"
	"git.azzamo.net/Michilis/Spanglish/photo-api/internal/storage"
	"git.azzamo.net/Michilis/Spanglish/photo-api/internal/store"
)

const (
	pollInterval = 10 * time.Second
	stuckAfter   = 10 * time.Minute
)

type Worker struct {
	db       *store.DB
	storage  storage.Storage
	heic     *imaging.HeicConverter
	tmpDir   string
	nudge    chan struct{}
	parallel int
}

func New(db *store.DB, st storage.Storage, heic *imaging.HeicConverter, tmpDir string, parallel int) *Worker {
	if parallel < 1 {
		parallel = 1
	}
	return &Worker{
		db:       db,
		storage:  st,
		heic:     heic,
		tmpDir:   tmpDir,
		nudge:    make(chan struct{}, 1),
		parallel: parallel,
	}
}

// Nudge wakes the worker after an upload so the poll interval is only a
// fallback.
func (w *Worker) Nudge() {
	select {
	case w.nudge <- struct{}{}:
	default:
	}
}

func (w *Worker) Run(ctx context.Context) {
	if n, err := w.db.RecoverStuckProcessing(ctx, stuckAfter); err != nil {
		log.Printf("worker: recover stuck: %v", err)
	} else if n > 0 {
		log.Printf("worker: requeued %d photos stuck in processing", n)
	}
	for i := 0; i < w.parallel; i++ {
		go w.loop(ctx)
	}
}

func (w *Worker) loop(ctx context.Context) {
	for {
		worked := w.drain(ctx)
		if ctx.Err() != nil {
			return
		}
		if worked {
			continue
		}
		select {
		case <-ctx.Done():
			return
		case <-w.nudge:
		case <-time.After(pollInterval):
		}
	}
}

// drain processes until the queue is empty; returns whether anything ran.
func (w *Worker) drain(ctx context.Context) bool {
	worked := false
	for ctx.Err() == nil {
		photo, err := w.db.ClaimNextPhoto(ctx)
		if err == store.ErrNotFound {
			return worked
		}
		if err != nil {
			log.Printf("worker: claim: %v", err)
			return worked
		}
		worked = true
		if err := w.process(ctx, photo); err != nil {
			log.Printf("worker: photo %s attempt %d failed: %v", photo.ID, photo.Attempts, err)
			if dberr := w.db.MarkPhotoFailed(ctx, photo.ID, photo.Attempts, err); dberr != nil {
				log.Printf("worker: mark failed: %v", dberr)
			}
		}
	}
	return worked
}

func (w *Worker) process(ctx context.Context, p store.Photo) error {
	work, err := os.MkdirTemp(w.tmpDir, "photo-*")
	if err != nil {
		return err
	}
	defer os.RemoveAll(work)

	// 1. Fetch the original to a local file.
	src := filepath.Join(work, "original")
	if err := w.download(ctx, p.OriginalKey, src); err != nil {
		return fmt.Errorf("fetch original: %w", err)
	}

	// 2. HEIC → JPEG via the external converter before the pure-Go pipeline.
	if p.ContentType == "image/heic" {
		if w.heic == nil {
			return fmt.Errorf("HEIC converter not installed on this host (install libvips-tools or libheif-examples)")
		}
		converted := filepath.Join(work, "converted.jpg")
		if err := w.heic.ToJPEG(src, converted); err != nil {
			return err
		}
		src = converted
	}

	// 3. Decode, orient, resize, encode.
	thumbPath := filepath.Join(work, "thumb.jpg")
	previewPath := filepath.Join(work, "preview.jpg")
	res, err := imaging.ProcessFile(src, thumbPath, previewPath)
	if err != nil {
		return err
	}

	// 4. Store variants next to the original's key.
	base := strings.TrimSuffix(filepath.Base(p.OriginalKey), filepath.Ext(p.OriginalKey))
	prefix := fmt.Sprintf("galleries/%s", p.GalleryID)
	thumbKey := fmt.Sprintf("%s/thumb/%s.jpg", prefix, base)
	previewKey := fmt.Sprintf("%s/preview/%s.jpg", prefix, base)
	if err := w.upload(ctx, thumbKey, thumbPath); err != nil {
		return fmt.Errorf("store thumb: %w", err)
	}
	if err := w.upload(ctx, previewKey, previewPath); err != nil {
		return fmt.Errorf("store preview: %w", err)
	}

	return w.db.MarkPhotoReady(ctx, p.ID, thumbKey, previewKey, res.Width, res.Height, res.TakenAt)
}

func (w *Worker) download(ctx context.Context, key, dst string) error {
	r, _, err := w.storage.Open(ctx, key)
	if err != nil {
		return err
	}
	defer r.Close()
	f, err := os.Create(dst)
	if err != nil {
		return err
	}
	defer f.Close()
	_, err = io.Copy(f, r)
	return err
}

func (w *Worker) upload(ctx context.Context, key, src string) error {
	f, err := os.Open(src)
	if err != nil {
		return err
	}
	defer f.Close()
	info, err := f.Stat()
	if err != nil {
		return err
	}
	return w.storage.Put(ctx, key, f, info.Size(), "image/jpeg")
}