fixes #1557 web_connector: fix dropped messages after network failure (#1808)

* fixes #1557 web_connector: fix dropped messages after network failure

Author: Fabilin

bot/connector-web/README.md

@@ -156,6 +156,17 @@ and attempts to send them if and when the SSE connection is re-established.
 The `tock_web_sse_keepalive_delay` optional property can be used to configure the number of seconds between
 two SSE pings (default: 10).
 
+The `tock_web_sse_messages_ttl_days` optional property can be used to configure the number of days after which
+enqueued messages get deleted. Note: this expiration only works on MongoDB v7.1.0 and up. Set to a
+negative value to disable (default: -1).
+
+The `tock_web_sse_message_queue_max_count` optional property can be used to configure the maximum number of enqueued
+messages in the database. When a new message is enqueued past this limit, the oldest message gets deleted (default: 50000).
+
+The `tock_web_sse_message_queue_max_size_kb` optional property can be used to configure the maximum size in kilobytes
+of the message queue in the database. When a new message is enqueued past this limit, the oldest messages get deleted
+(default: `2 * tock_web_sse_message_queue_max_count`).
+
 #### Push messages
 
 When SSE is enabled, the web connector allows sending push messages through the

bot/connector-web/src/main/kotlin/channel/ChannelCallback.kt

@@ -16,5 +16,6 @@
 package ai.tock.bot.connector.web.channel
 
 import ai.tock.bot.connector.web.WebConnectorResponse
+import io.vertx.core.Future
 
-internal typealias ChannelCallback = (webConnectorResponse: WebConnectorResponse) -> Unit
+internal typealias ChannelCallback = (webConnectorResponse: WebConnectorResponse) -> Future<Void>

bot/connector-web/src/main/kotlin/channel/ChannelEvent.kt

@@ -17,6 +17,7 @@ package ai.tock.bot.connector.web.channel
 
 import ai.tock.bot.connector.web.WebConnectorResponse
 import com.fasterxml.jackson.annotation.JsonValue
+import io.vertx.core.Future
 import java.time.Instant
 import org.litote.kmongo.Id
 import org.litote.kmongo.newId
@@ -42,6 +43,6 @@ internal data class ChannelEvent(
         /**
          * @return `true` if the event has been handled successfully
          */
-        operator fun invoke(event: ChannelEvent): Boolean
+        operator fun invoke(event: ChannelEvent): Future<Boolean>
     }
 }

bot/connector-web/src/main/kotlin/channel/ChannelMongoDAO.kt

@@ -20,11 +20,14 @@ import ai.tock.shared.TOCK_BOT_DATABASE
 import ai.tock.shared.ensureIndex
 import ai.tock.shared.error
 import ai.tock.shared.injector
+import ai.tock.shared.longProperty
 import ai.tock.shared.watch
 import com.github.salomonbrys.kodein.instance
 import com.mongodb.client.MongoCollection
 import com.mongodb.client.MongoDatabase
 import com.mongodb.client.model.CreateCollectionOptions
+import com.mongodb.client.model.IndexOptions
+import java.util.concurrent.TimeUnit
 import mu.KotlinLogging
 import org.litote.kmongo.and
 import org.litote.kmongo.eq
@@ -49,6 +52,10 @@ internal object ChannelMongoDAO : ChannelDAO {
     private fun MongoDatabase.collectionExists(collectionName: String): Boolean =
         listCollectionNames().contains(collectionName)
 
+    private val messageQueueTtl = longProperty("tock_web_sse_message_queue_ttl_days", -1)
+    private val messageQueueMaxCount = longProperty("tock_web_sse_message_queue_max_count", 50000)
+    private val messageQueueMaxSize = longProperty("tock_web_sse_message_queue_max_size_kb", 2 * messageQueueMaxCount)
+
     init {
         if (!database.collectionExists(COLLECTION_NAME)) {
             try {
@@ -57,8 +64,8 @@ internal object ChannelMongoDAO : ChannelDAO {
                         COLLECTION_NAME,
                         CreateCollectionOptions()
                             .capped(true)
-                            .sizeInBytes(100000000)
-                            .maxDocuments(50000)
+                            .sizeInBytes(messageQueueMaxSize * 1000)
+                            .maxDocuments(messageQueueMaxCount)
                     )
             } catch (e: Exception) {
                 logger.error(e)
@@ -68,7 +75,14 @@ internal object ChannelMongoDAO : ChannelDAO {
         webChannelResponseCol = database.getCollection<ChannelEvent>(COLLECTION_NAME)
         try {
             webChannelResponseCol.ensureIndex(ChannelEvent::appId, ChannelEvent::recipientId, ChannelEvent::status)
-            // TODO add an index with TTL on ChannelEvent::enqueuedAt once MongoDB supports it (cf. https://jira.mongodb.org/browse/SERVER-77586)
+            if (messageQueueTtl > 0) {
+                webChannelResponseCol.ensureIndex(
+                    ChannelEvent::enqueuedAt, indexOptions = IndexOptions().expireAfter(
+                        messageQueueTtl,
+                        TimeUnit.DAYS
+                    )
+                )
+            }
         } catch (e: Exception) {
             logger.error(e)
         }
@@ -101,11 +115,19 @@ internal object ChannelMongoDAO : ChannelDAO {
 
     private fun process(event: ChannelEvent, handler: ChannelEvent.Handler) {
         try {
-            if (handler(event)) {
-                webChannelResponseCol.updateOneById(event._id, ChannelEvent::status setTo ChannelEvent.Status.PROCESSED)
-            }
+            handler(event).onComplete({ processed ->
+                if (processed) {
+                    webChannelResponseCol.updateOneById(event._id, ChannelEvent::status setTo ChannelEvent.Status.PROCESSED)
+                }
+            }, { e ->
+                logger.error(e) {
+                    "Failed to send SSE message"
+                }
+            })
         } catch (e: Exception) {
-            logger.error(e)
+            logger.error(e) {
+                "Failed to send SSE message"
+            }
         }
     }
 

bot/connector-web/src/main/kotlin/channel/Channels.kt

@@ -19,6 +19,7 @@ import ai.tock.bot.connector.web.WebConnectorResponse
 import ai.tock.bot.engine.user.PlayerId
 import ai.tock.shared.injector
 import ai.tock.shared.provide
+import io.vertx.core.Future
 import java.util.UUID
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.CopyOnWriteArrayList
@@ -30,9 +31,9 @@ internal class Channels {
 
     init {
         channelDAO.listenChanges { (appId, recipientId, response) ->
-            channelsByUser[recipientId]?.filter { it.appId == appId }?.onEach { channel ->
+            Future.all((channelsByUser[recipientId] ?: emptyList()).filter { it.appId == appId }.map { channel ->
                 channel.onAction(response)
-            }?.isNotEmpty() ?: false
+            }).map { it.list<Unit?>().isNotEmpty() }
         }
     }
 
@@ -43,8 +44,7 @@ internal class Channels {
         val channel = Channel(appId, UUID.randomUUID(), userId, onAction)
         channels.add(channel)
         channelDAO.handleMissedEvents(appId, userId) { (_, _, response) ->
-            channel.onAction(response)
-            true
+            channel.onAction(response).map { true }
         }
         return channel
     }

bot/connector-web/src/test/kotlin/ChannelsTest.kt

@@ -28,6 +28,7 @@ import io.mockk.just
 import io.mockk.mockk
 import io.mockk.runs
 import io.mockk.slot
+import io.vertx.core.Future
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
@@ -66,6 +67,7 @@ internal class ChannelsTest {
         val responses = mutableListOf<WebConnectorResponse>()
         channels.register(appId, recipientId) {
             responses.add(it)
+            Future.succeededFuture()
         }
         assertEquals(expectedMissedResponses, responses)
         expectedNewResponses.forEach {