Generalise S3 buckets

common-lib/src/main/scala/com/gu/mediaservice/lib/BaseStore.scala

@@ -1,49 +1,25 @@
 package com.gu.mediaservice.lib
 
-import org.apache.pekko.actor.{Cancellable, Scheduler}
-import com.gu.mediaservice.lib.aws.S3
+import com.gu.mediaservice.lib.aws.{S3, S3Bucket}
 import com.gu.mediaservice.lib.config.CommonConfig
 import com.gu.mediaservice.lib.logging.GridLogging
+import org.apache.pekko.actor.{Cancellable, Scheduler}
 import org.joda.time.DateTime
-import software.amazon.awssdk.services.s3.model.{GetObjectRequest, ListObjectsV2Request}
 
 import java.util.concurrent.atomic.AtomicReference
-import java.io.InputStream
-import scala.jdk.CollectionConverters._
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 import scala.util.control.NonFatal
 
 
-abstract class BaseStore[TStoreKey, TStoreVal](bucket: String, config: CommonConfig)(implicit ec: ExecutionContext)
+abstract class BaseStore[TStoreKey, TStoreVal](bucket: S3Bucket, config: CommonConfig, s3: S3)(implicit ec: ExecutionContext)
   extends GridLogging {
 
-  val s3 = new S3(config)
-
   protected val store: AtomicReference[Map[TStoreKey, TStoreVal]] = new AtomicReference(Map.empty)
   protected val lastUpdated: AtomicReference[DateTime] = new AtomicReference(DateTime.now())
 
   protected def getS3Object(key: String): Option[String] = s3.getObjectAsString(bucket, key)
 
-  protected def getLatestS3Stream: Option[InputStream] = {
-    val objects = s3.client.listObjectsV2(ListObjectsV2Request.builder().bucket(bucket).build())
-      .contents().asScala.toList
-      .filterNot(_.key() == "AMAZON_SES_SETUP_NOTIFICATION")
-
-    if (objects.nonEmpty) {
-      val obj = objects.maxBy(_.lastModified())
-      logger.info(s"Latest key ${obj.key} in bucket $bucket")
-
-      val stream = s3.client.getObject(
-        GetObjectRequest.builder().key(obj.key()).bucket(bucket).build()
-      )
-      Some(stream)
-    } else {
-      logger.error(s"Bucket $bucket is empty")
-      None
-    }
-  }
-
   private var cancellable: Option[Cancellable] = None
 
   def scheduleUpdates(scheduler: Scheduler): Unit = {

common-lib/src/main/scala/com/gu/mediaservice/lib/ImageIngestOperations.scala

@@ -1,17 +1,15 @@
 package com.gu.mediaservice.lib
 
 
-import com.gu.mediaservice.lib.aws.S3Object
+import com.gu.mediaservice.lib.aws.{S3Bucket, S3Object}
 import com.gu.mediaservice.lib.config.CommonConfig
 import com.gu.mediaservice.lib.logging.LogMarker
 import com.gu.mediaservice.model.{Instance, MimeType}
 import com.typesafe.scalalogging.StrictLogging
 import org.joda.time.DateTime
-import software.amazon.awssdk.services.s3.model.{Delete, DeleteObjectsRequest, ObjectIdentifier}
 
 import java.io.File
 import scala.concurrent.Future
-import scala.jdk.CollectionConverters._
 
 object ImageIngestOperations {
   def fileKeyFromId(id: String)(implicit instance: Instance): String = instance.id + "/" + snippetForId(id)
@@ -21,7 +19,7 @@ object ImageIngestOperations {
   private def snippetForId(id: String) = id.take(6).mkString("/") + "/" + id
 }
 
-class ImageIngestOperations(imageBucket: String, thumbnailBucket: String, config: CommonConfig, isVersionedS3: Boolean = false)
+class ImageIngestOperations(imageBucket: S3Bucket, thumbnailBucket: S3Bucket, config: CommonConfig, isVersionedS3: Boolean = false)
   extends S3ImageStorage(config) with StrictLogging {
 
   import ImageIngestOperations.{fileKeyFromId, optimisedPngKeyFromId}
@@ -36,45 +34,31 @@ class ImageIngestOperations(imageBucket: String, thumbnailBucket: String, config
   private def storeOriginalImage(storableImage: StorableOriginalImage)
                         (implicit logMarker: LogMarker): Future[S3Object] = {
     val instanceSpecificKey = instanceAwareOriginalImageKey(storableImage)
-    logger.info(s"Storing original image to instance specific key:$imageBucket / $instanceSpecificKey")
+    logger.info(s"Storing original image to instance specific key:${imageBucket.name} / $instanceSpecificKey")
     storeImage(imageBucket, instanceSpecificKey, storableImage.file, Some(storableImage.mimeType),
       storableImage.meta, overwrite = false)
   }
 
   private def storeThumbnailImage(storableImage: StorableThumbImage)
                                  (implicit logMarker: LogMarker): Future[S3Object] = {
     val instanceSpecificKey = instanceAwareThumbnailImageKey(storableImage)
-    logger.info(s"Storing thumbnail to instance specific key: $thumbnailBucket / $instanceSpecificKey")
+    logger.info(s"Storing thumbnail to instance specific key: ${thumbnailBucket.name} / $instanceSpecificKey")
     storeImage(thumbnailBucket, instanceSpecificKey, storableImage.file, Some(storableImage.mimeType),
       overwrite = true)
   }
 
   private def storeOptimisedImage(storableImage: StorableOptimisedImage)
                                  (implicit logMarker: LogMarker): Future[S3Object] = {
     val instanceSpecificKey = optimisedPngKeyFromId(storableImage.id)(storableImage.instance)
-    logger.info(s"Storing optimised image to instance specific key: $thumbnailBucket / $instanceSpecificKey")
+    logger.info(s"Storing optimised image to instance specific key: ${thumbnailBucket.name} / $instanceSpecificKey")
     storeImage(imageBucket, instanceSpecificKey, storableImage.file, Some(storableImage.mimeType),
       overwrite = true)
   }
 
-
-  private def bulkDelete(bucket: String, keys: List[String]): Future[Map[String, Boolean]] = keys match {
+  private def bulkDelete(bucket: S3Bucket, keys: List[String]): Future[Map[String, Boolean]] = keys match {
     case Nil => Future.successful(Map.empty)
     case _ => Future {
-      val objects = keys.map { key =>
-        ObjectIdentifier.builder()
-          .key(key)
-          .build()
-      }.asJava
-      val response = client.deleteObjects(
-        DeleteObjectsRequest.builder().bucket(bucket)
-          .delete(Delete.builder().objects(objects).build())
-          .build()
-      )
-      val errorKeys = response.errors().asScala.toList.map(_.key())
-      keys.map { key =>
-        key -> !errorKeys.contains(key)
-      }.toMap
+      deleteObjects(bucket, keys)
     }
   }
 
@@ -106,7 +90,7 @@ sealed trait ImageWrapper {
   val instance: Instance
 }
 sealed trait StorableImage extends ImageWrapper {
-  def toProjectedS3Object(thumbBucket: String): S3Object = S3Object(
+  def toProjectedS3Object(thumbBucket: S3Bucket): S3Object = S3Object(
     thumbBucket,
     ImageIngestOperations.fileKeyFromId(id)(instance),
     file,
@@ -118,7 +102,7 @@ sealed trait StorableImage extends ImageWrapper {
 
 case class StorableThumbImage(id: String, file: File, mimeType: MimeType, meta: Map[String, String] = Map.empty, instance: Instance) extends StorableImage
 case class StorableOriginalImage(id: String, file: File, mimeType: MimeType, lastModified: DateTime, meta: Map[String, String] = Map.empty, instance: Instance) extends StorableImage {
-  override def toProjectedS3Object(thumbBucket: String): S3Object = S3Object(
+  override def toProjectedS3Object(thumbBucket: S3Bucket): S3Object = S3Object(
     thumbBucket,
     ImageIngestOperations.fileKeyFromId(id)(instance),
     file,
@@ -128,7 +112,7 @@ case class StorableOriginalImage(id: String, file: File, mimeType: MimeType, las
   )
 }
 case class StorableOptimisedImage(id: String, file: File, mimeType: MimeType, meta: Map[String, String] = Map.empty, instance: Instance) extends StorableImage {
-  override def toProjectedS3Object(thumbBucket: String): S3Object = S3Object(
+  override def toProjectedS3Object(thumbBucket: S3Bucket): S3Object = S3Object(
     thumbBucket,
     ImageIngestOperations.optimisedPngKeyFromId(id)(instance),
     file,

common-lib/src/main/scala/com/gu/mediaservice/lib/ImageQuarantineOperations.scala

@@ -2,13 +2,13 @@ package com.gu.mediaservice.lib
 
 import java.io.File
 import com.gu.mediaservice.lib.config.CommonConfig
-import com.gu.mediaservice.lib.aws.S3Object
+import com.gu.mediaservice.lib.aws.{S3Bucket, S3Object}
 import com.gu.mediaservice.lib.logging.LogMarker
 import com.gu.mediaservice.model.{Instance, MimeType}
 
 import scala.concurrent.Future
 
-class ImageQuarantineOperations(quarantineBucket: String, config: CommonConfig, isVersionedS3: Boolean = false)
+class ImageQuarantineOperations(quarantineBucket: S3Bucket, config: CommonConfig, isVersionedS3: Boolean = false)
   extends S3ImageStorage(config) {
 
   def storeQuarantineImage(id: String, file: File, mimeType: Option[MimeType], meta: Map[String, String] = Map.empty)

common-lib/src/main/scala/com/gu/mediaservice/lib/ImageStorage.scala

@@ -6,7 +6,7 @@ import java.io.File
 import scala.concurrent.{ExecutionContext, Future}
 import scala.concurrent.duration._
 import scala.language.postfixOps
-import com.gu.mediaservice.lib.aws.S3Object
+import com.gu.mediaservice.lib.aws.{S3Bucket, S3Object}
 import com.gu.mediaservice.lib.logging.LogMarker
 import com.gu.mediaservice.model.MimeType
 
@@ -37,9 +37,9 @@ trait ImageStorage {
   /** Store a copy of the given file and return the URI of that copy.
     * The file can safely be deleted afterwards.
     */
-  def storeImage(bucket: String, id: String, file: File, mimeType: Option[MimeType],
+  def storeImage(bucket: S3Bucket, id: String, file: File, mimeType: Option[MimeType],
                  meta: Map[String, String] = Map.empty, overwrite: Boolean)
                 (implicit logMarker: LogMarker): Future[S3Object]
 
-  def deleteImage(bucket: String, id: String)(implicit logMarker: LogMarker): Future[Unit]
+  def deleteImage(bucket: S3Bucket, id: String)(implicit logMarker: LogMarker): Future[Unit]
 }

common-lib/src/main/scala/com/gu/mediaservice/lib/S3ImageStorage.scala

@@ -1,10 +1,10 @@
 package com.gu.mediaservice.lib
 
-import com.gu.mediaservice.lib.aws.S3
+import com.gu.mediaservice.lib.aws.{S3, S3Bucket}
 import com.gu.mediaservice.lib.config.CommonConfig
 import com.gu.mediaservice.lib.logging.{GridLogging, LogMarker}
 import com.gu.mediaservice.model.MimeType
-import software.amazon.awssdk.services.s3.model.{DeleteObjectRequest, HeadObjectRequest, ListObjectsV2Request}
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request
 
 import java.io.File
 import scala.concurrent.Future
@@ -14,10 +14,10 @@ import scala.jdk.CollectionConverters._
 class S3ImageStorage(config: CommonConfig) extends S3(config) with ImageStorage with GridLogging {
 
   private val cacheSetting = Some(cacheForever)
-  def storeImage(bucket: String, id: String, file: File, mimeType: Option[MimeType],
+  def storeImage(bucket: S3Bucket, id: String, file: File, mimeType: Option[MimeType],
                  meta: Map[String, String] = Map.empty, overwrite: Boolean)
                 (implicit logMarker: LogMarker) = {
-    logger.info(logMarker, s"bucket: $bucket, id: $id, meta: $meta")
+    logger.info(logMarker, s"bucket: ${bucket.name}, id: $id, meta: $meta")
     val eventualObject = if (overwrite) {
       store(bucket, id, file, mimeType, meta, cacheSetting)
     } else {
@@ -27,27 +27,23 @@ class S3ImageStorage(config: CommonConfig) extends S3(config) with ImageStorage
     eventualObject
   }
 
-  def deleteImage(bucket: String, key: String)(implicit logMarker: LogMarker) = Future {
-    logger.info(logMarker, s"Deleted image $key from bucket $bucket")
-    client.deleteObject(
-      DeleteObjectRequest.builder().bucket(bucket).key(key).build())
+  def deleteImage(bucket: S3Bucket, key: String)(implicit logMarker: LogMarker) = Future {
+    deleteObject(bucket, key)
+    logger.info(logMarker, s"Deleted image $key from bucket ${bucket.name}")
   }
 
-  def deleteVersionedImage(bucket: String, id: String)(implicit logMarker: LogMarker) = Future {
-    val objectVersion = client.headObject(HeadObjectRequest.builder().bucket(bucket).key(id).build()).versionId()
-    client.deleteObject(DeleteObjectRequest.builder().bucket(bucket).key(id).versionId(objectVersion).build())
-    logger.info(logMarker, s"Deleted image $id from bucket $bucket (version: $objectVersion)")
+  def deleteVersionedImage(bucket: S3Bucket, id: String)(implicit logMarker: LogMarker) = Future {
+    val objectVersion = getMetadata(bucket, id).objectVersion.getOrElse(
+      throw new IllegalStateException(s"No version id found for $id in bucket ${bucket.name}")
+    )
+    deleteVersion(bucket, id, objectVersion)
+    logger.info(logMarker, s"Deleted image $id from bucket ${bucket.name} (version: $objectVersion)")
   }
 
-  def deleteFolder(bucket: String, id: String)(implicit logMarker: LogMarker) = Future {
-    val files = client.listObjectsV2(
-      ListObjectsV2Request.builder().bucket(bucket).prefix(id).build()
-    ).contents().asScala.toList
+  def deleteFolder(bucket: S3Bucket, id: String)(implicit logMarker: LogMarker): Future[Unit] = list(bucket, id).map { files =>
     logger.info(s"Found ${files.size} files to delete in folder $id")
-    files.foreach(file => client.deleteObject(
-      DeleteObjectRequest.builder().bucket(bucket).key(file.key()).build()
-    ))
-		logger.info(logMarker, s"Deleting images in folder $id from bucket $bucket")
-	}
+    files.foreach(file => deleteObject(bucket, bucket.keyFromURL(file.uri)))
+    logger.info(logMarker, s"Deleting images in folder $id from bucket $bucket")
+  }
 
 }

common-lib/src/main/scala/com/gu/mediaservice/lib/auth/KeyStore.scala

@@ -1,15 +1,15 @@
 package com.gu.mediaservice.lib.auth
 
 import com.gu.mediaservice.lib.BaseStore
+import com.gu.mediaservice.lib.aws.{S3, S3Bucket}
 import com.gu.mediaservice.lib.config.CommonConfig
 import com.gu.mediaservice.model.Instance
-import software.amazon.awssdk.services.s3.model.ListObjectsV2Request
 
-import scala.jdk.CollectionConverters._
-import scala.concurrent.ExecutionContext
+import scala.concurrent.duration._
+import scala.concurrent.{Await, ExecutionContext}
 
-class KeyStore(bucket: String, config: CommonConfig)(implicit ec: ExecutionContext)
-  extends BaseStore[String, ApiAccessor](bucket, config)(ec) {
+class KeyStore(bucket: S3Bucket, config: CommonConfig, s3: S3)(implicit ec: ExecutionContext)
+  extends BaseStore[String, ApiAccessor](bucket, config, s3)(ec) {
 
   def lookupIdentity(key: String)(implicit instance: Instance): Option[ApiAccessor] = store.get().get(instance.id + "/" + key)
 
@@ -20,9 +20,9 @@ class KeyStore(bucket: String, config: CommonConfig)(implicit ec: ExecutionConte
   }
 
   private def fetchAll: Map[String, ApiAccessor] = {
-    val contents = s3.client.listObjectsV2(ListObjectsV2Request.builder().bucket(bucket).build())
-      .contents().asScala.toList
-    val keys = contents.map(_.key())
+    val objects = Await.result(s3.listPaginating(bucket, None), 10.seconds)
+    logger.info(s"fetchAll found ${objects.size} objects")
+    val keys = objects.map( s3Object => bucket.keyFromURL(s3Object.uri))
     keys.flatMap(k => getS3Object(k).map(k -> ApiAccessor(_))).toMap
   }
 }