<commit_before>package lila.socket

import chess.Centis
import io.lettuce.core._
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection
import play.api.libs.json._
import scala.concurrent.Future

import lila.common.Chronometer
import lila.hub.actorApi.relation.ReloadOnlineFriends
import lila.hub.actorApi.round.{ MoveEvent, Mlat }
import lila.hub.actorApi.security.CloseAccount
import lila.hub.actorApi.socket.remote.{ TellSriIn, TellSriOut }
import lila.hub.actorApi.socket.{ SendTo, SendTos, WithUserIds }
import lila.hub.actorApi.{ Deploy, Announce }
import lila.socket.actorApi.SendToFlag
import Socket.Sri

final class RemoteSocket(
    redisClient: RedisClient,
    notificationActor: akka.actor.ActorSelection,
    setNb: Int => Unit,
    bus: lila.common.Bus,
    lifecycle: play.api.inject.ApplicationLifecycle
) {

  import RemoteSocket._, Protocol._

  private val connectedUserIds = collection.mutable.Set.empty[String]
  private val watchedGameIds = collection.mutable.Set.empty[String]

  private def connectToPubSub = redisClient.connectPubSub()

  val baseHandler: Handler = {
    case In.ConnectUser(userId) =>
      bus.publish(lila.hub.actorApi.socket.remote.ConnectUser(userId), 'userActive)
      connectedUserIds += userId
    case In.DisconnectUsers(userIds) => userIds foreach { userId =>
      connectedUserIds -= userId
    }
    case In.Watch(gameId) => watchedGameIds += gameId
    case In.Unwatch(gameId) => watchedGameIds -= gameId
    case In.NotifiedBatch(userIds) => notificationActor ! lila.hub.actorApi.notify.NotifiedBatch(userIds)
    case In.Connections(nb) => tick(nb)
    case In.FriendsBatch(userIds) => userIds foreach { userId =>
      bus.publish(ReloadOnlineFriends(userId), 'reloadOnlineFriends)
    }
    case In.Lags(lags) => lags foreach (UserLagCache.put _).tupled
    case In.TellSri(sri, userId, typ, msg) =>
      bus.publish(TellSriIn(sri.value, userId, msg), Symbol(s"remoteSocketIn:$typ"))
    case In.DisconnectAll =>
      logger.info("Remote socket disconnect all")
      connectedUserIds.clear
      watchedGameIds.clear
  }

  bus.subscribeFun('moveEvent, 'socketUsers, 'deploy, 'announce, 'mlat, 'sendToFlag, 'remoteSocketOut, 'accountClose) {
    case MoveEvent(gameId, fen, move) =>
      if (watchedGameIds(gameId)) send(Out.move(gameId, move, fen))
    case SendTos(userIds, payload) =>
      val connectedUsers = userIds intersect connectedUserIds
      if (connectedUsers.nonEmpty) send(Out.tellUsers(connectedUsers, payload))
    case SendTo(userId, payload) if connectedUserIds(userId) =>
      send(Out.tellUser(userId, payload))
    case d: Deploy =>
      send(Out.tellAll(Json.obj("t" -> d.key)))
    case Announce(msg) =>
      send(Out.tellAll(Json.obj("t" -> "announce", "d" -> Json.obj("msg" -> msg))))
    case Mlat(micros) =>
      send(Out.mlat(micros))
    case actorApi.SendToFlag(flag, payload) =>
      send(Out.tellFlag(flag, payload))
    case TellSriOut(sri, payload) =>
      send(Out.tellSri(Sri(sri), payload))
    case CloseAccount(userId) =>
      send(Out.disconnectUser(userId))
    case WithUserIds(f) =>
      f(connectedUserIds)
  }

  private def tick(nbConn: Int): Unit = {
    setNb(nbConn)
    mon.connections(nbConn)
    mon.sets.games(watchedGameIds.size)
    val nbUsers = connectedUserIds.size
    if (nbUsers >= 0) mon.sets.users(nbUsers)
    else if (nbUsers % 10 == 0) logger.warn(s"$nbUsers connectedUsers, first = ${connectedUserIds.take(5)}")
  }

  private val mon = lila.mon.socket.remote

  def makeSender(channel: Channel): Sender = new Sender(connectToPubSub, channel)

  private val send: String => Unit = makeSender("site-out").apply _

  def subscribe(channel: Channel, reader: In.Reader)(handler: Handler): Unit = {
    val conn = connectToPubSub
    conn.addListener(new pubsub.RedisPubSubAdapter[String, String] {
      override def message(_channel: String, message: String): Unit = {
        val raw = RawMsg(message)
        mon.redis.in.channel(channel)()
        mon.redis.in.path(channel, raw.path)()
        // println(message, s"in:$channel")
        reader(raw) collect handler match {<commit_msg>reduce logging for negative set size entries<commit_after>package lila.socket

import chess.Centis
import io.lettuce.core._
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection
import play.api.libs.json._
import scala.concurrent.Future

import lila.common.Chronometer
import lila.hub.actorApi.relation.ReloadOnlineFriends
import lila.hub.actorApi.round.{ MoveEvent, Mlat }
import lila.hub.actorApi.security.CloseAccount
import lila.hub.actorApi.socket.remote.{ TellSriIn, TellSriOut }
import lila.hub.actorApi.socket.{ SendTo, SendTos, WithUserIds }
import lila.hub.actorApi.{ Deploy, Announce }
import lila.socket.actorApi.SendToFlag
import Socket.Sri

final class RemoteSocket(
    redisClient: RedisClient,
    notificationActor: akka.actor.ActorSelection,
    setNb: Int => Unit,
    bus: lila.common.Bus,
    lifecycle: play.api.inject.ApplicationLifecycle
) {

  import RemoteSocket._, Protocol._

  private val connectedUserIds = collection.mutable.Set.empty[String]
  private val watchedGameIds = collection.mutable.Set.empty[String]

  private def connectToPubSub = redisClient.connectPubSub()

  val baseHandler: Handler = {
    case In.ConnectUser(userId) =>
      bus.publish(lila.hub.actorApi.socket.remote.ConnectUser(userId), 'userActive)
      connectedUserIds += userId
    case In.DisconnectUsers(userIds) => userIds foreach { userId =>
      connectedUserIds -= userId
    }
    case In.Watch(gameId) => watchedGameIds += gameId
    case In.Unwatch(gameId) => watchedGameIds -= gameId
    case In.NotifiedBatch(userIds) => notificationActor ! lila.hub.actorApi.notify.NotifiedBatch(userIds)
    case In.Connections(nb) => tick(nb)
    case In.FriendsBatch(userIds) => userIds foreach { userId =>
      bus.publish(ReloadOnlineFriends(userId), 'reloadOnlineFriends)
    }
    case In.Lags(lags) => lags foreach (UserLagCache.put _).tupled
    case In.TellSri(sri, userId, typ, msg) =>
      bus.publish(TellSriIn(sri.value, userId, msg), Symbol(s"remoteSocketIn:$typ"))
    case In.DisconnectAll =>
      logger.info("Remote socket disconnect all")
      connectedUserIds.clear
      watchedGameIds.clear
  }

  bus.subscribeFun('moveEvent, 'socketUsers, 'deploy, 'announce, 'mlat, 'sendToFlag, 'remoteSocketOut, 'accountClose) {
    case MoveEvent(gameId, fen, move) =>
      if (watchedGameIds(gameId)) send(Out.move(gameId, move, fen))
    case SendTos(userIds, payload) =>
      val connectedUsers = userIds intersect connectedUserIds
      if (connectedUsers.nonEmpty) send(Out.tellUsers(connectedUsers, payload))
    case SendTo(userId, payload) if connectedUserIds(userId) =>
      send(Out.tellUser(userId, payload))
    case d: Deploy =>
      send(Out.tellAll(Json.obj("t" -> d.key)))
    case Announce(msg) =>
      send(Out.tellAll(Json.obj("t" -> "announce", "d" -> Json.obj("msg" -> msg))))
    case Mlat(micros) =>
      send(Out.mlat(micros))
    case actorApi.SendToFlag(flag, payload) =>
      send(Out.tellFlag(flag, payload))
    case TellSriOut(sri, payload) =>
      send(Out.tellSri(Sri(sri), payload))
    case CloseAccount(userId) =>
      send(Out.disconnectUser(userId))
    case WithUserIds(f) =>
      f(connectedUserIds)
  }

  private def tick(nbConn: Int): Unit = {
    setNb(nbConn)
    mon.connections(nbConn)
    mon.sets.games(watchedGameIds.size)
    val nbUsers = connectedUserIds.size
    if (nbUsers >= 0) mon.sets.users(nbUsers)
    else if (nbUsers % 10 == 0) logger.warn(s"$nbUsers connectedUsers")
  }

  private val mon = lila.mon.socket.remote

  def makeSender(channel: Channel): Sender = new Sender(connectToPubSub, channel)

  private val send: String => Unit = makeSender("site-out").apply _

  def subscribe(channel: Channel, reader: In.Reader)(handler: Handler): Unit = {
    val conn = connectToPubSub
    conn.addListener(new pubsub.RedisPubSubAdapter[String, String] {
      override def message(_channel: String, message: String): Unit = {
        val raw = RawMsg(message)
        mon.redis.in.channel(channel)()
        mon.redis.in.path(channel, raw.path)()
        // println(message, s"in:$channel")
        reader(raw) collect handler match {
{'fraction_non_alphanumeric': 0.12318489835430785, 'fraction_numerical': 0.0010890609874152953, 'mean_word_length': 4.036563071297989, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 61, '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': '5703054', 'n_tokens_mistral': 2711, 'n_tokens_neox': 2672, 'n_words': 542}