<commit_before>package uk.ac.wellcome.platform.transformer.receive

import com.amazonaws.services.s3.AmazonS3
import com.google.inject.Inject
import com.twitter.inject.Logging
import io.circe.ParsingFailure
import uk.ac.wellcome.exceptions.GracefulFailureException
import uk.ac.wellcome.messaging.message.MessageWriter
import uk.ac.wellcome.messaging.sqs.SQSMessage
import uk.ac.wellcome.models.SourceMetadata
import uk.ac.wellcome.models.transformable.{
  CalmTransformable,
  MiroTransformable,
  SierraTransformable,
  Transformable
}
import uk.ac.wellcome.models.work.internal.UnidentifiedWork
import uk.ac.wellcome.monitoring.MetricsSender
import uk.ac.wellcome.storage.s3.{S3Config, S3ObjectLocation, S3StringStore, S3TypedObjectStore}
import uk.ac.wellcome.storage.vhs.HybridRecord
import uk.ac.wellcome.platform.transformer.transformers.{
  CalmTransformableTransformer,
  MiroTransformableTransformer,
  SierraTransformableTransformer
}
import uk.ac.wellcome.utils.GlobalExecutionContext.context
import uk.ac.wellcome.utils.JsonUtil._

import scala.concurrent.Future
import scala.util.Try

class SQSMessageReceiver @Inject()(
  messageWriter: MessageWriter[UnidentifiedWork],
  s3Client: AmazonS3,
  s3Config: S3Config,
  metricsSender: MetricsSender)
    extends Logging {

  val s3StringStore = new S3StringStore(
    s3Client = s3Client,
    s3Config = s3Config
  )

  def receiveMessage(message: SQSMessage): Future[Unit] = {
    debug(s"Starting to process message $message")
    metricsSender.timeAndCount(
      "transform-time",
      () => {
        val futurePublishAttempt = for {
          hybridRecord <- Future.fromTry(fromJson[HybridRecord](message.body))
          sourceMetadata <- Future.fromTry(
            fromJson[SourceMetadata](message.body))
          transformableRecord <- getTransformable(hybridRecord, sourceMetadata)
          cleanRecord <- Future.fromTry(
            transformTransformable(transformableRecord, hybridRecord.version))
          publishResult <- publishMessage(cleanRecord)
        } yield publishResult

        futurePublishAttempt
          .recover {
            case e: ParsingFailure =>
              info("Recoverable failure parsing HybridRecord from message", e)
              throw GracefulFailureException(e)
          }
          .map(_ => ())
      }
    )
  }

  val miroTransformableStore = new S3TypedObjectStore[MiroTransformable](s3StringStore)
  val calmTransformableStore = new S3TypedObjectStore[CalmTransformable](s3StringStore)
  val sierraTransformableStore = new S3TypedObjectStore[SierraTransformable](s3StringStore)

  private def getTransformable(
    hybridRecord: HybridRecord,
    sourceMetadata: SourceMetadata
  ) = {
    val s3ObjectLocation = S3ObjectLocation(
      bucket = s3Config.bucketName,
      key = hybridRecord.s3key
    )

    sourceMetadata.sourceName match {
      case "miro" => miroTransformableStore.get(s3ObjectLocation)
      case "calm" => calmTransformableStore.get(s3ObjectLocation)
      case "sierra" => sierraTransformableStore.get(s3ObjectLocation)
    }
  }

  private def transformTransformable(
    transformable: Transformable,
    version: Int
  ): Try[Option[UnidentifiedWork]] = {
    val transformableTransformer = chooseTransformer(transformable)
    transformableTransformer.transform(transformable, version) map {
      transformed =>
        debug(s"Transformed record to $transformed")
        transformed
    } recover {<commit_msg>Apply auto-formatting rules<commit_after>package uk.ac.wellcome.platform.transformer.receive

import com.amazonaws.services.s3.AmazonS3
import com.google.inject.Inject
import com.twitter.inject.Logging
import io.circe.ParsingFailure
import uk.ac.wellcome.exceptions.GracefulFailureException
import uk.ac.wellcome.messaging.message.MessageWriter
import uk.ac.wellcome.messaging.sqs.SQSMessage
import uk.ac.wellcome.models.SourceMetadata
import uk.ac.wellcome.models.transformable.{
  CalmTransformable,
  MiroTransformable,
  SierraTransformable,
  Transformable
}
import uk.ac.wellcome.models.work.internal.UnidentifiedWork
import uk.ac.wellcome.monitoring.MetricsSender
import uk.ac.wellcome.storage.s3.{
  S3Config,
  S3ObjectLocation,
  S3StringStore,
  S3TypedObjectStore
}
import uk.ac.wellcome.storage.vhs.HybridRecord
import uk.ac.wellcome.platform.transformer.transformers.{
  CalmTransformableTransformer,
  MiroTransformableTransformer,
  SierraTransformableTransformer
}
import uk.ac.wellcome.utils.GlobalExecutionContext.context
import uk.ac.wellcome.utils.JsonUtil._

import scala.concurrent.Future
import scala.util.Try

class SQSMessageReceiver @Inject()(
  messageWriter: MessageWriter[UnidentifiedWork],
  s3Client: AmazonS3,
  s3Config: S3Config,
  metricsSender: MetricsSender)
    extends Logging {

  val s3StringStore = new S3StringStore(
    s3Client = s3Client,
    s3Config = s3Config
  )

  def receiveMessage(message: SQSMessage): Future[Unit] = {
    debug(s"Starting to process message $message")
    metricsSender.timeAndCount(
      "transform-time",
      () => {
        val futurePublishAttempt = for {
          hybridRecord <- Future.fromTry(fromJson[HybridRecord](message.body))
          sourceMetadata <- Future.fromTry(
            fromJson[SourceMetadata](message.body))
          transformableRecord <- getTransformable(hybridRecord, sourceMetadata)
          cleanRecord <- Future.fromTry(
            transformTransformable(transformableRecord, hybridRecord.version))
          publishResult <- publishMessage(cleanRecord)
        } yield publishResult

        futurePublishAttempt
          .recover {
            case e: ParsingFailure =>
              info("Recoverable failure parsing HybridRecord from message", e)
              throw GracefulFailureException(e)
          }
          .map(_ => ())
      }
    )
  }

  val miroTransformableStore =
    new S3TypedObjectStore[MiroTransformable](s3StringStore)
  val calmTransformableStore =
    new S3TypedObjectStore[CalmTransformable](s3StringStore)
  val sierraTransformableStore =
    new S3TypedObjectStore[SierraTransformable](s3StringStore)

  private def getTransformable(
    hybridRecord: HybridRecord,
    sourceMetadata: SourceMetadata
  ) = {
    val s3ObjectLocation = S3ObjectLocation(
      bucket = s3Config.bucketName,
      key = hybridRecord.s3key
    )

    sourceMetadata.sourceName match {
      case "miro" => miroTransformableStore.get(s3ObjectLocation)
      case "calm" => calmTransformableStore.get(s3ObjectLocation)
      case "sierra" => sierraTransformableStore.get(s3ObjectLocation)
    }
  }

  private def transformTransformable(
    transformable: Transformable,
    version: Int
  ): Try[Option[UnidentifiedWork]] = {
    val transformableTransformer = chooseTransformer(transformable)
    transformableTransformer.transform(transformable, version) map {
      transformed =>
        debug(s"Transformed record to $transformed")
        transformed
    } recover {
{'fraction_non_alphanumeric': 0.08342906375646181, 'fraction_numerical': 0.00861573808156232, 'mean_word_length': 3.975, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1799709', 'n_tokens_mistral': 2043, 'n_tokens_neox': 2004, 'n_words': 347}