diff --git a/.env.example b/.env.example index ac66a60..5392ab3 100644 --- a/.env.example +++ b/.env.example @@ -28,3 +28,8 @@ APP_BASE_URL=http://localhost:8080 # SSE SSE_HEARTBEAT_INTERVAL=25s SSE_CLIENT_BUFFER=64 + +# Background worker (variant generation + object GC) +WORKER_INTERVAL=5s +WORKER_BATCH=10 +WORKER_PENDING_TTL=10m diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index cae1c25..bda98af 100644 --- a/backend/cmd/server/main.go +++ b/backend/cmd/server/main.go @@ -69,6 +69,8 @@ func main() { hub := sse.NewHub(cfg.SSE.ClientBuffer) server := app.New(cfg, pool, store, hub) + go server.StartWorker(ctx) + httpServer := &http.Server{ Addr: cfg.Server.Addr, Handler: server.Router(), diff --git a/backend/internal/albums/store.go b/backend/internal/albums/store.go index b4cd255..2ceaf5c 100644 --- a/backend/internal/albums/store.go +++ b/backend/internal/albums/store.go @@ -44,11 +44,12 @@ func (s *Store) GetByID(ctx context.Context, id string) (*models.Album, error) { var a models.Album err := s.pool.QueryRow(ctx, `SELECT a.id, a.owner_id, a.title, a.description, a.cover_photo_id, a.created_at, a.updated_at, - COALESCE(p.thumb_key, ''), + COALESCE(co.storage_key, ''), (SELECT count(*) FROM photos WHERE album_id = a.id), (SELECT count(*) FROM album_members WHERE album_id = a.id) FROM albums a - LEFT JOIN photos p ON p.id = a.cover_photo_id + LEFT JOIN photo_objects cpo ON cpo.photo_id = a.cover_photo_id AND cpo.role = 'thumb' + LEFT JOIN objects co ON co.id = cpo.object_id WHERE a.id = $1`, id). Scan(&a.ID, &a.OwnerID, &a.Title, &a.Description, &a.CoverPhotoID, &a.CreatedAt, &a.UpdatedAt, &a.CoverThumbKey, &a.PhotosCount, &a.MembersCount) if err != nil { @@ -63,12 +64,13 @@ func (s *Store) GetByID(ctx context.Context, id string) (*models.Album, error) { func (s *Store) ListForUser(ctx context.Context, userID string) ([]models.Album, error) { rows, err := s.pool.Query(ctx, `SELECT a.id, a.owner_id, a.title, a.description, a.cover_photo_id, a.created_at, a.updated_at, - COALESCE(p.thumb_key, ''), + COALESCE(co.storage_key, ''), (SELECT count(*) FROM photos WHERE album_id = a.id), (SELECT count(*) FROM album_members WHERE album_id = a.id) FROM albums a LEFT JOIN album_members m ON m.album_id = a.id AND m.user_id = $1 - LEFT JOIN photos p ON p.id = a.cover_photo_id + LEFT JOIN photo_objects cpo ON cpo.photo_id = a.cover_photo_id AND cpo.role = 'thumb' + LEFT JOIN objects co ON co.id = cpo.object_id WHERE a.owner_id = $1 OR m.user_id IS NOT NULL ORDER BY a.created_at DESC`, userID) if err != nil { diff --git a/backend/internal/app/app.go b/backend/internal/app/app.go index 6dfdc8b..0c323d5 100644 --- a/backend/internal/app/app.go +++ b/backend/internal/app/app.go @@ -1,6 +1,7 @@ package app import ( + "context" "io/fs" "net/http" "strings" @@ -37,6 +38,7 @@ type Server struct { membersSvc *members.Service photos *photos.Service + worker *photos.Worker social *social.Service @@ -61,6 +63,7 @@ func New(cfg *config.Config, pool *pgxpool.Pool, st *storage.Client, hub *sse.Hu photosStore := photos.NewStore(pool) photosSvc := photos.NewService(photosStore, albumsSvc, st, hub) + photosWorker := photos.NewWorker(photosSvc, cfg.Worker.Interval, cfg.Worker.Batch, cfg.Worker.PendingTTL) socialStore := social.NewStore(pool) socialSvc := social.NewService(socialStore, albumsSvc, photosStore, hub, notifSvc) @@ -78,12 +81,20 @@ func New(cfg *config.Config, pool *pgxpool.Pool, st *storage.Client, hub *sse.Hu auth: authSvc, membersSvc: membersSvc, photos: photosSvc, + worker: photosWorker, social: socialSvc, notifications: notifSvc, share: shareSvc, } } +// StartWorker runs the background reconciler until ctx is cancelled. +func (s *Server) StartWorker(ctx context.Context) { + if s.worker != nil { + s.worker.Run(ctx) + } +} + func (s *Server) Router() *gin.Engine { r := gin.New() r.Use(gin.Recovery()) diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index e5d7ee6..f99c843 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -48,12 +48,19 @@ type SSE struct { ClientBuffer int } +type Worker struct { + Interval time.Duration + Batch int + PendingTTL time.Duration +} + type Config struct { Postgres Postgres S3 S3 Auth Auth Server Server SSE SSE + Worker Worker } func Load() (*Config, error) { @@ -92,6 +99,21 @@ func Load() (*Config, error) { return nil, fmt.Errorf("invalid S3_USE_PATH_STYLE: %w", err) } + workerInterval, err := time.ParseDuration(getEnv("WORKER_INTERVAL", "5s")) + if err != nil { + return nil, fmt.Errorf("invalid WORKER_INTERVAL: %w", err) + } + + workerBatch, err := strconv.Atoi(getEnv("WORKER_BATCH", "10")) + if err != nil { + return nil, fmt.Errorf("invalid WORKER_BATCH: %w", err) + } + + workerPendingTTL, err := time.ParseDuration(getEnv("WORKER_PENDING_TTL", "10m")) + if err != nil { + return nil, fmt.Errorf("invalid WORKER_PENDING_TTL: %w", err) + } + cfg := &Config{ Postgres: Postgres{ Host: getEnv("PG_HOST", "pg.localnet"), @@ -124,6 +146,11 @@ func Load() (*Config, error) { HeartbeatInterval: heartbeat, ClientBuffer: bufSize, }, + Worker: Worker{ + Interval: workerInterval, + Batch: workerBatch, + PendingTTL: workerPendingTTL, + }, } if cfg.Auth.JWTSecret == "" { diff --git a/backend/internal/photos/copy_test.go b/backend/internal/photos/copy_test.go new file mode 100644 index 0000000..f04a41b --- /dev/null +++ b/backend/internal/photos/copy_test.go @@ -0,0 +1,92 @@ +package photos + +import ( + "context" + "testing" + "time" +) + +func TestCopySharesStorageObjects(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + svc, cl := newTestService(t, pool) + user := createUser(t, ctx, pool) + albumA := createAlbum(t, ctx, pool, user) + albumB := createAlbum(t, ctx, pool, user) + + p, err := svc.Upload(ctx, UploadInput{ + AlbumID: albumA, UserID: user, Filename: "p.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) }) + + w := NewWorker(svc, time.Second, 10, time.Minute) + w.processVariantJobs(ctx) + + srcIDs, err := svc.store.PhotoObjectIDs(ctx, p.ID) + if err != nil || len(srcIDs) != 3 { + t.Fatalf("source objects = %v err=%v", srcIDs, err) + } + + clones, err := svc.CopyBatch(ctx, []string{p.ID}, user, albumB) + if err != nil || len(clones) != 1 { + t.Fatalf("copy: %v len=%d", err, len(clones)) + } + clone := clones[0] + t.Cleanup(func() { cleanupPhoto(t, ctx, pool, cl, clone.ID) }) + + cloneIDs, err := svc.store.PhotoObjectIDs(ctx, clone.ID) + if err != nil || len(cloneIDs) != 3 { + t.Fatalf("clone objects = %v err=%v", cloneIDs, err) + } + if !sameSet(srcIDs, cloneIDs) { + t.Fatalf("clone does not share objects: src=%v clone=%v", srcIDs, cloneIDs) + } + for _, id := range cloneIDs { + if n, err := svc.store.ObjectRefCount(ctx, id); err != nil || n != 2 { + t.Fatalf("object %s ref count = %d err=%v, want 2", id, n, err) + } + } + + // Move the source into the same album; the storage key and shared objects + // must be untouched. + beforeKey := p.StorageKey + moved, err := svc.MoveBatch(ctx, []string{p.ID}, user, albumB) + if err != nil || moved != 1 { + t.Fatalf("move: %v moved=%d", err, moved) + } + after, err := svc.store.GetByID(ctx, p.ID, "") + if err != nil { + t.Fatalf("get after move: %v", err) + } + if after.AlbumID != albumB { + t.Fatalf("photo not moved: album=%q", after.AlbumID) + } + if after.StorageKey != beforeKey { + t.Fatalf("move changed storage key: %q -> %q", beforeKey, after.StorageKey) + } + for _, id := range cloneIDs { + if n, err := svc.store.ObjectRefCount(ctx, id); err != nil || n != 2 { + t.Fatalf("object %s ref count after move = %d err=%v, want 2", id, n, err) + } + } +} + +func sameSet(a, b []string) bool { + if len(a) != len(b) { + return false + } + m := make(map[string]int, len(a)) + for _, v := range a { + m[v]++ + } + for _, v := range b { + m[v]-- + if m[v] < 0 { + return false + } + } + return true +} diff --git a/backend/internal/photos/helpers_test.go b/backend/internal/photos/helpers_test.go new file mode 100644 index 0000000..56491a3 --- /dev/null +++ b/backend/internal/photos/helpers_test.go @@ -0,0 +1,149 @@ +package photos + +import ( + "bytes" + "context" + "image" + "image/color" + "image/png" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "shared-album/internal/auth" + "shared-album/internal/config" + "shared-album/internal/models" + "shared-album/internal/roles" + "shared-album/internal/sse" + "shared-album/internal/storage" +) + +func testPool(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := os.Getenv("TEST_DATABASE_URL") + if dsn == "" { + t.Skip("TEST_DATABASE_URL not set") + } + pool, err := pgxpool.New(context.Background(), dsn) + if err != nil { + t.Skipf("cannot connect: %v", err) + } + if err := pool.Ping(context.Background()); err != nil { + t.Skipf("cannot ping: %v", err) + } + t.Cleanup(pool.Close) + return pool +} + +func testEnvOr(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} + +type stubAlbums struct{} + +func (stubAlbums) Access(ctx context.Context, albumID, userID string) (*models.Album, string, error) { + return &models.Album{ID: albumID}, roles.Owner, nil +} + +func createUser(t *testing.T, ctx context.Context, pool *pgxpool.Pool) string { + t.Helper() + id := auth.NewID() + if _, err := pool.Exec(ctx, + `INSERT INTO users (id, email, password_hash, name) VALUES ($1, $2, 'x', 't')`, + id, id+"@test.local"); err != nil { + t.Fatalf("create user: %v", err) + } + t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM users WHERE id = $1`, id) }) + return id +} + +func createAlbum(t *testing.T, ctx context.Context, pool *pgxpool.Pool, ownerID string) string { + t.Helper() + id := auth.NewID() + if _, err := pool.Exec(ctx, + `INSERT INTO albums (id, owner_id, title) VALUES ($1, $2, 't')`, id, ownerID); err != nil { + t.Fatalf("create album: %v", err) + } + t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM albums WHERE id = $1`, id) }) + return id +} + +func createPhoto(t *testing.T, ctx context.Context, pool *pgxpool.Pool, uploaderID string) string { + t.Helper() + id := auth.NewID() + if _, err := pool.Exec(ctx, + `INSERT INTO photos (id, album_id, uploader_id) VALUES ($1, NULL, $2)`, id, uploaderID); err != nil { + t.Fatalf("create photo: %v", err) + } + return id +} + +func makePNG(t *testing.T) []byte { + t.Helper() + img := image.NewRGBA(image.Rect(0, 0, 4, 4)) + for x := 0; x < 4; x++ { + for y := 0; y < 4; y++ { + img.Set(x, y, color.RGBA{R: uint8(x * 40), G: uint8(y * 40), B: 128, A: 255}) + } + } + var buf bytes.Buffer + if err := png.Encode(&buf, img); err != nil { + t.Fatalf("encode png: %v", err) + } + return buf.Bytes() +} + +// newTestService builds a service backed by the real dev storage. It skips when +// S3 env is absent so unit runs stay hermetic. +func newTestService(t *testing.T, pool *pgxpool.Pool) (*Service, *storage.Client) { + t.Helper() + endpoint := os.Getenv("S3_ENDPOINT") + if endpoint == "" { + t.Skip("S3_ENDPOINT not set") + } + cl, err := storage.New(context.Background(), config.S3{ + Endpoint: endpoint, + Region: testEnvOr("S3_REGION", "garage"), + AccessKey: os.Getenv("S3_ACCESS_KEY"), + SecretKey: os.Getenv("S3_SECRET_KEY"), + Bucket: testEnvOr("S3_BUCKET", "galo-data"), + UsePathStyle: testEnvOr("S3_USE_PATH_STYLE", "true") != "false", + PresignTTL: time.Hour, + }) + if err != nil { + t.Skipf("storage init: %v", err) + } + return NewService(NewStore(pool), stubAlbums{}, cl, sse.NewHub(1)), cl +} + +// cleanupPhoto removes a photo, its objects (rows and blobs) and bindings. +func cleanupPhoto(t *testing.T, ctx context.Context, pool *pgxpool.Pool, cl *storage.Client, photoID string) { + t.Helper() + rows, err := pool.Query(ctx, + `SELECT o.id, o.storage_key FROM photo_objects po JOIN objects o ON o.id = po.object_id + WHERE po.photo_id = $1`, photoID) + if err != nil { + return + } + var ids, keys []string + for rows.Next() { + var id, key string + if rows.Scan(&id, &key) == nil { + ids = append(ids, id) + keys = append(keys, key) + } + } + rows.Close() + _, _ = pool.Exec(ctx, `DELETE FROM photos WHERE id = $1`, photoID) + for _, id := range ids { + _, _ = pool.Exec(ctx, `DELETE FROM objects WHERE id = $1`, id) + } + if cl != nil && len(keys) > 0 { + _ = cl.Delete(ctx, keys...) + } +} diff --git a/backend/internal/photos/keys.go b/backend/internal/photos/keys.go new file mode 100644 index 0000000..39f7c6a --- /dev/null +++ b/backend/internal/photos/keys.go @@ -0,0 +1,17 @@ +package photos + +import ( + "fmt" + "strings" +) + +// objectKey builds an opaque, album-free storage key for a physical object. +// The key never encodes album, photo or any other metadata entity; the mapping +// between a key and its photo binding lives only in the metadata layer. The +// role segment is cosmetic (for debugging) and is not authoritative. +func objectKey(objectID, role, ext string) string { + if ext != "" && !strings.HasPrefix(ext, ".") { + ext = "." + ext + } + return fmt.Sprintf("objects/%s/%s%s", objectID, role, ext) +} diff --git a/backend/internal/photos/keys_test.go b/backend/internal/photos/keys_test.go new file mode 100644 index 0000000..083f579 --- /dev/null +++ b/backend/internal/photos/keys_test.go @@ -0,0 +1,32 @@ +package photos + +import ( + "strings" + "testing" +) + +func TestObjectKeyIsAlbumFree(t *testing.T) { + cases := []struct { + name string + id string + role string + ext string + want string + }{ + {"original with dot", "obj1", RoleOriginal, ".jpg", "objects/obj1/original.jpg"}, + {"thumb without dot", "obj1", RoleThumb, "webp", "objects/obj1/thumb.webp"}, + {"preview", "obj2", RolePreview, ".webp", "objects/obj2/preview.webp"}, + {"empty ext", "obj3", RoleOriginal, "", "objects/obj3/original"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := objectKey(tc.id, tc.role, tc.ext) + if got != tc.want { + t.Fatalf("objectKey = %q, want %q", got, tc.want) + } + if strings.Contains(got, "album") { + t.Fatalf("key must not encode an album entity: %q", got) + } + }) + } +} diff --git a/backend/internal/photos/objects.go b/backend/internal/photos/objects.go new file mode 100644 index 0000000..36ce908 --- /dev/null +++ b/backend/internal/photos/objects.go @@ -0,0 +1,370 @@ +package photos + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" +) + +// Object lifecycle states. The DB row is the ledger entry that governs the +// lifecycle of the corresponding blob in the storage layer. +const ( + ObjectPending = "pending" + ObjectReady = "ready" + ObjectDeleting = "deleting" +) + +// Object roles as bound to a photo. The role is a property of the binding. +const ( + RoleOriginal = "original" + RoleThumb = "thumb" + RolePreview = "preview" +) + +var ( + // ErrObjectNotFound is returned when an object row does not exist. + ErrObjectNotFound = errors.New("object not found") + // ErrObjectNotReady is returned when binding an object that is not ready. + ErrObjectNotReady = errors.New("object is not ready") +) + +// Object is a physical file in the storage layer, described by metadata. +type Object struct { + ID string + StorageKey string + MimeType string + Size int64 + Width int + Height int + Checksum string + State string + CreatedAt time.Time +} + +// CreateObject inserts a new object row in the pending state. +func (s *Store) CreateObject(ctx context.Context, o *Object) error { + if o.State == "" { + o.State = ObjectPending + } + return s.pool.QueryRow(ctx, + `INSERT INTO objects (id, storage_key, mime_type, size, width, height, checksum, state) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + RETURNING created_at`, + o.ID, o.StorageKey, o.MimeType, o.Size, o.Width, o.Height, o.Checksum, o.State).Scan(&o.CreatedAt) +} + +// SetObjectState updates an object's lifecycle state. +func (s *Store) SetObjectState(ctx context.Context, id, state string) error { + tag, err := s.pool.Exec(ctx, `UPDATE objects SET state = $2 WHERE id = $1`, id, state) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrObjectNotFound + } + return nil +} + +// ActivateObject marks an object ready after its bytes have been stored. +func (s *Store) ActivateObject(ctx context.Context, id string) error { + return s.SetObjectState(ctx, id, ObjectReady) +} + +// MarkObjectDeleting marks an object as pending removal from storage. +func (s *Store) MarkObjectDeleting(ctx context.Context, id string) error { + return s.SetObjectState(ctx, id, ObjectDeleting) +} + +// DeleteObjectRow permanently removes an object row (after its blob is gone). +func (s *Store) DeleteObjectRow(ctx context.Context, id string) error { + tag, err := s.pool.Exec(ctx, `DELETE FROM objects WHERE id = $1`, id) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrObjectNotFound + } + return nil +} + +// BindObject attaches a ready object to a photo under the given role. It locks +// the object row and refuses objects that are not ready, so an object being +// garbage-collected can never be bound. +func (s *Store) BindObject(ctx context.Context, photoID, objectID, role string) error { + tx, err := s.pool.Begin(ctx) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + + var state string + if err := tx.QueryRow(ctx, `SELECT state FROM objects WHERE id = $1 FOR UPDATE`, objectID).Scan(&state); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return ErrObjectNotFound + } + return err + } + if state != ObjectReady { + return ErrObjectNotReady + } + if _, err := tx.Exec(ctx, + `INSERT INTO photo_objects (photo_id, object_id, role) VALUES ($1, $2, $3) + ON CONFLICT (photo_id, role) DO UPDATE SET object_id = EXCLUDED.object_id`, + photoID, objectID, role); err != nil { + return err + } + return tx.Commit(ctx) +} + +// PhotoObjectIDs returns the object ids currently bound to a photo. +func (s *Store) PhotoObjectIDs(ctx context.Context, photoID string) ([]string, error) { + rows, err := s.pool.Query(ctx, `SELECT object_id FROM photo_objects WHERE photo_id = $1`, photoID) + if err != nil { + return nil, err + } + defer rows.Close() + var ids []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +// ObjectRefCount counts how many photo bindings reference an object. +func (s *Store) ObjectRefCount(ctx context.Context, objectID string) (int, error) { + var n int + err := s.pool.QueryRow(ctx, `SELECT count(*) FROM photo_objects WHERE object_id = $1`, objectID).Scan(&n) + return n, err +} + +// OriginalObject returns the original object bound to a photo. +func (s *Store) OriginalObject(ctx context.Context, photoID string) (*Object, error) { + var o Object + err := s.pool.QueryRow(ctx, + `SELECT o.id, o.storage_key, o.mime_type, o.size, o.width, o.height, o.checksum, o.state, o.created_at + FROM photo_objects po JOIN objects o ON o.id = po.object_id + WHERE po.photo_id = $1 AND po.role = $2`, photoID, RoleOriginal). + Scan(&o.ID, &o.StorageKey, &o.MimeType, &o.Size, &o.Width, &o.Height, &o.Checksum, &o.State, &o.CreatedAt) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrObjectNotFound + } + return nil, err + } + return &o, nil +} + +// PhotosByOriginalObject returns the ids of all photos bound to the given +// original object, so generated variants can be shared between copies. +func (s *Store) PhotosByOriginalObject(ctx context.Context, originalObjectID string) ([]string, error) { + rows, err := s.pool.Query(ctx, + `SELECT photo_id FROM photo_objects WHERE object_id = $1 AND role = $2`, originalObjectID, RoleOriginal) + if err != nil { + return nil, err + } + defer rows.Close() + var ids []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +// ClaimGarbage claims collectible objects: ready objects with no bindings and +// objects already marked deleting. Rows are locked with SKIP LOCKED so +// concurrent workers do not contend and a concurrent bind cannot slip in. +func (s *Store) ClaimGarbage(ctx context.Context, limit int) ([]Object, error) { + if limit <= 0 { + limit = 10 + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer func() { _ = tx.Rollback(ctx) }() + + rows, err := tx.Query(ctx, + `SELECT o.id, o.storage_key FROM objects o + WHERE o.state = ANY($1) + AND NOT EXISTS (SELECT 1 FROM photo_objects po WHERE po.object_id = o.id) + ORDER BY o.created_at + LIMIT $2 + FOR UPDATE SKIP LOCKED`, []string{ObjectReady, ObjectDeleting}, limit) + if err != nil { + return nil, err + } + var objs []Object + for rows.Next() { + var o Object + if err := rows.Scan(&o.ID, &o.StorageKey); err != nil { + rows.Close() + return nil, err + } + objs = append(objs, o) + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, err + } + if len(objs) == 0 { + return nil, tx.Commit(ctx) + } + ids := make([]string, len(objs)) + for i, o := range objs { + ids[i] = o.ID + } + if _, err := tx.Exec(ctx, `UPDATE objects SET state = $2 WHERE id = ANY($1)`, ids, ObjectDeleting); err != nil { + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return objs, nil +} + +// ClaimStalePending marks abandoned pending objects (whose upload never +// completed) as deleting and returns them for cleanup. +func (s *Store) ClaimStalePending(ctx context.Context, olderThan time.Duration, limit int) ([]Object, error) { + if limit <= 0 { + limit = 10 + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer func() { _ = tx.Rollback(ctx) }() + + rows, err := tx.Query(ctx, + `SELECT id, storage_key FROM objects + WHERE state = $1 AND created_at < $2 + ORDER BY created_at + LIMIT $3 + FOR UPDATE SKIP LOCKED`, ObjectPending, time.Now().Add(-olderThan), limit) + if err != nil { + return nil, err + } + var objs []Object + for rows.Next() { + var o Object + if err := rows.Scan(&o.ID, &o.StorageKey); err != nil { + rows.Close() + return nil, err + } + objs = append(objs, o) + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, err + } + if len(objs) == 0 { + return nil, tx.Commit(ctx) + } + ids := make([]string, len(objs)) + for i, o := range objs { + ids[i] = o.ID + } + if _, err := tx.Exec(ctx, `UPDATE objects SET state = $2 WHERE id = ANY($1)`, ids, ObjectDeleting); err != nil { + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return objs, nil +} + +// ClaimVariantJobs leases due variant jobs and returns their photo ids. The +// lease pushes next_attempt_at forward so the same job is not picked up twice +// while it is being processed. +func (s *Store) ClaimVariantJobs(ctx context.Context, limit int, lease time.Duration) ([]string, error) { + if limit <= 0 { + limit = 10 + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, err + } + 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 + LIMIT $1 + FOR UPDATE SKIP LOCKED`, limit) + if err != nil { + return nil, err + } + var ids []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + rows.Close() + return nil, err + } + ids = append(ids, id) + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, err + } + if len(ids) == 0 { + return nil, tx.Commit(ctx) + } + if _, err := tx.Exec(ctx, + `UPDATE variant_jobs SET next_attempt_at = $2 WHERE photo_id = ANY($1)`, + ids, time.Now().Add(lease)); err != nil { + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return ids, nil +} + +// CompleteVariantJob removes a finished job. +func (s *Store) CompleteVariantJob(ctx context.Context, photoID string) error { + _, err := s.pool.Exec(ctx, `DELETE FROM variant_jobs WHERE photo_id = $1`, photoID) + return err +} + +// FailVariantJob records a failure and schedules a retry with exponential +// backoff. After maxAttempts the job is abandoned so a permanently broken photo +// cannot loop forever. +func (s *Store) FailVariantJob(ctx context.Context, photoID, errMsg string, maxAttempts int) error { + tx, err := s.pool.Begin(ctx) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + + var attempts int + err = tx.QueryRow(ctx, + `UPDATE variant_jobs + SET attempts = attempts + 1, + next_attempt_at = now() + make_interval(secs => least(300, 10 * power(2, attempts))), + last_error = $2 + WHERE photo_id = $1 + RETURNING attempts`, photoID, errMsg).Scan(&attempts) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return tx.Commit(ctx) + } + return err + } + if maxAttempts > 0 && attempts >= maxAttempts { + if _, err := tx.Exec(ctx, `DELETE FROM variant_jobs WHERE photo_id = $1`, photoID); err != nil { + return err + } + } + return tx.Commit(ctx) +} diff --git a/backend/internal/photos/objects_test.go b/backend/internal/photos/objects_test.go new file mode 100644 index 0000000..18fbf78 --- /dev/null +++ b/backend/internal/photos/objects_test.go @@ -0,0 +1,187 @@ +package photos + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "shared-album/internal/auth" +) + +func TestObjectStateTransitions(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + st := NewStore(pool) + + objID := auth.NewID() + obj := &Object{ID: objID, StorageKey: "objects/" + objID + "/original.jpg", MimeType: "image/jpeg", Size: 42} + t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM objects WHERE id = $1`, objID) }) + + if err := st.CreateObject(ctx, obj); err != nil { + t.Fatalf("create object: %v", err) + } + if obj.State != ObjectPending { + t.Fatalf("new object state = %q, want %q", obj.State, ObjectPending) + } + if obj.CreatedAt.IsZero() { + t.Fatalf("created_at not populated") + } + + if err := st.ActivateObject(ctx, objID); err != nil { + t.Fatalf("activate: %v", err) + } + if got := objectState(t, ctx, pool, objID); got != ObjectReady { + t.Fatalf("state after activate = %q", got) + } + if err := st.MarkObjectDeleting(ctx, objID); err != nil { + t.Fatalf("mark deleting: %v", err) + } + if got := objectState(t, ctx, pool, objID); got != ObjectDeleting { + t.Fatalf("state after mark = %q", got) + } + if err := st.DeleteObjectRow(ctx, objID); err != nil { + t.Fatalf("delete row: %v", err) + } + if err := st.DeleteObjectRow(ctx, objID); !errors.Is(err, ErrObjectNotFound) { + t.Fatalf("second delete = %v, want ErrObjectNotFound", err) + } +} + +func TestClaimStalePending(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + st := NewStore(pool) + + objID := auth.NewID() + obj := &Object{ID: objID, StorageKey: "objects/" + objID + "/original.jpg"} + t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM objects WHERE id = $1`, objID) }) + if err := st.CreateObject(ctx, obj); err != nil { + t.Fatalf("create object: %v", err) + } + if _, err := pool.Exec(ctx, `UPDATE objects SET created_at = now() - interval '1 hour' WHERE id = $1`, objID); err != nil { + t.Fatalf("age object: %v", err) + } + + objs, err := st.ClaimStalePending(ctx, 10*time.Minute, 100) + if err != nil { + t.Fatalf("claim stale: %v", err) + } + if !containsObject(objs, objID) { + t.Fatalf("stale pending object not claimed: %+v", objs) + } + if got := objectState(t, ctx, pool, objID); got != ObjectDeleting { + t.Fatalf("claimed stale object state = %q, want deleting", got) + } +} + +func TestBindObjectRequiresReadyAndShares(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + st := NewStore(pool) + + user := createUser(t, ctx, pool) + p1 := createPhoto(t, ctx, pool, user) + p2 := createPhoto(t, ctx, pool, user) + + objID := auth.NewID() + obj := &Object{ID: objID, StorageKey: "objects/" + objID + "/original.jpg"} + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM photo_objects WHERE object_id = $1`, objID) + _, _ = pool.Exec(ctx, `DELETE FROM objects WHERE id = $1`, objID) + }) + if err := st.CreateObject(ctx, obj); err != nil { + t.Fatalf("create object: %v", err) + } + + if err := st.BindObject(ctx, p1, objID, RoleOriginal); !errors.Is(err, ErrObjectNotReady) { + t.Fatalf("bind pending object = %v, want ErrObjectNotReady", err) + } + if err := st.ActivateObject(ctx, objID); err != nil { + t.Fatalf("activate: %v", err) + } + if err := st.BindObject(ctx, p1, objID, RoleOriginal); err != nil { + t.Fatalf("bind p1: %v", err) + } + if err := st.BindObject(ctx, p2, objID, RoleOriginal); err != nil { + t.Fatalf("bind p2: %v", err) + } + + if n, err := st.ObjectRefCount(ctx, objID); err != nil || n != 2 { + t.Fatalf("ref count = %d err=%v, want 2", n, err) + } + siblings, err := st.PhotosByOriginalObject(ctx, objID) + if err != nil || len(siblings) != 2 { + t.Fatalf("siblings = %v err=%v, want 2", siblings, err) + } +} + +func TestClaimGarbageRespectsReferences(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + st := NewStore(pool) + + user := createUser(t, ctx, pool) + photo := createPhoto(t, ctx, pool, user) + + boundID := auth.NewID() + freeID := auth.NewID() + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM photo_objects WHERE object_id IN ($1, $2)`, boundID, freeID) + _, _ = pool.Exec(ctx, `DELETE FROM objects WHERE id IN ($1, $2)`, boundID, freeID) + }) + for _, id := range []string{boundID, freeID} { + if err := st.CreateObject(ctx, &Object{ID: id, StorageKey: "objects/" + id + "/original.jpg"}); err != nil { + t.Fatalf("create object: %v", err) + } + if err := st.ActivateObject(ctx, id); err != nil { + t.Fatalf("activate: %v", err) + } + } + if err := st.BindObject(ctx, photo, boundID, RoleOriginal); err != nil { + t.Fatalf("bind: %v", err) + } + + claimed, err := st.ClaimGarbage(ctx, 100) + if err != nil { + t.Fatalf("claim garbage: %v", err) + } + if !containsObject(claimed, freeID) { + t.Fatalf("unreferenced object not claimed: %+v", claimed) + } + if containsObject(claimed, boundID) { + t.Fatalf("referenced object was claimed for deletion") + } + + // Removing the last reference makes both collectible. + if _, err := pool.Exec(ctx, `DELETE FROM photos WHERE id = $1`, photo); err != nil { + t.Fatalf("delete photo: %v", err) + } + claimed, err = st.ClaimGarbage(ctx, 100) + if err != nil { + t.Fatalf("claim garbage 2: %v", err) + } + if !containsObject(claimed, boundID) { + t.Fatalf("object with no remaining references not claimed: %+v", claimed) + } +} + +func objectState(t *testing.T, ctx context.Context, pool *pgxpool.Pool, id string) string { + t.Helper() + var state string + if err := pool.QueryRow(ctx, `SELECT state FROM objects WHERE id = $1`, id).Scan(&state); err != nil { + t.Fatalf("read state: %v", err) + } + return state +} + +func containsObject(objs []Object, id string) bool { + for _, o := range objs { + if o.ID == id { + return true + } + } + return false +} diff --git a/backend/internal/photos/pipeline.go b/backend/internal/photos/pipeline.go index 2ed36f6..928315f 100644 --- a/backend/internal/photos/pipeline.go +++ b/backend/internal/photos/pipeline.go @@ -2,6 +2,8 @@ package photos import ( "bytes" + "crypto/sha256" + "encoding/hex" "fmt" "image" "strconv" @@ -138,3 +140,9 @@ func buildVariants(data []byte) (thumb, preview []byte, err error) { } return thumbBuf.Bytes(), previewBuf.Bytes(), nil } + +// checksum returns the hex-encoded SHA-256 of the given bytes. +func checksum(data []byte) string { + sum := sha256.Sum256(data) + return hex.EncodeToString(sum[:]) +} diff --git a/backend/internal/photos/service.go b/backend/internal/photos/service.go index c58bfc4..f328f3f 100644 --- a/backend/internal/photos/service.go +++ b/backend/internal/photos/service.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "errors" - "fmt" "net/http" "path/filepath" "strings" @@ -64,22 +63,20 @@ func (s *Service) Upload(ctx context.Context, in UploadInput) (*models.Photo, er } meta := extractEXIF(in.Data) - id := auth.NewID() ext := strings.ToLower(filepath.Ext(in.Filename)) if ext == "" { ext = ".bin" } - storageKey := fmt.Sprintf("albums/%s/original/%s%s", in.AlbumID, id, ext) - if err := s.storage.Put(ctx, storageKey, bytes.NewReader(in.Data), int64(len(in.Data)), mime); err != nil { - return nil, err - } + photoID := auth.NewID() + objectID := auth.NewID() + key := objectKey(objectID, RoleOriginal, ext) p := &models.Photo{ - ID: id, + ID: photoID, AlbumID: in.AlbumID, UploaderID: in.UserID, - StorageKey: storageKey, + StorageKey: key, OriginalFilename: in.Filename, Size: int64(len(in.Data)), Width: width, @@ -90,12 +87,28 @@ func (s *Service) Upload(ctx context.Context, in UploadInput) (*models.Photo, er Longitude: meta.Longitude, EXIF: meta.Attributes, } - if err := s.store.Create(ctx, p); err != nil { - return nil, err + obj := &Object{ + ID: objectID, + StorageKey: key, + MimeType: mime, + Size: int64(len(in.Data)), + Width: width, + Height: height, + Checksum: checksum(in.Data), } - data := in.Data - go s.processVariants(p.ID, in.AlbumID, data) + // The metadata layer is the ledger of record: the row is written first, + // then the bytes are stored, then the object is activated. + if err := s.store.CreateWithOriginal(ctx, p, obj); err != nil { + return nil, err + } + if err := s.storage.Put(ctx, key, bytes.NewReader(in.Data), int64(len(in.Data)), mime); err != nil { + _ = s.store.DeleteUpload(ctx, photoID, objectID) + return nil, err + } + if err := s.store.ActivateObject(ctx, objectID); err != nil { + return nil, err + } if err := s.attachURLs(ctx, p); err != nil { return nil, err @@ -103,33 +116,6 @@ func (s *Service) Upload(ctx context.Context, in UploadInput) (*models.Photo, er return p, nil } -func (s *Service) processVariants(photoID, albumID string, data []byte) { - ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) - defer cancel() - - thumb, preview, err := buildVariants(data) - if err != nil { - return - } - thumbKey := fmt.Sprintf("albums/%s/thumb/%s.webp", albumID, photoID) - previewKey := fmt.Sprintf("albums/%s/preview/%s.webp", albumID, photoID) - if err := s.storage.Put(ctx, thumbKey, bytes.NewReader(thumb), int64(len(thumb)), "image/webp"); err != nil { - return - } - if err := s.storage.Put(ctx, previewKey, bytes.NewReader(preview), int64(len(preview)), "image/webp"); err != nil { - return - } - if err := s.store.SetVariants(ctx, photoID, thumbKey, previewKey); err != nil { - return - } - s.store.SetCoverIfEmpty(ctx, albumID, photoID) - if p, err := s.store.GetByID(ctx, photoID, ""); err == nil { - if err := s.attachURLs(ctx, p); err == nil { - s.hub.Publish(albumID, "photo.created", p) - } - } -} - func (s *Service) attachURLs(ctx context.Context, p *models.Photo) error { if p.StorageKey != "" { url, err := s.storage.PresignGet(ctx, p.StorageKey) @@ -416,21 +402,22 @@ func (s *Service) PermanentDelete(ctx context.Context, photoID, userID string) e if !canDeletePhoto(role, p.UploaderID == userID) { return httpx.Forbidden("forbidden", "you cannot delete this photo") } + objectIDs, err := s.store.PhotoObjectIDs(ctx, photoID) + if err != nil { + return err + } if err := s.store.HardDelete(ctx, photoID); err != nil { return err } if p.AlbumID != "" { s.publishDeleted(p) } - if n, err := s.store.StorageKeyRefCount(ctx, p.StorageKey); err == nil && n == 0 { - keys := []string{p.StorageKey} - if p.ThumbKey != "" { - keys = append(keys, p.ThumbKey) + // Mark objects that no longer have any binding for garbage collection. + // The background worker removes the blobs from storage. + for _, oid := range objectIDs { + if n, err := s.store.ObjectRefCount(ctx, oid); err == nil && n == 0 { + _ = s.store.MarkObjectDeleting(ctx, oid) } - if p.PreviewKey != "" { - keys = append(keys, p.PreviewKey) - } - _ = s.storage.Delete(ctx, keys...) } return nil } diff --git a/backend/internal/photos/store.go b/backend/internal/photos/store.go index 446c4a6..ff59950 100644 --- a/backend/internal/photos/store.go +++ b/backend/internal/photos/store.go @@ -23,9 +23,14 @@ type Store struct { func NewStore(pool *pgxpool.Pool) *Store { return &Store{pool: pool} } -const photoColumns = `p.id, COALESCE(p.album_id, ''), p.uploader_id, p.storage_key, - COALESCE(p.thumb_key, ''), COALESCE(p.preview_key, ''), - p.original_filename, p.size, p.width, p.height, p.mime_type, +// photoColumns selects photo rows together with the keys and physical facts of +// their bound objects (original, thumb, preview) and the original's size, +// dimensions and mime type. +const photoColumns = `p.id, COALESCE(p.album_id, ''), p.uploader_id, + COALESCE(o_o.storage_key, ''), + COALESCE(o_t.storage_key, ''), + COALESCE(o_p.storage_key, ''), + p.original_filename, COALESCE(o_o.size, 0), COALESCE(o_o.width, 0), COALESCE(o_o.height, 0), COALESCE(o_o.mime_type, ''), p.taken_at, p.latitude, p.longitude, COALESCE(p.caption, ''), p.created_at, u.id, u.name, COALESCE(u.avatar_url, ''), (SELECT count(*) FROM likes l WHERE l.photo_id = p.id), @@ -36,6 +41,14 @@ const photoColumns = `p.id, COALESCE(p.album_id, ''), p.uploader_id, p.storage_k FROM photo_tags pt JOIN tags t ON t.id = pt.tag_id WHERE pt.photo_id = p.id AND t.user_id = $2), '[]')::text` +const photoJoins = ` + LEFT JOIN photo_objects po_o ON po_o.photo_id = p.id AND po_o.role = 'original' + LEFT JOIN objects o_o ON o_o.id = po_o.object_id + LEFT JOIN photo_objects po_t ON po_t.photo_id = p.id AND po_t.role = 'thumb' + LEFT JOIN objects o_t ON o_t.id = po_t.object_id + LEFT JOIN photo_objects po_p ON po_p.photo_id = p.id AND po_p.role = 'preview' + LEFT JOIN objects o_p ON o_p.id = po_p.object_id` + func scanPhoto(row pgx.Row) (*models.Photo, error) { var p models.Photo var exifText, tagsText string @@ -76,31 +89,68 @@ func setExif(p *models.Photo, raw string) { } } -func (s *Store) Create(ctx context.Context, p *models.Photo) error { +// CreateWithOriginal records a new photo together with its pending original +// object, the original binding and a variant job — all in one transaction. The +// caller stores the bytes and then activates the object. The metadata layer is +// the ledger of record; the storage layer follows. +func (s *Store) CreateWithOriginal(ctx context.Context, p *models.Photo, obj *Object) error { exifJSON, err := json.Marshal(p.EXIF) if err != nil || len(exifJSON) == 0 { exifJSON = []byte("{}") } - return s.pool.QueryRow(ctx, - `INSERT INTO photos (id, album_id, uploader_id, storage_key, original_filename, - size, width, height, mime_type, taken_at, latitude, longitude, caption, exif) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14::jsonb) + tx, err := s.pool.Begin(ctx) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + + if _, err := tx.Exec(ctx, + `INSERT INTO objects (id, storage_key, mime_type, size, width, height, checksum, state) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`, + obj.ID, obj.StorageKey, obj.MimeType, obj.Size, obj.Width, obj.Height, obj.Checksum, ObjectPending); err != nil { + return err + } + if err := tx.QueryRow(ctx, + `INSERT INTO photos (id, album_id, uploader_id, original_filename, taken_at, latitude, longitude, caption, exif) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb) RETURNING created_at`, - p.ID, p.AlbumID, p.UploaderID, p.StorageKey, p.OriginalFilename, - p.Size, p.Width, p.Height, p.MimeType, p.TakenAt, p.Latitude, p.Longitude, p.Caption, string(exifJSON), - ).Scan(&p.CreatedAt) + p.ID, p.AlbumID, p.UploaderID, p.OriginalFilename, p.TakenAt, p.Latitude, p.Longitude, p.Caption, string(exifJSON), + ).Scan(&p.CreatedAt); err != nil { + return err + } + if _, err := tx.Exec(ctx, + `INSERT INTO photo_objects (photo_id, object_id, role) VALUES ($1, $2, $3)`, + p.ID, obj.ID, RoleOriginal); err != nil { + return err + } + if _, err := tx.Exec(ctx, + `INSERT INTO variant_jobs (photo_id) VALUES ($1) ON CONFLICT (photo_id) DO NOTHING`, p.ID); err != nil { + return err + } + return tx.Commit(ctx) } -func (s *Store) SetVariants(ctx context.Context, id, thumbKey, previewKey string) error { - _, err := s.pool.Exec(ctx, - `UPDATE photos SET thumb_key = $2, preview_key = $3 WHERE id = $1`, id, thumbKey, previewKey) - return err +// DeleteUpload rolls back a failed upload: the photo (and its bindings and job) +// plus the pending object row are removed. +func (s *Store) DeleteUpload(ctx context.Context, photoID, objectID string) error { + tx, err := s.pool.Begin(ctx) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + if _, err := tx.Exec(ctx, `DELETE FROM photos WHERE id = $1`, photoID); err != nil { + return err + } + if _, err := tx.Exec(ctx, `DELETE FROM objects WHERE id = $1`, objectID); err != nil { + return err + } + return tx.Commit(ctx) } func (s *Store) GetByID(ctx context.Context, id, viewerID string) (*models.Photo, error) { return scanPhoto(s.pool.QueryRow(ctx, `SELECT `+photoColumns+` - FROM photos p JOIN users u ON u.id = p.uploader_id + FROM photos p JOIN users u ON u.id = p.uploader_id`+photoJoins+` WHERE p.id = $1`, id, viewerID)) } @@ -118,7 +168,7 @@ func (s *Store) List(ctx context.Context, in ListParams) ([]models.Photo, error) } rows, err := s.pool.Query(ctx, `SELECT `+photoColumns+` - FROM photos p JOIN users u ON u.id = p.uploader_id + FROM photos p JOIN users u ON u.id = p.uploader_id`+photoJoins+` WHERE p.album_id = $1 AND p.deleted_at IS NULL AND ($3 = '' OR (p.created_at, p.id) < (SELECT created_at, id FROM photos WHERE id = $3)) @@ -181,30 +231,36 @@ func (s *Store) UpdateAlbum(ctx context.Context, id, albumID string) error { return nil } -// Clone creates a new photo row that references the same stored objects as src. +// Clone creates a new photo row in albumID that references the same objects as +// src. No storage operation is performed: the clone shares the physical files. func (s *Store) Clone(ctx context.Context, src *models.Photo, albumID, uploaderID, id string) error { exifJSON, err := json.Marshal(src.EXIF) if err != nil || len(exifJSON) == 0 { exifJSON = []byte("{}") } - _, err = s.pool.Exec(ctx, - `INSERT INTO photos (id, album_id, uploader_id, storage_key, thumb_key, preview_key, - original_filename, size, width, height, mime_type, taken_at, latitude, longitude, caption, exif) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16::jsonb)`, - id, albumID, uploaderID, src.StorageKey, nullIfEmpty(src.ThumbKey), nullIfEmpty(src.PreviewKey), - src.OriginalFilename, src.Size, src.Width, src.Height, src.MimeType, - src.TakenAt, src.Latitude, src.Longitude, src.Caption, string(exifJSON)) - return err + tx, err := s.pool.Begin(ctx) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + + if _, err := tx.Exec(ctx, + `INSERT INTO photos (id, album_id, uploader_id, original_filename, taken_at, latitude, longitude, caption, exif) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb)`, + id, albumID, uploaderID, src.OriginalFilename, src.TakenAt, src.Latitude, src.Longitude, src.Caption, string(exifJSON)); err != nil { + return err + } + if _, err := tx.Exec(ctx, + `INSERT INTO photo_objects (photo_id, object_id, role) + SELECT $1, object_id, role FROM photo_objects WHERE photo_id = $2`, + id, src.ID); err != nil { + return err + } + return tx.Commit(ctx) } -// StorageKeyRefCount counts photo rows (any state) referencing the stored object. -func (s *Store) StorageKeyRefCount(ctx context.Context, storageKey string) (int, error) { - var n int - err := s.pool.QueryRow(ctx, `SELECT count(*) FROM photos WHERE storage_key = $1`, storageKey).Scan(&n) - return n, err -} - -// HardDelete permanently removes the photo row. +// HardDelete permanently removes the photo row. Bound objects are not touched +// here; the caller marks orphaned objects for collection. func (s *Store) HardDelete(ctx context.Context, id string) error { tag, err := s.pool.Exec(ctx, `DELETE FROM photos WHERE id = $1`, id) if err != nil { @@ -223,7 +279,7 @@ func (s *Store) ListTrash(ctx context.Context, userID string) ([]models.Photo, e `SELECT `+photoColumns+`, COALESCE(a.title, '') FROM photos p JOIN users u ON u.id = p.uploader_id - LEFT JOIN albums a ON a.id = p.album_id + LEFT JOIN albums a ON a.id = p.album_id`+photoJoins+` WHERE p.deleted_at IS NOT NULL AND ( (a.id IS NOT NULL AND (a.owner_id = $1 OR EXISTS( @@ -362,13 +418,6 @@ func (s *Store) TagOwnedBy(ctx context.Context, tagID, userID string) (bool, err return ok, err } -func nullIfEmpty(s string) *string { - if s == "" { - return nil - } - return &s -} - func (s *Store) SetCoverIfEmpty(ctx context.Context, albumID, photoID string) { _, _ = s.pool.Exec(ctx, `UPDATE albums SET cover_photo_id = $2 WHERE id = $1 AND cover_photo_id IS NULL`, @@ -391,11 +440,16 @@ type backfillRow struct { StorageKey string } +// BackfillEXIF re-reads the original object of every photo with empty exif and +// fills the exif/geo fields on the photo. func (s *Store) BackfillEXIF(ctx context.Context, fetch func(context.Context, string) ([]byte, error)) (int, error) { rows, err := s.pool.Query(ctx, - `SELECT id, storage_key FROM photos - WHERE exif IS NULL OR exif = '{}'::jsonb - ORDER BY created_at`) + `SELECT p.id, o.storage_key + FROM photos p + JOIN photo_objects po ON po.photo_id = p.id AND po.role = 'original' + JOIN objects o ON o.id = po.object_id + WHERE p.exif IS NULL OR p.exif = '{}'::jsonb + ORDER BY p.created_at`) if err != nil { return 0, err } diff --git a/backend/internal/photos/upload_test.go b/backend/internal/photos/upload_test.go new file mode 100644 index 0000000..3ccf887 --- /dev/null +++ b/backend/internal/photos/upload_test.go @@ -0,0 +1,70 @@ +package photos + +import ( + "context" + "strings" + "testing" +) + +func TestUploadCreatesLedgerAndAlbumFreeKey(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + svc, cl := newTestService(t, pool) + 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) }) + + if !strings.HasPrefix(p.StorageKey, "objects/") { + t.Fatalf("storage key %q is not an object key", p.StorageKey) + } + if strings.Contains(p.StorageKey, album) { + t.Fatalf("storage key leaks album id: %q", p.StorageKey) + } + if p.OriginalURL == "" { + t.Fatalf("original url not set") + } + if p.ThumbURL != "" || p.PreviewURL != "" { + t.Fatalf("variants should be pending, got thumb=%q preview=%q", p.ThumbURL, p.PreviewURL) + } + + orig, err := svc.store.OriginalObject(ctx, p.ID) + if err != nil { + t.Fatalf("original object: %v", err) + } + if orig.State != ObjectReady { + t.Fatalf("original object state = %q, want ready", orig.State) + } + if n, err := svc.store.ObjectRefCount(ctx, orig.ID); err != nil || n != 1 { + t.Fatalf("original ref count = %d err=%v, want 1", n, err) + } + + 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 != 1 { + t.Fatalf("variant jobs = %d, want 1", jobs) + } + + list, _, err := svc.List(ctx, album, user, "", nil, 50) + if err != nil || len(list) != 1 { + t.Fatalf("list: %v len=%d", err, len(list)) + } + if list[0].MimeType != "image/png" || list[0].Width != 4 || list[0].Height != 4 { + t.Fatalf("derived fields wrong: mime=%q %dx%d", list[0].MimeType, list[0].Width, list[0].Height) + } + if list[0].OriginalURL == "" { + t.Fatalf("list original url not set") + } +} diff --git a/backend/internal/photos/worker.go b/backend/internal/photos/worker.go new file mode 100644 index 0000000..9f1352d --- /dev/null +++ b/backend/internal/photos/worker.go @@ -0,0 +1,210 @@ +package photos + +import ( + "bytes" + "context" + "errors" + "log/slog" + "time" + + "shared-album/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 +} + +// NewWorker builds a worker. Parameters are clamped to sane defaults. +func NewWorker(svc *Service, interval time.Duration, batch int, pendingTTL time.Duration) *Worker { + if interval <= 0 { + interval = 5 * time.Second + } + if batch <= 0 { + batch = 10 + } + if pendingTTL <= 0 { + pendingTTL = 10 * time.Minute + } + return &Worker{ + svc: svc, + interval: interval, + batch: batch, + pendingTTL: pendingTTL, + lease: interval * 4, + } +} + +// 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 + } + for _, id := range ids { + w.generateVariants(ctx, id) + } +} + +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 + } + w.svc.hub.Publish(p.AlbumID, "photo.created", 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) + } +} diff --git a/backend/internal/photos/worker_test.go b/backend/internal/photos/worker_test.go new file mode 100644 index 0000000..bbd5532 --- /dev/null +++ b/backend/internal/photos/worker_test.go @@ -0,0 +1,79 @@ +package photos + +import ( + "context" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestWorkerGeneratesVariantsAndCollects(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + svc, cl := newTestService(t, pool) + 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) }) + + w := NewWorker(svc, time.Second, 10, time.Minute) + w.processVariantJobs(ctx) + + refreshed, err := svc.store.GetByID(ctx, p.ID, "") + if err != nil { + t.Fatalf("get refreshed: %v", err) + } + if refreshed.ThumbKey == "" || refreshed.PreviewKey == "" { + t.Fatalf("variants not bound: thumb=%q preview=%q", refreshed.ThumbKey, refreshed.PreviewKey) + } + if err := svc.attachURLs(ctx, refreshed); err != nil { + t.Fatalf("attach urls: %v", err) + } + if refreshed.ThumbURL == "" || 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) != 3 { + t.Fatalf("expected 3 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 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 +} diff --git a/backend/internal/repo/repo_test.go b/backend/internal/repo/repo_test.go index 9449131..df56c88 100644 --- a/backend/internal/repo/repo_test.go +++ b/backend/internal/repo/repo_test.go @@ -29,6 +29,7 @@ func testPool(t *testing.T) *pgxpool.Pool { if err := pool.Ping(context.Background()); err != nil { t.Skipf("cannot ping: %v", err) } + t.Cleanup(pool.Close) return pool } @@ -48,7 +49,6 @@ func newUser(t *testing.T, ctx context.Context, pool *pgxpool.Pool, name string) func TestUserRepository(t *testing.T) { pool := testPool(t) - defer pool.Close() ctx := context.Background() store := auth.NewStore(pool) @@ -82,7 +82,6 @@ func TestUserRepository(t *testing.T) { func TestAlbumRepository(t *testing.T) { pool := testPool(t) - defer pool.Close() ctx := context.Background() store := albums.NewStore(pool) @@ -126,7 +125,6 @@ func TestAlbumRepository(t *testing.T) { func TestAlbumCoverRepository(t *testing.T) { pool := testPool(t) - defer pool.Close() ctx := context.Background() store := albums.NewStore(pool) @@ -138,18 +136,33 @@ func TestAlbumCoverRepository(t *testing.T) { t.Fatalf("create album: %v", err) } - insertPhoto := func(id, thumbKey string) { + insertPhoto := func(id string) string { t.Helper() + origObj := auth.NewID() + thumbObj := auth.NewID() + thumbKey := "objects/" + thumbObj + "/thumb.webp" if _, err := pool.Exec(ctx, - `INSERT INTO photos (id, album_id, uploader_id, storage_key, thumb_key) - VALUES ($1, $2, $3, $4, $5)`, - id, a.ID, owner.ID, "albums/"+a.ID+"/original/"+id+".jpg", thumbKey); err != nil { + `INSERT INTO objects (id, storage_key, mime_type, state) VALUES + ($1, $2, 'image/jpeg', 'ready'), + ($3, $4, 'image/webp', 'ready')`, + origObj, "objects/"+origObj+"/original.jpg", thumbObj, thumbKey); err != nil { + t.Fatalf("insert objects: %v", err) + } + if _, err := pool.Exec(ctx, + `INSERT INTO photos (id, album_id, uploader_id) VALUES ($1, $2, $3)`, + id, a.ID, owner.ID); err != nil { t.Fatalf("insert photo: %v", err) } + if _, err := pool.Exec(ctx, + `INSERT INTO photo_objects (photo_id, object_id, role) VALUES ($1, $2, 'original'), ($1, $3, 'thumb')`, + id, origObj, thumbObj); err != nil { + t.Fatalf("insert bindings: %v", err) + } + return thumbKey } photo1 := auth.NewID() - insertPhoto(photo1, "albums/"+a.ID+"/thumb/"+photo1+".webp") + thumb1 := insertPhoto(photo1) got, err := store.GetByID(ctx, a.ID) if err != nil || got.CoverThumbKey != "" { @@ -162,12 +175,12 @@ func TestAlbumCoverRepository(t *testing.T) { if got.CoverPhotoID == nil || *got.CoverPhotoID != photo1 { t.Fatalf("cover not set: %+v", got.CoverPhotoID) } - if want := "albums/" + a.ID + "/thumb/" + photo1 + ".webp"; got.CoverThumbKey != want { - t.Fatalf("cover thumb key = %q want %q", got.CoverThumbKey, want) + if got.CoverThumbKey != thumb1 { + t.Fatalf("cover thumb key = %q want %q", got.CoverThumbKey, thumb1) } photo2 := auth.NewID() - insertPhoto(photo2, "thumb-2") + thumb2 := insertPhoto(photo2) ps.SetCoverIfEmpty(ctx, a.ID, photo2) got, _ = store.GetByID(ctx, a.ID) if got.CoverPhotoID == nil || *got.CoverPhotoID != photo1 { @@ -184,14 +197,13 @@ func TestAlbumCoverRepository(t *testing.T) { ps.SetCoverIfEmpty(ctx, a.ID, photo2) list, err := store.ListForUser(ctx, owner.ID) - if err != nil || len(list) != 1 || list[0].CoverThumbKey != "thumb-2" { + if err != nil || len(list) != 1 || list[0].CoverThumbKey != thumb2 { t.Fatalf("list cover key: %v %+v", err, list) } } func TestMemberRepository(t *testing.T) { pool := testPool(t) - defer pool.Close() ctx := context.Background() owner := newUser(t, ctx, pool, "Owner") @@ -255,7 +267,6 @@ func TestMemberRepository(t *testing.T) { func TestShareLinkRepository(t *testing.T) { pool := testPool(t) - defer pool.Close() ctx := context.Background() owner := newUser(t, ctx, pool, "Owner") diff --git a/backend/internal/storage/storage.go b/backend/internal/storage/storage.go index 19b5b93..91de32e 100644 --- a/backend/internal/storage/storage.go +++ b/backend/internal/storage/storage.go @@ -104,24 +104,6 @@ func (c *Client) Delete(ctx context.Context, keys ...string) error { return nil } -func (c *Client) DeletePrefix(ctx context.Context, prefix string) error { - var keys []string - paginator := s3.NewListObjectsV2Paginator(c.s3, &s3.ListObjectsV2Input{ - Bucket: aws.String(c.bucket), - Prefix: aws.String(prefix), - }) - for paginator.HasMorePages() { - page, err := paginator.NextPage(ctx) - if err != nil { - return fmt.Errorf("list objects: %w", err) - } - for _, o := range page.Contents { - keys = append(keys, aws.ToString(o.Key)) - } - } - return c.Delete(ctx, keys...) -} - func (c *Client) Get(ctx context.Context, key string) ([]byte, error) { out, err := c.s3.GetObject(ctx, &s3.GetObjectInput{ Bucket: aws.String(c.bucket), diff --git a/backend/migrations/000008_media_objects.down.sql b/backend/migrations/000008_media_objects.down.sql new file mode 100644 index 0000000..82da35a --- /dev/null +++ b/backend/migrations/000008_media_objects.down.sql @@ -0,0 +1,3 @@ +DROP TABLE IF EXISTS variant_jobs; +DROP TABLE IF EXISTS photo_objects; +DROP TABLE IF EXISTS objects; diff --git a/backend/migrations/000008_media_objects.up.sql b/backend/migrations/000008_media_objects.up.sql new file mode 100644 index 0000000..7dc3021 --- /dev/null +++ b/backend/migrations/000008_media_objects.up.sql @@ -0,0 +1,39 @@ +-- Physical objects stored in S3. Keys are opaque and never encode album, photo +-- or any other metadata entity. The row is the ledger entry that governs the +-- lifecycle of the corresponding blob. +CREATE TABLE IF NOT EXISTS objects ( + id text PRIMARY KEY, + storage_key text NOT NULL UNIQUE, + mime_type text NOT NULL DEFAULT '', + size bigint NOT NULL DEFAULT 0, + width integer NOT NULL DEFAULT 0, + height integer NOT NULL DEFAULT 0, + checksum text NOT NULL DEFAULT '', + state text NOT NULL DEFAULT 'pending' + CHECK (state IN ('pending', 'ready', 'deleting')), + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS objects_state_created_idx ON objects (state, created_at); + +-- Binding of a photo to its objects by role. The role is a property of the +-- binding, not of the object itself. +CREATE TABLE IF NOT EXISTS photo_objects ( + photo_id text NOT NULL REFERENCES photos(id) ON DELETE CASCADE, + object_id text NOT NULL REFERENCES objects(id) ON DELETE RESTRICT, + role text NOT NULL CHECK (role IN ('original', 'thumb', 'preview')), + PRIMARY KEY (photo_id, role) +); + +CREATE INDEX IF NOT EXISTS photo_objects_object_idx ON photo_objects (object_id); + +-- Durable queue of pending variant generation work, one job per photo. +CREATE TABLE IF NOT EXISTS variant_jobs ( + photo_id text PRIMARY KEY REFERENCES photos(id) ON DELETE CASCADE, + attempts integer NOT NULL DEFAULT 0, + next_attempt_at timestamptz NOT NULL DEFAULT now(), + last_error text NOT NULL DEFAULT '', + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS variant_jobs_due_idx ON variant_jobs (next_attempt_at); diff --git a/backend/migrations/000009_drop_photo_keys.down.sql b/backend/migrations/000009_drop_photo_keys.down.sql new file mode 100644 index 0000000..0288ae3 --- /dev/null +++ b/backend/migrations/000009_drop_photo_keys.down.sql @@ -0,0 +1,8 @@ +ALTER TABLE photos + ADD COLUMN IF NOT EXISTS storage_key text NOT NULL DEFAULT '', + ADD COLUMN IF NOT EXISTS thumb_key text, + ADD COLUMN IF NOT EXISTS preview_key text, + ADD COLUMN IF NOT EXISTS size bigint NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS width integer NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS height integer NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS mime_type text NOT NULL DEFAULT ''; diff --git a/backend/migrations/000009_drop_photo_keys.up.sql b/backend/migrations/000009_drop_photo_keys.up.sql new file mode 100644 index 0000000..72365e9 --- /dev/null +++ b/backend/migrations/000009_drop_photo_keys.up.sql @@ -0,0 +1,10 @@ +-- Physical facts move to the objects table; photos keeps only membership and +-- presentation data. size/width/height/mime_type are derived from the original +-- object via photo_objects + objects. +ALTER TABLE photos DROP COLUMN IF EXISTS storage_key; +ALTER TABLE photos DROP COLUMN IF EXISTS thumb_key; +ALTER TABLE photos DROP COLUMN IF EXISTS preview_key; +ALTER TABLE photos DROP COLUMN IF EXISTS size; +ALTER TABLE photos DROP COLUMN IF EXISTS width; +ALTER TABLE photos DROP COLUMN IF EXISTS height; +ALTER TABLE photos DROP COLUMN IF EXISTS mime_type; diff --git a/openspec/changes/media-object-layer/.openspec.yaml b/openspec/changes/media-object-layer/.openspec.yaml new file mode 100644 index 0000000..53f62d7 --- /dev/null +++ b/openspec/changes/media-object-layer/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-10-10 diff --git a/openspec/changes/media-object-layer/design.md b/openspec/changes/media-object-layer/design.md new file mode 100644 index 0000000..bcf74c6 --- /dev/null +++ b/openspec/changes/media-object-layer/design.md @@ -0,0 +1,134 @@ +# Design + +## Context + +См. `proposal.md` — Why. Текущее состояние, которое формирует подход: + +- `photos` хранит `storage_key`, `thumb_key`, `preview_key` (`migrations/000001_init.up.sql:45-47`), + а ключи строятся как `albums/{album_id}/...` (`internal/photos/service.go:72,114,115`). +- `Clone` копирует строку фото вместе с ключами, `MoveBatch` меняет только `album_id`, + `PermanentDelete` удаляет объекты по счётчику ссылок `StorageKeyRefCount` + (`internal/photos/store.go:201`). +- `DeletePrefix` (`internal/storage/storage.go:107`) существует, но не вызывается. +- Очередь вариантов отсутствует: генерация запускается goroutine из HTTP-запроса + (`internal/photos/service.go:98`), результат не переживает рестарт. +- Внешний API уже отдаёт `thumb_url/preview_url/original_url` и + `size/width/height/mime_type`; это контракт, который надо сохранить. +- В БД только дев-данные, миграция может быть разрушительной. + +## Goals / Non-Goals + +**Goals:** +- Убрать сущность «альбом» (и любую модель метаданных) из ключей и операций S3. +- Сделать БД журналом истины: физический объект существует, пока на него есть ссылка. +- Дать явный, устойчивый к сбоям сборщик мусора и самовосстанавливающуюся генерацию вариантов. +- Сохранить внешний JSON API без изменений. + +**Non-Goals:** +- Content-addressing и дедупликация одинаковых байтов между разными загрузками + (задел — колонка `checksum`, политика позже). +- Мультивариантность сверх фиксированных original/thumb/preview (join-таблица это выдержит). +- Перенос существующих данных: миграция разрушительная. + +## Decisions + +### D1. Join-таблица `photo_objects` вместо колонок-FK +Фото ссылается на объекты через `photo_objects(photo_id, object_id, role)` с +`PK(photo_id, role)`. Роль — свойство **связки**, не объекта: сам объект не знает, +кто он. Альтернатива (три FK-колонки на `photos`) проще, но фиксирует ровно три +варианта и снова смешивает представление с физикой. Выбран вариант, который +расширяется до нескольких размеров/видео без переделки схемы. + +### D2. Разделение фактов: физические vs логические +`objects` — только физические факты: `storage_key`, `mime_type`, `size`, +`width`, `height`, `checksum`, `state`, `created_at`. +`photos` — членство и представление: `album_id`, `uploader_id`, +`original_filename`, `caption`, `exif`, `taken_at`, `latitude`, `longitude`, +`deleted_at`, `created_at`. +`size/width/height/mime_type` в API выводятся джойном к объекту-оригиналу. + +```sql +objects( + id text pk, storage_key text unique not null, mime_type text, + size bigint, width int, height int, checksum text, + state text, created_at timestamptz) + +photo_objects( + photo_id -> photos(id) on delete cascade, + object_id -> objects(id) on delete restrict, + role text check (role in ('original','thumb','preview')), + primary key (photo_id, role)) +create index on photo_objects(object_id); + +variant_jobs( + photo_id -> photos(id) on delete cascade, + attempts int, next_attempt_at timestamptz, last_error text, + primary key (photo_id)) +``` + +У `photos` удаляются `storage_key`, `thumb_key`, `preview_key`. + +### D3. Album-free ключи, случайный id объекта +Ключ: `objects/{object_id}/{role}.{ext}`, где `object_id` — случайный +(`auth.NewID()`). Роль в ключе — косметика для отладки, истина о роли — в +`photo_objects`. Бакет, region, path-style не меняются. `DeletePrefix` удаляется. + +### D4. Ledger состояний и явный GC +Объект: `pending -> ready -> deleting`. Порядок операций — **сначала строка, потом +Put**, поэтому неубиваемый мусор «файл есть, строки нет» не возникает. + +``` + Upload : INSERT objects(pending) -> S3 Put -> ready -> bind original + INSERT variant_jobs + Copy : INSERT photos + bind те же objects -- 0 S3 + Move : UPDATE photos.album_id -- 0 S3 + Delete : tx DELETE photos -> cascade unbind -> mark orphaned 'deleting' + Worker : deleting -> re-check 0 ссылок -> S3 Delete -> DELETE row + pending старше T (Put не завершился) -> deleting -> S3 Delete -> DELETE row +``` + +Ссылочные гонки закрываются блокировками: +- Пометить `deleting` можно только объект с нулём связок, под `SELECT ... FOR UPDATE`. +- Перед S3 Delete воркер **перепроверяет** ноль связок под `FOR UPDATE`. +- Привязка объекта (bind) берёт `FOR UPDATE` и отвергает `state != 'ready'`. + +### D5. Генерация вариантов через `variant_jobs` +Вместо goroutine из запроса — строка задания в той же транзакции, что и загрузка. +Воркер забирает задания через `FOR UPDATE SKIP LOCKED`, регенерирует thumb/preview, +создаёт объекты (pending->Put->ready), вставляет связки, публикует SSE +`photo.created`, при ошибке увеличивает `attempts` и переносит `next_attempt_at`. +Это даёт однократное выполнение, ретраи и восстановление после рестарта. + +### D6. Один фоновый воркер в том же бинаре +Петля-тикер внутри процесса (соответствует модели «один бинарь + systemd»): +(1) доделать варианты, (2) удалить `deleting`, (3) дососить `pending`. Интервал и +батч — конфигурируемые. Пометка объектов при удалении — синхронно в транзакции. + +## Risks / Trade-offs + +- [Двойная генерация вариантов после рестарта] → задание живёт в `variant_jobs`, + забирается под `SKIP LOCKED`; повторное создание объекта безопасно, лишний объект + соберёт GC. +- [Удаление объекта, который вот-вот свяжут] → `FOR UPDATE` на объекте в bind и + повторная проверка ссылок в GC. +- [Строка `photos` есть, объекта нет при сбое Put] → состояние `pending` + sweeper. +- [Джойн добавляет стоимость в список фото] → один join к объекту-оригиналу; индексы + по `photo_objects(photo_id)` (PK) и `photo_objects(object_id)`. +- [Ошибка в GC удалит живые байты необратимо] → удаляем только при доказанном нуле + ссылок; удаление идемпотентно, состояние фиксируется до S3-операции. +- [API-совместимость] → JSON-поля сохраняются, меняется только источник значений. + +## Migration Plan + +Разрушительная (только дев-данные): + +1. Миграция создаёт `objects`, `photo_objects`, `variant_jobs`. +2. Миграция удаляет `photos.storage_key`, `photos.thumb_key`, `photos.preview_key`. +3. Существующие строки фото и объекты S3 не переносятся. +4. Откат: down-миграция возвращает колонки и удаляет новые таблицы. +5. `project.md` §10 приводится к новой структуре ключей. + +## Open Questions + +- Точные значения интервала воркера, размера батча и порога `T` для `pending` — + подбираются при реализации, на схему и контракты не влияют. +- Формат будущего дедупа по `checksum` (политика, не схема) — вне этого change. diff --git a/openspec/changes/media-object-layer/proposal.md b/openspec/changes/media-object-layer/proposal.md new file mode 100644 index 0000000..169aca7 --- /dev/null +++ b/openspec/changes/media-object-layer/proposal.md @@ -0,0 +1,49 @@ +# Proposal + +## Why + +Слой хранения S3 сейчас знает о сущности «альбом»: ключи объектов имеют вид +`albums/{album_id}/original/{photo_id}.{ext}` (`photos/service.go:72,114,115`). +Это протекание модели метаданных в физический слой: перенос фото оставляет ключ в +старом namespace, копия делит объекты под префиксом исходного альбома, а +неиспользуемый `DeletePrefix` (`storage/storage.go:107`) создаёт латентный риск +удаления живых объектов. Нужно развести слои: S3 хранит непрозрачные объекты, а +сущности и их жизненный цикл живут в БД. + +## What Changes + +- Ввести таблицу `objects` — физические блобы (ключ, mime, размер, габариты, + checksum, состояние) и `photo_objects` — связка photo↔object с ролью + (`original|thumb|preview`). +- **BREAKING**: удалить `storage_key`, `thumb_key`, `preview_key` из `photos`; + ключи фото выводятся через джойн к объекту-оригиналу. Миграция разрушительная + (в базе только дев-данные). +- Сделать ключи S3 album-free: `objects/{id}/{role}.{ext}`, где `id` — случайный + идентификатор объекта. +- Ввести явный GC через ledger-состояния объектов `pending | ready | deleting`: + физический объект удаляется только когда на него не осталось ни одной связки. +- Перенести генерацию вариантов в фоновый воркер: таблица `variant_jobs` + (ровно-один-раз, ретраи с backoff), воркер также удаляет осиротевшие объекты и + досасывает зависшие `pending`. SSE `photo.created` публикует воркер. +- Удалить `DeletePrefix` из клиента хранения — entity-namespace больше нет. + +## Capabilities + +### New Capabilities +- `media-objects`: физическое объектное хранилище с непрозрачными ключами, журнал + объектов и их жизненный цикл, сборка мусора по отсутствию ссылок, привязка + объектов к фото по ролям. + +### Modified Capabilities +- `photos`: уточняется семантика удаления (корзина не удаляет объекты хранилища) и + разделение объектов между фото при переносе и копировании. + +## Impact + +- Backend: `internal/photos/{service,store}.go`, `internal/storage/storage.go`, + `internal/albums/*` (косвенно, через ключи обложки), wiring фонового воркера, + `migrations/` (новая схема + удаление колонок). +- Данные: разрушительная миграция `photos` (dev-only), новые таблицы. +- Внешний JSON API не меняется: `thumb_url/preview_url/original_url` и + `size/width/height/mime_type` по-прежнему отдаются, но выводятся из объектов. +- Документация: `project.md` §10 (структура в S3). diff --git a/openspec/changes/media-object-layer/specs/media-objects/spec.md b/openspec/changes/media-object-layer/specs/media-objects/spec.md new file mode 100644 index 0000000..5e6b6a5 --- /dev/null +++ b/openspec/changes/media-object-layer/specs/media-objects/spec.md @@ -0,0 +1,90 @@ +# Spec Delta + +## Purpose + +Физическое объектное хранилище с непрозрачными ключами и журналом объектов: +привязка объектов к фото по ролям, совместное использование объектов несколькими +фото и сборка мусора по отсутствию ссылок. + +## ADDED Requirements + +### Requirement: Непрозрачные ключи хранилища +Слой хранения SHALL адресоваться непрозрачными ключами объектов, не содержащими +идентификаторов сущностей метаданных (таких как альбом или фото); сопоставление +ключей и сущностей SHALL храниться только в слое метаданных. + +#### Scenario: Ключ без альбома +- **WHEN** система сохраняет объект в хранилище +- **THEN** ключ объекта не содержит идентификатора альбома + +#### Scenario: Перенос не трогает ключи +- **WHEN** фото переносится в другой альбом +- **THEN** ключи его объектов остаются неизменными + +### Requirement: Журнал объектов +Каждый физический объект SHALL иметь ровно одну запись в слое метаданных с ключом, +MIME-типом, размером, габаритами и состоянием; фото SHALL ссылаться на объекты по +ролям original, thumb и preview. + +#### Scenario: Привязка ролей +- **WHEN** фото загружено и обработано +- **THEN** фото ссылается на объект-оригинал и, когда они готовы, на объекты thumb и preview + +#### Scenario: Оригинал без вариантов +- **WHEN** варианты ещё не сгенерированы +- **THEN** фото доступно с оригиналом, а thumb и preview отсутствуют + +### Requirement: Совместные объекты +Несколько фото SHALL иметь возможность ссылаться на один и тот же физический +объект; перенос и копирование SHALL изменять только ссылки и SHALL не изменять +объекты хранилища. + +#### Scenario: Копирование +- **WHEN** пользователь копирует фото в другой альбом +- **THEN** создаётся новая запись, ссылающаяся на те же объекты, без операций с хранилищем + +#### Scenario: Перенос +- **WHEN** пользователь переносит фото в другой альбом +- **THEN** изменяется только привязка к альбому, объекты хранилища не меняются + +### Requirement: Сборка мусора по ссылкам +Объект хранилища SHALL удаляться только когда на него не ссылается ни одно фото; +удаление альбома и попадание фото в корзину SHALL сохранять объекты. + +#### Scenario: Последняя ссылка удалена +- **WHEN** фото удаляется навсегда и больше ни одно фото не ссылается на его объекты +- **THEN** объекты удаляются из хранилища + +#### Scenario: Остались ссылки +- **WHEN** на объект ссылается другое фото +- **THEN** объект сохраняется в хранилище + +#### Scenario: Удаление альбома +- **WHEN** владелец удаляет альбом +- **THEN** объекты хранилища не удаляются + +### Requirement: Устойчивость жизненного цикла объекта +Объект SHALL иметь состояние жизненного цикла, и незавершённые операции SHALL +самовосстанавливаться: неудачная загрузка не оставляет осиротевших объектов, а +удаление объекта SHALL доводиться до конца после сбоя. + +#### Scenario: Сбой загрузки +- **WHEN** запись объекта создана, но загрузка в хранилище не завершилась +- **THEN** объект очищается без ручного вмешательства + +#### Scenario: Сбой удаления +- **WHEN** удаление объекта из хранилища не удалось +- **THEN** попытка повторяется до успешного удаления + +### Requirement: Генерация вариантов +Генерация thumb и preview SHALL выполняться через надёжную очередь заданий с +однократным выполнением и повтором при сбое; готовый результат SHALL +транслироваться подписчикам альбома. + +#### Scenario: Повтор после рестарта +- **WHEN** процесс перезапускается с незавершёнными заданиями +- **THEN** генерация вариантов продолжается + +#### Scenario: Событие после генерации +- **WHEN** варианты фото готовы +- **THEN** подписчикам альбома публикуется событие с этим фото diff --git a/openspec/changes/media-object-layer/specs/photos/spec.md b/openspec/changes/media-object-layer/specs/photos/spec.md new file mode 100644 index 0000000..296a147 --- /dev/null +++ b/openspec/changes/media-object-layer/specs/photos/spec.md @@ -0,0 +1,14 @@ +# Spec Delta + +## MODIFIED Requirements + +### Requirement: Удаление фото +Автор SHALL иметь возможность удалить своё фото; владелец и админ SHALL иметь возможность удалить любое фото. Удаление SHALL помещать фото в корзину и SHALL не удалять объекты хранилища. + +#### Scenario: Удаление своего фото +- **WHEN** автор удаляет своё фото +- **THEN** фото помещается в корзину, а объекты хранилища сохраняются + +#### Scenario: Удаление чужого фото участником +- **WHEN** участник с ролью member пытается удалить чужое фото +- **THEN** система отклоняет запрос diff --git a/openspec/changes/media-object-layer/tasks.md b/openspec/changes/media-object-layer/tasks.md new file mode 100644 index 0000000..72b16df --- /dev/null +++ b/openspec/changes/media-object-layer/tasks.md @@ -0,0 +1,37 @@ +# Tasks + +## 1. Схема и миграции + +- [x] 1.1 Добавить миграцию, создающую `objects`, `photo_objects` (PK `photo_id,role`, индекс `object_id`) и `variant_jobs`; проверить, что `make migrate` применяется и таблицы видны в БД +- [x] 1.2 Добавить миграцию, удаляющую `photos.storage_key`, `photos.thumb_key`, `photos.preview_key`, `photos.size`, `photos.width`, `photos.height`, `photos.mime_type`; проверить, что up и down миграции проходят в обе стороны без ошибок + +## 2. Слой хранения + +- [x] 2.1 Удалить `DeletePrefix` из `internal/storage/storage.go`; проверить `go build ./...` и отсутствие вызовов метода в коде +- [x] 2.2 Ввести хелпер ключа объекта вида `objects/{id}/{role}.{ext}` без идентификатора альбома; покрыть юнит-тестом, что ключ не содержит album_id + +## 3. Метаданные объектов + +- [x] 3.1 Реализовать стор объектов: создание `pending`, активация `ready`, пометка `deleting`, удаление строки, выборка `pending` старше порога; покрыть тестами переходы состояний +- [x] 3.2 Реализовать привязку `photo_objects` с блокировкой объекта `FOR UPDATE` и отказом при `state != 'ready'`; тест на повторную привязку одного объекта к двум фото +- [x] 3.3 Реализовать запрос GC (объекты `ready` без связок) и тесты сценариев «последняя ссылка удалена» и «остались ссылки» + +## 4. Сервис фото + +- [x] 4.1 Переписать `Upload`: строка объекта `pending` → S3 Put → `ready` → привязка `original` + задание в `variant_jobs`; тест, что загрузка даёт фото с URL оригинала и ключом без альбома +- [x] 4.2 Переписать `List`/`Get`/`attachURLs` на вывод ключей и `size/width/height/mime_type` джойном к объекту-оригиналу; тест, что JSON-поля и URL присутствуют как раньше +- [x] 4.3 Переписать `CopyBatch`/`MoveBatch` на работу только через связки; тест, что объекты хранилища не меняются и делятся между фото +- [x] 4.4 Переписать `PermanentDelete`: удалить связки и пометить осиротевшие объекты `deleting`; тест, что объект с нулём ссылок становится кандидатом на GC +- [x] 4.5 Убрать обращения к удалённым колонкам во всех запросах и моделях; проверить `go build ./...` и прохождение `internal/repo` тестов + +## 5. Фоновый воркер + +- [x] 5.1 Реализовать цикл забора `variant_jobs` через `FOR UPDATE SKIP LOCKED`, генерацию thumb/preview, привязку и публикацию SSE; тест обработки задания и повтора при ошибке +- [x] 5.2 Реализовать в воркере удаление `deleting` (с повторной проверкой нуля ссылок) и досасывание зависших `pending`; тест повторной попытки S3 Delete и очистки `pending` +- [x] 5.3 Встроить старт/останов воркера в жизненный цикл приложения; проверить, что сервер запускается и воркер выполняет петлю (по логам/интеграционному тесту) + +## 6. Интеграция и документация + +- [x] 6.1 Перевести обложку альбома (`CoverThumbKey`) на вывод через `photo_objects`+`objects`; тест, что список альбомов отдаёт URL обложки +- [x] 6.2 Обновить `project.md` §10 (структура в S3 и слои); проверить, что описание совпадает с реализованными ключами +- [x] 6.3 Сквозная проверка: загрузка → генерация вариантов → копирование между альбомами (0 операций S3) → перенос → окончательное удаление → сборка объектов; проверить, что объекты удаляются только при нуле ссылок diff --git a/project.md b/project.md index 929bc3c..bf0bd8b 100644 --- a/project.md +++ b/project.md @@ -328,23 +328,41 @@ colors: { - **albums** (id, owner_id, title, description, cover_photo_id, privacy, created_at) - **album_members** (album_id, user_id, role, invited_by, joined_at) - **invites** (id, album_id, token, created_by, expires_at, max_uses, uses) -- **photos** (id, album_id, uploader_id, storage_key, thumb_key, preview_key, - original_filename, size, width, height, mime_type, - taken_at, latitude, longitude, caption, created_at) +- **photos** (id, album_id, uploader_id, original_filename, + taken_at, latitude, longitude, caption, exif, deleted_at, created_at) +- **objects** (id, storage_key, mime_type, size, width, height, checksum, state, created_at) +- **photo_objects** (photo_id, object_id, role) +- **variant_jobs** (photo_id, attempts, next_attempt_at, last_error) - **comments** (id, photo_id, user_id, text, created_at, updated_at) - **likes** (photo_id, user_id, created_at) - **notifications** (id, user_id, type, payload_json, read_at, created_at) -## 10. Структура в S3 (Garage) +## 10. Слой хранения и слой метаданных + +**S3 (Garage) — непрозрачное объектное хранилище.** Ключи не содержат сущностей +метаданных (альбома, фото). Каждый физический файл адресуется своим `object_id`: ``` {bucket}/ -└── albums/{album_id}/ - ├── original/{photo_id}.{ext} - ├── preview/{photo_id}.webp - └── thumb/{photo_id}.webp +└── objects/{object_id}/ + ├── original.{ext} + ├── thumb.webp + └── preview.webp ``` +**Postgres — журнал истины.** Таблица `objects` описывает каждый блоб (ключ, mime, +размер, габариты, checksum, состояние `pending|ready|deleting`), таблица +`photo_objects` связывает фото с объектами по ролям (`original|thumb|preview`), +таблица `variant_jobs` — очередь генерации вариантов. Альбом, принадлежность и +представление живут только в `photos`/`albums`. + +Следствия: +- **Копирование** между альбомами — только новые строки `photos`+`photo_objects`, + ссылающиеся на те же объекты; операции с S3 не выполняются. +- **Перенос** — только `UPDATE photos.album_id`; объекты не трогаются. +- **Удаление** фото удаляет строки; фоновый воркер удаляет блоб лишь когда на + объект не осталось ни одной ссылки (сборка мусора по ledger-состояниям). + Доступ: **presigned URL** со сроком 1 час (бакет приватный). Garage поддерживает presigned URL через SigV4 query-string signatures.