From d7a37328f0b45250fbd5cfa8371c524654078414 Mon Sep 17 00:00:00 2001 From: Iu Date: Sun, 11 Oct 2026 14:15:59 +0300 Subject: [PATCH] fix(photos): faster large-album loading and album cover lifecycle Large albums opened slowly and flickered; the default album cover was also lost in the media-objects refactor. Large-album loading: - quantize signed object URLs within a window and send Cache-Control so previews are reused instead of re-fetched - publish photo.updated (not photo.created) when variants are ready, so clients update in place without reordering the grid - worker concurrency + newest-first variant priority; raise default batch Album cover: - first uploaded photo becomes the cover when none is set - clear the cover when the cover photo is trashed or moved out - "set as album cover" in the photo context menu and actions panel OpenSpec changes: fix-large-album-load, fix-album-cover --- .env.example | 12 +- backend/internal/app/app.go | 19 ++- backend/internal/app/health_test.go | 2 +- backend/internal/app/objects_test.go | 8 +- backend/internal/app/share_flow_test.go | 4 +- backend/internal/config/config.go | 51 ++++-- backend/internal/e2e/sqlite_test.go | 147 +++++++++++++++++- backend/internal/photos/copy_test.go | 2 +- backend/internal/photos/helpers_test.go | 4 +- backend/internal/photos/objects.go | 12 +- backend/internal/photos/repository.go | 1 + backend/internal/photos/service.go | 7 + backend/internal/photos/sqlite_objects.go | 10 +- backend/internal/photos/sqlite_store.go | 8 + backend/internal/photos/store.go | 8 + backend/internal/photos/worker.go | 61 ++++++-- backend/internal/photos/worker_test.go | 60 ++++++- backend/internal/repo/contract/contract.go | 25 +++ backend/internal/storage/fs_test.go | 2 +- backend/internal/storage/s3_test.go | 2 +- backend/internal/storage/signer.go | 34 +++- backend/internal/storage/signer_test.go | 63 ++++++++ backend/internal/storage/storage.go | 2 +- frontend/src/lib/components/AlbumPanel.svelte | 7 + frontend/src/lib/components/PhotoPanel.svelte | 15 ++ frontend/src/lib/sse.ts | 1 + .../src/routes/(app)/albums/[id]/+page.svelte | 12 ++ .../changes/fix-album-cover/.openspec.yaml | 2 + openspec/changes/fix-album-cover/design.md | 98 ++++++++++++ openspec/changes/fix-album-cover/proposal.md | 37 +++++ .../fix-album-cover/specs/albums/spec.md | 20 +++ .../fix-album-cover/specs/photos/spec.md | 24 +++ openspec/changes/fix-album-cover/tasks.md | 21 +++ .../fix-large-album-load/.openspec.yaml | 2 + .../changes/fix-large-album-load/design.md | 115 ++++++++++++++ .../changes/fix-large-album-load/proposal.md | 46 ++++++ .../specs/media-objects/spec.md | 25 +++ .../fix-large-album-load/specs/photos/spec.md | 32 ++++ .../specs/realtime/spec.md | 31 ++++ .../changes/fix-large-album-load/tasks.md | 24 +++ 40 files changed, 999 insertions(+), 57 deletions(-) create mode 100644 backend/internal/storage/signer_test.go create mode 100644 openspec/changes/fix-album-cover/.openspec.yaml create mode 100644 openspec/changes/fix-album-cover/design.md create mode 100644 openspec/changes/fix-album-cover/proposal.md create mode 100644 openspec/changes/fix-album-cover/specs/albums/spec.md create mode 100644 openspec/changes/fix-album-cover/specs/photos/spec.md create mode 100644 openspec/changes/fix-album-cover/tasks.md create mode 100644 openspec/changes/fix-large-album-load/.openspec.yaml create mode 100644 openspec/changes/fix-large-album-load/design.md create mode 100644 openspec/changes/fix-large-album-load/proposal.md create mode 100644 openspec/changes/fix-large-album-load/specs/media-objects/spec.md create mode 100644 openspec/changes/fix-large-album-load/specs/photos/spec.md create mode 100644 openspec/changes/fix-large-album-load/specs/realtime/spec.md create mode 100644 openspec/changes/fix-large-album-load/tasks.md diff --git a/.env.example b/.env.example index 70b85fd..0e77923 100644 --- a/.env.example +++ b/.env.example @@ -28,6 +28,11 @@ FS_ROOT=./data URL_SECRET= # Lifetime of signed object URLs OBJECT_URL_TTL=1h +# Quantizes URL expiry so the same object keeps an identical URL within a window +# (capped at OBJECT_URL_TTL), letting clients reuse cached previews +OBJECT_URL_SIGN_WINDOW=10m +# Cache-Control max-age for object responses (blobs are immutable) +OBJECT_CACHE_MAX_AGE=1h # Auth. The token signing secret is generated on first start and stored in the # metadata layer (app_settings); it is not configured here. @@ -42,9 +47,12 @@ APP_BASE_URL=http://localhost:8080 SSE_HEARTBEAT_INTERVAL=25s SSE_CLIENT_BUFFER=64 -# Background worker (variant generation + object GC) +# Background worker (variant generation + object GC). +# Throughput is roughly WORKER_BATCH per WORKER_INTERVAL; WORKER_CONCURRENCY caps +# how many of those jobs are processed in parallel (CPU-bound, scales with cores). WORKER_INTERVAL=5s -WORKER_BATCH=10 +WORKER_BATCH=100 +WORKER_CONCURRENCY=4 WORKER_PENDING_TTL=10m # Logging. LOG_LEVEL=debug adds per-request access logs and expected 4xx errors. diff --git a/backend/internal/app/app.go b/backend/internal/app/app.go index ac9ba51..d990c36 100644 --- a/backend/internal/app/app.go +++ b/backend/internal/app/app.go @@ -3,6 +3,7 @@ package app import ( "context" "errors" + "fmt" "io/fs" "net/http" "strings" @@ -59,7 +60,7 @@ func New(cfg *config.Config, repos store.Repos, ping func(ctx context.Context) e membersSvc := members.NewService(repos.Members, repos.Albums, hub, notifSvc) photosSvc := photos.NewService(repos.Photos, albumsSvc, st, hub) - photosWorker := photos.NewWorker(photosSvc, cfg.Worker.Interval, cfg.Worker.Batch, cfg.Worker.PendingTTL) + photosWorker := photos.NewWorker(photosSvc, cfg.Worker.Interval, cfg.Worker.Batch, cfg.Worker.PendingTTL, cfg.Worker.Concurrency) socialSvc := social.NewService(repos.Social, albumsSvc, repos.Photos, hub, notifSvc) @@ -198,20 +199,36 @@ func (s *Server) health(c *gin.Context) { func (s *Server) serveObject(c *gin.Context) { key := strings.TrimPrefix(c.Param("key"), "/") + // Objects are immutable (one key maps to one blob), so successful responses + // are safe to cache. The header is dropped on error paths below. + c.Writer.Header().Set("Cache-Control", cacheControlValue(s.cfg.Storage.CacheMaxAge)) switch err := s.storage.ServeObject(c.Writer, c.Request, key); { case err == nil: return case errors.Is(err, storage.ErrSignedURLInvalid): + c.Writer.Header().Del("Cache-Control") c.Status(http.StatusForbidden) case errors.Is(err, storage.ErrObjectNotFound): + c.Writer.Header().Del("Cache-Control") httpx.RespondError(c, httpx.NotFound("object_not_found", "object not found")) case errors.Is(err, storage.ErrRangeNotSatisfiable): + c.Writer.Header().Del("Cache-Control") c.Status(http.StatusRequestedRangeNotSatisfiable) default: + c.Writer.Header().Del("Cache-Control") c.Status(http.StatusInternalServerError) } } +// cacheControlValue renders the Cache-Control value for object responses. +// Non-positive values fall back to a one-hour freshness window. +func cacheControlValue(maxAge time.Duration) string { + if maxAge <= 0 { + maxAge = time.Hour + } + return fmt.Sprintf("private, max-age=%d", int(maxAge.Seconds())) +} + func (s *Server) albumSSE(c *gin.Context) { userID := auth.UserID(c) albumID := c.Param("id") diff --git a/backend/internal/app/health_test.go b/backend/internal/app/health_test.go index b9588e4..8d1db3a 100644 --- a/backend/internal/app/health_test.go +++ b/backend/internal/app/health_test.go @@ -16,7 +16,7 @@ import ( func mustSigner(t *testing.T) *storage.Signer { t.Helper() - signer, err := storage.NewSigner("http://app.test", "secret", time.Hour) + signer, err := storage.NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute) if err != nil { t.Fatalf("signer: %v", err) } diff --git a/backend/internal/app/objects_test.go b/backend/internal/app/objects_test.go index 4455452..a0fee96 100644 --- a/backend/internal/app/objects_test.go +++ b/backend/internal/app/objects_test.go @@ -22,7 +22,7 @@ import ( func newTestServerWithFS(t *testing.T) (*storage.FS, *gin.Engine) { t.Helper() gin.SetMode(gin.TestMode) - signer, err := storage.NewSigner("http://app.test", "secret", time.Hour) + signer, err := storage.NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute) if err != nil { t.Fatalf("signer: %v", err) } @@ -70,6 +70,9 @@ func TestServeObjectRoute(t *testing.T) { if rec.Body.String() != "0123456789" { t.Fatalf("body = %q", rec.Body.String()) } + if cc := rec.Header().Get("Cache-Control"); cc == "" || !strings.Contains(cc, "max-age=") { + t.Fatalf("cache-control = %q, want max-age directive", cc) + } }) t.Run("range", func(t *testing.T) { @@ -89,6 +92,9 @@ func TestServeObjectRoute(t *testing.T) { if rec.Code != http.StatusForbidden { t.Fatalf("status = %d, want 403", rec.Code) } + if cc := rec.Header().Get("Cache-Control"); cc != "" { + t.Fatalf("cache-control = %q, want empty on error", cc) + } }) t.Run("forged", func(t *testing.T) { diff --git a/backend/internal/app/share_flow_test.go b/backend/internal/app/share_flow_test.go index 5f17255..ef7592b 100644 --- a/backend/internal/app/share_flow_test.go +++ b/backend/internal/app/share_flow_test.go @@ -149,7 +149,7 @@ func runShareFlow(t *testing.T, backend storage.Backend, verify func(t *testing. } func TestShareFlowWithFS(t *testing.T) { - signer, err := storage.NewSigner("http://app.test", "secret", time.Hour) + signer, err := storage.NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute) if err != nil { t.Fatalf("signer: %v", err) } @@ -178,7 +178,7 @@ func TestShareFlowWithS3(t *testing.T) { if endpoint == "" { t.Skip("S3_ENDPOINT not set") } - signer, err := storage.NewSigner("http://app.test", "secret", time.Hour) + signer, err := storage.NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute) if err != nil { t.Skipf("signer init: %v", err) } diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index be4b1db..d5888c0 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -48,16 +48,19 @@ type SSE struct { } type Worker struct { - Interval time.Duration - Batch int - PendingTTL time.Duration + Interval time.Duration + Batch int + PendingTTL time.Duration + Concurrency int } type Storage struct { - Backend string - FSRoot string - URLSecret string - URLTTL time.Duration + Backend string + FSRoot string + URLSecret string + URLTTL time.Duration + URLSignWindow time.Duration + CacheMaxAge time.Duration } type Database struct { @@ -103,6 +106,16 @@ func Load() (*Config, error) { return nil, fmt.Errorf("invalid OBJECT_URL_TTL: %w", err) } + urlSignWindow, err := time.ParseDuration(getEnv("OBJECT_URL_SIGN_WINDOW", "10m")) + if err != nil { + return nil, fmt.Errorf("invalid OBJECT_URL_SIGN_WINDOW: %w", err) + } + + cacheMaxAge, err := time.ParseDuration(getEnv("OBJECT_CACHE_MAX_AGE", "1h")) + if err != nil { + return nil, fmt.Errorf("invalid OBJECT_CACHE_MAX_AGE: %w", err) + } + heartbeat, err := time.ParseDuration(getEnv("SSE_HEARTBEAT_INTERVAL", "25s")) if err != nil { return nil, fmt.Errorf("invalid SSE_HEARTBEAT_INTERVAL: %w", err) @@ -123,7 +136,7 @@ func Load() (*Config, error) { return nil, fmt.Errorf("invalid WORKER_INTERVAL: %w", err) } - workerBatch, err := strconv.Atoi(getEnv("WORKER_BATCH", "10")) + workerBatch, err := strconv.Atoi(getEnv("WORKER_BATCH", "100")) if err != nil { return nil, fmt.Errorf("invalid WORKER_BATCH: %w", err) } @@ -133,6 +146,11 @@ func Load() (*Config, error) { return nil, fmt.Errorf("invalid WORKER_PENDING_TTL: %w", err) } + workerConcurrency, err := strconv.Atoi(getEnv("WORKER_CONCURRENCY", "4")) + if err != nil { + return nil, fmt.Errorf("invalid WORKER_CONCURRENCY: %w", err) + } + logLevel := getEnv("LOG_LEVEL", "info") if _, err := logging.ParseLevel(logLevel); err != nil { return nil, err @@ -160,10 +178,12 @@ func Load() (*Config, error) { UsePathStyle: pathStyle, }, Storage: Storage{ - Backend: getEnv("STORAGE_BACKEND", "fs"), - FSRoot: getEnv("FS_ROOT", "./data"), - URLSecret: getEnv("URL_SECRET", ""), - URLTTL: urlTTL, + Backend: getEnv("STORAGE_BACKEND", "fs"), + FSRoot: getEnv("FS_ROOT", "./data"), + URLSecret: getEnv("URL_SECRET", ""), + URLTTL: urlTTL, + URLSignWindow: urlSignWindow, + CacheMaxAge: cacheMaxAge, }, Database: Database{ Backend: getEnv("DB_BACKEND", "sqlite"), @@ -182,9 +202,10 @@ func Load() (*Config, error) { ClientBuffer: bufSize, }, Worker: Worker{ - Interval: workerInterval, - Batch: workerBatch, - PendingTTL: workerPendingTTL, + Interval: workerInterval, + Batch: workerBatch, + PendingTTL: workerPendingTTL, + Concurrency: workerConcurrency, }, Log: Log{ Level: logLevel, diff --git a/backend/internal/e2e/sqlite_test.go b/backend/internal/e2e/sqlite_test.go index 91e2c0e..12b8e0a 100644 --- a/backend/internal/e2e/sqlite_test.go +++ b/backend/internal/e2e/sqlite_test.go @@ -43,6 +43,7 @@ type stack struct { photos *photos.Service albums *albums.Service auth *auth.Service + hub *sse.Hub } // openStack opens a migrated SQLite database at path and builds the services. @@ -61,7 +62,7 @@ func openStack(t *testing.T, path, blobRoot string) (*stack, func()) { } repos := store.NewSQLite(database) hub := sse.NewHub(1) - signer, err := storage.NewSigner("http://app.test", "secret", time.Hour) + signer, err := storage.NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute) if err != nil { t.Fatalf("signer: %v", err) } @@ -72,7 +73,145 @@ func openStack(t *testing.T, path, blobRoot string) (*stack, func()) { authSvc := auth.NewService(repos.Auth, auth.NewTokenManager("secret", 15*time.Minute, time.Hour)) albumsSvc := albums.NewService(repos.Albums, repos.Members, hub, blob) photosSvc := photos.NewService(repos.Photos, albumsSvc, blob, hub) - return &stack{db: database, photos: photosSvc, albums: albumsSvc, auth: authSvc}, func() { database.Close() } + return &stack{db: database, photos: photosSvc, albums: albumsSvc, auth: authSvc, hub: hub}, func() { database.Close() } +} + +func TestUploadSetsDefaultCover(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "galo.db") + st, closeDB := openStack(t, path, t.TempDir()) + defer closeDB() + + user, _, err := st.auth.Register(ctx, "cover@test.local", "password123", "C") + if err != nil { + t.Fatalf("register: %v", err) + } + album, err := st.albums.Create(ctx, user.ID, albums.CreateInput{Title: "Cover"}) + if err != nil { + t.Fatalf("create album: %v", err) + } + + first, err := st.photos.Upload(ctx, photos.UploadInput{ + AlbumID: album.ID, UserID: user.ID, Filename: "a.png", Data: makePNG(t), + }) + if err != nil { + t.Fatalf("upload first: %v", err) + } + got, _, err := st.albums.Get(ctx, album.ID, user.ID) + if err != nil { + t.Fatalf("get album: %v", err) + } + if got.CoverPhotoID == nil || *got.CoverPhotoID != first.ID { + t.Fatalf("cover = %v, want first photo %s", got.CoverPhotoID, first.ID) + } + + second, err := st.photos.Upload(ctx, photos.UploadInput{ + AlbumID: album.ID, UserID: user.ID, Filename: "b.png", Data: makePNG(t), + }) + if err != nil { + t.Fatalf("upload second: %v", err) + } + got, _, err = st.albums.Get(ctx, album.ID, user.ID) + if err != nil { + t.Fatalf("get album after second: %v", err) + } + if got.CoverPhotoID == nil || *got.CoverPhotoID != first.ID { + t.Fatalf("cover changed to %v, want still first %s (second=%s)", got.CoverPhotoID, first.ID, second.ID) + } +} + +func TestCoverClearedWhenCoverPhotoLeaves(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "galo.db") + st, closeDB := openStack(t, path, t.TempDir()) + defer closeDB() + + user, _, err := st.auth.Register(ctx, "clear@test.local", "password123", "C") + if err != nil { + t.Fatalf("register: %v", err) + } + albumA, err := st.albums.Create(ctx, user.ID, albums.CreateInput{Title: "A"}) + if err != nil { + t.Fatalf("create A: %v", err) + } + albumB, err := st.albums.Create(ctx, user.ID, albums.CreateInput{Title: "B"}) + if err != nil { + t.Fatalf("create B: %v", err) + } + + cover, err := st.photos.Upload(ctx, photos.UploadInput{ + AlbumID: albumA.ID, UserID: user.ID, Filename: "cover.png", Data: makePNG(t), + }) + if err != nil { + t.Fatalf("upload cover: %v", err) + } + + // Trashing the cover clears it. + if err := st.photos.Delete(ctx, cover.ID, user.ID); err != nil { + t.Fatalf("delete cover: %v", err) + } + goal, _, _ := st.albums.Get(ctx, albumA.ID, user.ID) + if goal.CoverPhotoID != nil { + t.Fatalf("cover not cleared on delete: %v", goal.CoverPhotoID) + } + + // A fresh upload becomes the cover, and moving it out clears it again. + next, err := st.photos.Upload(ctx, photos.UploadInput{ + AlbumID: albumA.ID, UserID: user.ID, Filename: "next.png", Data: makePNG(t), + }) + if err != nil { + t.Fatalf("upload next: %v", err) + } + goal, _, _ = st.albums.Get(ctx, albumA.ID, user.ID) + if goal.CoverPhotoID == nil || *goal.CoverPhotoID != next.ID { + t.Fatalf("cover = %v, want %s", goal.CoverPhotoID, next.ID) + } + if _, err := st.photos.MoveBatch(ctx, []string{next.ID}, user.ID, albumB.ID); err != nil { + t.Fatalf("move: %v", err) + } + goal, _, _ = st.albums.Get(ctx, albumA.ID, user.ID) + if goal.CoverPhotoID != nil { + t.Fatalf("cover not cleared on move: %v", goal.CoverPhotoID) + } +} + +func TestWorkerPublishesPhotoUpdated(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "galo.db") + st, closeDB := openStack(t, path, t.TempDir()) + defer closeDB() + + worker := photos.NewWorker(st.photos, 20*time.Millisecond, 10, time.Minute, 2) + wctx, cancel := context.WithCancel(ctx) + defer cancel() + go worker.Run(wctx) + + user, _, err := st.auth.Register(ctx, "evt@test.local", "password123", "E") + if err != nil { + t.Fatalf("register: %v", err) + } + album, err := st.albums.Create(ctx, user.ID, albums.CreateInput{Title: "Trip"}) + if err != nil { + t.Fatalf("create album: %v", err) + } + + sub := st.hub.SubscribeAlbum(album.ID) + defer sub.Close() + + if _, err := st.photos.Upload(ctx, photos.UploadInput{ + AlbumID: album.ID, UserID: user.ID, Filename: "a.png", Data: makePNG(t), + }); err != nil { + t.Fatalf("upload: %v", err) + } + + select { + case ev := <-sub.Events(): + if ev.Type != "photo.updated" { + t.Fatalf("variant-ready event = %q, want photo.updated", ev.Type) + } + case <-time.After(5 * time.Second): + t.Fatalf("no photo.updated event published for generated variants") + } } func TestSQLiteEndToEndWithRestart(t *testing.T) { @@ -81,7 +220,7 @@ func TestSQLiteEndToEndWithRestart(t *testing.T) { blobs := t.TempDir() st, closeDB := openStack(t, path, blobs) - worker := photos.NewWorker(st.photos, 20*time.Millisecond, 10, time.Minute) + worker := photos.NewWorker(st.photos, 20*time.Millisecond, 10, time.Minute, 2) wctx, cancel := context.WithCancel(ctx) go worker.Run(wctx) @@ -127,7 +266,7 @@ func TestSQLiteEndToEndWithRestart(t *testing.T) { st2, close2 := openStack(t, path, blobs) defer close2() - worker2 := photos.NewWorker(st2.photos, 20*time.Millisecond, 10, time.Minute) + worker2 := photos.NewWorker(st2.photos, 20*time.Millisecond, 10, time.Minute, 2) wctx2, cancel2 := context.WithCancel(ctx) defer cancel2() go worker2.Run(wctx2) diff --git a/backend/internal/photos/copy_test.go b/backend/internal/photos/copy_test.go index f04a41b..e060616 100644 --- a/backend/internal/photos/copy_test.go +++ b/backend/internal/photos/copy_test.go @@ -22,7 +22,7 @@ func TestCopySharesStorageObjects(t *testing.T) { } t.Cleanup(func() { cleanupPhoto(t, ctx, pool, cl, p.ID) }) - w := NewWorker(svc, time.Second, 10, time.Minute) + w := NewWorker(svc, time.Second, 10, time.Minute, 2) w.processVariantJobs(ctx) srcIDs, err := svc.store.PhotoObjectIDs(ctx, p.ID) diff --git a/backend/internal/photos/helpers_test.go b/backend/internal/photos/helpers_test.go index f68c825..6209e53 100644 --- a/backend/internal/photos/helpers_test.go +++ b/backend/internal/photos/helpers_test.go @@ -106,7 +106,7 @@ func newTestService(t *testing.T, pool *pgxpool.Pool) (*Service, storage.Backend if endpoint == "" { t.Skip("S3_ENDPOINT not set") } - signer, err := storage.NewSigner("http://app.test", "test-secret", time.Hour) + signer, err := storage.NewSigner("http://app.test", "test-secret", time.Hour, 10*time.Minute) if err != nil { t.Skipf("signer init: %v", err) } @@ -128,7 +128,7 @@ func newTestService(t *testing.T, pool *pgxpool.Pool) (*Service, storage.Backend // storage backend so the full flow can run without external infrastructure. func newTestServiceFS(t *testing.T, pool *pgxpool.Pool) (*Service, storage.Backend) { t.Helper() - signer, err := storage.NewSigner("http://app.test", "test-secret", time.Hour) + signer, err := storage.NewSigner("http://app.test", "test-secret", time.Hour, 10*time.Minute) if err != nil { t.Fatalf("storage init: %v", err) } diff --git a/backend/internal/photos/objects.go b/backend/internal/photos/objects.go index f28cfff..e0edaaf 100644 --- a/backend/internal/photos/objects.go +++ b/backend/internal/photos/objects.go @@ -296,11 +296,15 @@ func (s *Store) ClaimVariantJobs(ctx context.Context, limit int, lease time.Dura defer func() { _ = tx.Rollback(ctx) }() rows, err := tx.Query(ctx, - `SELECT photo_id FROM variant_jobs - WHERE next_attempt_at <= now() - ORDER BY next_attempt_at + `SELECT vj.photo_id FROM variant_jobs vj + JOIN photos p ON p.id = vj.photo_id + WHERE vj.next_attempt_at <= now() + ORDER BY (vj.attempts > 0) ASC, + CASE WHEN vj.attempts = 0 THEN p.created_at END DESC NULLS LAST, + vj.next_attempt_at ASC, + vj.photo_id DESC LIMIT $1 - FOR UPDATE SKIP LOCKED`, limit) + FOR UPDATE OF vj SKIP LOCKED`, limit) if err != nil { return nil, err } diff --git a/backend/internal/photos/repository.go b/backend/internal/photos/repository.go index f352dc2..66c0a36 100644 --- a/backend/internal/photos/repository.go +++ b/backend/internal/photos/repository.go @@ -28,6 +28,7 @@ type Repository interface { DetachTag(ctx context.Context, photoID, tagID string) error TagOwnedBy(ctx context.Context, tagID, userID string) (bool, error) SetCoverIfEmpty(ctx context.Context, albumID, photoID string) + ClearCoverIfMatches(ctx context.Context, albumID, photoID string) UpdateFilename(ctx context.Context, id, filename string) error BackfillEXIF(ctx context.Context, fetch func(context.Context, string) ([]byte, error)) (int, error) diff --git a/backend/internal/photos/service.go b/backend/internal/photos/service.go index c128652..22bf691 100644 --- a/backend/internal/photos/service.go +++ b/backend/internal/photos/service.go @@ -110,6 +110,9 @@ func (s *Service) Upload(ctx context.Context, in UploadInput) (*models.Photo, er return nil, err } + // The first photo uploaded to an album becomes its default cover. + s.store.SetCoverIfEmpty(ctx, in.AlbumID, p.ID) + if err := s.attachURLs(ctx, p); err != nil { return nil, err } @@ -298,6 +301,8 @@ func (s *Service) Delete(ctx context.Context, photoID, userID string) error { if err := s.store.SetDeleted(ctx, photoID, true); err != nil { return err } + // A trashed cover leaves the album without one. + s.store.ClearCoverIfMatches(ctx, p.AlbumID, photoID) s.publishDeleted(p) return nil } @@ -451,6 +456,8 @@ func (s *Service) MoveBatch(ctx context.Context, ids []string, userID, targetAlb if err := s.store.UpdateAlbum(ctx, id, targetAlbumID); err != nil { continue } + // Moving a cover photo out of its album leaves the album without one. + s.store.ClearCoverIfMatches(ctx, oldAlbum, id) p.AlbumID = targetAlbumID s.hub.Publish(oldAlbum, "photo.deleted", map[string]string{ "id": p.ID, "album_id": oldAlbum, "uploader_id": p.UploaderID, diff --git a/backend/internal/photos/sqlite_objects.go b/backend/internal/photos/sqlite_objects.go index 9fa6da3..120e730 100644 --- a/backend/internal/photos/sqlite_objects.go +++ b/backend/internal/photos/sqlite_objects.go @@ -240,9 +240,13 @@ func (s *SQLiteStore) ClaimVariantJobs(ctx context.Context, limit int, lease tim defer func() { _ = tx.Rollback() }() rows, err := tx.QueryContext(ctx, - `SELECT photo_id FROM variant_jobs - WHERE next_attempt_at <= $1 - ORDER BY next_attempt_at + `SELECT vj.photo_id FROM variant_jobs vj + JOIN photos p ON p.id = vj.photo_id + WHERE vj.next_attempt_at <= $1 + ORDER BY (vj.attempts > 0) ASC, + CASE WHEN vj.attempts = 0 THEN p.created_at END DESC NULLS LAST, + vj.next_attempt_at ASC, + vj.photo_id DESC LIMIT $2`, now, limit) if err != nil { return nil, err diff --git a/backend/internal/photos/sqlite_store.go b/backend/internal/photos/sqlite_store.go index c943fe1..e7958df 100644 --- a/backend/internal/photos/sqlite_store.go +++ b/backend/internal/photos/sqlite_store.go @@ -403,6 +403,14 @@ func (s *SQLiteStore) SetCoverIfEmpty(ctx context.Context, albumID, photoID stri albumID, photoID) } +// ClearCoverIfMatches removes the album cover only when it currently points at +// photoID, so a concurrent cover change is not clobbered. +func (s *SQLiteStore) ClearCoverIfMatches(ctx context.Context, albumID, photoID string) { + _, _ = s.db.ExecContext(ctx, + `UPDATE albums SET cover_photo_id = NULL WHERE id = $1 AND cover_photo_id = $2`, + albumID, photoID) +} + func (s *SQLiteStore) UpdateFilename(ctx context.Context, id, filename string) error { res, err := s.db.ExecContext(ctx, `UPDATE photos SET original_filename = $2 WHERE id = $1`, id, filename) if err != nil { diff --git a/backend/internal/photos/store.go b/backend/internal/photos/store.go index 66ff25b..9f44e56 100644 --- a/backend/internal/photos/store.go +++ b/backend/internal/photos/store.go @@ -424,6 +424,14 @@ func (s *Store) SetCoverIfEmpty(ctx context.Context, albumID, photoID string) { albumID, photoID) } +// ClearCoverIfMatches removes the album cover only when it currently points at +// photoID, so a concurrent cover change is not clobbered. +func (s *Store) ClearCoverIfMatches(ctx context.Context, albumID, photoID string) { + _, _ = s.pool.Exec(ctx, + `UPDATE albums SET cover_photo_id = NULL WHERE id = $1 AND cover_photo_id = $2`, + albumID, photoID) +} + func (s *Store) UpdateFilename(ctx context.Context, id, filename string) error { tag, err := s.pool.Exec(ctx, `UPDATE photos SET original_filename = $2 WHERE id = $1`, id, filename) if err != nil { diff --git a/backend/internal/photos/worker.go b/backend/internal/photos/worker.go index 80d76b0..ebaec6d 100644 --- a/backend/internal/photos/worker.go +++ b/backend/internal/photos/worker.go @@ -5,6 +5,7 @@ import ( "context" "errors" "log/slog" + "sync" "time" "galo/internal/auth" @@ -16,15 +17,16 @@ const maxVariantAttempts = 5 // 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 + 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) *Worker { +func NewWorker(svc *Service, interval time.Duration, batch int, pendingTTL time.Duration, concurrency int) *Worker { if interval <= 0 { interval = 5 * time.Second } @@ -34,12 +36,16 @@ func NewWorker(svc *Service, interval time.Duration, batch int, pendingTTL time. 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, + svc: svc, + interval: interval, + batch: batch, + pendingTTL: pendingTTL, + lease: interval * 4, + concurrency: concurrency, } } @@ -71,9 +77,36 @@ func (w *Worker) processVariantJobs(ctx context.Context) { slog.Error("claim variant jobs", "error", err) return } - for _, id := range ids { + 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) { @@ -162,7 +195,9 @@ func (w *Worker) notify(ctx context.Context, photoID string) { if err := w.svc.attachURLs(ctx, p); err != nil { return } - w.svc.hub.Publish(p.AlbumID, "photo.created", p) + // 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) { diff --git a/backend/internal/photos/worker_test.go b/backend/internal/photos/worker_test.go index 5b7f165..0273a3e 100644 --- a/backend/internal/photos/worker_test.go +++ b/backend/internal/photos/worker_test.go @@ -40,9 +40,21 @@ func runWorkerFlow(t *testing.T, pool *pgxpool.Pool, svc *Service, cl storage.Ba } t.Cleanup(func() { cleanupPhoto(t, ctx, pool, cl, p.ID) }) - w := NewWorker(svc, time.Second, 10, time.Minute) + 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") + } + refreshed, err := svc.store.GetByID(ctx, p.ID, "") if err != nil { t.Fatalf("get refreshed: %v", err) @@ -82,6 +94,52 @@ func runWorkerFlow(t *testing.T, pool *pgxpool.Pool, svc *Service, cl storage.Ba } } +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 countObjects(t *testing.T, ctx context.Context, pool *pgxpool.Pool, ids []string) int { t.Helper() var n int diff --git a/backend/internal/repo/contract/contract.go b/backend/internal/repo/contract/contract.go index 25cd9b8..e257650 100644 --- a/backend/internal/repo/contract/contract.go +++ b/backend/internal/repo/contract/contract.go @@ -382,6 +382,18 @@ func testPhotos(t *testing.T, ctx context.Context, r Repos) { t.Fatalf("cover not set: %+v", al.CoverPhotoID) } + // Clearing the cover only affects a matching photo id. + r.Photos.ClearCoverIfMatches(ctx, a.ID, newID()) + al, _ = r.Albums.GetByID(ctx, a.ID) + if al.CoverPhotoID == nil || *al.CoverPhotoID != p.ID { + t.Fatalf("cover cleared by non-matching id: %+v", al.CoverPhotoID) + } + r.Photos.ClearCoverIfMatches(ctx, a.ID, p.ID) + al, _ = r.Albums.GetByID(ctx, a.ID) + if al.CoverPhotoID != nil { + t.Fatalf("cover not cleared: %+v", al.CoverPhotoID) + } + if err := r.Photos.SetDeleted(ctx, p.ID, true); err != nil { t.Fatalf("soft delete: %v", err) } @@ -524,6 +536,19 @@ func testObjects(t *testing.T, ctx context.Context, r Repos) { if jobs, _ = r.Photos.ClaimVariantJobs(ctx, 10, time.Minute); contains(jobs, p3.ID) { t.Fatalf("abandoned job still present: %+v", jobs) } + + // Fresh jobs are prioritized by photo freshness: the newest photo's job is + // claimed first. The delay keeps the two photos' timestamps distinct. + pOlder := seedPhoto(t, ctx, r, owner.ID, a.ID) + time.Sleep(5 * time.Millisecond) + pNewer := seedPhoto(t, ctx, r, owner.ID, a.ID) + order, err := r.Photos.ClaimVariantJobs(ctx, 2, time.Minute) + if err != nil { + t.Fatalf("claim ordered jobs: %v", err) + } + if len(order) < 2 || order[0] != pNewer.ID || order[1] != pOlder.ID { + t.Fatalf("claim order = %+v, want newest first [%s %s]", order, pNewer.ID, pOlder.ID) + } } func testSettings(t *testing.T, ctx context.Context, r Repos) { diff --git a/backend/internal/storage/fs_test.go b/backend/internal/storage/fs_test.go index 2223608..6c68f27 100644 --- a/backend/internal/storage/fs_test.go +++ b/backend/internal/storage/fs_test.go @@ -16,7 +16,7 @@ import ( func newTestFS(t *testing.T) *FS { t.Helper() - signer, err := NewSigner("http://app.test", "test-secret", time.Hour) + signer, err := NewSigner("http://app.test", "test-secret", time.Hour, 10*time.Minute) if err != nil { t.Fatalf("signer: %v", err) } diff --git a/backend/internal/storage/s3_test.go b/backend/internal/storage/s3_test.go index 95812db..0b3ec65 100644 --- a/backend/internal/storage/s3_test.go +++ b/backend/internal/storage/s3_test.go @@ -22,7 +22,7 @@ func newTestS3(t *testing.T) (*S3, *Signer) { if endpoint == "" { t.Skip("S3_ENDPOINT not set") } - signer, err := NewSigner("http://app.test", "secret", time.Hour) + signer, err := NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute) if err != nil { t.Fatalf("signer: %v", err) } diff --git a/backend/internal/storage/signer.go b/backend/internal/storage/signer.go index 79f20b7..297c024 100644 --- a/backend/internal/storage/signer.go +++ b/backend/internal/storage/signer.go @@ -20,33 +20,59 @@ type Signer struct { secret []byte baseURL string ttl time.Duration + window time.Duration + now func() time.Time } // NewSigner builds a signer. secret signs the URLs (required); baseURL is the -// public base URL of the app; ttl limits URL lifetime (defaults to one hour). -func NewSigner(baseURL, secret string, ttl time.Duration) (*Signer, error) { +// public base URL of the app; ttl limits URL lifetime (defaults to one hour); +// window quantizes the expiry so the same object yields an identical URL within +// one window (defaults to ten minutes, and is capped at ttl). A non-positive +// window keeps the URL stable for its whole lifetime. +func NewSigner(baseURL, secret string, ttl, window time.Duration) (*Signer, error) { if strings.TrimSpace(secret) == "" { return nil, errors.New("url secret is required") } if ttl <= 0 { ttl = time.Hour } + if window <= 0 { + window = 10 * time.Minute + } + if window > ttl { + window = ttl + } return &Signer{ secret: []byte(secret), baseURL: strings.TrimRight(baseURL, "/"), ttl: ttl, + window: window, + now: time.Now, }, nil } -// URL returns a signed app URL for the given opaque key. +// URL returns a signed app URL for the given opaque key. The expiry is +// quantized to the signing window so repeated calls within that window return +// the identical URL, letting clients reuse cached objects. func (s *Signer) URL(key string) string { - exp := time.Now().Add(s.ttl).Unix() + exp := s.expiry() q := url.Values{} q.Set("e", strconv.FormatInt(exp, 10)) q.Set("s", s.sign(key, exp)) return fmt.Sprintf("%s/api/objects/%s?%s", s.baseURL, escapeKey(key), q.Encode()) } +// expiry is the quantized Unix expiry shared by all URLs issued in the current +// window: the window start plus the TTL. It is never earlier than now. +func (s *Signer) expiry() int64 { + now := s.now() + exp := now.Truncate(s.window).Add(s.ttl) + if exp.Before(now) { + exp = now.Add(s.ttl) + } + return exp.Unix() +} + // VerifyRequest checks the signature and expiry carried by an object request. // It returns ErrSignedURLInvalid when the request is not authorized. func (s *Signer) VerifyRequest(key string, r *http.Request) error { diff --git a/backend/internal/storage/signer_test.go b/backend/internal/storage/signer_test.go new file mode 100644 index 0000000..5b432df --- /dev/null +++ b/backend/internal/storage/signer_test.go @@ -0,0 +1,63 @@ +package storage + +import ( + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "testing" + "time" +) + +func expiryOf(t *testing.T, raw string) int64 { + t.Helper() + u, err := url.Parse(raw) + if err != nil { + t.Fatalf("parse url: %v", err) + } + e, err := strconv.ParseInt(u.Query().Get("e"), 10, 64) + if err != nil { + t.Fatalf("parse expiry: %v", err) + } + return e +} + +func TestSignerURLStableWithinWindow(t *testing.T) { + s, err := NewSigner("http://app.test", "secret", time.Hour, time.Hour) + if err != nil { + t.Fatalf("new signer: %v", err) + } + first := s.URL("objects/a/thumb.webp") + second := s.URL("objects/a/thumb.webp") + if first != second { + t.Fatalf("URL changed within window:\n %q\n %q", first, second) + } +} + +func TestSignerURLChangesAcrossWindow(t *testing.T) { + s, err := NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute) + if err != nil { + t.Fatalf("new signer: %v", err) + } + base := time.Now() + s.now = func() time.Time { return base } + first := expiryOf(t, s.URL("objects/a/thumb.webp")) + s.now = func() time.Time { return base.Add(s.window) } + second := expiryOf(t, s.URL("objects/a/thumb.webp")) + if first == second { + t.Fatalf("expiry did not advance across window: %d", first) + } +} + +func TestSignerURLVerifies(t *testing.T) { + s, err := NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute) + if err != nil { + t.Fatalf("new signer: %v", err) + } + key := "objects/a/thumb.webp" + raw := s.URL(key) + req := httptest.NewRequest(http.MethodGet, raw, nil) + if err := s.VerifyRequest(key, req); err != nil { + t.Fatalf("verify issued URL: %v", err) + } +} diff --git a/backend/internal/storage/storage.go b/backend/internal/storage/storage.go index 6e938bc..fe56fa1 100644 --- a/backend/internal/storage/storage.go +++ b/backend/internal/storage/storage.go @@ -108,7 +108,7 @@ func Open(ctx context.Context, cfg *config.Config, jwtSecret string) (Backend, e if secret == "" { return nil, errors.New("object URL secret is not available") } - signer, err := NewSigner(cfg.Server.BaseURL, secret, cfg.Storage.URLTTL) + signer, err := NewSigner(cfg.Server.BaseURL, secret, cfg.Storage.URLTTL, cfg.Storage.URLSignWindow) if err != nil { return nil, err } diff --git a/frontend/src/lib/components/AlbumPanel.svelte b/frontend/src/lib/components/AlbumPanel.svelte index 71e2e16..c9656ac 100644 --- a/frontend/src/lib/components/AlbumPanel.svelte +++ b/frontend/src/lib/components/AlbumPanel.svelte @@ -75,6 +75,13 @@ .catch((e) => (error = (e as Error).message)); }); + $effect(() => { + // Keep the local album in sync with cover changes made elsewhere + // (context menu, photo panel) that go through the albums store. + const updated = albums.list.find((a) => a.id === albumId); + if (updated) album = updated; + }); + $effect(() => { const id = photoId; if (!id) { diff --git a/frontend/src/lib/components/PhotoPanel.svelte b/frontend/src/lib/components/PhotoPanel.svelte index 4172750..a4a03c2 100644 --- a/frontend/src/lib/components/PhotoPanel.svelte +++ b/frontend/src/lib/components/PhotoPanel.svelte @@ -6,6 +6,7 @@ import PanelSection from './PanelSection.svelte'; import PhotoInfoSettingsDialog from './PhotoInfoSettingsDialog.svelte'; import { auth } from '$lib/stores/auth.svelte'; + import { albums } from '$lib/stores/albums.svelte'; import { transfer } from '$lib/stores/transfer.svelte'; import { viewerActions } from '$lib/stores/viewer-actions.svelte'; import { buildExifRows, EXIF_BASE, EXIF_CATALOG, EXIF_LABELS } from '$lib/exif'; @@ -179,6 +180,14 @@ } } + async function setAsCover() { + try { + await albums.setCover(photo.album_id, photo.id); + } catch { + /* ignore */ + } + } + async function confirmDelete() { deleting = true; try { @@ -273,6 +282,12 @@ Копировать + {#if canEdit} + + {/if} {#if canDelete}