bulk challenge WIP - for lichess-org#8059

app/controllers/Challenge.scala

@@ -309,17 +309,17 @@ final class Challenge(
         )
     }
 
-  def bulk(userId: String) =
+  def bulk =
     ScopedBody(_.Challenge.Bulk) { implicit req => me =>
       implicit val lang = reqLang
-      lila.setup.BulkChallenge.form
+      lila.setup.SetupBulk.form
         .bindFromRequest()
         .fold(
           newJsonFormError,
           data =>
-            env.setup.bulk(data) flatMap {
+            env.setup.bulk(data) map {
               case Left(badTokens) =>
-                import lila.setup.BulkChallenge.BadToken
+                import lila.setup.SetupBulk.BadToken
                 import play.api.libs.json._
                 BadRequest(
                   Json.obj(
@@ -329,10 +329,15 @@ final class Challenge(
                       }
                     }
                   )
-                ).fuccess
+                )
               case Right(bulk) =>
-                println(bulk)
-                ???
+                env.challenge.bulk(me, bulk).thenPp
+                Ok(Json.obj("games" -> bulk.games.map { g =>
+                  Json.obj(
+                    "gameId"  -> g.id,
+                    "userIds" -> Json.arr(g.white, g.black)
+                  )
+                })) as JSON
             }
         )
     }

conf/routes

@@ -612,12 +612,12 @@ GET   /api/account/preferences         controllers.Pref.apiGet
 POST  /api/challenge/ai                controllers.Setup.apiAi
 POST  /api/challenge/open              controllers.Challenge.openCreate
 POST  /api/challenge/admin/:orig/:dest controllers.Challenge.apiCreateAdmin(orig: String, dest: String)
+POST  /api/challenge/bulk              controllers.Challenge.bulk
 POST  /api/challenge/:user             controllers.Challenge.apiCreate(user: String)
 POST  /api/challenge/$id<\w{8}>/accept controllers.Challenge.apiAccept(id: String)
 POST  /api/challenge/$id<\w{8}>/decline controllers.Challenge.apiDecline(id: String)
 POST  /api/challenge/$id<\w{8}>/cancel controllers.Challenge.apiCancel(id: String)
 POST  /api/challenge/$id<\w{8}>/start-clocks  controllers.Challenge.apiStartClocks(id: String)
-POST  /api/challenge/bulk              controllers.Challenge.bulk
 POST  /api/round/$id<\w{8}>/add-time/:seconds controllers.Round.apiAddTime(id: String, seconds: Int)
 GET   /api/cloud-eval                  controllers.Api.cloudEval
 GET   /api/broadcast                   controllers.Relay.apiIndex

modules/challenge/src/main/ChallengeBulk.scala

@@ -0,0 +1,63 @@
+package lila.challenge
+
+import akka.stream.scaladsl._
+import chess.variant.Variant
+import chess.{ Clock, Mode, Situation, Speed }
+import org.joda.time.DateTime
+import scala.util.chaining._
+
+import lila.common.LilaStream
+import lila.game.{ Game, Player }
+import lila.rating.PerfType
+import lila.setup.SetupBulk.ScheduledBulk
+import lila.user.User
+
+final class ChallengeBulkApi(
+    gameRepo: lila.game.GameRepo,
+    userRepo: lila.user.UserRepo,
+    onStart: lila.round.OnStart
+)(implicit
+    ec: scala.concurrent.ExecutionContext,
+    mat: akka.stream.Materializer
+) {
+
+  def apply(
+      by: User,
+      scheduled: ScheduledBulk
+  ): Fu[Int] = {
+    val perfType   = PerfType(scheduled.variant, Speed(scheduled.clock))
+    val startClock = scheduled.startClocksAt isBefore DateTime.now
+    Source(scheduled.games)
+      .mapAsyncUnordered(8) { game =>
+        userRepo.pair(game.white, game.black) map2 { case (white, black) =>
+          (game.id, white, black)
+        }
+      }
+      .mapConcat(_.toList)
+      .map[Game] { case (id, white, black) =>
+        Game
+          .make(
+            chess =
+              chess.Game(situation = Situation(scheduled.variant), clock = scheduled.clock.toClock.some),
+            whitePlayer = Player.make(chess.White, white.some, _(perfType)),
+            blackPlayer = Player.make(chess.Black, black.some, _(perfType)),
+            mode = scheduled.mode,
+            source = lila.game.Source.Api,
+            pgnImport = None
+          )
+          .withId(id)
+          .start
+          .pipe { g =>
+            if (startClock) g.startClock.fold(g)(_.game) else g
+          }
+      }
+      .mapAsyncUnordered(8) { game =>
+        (gameRepo insertDenormalized game) >>- onStart(game.id)
+      }
+      .toMat(LilaStream.sinkCount)(Keep.right)
+      .run()
+      .addEffect { nb =>
+        lila.mon.api.challenge.bulk.createNb(by.id).increment(nb).unit
+      }
+  }
+}

modules/challenge/src/main/Env.scala

@@ -50,6 +50,8 @@ final class Env(
 
   lazy val jsonView = wire[JsonView]
 
+  lazy val bulk = wire[ChallengeBulkApi]
+
   val forms = new ChallengeForm
 
   system.scheduler.scheduleWithFixedDelay(10 seconds, 3 seconds) { () =>

modules/common/src/main/mon.scala

@@ -602,6 +602,12 @@ object mon {
     val users     = counter("api.cost").withTag("endpoint", "users")
     val game      = counter("api.cost").withTag("endpoint", "game")
     val activity  = counter("api.cost").withTag("endpoint", "activity")
+    object challenge {
+      object bulk {
+        def scheduleNb(byUserId: String) = counter("api.challenge.bulk.schedule.nb").withTag("by", byUserId)
+        def createNb(byUserId: String)   = counter("api.challenge.bulk.create.nb").withTag("by", byUserId)
+      }
+    }
   }
   object export {
     object pgn {

modules/oauth/src/main/OAuthScope.scala

@@ -65,6 +65,7 @@ object OAuthScope {
     Email.Read,
     Challenge.Read,
     Challenge.Write,
+    Challenge.Bulk,
     Study.Read,
     Study.Write,
     Tournament.Write,

modules/setup/src/main/Env.scala

@@ -10,6 +10,7 @@ import lila.oauth.OAuthServer
 final class Env(
     appConfig: Configuration,
     gameRepo: lila.game.GameRepo,
+    idGenerator: lila.game.IdGenerator,
     fishnetPlayer: lila.fishnet.Player,
     onStart: lila.round.OnStart,
     gameCache: lila.game.Cached,

modules/setup/src/main/BulkChallenge.scala → modules/setup/src/main/SetupBulk.scala

@@ -3,18 +3,19 @@ package lila.setup
 import akka.stream.scaladsl._
 import chess.format.FEN
 import chess.variant.Variant
-import chess.{ Clock, Speed }
+import chess.{ Clock, Mode, Speed }
 import org.joda.time.DateTime
 import play.api.data._
 import play.api.data.Forms._
 
 import lila.game.Game
+import lila.game.IdGenerator
 import lila.oauth.AccessToken
 import lila.oauth.OAuthScope
 import lila.oauth.OAuthServer
 import lila.user.User
 
-object BulkChallenge {
+object SetupBulk {
 
   val maxGames = 500
 
@@ -23,8 +24,15 @@ object BulkChallenge {
   val form = Form[BulkFormData](
     mapping(
       "tokens" -> nonEmptyText
-        .verifying("Not enough tokens", t => extractTokenPairs(t).isEmpty)
-        .verifying(s"Too many tokens (max: ${maxGames * 2})", t => extractTokenPairs(t).sizeIs > maxGames),
+        .verifying("Not enough tokens", t => extractTokenPairs(t).nonEmpty)
+        .verifying(s"Too many tokens (max: ${maxGames * 2})", t => extractTokenPairs(t).sizeIs < maxGames)
+        .verifying(
+          "Tokens must be unique",
+          t => {
+            val tokens = extractTokenPairs(t).view.flatMap { case (w, b) => Vector(w, b) }.toVector
+            tokens.size == tokens.distinct.size
+          }
+        ),
       SetupForm.api.variant,
       "clock" -> SetupForm.api.clockMapping,
       "rated" -> boolean
@@ -48,24 +56,26 @@ object BulkChallenge {
 
   case class BadToken(token: AccessToken.Id, error: OAuthServer.AuthError)
 
-  case class ScheduledBulkPairing(
-      players: List[(User.ID, User.ID)],
+  case class ScheduledGame(id: Game.ID, white: User.ID, black: User.ID)
+
+  case class ScheduledBulk(
+      games: List[ScheduledGame],
       variant: Variant,
       clock: Clock.Config,
-      rated: Boolean,
+      mode: Mode,
       pairAt: DateTime,
       startClocksAt: DateTime
   )
 }
 
-final class BulkChallengeApi(oauthServer: OAuthServer)(implicit
+final class BulkChallengeApi(oauthServer: OAuthServer, idGenerator: IdGenerator)(implicit
     ec: scala.concurrent.ExecutionContext,
     mat: akka.stream.Materializer
 ) {
 
-  import BulkChallenge._
+  import SetupBulk._
 
-  def apply(data: BulkFormData): Fu[Either[List[BadToken], ScheduledBulkPairing]] =
+  def apply(data: BulkFormData): Fu[Either[List[BadToken], ScheduledBulk]] =
     Source(extractTokenPairs(data.tokens))
       .mapConcat { case (whiteToken, blackToken) =>
         List(whiteToken, blackToken) // flatten now, re-pair later!
@@ -81,24 +91,26 @@ final class BulkChallengeApi(oauthServer: OAuthServer)(implicit
         case (Right(_), Left(bad))         => Left(bad :: Nil)
         case (Right(users), Right(scoped)) => Right(scoped.user.id :: users)
       }
-      .map {
-        _.map {
-          _.reverse
+      .flatMap {
+        case Left(errors) => fuccess(Left(errors.reverse))
+        case Right(allPlayers) =>
+          val pairs = allPlayers.reverse
             .grouped(2)
             .collect { case List(w, b) => (w, b) }
             .toList
-        }.left.map(_.reverse)
-      }
-      .map {
-        _.map { players =>
-          ScheduledBulkPairing(
-            players,
-            data.variant,
-            data.clock,
-            data.rated,
-            DateTime.now,
-            DateTime.now
-          )
-        }
+          idGenerator
+            .games(pairs.size)
+            .map {
+              _.toList zip pairs
+            }
+            .map {
+              _.map { case (id, (w, b)) =>
+                ScheduledGame(id, w, b)
+              }
+            }
+            .dmap {
+              ScheduledBulk(_, data.variant, data.clock, Mode(data.rated), DateTime.now, DateTime.now)
+            }
+            .dmap(Right.apply)
       }
 }