<commit_before>package im.actor.server.api.rpc.service.files

import java.io.{ File, FileOutputStream }
import java.nio.file.Files

import scala.collection.mutable
import scala.concurrent.duration._
import scala.concurrent.forkjoin.ThreadLocalRandom
import scala.concurrent.{ ExecutionContext, Future }

import akka.actor._
import com.amazonaws.HttpMethod
import com.amazonaws.services.s3.model._
import com.amazonaws.services.s3.transfer.TransferManager
import com.github.dwhjames.awswrap.s3.{ AmazonS3ScalaClient, FutureTransfer }
import org.apache.commons.io.FileUtils
import slick.driver.PostgresDriver.api._

import im.actor.api.rpc.files._
import im.actor.api.rpc.{ ClientData, _ }
import im.actor.server.api.util.ACL
import im.actor.server.{ models, persist }

class FilesServiceImpl(bucketName: String)(implicit
  s3Client: AmazonS3ScalaClient,
                                           transferManager: TransferManager,
                                           db:              Database,
                                           actorSystem:     ActorSystem) extends FilesService {

  import scala.collection.JavaConverters._

  override implicit val ec: ExecutionContext = actorSystem.dispatcher

  object Errors {
    val FileNotFound = RpcError(404, "FILE_NOT_FOUND", "File not found.", false, None)
    val LocationInvalid = RpcError(400, "LOCATION_INVALID", "", false, None)
  }

  override def jhandleGetFileUrl(location: FileLocation, clientData: ClientData): Future[HandlerResult[ResponseGetFileUrl]] = {
    val authorizedAction = requireAuth(clientData).map { client ⇒
      persist.File.find(location.fileId) flatMap {
        case Some(file) ⇒
          if (ACL.fileAccessHash(file.id, file.accessSalt) == location.accessHash) {
            val presignedRequest = new GeneratePresignedUrlRequest(bucketName, s"file_${file.id}")
            val timeout = 1.day

            val expiration = new java.util.Date
            expiration.setTime(expiration.getTime + timeout.toMillis)
            presignedRequest.setExpiration(expiration)
            presignedRequest.setMethod(HttpMethod.GET)

            for {
              url ← DBIO.from(s3Client.generatePresignedUrlRequest(presignedRequest))
            } yield {
              Ok(ResponseGetFileUrl(url.toString, timeout.toSeconds.toInt))
            }
          } else {
            DBIO.successful(Error(Errors.LocationInvalid))
          }
        case None ⇒ DBIO.successful(Error(Errors.LocationInvalid))
      }
    }

    db.run(toDBIOAction(authorizedAction))
  }

  override def jhandleGetFileUploadUrl(expectedSize: Int, clientData: ClientData): Future[HandlerResult[ResponseGetFileUploadUrl]] = {
    val authorizedAction = requireAuth(clientData).map { client ⇒
      val rnd = ThreadLocalRandom.current()
      val id = rnd.nextLong()
      val salt = ACL.nextAccessSalt(rnd)

      val key = s"upload_${id}"
      val presignedRequest = new GeneratePresignedUrlRequest(bucketName, key)
      val expiration = new java.util.Date
      expiration.setTime(expiration.getTime + 1.day.toMillis)
      presignedRequest.setExpiration(expiration)
      presignedRequest.setMethod(HttpMethod.PUT)

      for {
        _ ← persist.File.create(id, salt, key)
        url ← DBIO.from(s3Client.generatePresignedUrlRequest(presignedRequest))
      } yield {
        Ok(ResponseGetFileUploadUrl(url.toString, key.getBytes()))
      }
    }

    db.run(toDBIOAction(authorizedAction))
  }

  override def jhandleGetFileUploadPartUrl(partNumber: Int, partSize: Int, uploadKey: Array[Byte], clientData: ClientData): Future[HandlerResult[ResponseGetFileUploadPartUrl]] = {
    val authorizedAction = requireAuth(clientData).map { client ⇒
      val key = new String(uploadKey)

      persist.File.findByKey(key) flatMap {
        case Some(file) ⇒
          val partKey = s"upload_part_${file.s3UploadKey}_${partNumber}"
          val request = new GeneratePresignedUrlRequest(bucketName, partKey)
          val expiration = new java.util.Date
          expiration.setTime(expiration.getTime + 1.day.toMillis)
          request.setMethod(HttpMethod.PUT)
          request.setExpiration(expiration)
          request.setContentType("multipart/form-data")

          for {
            url ← DBIO.from(s3Client.generatePresignedUrlRequest(request))
            _ ← persist.FilePart.create(file.id, partNumber, partSize, partKey)
          } yield {
            Ok(ResponseGetFileUploadPartUrl(url.toString))
          }
        case None ⇒
          DBIO.successful(Error(Errors.FileNotFound))
      }
    }

    db.run(toDBIOAction(authorizedAction))
  }

  override def jhandleCommitFileUpload(uploadKey: Array[Byte], clientData: ClientData): Future[HandlerResult[ResponseCommitFileUpload]] = {
    val authorizedAction = requireAuth(clientData).map { client ⇒
      val key = new String(uploadKey)

      persist.File.findByKey(key) flatMap {
        case Some(file) ⇒
          for {
            parts ← persist.FilePart.findByFileId(file.id)
            tempDir ← DBIO.from(createTempDir())
            download = FutureTransfer.listenFor {
              transferManager.downloadDirectory(bucketName, s"upload_part_${file.s3UploadKey}", tempDir)
            } map (_.waitForCompletion())
            _ ← DBIO.from(download)
            concatFile ← DBIO.from(concatFiles(tempDir, parts map (_.s3UploadKey)))
            upload = FutureTransfer.listenFor {
              transferManager.upload(bucketName, s"file_${file.id}", concatFile)
            } map (_.waitForCompletion())
            _ ← DBIO.from(upload)
            _ ← DBIO.from(deleteDir(tempDir))
            _ ← persist.File.setUploaded(file.id)
          } yield {
            Ok(ResponseCommitFileUpload(FileLocation(file.id, ACL.fileAccessHash(file.id, file.accessSalt))))
          }
        case None ⇒
          DBIO.successful(Error(Errors.FileNotFound))
      }
    }

    db.run(toDBIOAction(authorizedAction))
  }

  private def copyPartRequest(part: models.FilePart, destinationUploadId: String, destinationKey: String): CopyPartRequest = {
    (new CopyPartRequest)
      .withDestinationBucketName(bucketName)
      .withDestinationKey(destinationKey)
      .withSourceBucketName(bucketName)
      .withSourceKey(part.s3UploadKey)
      .withUploadId(destinationUploadId)
      .withPartNumber(part.number)
  }

  private def etags(responses: Seq[CopyPartResult]): java.util.List[PartETag] = {
    val xs = responses map { response ⇒
      new PartETag(response.getPartNumber(), response.getETag())
    }

    mutable.Seq(xs: _*).asJava
  }

  // FIXME: #perf pinned dispatcher

  private def createTempDir(): Future[File] = {
    Future {
      com.google.common.io.Files.createTempDir()
    }
  }

  // FIXME: #perf use nio and pinned dispatcher

  private def concatFiles(dir: File, fileNames: Seq[String]): Future[File] = {
    Future {
      val dirPath = dir.toPath
      val concatFile = dirPath.resolve("concatenated").toFile

      val outStream = new FileOutputStream(concatFile)

      fileNames foreach { fileName ⇒
        Files.copy(dirPath.resolve(fileName), outStream)
      }

      outStream.close()

      concatFile
    }
  }

  private def deleteDir(dir: File): Future[Unit] = {
    Future {
      FileUtils.deleteDirectory(dir)
    }
  }
}
<commit_msg>fix(files): application/octet-stream for files<commit_after>package im.actor.server.api.rpc.service.files

import java.io.{ File, FileOutputStream }
import java.nio.file.Files

import scala.collection.mutable
import scala.concurrent.duration._
import scala.concurrent.forkjoin.ThreadLocalRandom
import scala.concurrent.{ ExecutionContext, Future }

import akka.actor._
import com.amazonaws.HttpMethod
import com.amazonaws.services.s3.model._
import com.amazonaws.services.s3.transfer.TransferManager
import com.github.dwhjames.awswrap.s3.{ AmazonS3ScalaClient, FutureTransfer }
import org.apache.commons.io.FileUtils
import slick.driver.PostgresDriver.api._

import im.actor.api.rpc.files._
import im.actor.api.rpc.{ ClientData, _ }
import im.actor.server.api.util.ACL
import im.actor.server.{ models, persist }

class FilesServiceImpl(bucketName: String)(implicit
  s3Client: AmazonS3ScalaClient,
                                           transferManager: TransferManager,
                                           db:              Database,
                                           actorSystem:     ActorSystem) extends FilesService {

  import scala.collection.JavaConverters._

  override implicit val ec: ExecutionContext = actorSystem.dispatcher

  object Errors {
    val FileNotFound = RpcError(404, "FILE_NOT_FOUND", "File not found.", false, None)
    val LocationInvalid = RpcError(400, "LOCATION_INVALID", "", false, None)
  }

  override def jhandleGetFileUrl(location: FileLocation, clientData: ClientData): Future[HandlerResult[ResponseGetFileUrl]] = {
    val authorizedAction = requireAuth(clientData).map { client ⇒
      persist.File.find(location.fileId) flatMap {
        case Some(file) ⇒
          if (ACL.fileAccessHash(file.id, file.accessSalt) == location.accessHash) {
            val presignedRequest = new GeneratePresignedUrlRequest(bucketName, s"file_${file.id}")
            val timeout = 1.day

            val expiration = new java.util.Date
            expiration.setTime(expiration.getTime + timeout.toMillis)
            presignedRequest.setExpiration(expiration)
            presignedRequest.setMethod(HttpMethod.GET)

            for {
              url ← DBIO.from(s3Client.generatePresignedUrlRequest(presignedRequest))
            } yield {
              Ok(ResponseGetFileUrl(url.toString, timeout.toSeconds.toInt))
            }
          } else {
            DBIO.successful(Error(Errors.LocationInvalid))
          }
        case None ⇒ DBIO.successful(Error(Errors.LocationInvalid))
      }
    }

    db.run(toDBIOAction(authorizedAction))
  }

  override def jhandleGetFileUploadUrl(expectedSize: Int, clientData: ClientData): Future[HandlerResult[ResponseGetFileUploadUrl]] = {
    val authorizedAction = requireAuth(clientData).map { client ⇒
      val rnd = ThreadLocalRandom.current()
      val id = rnd.nextLong()
      val salt = ACL.nextAccessSalt(rnd)

      val key = s"upload_${id}"
      val presignedRequest = new GeneratePresignedUrlRequest(bucketName, key)
      val expiration = new java.util.Date
      expiration.setTime(expiration.getTime + 1.day.toMillis)
      presignedRequest.setExpiration(expiration)
      presignedRequest.setMethod(HttpMethod.PUT)

      for {
        _ ← persist.File.create(id, salt, key)
        url ← DBIO.from(s3Client.generatePresignedUrlRequest(presignedRequest))
      } yield {
        Ok(ResponseGetFileUploadUrl(url.toString, key.getBytes()))
      }
    }

    db.run(toDBIOAction(authorizedAction))
  }

  override def jhandleGetFileUploadPartUrl(partNumber: Int, partSize: Int, uploadKey: Array[Byte], clientData: ClientData): Future[HandlerResult[ResponseGetFileUploadPartUrl]] = {
    val authorizedAction = requireAuth(clientData).map { client ⇒
      val key = new String(uploadKey)

      persist.File.findByKey(key) flatMap {
        case Some(file) ⇒
          val partKey = s"upload_part_${file.s3UploadKey}_${partNumber}"
          val request = new GeneratePresignedUrlRequest(bucketName, partKey)
          val expiration = new java.util.Date
          expiration.setTime(expiration.getTime + 1.day.toMillis)
          request.setMethod(HttpMethod.PUT)
          request.setExpiration(expiration)
          request.setContentType("application/octet-stream")

          for {
            url ← DBIO.from(s3Client.generatePresignedUrlRequest(request))
            _ ← persist.FilePart.create(file.id, partNumber, partSize, partKey)
          } yield {
            Ok(ResponseGetFileUploadPartUrl(url.toString))
          }
        case None ⇒
          DBIO.successful(Error(Errors.FileNotFound))
      }
    }

    db.run(toDBIOAction(authorizedAction))
  }

  override def jhandleCommitFileUpload(uploadKey: Array[Byte], clientData: ClientData): Future[HandlerResult[ResponseCommitFileUpload]] = {
    val authorizedAction = requireAuth(clientData).map { client ⇒
      val key = new String(uploadKey)

      persist.File.findByKey(key) flatMap {
        case Some(file) ⇒
          for {
            parts ← persist.FilePart.findByFileId(file.id)
            tempDir ← DBIO.from(createTempDir())
            download = FutureTransfer.listenFor {
              transferManager.downloadDirectory(bucketName, s"upload_part_${file.s3UploadKey}", tempDir)
            } map (_.waitForCompletion())
            _ ← DBIO.from(download)
            concatFile ← DBIO.from(concatFiles(tempDir, parts map (_.s3UploadKey)))
            upload = FutureTransfer.listenFor {
              transferManager.upload(bucketName, s"file_${file.id}", concatFile)
            } map (_.waitForCompletion())
            _ ← DBIO.from(upload)
            _ ← DBIO.from(deleteDir(tempDir))
            _ ← persist.File.setUploaded(file.id)
          } yield {
            Ok(ResponseCommitFileUpload(FileLocation(file.id, ACL.fileAccessHash(file.id, file.accessSalt))))
          }
        case None ⇒
          DBIO.successful(Error(Errors.FileNotFound))
      }
    }

    db.run(toDBIOAction(authorizedAction))
  }

  private def copyPartRequest(part: models.FilePart, destinationUploadId: String, destinationKey: String): CopyPartRequest = {
    (new CopyPartRequest)
      .withDestinationBucketName(bucketName)
      .withDestinationKey(destinationKey)
      .withSourceBucketName(bucketName)
      .withSourceKey(part.s3UploadKey)
      .withUploadId(destinationUploadId)
      .withPartNumber(part.number)
  }

  private def etags(responses: Seq[CopyPartResult]): java.util.List[PartETag] = {
    val xs = responses map { response ⇒
      new PartETag(response.getPartNumber(), response.getETag())
    }

    mutable.Seq(xs: _*).asJava
  }

  // FIXME: #perf pinned dispatcher

  private def createTempDir(): Future[File] = {
    Future {
      com.google.common.io.Files.createTempDir()
    }
  }

  // FIXME: #perf use nio and pinned dispatcher

  private def concatFiles(dir: File, fileNames: Seq[String]): Future[File] = {
    Future {
      val dirPath = dir.toPath
      val concatFile = dirPath.resolve("concatenated").toFile

      val outStream = new FileOutputStream(concatFile)

      fileNames foreach { fileName ⇒
        Files.copy(dirPath.resolve(fileName), outStream)
      }

      outStream.close()

      concatFile
    }
  }

  private def deleteDir(dir: File): Future[Unit] = {
    Future {
      FileUtils.deleteDirectory(dir)
    }
  }
}

{'fraction_non_alphanumeric': 0.1001361470388019, 'fraction_numerical': 0.0029952348536419334, 'mean_word_length': 2.90718085106383, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6444524', 'n_tokens_mistral': 4312, 'n_tokens_neox': 4257, 'n_words': 781}