Merge pull request #4244 from bbc/conal/IMAGEDAM-1593-sendToCaptureSy…

…ndication

Conal/imagedam 1593 send to capture syndication

common-lib/src/main/scala/com/gu/mediaservice/lib/elasticsearch/MappingTest.scala

@@ -175,7 +175,8 @@ object MappingTest {
         composerUrl = Some(new URI("https://composer/api/2345678987654321345678"))
       )),
       syndicationUsageMetadata = Some(SyndicationUsageMetadata(
-        partnerName = "friends of ours"
+        partnerName = "friends of ours",
+        syndicatedBy = Some("Bob")
       )),
       frontUsageMetadata = Some(FrontUsageMetadata(
         addedBy = "me",

common-lib/src/main/scala/com/gu/mediaservice/lib/elasticsearch/Mappings.scala

@@ -281,7 +281,8 @@ object Mappings {
   ))
 
   def syndicationUsageMetadata(name: String): ObjectField = nonDynamicObjectField(name).copy(properties = Seq(
-    keywordField("partnerName")
+    keywordField("partnerName"),
+    keywordField("syndicatedBy")
   ))
 
   def frontUsageMetadata(name: String): ObjectField = nonDynamicObjectField(name).copy(properties = Seq(
@@ -293,7 +294,7 @@ object Mappings {
     keywordField("downloadedBy")
   ))
 
-  def usagesMapping(name: String): NestedField = nestedField(name).copy( properties = Seq(
+  def usagesMapping(name: String): NestedField = nestedField(name).copy(properties = Seq(
     keywordField("id"),
     sStemmerAnalysed("title"),
     usageReference("references"),

common-lib/src/main/scala/com/gu/mediaservice/lib/usage/ItemToMediaUsage.scala

@@ -1,7 +1,6 @@
 package com.gu.mediaservice.lib.usage
 
 import java.net.URI
-
 import com.amazonaws.services.dynamodbv2.document.Item
 import com.gu.mediaservice.model.usage._
 import org.joda.time.DateTime
@@ -52,7 +51,8 @@ object ItemToMediaUsage {
   private def buildSyndication(metadataMap: Map[String, Any]): Option[SyndicationUsageMetadata] = {
     Try {
       SyndicationUsageMetadata(
-        metadataMap("partnerName").asInstanceOf[String]
+        metadataMap("partnerName").asInstanceOf[String],
+        metadataMap.get("syndicatedBy").map(x => x.asInstanceOf[String])
       )
     }.toOption
   }

common-lib/src/main/scala/com/gu/mediaservice/lib/usage/UsageBuilder.scala

@@ -64,7 +64,7 @@ object UsageBuilder {
   private def buildSyndicationUsageReference(usage: MediaUsage): List[UsageReference] = usage.syndicationUsageMetadata.map (metadata => {
     List(
       UsageReference(
-        SyndicationUsageReference, None, Some(metadata.partnerName)
+        SyndicationUsageReference, None, metadata.syndicatedBy.map(_ => s"${metadata.partnerName}, ${metadata.syndicatedBy.get}").orElse(Some(metadata.partnerName))
       )
     )
   }).getOrElse(

common-lib/src/main/scala/com/gu/mediaservice/model/ThrallMessage.scala

@@ -76,6 +76,7 @@ object ExternalThrallMessage{
   implicit val updateImagePhotoshootMetadataMessage = Json.format[UpdateImagePhotoshootMetadataMessage]
   implicit val deleteUsagesMessage = Json.format[DeleteUsagesMessage]
   implicit val deleteSingleUsageMessage = Json.format[DeleteSingleUsageMessage]
+  implicit val updateUsageStatusMessage = Json.format[UpdateUsageStatusMessage]
   implicit val updateImageUsagesMessage = Json.format[UpdateImageUsagesMessage]
   implicit val addImageLeaseMessage = Json.format[AddImageLeaseMessage]
   implicit val removeImageLeaseMessage = Json.format[RemoveImageLeaseMessage]
@@ -124,6 +125,8 @@ case class DeleteSingleUsageMessage(id: String, lastModified: DateTime, usageId:
 
 case class DeleteUsagesMessage(id: String, lastModified: DateTime) extends ExternalThrallMessage
 
+case class UpdateUsageStatusMessage(id: String, usageNotice: UsageNotice, lastModified: DateTime) extends ExternalThrallMessage
+
 object DeleteUsagesMessage {
   implicit val yourJodaDateReads = JodaReads.DefaultJodaDateTimeReads.map(d => d.withZone(DateTimeZone.UTC))
   implicit val yourJodaDateWrites = JodaWrites.JodaDateTimeWrites

common-lib/src/main/scala/com/gu/mediaservice/model/usage/SyndicationUsageMetadata.scala

@@ -3,11 +3,12 @@ package com.gu.mediaservice.model.usage
 import play.api.libs.json._
 
 case class SyndicationUsageMetadata(
-  partnerName: String
+  partnerName: String,
+  syndicatedBy: Option[String] = None
 ) extends UsageMetadata {
   override def toMap: Map[String, Any] = Map(
     "partnerName" -> partnerName
-  )
+  ) ++ syndicatedBy.map("syndicatedBy" -> _)
 }
 
 object SyndicationUsageMetadata {

common-lib/src/main/scala/com/gu/mediaservice/model/usage/UsageStatus.scala

@@ -9,6 +9,7 @@ sealed trait UsageStatus {
     case RemovedUsageStatus => "removed"
     case SyndicatedUsageStatus => "syndicated"
     case DownloadedUsageStatus => "downloaded"
+    case FailedUsageStatus => "failed"
     case UnknownUsageStatus => "unknown"
   }
 }
@@ -20,7 +21,9 @@ object UsageStatus {
     case "removed" => RemovedUsageStatus
     case "syndicated" => SyndicatedUsageStatus
     case "downloaded" => DownloadedUsageStatus
+    case "failed" => FailedUsageStatus
     case "unknown" => UnknownUsageStatus
+    case _ => throw new IllegalArgumentException("Invalid usage status")
   }
 
   implicit val reads: Reads[UsageStatus] = JsPath.read[String].map(UsageStatus(_))
@@ -35,6 +38,7 @@ object PublishedUsageStatus extends UsageStatus
 object RemovedUsageStatus extends UsageStatus
 object SyndicatedUsageStatus extends UsageStatus
 object DownloadedUsageStatus extends UsageStatus
+object FailedUsageStatus extends UsageStatus
 
 // For Fronts usages as we don't know if a front is in draft or is live
 // TODO remove this once we do!

common-lib/src/main/scala/com/gu/mediaservice/syntax/MessageSubjects.scala

@@ -16,6 +16,7 @@ trait MessageSubjects {
   val AddImageLease = "add-image-lease"
   val RemoveImageLease = "remove-image-lease"
   val SetImageCollections = "set-image-collections"
+  val UpdateUsageStatus = "update-usage-status"
   val DeleteUsages = "delete-usages"
   val DeleteSingleUsage = "delete-single-usage"
   val UpdateImageSyndicationMetadata = "update-image-syndication-metadata"

thrall/app/lib/elasticsearch/ElasticSearch.scala

@@ -8,7 +8,7 @@ import com.gu.mediaservice.lib.formatting.printDateTime
 import com.gu.mediaservice.lib.logging.{LogMarker, MarkerMap}
 import com.gu.mediaservice.model._
 import com.gu.mediaservice.model.leases.MediaLease
-import com.gu.mediaservice.model.usage.Usage
+import com.gu.mediaservice.model.usage.{Usage, UsageNotice}
 import com.gu.mediaservice.syntax._
 import com.sksamuel.elastic4s.ElasticDsl._
 import com.sksamuel.elastic4s.requests.script.Script
@@ -20,6 +20,7 @@ import com.sksamuel.elastic4s.requests.update.UpdateRequest
 import com.sksamuel.elastic4s.{ElasticDsl, Executor, Functor, Handler, Response}
 import lib.{BatchDeletionIds, ThrallMetrics}
 import org.joda.time.DateTime
+import play.api.libs.json.JsValue.jsValueToJsLookup
 import play.api.libs.json._
 
 import scala.annotation.nowarn
@@ -503,6 +504,33 @@ class ElasticSearch(
     }))
   }
 
+  def updateUsageStatus(id: String, usages: Seq[Usage], lastModified: DateTime)
+                       (implicit ex: ExecutionContext, logMarker: LogMarker): List[Future[ElasticSearchUpdateResponse]] = {
+
+  val updateUsageStatusScript =
+    s"""
+       | for(int i = 0; i < ctx._source.usages.size(); i++) {
+       |    if(ctx._source.usages[i].id == params.usage.id) {
+       |        ctx._source.usages[i].status = params.usage.status;
+       |        ctx._source.usages[i].lastModified = params.lastModified;
+       |        ctx._source.usagesLastModified = params.lastModified;
+       |    }
+       | }
+       |""".stripMargin
+
+    val scriptSource = loadUpdatingModificationPainless(updateUsageStatusScript)
+
+    val usageParameters = JsDefined(Json.toJson(usages.head)).toOption.map(_.as[Usage]).map(i => asNestedMap(Json.toJson(i))).orNull
+
+    List(migrationAwareUpdater(
+      requestFromIndexName = indexName =>
+        prepareUpdateRequest(indexName, id, scriptSource, lastModified, ("usage", usageParameters)),
+      logMessageFromIndexName = indexName =>
+        s"ES6 updating usagesRights on image $id and usages id ${usageParameters.get("id")} " +
+          s"in index $indexName with usage $usageParameters"
+    ).map(_ => ElasticSearchUpdateResponse()))
+  }
+
   def deleteSyndicationRights(id: String, lastModified: DateTime)
                              (implicit ex: ExecutionContext, logMarker: LogMarker): List[Future[ElasticSearchUpdateResponse]] = {
     val deleteSyndicationRightsScript = s"""
@@ -777,4 +805,5 @@ class ElasticSearch(
 
     image.transform(removeUploadInformation()).get
   }
+
 }

thrall/app/lib/kinesis/MessageProcessor.scala

@@ -49,6 +49,7 @@ class MessageProcessor(
       case message: CreateMigrationIndexMessage => createMigrationIndex(message, logMarker)
       case message: MigrateImageMessage => migrateImage(message, logMarker)
       case message: UpsertFromProjectionMessage => upsertImageFromProjection(message, logMarker)
+      case message: UpdateUsageStatusMessage => updateUsageStatus(message, logMarker)
       case _: CompleteMigrationMessage => completeMigration(logMarker)
     }
   }
@@ -182,6 +183,14 @@ class MessageProcessor(
     Future.sequence(es.deleteSingleImageUsage(message.id, message.usageId, message.lastModified)(ec, logMarker))
   }
 
+  private def updateUsageStatus(message: UpdateUsageStatusMessage, logMarker: LogMarker)(implicit ec: ExecutionContext): Future[List[ElasticSearchUpdateResponse]] = {
+      implicit val lm: LogMarker = combineMarkers(message, logMarker)
+      val usage = message.usageNotice.usageJson.as[Seq[Usage]]
+      Future.traverse(es.updateUsageStatus(message.id, usage, message.lastModified ))(_.recoverWith {
+        case ElasticNotFoundException => Future.successful(ElasticSearchUpdateResponse())
+    })
+  }
+
   def upsertSyndicationRightsOnly(message: UpdateImageSyndicationMetadataMessage, logMarker: LogMarker)(implicit ec: ExecutionContext): Future[Any] = {
     implicit val marker: LogMarker = logMarker ++ imageIdMarker(ImageId(message.id))
     es.getImage(message.id) map {

thrall/app/lib/kinesis/MessageTranslator.scala

@@ -73,6 +73,10 @@ object MessageTranslator extends GridLogging {
         case (Some(id), Some(edits)) => Right(UpdateImagePhotoshootMetadataMessage(id, updateMessage.lastModified, edits))
         case _ => Left(MissingFieldsException(updateMessage.subject))
       }
+      case UpdateUsageStatus => (updateMessage.id, updateMessage.usageNotice ) match {
+        case (Some(id), Some(usageNotice)) => Right(UpdateUsageStatusMessage(id, usageNotice, updateMessage.lastModified))
+        case _ => Left(MissingFieldsException(updateMessage.subject))
+      }
       case _ => Left(ProcessorNotFoundException(updateMessage.subject))
     }
   }

thrall/test/lib/elasticsearch/ElasticSearchTest.scala

@@ -5,6 +5,7 @@ import com.gu.mediaservice.lib.logging.{LogMarker, MarkerMap}
 import com.gu.mediaservice.model
 import com.gu.mediaservice.model._
 import com.gu.mediaservice.model.leases.{LeasesByMedia, MediaLease}
+import com.gu.mediaservice.model.usage.{PublishedUsageStatus, SyndicatedUsageStatus}
 import com.gu.mediaservice.model.usage.Usage
 import com.sksamuel.elastic4s.ElasticDsl
 import com.sksamuel.elastic4s.ElasticDsl._
@@ -534,6 +535,18 @@ class ElasticSearchTest extends ElasticSearchTestBase {
 
         reloadedImage(id).get.usages.head.id shouldEqual ("recent")
       }
+
+      "can update usage status for single image" in {
+        val id = UUID.randomUUID().toString
+        val imageWithUsages = createImageForSyndication(id = UUID.randomUUID().toString, true, Some(now), None).copy(usages = List(usage(), usage()))
+        val usageWithUpdatedUsageStatus = imageWithUsages.usages.head.copy(status = SyndicatedUsageStatus)
+
+        Await.result(ES.migrationAwareIndexImage(id, imageWithUsages, now), fiveSeconds)
+
+        Await.result(Future.sequence(ES.updateUsageStatus(id, List(usageWithUpdatedUsageStatus), now)), fiveSeconds)
+        reloadedImage(id).get.usages.head.status shouldBe (SyndicatedUsageStatus)
+        reloadedImage(id).get.usages.last.status shouldBe (PublishedUsageStatus)
+      }
     }
 
     "syndication rights" - {

usage/app/controllers/UsageApi.scala

@@ -9,11 +9,11 @@ import com.gu.mediaservice.lib.aws.UpdateMessage
 import com.gu.mediaservice.lib.logging.{LogMarker, MarkerMap}
 import com.gu.mediaservice.lib.play.RequestLoggingFilter
 import com.gu.mediaservice.lib.usage.UsageBuilder
-import com.gu.mediaservice.model.usage.{MediaUsage, Usage}
+import com.gu.mediaservice.model.usage.{MediaUsage, SyndicatedUsageStatus, Usage, UsageNotice, UsageStatus}
 import com.gu.mediaservice.syntax.MessageSubjects
 import lib._
 import model._
-import play.api.libs.json.{JsError, JsValue}
+import play.api.libs.json.{JsArray, JsError, JsValue, Json}
 import play.api.mvc._
 import play.utils.UriEncoding
 import rx.lang.scala.Subject
@@ -256,6 +256,45 @@ class UsageApi(
     )
   }}
 
+  def updateUsageStatus(mediaId: String, usageId: String) = auth.async(parse.json) {req => {
+    val request = (req.body \ "data").validate[UsageStatus]
+    request.fold(
+      e => Future.successful(
+        respondError(
+          BadRequest,
+          errorKey = "update-image-usage-status-failed",
+          errorMessage = JsError.toJson(e).toString()
+        )
+      ),
+      usageStatus => {
+        implicit val logMarker: LogMarker = MarkerMap(
+          "requestType" -> "update-usage-status",
+          "requestId" -> RequestLoggingFilter.getRequestId(req),
+          "usageStatus" -> usageStatus.toString,
+          "image-id" -> mediaId,
+          "usage-id" -> usageId,
+        ) ++ apiKeyMarkers(req.user.accessor)
+        logger.info(logMarker, "recording usage status  update")
+
+        usageTable.queryByUsageId(usageId).map {
+          case Some(mediaUsage) =>
+            val updatedStatusMediaUsage = mediaUsage.copy(status = usageStatus)
+            usageTable.update(updatedStatusMediaUsage)
+            val usageNotice = UsageNotice(mediaId,
+              JsArray(Seq(Json.toJson(UsageBuilder.build(updatedStatusMediaUsage)))))
+            val updateMessage = UpdateMessage(
+              subject = UpdateUsageStatus, id = Some(mediaId),
+              usageNotice = Some(usageNotice)
+            )
+            notifications.publish(updateMessage)
+            Ok
+          case None =>
+            NotFound
+        }
+      }
+    )
+  }}
+
   def deleteSingleUsage(mediaId: String, usageId: String) = AuthenticatedAndAuthorisedToDelete.async { req =>
     implicit val logMarker: LogMarker = MarkerMap(
       "requestType" -> "delete-usage",

usage/app/model/SyndicationUsageRequest.scala

@@ -1,16 +1,21 @@
 package model
 
-import com.gu.mediaservice.model.usage.{SyndicatedUsageStatus, SyndicationUsageMetadata, UsageStatus}
+import com.gu.mediaservice.model.usage.{PendingUsageStatus, SyndicatedUsageStatus, SyndicationUsageMetadata, UsageStatus}
 import org.joda.time.DateTime
 import play.api.libs.json._
 
 case class SyndicationUsageRequest (
   partnerName: String,
+  syndicatedBy: Option[String],
+  startPending: Option[Boolean],
   mediaId: String,
   dateAdded: DateTime
 ) {
-  val status: UsageStatus = SyndicatedUsageStatus
-  val metadata: SyndicationUsageMetadata = SyndicationUsageMetadata(partnerName)
+  val status: UsageStatus = startPending match {
+    case Some(true) => PendingUsageStatus
+    case _ => SyndicatedUsageStatus
+  }
+  val metadata: SyndicationUsageMetadata = SyndicationUsageMetadata(partnerName, syndicatedBy)
 }
 object SyndicationUsageRequest {
   import JodaWrites._

usage/app/model/UsageGroup.scala

@@ -29,6 +29,7 @@ class UsageGroupOps(config: UsageConfig, mediaWrapperOps: MediaWrapperOps)
   def buildId(syndicationUsageRequest: SyndicationUsageRequest): String = s"syndication/${
     MD5.hash(List(
       syndicationUsageRequest.metadata.partnerName,
+      syndicationUsageRequest.metadata.syndicatedBy,
       syndicationUsageRequest.mediaId
     ).mkString("_"))
   }"

usage/app/model/UsageIdBuilder.scala

@@ -23,6 +23,7 @@ object UsageIdBuilder {
   def build(syndicationUsageRequest: SyndicationUsageRequest) = buildId(List(
     Some(syndicationUsageRequest.mediaId),
     Some(syndicationUsageRequest.metadata.partnerName),
+    syndicationUsageRequest.metadata.syndicatedBy,
     Some(syndicationUsageRequest.status)
   ))
 

usage/conf/routes

@@ -8,7 +8,7 @@ POST    /usages/print                                   controllers.UsageApi.set
 POST    /usages/syndication                             controllers.UsageApi.setSyndicationUsages
 POST    /usages/front                                   controllers.UsageApi.setFrontUsages
 POST    /usages/download                                controllers.UsageApi.setDownloadUsages
-
+PUT     /usages/status/update/:mediaId/*usageId         controllers.UsageApi.updateUsageStatus(mediaId: String, usageId: String)
 GET     /usages/digital/content/*contentId/reindex      controllers.UsageApi.reindexForContent(contentId: String)
 
 # Management