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}