<commit_before>package fcmworker

import _root_.models.{Android, Notification, ShardRange}
import cats.data.NonEmptyList
import cats.effect.{Async, Concurrent}
import cats.syntax.either._
import cats.syntax.functor._
import cats.syntax.list._
import db.{RegistrationService, Topic}
import fcmworker.FcmClient.{FcmResponse, FcmSuccess, MessageId, Token}
import fcmworker.models.FcmException.{FcmGenericFailure, FcmInvalidPayload, FcmInvalidTopics}
import fcmworker.models.payload.FcmPayload
import fcmworker.models.{FcmConfig, FcmException}
import fs2.{Pipe, Stream}

import scala.concurrent.ExecutionContextExecutor

class Fcm[F[_]](registrationService: RegistrationService[F, Stream], config: FcmConfig)
  (implicit executionContext: ExecutionContextExecutor, concurrent: Concurrent[F], F: Async[F]) {

  private val fcmClientF: F[FcmClient] =
    FcmClient(config).fold(e => F.raiseError(e), c => F.delay(c))

  def send(notification: Notification, shardRange: ShardRange) = {

    def sendAsync(token: Token, payload: FcmPayload)(client: FcmClient): F[FcmSuccess] =
      Async[F].async[FcmSuccess] { (cb: FcmResponse => Unit) =>
        client.sendNotification(notification.id, token, payload)(cb)
      }

    val topicsF: F[NonEmptyList[Topic]] = notification
      .topic
      .map(t => Topic(t.name))
      .toNel
      .map(nel => F.delay(nel))
      .getOrElse(F.raiseError(FcmInvalidTopics(notification.id)))

    def tokens: Stream[F, Token] = for {
        topics <- Stream.eval(topicsF)
        res <- registrationService.findTokens(topics, Some(Android), Some(shardRange))
    } yield res

    def sending(token: Token, payload: FcmPayload, fcmClient: FcmClient): Stream[F, FcmResponse] = {

      def toFcmResponse: Pipe[F, FcmSuccess, FcmResponse] =
        _.attempt.map {
          _.leftMap {
            _ match {
              case fe: FcmException => fe
              case e: Throwable => FcmGenericFailure(notification.id, token, e)
            }
          }
        }

      Stream
        .eval(sendAsync(token, payload)(fcmClient))
        .through(toFcmResponse)
    }

    val payloadF = FcmPayload(notification, config.debug)
      .fold[F[FcmPayload]](F.raiseError(FcmInvalidPayload(notification.id)))(p => F.delay(p))

    for {
      fcmClient <- Stream.eval(fcmClientF)
      payload <- Stream.eval(payloadF)
      res <- tokens
        .map(token => sending(token, payload, fcmClient))
        .parJoinUnbounded
    } yield res
  }

  def close(): F[Unit] = fcmClientF.map(_.close())
}
<commit_msg>fcmworker: limit concurrency when calling Firebase sdk<commit_after>package fcmworker

import _root_.models.{Android, Notification, ShardRange}
import cats.data.NonEmptyList
import cats.effect.{Async, Concurrent}
import cats.syntax.either._
import cats.syntax.functor._
import cats.syntax.list._
import db.{RegistrationService, Topic}
import fcmworker.FcmClient.{FcmResponse, FcmSuccess, MessageId, Token}
import fcmworker.models.FcmException.{FcmGenericFailure, FcmInvalidPayload, FcmInvalidTopics}
import fcmworker.models.payload.FcmPayload
import fcmworker.models.{FcmConfig, FcmException}
import fs2.{Pipe, Stream}

import scala.concurrent.ExecutionContextExecutor

class Fcm[F[_]](registrationService: RegistrationService[F, Stream], config: FcmConfig)
  (implicit executionContext: ExecutionContextExecutor, concurrent: Concurrent[F], F: Async[F]) {

  private val fcmClientF: F[FcmClient] =
    FcmClient(config).fold(e => F.raiseError(e), c => F.delay(c))

  def send(notification: Notification, shardRange: ShardRange) = {

    def sendAsync(token: Token, payload: FcmPayload)(client: FcmClient): F[FcmSuccess] =
      Async[F].async[FcmSuccess] { (cb: FcmResponse => Unit) =>
        client.sendNotification(notification.id, token, payload)(cb)
      }

    val topicsF: F[NonEmptyList[Topic]] = notification
      .topic
      .map(t => Topic(t.name))
      .toNel
      .map(nel => F.delay(nel))
      .getOrElse(F.raiseError(FcmInvalidTopics(notification.id)))

    def tokens: Stream[F, Token] = for {
        topics <- Stream.eval(topicsF)
        res <- registrationService.findTokens(topics, Some(Android), Some(shardRange))
    } yield res

    def sending(token: Token, payload: FcmPayload, fcmClient: FcmClient): Stream[F, FcmResponse] = {

      def toFcmResponse: Pipe[F, FcmSuccess, FcmResponse] =
        _.attempt.map {
          _.leftMap {
            _ match {
              case fe: FcmException => fe
              case e: Throwable => FcmGenericFailure(notification.id, token, e)
            }
          }
        }

      Stream
        .eval(sendAsync(token, payload)(fcmClient))
        .through(toFcmResponse)
    }

    val payloadF = FcmPayload(notification, config.debug)
      .fold[F[FcmPayload]](F.raiseError(FcmInvalidPayload(notification.id)))(p => F.delay(p))

    val maxConcurrency = 500
    for {
      fcmClient <- Stream.eval(fcmClientF)
      payload <- Stream.eval(payloadF)
      res <- tokens
        .map(token => sending(token, payload, fcmClient))
        .parJoin(maxConcurrency)
    } yield res
  }

  def close(): F[Unit] = fcmClientF.map(_.close())
}

{'fraction_non_alphanumeric': 0.12206025267249757, 'fraction_numerical': 0.0009718172983479105, 'mean_word_length': 3.574222222222222, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 23, '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': '7477199', 'n_tokens_mistral': 1666, 'n_tokens_neox': 1613, 'n_words': 325}