recompute all ublog ranks on factor change

app/controllers/Ublog.scala

@@ -183,7 +183,7 @@ final class Ublog(env: Env) extends LilaController(env) {
               for {
                 user <- env.user.repo.byId(blog.userId) orFail "Missing blog user!" dmap Suspect
                 _    <- env.ublog.api.setTier(blog.id, tier)
-                _    <- env.ublog.rank.recomputeRankOfAllPosts(blog.id)
+                _    <- env.ublog.rank.recomputeRankOfAllPostsOfBlog(blog.id)
                 _ <- env.mod.logApi
                   .blogTier(lila.report.Mod(me.user), user, blog.id.full, UblogBlog.Tier.name(tier))
               } yield Redirect(urlOfBlog(blog)).flashSuccess

modules/ublog/src/main/Env.scala

@@ -21,7 +21,8 @@ final class Env(
     settingStore: lila.memo.SettingStore.Builder,
     net: NetConfig
 )(implicit
-    ec: scala.concurrent.ExecutionContext
+    ec: scala.concurrent.ExecutionContext,
+    mat: akka.stream.Materializer
 ) {
 
   import net.{ assetBaseUrl, baseUrl }
@@ -31,12 +32,13 @@ final class Env(
   val rankFactorSetting = settingStore[Int](
     "ublogRankFactor",
     default = 5,
-    text = "Ublog rank factor".some
+    text = "Ublog rank factor".some,
+    onSet = _ => rank.recomputeRankOfAllPosts
   ).taggedWith[UblogRankFactor]
 
   val topic = wire[UblogTopicApi]
 
-  val rank = wire[UblogRank]
+  val rank: UblogRank = wire[UblogRank]
 
   val api: UblogApi = wire[UblogApi]
 
@@ -55,7 +57,7 @@ final class Env(
 
   lila.common.Bus.subscribeFun("shadowban") { case lila.hub.actorApi.mod.Shadowban(userId, v) =>
     api.setShadowban(userId, v) >>
-      rank.recomputeRankOfAllPosts(UblogBlog.Id.User(userId))
+      rank.recomputeRankOfAllPostsOfBlog(UblogBlog.Id.User(userId))
     ()
   }
 }

modules/ublog/src/main/UblogRank.scala

@@ -1,10 +1,11 @@
 package lila.ublog
 
-
+import akka.stream.scaladsl._
 import cats.implicits._
 import com.softwaremill.tagging._
 import org.joda.time.DateTime
 import play.api.i18n.Lang
+import reactivemongo.akkastream.cursorProducer
 import reactivemongo.api._
 import scala.concurrent.ExecutionContext
 
@@ -17,7 +18,7 @@ final class UblogRank(
     colls: UblogColls,
     timeline: lila.hub.actors.Timeline,
     factor: SettingStore[Int] @@ UblogRankFactor
-)(implicit ec: ExecutionContext) {
+)(implicit ec: ExecutionContext, mat: akka.stream.Materializer) {
 
   import UblogBsonHandlers._
 
@@ -76,33 +77,45 @@ final class UblogRank(
         }
       }
 
-  def recomputeRankOfAllPosts(blogId: UblogBlog.Id): Funit =
+  def recomputeRankOfAllPostsOfBlog(blogId: UblogBlog.Id): Funit =
     colls.blog.byId[UblogBlog](blogId.full) flatMap {
-      _ ?? { blog =>
-        colls.post
-          .find(
-            $doc("blog" -> blog.id),
-            $doc("topics" -> true, "likes" -> true, "lived" -> true, "language" -> true).some
-          )
-          .cursor[Bdoc]()
-          .list() flatMap { docs =>
-          lila.common.Future.applySequentially(docs) { doc =>
-            (
-              doc.string("_id"),
-              doc.getAsOpt[List[UblogTopic]]("topics"),
-              doc.getAsOpt[UblogPost.Likes]("likes"),
-              doc.getAsOpt[UblogPost.Recorded]("lived"),
-              doc.getAsOpt[Lang]("language")
-            ).tupled ?? { case (id, topics, likes, lived, language) =>
-              colls.post
-                .updateField($id(id), "rank", computeRank(topics, likes, lived.at, language, blog.tier))
-                .void
-            }
-          }
+      _ ?? recomputeRankOfAllPostsOfBlog
+    }
+
+  def recomputeRankOfAllPostsOfBlog(blog: UblogBlog): Funit =
+    colls.post
+      .find(
+        $doc("blog" -> blog.id),
+        $doc("topics" -> true, "likes" -> true, "lived" -> true, "language" -> true).some
+      )
+      .cursor[Bdoc](ReadPreference.secondaryPreferred)
+      .list() flatMap { docs =>
+      lila.common.Future.applySequentially(docs) { doc =>
+        (
+          doc.string("_id"),
+          doc.getAsOpt[List[UblogTopic]]("topics"),
+          doc.getAsOpt[UblogPost.Likes]("likes"),
+          doc.getAsOpt[UblogPost.Recorded]("lived"),
+          doc.getAsOpt[Lang]("language")
+        ).tupled ?? { case (id, topics, likes, lived, language) =>
+          colls.post
+            .updateField($id(id), "rank", computeRank(topics, likes, lived.at, language, blog.tier))
+            .void
         }
       }
     }
 
+  def recomputeRankOfAllPosts: Funit =
+    colls.blog
+      .find($empty)
+      .sort($sort desc "tier")
+      .cursor[UblogBlog](ReadPreference.secondaryPreferred)
+      .documentSource()
+      .mapAsyncUnordered(4)(recomputeRankOfAllPostsOfBlog)
+      .toMat(lila.common.LilaStream.sinkCount)(Keep.right)
+      .run()
+      .map(nb => println(s"Recomputed rank of $nb blogs"))
+
   def computeRank(blog: UblogBlog, post: UblogPost): Option[UblogPost.Rank] =
     post.lived map { lived =>
       computeRank(post.topics, post.likes, lived.at, post.language, blog.tier)