Merge pull request #4145 from guardian/reaper-in-thrall

.github/dependabot.yml

@@ -47,12 +47,6 @@ updates:
 #
 #  # things that run in an AWS Lambda environment
 #  - package-ecosystem: 'npm'
-#    directory: '/reaper'
-#    schedule:
-#      interval: 'weekly'
-#    commit-message:
-#      prefix: "chore(deps): "
-#  - package-ecosystem: 'npm'
 #    directory: '/image-counter-lambda'
 #    schedule:
 #      interval: 'weekly'

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

@@ -1,5 +1,7 @@
 package com.gu.mediaservice.lib
 
+import com.amazonaws.services.s3.model.{DeleteObjectsRequest, MultiObjectDeleteException}
+
 import java.io.File
 import com.gu.mediaservice.lib.config.CommonConfig
 import com.gu.mediaservice.lib.aws.S3Object
@@ -9,6 +11,8 @@ import org.joda.time.DateTime
 
 import scala.concurrent.Future
 
+import scala.jdk.CollectionConverters.iterableAsScalaIterableConverter
+
 object ImageIngestOperations {
   def fileKeyFromId(id: String): String = id.take(6).mkString("/") + "/" + id
 
@@ -42,9 +46,33 @@ class ImageIngestOperations(imageBucket: String, thumbnailBucket: String, config
     storeImage(imageBucket, optimisedPngKeyFromId(storableImage.id), storableImage.file, Some(storableImage.mimeType),
       overwrite = true)
 
+  private def bulkDelete(bucket: String, keys: List[String]): Future[Map[String, Boolean]] = keys match {
+    case Nil => Future.successful(Map.empty)
+    case _ => Future {
+      try {
+        client.deleteObjects(
+          new DeleteObjectsRequest(bucket).withKeys(keys: _*)
+        )
+        keys.map { key =>
+          key -> true
+        }.toMap
+      } catch {
+        case partialFailure: MultiObjectDeleteException =>
+          logger.warn(s"Partial failure when deleting images from $bucket: ${partialFailure.getMessage} ${partialFailure.getErrors}")
+          val errorKeys = partialFailure.getErrors.asScala.map(_.getKey).toSet
+          keys.map { key =>
+            key -> !errorKeys.contains(key)
+          }.toMap
+      }
+    }
+  }
+
   def deleteOriginal(id: String)(implicit logMarker: LogMarker): Future[Unit] = if(isVersionedS3) deleteVersionedImage(imageBucket, fileKeyFromId(id)) else deleteImage(imageBucket, fileKeyFromId(id))
+  def deleteOriginals(ids: Set[String]) = bulkDelete(imageBucket, ids.map(fileKeyFromId).toList)
   def deleteThumbnail(id: String)(implicit logMarker: LogMarker): Future[Unit] = deleteImage(thumbnailBucket, fileKeyFromId(id))
-  def deletePng(id: String)(implicit logMarker: LogMarker): Future[Unit] = deleteImage(imageBucket, optimisedPngKeyFromId(id))
+  def deleteThumbnails(ids: Set[String]) = bulkDelete(thumbnailBucket, ids.map(fileKeyFromId).toList)
+  def deletePNG(id: String)(implicit logMarker: LogMarker): Future[Unit] = deleteImage(imageBucket, optimisedPngKeyFromId(id))
+  def deletePNGs(ids: Set[String]) = bulkDelete(imageBucket, ids.map(optimisedPngKeyFromId).toList)
 
   def doesOriginalExist(id: String): Boolean =
     client.doesObjectExist(imageBucket, fileKeyFromId(id))

common-lib/src/main/scala/com/gu/mediaservice/lib/metadata/SoftDeletedMetadataTable.scala

@@ -1,12 +1,14 @@
 package com.gu.mediaservice.lib.metadata
 
+import com.amazonaws.services.dynamodbv2.model.BatchWriteItemResult
 import com.gu.mediaservice.lib.aws.DynamoDB
 import com.gu.mediaservice.lib.config.CommonConfig
 import com.gu.mediaservice.model.ImageStatusRecord
 import com.gu.scanamo._
 import com.gu.scanamo.syntax._
 
-import scala.concurrent.ExecutionContext
+import scala.concurrent.{ExecutionContext, Future}
+import scala.jdk.CollectionConverters.collectionAsScalaIterableConverter
 
 class SoftDeletedMetadataTable(config: CommonConfig) extends DynamoDB[ImageStatusRecord](config, config.softDeletedMetadataTable) {
   private val softDeletedMetadataTable = Table[ImageStatusRecord](table.getTableName)
@@ -19,6 +21,19 @@ class SoftDeletedMetadataTable(config: CommonConfig) extends DynamoDB[ImageStatu
     ScanamoAsync.exec(client)(softDeletedMetadataTable.put(imageStatus))
   }
 
+  private def extractUnprocessedIds(results: List[BatchWriteItemResult]): List[String] =
+    results.flatMap(_.getUnprocessedItems.values().asScala.flatMap(_.asScala.map(_.getPutRequest.getItem.get("id").getS)))
+
+  def setStatuses(imageStatuses: Set[ImageStatusRecord])(implicit ex: ExecutionContext) = {
+    if (imageStatuses.isEmpty) Future.successful(List.empty)
+    else ScanamoAsync.exec(client)(softDeletedMetadataTable.putAll(imageStatuses)).map(extractUnprocessedIds)
+  }
+
+  def clearStatuses(imageIds: Set[String])(implicit ex: ExecutionContext) = {
+    if (imageIds.isEmpty) Future.successful(List.empty)
+    else ScanamoAsync.exec(client)(softDeletedMetadataTable.deleteAll('id -> imageIds)).map(extractUnprocessedIds)
+  }
+
   def updateStatus(imageId: String, isDeleted: Boolean)(implicit ex: ExecutionContext) = {
     val updateExpression = set('isDeleted -> isDeleted)
     ScanamoAsync.exec(client)(