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

package photos

import (
	"context"
	"database/sql"
	"errors"
	"time"

	"galo/internal/sqliteutil"
)

// SetEditParams stores the non-destructive editing parameters of a photo.
func (s *SQLiteStore) SetEditParams(ctx context.Context, photoID string, params []byte) error {
	_, err := s.db.ExecContext(ctx, `UPDATE photos SET edit_params = $2 WHERE id = $1`, photoID, string(params))
	return err
}

// ClearEditParams removes the editing parameters of a photo, returning it to
// its unedited state.
func (s *SQLiteStore) ClearEditParams(ctx context.Context, photoID string) error {
	_, err := s.db.ExecContext(ctx, `UPDATE photos SET edit_params = '{}' 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 *SQLiteStore) EnqueueEditJob(ctx context.Context, photoID string) error {
	_, err := s.db.ExecContext(ctx,
		`INSERT INTO edit_jobs (photo_id) VALUES ($1)
		 ON CONFLICT (photo_id) DO UPDATE SET attempts = 0, next_attempt_at = $2, last_error = ''`,
		photoID, sqliteutil.FormatTime(nowUTC()))
	return err
}

// DeleteEditJob drops the pending render of a photo.
func (s *SQLiteStore) DeleteEditJob(ctx context.Context, photoID string) error {
	_, err := s.db.ExecContext(ctx, `DELETE FROM edit_jobs WHERE photo_id = $1`, photoID)
	return err
}

// ClaimEditJobs leases due edit renders and returns their photo ids.
func (s *SQLiteStore) ClaimEditJobs(ctx context.Context, limit int, lease time.Duration) ([]string, error) {
	if limit <= 0 {
		limit = 10
	}
	now := sqliteutil.FormatTime(nowUTC())
	next := sqliteutil.FormatTime(nowUTC().Add(lease))
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback() }()

	rows, err := tx.QueryContext(ctx,
		`SELECT ej.photo_id FROM edit_jobs ej
		 JOIN photos p ON p.id = ej.photo_id
		 WHERE ej.next_attempt_at <= $1
		 ORDER BY (ej.attempts > 0) ASC, ej.next_attempt_at ASC, ej.photo_id DESC
		 LIMIT $2`, now, 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()
	}
	args := make([]any, 0, len(ids)+1)
	args = append(args, next)
	for _, id := range ids {
		args = append(args, id)
	}
	if _, err := tx.ExecContext(ctx,
		`UPDATE edit_jobs SET next_attempt_at = $1 WHERE photo_id IN (`+sqliteutil.Placeholders(2, len(ids))+`)`,
		args...); err != nil {
		return nil, err
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return ids, nil
}

// CompleteEditJob removes a finished edit render.
func (s *SQLiteStore) CompleteEditJob(ctx context.Context, photoID string) error {
	_, err := s.db.ExecContext(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 *SQLiteStore) FailEditJob(ctx context.Context, photoID, errMsg string, maxAttempts int) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	var attempts int
	err = tx.QueryRowContext(ctx,
		`UPDATE edit_jobs
		 SET attempts = attempts + 1,
		     next_attempt_at = $2,
		     last_error = $3
		 WHERE photo_id = $1
		 RETURNING attempts`, photoID,
		sqliteutil.FormatTime(nowUTC().Add(backoff(attempts))), errMsg).Scan(&attempts)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return tx.Commit()
		}
		return err
	}
	if maxAttempts > 0 && attempts >= maxAttempts {
		if _, err := tx.ExecContext(ctx, `DELETE FROM edit_jobs WHERE photo_id = $1`, photoID); err != nil {
			return err
		}
	}
	return tx.Commit()
}

// backoff returns the delay before the next attempt of a failed job.
func backoff(attempts int) time.Duration {
	d := 10 * time.Second
	for i := 0; i < attempts && d < 5*time.Minute; i++ {
		d *= 2
	}
	if d > 5*time.Minute {
		d = 5 * time.Minute
	}
	return d
}

// 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 *SQLiteStore) DeleteObjectBindings(ctx context.Context, photoID string, roles ...string) error {
	if len(roles) == 0 {
		return nil
	}
	args := make([]any, 0, len(roles)+1)
	args = append(args, photoID)
	for _, role := range roles {
		args = append(args, role)
	}
	_, err := s.db.ExecContext(ctx,
		`DELETE FROM photo_objects WHERE photo_id = $1 AND role IN (`+sqliteutil.Placeholders(2, len(roles))+`)`,
		args...)
	return err
}