persist round versioned events to ease deployments - fixes #570

modules/hub/src/main/ActorMap.scala

@@ -6,7 +6,6 @@ import actorApi.map._
 import akka.actor._
 import akka.pattern.{ ask, pipe }
 import makeTimeout.short
-import scalaz.Monoid
 
 trait ActorMap extends Actor {
 

modules/round/src/main/Env.scala

@@ -33,7 +33,6 @@ final class Env(
     scheduler: lila.common.Scheduler) {
 
   private val settings = new {
-    val MessageTtl = config duration "message.ttl"
     val UidTimeout = config duration "uid.timeout"
     val PlayerDisconnectTimeout = config duration "player.disconnect.timeout"
     val PlayerRagequitTimeout = config duration "player.ragequit.timeout"
@@ -48,10 +47,11 @@ final class Env(
     val CasualOnly = config getBoolean "casual_only"
     val ActiveTtl = config duration "active.ttl"
     val CollectionNote = config getString "collection.note"
+    val CollectionHistory = config getString "collection.history"
   }
   import settings._
 
-  lazy val history = () => new History(ttl = MessageTtl)
+  lazy val eventHistory = History(db(CollectionHistory)) _
 
   val roundMap = system.actorOf(Props(new lila.hub.ActorMap {
     def mkActor(id: String) = new Round(
@@ -80,7 +80,7 @@ final class Env(
       Props(new lila.socket.SocketHubActor[Socket] {
         def mkActor(id: String) = new Socket(
           gameId = id,
-          history = history(),
+          history = eventHistory(id),
           lightUser = lightUser,
           uidTimeout = UidTimeout,
           socketTimeout = SocketTimeout,

modules/round/src/main/History.scala

@@ -4,38 +4,74 @@ import scala.concurrent.duration.Duration
 
 import actorApi._
 import akka.actor._
+import reactivemongo.bson._
 
+import lila.db.Types.Coll
 import lila.game.Event
 import lila.socket.actorApi.GetVersion
 
-private[round] final class History(ttl: Duration) {
+/**
+ * NOT THREAD SAFE
+ * Designed for use within a sequential actor
+ */
+private[round] final class History(
+    load: Fu[VersionedEvents],
+    persist: VersionedEvents => Unit) {
 
-  private var version = 0
-  private val events = lila.memo.Builder.expiry[Int, VersionedEvent](ttl)
+  // private var version = 0
+  private var events: VersionedEvents = _
 
-  def getVersion = version
+  // TODO optimize
+  def getVersion: Int = {
+    waitForLoadedEvents
+    events.headOption.??(_.version)
+  }
 
   // none if version asked is > to history version
   // none if an event is missing (asked too old version)
-  def getEventsSince(v: Int): Option[List[VersionedEvent]] =
+  def getEventsSince(v: Int): Option[List[VersionedEvent]] = {
+    waitForLoadedEvents
+    val version = getVersion
     if (v > version) None
     else if (v == version) Some(Nil)
-    else {
-      val events = (v + 1 to version).toList flatMap get
-      (events.size == (version - v)) option events
-    }
-
-  def addEvents(xs: List[Event]) = xs map { e =>
-    version = version + 1
-    VersionedEvent(
-      version = version,
-      typ = e.typ,
-      data = e.data,
-      only = e.only,
-      owner = e.owner,
-      watcher = e.watcher,
-      troll = e.troll) ~ { events.put(version, _) }
+    else events.filter(_.version > v).some
   }
 
-  private def get(v: Int): Option[VersionedEvent] = Option(events getIfPresent v)
+  def addEvents(xs: List[Event]): VersionedEvents = {
+    waitForLoadedEvents
+    val vevs = xs.foldLeft(List.empty[VersionedEvent] -> getVersion) {
+      case ((vevs, v), e) => (VersionedEvent(e, v + 1) :: vevs, v + 1)
+    }._1
+    events = (vevs ::: events) take History.size
+    vevs.reverse ~ persist
+  }
+
+  private def waitForLoadedEvents {
+    if (events == null) events = load.await.reverse
+  }
+}
+
+private[round] object History {
+
+  val size = 30
+
+  def apply(coll: Coll)(gameId: String): History = new History(
+    load = load(coll, gameId),
+    persist = persist(coll, gameId) _)
+
+  private def load(coll: Coll, gameId: String): Fu[VersionedEvents] =
+    coll.find(BSONDocument("_id" -> gameId)).one[BSONDocument].map {
+      _.flatMap(_.getAs[VersionedEvents]("e")) | Nil
+    }
+
+  private def persist(coll: Coll, gameId: String)(vevs: List[VersionedEvent]) {
+    coll.uncheckedUpdate(
+      BSONDocument("_id" -> gameId),
+      BSONDocument("$push" -> BSONDocument(
+        "e" -> BSONDocument(
+          "$each" -> vevs,
+          "$slice" -> -History.size))),
+      upsert = true
+    )
+  }
 }

modules/round/src/main/VersionedEvent.scala

@@ -26,4 +26,40 @@ case class VersionedEvent(
     else if (owner && m.watcher) false
     else if (troll && !m.troll) false
     else only.fold(true)(_ == m.color)
+
+  override def toString = s"Event $version $typ"
+}
+
+private[round] object VersionedEvent {
+
+  def apply(e: Event, v: Int): VersionedEvent = VersionedEvent(
+    version = v,
+    typ = e.typ,
+    data = e.data,
+    only = e.only,
+    owner = e.owner,
+    watcher = e.watcher,
+    troll = e.troll)
+
+  import lila.db.BSON
+  import reactivemongo.bson._
+
+  implicit val versionedEventHandler = new BSON[VersionedEvent] {
+    def reads(r: BSON.Reader) = VersionedEvent(
+      version = r int "v",
+      typ = r str "t",
+      data = r.strO("d").fold[JsValue](JsNull)(Json.parse),
+      only = r boolO "o" map Color.apply,
+      owner = r boolD "ow",
+      watcher = r boolD "w",
+      troll = r boolD "r")
+    def writes(w: BSON.Writer, o: VersionedEvent) = BSONDocument(
+      "v" -> o.version,
+      "t" -> o.typ,
+      "d" -> (o.data != JsNull).option(Json stringify o.data),
+      "o" -> o.only.map(_.white),
+      "ow" -> w.boolO(o.owner),
+      "w" -> w.boolO(o.watcher),
+      "t" -> w.boolO(o.troll))
+  }
 }

modules/round/src/main/package.scala

@@ -6,6 +6,8 @@ import lila.socket.WithSocket
 package object round extends PackageObject with WithPlay with WithSocket {
 
   private[round] type Events = List[Event]
+
+  private[round] type VersionedEvents = List[VersionedEvent]
 }
 
 package round {