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

package photos

import (
	"context"
	"testing"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"galo/internal/config"
	"galo/internal/storage"
	"galo/internal/video"
)

func TestWorkerGeneratesVariantsAndCollects(t *testing.T) {
	pool := testPool(t)
	svc, cl := newTestService(t, pool)
	runWorkerFlow(t, pool, svc, cl, nil)
}

func TestWorkerGeneratesVariantsWithFS(t *testing.T) {
	pool := testPool(t)
	svc, cl, albums := newRecordingServiceFS(t, pool)
	runWorkerFlow(t, pool, svc, cl, albums)
}

func runWorkerFlow(t *testing.T, pool *pgxpool.Pool, svc *Service, cl storage.Backend, albums *recordingAlbums) {
	t.Helper()
	ctx := context.Background()
	user := createUser(t, ctx, pool)
	album := createAlbum(t, ctx, pool, user)

	p, err := svc.Upload(ctx, UploadInput{
		AlbumID:  album,
		UserID:   user,
		Filename: "photo.png",
		MimeType: "image/png",
		Data:     makePNG(t),
	})
	if err != nil {
		t.Fatalf("upload: %v", err)
	}
	t.Cleanup(func() { cleanupPhoto(t, ctx, pool, cl, p.ID) })

	sub := svc.hub.SubscribeAlbum(album)
	defer sub.Close()

	w := NewWorker(svc, time.Second, 10, time.Minute, 2)
	w.processVariantJobs(ctx)

	select {
	case ev := <-sub.Events():
		if ev.Type != "photo.updated" {
			t.Fatalf("variant-ready event = %q, want photo.updated", ev.Type)
		}
	case <-time.After(time.Second):
		t.Fatalf("no variant-ready event published")
	}

	// Variant readiness must also announce the album to personal streams, because
	// the album object only gains a cover link once the variants exist.
	if albums != nil && !albums.coverReady(album, p.ID) {
		t.Fatalf("cover readiness was not announced for the uploaded photo")
	}

	refreshed, err := svc.store.GetByID(ctx, p.ID, "")
	if err != nil {
		t.Fatalf("get refreshed: %v", err)
	}
	if refreshed.ThumbKey == "" || refreshed.MediumKey == "" || refreshed.PreviewKey == "" {
		t.Fatalf("variants not bound: thumb=%q medium=%q preview=%q", refreshed.ThumbKey, refreshed.MediumKey, refreshed.PreviewKey)
	}
	if err := svc.attachURLs(ctx, refreshed); err != nil {
		t.Fatalf("attach urls: %v", err)
	}
	if refreshed.ThumbURL == "" || refreshed.MediumURL == "" || refreshed.PreviewURL == "" {
		t.Fatalf("variant urls not set")
	}

	var jobs int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM variant_jobs WHERE photo_id = $1`, p.ID).Scan(&jobs); err != nil {
		t.Fatalf("count jobs: %v", err)
	}
	if jobs != 0 {
		t.Fatalf("variant job not completed: %d left", jobs)
	}

	objectIDs, err := svc.store.PhotoObjectIDs(ctx, p.ID)
	if err != nil || len(objectIDs) != 4 {
		t.Fatalf("expected 4 bound objects, got %v err=%v", objectIDs, err)
	}

	if err := svc.PermanentDelete(ctx, p.ID, user); err != nil {
		t.Fatalf("permanent delete: %v", err)
	}
	// Run the collector until the objects are gone (bounded to avoid hangs).
	for attempt := 0; attempt < 100 && countObjects(t, ctx, pool, objectIDs) > 0; attempt++ {
		w.collectGarbage(ctx)
	}
	if n := countObjects(t, ctx, pool, objectIDs); n != 0 {
		t.Fatalf("%d objects were not collected", n)
	}
}

func TestWorkerConcurrencyPool(t *testing.T) {
	w := &Worker{concurrency: 3}
	ids := []string{"a", "b", "c", "d", "e"}
	started := make(chan struct{}, len(ids))
	release := make(chan struct{})
	done := make(chan struct{})
	go func() {
		w.forEachConcurrent(ids, func(string) {
			started <- struct{}{}
			<-release
		})
		close(done)
	}()

	for i := 0; i < 3; i++ {
		select {
		case <-started:
		case <-time.After(2 * time.Second):
			t.Fatalf("only %d tasks started concurrently, want 3", i)
		}
	}
	select {
	case <-started:
		t.Fatalf("more than concurrency tasks started concurrently")
	case <-time.After(50 * time.Millisecond):
	}

	close(release)
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatalf("pool did not finish")
	}
}

func TestWorkerConcurrencySequentialAtOne(t *testing.T) {
	w := &Worker{concurrency: 1}
	var order []string
	w.forEachConcurrent([]string{"a", "b", "c"}, func(id string) {
		order = append(order, id)
	})
	if len(order) != 3 || order[0] != "a" || order[1] != "b" || order[2] != "c" {
		t.Fatalf("sequential order = %v, want [a b c]", order)
	}
}

func TestProcessVideoJobsSkipsWithoutTools(t *testing.T) {
	// With no toolchain the video pass must return before touching the store,
	// so a nil service is safe here.
	w := NewWorker(&Service{}, time.Second, 10, time.Minute, 2)
	w.processVideoJobs(context.Background())
}

func TestSetVideoConfiguresPool(t *testing.T) {
	tools := video.Detect(config.Video{FFmpegPath: "definitely-missing-xyz", FFprobePath: "definitely-missing-xyz"})
	w := &Worker{}
	w.SetVideo(tools, 2)
	if w.videoConc != 2 {
		t.Fatalf("video concurrency = %d, want 2", w.videoConc)
	}
	if w.videoLease != videoJobLease {
		t.Fatalf("video lease = %v, want %v", w.videoLease, videoJobLease)
	}
	if tools.Available() {
		t.Fatal("missing toolchain should be unavailable")
	}
}

func countObjects(t *testing.T, ctx context.Context, pool *pgxpool.Pool, ids []string) int {
	t.Helper()
	var n int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM objects WHERE id = ANY($1)`, ids).Scan(&n); err != nil {
		t.Fatalf("count objects: %v", err)
	}
	return n
}