Prefer CopyRequest for move API

modules/aws/src/main/scala/com/kevel/apso/aws/S3Bucket.scala

@@ -379,6 +379,42 @@ class S3Bucket(
       .join()
   }.isDefined
 
+  /** Copies the object in the location specified by `sourceKey` to `destinationKey`, in the same bucket.
+    *
+    * The copy is performed by S3 itself, so the object's bytes never travel through this process. The Transfer Manager
+    * uses `CopyObject` for small objects and multipart copy for large objects.
+    *
+    * @see
+    *   [[https://docs.aws.amazon.com/AmazonS3/latest/userguide/copy-object.html Copying objects in Amazon S3]]
+    * @see
+    *   [[https://docs.aws.amazon.com/java/api/latest/software/amazon/awssdk/transfer/s3/S3TransferManager.html#copy(software.amazon.awssdk.transfer.s3.model.CopyRequest) S3TransferManager.copy]]
+    *
+    * @param sourceKey
+    *   the remote pathname to copy from
+    * @param destinationKey
+    *   the remote pathname to copy to
+    * @return
+    *   true if the copy was successful, false otherwise.
+    */
+  def copy(sourceKey: String, destinationKey: String): Boolean = retry {
+    Using(getTransferManager) {
+      _.copy(
+        model.CopyRequest
+          .builder()
+          .copyObjectRequest(
+            CopyObjectRequest
+              .builder()
+              .sourceBucket(bucketName)
+              .sourceKey(sanitizeKey(sourceKey))
+              .destinationBucket(bucketName)
+              .destinationKey(sanitizeKey(destinationKey))
+              .build()
+          )
+          .build()
+      ).completionFuture().join()
+    }.get
+  }.isDefined
+
   /** Backups a remote file with the given `key`. A backup consists in copying the supplied file to a backup folder
     * under the same bucket and folder the file is currently in.
     *

modules/gcp/src/main/scala/com/kevel/apso/gcp/GCSBucket.scala

@@ -10,7 +10,7 @@ import scala.jdk.CollectionConverters.*
 import scala.util.Try
 
 import com.google.cloud.BaseServiceException
-import com.google.cloud.storage.Storage.{BlobListOption, BlobSourceOption, BlobWriteOption}
+import com.google.cloud.storage.Storage.{BlobListOption, BlobSourceOption, BlobWriteOption, CopyRequest}
 import com.google.cloud.storage.{Blob, BlobId, BlobInfo, Storage, StorageException}
 import com.typesafe.config.ConfigFactory
 import com.typesafe.scalalogging.LazyLogging
@@ -208,6 +208,21 @@ final class GCSBucket(
     storage.create(info, Array.emptyByteArray)
   }.isDefined
 
+  /** Copies the object in the location specified by `sourceKey` to `destinationKey`, in the same bucket.
+    *
+    * The copy is performed by GCS itself (the `rewrite` API), so the object's bytes never travel through this process.
+    *
+    * @param sourceKey
+    *   the remote pathname to copy from
+    * @param destinationKey
+    *   the remote pathname to copy to
+    * @return
+    *   true if the copy was successful, false otherwise.
+    */
+  def copy(sourceKey: String, destinationKey: String): Boolean = retry {
+    storage.copy(CopyRequest.of(blobId(sourceKey), blobId(destinationKey))).getResult()
+  }.isDefined
+
   /** Pulls a remote file with the given `key`, to the local storage in the pathname provided by `destination`.
     *
     * @param key

modules/io/src/main/scala/com/kevel/apso/io/GCSFileDescriptor.scala

@@ -3,8 +3,6 @@ package com.kevel.apso.io
 import java.io.InputStream
 import java.net.URI
 
-import scala.util.Using
-
 import com.google.cloud.storage.{Blob, Storage, StorageOptions}
 
 import com.kevel.apso.gcp.GCSBucket
@@ -99,10 +97,10 @@ case class GCSFileDescriptor(
   def move(pathString: String): Option[GCSFileDescriptor] = {
     val destination = mvLocation(pathString)
     if (destination.elements == elements) Some(this)
-    else {
-      val copied = Using.resource(stream())(input => destination.upload(input, Some(size)))
-      if (copied && delete()) Some(destination.copy(summary = None)) else None
-    }
+    else if (bucket.copy(builtPath, destination.builtPath) && delete())
+      Some(destination.copy(summary = None))
+    else
+      None
   }
 
   override def toString: String = uri.toString

modules/io/src/main/scala/com/kevel/apso/io/S3FileDescriptor.scala

@@ -4,7 +4,6 @@ import java.io.InputStream
 import java.net.URI
 
 import scala.collection.concurrent.TrieMap
-import scala.util.Using
 
 import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider}
 import software.amazon.awssdk.services.s3.model.S3Object
@@ -118,10 +117,10 @@ case class S3FileDescriptor(
   def move(pathString: String): Option[S3FileDescriptor] = {
     val destination = mvLocation(pathString)
     if (destination.elements == elements) Some(this)
-    else {
-      val copied = Using.resource(stream())(input => destination.upload(input, Some(size)))
-      if (copied && delete()) Some(destination.copy(summary = None)) else None
-    }
+    else if (bucket.copy(builtPath, destination.builtPath) && delete())
+      Some(destination.copy(summary = None))
+    else
+      None
   }
 
   override def toString: String =