package photos import ( "context" "database/sql" "errors" "time" "shared-album/internal/sqliteutil" ) // CreateObject inserts a new object row in the pending state. func (s *SQLiteStore) CreateObject(ctx context.Context, o *Object) error { if o.State == "" { o.State = ObjectPending } var createdAt string err := s.db.QueryRowContext(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(&createdAt) if err != nil { return err } o.CreatedAt = sqliteutil.ParseTime(createdAt) return nil } // SetObjectState updates an object's lifecycle state. func (s *SQLiteStore) SetObjectState(ctx context.Context, id, state string) error { res, err := s.db.ExecContext(ctx, `UPDATE objects SET state = $2 WHERE id = $1`, id, state) if err != nil { return err } if n, _ := res.RowsAffected(); n == 0 { return ErrObjectNotFound } return nil } // ActivateObject marks an object ready after its bytes have been stored. func (s *SQLiteStore) ActivateObject(ctx context.Context, id string) error { return s.SetObjectState(ctx, id, ObjectReady) } // MarkObjectDeleting marks an object as pending removal from storage. func (s *SQLiteStore) 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 *SQLiteStore) DeleteObjectRow(ctx context.Context, id string) error { res, err := s.db.ExecContext(ctx, `DELETE FROM objects WHERE id = $1`, id) if err != nil { return err } if n, _ := res.RowsAffected(); n == 0 { return ErrObjectNotFound } return nil } // BindObject attaches a ready object to a photo under the given role. The write // lock of the immediate transaction serializes it against garbage collection. func (s *SQLiteStore) BindObject(ctx context.Context, photoID, objectID, role string) error { tx, err := s.db.BeginTx(ctx, nil) if err != nil { return err } defer func() { _ = tx.Rollback() }() var state string if err := tx.QueryRowContext(ctx, `SELECT state FROM objects WHERE id = $1`, objectID).Scan(&state); err != nil { if errors.Is(err, sql.ErrNoRows) { return ErrObjectNotFound } return err } if state != ObjectReady { return ErrObjectNotReady } if _, err := tx.ExecContext(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() } // PhotoObjectIDs returns the object ids currently bound to a photo. func (s *SQLiteStore) PhotoObjectIDs(ctx context.Context, photoID string) ([]string, error) { rows, err := s.db.QueryContext(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 *SQLiteStore) ObjectRefCount(ctx context.Context, objectID string) (int, error) { var n int err := s.db.QueryRowContext(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 *SQLiteStore) OriginalObject(ctx context.Context, photoID string) (*Object, error) { var o Object var createdAt string err := s.db.QueryRowContext(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, &createdAt) if err != nil { if errors.Is(err, sql.ErrNoRows) { return nil, ErrObjectNotFound } return nil, err } o.CreatedAt = sqliteutil.ParseTime(createdAt) 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 *SQLiteStore) PhotosByOriginalObject(ctx context.Context, originalObjectID string) ([]string, error) { rows, err := s.db.QueryContext(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. func (s *SQLiteStore) ClaimGarbage(ctx context.Context, limit int) ([]Object, error) { if limit <= 0 { limit = 10 } tx, err := s.db.BeginTx(ctx, nil) if err != nil { return nil, err } defer func() { _ = tx.Rollback() }() rows, err := tx.QueryContext(ctx, `SELECT o.id, o.storage_key FROM objects o WHERE o.state IN ('ready', 'deleting') AND NOT EXISTS (SELECT 1 FROM photo_objects po WHERE po.object_id = o.id) ORDER BY o.created_at LIMIT $1`, limit) if err != nil { return nil, err } objs, err := scanObjectKeys(rows) if err != nil { return nil, err } if len(objs) == 0 { return nil, tx.Commit() } if err := markDeleting(ctx, tx, objs); err != nil { return nil, err } if err := tx.Commit(); err != nil { return nil, err } return objs, nil } // ClaimStalePending marks abandoned pending objects as deleting and returns them. func (s *SQLiteStore) ClaimStalePending(ctx context.Context, olderThan time.Duration, limit int) ([]Object, error) { if limit <= 0 { limit = 10 } threshold := sqliteutil.FormatTime(time.Now().UTC().Add(-olderThan)) tx, err := s.db.BeginTx(ctx, nil) if err != nil { return nil, err } defer func() { _ = tx.Rollback() }() rows, err := tx.QueryContext(ctx, `SELECT id, storage_key FROM objects WHERE state = 'pending' AND created_at < $1 ORDER BY created_at LIMIT $2`, threshold, limit) if err != nil { return nil, err } objs, err := scanObjectKeys(rows) if err != nil { return nil, err } if len(objs) == 0 { return nil, tx.Commit() } if err := markDeleting(ctx, tx, objs); err != nil { return nil, err } if err := tx.Commit(); err != nil { return nil, err } return objs, nil } // ClaimVariantJobs leases due variant jobs and returns their photo ids. func (s *SQLiteStore) ClaimVariantJobs(ctx context.Context, limit int, lease time.Duration) ([]string, error) { if limit <= 0 { limit = 10 } now := sqliteutil.FormatTime(time.Now().UTC()) next := sqliteutil.FormatTime(time.Now().UTC().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 photo_id FROM variant_jobs WHERE next_attempt_at <= $1 ORDER BY next_attempt_at 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 variant_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 } // CompleteVariantJob removes a finished job. func (s *SQLiteStore) CompleteVariantJob(ctx context.Context, photoID string) error { _, err := s.db.ExecContext(ctx, `DELETE FROM variant_jobs WHERE photo_id = $1`, photoID) return err } // FailVariantJob records a failure and schedules a retry with exponential // backoff, abandoning the job after maxAttempts. func (s *SQLiteStore) FailVariantJob(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, `SELECT attempts FROM variant_jobs WHERE photo_id = $1`, photoID).Scan(&attempts) if err != nil { if errors.Is(err, sql.ErrNoRows) { return tx.Commit() } return err } newAttempts := attempts + 1 if maxAttempts > 0 && newAttempts >= maxAttempts { if _, err := tx.ExecContext(ctx, `DELETE FROM variant_jobs WHERE photo_id = $1`, photoID); err != nil { return err } } else { secs := 10 * (1 << attempts) if secs > 300 { secs = 300 } next := sqliteutil.FormatTime(time.Now().UTC().Add(time.Duration(secs) * time.Second)) if _, err := tx.ExecContext(ctx, `UPDATE variant_jobs SET attempts = $2, next_attempt_at = $3, last_error = $4 WHERE photo_id = $1`, photoID, newAttempts, next, errMsg); err != nil { return err } } return tx.Commit() } func scanObjectKeys(rows *sql.Rows) ([]Object, error) { 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() return objs, rows.Err() } func markDeleting(ctx context.Context, tx *sql.Tx, objs []Object) error { args := make([]any, len(objs)) for i, o := range objs { args[i] = o.ID } _, err := tx.ExecContext(ctx, `UPDATE objects SET state = 'deleting' WHERE id IN (`+sqliteutil.Placeholders(1, len(objs))+`)`, args...) return err } // AllStorageKeys returns every key registered in the object journal. func (s *SQLiteStore) AllStorageKeys(ctx context.Context) ([]string, error) { rows, err := s.db.QueryContext(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() }