ds-galo/backend/internal/photos/worker.go

package photos

import (
	"bytes"
	"context"
	"errors"
	"log/slog"
	"sync"
	"time"

	"galo/internal/auth"
)

const maxVariantAttempts = 5

// Worker is the background reconciler. It drives the variant generation queue
// and the object garbage collector entirely from the state recorded in the
// metadata layer, so it resumes correctly after a restart.
type Worker struct {
	svc         *Service
	interval    time.Duration
	batch       int
	pendingTTL  time.Duration
	lease       time.Duration
	concurrency int
}

// NewWorker builds a worker. Parameters are clamped to sane defaults.
func NewWorker(svc *Service, interval time.Duration, batch int, pendingTTL time.Duration, concurrency int) *Worker {
	if interval <= 0 {
		interval = 5 * time.Second
	}
	if batch <= 0 {
		batch = 10
	}
	if pendingTTL <= 0 {
		pendingTTL = 10 * time.Minute
	}
	if concurrency <= 0 {
		concurrency = 4
	}
	return &Worker{
		svc:         svc,
		interval:    interval,
		batch:       batch,
		pendingTTL:  pendingTTL,
		lease:       interval * 4,
		concurrency: concurrency,
	}
}

// Run loops until the context is cancelled.
func (w *Worker) Run(ctx context.Context) {
	slog.Info("background worker started", "interval", w.interval)
	defer slog.Info("background worker stopped")
	ticker := time.NewTicker(w.interval)
	defer ticker.Stop()
	for {
		w.tick(ctx)
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

func (w *Worker) tick(ctx context.Context) {
	w.processVariantJobs(ctx)
	w.collectGarbage(ctx)
	w.reapStalePending(ctx)
}

func (w *Worker) processVariantJobs(ctx context.Context) {
	ids, err := w.svc.store.ClaimVariantJobs(ctx, w.batch, w.lease)
	if err != nil {
		slog.Error("claim variant jobs", "error", err)
		return
	}
	w.forEachConcurrent(ids, func(id string) {
		w.generateVariants(ctx, id)
	})
}

// forEachConcurrent runs fn over ids with at most w.concurrency functions in
// flight. Generation is CPU-bound (decode + resize + encode), so parallelism
// scales with cores; a concurrency of 1 stays strictly sequential.
func (w *Worker) forEachConcurrent(ids []string, fn func(string)) {
	if len(ids) == 0 {
		return
	}
	if w.concurrency <= 1 {
		for _, id := range ids {
			fn(id)
		}
		return
	}
	sem := make(chan struct{}, w.concurrency)
	var wg sync.WaitGroup
	for _, id := range ids {
		wg.Add(1)
		sem <- struct{}{}
		go func(id string) {
			defer wg.Done()
			defer func() { <-sem }()
			fn(id)
		}(id)
	}
	wg.Wait()
}

func (w *Worker) generateVariants(ctx context.Context, photoID string) {
	orig, err := w.svc.store.OriginalObject(ctx, photoID)
	if err != nil {
		// The photo or its original binding is gone; nothing to do.
		_ = w.svc.store.CompleteVariantJob(ctx, photoID)
		return
	}
	if orig.State != ObjectReady {
		w.fail(ctx, photoID, "original object is not ready")
		return
	}
	data, err := w.svc.storage.Get(ctx, orig.StorageKey)
	if err != nil {
		w.fail(ctx, photoID, "fetch original: "+err.Error())
		return
	}
	thumb, preview, err := buildVariants(data)
	if err != nil {
		w.fail(ctx, photoID, "build variants: "+err.Error())
		return
	}
	siblings, err := w.svc.store.PhotosByOriginalObject(ctx, orig.ID)
	if err != nil || len(siblings) == 0 {
		siblings = []string{photoID}
	}
	if err := w.createVariant(ctx, siblings, RoleThumb, thumb); err != nil {
		w.fail(ctx, photoID, "store thumb: "+err.Error())
		return
	}
	if err := w.createVariant(ctx, siblings, RolePreview, preview); err != nil {
		w.fail(ctx, photoID, "store preview: "+err.Error())
		return
	}
	_ = w.svc.store.CompleteVariantJob(ctx, photoID)
	for _, pid := range siblings {
		w.notify(ctx, pid)
	}
}

// createVariant stores one derived object and binds it to every photo sharing
// the original, so copies reuse the same physical variant.
func (w *Worker) createVariant(ctx context.Context, photoIDs []string, role string, data []byte) error {
	objID := auth.NewID()
	key := objectKey(objID, role, ".webp")
	obj := &Object{
		ID:         objID,
		StorageKey: key,
		MimeType:   "image/webp",
		Size:       int64(len(data)),
		Checksum:   checksum(data),
	}
	if err := w.svc.store.CreateObject(ctx, obj); err != nil {
		return err
	}
	if err := w.svc.storage.Put(ctx, key, bytes.NewReader(data), int64(len(data)), "image/webp"); err != nil {
		_ = w.svc.store.DeleteObjectRow(ctx, objID)
		return err
	}
	if err := w.svc.store.ActivateObject(ctx, objID); err != nil {
		return err
	}
	bound := false
	for _, pid := range photoIDs {
		if err := w.svc.store.BindObject(ctx, pid, objID, role); err != nil {
			if errors.Is(err, ErrObjectNotReady) || errors.Is(err, ErrObjectNotFound) {
				continue
			}
			return err
		}
		bound = true
	}
	if !bound {
		// No photo could take the variant; let the collector reclaim it.
		_ = w.svc.store.MarkObjectDeleting(ctx, objID)
	}
	return nil
}

func (w *Worker) notify(ctx context.Context, photoID string) {
	p, err := w.svc.store.GetByID(ctx, photoID, "")
	if err != nil {
		return
	}
	if err := w.svc.attachURLs(ctx, p); err != nil {
		return
	}
	// Variant readiness updates an existing photo; it is not a new photo in the
	// album, so clients update it in place instead of inserting at the top.
	w.svc.hub.Publish(p.AlbumID, "photo.updated", p)
}

func (w *Worker) fail(ctx context.Context, photoID, msg string) {
	slog.Warn("variant job failed", "photo", photoID, "error", msg)
	_ = w.svc.store.FailVariantJob(ctx, photoID, msg, maxVariantAttempts)
}

func (w *Worker) collectGarbage(ctx context.Context) {
	objs, err := w.svc.store.ClaimGarbage(ctx, w.batch)
	if err != nil {
		slog.Error("claim garbage", "error", err)
		return
	}
	for _, o := range objs {
		w.deleteObject(ctx, o)
	}
}

func (w *Worker) reapStalePending(ctx context.Context) {
	objs, err := w.svc.store.ClaimStalePending(ctx, w.pendingTTL, w.batch)
	if err != nil {
		slog.Error("claim stale pending", "error", err)
		return
	}
	for _, o := range objs {
		w.deleteObject(ctx, o)
	}
}

// deleteObject removes an object's blob then its row. If a binding appeared
// meanwhile the object is returned to ready; if the storage delete fails the
// row stays deleting and is retried on the next pass.
func (w *Worker) deleteObject(ctx context.Context, o Object) {
	if n, err := w.svc.store.ObjectRefCount(ctx, o.ID); err == nil && n > 0 {
		_ = w.svc.store.SetObjectState(ctx, o.ID, ObjectReady)
		return
	}
	if err := w.svc.storage.Delete(ctx, o.StorageKey); err != nil {
		slog.Warn("delete object blob", "object", o.ID, "error", err)
		return
	}
	if err := w.svc.store.DeleteObjectRow(ctx, o.ID); err != nil && !errors.Is(err, ErrObjectNotFound) {
		slog.Warn("delete object row", "object", o.ID, "error", err)
	}
}