<commit_before>package im.actor.server.group

import im.actor.server.event.TSEvent

import scala.concurrent.duration._
import scala.concurrent.{ ExecutionContext, Future }
import scala.util.{ Failure, Success }

import akka.actor._
import akka.contrib.pattern.ShardRegion
import akka.pattern.pipe
import akka.persistence.{ RecoveryCompleted, RecoveryFailure }
import akka.util.Timeout
import com.github.benmanes.caffeine.cache.Cache
import org.joda.time.DateTime
import slick.driver.PostgresDriver.api._

import im.actor.server.commons.serialization.ActorSerializer
import im.actor.server.file.Avatar
import im.actor.server.office.PeerProcessor.MessageSentComplete
import im.actor.server.office.{ ProcessorState, PeerProcessor, StopOffice }
import im.actor.server.push.SeqUpdatesManagerRegion
import im.actor.server.sequence.SeqStateDate
import im.actor.server.user.{ UserViewRegion, UserProcessorRegion }
import im.actor.server.util.FileStorageAdapter
import im.actor.utils.cache.CacheHelpers._

private[group] case class Member(
  userId:        Int,
  inviterUserId: Int,
  invitedAt:     DateTime,
  isAdmin:       Boolean
)

private[group] case class Bot(
  userId: Int,
  token:  String
)

private[group] case class Group(
  id:               Int,
  accessHash:       Long,
  creatorUserId:    Int,
  createdAt:        DateTime,
  members:          Map[Int, Member],
  invitedUserIds:   Set[Int],
  title:            String,
  about:            Option[String],
  isPublic:         Boolean,
  lastSenderId:     Option[Int],
  lastReceivedDate: Option[DateTime],
  lastReadDate:     Option[DateTime],
  bot:              Option[Bot],
  avatar:           Option[Avatar],
  topic:            Option[String]
) extends ProcessorState

trait GroupCommand {
  val groupId: Int
}

trait GroupEvent

trait GroupQuery {
  val groupId: Int
}

object GroupProcessor {

  private case class Initialized(groupUsersIds: Set[Int], invitedUsersIds: Set[Int], isPublic: Boolean)

  def register(): Unit = {
    ActorSerializer.register(20001, classOf[GroupCommands.Create])
    ActorSerializer.register(20002, classOf[GroupCommands.CreateAck])
    ActorSerializer.register(20003, classOf[GroupCommands.Invite])
    ActorSerializer.register(20004, classOf[GroupCommands.Join])
    ActorSerializer.register(20005, classOf[GroupCommands.Kick])
    ActorSerializer.register(20006, classOf[GroupCommands.Leave])
    ActorSerializer.register(20007, classOf[GroupCommands.SendMessage])
    ActorSerializer.register(20008, classOf[GroupCommands.MessageReceived])
    ActorSerializer.register(20009, classOf[GroupCommands.MessageRead])
    ActorSerializer.register(20010, classOf[GroupCommands.UpdateAvatar])
    ActorSerializer.register(20011, classOf[GroupCommands.MakePublic])
    ActorSerializer.register(20012, classOf[GroupCommands.MakePublicAck])
    ActorSerializer.register(20013, classOf[GroupCommands.UpdateTitle])
    ActorSerializer.register(20014, classOf[GroupCommands.UpdateTitle])
    ActorSerializer.register(20015, classOf[GroupCommands.ChangeTopic])
    ActorSerializer.register(20016, classOf[GroupCommands.ChangeAbout])
    ActorSerializer.register(20017, classOf[GroupCommands.MakeUserAdmin])
    ActorSerializer.register(20018, classOf[GroupCommands.RevokeIntegrationToken])
    ActorSerializer.register(20020, classOf[GroupCommands.RevokeIntegrationTokenAck])

    ActorSerializer.register(21001, classOf[GroupQueries.GetIntegrationToken])
    ActorSerializer.register(21002, classOf[GroupQueries.GetIntegrationTokenResponse])

    ActorSerializer.register(22001, classOf[GroupEvents.MessageRead])
    ActorSerializer.register(22002, classOf[GroupEvents.MessageReceived])
    ActorSerializer.register(22003, classOf[GroupEvents.UserInvited])
    ActorSerializer.register(22004, classOf[GroupEvents.UserJoined])
    ActorSerializer.register(22005, classOf[GroupEvents.Created])
    ActorSerializer.register(22006, classOf[GroupEvents.BotAdded])
    ActorSerializer.register(22007, classOf[GroupEvents.UserKicked])
    ActorSerializer.register(22008, classOf[GroupEvents.UserLeft])
    ActorSerializer.register(22009, classOf[GroupEvents.AvatarUpdated])
    ActorSerializer.register(22010, classOf[GroupEvents.BecamePublic])
    ActorSerializer.register(22011, classOf[GroupEvents.AboutUpdated])
    ActorSerializer.register(22012, classOf[GroupEvents.TitleUpdated])
    ActorSerializer.register(22013, classOf[GroupEvents.TopicUpdated])
    ActorSerializer.register(22014, classOf[GroupEvents.AboutUpdated])
    ActorSerializer.register(22015, classOf[GroupEvents.UserBecameAdmin])
    ActorSerializer.register(22016, classOf[GroupEvents.IntegrationTokenRevoked])
  }

  def props(
    implicit
    db:                  Database,
    seqUpdManagerRegion: SeqUpdatesManagerRegion,
    userProcessorRegion: UserProcessorRegion,
    userViewRegion:      UserViewRegion,
    fsAdapter:           FileStorageAdapter
  ): Props = Props(classOf[GroupProcessor], db, seqUpdManagerRegion, userProcessorRegion, userViewRegion, fsAdapter)
}

private[group] final class GroupProcessor(
  implicit
  protected val db:                  Database,
  protected val seqUpdManagerRegion: SeqUpdatesManagerRegion,
  protected val userOfficeRegion:    UserProcessorRegion,
  protected val userViewRegion:      UserViewRegion,
  protected val fsAdapter:           FileStorageAdapter
) extends PeerProcessor[Group, TSEvent] with GroupCommandHandlers with GroupQueryHandlers with ActorLogging with Stash with GroupsImplicits {

  import GroupCommands._
  import GroupErrors._

  implicit protected val system: ActorSystem = context.system
  implicit protected val ec: ExecutionContext = context.dispatcher

  implicit protected val timeout: Timeout = Timeout(10.seconds)

  protected val groupId = self.path.name.toInt

  override def persistenceId = GroupOffice.persistenceIdFor(groupId)

  context.setReceiveTimeout(1.hour)

  private val MaxCacheSize = 100L

  implicit val sendResponseCache: Cache[AuthIdRandomId, Future[SeqStateDate]] =
    createCache[AuthIdRandomId, Future[SeqStateDate]](MaxCacheSize)

  def updatedState(evt: TSEvent, state: Group): Group = {
    evt match {
      case TSEvent(_, GroupEvents.BotAdded(userId, token)) ⇒
        state.copy(bot = Some(Bot(userId, token)))
      case TSEvent(_, GroupEvents.MessageReceived(date)) ⇒
        state.copy(lastReceivedDate = Some(new DateTime(date)))
      case TSEvent(_, GroupEvents.MessageRead(userId, date)) ⇒
        state.copy(
          lastReadDate = Some(new DateTime(date)),<commit_msg>perf(server): passivate GroupProcessor in 5 hours of inactivity<commit_after>package im.actor.server.group

import im.actor.server.event.TSEvent

import scala.concurrent.duration._
import scala.concurrent.{ ExecutionContext, Future }
import scala.util.{ Failure, Success }

import akka.actor._
import akka.contrib.pattern.ShardRegion
import akka.pattern.pipe
import akka.persistence.{ RecoveryCompleted, RecoveryFailure }
import akka.util.Timeout
import com.github.benmanes.caffeine.cache.Cache
import org.joda.time.DateTime
import slick.driver.PostgresDriver.api._

import im.actor.server.commons.serialization.ActorSerializer
import im.actor.server.file.Avatar
import im.actor.server.office.PeerProcessor.MessageSentComplete
import im.actor.server.office.{ ProcessorState, PeerProcessor, StopOffice }
import im.actor.server.push.SeqUpdatesManagerRegion
import im.actor.server.sequence.SeqStateDate
import im.actor.server.user.{ UserViewRegion, UserProcessorRegion }
import im.actor.server.util.FileStorageAdapter
import im.actor.utils.cache.CacheHelpers._

private[group] case class Member(
  userId:        Int,
  inviterUserId: Int,
  invitedAt:     DateTime,
  isAdmin:       Boolean
)

private[group] case class Bot(
  userId: Int,
  token:  String
)

private[group] case class Group(
  id:               Int,
  accessHash:       Long,
  creatorUserId:    Int,
  createdAt:        DateTime,
  members:          Map[Int, Member],
  invitedUserIds:   Set[Int],
  title:            String,
  about:            Option[String],
  isPublic:         Boolean,
  lastSenderId:     Option[Int],
  lastReceivedDate: Option[DateTime],
  lastReadDate:     Option[DateTime],
  bot:              Option[Bot],
  avatar:           Option[Avatar],
  topic:            Option[String]
) extends ProcessorState

trait GroupCommand {
  val groupId: Int
}

trait GroupEvent

trait GroupQuery {
  val groupId: Int
}

object GroupProcessor {

  private case class Initialized(groupUsersIds: Set[Int], invitedUsersIds: Set[Int], isPublic: Boolean)

  def register(): Unit = {
    ActorSerializer.register(20001, classOf[GroupCommands.Create])
    ActorSerializer.register(20002, classOf[GroupCommands.CreateAck])
    ActorSerializer.register(20003, classOf[GroupCommands.Invite])
    ActorSerializer.register(20004, classOf[GroupCommands.Join])
    ActorSerializer.register(20005, classOf[GroupCommands.Kick])
    ActorSerializer.register(20006, classOf[GroupCommands.Leave])
    ActorSerializer.register(20007, classOf[GroupCommands.SendMessage])
    ActorSerializer.register(20008, classOf[GroupCommands.MessageReceived])
    ActorSerializer.register(20009, classOf[GroupCommands.MessageRead])
    ActorSerializer.register(20010, classOf[GroupCommands.UpdateAvatar])
    ActorSerializer.register(20011, classOf[GroupCommands.MakePublic])
    ActorSerializer.register(20012, classOf[GroupCommands.MakePublicAck])
    ActorSerializer.register(20013, classOf[GroupCommands.UpdateTitle])
    ActorSerializer.register(20014, classOf[GroupCommands.UpdateTitle])
    ActorSerializer.register(20015, classOf[GroupCommands.ChangeTopic])
    ActorSerializer.register(20016, classOf[GroupCommands.ChangeAbout])
    ActorSerializer.register(20017, classOf[GroupCommands.MakeUserAdmin])
    ActorSerializer.register(20018, classOf[GroupCommands.RevokeIntegrationToken])
    ActorSerializer.register(20020, classOf[GroupCommands.RevokeIntegrationTokenAck])

    ActorSerializer.register(21001, classOf[GroupQueries.GetIntegrationToken])
    ActorSerializer.register(21002, classOf[GroupQueries.GetIntegrationTokenResponse])

    ActorSerializer.register(22001, classOf[GroupEvents.MessageRead])
    ActorSerializer.register(22002, classOf[GroupEvents.MessageReceived])
    ActorSerializer.register(22003, classOf[GroupEvents.UserInvited])
    ActorSerializer.register(22004, classOf[GroupEvents.UserJoined])
    ActorSerializer.register(22005, classOf[GroupEvents.Created])
    ActorSerializer.register(22006, classOf[GroupEvents.BotAdded])
    ActorSerializer.register(22007, classOf[GroupEvents.UserKicked])
    ActorSerializer.register(22008, classOf[GroupEvents.UserLeft])
    ActorSerializer.register(22009, classOf[GroupEvents.AvatarUpdated])
    ActorSerializer.register(22010, classOf[GroupEvents.BecamePublic])
    ActorSerializer.register(22011, classOf[GroupEvents.AboutUpdated])
    ActorSerializer.register(22012, classOf[GroupEvents.TitleUpdated])
    ActorSerializer.register(22013, classOf[GroupEvents.TopicUpdated])
    ActorSerializer.register(22014, classOf[GroupEvents.AboutUpdated])
    ActorSerializer.register(22015, classOf[GroupEvents.UserBecameAdmin])
    ActorSerializer.register(22016, classOf[GroupEvents.IntegrationTokenRevoked])
  }

  def props(
    implicit
    db:                  Database,
    seqUpdManagerRegion: SeqUpdatesManagerRegion,
    userProcessorRegion: UserProcessorRegion,
    userViewRegion:      UserViewRegion,
    fsAdapter:           FileStorageAdapter
  ): Props = Props(classOf[GroupProcessor], db, seqUpdManagerRegion, userProcessorRegion, userViewRegion, fsAdapter)
}

private[group] final class GroupProcessor(
  implicit
  protected val db:                  Database,
  protected val seqUpdManagerRegion: SeqUpdatesManagerRegion,
  protected val userOfficeRegion:    UserProcessorRegion,
  protected val userViewRegion:      UserViewRegion,
  protected val fsAdapter:           FileStorageAdapter
) extends PeerProcessor[Group, TSEvent] with GroupCommandHandlers with GroupQueryHandlers with ActorLogging with Stash with GroupsImplicits {

  import GroupCommands._
  import GroupErrors._

  implicit protected val system: ActorSystem = context.system
  implicit protected val ec: ExecutionContext = context.dispatcher

  implicit protected val timeout: Timeout = Timeout(10.seconds)

  protected val groupId = self.path.name.toInt

  override def persistenceId = GroupOffice.persistenceIdFor(groupId)

  context.setReceiveTimeout(5.hours)

  private val MaxCacheSize = 100L

  implicit val sendResponseCache: Cache[AuthIdRandomId, Future[SeqStateDate]] =
    createCache[AuthIdRandomId, Future[SeqStateDate]](MaxCacheSize)

  def updatedState(evt: TSEvent, state: Group): Group = {
    evt match {
      case TSEvent(_, GroupEvents.BotAdded(userId, token)) ⇒
        state.copy(bot = Some(Bot(userId, token)))
      case TSEvent(_, GroupEvents.MessageReceived(date)) ⇒
        state.copy(lastReceivedDate = Some(new DateTime(date)))
      case TSEvent(_, GroupEvents.MessageRead(userId, date)) ⇒
        state.copy(
          lastReadDate = Some(new DateTime(date)),
{'fraction_non_alphanumeric': 0.08928979218999772, 'fraction_numerical': 0.029154297023673593, 'mean_word_length': 5.004570383912249, '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': '4049332', 'n_tokens_mistral': 4015, 'n_tokens_neox': 3748, 'n_words': 694}