<commit_before>package im.actor.server.dialog

import java.time.Instant

import akka.actor.Status
import akka.http.scaladsl.util.FastFuture
import akka.pattern.pipe
import com.google.protobuf.wrappers.Int64Value
import im.actor.api.rpc.PeersImplicits
import im.actor.api.rpc.messaging._
import im.actor.server.ApiConversions._
import im.actor.server.dialog.HistoryUtils._
import im.actor.server.model._
import im.actor.server.persist.messaging.ReactionEventRepo
import im.actor.server.pubsub.{ PeerMessage, PubSubExtension }
import im.actor.server.sequence.{ SeqState, SeqStateDate }
import im.actor.server.social.SocialManager
import im.actor.util.cache.CacheHelpers._
import org.joda.time.DateTime

import scala.concurrent.Future
import scala.util.Failure

trait DialogCommandHandlers extends PeersImplicits with UserAcl {
  this: DialogProcessor ⇒

  import DialogCommands._
  import DialogEvents._

  protected def sendMessage(sm: SendMessage): Unit = {
    becomeStashing(replyTo ⇒ ({
      case seq: SeqStateDate ⇒
        persist(NewMessage(sm.randomId, Instant.ofEpochMilli(seq.date), sm.getOrigin.id, sm.message.header)) { e ⇒
          commit(e)
          replyTo ! seq
          unstashAll()
          context.become(receiveCommand)
        }
      case fail: Status.Failure ⇒
        log.error(fail.cause, "Failed to send message")
        replyTo forward fail
        context.become(receiveCommand)
        unstashAll()
    }: Receive) orElse reactions, discardOld = true)

    withValidAccessHash(sm.getDest, sm.senderAuthId map (_.value), sm.accessHash map (_.value)) {
      withCachedFuture[AuthSidRandomId, SeqStateDate](sm.senderAuthSid → sm.randomId) {
        val sendDate = calcSendDate
        val message = sm.message
        PubSubExtension(system).publish(PeerMessage(sm.getOrigin, sm.getDest, sm.randomId, sendDate, message))

        withNonBlockedPeer[SeqStateDate](userId, sm.getDest)(
          default = for {
          _ ← dialogExt.ackSendMessage(peer, sm.copy(date = Some(Int64Value(sendDate))))
          _ ← db.run(writeHistoryMessage(selfPeer, peer, new DateTime(sendDate), sm.randomId, message.header, message.toByteArray))
          //_ = dialogExt.updateCounters(peer, userId)
          SeqState(seq, state) ← deliveryExt.senderDelivery(userId, sm.senderAuthSid, peer, sm.randomId, sendDate, message, sm.isFat)
        } yield SeqStateDate(seq, state, sendDate),
          failed = for {
          _ ← db.run(writeHistoryMessageSelf(userId, peer, userId, new DateTime(sendDate), sm.randomId, message.header, message.toByteArray))
          SeqState(seq, state) ← deliveryExt.senderDelivery(userId, sm.senderAuthSid, peer, sm.randomId, sendDate, message, sm.isFat)
        } yield SeqStateDate(seq, state, sendDate)
        )
      }
    }
  }

  protected def ackSendMessage(sm: SendMessage): Unit = {
    val messageDate = sm.date getOrElse {
      throw new RuntimeException("No message date found in SendMessage")
    }

    persist(NewMessage(sm.randomId, Instant.ofEpochMilli(messageDate.value), sm.getOrigin.id, sm.message.header)) { e ⇒
      commit(e)

      if (peer.typ == PeerType.Private) {
        SocialManager.recordRelation(sm.getOrigin.id, userId)
        SocialManager.recordRelation(userId, sm.getOrigin.id)
      }

      deliveryExt
        .receiverDelivery(userId, sm.getOrigin.id, peer, sm.randomId, messageDate.value, sm.message, sm.isFat)
        .map(_ ⇒ SendMessageAck())
        .pipeTo(sender())

      deliveryExt.sendCountersUpdate(userId)
    }
  }

  protected def writeMessageSelf(
    senderUserId: Int,
    dateMillis:   Long,
    randomId:     Long,
    message:      ApiMessage
  ): Unit = {
    if (peer.`type` == PeerType.Private && peer.id != senderUserId && userId != senderUserId) {
      sender() ! Status.Failure(new RuntimeException(s"writeMessageSelf with senderUserId $senderUserId in dialog of user $userId with user ${peer.id}"))
    } else {
      persist(NewMessage(randomId, Instant.ofEpochMilli(dateMillis), senderUserId, message.header)) { e ⇒
        commit(e)
        db.run(writeHistoryMessageSelf(userId, peer, senderUserId, new DateTime(dateMillis), randomId, message.header, message.toByteArray))
          .map(_ ⇒ WriteMessageSelfAck()) pipeTo sender()
      }
    }
  }

  protected def messageReceived(mr: MessageReceived): Unit = {
    val mustReceive = mustMakeReceive(state, mr)

    if (mustReceive) {
      (for {
        _ ← dialogExt.ackMessageReceived(peer, mr)
      } yield MessageReceivedAck()) pipeTo sender()
    } else {
      sender() ! MessageReceivedAck()
    }
  }

  protected def ackMessageReceived(mr: MessageReceived): Unit = {
    persist(MessagesReceived(Instant.ofEpochMilli(mr.date))) { e ⇒
      commit(e)

      (deliveryExt.notifyReceive(userId, peer, mr.date, mr.now) map { _ ⇒ MessageReceivedAck() }) pipeTo sender() andThen {
        case Failure(err) ⇒ log.error(err, "Failed to ack MessageReceived")
      }
    }
  }

  protected def messageRead(mr: MessageRead): Unit = {
    val mustRead = mustMakeRead(state, mr)
    log.debug(s"mustRead is ${mustRead}")

    if (mustRead) {
      persist(MessagesRead(Instant.ofEpochMilli(mr.date), mr.getOrigin.id)) { e ⇒
        log.debug(s"persisted MessagesRead, origin=${mr.getOrigin.id}, date=${Instant.ofEpochMilli(mr.date)}, counter=${state.counter}, unreadMessages=${state.unreadMessages}")
        commit(e)
        log.debug(s"after commit: counter=${state.counter}, unreadMessages=${state.unreadMessages}")

        (for {
          _ ← dialogExt.ackMessageRead(peer, mr)
          _ ← deliveryExt.read(userId, mr.readerAuthSid, peer, mr.date, state.counter)
          _ = deliveryExt.sendCountersUpdate(userId)
        } yield MessageReadAck()) pipeTo sender()
      }
    } else {
      sender() ! MessageReadAck()
    }
  }

  protected def ackMessageRead(mr: MessageRead): Unit = {
    require(mr.getOrigin.typ.isPrivate)
    persist(MessagesRead(Instant.ofEpochMilli(mr.date), mr.getOrigin.id)) { e ⇒
      commit(e)
      log.debug(s"=== new lastReadDate is ${state.lastReadDate}")
      (deliveryExt.notifyRead(userId, peer, mr.date, mr.now) map { _ ⇒ MessageReadAck() }) pipeTo sender() andThen {
        case Failure(err) ⇒ log.error(err, "Failed to ack MessageRead")
      }
    }
  }

  protected def setReaction(sr: SetReaction): Unit = {
    (for {
      reactions ← db.run {
        ReactionEventRepo.create(DialogId(peer, userId), sr.randomId, sr.code, userId)
          .andThen(dialogExt.fetchReactions(peer, userId, sr.randomId))
      }
      seqstate ← seqUpdExt.deliverSingleUpdate(
        userId,
        UpdateReactionsUpdate(peer.asStruct, sr.randomId, reactions.toVector)
      )
      _ ← dialogExt.ackSetReaction(peer, sr)
    } yield SetReactionAck(Some(seqstate), reactions)) pipeTo sender()
  }

  protected def ackSetReaction(sr: SetReaction): Unit = {<commit_msg>perf(server:messaging): persistAsync in DialogCommandHandlers<commit_after>package im.actor.server.dialog

import java.time.Instant

import akka.actor.Status
import akka.http.scaladsl.util.FastFuture
import akka.pattern.pipe
import com.google.protobuf.wrappers.Int64Value
import im.actor.api.rpc.PeersImplicits
import im.actor.api.rpc.messaging._
import im.actor.server.ApiConversions._
import im.actor.server.dialog.HistoryUtils._
import im.actor.server.model._
import im.actor.server.persist.messaging.ReactionEventRepo
import im.actor.server.pubsub.{ PeerMessage, PubSubExtension }
import im.actor.server.sequence.{ SeqState, SeqStateDate }
import im.actor.server.social.SocialManager
import im.actor.util.cache.CacheHelpers._
import org.joda.time.DateTime

import scala.concurrent.Future
import scala.util.Failure

trait DialogCommandHandlers extends PeersImplicits with UserAcl {
  this: DialogProcessor ⇒

  import DialogCommands._
  import DialogEvents._

  protected def sendMessage(sm: SendMessage): Unit = {
    becomeStashing(replyTo ⇒ ({
      case seq: SeqStateDate ⇒
        persist(NewMessage(sm.randomId, Instant.ofEpochMilli(seq.date), sm.getOrigin.id, sm.message.header)) { e ⇒
          commit(e)
          replyTo ! seq
          unstashAll()
          context.become(receiveCommand)
        }
      case fail: Status.Failure ⇒
        log.error(fail.cause, "Failed to send message")
        replyTo forward fail
        context.become(receiveCommand)
        unstashAll()
    }: Receive) orElse reactions, discardOld = true)

    withValidAccessHash(sm.getDest, sm.senderAuthId map (_.value), sm.accessHash map (_.value)) {
      withCachedFuture[AuthSidRandomId, SeqStateDate](sm.senderAuthSid → sm.randomId) {
        val sendDate = calcSendDate
        val message = sm.message
        PubSubExtension(system).publish(PeerMessage(sm.getOrigin, sm.getDest, sm.randomId, sendDate, message))

        withNonBlockedPeer[SeqStateDate](userId, sm.getDest)(
          default = for {
          _ ← dialogExt.ackSendMessage(peer, sm.copy(date = Some(Int64Value(sendDate))))
          _ ← db.run(writeHistoryMessage(selfPeer, peer, new DateTime(sendDate), sm.randomId, message.header, message.toByteArray))
          //_ = dialogExt.updateCounters(peer, userId)
          SeqState(seq, state) ← deliveryExt.senderDelivery(userId, sm.senderAuthSid, peer, sm.randomId, sendDate, message, sm.isFat)
        } yield SeqStateDate(seq, state, sendDate),
          failed = for {
          _ ← db.run(writeHistoryMessageSelf(userId, peer, userId, new DateTime(sendDate), sm.randomId, message.header, message.toByteArray))
          SeqState(seq, state) ← deliveryExt.senderDelivery(userId, sm.senderAuthSid, peer, sm.randomId, sendDate, message, sm.isFat)
        } yield SeqStateDate(seq, state, sendDate)
        )
      }
    }
  }

  protected def ackSendMessage(sm: SendMessage): Unit = {
    val messageDate = sm.date getOrElse {
      throw new RuntimeException("No message date found in SendMessage")
    }

    persistAsync(NewMessage(sm.randomId, Instant.ofEpochMilli(messageDate.value), sm.getOrigin.id, sm.message.header)) { e ⇒
      commit(e)

      if (peer.typ == PeerType.Private) {
        SocialManager.recordRelation(sm.getOrigin.id, userId)
        SocialManager.recordRelation(userId, sm.getOrigin.id)
      }

      deliveryExt
        .receiverDelivery(userId, sm.getOrigin.id, peer, sm.randomId, messageDate.value, sm.message, sm.isFat)
        .map(_ ⇒ SendMessageAck())
        .pipeTo(sender())

      deliveryExt.sendCountersUpdate(userId)
    }
  }

  protected def writeMessageSelf(
    senderUserId: Int,
    dateMillis:   Long,
    randomId:     Long,
    message:      ApiMessage
  ): Unit = {
    if (peer.`type` == PeerType.Private && peer.id != senderUserId && userId != senderUserId) {
      sender() ! Status.Failure(new RuntimeException(s"writeMessageSelf with senderUserId $senderUserId in dialog of user $userId with user ${peer.id}"))
    } else {
      persist(NewMessage(randomId, Instant.ofEpochMilli(dateMillis), senderUserId, message.header)) { e ⇒
        commit(e)
        db.run(writeHistoryMessageSelf(userId, peer, senderUserId, new DateTime(dateMillis), randomId, message.header, message.toByteArray))
          .map(_ ⇒ WriteMessageSelfAck()) pipeTo sender()
      }
    }
  }

  protected def messageReceived(mr: MessageReceived): Unit = {
    val mustReceive = mustMakeReceive(state, mr)

    if (mustReceive) {
      (for {
        _ ← dialogExt.ackMessageReceived(peer, mr)
      } yield MessageReceivedAck()) pipeTo sender()
    } else {
      sender() ! MessageReceivedAck()
    }
  }

  protected def ackMessageReceived(mr: MessageReceived): Unit = {
    persistAsync(MessagesReceived(Instant.ofEpochMilli(mr.date))) { e ⇒
      commit(e)

      (deliveryExt.notifyReceive(userId, peer, mr.date, mr.now) map { _ ⇒ MessageReceivedAck() }) pipeTo sender() andThen {
        case Failure(err) ⇒ log.error(err, "Failed to ack MessageReceived")
      }
    }
  }

  protected def messageRead(mr: MessageRead): Unit = {
    val mustRead = mustMakeRead(state, mr)
    log.debug(s"mustRead is ${mustRead}")

    if (mustRead) {
      persistAsync(MessagesRead(Instant.ofEpochMilli(mr.date), mr.getOrigin.id)) { e ⇒
        log.debug(s"persisted MessagesRead, origin=${mr.getOrigin.id}, date=${Instant.ofEpochMilli(mr.date)}, counter=${state.counter}, unreadMessages=${state.unreadMessages}")
        commit(e)
        log.debug(s"after commit: counter=${state.counter}, unreadMessages=${state.unreadMessages}")

        (for {
          _ ← dialogExt.ackMessageRead(peer, mr)
          _ ← deliveryExt.read(userId, mr.readerAuthSid, peer, mr.date, state.counter)
          _ = deliveryExt.sendCountersUpdate(userId)
        } yield MessageReadAck()) pipeTo sender()
      }
    } else {
      sender() ! MessageReadAck()
    }
  }

  protected def ackMessageRead(mr: MessageRead): Unit = {
    require(mr.getOrigin.typ.isPrivate)
    persistAsync(MessagesRead(Instant.ofEpochMilli(mr.date), mr.getOrigin.id)) { e ⇒
      commit(e)
      log.debug(s"=== new lastReadDate is ${state.lastReadDate}")
      (deliveryExt.notifyRead(userId, peer, mr.date, mr.now) map { _ ⇒ MessageReadAck() }) pipeTo sender() andThen {
        case Failure(err) ⇒ log.error(err, "Failed to ack MessageRead")
      }
    }
  }

  protected def setReaction(sr: SetReaction): Unit = {
    (for {
      reactions ← db.run {
        ReactionEventRepo.create(DialogId(peer, userId), sr.randomId, sr.code, userId)
          .andThen(dialogExt.fetchReactions(peer, userId, sr.randomId))
      }
      seqstate ← seqUpdExt.deliverSingleUpdate(
        userId,
        UpdateReactionsUpdate(peer.asStruct, sr.randomId, reactions.toVector)
      )
      _ ← dialogExt.ackSetReaction(peer, sr)
    } yield SetReactionAck(Some(seqstate), reactions)) pipeTo sender()
  }

  protected def ackSetReaction(sr: SetReaction): Unit = {
{'fraction_non_alphanumeric': 0.11687373591447558, 'fraction_numerical': 0.0005778676683039584, 'mean_word_length': 3.409235668789809, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5111325', 'n_tokens_mistral': 4422, 'n_tokens_neox': 4439, 'n_words': 883}