package photos import ( "context" "errors" "time" "github.com/jackc/pgx/v5" ) // SetEditParams stores the non-destructive editing parameters of a photo. func (s *Store) SetEditParams(ctx context.Context, photoID string, params []byte) error { _, err := s.pool.Exec(ctx, `UPDATE photos SET edit_params = $2::jsonb WHERE id = $1`, photoID, string(params)) return err } // ClearEditParams removes the editing parameters of a photo, returning it to // its unedited state. func (s *Store) ClearEditParams(ctx context.Context, photoID string) error { _, err := s.pool.Exec(ctx, `UPDATE photos SET edit_params = '{}'::jsonb WHERE id = $1`, photoID) return err } // EnqueueEditJob schedules the render of the edited version of a photo. The // job is keyed by the photo, so a new edit replaces the pending one. func (s *Store) EnqueueEditJob(ctx context.Context, photoID string) error { _, err := s.pool.Exec(ctx, `INSERT INTO edit_jobs (photo_id) VALUES ($1) ON CONFLICT (photo_id) DO UPDATE SET attempts = 0, next_attempt_at = now(), last_error = ''`, photoID) return err } // DeleteEditJob drops the pending render of a photo. func (s *Store) DeleteEditJob(ctx context.Context, photoID string) error { _, err := s.pool.Exec(ctx, `DELETE FROM edit_jobs WHERE photo_id = $1`, photoID) return err } // ClaimEditJobs leases due edit renders 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) ClaimEditJobs(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 ej.photo_id FROM edit_jobs ej JOIN photos p ON p.id = ej.photo_id WHERE ej.next_attempt_at <= now() ORDER BY (ej.attempts > 0) ASC, ej.next_attempt_at ASC, ej.photo_id DESC LIMIT $1 FOR UPDATE OF ej 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 edit_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 } // CompleteEditJob removes a finished edit render. func (s *Store) CompleteEditJob(ctx context.Context, photoID string) error { _, err := s.pool.Exec(ctx, `DELETE FROM edit_jobs WHERE photo_id = $1`, photoID) return err } // FailEditJob records a failure and schedules a retry with exponential // backoff. After maxAttempts the job is abandoned. func (s *Store) FailEditJob(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 edit_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 edit_jobs WHERE photo_id = $1`, photoID); err != nil { return err } } return tx.Commit(ctx) } // DeleteObjectBindings removes the bindings of a photo for the given roles. // The objects themselves become unreferenced and are reclaimed by the garbage // collector. func (s *Store) DeleteObjectBindings(ctx context.Context, photoID string, roles ...string) error { if len(roles) == 0 { return nil } _, err := s.pool.Exec(ctx, `DELETE FROM photo_objects WHERE photo_id = $1 AND role = ANY($2)`, photoID, roles) return err }