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) }