ds-galo/backend/internal/photos/edit_store.go

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
}