<commit_before>/*
* This file is part of the sohva project.
*
* 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 gnieh.sohva
package async

import strategy.Strategy

import resource._

import java.io.{
  File,
  InputStream,
  FileOutputStream,
  ByteArrayInputStream,
  BufferedInputStream
}

import net.liftweb.json._

import gnieh.diffson.JsonPatch

import scala.concurrent.Future

import spray.http._
import spray.client.pipelining._
import spray.httpx.marshalling._

import akka.actor._

/** Gives the user access to the different operations available on a database.
 *  Among other operations this is the key class to get access to the documents
 *  of this database.
 *
 *  It also exposes the change handler interface, that allows people to react to change notifications. This
 *  is a low-level API, that handles raw Json objects
 *
 *  @param credit The credit assigned to the conflict resolver. It represents the number of times the client tries to save the document before giving up.
 *  @param strategy The strategy being used to resolve conflicts
 *
 *  @author <NAME>
 */
class Database private[sohva] (
  val name: String,
  val couch: CouchDB,
  val credit: Int,
  val strategy: Strategy)
    extends gnieh.sohva.Database[Future]
    with LiftMarshalling {

  import couch.serializer

  implicit def ec =
    couch.ec

  implicit def formats =
    couch.formats

  /* the resolver is responsible for applying the merging strategy on conflict and retrying
   * to save the document after resolution process */
  private def resolver(credit: Int, docId: String, baseRev: Option[String], current: JValue): Future[JValue] =
    couch.http(Put(uri / docId, current)).recoverWith {
      case CouchException(409, _) if credit > 0 =>
        // try to resolve the conflict and save again
        for {
          // get the base document if any
          base <- getRawDocById(docId, baseRev)
          // get the last document
          last <- getRawDocById(docId)
          // apply the merge strategy between base, last and current revision of the document
          lastRev = last map (d => (d \ "_rev").toString)
          resolved = strategy(base, last, current)
          res <- resolver(credit - 1, docId, lastRev, resolved)
        } yield res
    }

  def info: Future[Option[InfoResult]] =
    for (info <- couch.optHttp(Get(uri)))
      yield info.map(infoResult)

  def exists: Future[Boolean] =
    for (h <- couch.optHttp(Head(uri)))
      yield h.isDefined

  def changes(since: Option[Int] = None, filter: Option[String] = None): ChangeStream =
    new ChangeStream(this, since, filter)

  def create: Future[Boolean] =
    for {
      exist <- exists
      ok <- create(exist)
    } yield ok

  private[this] def create(exist: Boolean) =
    if (exist) {
      Future.successful(false)
    } else {
      for (result <- couch.http(Put(uri)))
        yield couch.ok(result)
    }

  def delete: Future[Boolean] =
    for {
      exist <- exists
      ok <- delete(exist)
    } yield ok

  private[this] def delete(exist: Boolean) =
    if (exist) {
      for (result <- couch.http(Delete(uri)))
        yield couch.ok(result)
    } else {
      Future.successful(false)
    }

  def _all_docs(key: Option[String] = None,
    keys: List[String] = Nil,
    startkey: Option[String] = None,
    startkey_docid: Option[String] = None,
    endkey: Option[String] = None,
    endkey_docid: Option[String] = None,
    limit: Int = -1,
    stale: Option[String] = None,
    descending: Boolean = false,
    skip: Int = 0,
    inclusive_end: Boolean = true): Future[List[String]] =
    for {
      res <- builtInView[String, Map[String, String], Any]("_all_docs").query(
        key = key,
        keys = keys,
        startkey = startkey,
        startkey_docid = startkey_docid,
        endkey = endkey,
        endkey_docid = endkey_docid,
        limit = limit,
        stale = stale,
        descending = descending,
        skip = skip,
        inclusive_end = inclusive_end
      )
    } yield for (Row(Some(id), _, _, _) <- res.rows) yield id

  def getDocById[T: Manifest](id: String, revision: Option[String] = None): Future[Option[T]] =
    for (raw <- getRawDocById(id, revision))
      yield raw.map(docResult[T])

  def getDocsById[T: Manifest](ids: List[String]): Future[List[T]] =
    for {
      res <- builtInView[String, Map[String, String], T]("_all_docs").query(keys = ids, include_docs = true)
    } yield res.rows.flatMap { case Row(_, _, _, doc) => doc }

  def getRawDocById(id: String, revision: Option[String] = None): Future[Option[JValue]] =
    couch.optHttp(Get(uri / id <<? revision.map("rev" -> _)))

  def getDocRevision(id: String): Future[Option[String]] =
    couch.pipeline(couch.prepare((Head(uri / id)))).flatMap(extractRev _)

  def getDocRevisions(ids: List[String]): Future[List[(String, String)]] =
    for {
      res <- builtInView[String, Map[String, String], Any]("_all_docs").query(keys = ids)
    } yield res.rows.map { case Row(Some(id), _, value, _) => (id, value("rev")) }

  def saveDoc[T <% IdRev: Manifest](doc: T): Future[Option[T]] =
    for {
      upd <- resolver(credit, doc._id, doc._rev, serializer.toJson(doc))
      res <- update(upd.extract[DocUpdate])
    } yield res

  def saveRawDoc(doc: JValue): Future[Option[JValue]] = serializer.fromCouchJson(doc) match {
    case Some((id, rev)) =>
      for {
        upd <- resolver(credit, id, rev, doc)
        res <- updateRaw(docUpdateResult(upd))
      } yield res<commit_msg>First attempt at error handling.<commit_after>/*
* This file is part of the sohva project.
*
* 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 gnieh.sohva
package async

import strategy.Strategy

import resource._

import java.io.{
  File,
  InputStream,
  FileOutputStream,
  ByteArrayInputStream,
  BufferedInputStream
}

import net.liftweb.json._

import gnieh.diffson.JsonPatch

import scala.concurrent.Future

import spray.http._
import spray.client.pipelining._
import spray.httpx.marshalling._

import akka.actor._

/** Gives the user access to the different operations available on a database.
 *  Among other operations this is the key class to get access to the documents
 *  of this database.
 *
 *  It also exposes the change handler interface, that allows people to react to change notifications. This
 *  is a low-level API, that handles raw Json objects
 *
 *  @param credit The credit assigned to the conflict resolver. It represents the number of times the client tries to save the document before giving up.
 *  @param strategy The strategy being used to resolve conflicts
 *
 *  @author <NAME>
 */
class Database private[sohva] (
  val name: String,
  val couch: CouchDB,
  val credit: Int,
  val strategy: Strategy)
    extends gnieh.sohva.Database[Future]
    with LiftMarshalling {

  import couch.serializer

  implicit def ec =
    couch.ec

  implicit def formats =
    couch.formats

  /* the resolver is responsible for applying the merging strategy on conflict and retrying
   * to save the document after resolution process */
  private def resolver(credit: Int, docId: String, baseRev: Option[String], current: JValue): Future[JValue] =
    couch.http(Put(uri / docId, current)).recoverWith {
      case CouchException(409, _) if credit > 0 =>
        // try to resolve the conflict and save again
        for {
          // get the base document if any
          base <- getRawDocById(docId, baseRev)
          // get the last document
          last <- getRawDocById(docId)
          // apply the merge strategy between base, last and current revision of the document
          lastRev = last map (d => (d \ "_rev").toString)
          resolved = strategy(base, last, current)
          res <- resolver(credit - 1, docId, lastRev, resolved)
        } yield res
    }

  def info: Future[Option[InfoResult]] =
    for (info <- couch.optHttp(Get(uri)))
      yield info.map(infoResult)

  def exists: Future[Boolean] =
    for (h <- couch.optHttp(Head(uri)))
      yield h.isDefined

  def changes(since: Option[Int] = None, filter: Option[String] = None): ChangeStream =
    new ChangeStream(this, since, filter)

  def create: Future[Boolean] =
    for {
      exist <- exists
      ok <- create(exist)
    } yield ok

  private[this] def create(exist: Boolean) =
    if (exist) {
      Future.successful(false)
    } else {
      for (result <- couch.http(Put(uri)))
        yield couch.ok(result)
    }

  def delete: Future[Boolean] =
    for {
      exist <- exists
      ok <- delete(exist)
    } yield ok

  private[this] def delete(exist: Boolean) =
    if (exist) {
      for (result <- couch.http(Delete(uri)))
        yield couch.ok(result)
    } else {
      Future.successful(false)
    }

  def _all_docs(key: Option[String] = None,
    keys: List[String] = Nil,
    startkey: Option[String] = None,
    startkey_docid: Option[String] = None,
    endkey: Option[String] = None,
    endkey_docid: Option[String] = None,
    limit: Int = -1,
    stale: Option[String] = None,
    descending: Boolean = false,
    skip: Int = 0,
    inclusive_end: Boolean = true): Future[List[String]] =
    for {
      res <- builtInView[String, Map[String, String], Any]("_all_docs").query(
        key = key,
        keys = keys,
        startkey = startkey,
        startkey_docid = startkey_docid,
        endkey = endkey,
        endkey_docid = endkey_docid,
        limit = limit,
        stale = stale,
        descending = descending,
        skip = skip,
        inclusive_end = inclusive_end
      )
    } yield for (Row(Some(id), _, _, _) <- res.rows) yield id

  def getDocById[T: Manifest](id: String, revision: Option[String] = None): Future[Option[T]] =
    for (raw <- getRawDocById(id, revision))
      yield raw.map(docResult[T])

  def getDocsById[T: Manifest](ids: List[String]): Future[List[T]] =
    for {
      res <- builtInView[String, Map[String, String], T]("_all_docs").query(keys = ids, include_docs = true)
    } yield res.rows.flatMap { case Row(_, _, _, doc) => doc }

  def getRawDocById(id: String, revision: Option[String] = None): Future[Option[JValue]] =
    couch.optHttp(Get(uri / id <<? revision.map("rev" -> _)))

  def getDocRevision(id: String): Future[Option[String]] =
    couch.pipeline(couch.prepare((Head(uri / id)))).flatMap(extractRev _)

  def getDocRevisions(ids: List[String]): Future[List[(String, String)]] =
    for {
      res <- builtInView[String, Map[String, String], Any]("_all_docs").query(keys = ids)
    } yield res.rows.map { case Row(Some(id), _, value, _) => (id, value("rev")) }

  def saveDoc[T <% IdRev: Manifest](doc: T): Future[Option[T]] =
    for {
      upd <- resolver(credit, doc._id, doc._rev, serializer.toJson(doc)) recover
        { case e: Exception => throw new Exception(f"Unable to update document with ID $doc._id at revision $doc._rev", e) }
      res <- update(upd.extract[DocUpdate])
    } yield res

  def saveRawDoc(doc: JValue): Future[Option[JValue]] = serializer.fromCouchJson(doc) match {
    case Some((id, rev)) =>
      for {
        upd <- resolver(credit, id, rev, doc)
        res <- updateRaw(docUpdateResult(upd))
      } yield res
{'fraction_non_alphanumeric': 0.1139500492934604, 'fraction_numerical': 0.001807426881367072, 'mean_word_length': 3.0903897849462365, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 20, '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': '4821937', 'n_tokens_mistral': 3772, 'n_tokens_neox': 3667, 'n_words': 1172}