<commit_before>/**
 * Copyright (C) 2015 DANS - Data Archiving and Networked Services (<EMAIL>)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package nl.knaw.dans.easy.sword2

import java.io.{ File, IOException }
import java.net.{ MalformedURLException, URL, UnknownHostException }
import java.nio.charset.StandardCharsets
import java.nio.file._
import java.nio.file.attribute.{ BasicFileAttributes, PosixFilePermissions }
import java.util.regex.Pattern
import java.util.{ Collections, NoSuchElementException }

import gov.loc.repository.bagit.FetchTxt.FilenameSizeUrl
import gov.loc.repository.bagit.transformer.impl.TagManifestCompleter
import gov.loc.repository.bagit.utilities.SimpleResult
import gov.loc.repository.bagit.verify.CompleteVerifier
import gov.loc.repository.bagit.writer.impl.FileSystemWriter
import gov.loc.repository.bagit.{ Bag, BagFactory, FetchTxt }
import net.lingala.zip4j.core.ZipFile
import net.lingala.zip4j.model.FileHeader
import nl.knaw.dans.easy.sword2.State._
import nl.knaw.dans.lib.error.{ CompositeException, TraversableTryExtensions }
import org.apache.abdera.i18n.iri.IRI
import org.apache.commons.codec.digest.DigestUtils
import org.apache.commons.io.FileUtils._
import org.joda.time.{ DateTime, DateTimeZone }
import org.slf4j.{ Logger, LoggerFactory }
import org.swordapp.server.{ Deposit, DepositReceipt, SwordError }
import resource.Using
import rx.lang.scala.schedulers.NewThreadScheduler
import rx.lang.scala.subjects.PublishSubject

import scala.collection.JavaConverters._
import scala.collection.convert.Wrappers.JListWrapper
import scala.util.control.NonFatal
import scala.util.{ Failure, Success, Try }

object DepositHandler {
  val log: Logger = LoggerFactory.getLogger(getClass)
  implicit val bagFactory = new BagFactory

  private val depositProcessingStream = PublishSubject[(String, Deposit)]()

  def startDepositProcessingStream(settings: Settings): Unit = {
    depositProcessingStream
      .onBackpressureBuffer
      .observeOn(NewThreadScheduler())
      .doOnEach(_ match { case (id, deposit) => finalizeDeposit(deposit.getMimeType)(settings, id) })
      .subscribe(_ match { case (id, deposit) => log.info(s"[$id] Done finalizing deposit") })
  }

  def handleDeposit(deposit: Deposit)(implicit settings: Settings, id: String): Try[DepositReceipt] = {
    val payload = Paths.get(settings.tempDir.toString, id, deposit.getFilename.split("/").last).toFile
    for {
      _ <- copyPayloadToFile(deposit, payload)
      _ <- doesHashMatch(payload, deposit.getMd5)
      _ <- handleDepositAsync(deposit)
      dr = createDepositReceipt(settings, id)
      _ = dr.setVerboseDescription("received successfully: " + deposit.getFilename + "; MD5: " + deposit.getMd5)
    } yield dr
  }

  def genericErrorMessage(implicit settings: Settings, id: String): String = {
    val mailaddress = settings.supportMailAddress
    val timestamp = DateTime.now(DateTimeZone.UTC).toString

    s"""The server encountered an unexpected condition.
       |Please contact the SWORD service administrator at $mailaddress.
       |The error occurred at $timestamp. Your 'DepositID' is $id.
    """.stripMargin
  }

  def finalizeDeposit(mimeType: String)(implicit settings: Settings, id: String): Try[Unit] = {
    log.info(s"[$id] Finalizing deposit")
    implicit val bagStoreSettings: Option[BagStoreSettings] = settings.bagStoreSettings
    val depositDir = new File(settings.tempDir, id)

    val result = for {
      _ <- extractBag(mimeType)
      bagDir <- getBagDir(depositDir)
      _ <- checkFetchItemUrls(bagDir, settings.urlPattern)
      _ <- checkBagVirtualValidity(bagDir)
      props <- DepositProperties(id)
      _ <- props.setState(SUBMITTED, "Deposit is valid and ready for post-submission processing")
      _ <- props.save()
      _ <- SampleTestData.sampleData(id, depositDir, props)(settings.sample)
      _ <- removeZipFiles(depositDir)
      dataDir <- moveBagToStorage()
    } yield ()

    result.recover {
      case InvalidDepositException(_, msg, cause) =>
        log.error(s"[$id] Invalid deposit", cause)
        for {
          props <- DepositProperties(id)
          _ <- props.setState(INVALID, msg)
          _ <- props.save()
          // we don't sample in this case, given that the deposit is invalid and we cannot automate
          // replacing sensitive data
        } yield ()
      case RejectedDepositException(_, msg, cause) =>
        log.error(s"[$id] Rejected deposit", cause)
        for {
          // Currently, the only reason for SWORD 2 to reject a deposit, is insufficient disk space,
          // so we clean up, here.
          _ <- removeZipFiles(depositDir)
          props <- DepositProperties(id)
          _ <- props.setState(REJECTED, msg)
          _ <- props.save()
          // we don't sample in this case, given that this state can only occur when there is
          // insufficient disk space
        } yield ()
      case NonFatal(e) =>
        log.error(s"[$id] Internal failure in deposit service", e)
        for {
          props <- DepositProperties(id)
          _ <- props.setState(FAILED, genericErrorMessage)
          _ <- props.save()
        } yield ()
    }
  }

  private def removeZipFiles(bagDir: File): Try[Unit] = Try {
    log.debug("Removing zip files")
    for (file <- bagDir.listFiles().toList
         if isPartOfDeposit(file)
         if file.isFile) {
      log.debug(s"Removing $file")
      deleteQuietly(file)
    }
  }

  private def extractBag(mimeType: String)(implicit settings: Settings, id: String): Try[File] = {
    def checkAvailableDiskspace(file: File): Try[Unit] = {
      val zipFile = new ZipFile(file.getPath)
      val headers = zipFile.getFileHeaders.asScala.asInstanceOf[JListWrapper[FileHeader]] // Look out! Not sure how robust this cast is!
      val uncompressedSize = headers.map(_.getUncompressedSize).sum
      val availableDiskSize = Files.getFileStore(file.toPath).getUsableSpace
      log.debug(s"Available (usable) disk space currently $availableDiskSize bytes. Spaces needed: $uncompressedSize bytes. Margin required: ${ settings.marginDiskSpace } bytes.")
      if (uncompressedSize + settings.marginDiskSpace > availableDiskSize)
        Failure(RejectedDepositException(id, "Not enough disk space to process deposit.",
          new IllegalStateException(s"Required disk space for unzipping: ${ uncompressedSize + settings.marginDiskSpace } (including ${ settings.marginDiskSpace } margin). Available: $availableDiskSize")))
      else Success(())
    }<commit_msg>cleanup implementation<commit_after>/**
 * Copyright (C) 2015 DANS - Data Archiving and Networked Services (<EMAIL>)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package nl.knaw.dans.easy.sword2

import java.io.{ File, IOException }
import java.net.{ MalformedURLException, URL, UnknownHostException }
import java.nio.charset.StandardCharsets
import java.nio.file._
import java.nio.file.attribute.{ BasicFileAttributes, PosixFilePermissions }
import java.util.regex.Pattern
import java.util.{ Collections, NoSuchElementException }

import gov.loc.repository.bagit.FetchTxt.FilenameSizeUrl
import gov.loc.repository.bagit.transformer.impl.TagManifestCompleter
import gov.loc.repository.bagit.utilities.SimpleResult
import gov.loc.repository.bagit.verify.CompleteVerifier
import gov.loc.repository.bagit.writer.impl.FileSystemWriter
import gov.loc.repository.bagit.{ Bag, BagFactory, FetchTxt }
import net.lingala.zip4j.core.ZipFile
import net.lingala.zip4j.model.FileHeader
import nl.knaw.dans.easy.sword2.State._
import nl.knaw.dans.lib.error.{ CompositeException, TraversableTryExtensions }
import org.apache.abdera.i18n.iri.IRI
import org.apache.commons.codec.digest.DigestUtils
import org.apache.commons.io.FileUtils._
import org.joda.time.{ DateTime, DateTimeZone }
import org.slf4j.{ Logger, LoggerFactory }
import org.swordapp.server.{ Deposit, DepositReceipt, SwordError }
import resource.Using
import rx.lang.scala.schedulers.NewThreadScheduler
import rx.lang.scala.subjects.PublishSubject

import scala.collection.JavaConverters._
import scala.collection.convert.Wrappers.JListWrapper
import scala.util.control.NonFatal
import scala.util.{ Failure, Success, Try }

object DepositHandler {
  val log: Logger = LoggerFactory.getLogger(getClass)
  implicit val bagFactory = new BagFactory

  private val depositProcessingStream = PublishSubject[(String, Deposit)]()

  def startDepositProcessingStream(settings: Settings): Unit = {
    depositProcessingStream
      .onBackpressureBuffer
      .observeOn(NewThreadScheduler())
      .doOnEach(_ match { case (id, deposit) => finalizeDeposit(deposit.getMimeType)(settings, id) })
      .subscribe(_ match { case (id, deposit) => log.info(s"[$id] Done finalizing deposit") })
  }

  def handleDeposit(deposit: Deposit)(implicit settings: Settings, id: String): Try[DepositReceipt] = {
    val payload = Paths.get(settings.tempDir.toString, id, deposit.getFilename.split("/").last).toFile
    for {
      _ <- copyPayloadToFile(deposit, payload)
      _ <- doesHashMatch(payload, deposit.getMd5)
      _ <- handleDepositAsync(deposit)
      dr = createDepositReceipt(settings, id)
      _ = dr.setVerboseDescription("received successfully: " + deposit.getFilename + "; MD5: " + deposit.getMd5)
    } yield dr
  }

  def genericErrorMessage(implicit settings: Settings, id: String): String = {
    val mailaddress = settings.supportMailAddress
    val timestamp = DateTime.now(DateTimeZone.UTC).toString

    s"""The server encountered an unexpected condition.
       |Please contact the SWORD service administrator at $mailaddress.
       |The error occurred at $timestamp. Your 'DepositID' is $id.
    """.stripMargin
  }

  def finalizeDeposit(mimeType: String)(implicit settings: Settings, id: String): Try[Unit] = {
    log.info(s"[$id] Finalizing deposit")
    implicit val bagStoreSettings: Option[BagStoreSettings] = settings.bagStoreSettings
    val depositDir = new File(settings.tempDir, id)

    val result = for {
      _ <- extractBag(mimeType)
      bagDir <- getBagDir(depositDir)
      _ <- checkFetchItemUrls(bagDir, settings.urlPattern)
      _ <- checkBagVirtualValidity(bagDir)
      props <- DepositProperties(id)
      _ <- props.setState(SUBMITTED, "Deposit is valid and ready for post-submission processing")
      _ <- props.save()
      _ <- SampleTestData.sampleData(id, depositDir, props)(settings.sample)
      _ <- removeZipFiles(depositDir)
      _ <- moveBagToStorage()
    } yield ()

    result.recover {
      case InvalidDepositException(_, msg, cause) =>
        log.error(s"[$id] Invalid deposit", cause)
        for {
          props <- DepositProperties(id)
          _ <- props.setState(INVALID, msg)
          _ <- props.save()
          // we don't sample in this case, given that the deposit is invalid and we cannot automate
          // replacing sensitive data
          _ <- cleanupFiles(depositDir, INVALID)
        } yield ()
      case RejectedDepositException(_, msg, cause) =>
        log.error(s"[$id] Rejected deposit", cause)
        for {
          // Currently, the only reason for SWORD 2 to reject a deposit, is insufficient disk space,
          // so we clean up, here.
          props <- DepositProperties(id)
          _ <- props.setState(REJECTED, msg)
          _ <- props.save()
          // we don't sample in this case, given that this state can only occur when there is
          // insufficient disk space
          _ <- cleanupFiles(depositDir, REJECTED)
        } yield ()
      case NonFatal(e) =>
        log.error(s"[$id] Internal failure in deposit service", e)
        for {
          props <- DepositProperties(id)
          _ <- props.setState(FAILED, genericErrorMessage)
          _ <- props.save()
          _ <- cleanupFiles(depositDir, FAILED)
        } yield ()
    }
  }

  private def cleanupFiles(depositDir: File, state: State)(implicit settings: Settings): Try[Unit] = {
    if (settings.cleanup.getOrElse(state, false)) {
      log.debug(s"cleaning up zip files and bag directory for deposit due to state $state")
      for {
        _ <- removeZipFiles(depositDir)
        bagDir <- getBagDir(depositDir)
        _ <- Try {
          if (bagDir.exists()) {
            log.debug(s"removing bag $bagDir")
            deleteQuietly(bagDir)
          }
        }
      } yield ()
    }
    else
      Success(())
  }

  private def removeZipFiles(bagDir: File): Try[Unit] = Try {
    log.debug("Removing zip files")
    for (file <- bagDir.listFiles().toList
         if isPartOfDeposit(file)
         if file.isFile) {
      log.debug(s"Removing $file")
      deleteQuietly(file)
    }
  }

  private def extractBag(mimeType: String)(implicit settings: Settings, id: String): Try[File] = {
    def checkAvailableDiskspace(file: File): Try[Unit] = {
      val zipFile = new ZipFile(file.getPath)
      val headers = zipFile.getFileHeaders.asScala.asInstanceOf[JListWrapper[FileHeader]] // Look out! Not sure how robust this cast is!
      val uncompressedSize = headers.map(_.getUncompressedSize).sum
      val availableDiskSize = Files.getFileStore(file.toPath).getUsableSpace
      log.debug(s"Available (usable) disk space currently $availableDiskSize bytes. Spaces needed: $uncompressedSize bytes. Margin required: ${ settings.marginDiskSpace } bytes.")
      if (uncompressedSize + settings.marginDiskSpace > availableDiskSize)
        Failure(RejectedDepositException(id, "Not enough disk space to process deposit.",
          new IllegalStateException(s"Required disk space for unzipping: ${ uncompressedSize + settings.marginDiskSpace } (including ${ settings.marginDiskSpace } margin). Available: $availableDiskSize")))
      else Success(())
    }
{'fraction_non_alphanumeric': 0.10352765928634301, 'fraction_numerical': 0.0025729568691177465, 'mean_word_length': 3.5140586797066016, 'pattern_counts': {'":': 0, '<': 58, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1888229', 'n_tokens_mistral': 4433, 'n_tokens_neox': 4235, 'n_words': 1108}