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

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 vj.photo_id FROM variant_jobs vj
		 JOIN photos p ON p.id = vj.photo_id
		 WHERE vj.next_attempt_at <= now()
		 ORDER BY (vj.attempts > 0) ASC,
		          CASE WHEN vj.attempts = 0 THEN p.created_at END DESC NULLS LAST,
		          vj.next_attempt_at ASC,
		          vj.photo_id DESC
		 LIMIT $1
		 FOR UPDATE OF vj 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)
}

// AllStorageKeys returns every key registered in the object journal.
func (s *Store) AllStorageKeys(ctx context.Context) ([]string, error) {
	rows, err := s.pool.Query(ctx, `SELECT storage_key FROM objects`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var keys []string
	for rows.Next() {
		var k string
		if err := rows.Scan(&k); err != nil {
			return nil, err
		}
		keys = append(keys, k)
	}
	return keys, rows.Err()
}