Repository: BlueBrain/nexus (scala)

Issue:
Set the content type when copying a file in S3
**Motivation**
When copying a file between 2 buckets, we should be able to provide an optional content type to be set on the new object

**Acceptance criteria**
1. The provided content type is set on the new object if provided
2. The original content type is kept otherwise

Pull request:
Fixes #4968 

Bonus: do not copy the file again if it exists at the given key

Patch:
diff --git a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/CopyOptions.scala b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/CopyOptions.scala
new file mode 100644
index 0000000000..e4a3be447a
--- /dev/null
+++ b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/CopyOptions.scala
@@ -0,0 +1,12 @@
+package ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3
+
+import akka.http.scaladsl.model.ContentType
+
+/**
+  * Options when copying an object in S3
+  * @param overwriteTarget
+  *   if we want to overwrite the target if it exists
+  * @param newContentType
+  *   if we want to set a new content type on the created object
+  */
+final case class CopyOptions(overwriteTarget: Boolean, newContentType: Option[ContentType])
diff --git a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClient.scala b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClient.scala
index b3a1788732..b0516a4471 100644
--- a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClient.scala
+++ b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClient.scala
@@ -2,7 +2,7 @@ package ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.cli
 
 import cats.effect.{IO, Resource}
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.StoragesConfig.S3StorageConfig
-import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.HeadObject
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.{CopyOptions, HeadObject}
 import fs2.Stream
 import io.laserdisc.pure.s3.tagless.{Interpreter, S3AsyncClientOp}
 import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, AwsCredentialsProvider, DefaultCredentialsProvider, StaticCredentialsProvider}
@@ -34,15 +34,17 @@ trait S3StorageClient {
       sourceBucket: String,
       sourceKey: String,
       destinationBucket: String,
-      destinationKey: String
-  ): IO[CopyObjectResponse]
+      destinationKey: String,
+      options: CopyOptions
+  ): IO[Unit]
 
   def copyObjectMultiPart(
       sourceBucket: String,
       sourceKey: String,
       destinationBucket: String,
-      destinationKey: String
-  ): IO[CompleteMultipartUploadResponse]
+      destinationKey: String,
+      options: CopyOptions
+  ): IO[Unit]
 
   def uploadFile(
       fileData: Stream[IO, ByteBuffer],
diff --git a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClientDisabled.scala b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClientDisabled.scala
index 32e3fceaf1..f659a7f969 100644
--- a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClientDisabled.scala
+++ b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClientDisabled.scala
@@ -1,7 +1,7 @@
 package ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.client
 
 import cats.effect.IO
-import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.HeadObject
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.{CopyOptions, HeadObject}
 import ch.epfl.bluebrain.nexus.delta.sdk.error.ServiceError.FeatureDisabled
 import fs2.Stream
 import software.amazon.awssdk.services.s3.model._
@@ -24,8 +24,9 @@ private[client] object S3StorageClientDisabled extends S3StorageClient {
       sourceBucket: String,
       sourceKey: String,
       destinationBucket: String,
-      destinationKey: String
-  ): IO[CopyObjectResponse] = raiseDisabledErr
+      destinationKey: String,
+      options: CopyOptions
+  ): IO[Unit] = raiseDisabledErr
 
   override def objectExists(bucket: String, key: String): IO[Boolean] = raiseDisabledErr
 
@@ -42,8 +43,9 @@ private[client] object S3StorageClientDisabled extends S3StorageClient {
       sourceBucket: String,
       sourceKey: String,
       destinationBucket: String,
-      destinationKey: String
-  ): IO[CompleteMultipartUploadResponse] = raiseDisabledErr
+      destinationKey: String,
+      options: CopyOptions
+  ): IO[Unit] = raiseDisabledErr
 
   override def readFileMultipart(bucket: String, fileKey: String): Stream[IO, Byte] = throw disabledErr
 }
diff --git a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClientImpl.scala b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClientImpl.scala
index 3ae1af72a3..71ff2e44e0 100644
--- a/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClientImpl.scala
+++ b/delta/plugins/storage/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClientImpl.scala
@@ -1,9 +1,10 @@
 package ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.client
 
+import akka.http.scaladsl.model.ContentType
 import cats.effect.IO
 import cats.implicits._
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.model.StorageRejection.StorageNotAccessible
-import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.HeadObject
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.{CopyOptions, HeadObject}
 import eu.timepit.refined.refineMV
 import eu.timepit.refined.types.string.NonEmptyString
 import fs2.aws.s3.S3
@@ -57,34 +58,56 @@ final private[client] class S3StorageClientImpl(client: S3AsyncClientOp[IO]) ext
       sourceBucket: String,
       sourceKey: String,
       destinationBucket: String,
-      destinationKey: String
-  ): IO[CopyObjectResponse] =
-    client.copyObject(
-      CopyObjectRequest
-        .builder()
-        .sourceBucket(sourceBucket)
-        .sourceKey(sourceKey)
-        .destinationBucket(destinationBucket)
-        .destinationKey(destinationKey)
-        .checksumAlgorithm(checksumAlgorithm)
-        .build()
-    )
+      destinationKey: String,
+      options: CopyOptions
+  ): IO[Unit] =
+    approveCopy(destinationBucket, destinationKey, options.overwriteTarget).flatMap { approved =>
+      IO.whenA(approved) {
+        val requestBuilder     = CopyObjectRequest
+          .builder()
+          .sourceBucket(sourceBucket)
+          .sourceKey(sourceKey)
+          .destinationBucket(destinationBucket)
+          .destinationKey(destinationKey)
+          .checksumAlgorithm(checksumAlgorithm)
+        val requestWithOptions = options.newContentType.fold(requestBuilder) { contentType =>
+          requestBuilder
+            .contentType(contentType.value)
+            .metadataDirective(MetadataDirective.REPLACE)
+        }
+        client.copyObject(requestWithOptions.build()).void
+      }
+    }
 
   def copyObjectMultiPart(
       sourceBucket: String,
       sourceKey: String,
       destinationBucket: String,
-      destinationKey: String
-  ): IO[CompleteMultipartUploadResponse] = {
+      destinationKey: String,
+      options: CopyOptions
+  ): IO[Unit] =
+    approveCopy(destinationBucket, destinationKey, options.overwriteTarget).flatMap { approved =>
+      IO.whenA(approved) {
+        copyObjectMultiPart(sourceBucket, sourceKey, destinationBucket, destinationKey, options.newContentType)
+      }
+    }
+
+  private def copyObjectMultiPart(
+      sourceBucket: String,
+      sourceKey: String,
+      destinationBucket: String,
+      destinationKey: String,
+      newContentType: Option[ContentType]
+  ): IO[Unit] = {
     val partSize = 5_000_000_000L // 5GB
     for {
       // Initiate the multipart upload
-      createMultipartUploadResponse   <-
-        client.createMultipartUpload(createMultipartUploadRequest(destinationBucket, destinationKey, checksumAlgorithm))
+      createMultipartUploadResponse <-
+        client.createMultipartUpload(createMultipartUploadRequest(destinationBucket, destinationKey, newContentType))
       // Get the object size
-      objectSize                      <- headObject(sourceBucket, sourceKey).map(_.fileSize)
+      objectSize                    <- headObject(sourceBucket, sourceKey).map(_.fileSize)
       // Copy the object using 5 MB parts
-      completedParts                  <- {
+      completedParts                <- {
         val uploadParts = (0L until objectSize by partSize).zipWithIndex.map { case (start, partNumber) =>
           val lastByte              = Math.min(start + partSize - 1, objectSize - 1)
           val uploadPartCopyRequest = UploadPartCopyRequest.builder
@@ -109,18 +132,23 @@ final private[client] class S3StorageClientImpl(client: S3AsyncClientOp[IO]) ext
         uploadParts.toList.sequence
       }
       // Complete the upload request
-      completeMultipartUploadResponse <-
-        client.completeMultipartUpload(
-          CompleteMultipartUploadRequest.builder
-            .bucket(destinationBucket)
-            .key(destinationKey)
-            .uploadId(createMultipartUploadResponse.uploadId)
-            .multipartUpload(CompletedMultipartUpload.builder.parts(completedParts.asJava).build)
-            .build
-        )
-    } yield completeMultipartUploadResponse
+      _                             <- client.completeMultipartUpload(
+                                         CompleteMultipartUploadRequest.builder
+                                           .bucket(destinationBucket)
+                                           .key(destinationKey)
+                                           .uploadId(createMultipartUploadResponse.uploadId)
+                                           .multipartUpload(CompletedMultipartUpload.builder.parts(completedParts.asJava).build)
+                                           .build
+                                       )
+    } yield ()
   }
 
+  private def approveCopy(destinationBucket: String, destinationKey: String, overwriteEnabled: Boolean) =
+    if (overwriteEnabled)
+      IO.pure(true)
+    else
+      objectExists(destinationBucket, destinationKey).map { exists => !exists }
+
   override def objectExists(bucket: String, key: String): IO[Boolean] = {
     headObject(bucket, key)
       .redeemWith(
@@ -177,11 +205,13 @@ final private[client] class S3StorageClientImpl(client: S3AsyncClientOp[IO]) ext
       .build()
   }
 
-  private def createMultipartUploadRequest(bucket: String, fileKey: String, checksumAlgorithm: ChecksumAlgorithm) =
-    CreateMultipartUploadRequest.builder
-      .bucket(bucket)
-      .key(fileKey)
-      .checksumAlgorithm(checksumAlgorithm)
-      .build
+  private def createMultipartUploadRequest(bucket: String, fileKey: String, newContentType: Option[ContentType]) = {
+    val requestBuilder     =
+      CreateMultipartUploadRequest.builder.bucket(bucket).key(fileKey).checksumAlgorithm(checksumAlgorithm)
+    val requestWithOptions = newContentType.fold(requestBuilder) { contentType =>
+      requestBuilder.contentType(contentType.value)
+    }
+    requestWithOptions.build
+  }
 
 }
diff --git a/ship/src/main/scala/ch/epfl/bluebrain/nexus/ship/files/FileCopier.scala b/ship/src/main/scala/ch/epfl/bluebrain/nexus/ship/files/FileCopier.scala
index 61ca77b045..80b30aa846 100644
--- a/ship/src/main/scala/ch/epfl/bluebrain/nexus/ship/files/FileCopier.scala
+++ b/ship/src/main/scala/ch/epfl/bluebrain/nexus/ship/files/FileCopier.scala
@@ -5,16 +5,18 @@ import cats.effect.IO
 import ch.epfl.bluebrain.nexus.delta.kernel.Logger
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.UrlUtils
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.files.model.FileAttributes
-import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.S3LocationGenerator
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.{CopyOptions, S3LocationGenerator}
 import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.client.S3StorageClient
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.ProjectRef
 import ch.epfl.bluebrain.nexus.ship.config.FileProcessingConfig
+import ch.epfl.bluebrain.nexus.ship.files.FileCopier.CopyResult
+import ch.epfl.bluebrain.nexus.ship.files.FileCopier.CopyResult.{CopySkipped, CopySuccess}
 
 import scala.concurrent.duration.DurationInt
 
 trait FileCopier {
 
-  def copyFile(project: ProjectRef, attributes: FileAttributes): IO[Option[Uri.Path]]
+  def copyFile(project: ProjectRef, attributes: FileAttributes): IO[CopyResult]
 
 }
 
@@ -24,6 +26,16 @@ object FileCopier {
 
   private val longCopyThreshold = 5.seconds
 
+  sealed trait CopyResult extends Product with Serializable
+
+  object CopyResult {
+
+    final case class CopySuccess(newPath: Uri.Path) extends CopyResult
+
+    final case object CopySkipped extends CopyResult
+
+  }
+
   def apply(
       s3StorageClient: S3StorageClient,
       config: FileProcessingConfig
@@ -46,11 +58,13 @@ object FileCopier {
       val originKey = UrlUtils.decode(origin)
       val targetKey = UrlUtils.decode(target)
 
+      val copyOptions = CopyOptions(overwriteTarget = false, attributes.mediaType)
+
       def copy = {
         if (attributes.bytes >= FIVE_GB)
-          s3StorageClient.copyObjectMultiPart(importBucket, originKey, targetBucket, targetKey).void
+          s3StorageClient.copyObjectMultiPart(importBucket, originKey, targetBucket, targetKey, copyOptions)
         else
-          s3StorageClient.copyObject(importBucket, originKey, targetBucket, targetKey).void
+          s3StorageClient.copyObject(importBucket, originKey, targetBucket, targetKey, copyOptions)
       }.timed.flatMap { case (duration, _) =>
         IO.whenA(duration > longCopyThreshold)(
           logger.info(s"Copy file ${attributes.path} of size ${attributes.bytes} took ${duration.toSeconds} seconds.")
@@ -66,10 +80,10 @@ object FileCopier {
         _        <- IO.whenA(!isFolder && !isObject) {
                       logger.error(s"$target is neither an object or folder, something is wrong.")
                     }
-      } yield Option.when(isObject)(target)
+      } yield if (isObject) CopySuccess(target) else CopySkipped
     }
   }
 
-  def apply(): FileCopier = (_: ProjectRef, attributes: FileAttributes) => IO.some(attributes.path)
+  def apply(): FileCopier = (_: ProjectRef, attributes: FileAttributes) => IO.pure(CopySuccess(attributes.path))
 
 }
diff --git a/ship/src/main/scala/ch/epfl/bluebrain/nexus/ship/files/FileProcessor.scala b/ship/src/main/scala/ch/epfl/bluebrain/nexus/ship/files/FileProcessor.scala
index 4598842aa8..2554a50c85 100644
--- a/ship/src/main/scala/ch/epfl/bluebrain/nexus/ship/files/FileProcessor.scala
+++ b/ship/src/main/scala/ch/epfl/bluebrain/nexus/ship/files/FileProcessor.scala
@@ -21,6 +21,7 @@ import ch.epfl.bluebrain.nexus.delta.sourcing.{ScopedEventLog, Transactors}
 import ch.epfl.bluebrain.nexus.ship._
 import ch.epfl.bluebrain.nexus.ship.acls.AclWiring.alwaysAuthorize
 import ch.epfl.bluebrain.nexus.ship.config.InputConfig
+import ch.epfl.bluebrain.nexus.ship.files.FileCopier.CopyResult.{CopySkipped, CopySuccess}
 import ch.epfl.bluebrain.nexus.ship.files.FileProcessor.logger
 import ch.epfl.bluebrain.nexus.ship.files.FileWiring._
 import ch.epfl.bluebrain.nexus.ship.storages.StorageWiring
@@ -66,22 +67,22 @@ class FileProcessor private (
         val attrs          = e.attributes
         val customMetadata = Some(getCustomMetadata(attrs))
         fileCopier.copyFile(e.project, attrs).flatMap {
-          case Some(newPath) =>
+          case CopySuccess(newPath) =>
             files
               .registerFile(fileId, None, customMetadata, newPath, e.tag, attrs.mediaType)
               .as(ImportStatus.Success)
-          case None          => IO.pure(ImportStatus.Dropped)
+          case CopySkipped          => IO.pure(ImportStatus.Dropped)
         }
       case _: FileCreated                                        => IO.pure(ImportStatus.Dropped)
       case e: FileUpdated if e.attributes.bytes < 5_000_000_000L =>
         val attrs          = e.attributes
         val customMetadata = Some(getCustomMetadata(attrs))
         fileCopier.copyFile(e.project, attrs).flatMap {
-          case Some(newPath) =>
+          case CopySuccess(newPath) =>
             files
               .updateRegisteredFile(fileId, None, customMetadata, cRev, newPath, e.tag, attrs.mediaType)
               .as(ImportStatus.Success)
-          case None          => IO.pure(ImportStatus.Dropped)
+          case CopySkipped          => IO.pure(ImportStatus.Dropped)
         }
       case _: FileUpdated                                        => IO.pure(ImportStatus.Dropped)
       case e: FileCustomMetadataUpdated                          =>


Test patch:
diff --git a/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/S3Helpers.scala b/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/S3Helpers.scala
index 5e717fb735..a7f7a64779 100644
--- a/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/S3Helpers.scala
+++ b/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/S3Helpers.scala
@@ -44,4 +44,18 @@ trait S3Helpers { self: Generators =>
     client.uploadFile(Stream.emit(ByteBuffer.wrap(bytes)), bucket, key, bytes.length.toLong) >> test(key)
   }
 
+  def givenFilesInABucket(bucket: String, contents1: String, contents2: String)(
+      test: (String, String) => IO[Unit]
+  )(implicit client: S3StorageClient): IO[Unit] = {
+    val bytes1 = contents1.getBytes(StandardCharsets.UTF_8)
+    val key1   = genString()
+    val bytes2 = contents2.getBytes(StandardCharsets.UTF_8)
+    val key2   = genString()
+    for {
+      _ <- client.uploadFile(Stream.emit(ByteBuffer.wrap(bytes1)), bucket, key1, bytes1.length.toLong)
+      _ <- client.uploadFile(Stream.emit(ByteBuffer.wrap(bytes2)), bucket, key2, bytes2.length.toLong)
+      _ <- test(key1, key2)
+    } yield ()
+  }
+
 }
diff --git a/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClientSuite.scala b/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClientSuite.scala
new file mode 100644
index 0000000000..7ace13a50f
--- /dev/null
+++ b/delta/plugins/storage/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/storage/storages/operations/s3/client/S3StorageClientSuite.scala
@@ -0,0 +1,91 @@
+package ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.client
+
+import akka.http.scaladsl.model.ContentTypes
+import cats.effect.IO
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.StoragesConfig.S3StorageConfig
+import ch.epfl.bluebrain.nexus.delta.plugins.storage.storages.operations.s3.{CopyOptions, LocalStackS3StorageClient, S3Helpers}
+import ch.epfl.bluebrain.nexus.testkit.mu.NexusSuite
+import io.laserdisc.pure.s3.tagless.S3AsyncClientOp
+import munit.AnyFixture
+
+class S3StorageClientSuite extends NexusSuite with LocalStackS3StorageClient.Fixture with S3Helpers {
+
+  implicit private lazy val (s3StorageClient: S3StorageClient, underlying: S3AsyncClientOp[IO], _: S3StorageConfig) =
+    localStackS3Client()
+
+  private val fileContents  = "file content"
+  private val contentLength = fileContents.length.toLong
+
+  private val anotherContent       = "Another content"
+  private val anotherContentLength = anotherContent.length.toLong
+
+  private val defaultS3ContentType = ContentTypes.`application/octet-stream`
+  private val contentType          = ContentTypes.`application/json`
+
+  override def munitFixtures: Seq[AnyFixture[_]] = List(localStackS3Client)
+
+  test("Copy the file to its new location if none is already there without a content type") {
+    givenAnS3Bucket { bucket =>
+      val options = CopyOptions(overwriteTarget = false, None)
+      givenAFileInABucket(bucket, fileContents) { key =>
+        val newKey = genString()
+        for {
+          _    <- s3StorageClient.copyObject(bucket, key, bucket, newKey, options)
+          head <- s3StorageClient.headObject(bucket, newKey)
+        } yield {
+          assertEquals(head.fileSize, contentLength)
+          assertEquals(head.contentType, Some(defaultS3ContentType))
+        }
+      }
+    }
+  }
+
+  test("Copy the file to its new location if none is already there setting a content type") {
+    givenAnS3Bucket { bucket =>
+      val options = CopyOptions(overwriteTarget = false, Some(contentType))
+      givenAFileInABucket(bucket, fileContents) { key =>
+        val newKey = genString()
+        for {
+          _    <- s3StorageClient.copyObject(bucket, key, bucket, newKey, options)
+          head <- s3StorageClient.headObject(bucket, newKey)
+        } yield {
+          assertEquals(head.fileSize, contentLength)
+          assertEquals(head.contentType, Some(contentType))
+        }
+      }
+    }
+  }
+
+  test("Do not overwrite an existing object") {
+    givenAnS3Bucket { bucket =>
+      val options = CopyOptions(overwriteTarget = false, Some(contentType))
+      givenFilesInABucket(bucket, fileContents, anotherContent) { case (sourceKey, existingTargetKey) =>
+        for {
+          _    <- s3StorageClient.copyObject(bucket, sourceKey, bucket, existingTargetKey, options)
+          head <- s3StorageClient.headObject(bucket, existingTargetKey)
+        } yield {
+          val clue = "The file should not have been overwritten"
+          assertEquals(head.fileSize, anotherContentLength, clue)
+          assertEquals(head.contentType, Some(defaultS3ContentType), clue)
+        }
+      }
+    }
+  }
+
+  test("Overwrite an existing object") {
+    givenAnS3Bucket { bucket =>
+      val options = CopyOptions(overwriteTarget = true, Some(contentType))
+      givenFilesInABucket(bucket, fileContents, anotherContent) { case (sourceKey, existingTargetKey) =>
+        for {
+          _    <- s3StorageClient.copyObject(bucket, sourceKey, bucket, existingTargetKey, options)
+          head <- s3StorageClient.headObject(bucket, existingTargetKey)
+        } yield {
+          val clue = "The file should have been overwritten"
+          assertEquals(head.fileSize, contentLength, clue)
+          assertEquals(head.contentType, Some(contentType), clue)
+        }
+      }
+    }
+  }
+
+}
diff --git a/ship/src/test/resources/import/file-import.json b/ship/src/test/resources/import/file-import.json
index 4018a65aae..5b05145488 100644
--- a/ship/src/test/resources/import/file-import.json
+++ b/ship/src/test/resources/import/file-import.json
@@ -1,5 +1,5 @@
 {"ordering":1,"type":"project","org":"public","project":"sscx","id":"projects/public/sscx"                                     ,"rev":1,"value":{"rev": 1, "base": "https://bbp.epfl.ch/neurosciencegraph/data/", "uuid": "c7d70522-4305-480a-b190-75d757ed9a49", "@type": "ProjectCreated", "label": "sscx", "vocab": "https://bbp.epfl.ch/nexus/v1/resources/public/sscx/_/", "instant": "2020-01-22T16:03:33.105Z", "subject": {"@type": "User", "realm": "bbp", "subject": "bob"}, "apiMappings": {"prov": "http://www.w3.org/ns/prov#", "context": "https://incf.github.io/neuroshapes/contexts/", "schemaorg": "http://schema.org/", "datashapes": "https://neuroshapes.org/dash/", "ontologies": "https://neuroshapes.org/dash/ontology", "taxonomies": "https://neuroshapes.org/dash/taxonomy", "commonshapes": "https://neuroshapes.org/commons/", "provdatashapes": "https://provshapes.org/datashapes/", "provcommonshapes": "https://provshapes.org/commons/"}, "description": "This project contains somatosensorycortex dissemination publication data.", "organizationUuid": "d098a020-508b-4131-a28d-75f73b7f5f0e", "organizationLabel": "public"},"instant":"2020-01-22T17:03:33.105+01:00"}
-{"ordering":2,"type":"file"   ,"org":"public","project":"sscx","id":"https://bbp.epfl.ch/neurosciencegraph/data/old-path","rev":1,"value":{"id": "https://bbp.epfl.ch/neurosciencegraph/data/old-path", "rev": 1, "@type": "FileCreated", "instant": "2020-04-15T14:34:05.190Z", "project": "public/sscx", "storage": "https://bbp.epfl.ch/neurosciencegraph/data/storage?rev=1", "subject": {"@type": "User", "realm": "bbp", "subject": "bob"}, "attributes": {"path": "75b85666-b66f-4d90-8fd2-c6fb04beb5c6/0/a/7/9/a/d/1/d/002_160120B3_OH.nwb", "uuid": "0a79ad1d-4e74-4234-8efc-7fc8d41bafe6", "bytes": 9917692, "digest": {"@type": "ComputedDigest", "value": "4b9fcb8143f50d8a3949eb7f4be1dfb5790e13f3aa681211a637ef345dfafa00", "algorithm": "SHA-256"}, "origin": "Client", "filename": "002_160120B3_OH.nwb", "location": "file:///path/nexus/75b85666-b66f-4d90-8fd2-c6fb04beb5c6/0/a/7/9/a/d/1/d/002_160120B3_OH.nwb", "mediaType": "application/octet-stream"}, "storageType": "RemoteDiskStorage"},"instant":"2020-04-15T16:34:05.19+02:00"}
+{"ordering":2,"type":"file"   ,"org":"public","project":"sscx","id":"https://bbp.epfl.ch/neurosciencegraph/data/old-path","rev":1,"value":{"id": "https://bbp.epfl.ch/neurosciencegraph/data/old-path", "rev": 1, "@type": "FileCreated", "instant": "2020-04-15T14:34:05.190Z", "project": "public/sscx", "storage": "https://bbp.epfl.ch/neurosciencegraph/data/storage?rev=1", "subject": {"@type": "User", "realm": "bbp", "subject": "bob"}, "attributes": {"path": "75b85666-b66f-4d90-8fd2-c6fb04beb5c6/0/a/7/9/a/d/1/d/002_160120B3_OH.nwb", "uuid": "0a79ad1d-4e74-4234-8efc-7fc8d41bafe6", "bytes": 9917692, "digest": {"@type": "ComputedDigest", "value": "4b9fcb8143f50d8a3949eb7f4be1dfb5790e13f3aa681211a637ef345dfafa00", "algorithm": "SHA-256"}, "origin": "Client", "filename": "002_160120B3_OH.nwb", "location": "file:///path/nexus/75b85666-b66f-4d90-8fd2-c6fb04beb5c6/0/a/7/9/a/d/1/d/002_160120B3_OH.nwb", "mediaType": "application/nwb"}, "storageType": "RemoteDiskStorage"},"instant":"2020-04-15T16:34:05.19+02:00"}
 {"ordering":3,"type":"file"   ,"org":"public","project":"sscx","id":"https://bbp.epfl.ch/neurosciencegraph/data/old-path","rev":2,"value":{"id": "https://bbp.epfl.ch/neurosciencegraph/data/old-path", "rev": 2, "@type": "FileAttributesUpdated", "bytes": 10488964, "digest": {"@type": "ComputedDigest", "value": "e237a2e1a4d56ffd63f9d2222788bc4fd9fa5e9c05466570b58675fde255dcc0", "algorithm": "SHA-256"}, "instant": "2020-04-27T07:49:39.161Z", "project": "public/sscx", "storage": "https://bbp.epfl.ch/neurosciencegraph/data/0617f3c3-8d30-4054-807f-c3708922f973?rev=1", "subject": {"@type": "User", "realm": "bbp", "subject": "alice"}, "mediaType": "application/x-hdf", "storageType": "RemoteDiskStorage"},"instant":"2020-04-27T09:49:39.161+02:00"}
 {"ordering":4,"type":"file"   ,"org":"public","project":"sscx","id":"https://bbp.epfl.ch/neurosciencegraph/data/old-path","rev":3,"value":{"id": "https://bbp.epfl.ch/neurosciencegraph/data/old-path", "rev": 3, "@type": "FileUpdated", "instant": "2020-09-24T09:58:43.479Z", "project": "public/sscx", "storage": "https://bbp.epfl.ch/neurosciencegraph/data/storage?rev=1", "subject": {"@type": "User", "realm": "bbp", "subject": "bob"}, "attributes": {"path": "75b85666-b66f-4d90-8fd2-c6fb04beb5c6/8/9/5/4/c/3/e/c/002_160120B3_OH_updated.nwb", "uuid": "8954c3ec-da81-47b9-bcec-b72a1706a6a3", "bytes": 10701815, "digest": {"@type": "NotComputedDigest"}, "origin": "Storage", "filename": "002_160120B3_OH_updated.nwb", "location": "file:///path/nexus/75b85666-b66f-4d90-8fd2-c6fb04beb5c6/8/9/5/4/c/3/e/c/002_160120B3_updated.nwb", "mediaType": "application/nwb"}, "storageType": "RemoteDiskStorage"},"instant":"2020-09-24T11:58:43.479+02:00"}
 {"ordering":5,"type":"file"   ,"org":"public","project":"sscx","id":"https://bbp.epfl.ch/neurosciencegraph/data/empty-filename","rev":1,"value":{"id": "https://bbp.epfl.ch/neurosciencegraph/data/empty-filename", "rev": 1, "@type": "FileCreated", "instant": "2023-03-02T10:46:36.969Z", "project": "public/sscx", "storage": "https://bbp.epfl.ch/neurosciencegraph/data/storage?rev=1", "subject": {"@type": "User", "realm": "bbp", "subject": "alice"}, "attributes": {"path": "public/sscx/2/b/3/9/7/9/3/0/", "uuid": "2b397930-0f69-4dad-bf6a-51825e940e12", "bytes": 538, "digest": {"@type": "ComputedDigest", "value": "b39a754a0988ca1f62e04a34d70479e9610b87beab91c58766f80c6ef6f93f3d", "algorithm": "SHA-256"}, "origin": "Client", "filename": "", "location": "file:///path/public/sscx/2/b/3/9/7/9/3/0", "mediaType": "text/plain"}, "storageType": "RemoteDiskStorage"},"instant":"2023-03-02T11:46:36.969+01:00"}
diff --git a/ship/src/test/scala/ch/epfl/bluebrain/nexus/ship/RunShipSuite.scala b/ship/src/test/scala/ch/epfl/bluebrain/nexus/ship/RunShipSuite.scala
index 8e34334464..7887a108b9 100644
--- a/ship/src/test/scala/ch/epfl/bluebrain/nexus/ship/RunShipSuite.scala
+++ b/ship/src/test/scala/ch/epfl/bluebrain/nexus/ship/RunShipSuite.scala
@@ -1,5 +1,6 @@
 package ch.epfl.bluebrain.nexus.ship
 
+import akka.http.scaladsl.model.{ContentType, MediaTypes}
 import cats.effect.IO
 import ch.epfl.bluebrain.nexus.delta.kernel.Hex
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.UrlUtils
@@ -133,6 +134,7 @@ class RunShipSuite
     report.progress == Map(Projects.entityType -> Statistics(1L, 0L))
 
   test("Import files in S3 and in the primary store") {
+    val textPlain = MediaTypes.`text/plain`.withMissingCharset
     for {
       events               <- eventsStream("import/file-import.json")
       report               <- RunShip(events, s3Client, inputConfig, xas)
@@ -141,21 +143,22 @@ class RunShipSuite
       oldPathFileId         = iri"https://bbp.epfl.ch/neurosciencegraph/data/old-path"
       oldPathLocationRev1   = "/prefix/public/sscx/files/0/a/7/9/a/d/1/d/002_160120B3_OH.nwb"
       oldPathLocationRev2   = "/prefix/public/sscx/files/8/9/5/4/c/3/e/c/002_160120B3_OH_updated.nwb"
+      oldPathContentType    = ContentType.parse("application/nwb").toOption
       _                    <- checkFor("file", oldPathFileId, xas).assertEquals(3)
-      _                    <- assertS3Object(oldPathLocationRev1)
-      _                    <- assertS3Object(oldPathLocationRev2)
+      _                    <- assertS3Object(oldPathLocationRev1, oldPathContentType)
+      _                    <- assertS3Object(oldPathLocationRev2, oldPathContentType)
       _                    <- assertFileAttributes(project, oldPathFileId)(oldPathLocationRev2, "002_160120B3_OH_updated.nwb")
       // File with a blank filename
       blankFilenameId       = iri"https://bbp.epfl.ch/neurosciencegraph/data/empty-filename"
       blankFilenameLocation = "/prefix/public/sscx/files/2/b/3/9/7/9/3/0/file"
       _                    <- checkFor("file", blankFilenameId, xas).assertEquals(1)
-      _                    <- assertS3Object(blankFilenameLocation)
+      _                    <- assertS3Object(blankFilenameLocation, Some(textPlain))
       _                    <- assertFileAttributes(project, blankFilenameId)(blankFilenameLocation, "file")
       // File with special characters in the filename
       specialCharsId        = iri"https://bbp.epfl.ch/neurosciencegraph/data/special-chars-filename"
       specialCharsLocation  = "/prefix/public/sscx/files/1/2/3/4/5/6/7/8/special [file].json"
       _                    <- checkFor("file", specialCharsId, xas).assertEquals(1)
-      _                    <- assertS3Object(specialCharsLocation)
+      _                    <- assertS3Object(specialCharsLocation, Some(textPlain))
       _                    <- assertFileAttributes(project, specialCharsId)(specialCharsLocation, "special [file].json")
       // Directory, should be skipped
       directoryId           = iri"https://bbp.epfl.ch/neurosciencegraph/data/directory"
@@ -175,12 +178,13 @@ class RunShipSuite
       assertEquals(attributes.filename, expectedFileName)
     }
 
-  private def assertS3Object(key: String)(implicit location: Location): IO[Unit] =
+  private def assertS3Object(key: String, contentType: Option[ContentType])(implicit location: Location): IO[Unit] =
     s3Client
       .headObject(targetBucket, key)
       .map { head =>
         assertEquals(head.fileSize, contentLength)
         assertEquals(head.digest, fileDigest)
+        assertEquals(head.contentType, contentType)
       }
       .recoverWith { e =>
         s3Client.listObjectsV2(targetBucket).map { listResponse =>