Immediately notify unread messages with a websocket event

app/core/CoreEnv.scala

@@ -37,7 +37,8 @@ final class CoreEnv private (application: Application, val settings: Settings) {
   lazy val message = new lila.message.MessageEnv(
     settings = settings,
     mongodb = mongodb.apply _,
-    userRepo = user.userRepo)
+    userRepo = user.userRepo,
+    notifyUnread = metaHub.notifyUnread)
 
   lazy val wiki = new lila.wiki.WikiEnv(
     settings = settings,

app/message/Api.scala

@@ -11,7 +11,8 @@ final class Api(
     threadRepo: ThreadRepo,
     unreadCache: UnreadCache,
     userRepo: UserRepo,
-    maxPerPage: Int) {
+    maxPerPage: Int,
+    notifyUnread: (String, Int) ⇒ Unit) {
 
   def inbox(me: User, page: Int): Paginator[Thread] = Paginator(
     SalatAdapter(
@@ -27,7 +28,7 @@ final class Api(
     _ ← threadOption.filter(_ isUnReadBy me).fold(
       thread ⇒ for {
         _ ← threadRepo setRead thread
-        _ ← io(unreadCache invalidate me)
+        _ ← updateUser(me)
       } yield (),
       io()
     )
@@ -41,7 +42,7 @@ final class Api(
       invited = data.user)
     for {
       _ ← threadRepo saveIO thread
-      _ ← io(unreadCache invalidate data.user)
+      _ ← updateUser(data.user)
     } yield thread
   }
 
@@ -53,13 +54,14 @@ final class Api(
     for {
       _ ← threadRepo saveIO newThread
       receiver ← userRepo byId (thread receiverOf post)
-      _ ← receiver.fold(
-        r ⇒ io(unreadCache invalidate r),
-        io()
-      )
+      _ ← receiver.fold(updateUser, io())
     } yield thread
   }
 
+  private def updateUser(user: User) = io {
+    notifyUnread(user.id, unreadCache refresh user)
+  }
+
   def deleteThread(id: String, me: User): IO[Unit] = for {
     threadOption ← thread(id, me)
     _ ← threadOption.fold(threadRepo.deleteFor(me), io())

app/message/MessageEnv.scala

@@ -9,7 +9,8 @@ import com.mongodb.casbah.MongoCollection
 final class MessageEnv(
     settings: Settings,
     mongodb: String ⇒ MongoCollection,
-    userRepo: UserRepo) {
+    userRepo: UserRepo,
+    notifyUnread: (String, Int) ⇒ Unit) {
 
   import settings._
 
@@ -21,7 +22,8 @@ final class MessageEnv(
     threadRepo = threadRepo,
     unreadCache = unreadCache,
     userRepo = userRepo,
-    maxPerPage = MessageThreadMaxPerPage)
-  
+    maxPerPage = MessageThreadMaxPerPage,
+    notifyUnread = notifyUnread)
+
   lazy val forms = new DataForm(userRepo)
 }

app/message/UnreadCache.scala

@@ -17,7 +17,8 @@ final class UnreadCache(threadRepo: ThreadRepo) {
       (threadRepo userNbUnread username).unsafePerformIO
     })
 
-  def invalidate(user: User) {
+  def refresh(user: User): Int = {
     cache -= user.id
+    get(user)
   }
 }

app/round/HubMaster.scala

@@ -1,7 +1,7 @@
 package lila
 package round
 
-import socket.{ Broom, Close, GetNbMembers, GetUsernames, NbMembers }
+import socket.{ Broom, Close, GetNbMembers, GetUsernames, NbMembers, SendTo }
 
 import akka.actor._
 import akka.actor.ReceiveTimeout
@@ -28,6 +28,8 @@ final class HubMaster(
 
     case Broom                            ⇒ hubs.values foreach (_ ! Broom)
 
+    case msg @ SendTo(_, _)               ⇒ hubs.values foreach (_ ! msg)
+
     case msg @ GameEvents(gameId, events) ⇒ hubs get gameId foreach (_ forward msg)
 
     case GetHub(gameId: String) ⇒ sender ! {

app/socket/HubActor.scala

@@ -1,8 +1,6 @@
 package lila
 package socket
 
-import core.Global.env // fuck. need it for message unread cache
-
 import akka.actor._
 import play.api.libs.json._
 
@@ -18,18 +16,20 @@ abstract class HubActor[M <: SocketMember](uidTimeout: Int) extends Actor {
   // generic message handler
   def receiveGeneric: Receive = {
 
-    case Ping(uid)     ⇒ ping(uid)
+    case Ping(uid)            ⇒ ping(uid)
 
-    case Broom         ⇒ broom()
+    case Broom                ⇒ broom()
 
     // when a member quits
-    case Quit(uid)     ⇒ quit(uid)
+    case Quit(uid)            ⇒ quit(uid)
 
-    case GetNbMembers  ⇒ sender ! members.size
+    case GetNbMembers         ⇒ sender ! members.size
 
-    case NbMembers(nb) ⇒ pong = makePong(nb)
+    case NbMembers(nb)        ⇒ pong = makePong(nb)
 
-    case GetUsernames  ⇒ sender ! usernames
+    case GetUsernames         ⇒ sender ! usernames
+
+    case SendTo(userId, msg) ⇒ sendTo(userId, msg)
   }
 
   def receive = receiveSpecific orElse receiveGeneric
@@ -46,11 +46,11 @@ abstract class HubActor[M <: SocketMember](uidTimeout: Int) extends Actor {
 
   def ping(uid: String) {
     setAlive(uid)
-    member(uid) foreach { m ⇒
-      m.channel push m.username.fold(
-        u ⇒ pong ++ JsObject(Seq("m" -> JsNumber(unreadMessages(u)))),
-        pong)
-    }
+    member(uid) foreach (_.channel push pong)
+  }
+
+  def sendTo(userId: String, msg: JsObject) {
+    memberByUserId(userId) foreach (_.channel push msg)
   }
 
   def broom() {
@@ -82,8 +82,10 @@ abstract class HubActor[M <: SocketMember](uidTimeout: Int) extends Actor {
 
   def member(uid: String): Option[M] = members get uid
 
-  def usernames: Iterable[String] = members.values.map(_.username).flatten
+  def memberByUserId(userId: String): Option[M] = {
+    val someId = Some(userId)
+    members.values find (_.userId == someId)
+  }
 
-  private def unreadMessages(username: String): Int = 
-    env.message.unreadCache get username
+  def usernames: Iterable[String] = members.values.map(_.username).flatten
 }

app/socket/MetaHub.scala

@@ -8,6 +8,7 @@ import akka.util.duration._
 import akka.util.{ Duration, Timeout }
 import play.api.libs.concurrent._
 import play.api.Play.current
+import play.api.libs.json._
 
 final class MetaHub(hubs: List[ActorRef]) {
 
@@ -22,4 +23,9 @@ final class MetaHub(hubs: List[ActorRef]) {
     Future.traverse(hubs) { hub ⇒
       hub ? message mapTo m
     }
+  
+  def notifyUnread(userId: String, nb: Int) = this ! SendTo(
+    userId,
+    JsObject(Seq("t" -> JsString("nbm"), "d" -> JsNumber(nb)))
+  )
 }

app/socket/actorApi.scala

@@ -1,9 +1,13 @@
 package lila
 package socket
 
+import play.api.libs.json.JsObject
+
 trait SocketMember {
   val channel: Channel
   val username: Option[String]
+
+  lazy val userId: Option[String] = username map (_.toLowerCase)
 }
 case object Close
 case object GetUsernames
@@ -12,3 +16,5 @@ case class NbMembers(nb: Int)
 case class Ping(uid: String)
 case object Broom
 case class Quit(uid: String)
+
+case class SendTo(userId: String, message: JsObject)

public/javascripts/ctrl.js

@@ -12,6 +12,10 @@ var lichess = {
       n: function(e) {
         var $tag = $('#nb_connected_players');
         $tag.html($tag.html().replace(/\d+/, e)).removeClass('none');
+      },
+      nbm: function(e) {
+        var $tag = $('#nb_messages');
+        $tag.text(e).toggleClass("unread", e > 0);
       }
     },
     options: {
@@ -137,8 +141,6 @@ $(function() {
     $('#top').find('div.security').removeClass('show_signin_form');
   });
 
-  $('#lichess_message input[value=""]:first, #fos_user_registration_form_username').focus();
-
   $('#lichess_translation_form_code').change(function() {
     if ("0" != $(this).val()) {
       location.href = $(this).closest('form').attr('data-change-url').replace(/__/, $(this).val());