<commit_before>/**
  * Copyright 2015 <NAME>
  *
  * 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 controllers

import actions.RequestMonitor
import akka.util.ByteString
import cmwell.domain._
import cmwell.tracking._
import cmwell.util.concurrent._
import cmwell.util.collections.opfut
import cmwell.util.formats.JsonEncoder
import cmwell.web.ld.cmw.CMWellRDFHelper
import cmwell.web.ld.exceptions.UnretrievableIdentifierException
import cmwell.web.ld.util.LDFormatParser.ParsingResponse
import cmwell.web.ld.util._
import cmwell.ws.{AggregateBothOldAndNewTypesCaches, Settings}
import cmwell.ws.util.{FieldKeyParser, TypeHelpers}
import com.typesafe.scalalogging.LazyLogging
import logic.{CRUDServiceFS, InfotonValidator}
import play.api.libs.json._
import play.api.mvc._
import security.{AuthUtils, PermissionLevel}
import wsutil._
import javax.inject._

import filters.Attrs

import scala.concurrent.duration._
import scala.concurrent._
import scala.language.postfixOps
import scala.util._

@Singleton
class InputHandler @Inject() (ingestPushback: IngestPushback,
                              crudService: CRUDServiceFS,
                              tbg: NbgToggler,
                              authUtils: AuthUtils,
                              cmwellRDFHelper: CMWellRDFHelper,
                              formatterManager: FormatterManager) extends InjectedController with LazyLogging with TypeHelpers { self =>

  val aggregateBothOldAndNewTypesCaches = new AggregateBothOldAndNewTypesCaches(crudService,tbg)
  val bo1 = collection.breakOut[List[Infoton],String,Set[String]]
  val bo2 = collection.breakOut[Vector[Infoton],String,Set[String]]

  /**
   *
   * @param format
   * @return
   */
  def handlePost(format: String = "") = ingestPushback.async(parse.raw) { implicit req =>
    import scala.concurrent.ExecutionContext.Implicits.global
    RequestMonitor.add("in", req.path, req.rawQueryString, req.body.asBytes().fold("")(_.utf8String),req.attrs(Attrs.RequestReceivedTimestamp))
    // first checking "priority" query string. Only if it is present we will consult the UserInfoton which is more expensive (order of && below matters):
    if (req.getQueryString("priority").isDefined && !authUtils.isOperationAllowedForUser(security.PriorityWrite, authUtils.extractTokenFrom(req), req.attrs(Attrs.Nbg), evenForNonProdEnv = true)) {
      Future.successful(Forbidden(Json.obj("success" -> false, "message" -> "User not authorized for priority write")))
    } else {
      val resp = if ("jsonw" == format.toLowerCase) handlePostWrapped(req) -> Future.successful(Seq.empty[(String, String)]) else handlePostRDF(req)
      resp._2.flatMap { headers =>
        keepAliveByDrippingNewlines(resp._1, headers)
      }.recover(errorHandler)
    }
  }

  /**
   *
   * @param path
   * @return
   */
  private def escapePath(path : String) : String = {
    path.replace(" ", "%20")
  }

  def handlePostForDCOverwrites =  ingestPushback.async(parse.raw) { implicit req =>
    import scala.concurrent.ExecutionContext.Implicits.global

    val tokenOpt = authUtils.extractTokenFrom(req)
    if (!authUtils.isOperationAllowedForUser(security.Overwrite, tokenOpt, req.attrs(Attrs.Nbg), evenForNonProdEnv = true))
      Future.successful(Forbidden("not authorized"))
    else {
      Try {
        parseRDF(req, false, true).flatMap {
          case ParsingResponse(infotonsMap, metaDataMap, cmwHostsSet, tmpDeleteMap, deleteValsMap, deletePaths, atomicUpdates) => {

            require(tmpDeleteMap.isEmpty && deleteValsMap.isEmpty && deletePaths.isEmpty && atomicUpdates.isEmpty, "can't use meta operations here! this API is used internaly, and only for overwrites!")
            require(metaDataMap.forall {
              case (path, MetaData(mdType, date, data, text, mimeType, linkType, linkTo, dataCenter, indexTime)) => {
                indexTime.isDefined &&
                  dataCenter.isDefined &&
                  dataCenter.get != Settings.dataCenter &&
                  mdType.isDefined &&
                  ((mdType.get == LinkMetaData && linkType.isDefined && linkTo.isDefined) ||
                    (mdType.get == FileMetaData && mimeType.isDefined && (data.isDefined || text.isDefined)) ||
                    (mdType.get == ObjectMetaData) ||
                    (mdType.get == DeletedMetaData))
              }
            }, "in overwrites API all meta data must be present! no implicit inference is allowed. (every infoton must have all relevant system fields added)")
            enforceForceIfNeededAndReturnMetaFieldsInfotons(infotonsMap, true).flatMap { metaFields =>
              val infotonsWithoutFields = metaDataMap.keySet.filterNot(infotonsMap.keySet.apply) //meaning FileInfotons without extra data...

              val allInfotons = (infotonsMap.toVector map {
                case (path, fields) => {
                  require(path.nonEmpty, "path cannot be empty!")
                  val escapedPath = escapePath(path)
                  InfotonValidator.validateValueSize(fields)
                  val fs = fields.map {
                    case (fk, vs) => fk.internalKey -> vs
                  }
                  infotonFromMaps(cmwHostsSet, escapedPath, Some(fs), metaDataMap.get(escapedPath))
                }
              }) ++ infotonsWithoutFields.map(p => infotonFromMaps(cmwHostsSet, p, None, metaDataMap.get(p))) ++ metaFields

              //logger.info(s"infotonsToPut: ${allInfotons.collect { case o: ObjectInfoton => o.toString }.mkString("[", ",", "]")}")

              val (metaInfotons, infotonsToPut) = allInfotons.partition(_.path.startsWith("/meta/"))

              val f = crudService.putInfotons(metaInfotons)
              crudService.putOverwrites(infotonsToPut).flatMap { b =>
                f.map {
                  case true if b => Ok(Json.obj("success" -> true))
                  case _ => BadRequest(Json.obj("success" -> false))
                }
              }.recover {
                case err: Throwable => {
                  logger.error(s"bad data received: ${err.getMessage}: ${req.body.asBytes().fold("NOTHING")(_.utf8String)}")
                  wsutil.exceptionToResponse(err)
                }<commit_msg>cmwell #412 - better error reporting for failed _ow ingests<commit_after>/**
  * Copyright 2015 <NAME>
  *
  * 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 controllers

import actions.RequestMonitor
import akka.util.ByteString
import cmwell.domain._
import cmwell.tracking._
import cmwell.util.concurrent._
import cmwell.util.collections.opfut
import cmwell.util.formats.JsonEncoder
import cmwell.web.ld.cmw.CMWellRDFHelper
import cmwell.web.ld.exceptions.UnretrievableIdentifierException
import cmwell.web.ld.util.LDFormatParser.ParsingResponse
import cmwell.web.ld.util._
import cmwell.ws.{AggregateBothOldAndNewTypesCaches, Settings}
import cmwell.ws.util.{FieldKeyParser, TypeHelpers}
import com.typesafe.scalalogging.LazyLogging
import logic.{CRUDServiceFS, InfotonValidator}
import play.api.libs.json._
import play.api.mvc._
import security.{AuthUtils, PermissionLevel}
import wsutil._
import javax.inject._

import filters.Attrs

import scala.concurrent.duration._
import scala.concurrent._
import scala.language.postfixOps
import scala.util._

@Singleton
class InputHandler @Inject() (ingestPushback: IngestPushback,
                              crudService: CRUDServiceFS,
                              tbg: NbgToggler,
                              authUtils: AuthUtils,
                              cmwellRDFHelper: CMWellRDFHelper,
                              formatterManager: FormatterManager) extends InjectedController with LazyLogging with TypeHelpers { self =>

  val aggregateBothOldAndNewTypesCaches = new AggregateBothOldAndNewTypesCaches(crudService,tbg)
  val bo1 = collection.breakOut[List[Infoton],String,Set[String]]
  val bo2 = collection.breakOut[Vector[Infoton],String,Set[String]]

  /**
   *
   * @param format
   * @return
   */
  def handlePost(format: String = "") = ingestPushback.async(parse.raw) { implicit req =>
    import scala.concurrent.ExecutionContext.Implicits.global
    RequestMonitor.add("in", req.path, req.rawQueryString, req.body.asBytes().fold("")(_.utf8String),req.attrs(Attrs.RequestReceivedTimestamp))
    // first checking "priority" query string. Only if it is present we will consult the UserInfoton which is more expensive (order of && below matters):
    if (req.getQueryString("priority").isDefined && !authUtils.isOperationAllowedForUser(security.PriorityWrite, authUtils.extractTokenFrom(req), req.attrs(Attrs.Nbg), evenForNonProdEnv = true)) {
      Future.successful(Forbidden(Json.obj("success" -> false, "message" -> "User not authorized for priority write")))
    } else {
      val resp = if ("jsonw" == format.toLowerCase) handlePostWrapped(req) -> Future.successful(Seq.empty[(String, String)]) else handlePostRDF(req)
      resp._2.flatMap { headers =>
        keepAliveByDrippingNewlines(resp._1, headers)
      }.recover(errorHandler)
    }
  }

  /**
   *
   * @param path
   * @return
   */
  private def escapePath(path : String) : String = {
    path.replace(" ", "%20")
  }

  def handlePostForDCOverwrites =  ingestPushback.async(parse.raw) { implicit req =>
    import scala.concurrent.ExecutionContext.Implicits.global

    val tokenOpt = authUtils.extractTokenFrom(req)
    if (!authUtils.isOperationAllowedForUser(security.Overwrite, tokenOpt, req.attrs(Attrs.Nbg), evenForNonProdEnv = true))
      Future.successful(Forbidden("not authorized"))
    else {
      Try {
        parseRDF(req, false, true).flatMap {
          case ParsingResponse(infotonsMap, metaDataMap, cmwHostsSet, tmpDeleteMap, deleteValsMap, deletePaths, atomicUpdates) => {

            require(tmpDeleteMap.isEmpty && deleteValsMap.isEmpty && deletePaths.isEmpty && atomicUpdates.isEmpty, "can't use meta operations here! this API is used internaly, and only for overwrites!")
            val (errs,_) = cmwell.util.collections.partitionWith(metaDataMap){
              case (path, MetaData(mdType, date, data, text, mimeType, linkType, linkTo, dataCenter, indexTime)) =>
                var errors = List.empty[String]
                if(indexTime.isEmpty) {
                  errors = "indexTime should be defined" :: errors
                }
                if(dataCenter.isEmpty) {
                  errors = "dataCenter should be defined" :: errors
                }
                else if(dataCenter.get == Settings.dataCenter) {
                  errors = "dataCenter cannot be equal to current ID" :: errors
                }
                if(mdType.isEmpty) {
                  errors = "infoton's kind (type) must be defined" :: errors
                }
                else if(mdType.get == LinkMetaData){
                  if(linkType.isEmpty) {
                    errors = "link kind (type) must be defined" :: errors
                  }
                  if(linkTo.isEmpty) {
                    errors = "link destination (to) must be defined" :: errors
                  }
                }
                else if(mdType.get == FileMetaData){
                  if(mimeType.isEmpty) {
                    errors = "file's media type (mimeType) must be defined" :: errors
                  }
                  if(data.isEmpty && text.isEmpty) {
                    errors = "file's content must be defined" :: errors
                  }
                }
                else if(mdType.get != ObjectMetaData && mdType.get != DeletedMetaData){
                  errors = s"infoton's kind (type) isn't recognized" :: errors
                }

                if(errors.isEmpty) Right(())
                else Left(errors.mkString(s"path [$path] failed due to:\n\t","\n\t",""))
            }
            require(errs.isEmpty,errs.mkString("overwrites API failed to validate the request.\n\n","\n",""))

            enforceForceIfNeededAndReturnMetaFieldsInfotons(infotonsMap, true).flatMap { metaFields =>
              val infotonsWithoutFields = metaDataMap.keySet.filterNot(infotonsMap.keySet.apply) //meaning FileInfotons without extra data...

              val allInfotons = (infotonsMap.toVector map {
                case (path, fields) => {
                  require(path.nonEmpty, "path cannot be empty!")
                  val escapedPath = escapePath(path)
                  InfotonValidator.validateValueSize(fields)
                  val fs = fields.map {
                    case (fk, vs) => fk.internalKey -> vs
                  }
                  infotonFromMaps(cmwHostsSet, escapedPath, Some(fs), metaDataMap.get(escapedPath))
                }
              }) ++ infotonsWithoutFields.map(p => infotonFromMaps(cmwHostsSet, p, None, metaDataMap.get(p))) ++ metaFields

              //logger.info(s"infotonsToPut: ${allInfotons.collect { case o: ObjectInfoton => o.toString }.mkString("[", ",", "]")}")

              val (metaInfotons, infotonsToPut) = allInfotons.partition(_.path.startsWith("/meta/"))

              val f = crudService.putInfotons(metaInfotons)
              crudService.putOverwrites(infotonsToPut).flatMap { b =>
                f.map {
                  case true if b => Ok(Json.obj("success" -> true))
                  case _ => BadRequest(Json.obj("success" -> false))
                }
              }.recover {
                case err: Throwable => {
                  logger.error(s"bad data received: ${err.getMessage}: ${req.body.asBytes().fold("NOTHING")(_.utf8String)}")
                  wsutil.exceptionToResponse(err)
                }
{'fraction_non_alphanumeric': 0.10550051247010592, 'fraction_numerical': 0.0023915271609156134, 'mean_word_length': 2.6299603174603177, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 47, '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': '5521254', 'n_tokens_mistral': 4194, 'n_tokens_neox': 4083, 'n_words': 948}