chore: Adopt V9 Changes #WPB-18787

* Remove MessageCount notification event
* Add Synchronization notification event
* Generate new sync marker everytime the websocket is connected
* Acknowledge synchronization event and add logs

Author: alexandreferris

lib/src/main/kotlin/com/wire/integrations/jvm/client/BackendClient.kt

@@ -30,8 +30,11 @@ import com.wire.integrations.jvm.model.http.client.RegisterClientResponse
 import com.wire.integrations.jvm.model.http.conversation.ConversationResponse
 import com.wire.integrations.jvm.model.http.user.UserResponse
 import io.ktor.client.plugins.websocket.DefaultClientWebSocketSession
+import java.util.UUID
 
 interface BackendClient {
+    fun getCurrentSyncMarker(): UUID?
+
     suspend fun connectWebSocket(handleFrames: suspend (DefaultClientWebSocketSession) -> Unit)
 
     suspend fun getAvailableApiVersions(): ApiVersionResponse

lib/src/main/kotlin/com/wire/integrations/jvm/client/BackendClientDemo.kt

@@ -81,17 +81,29 @@ internal class BackendClientDemo(
     private var cachedFeatures: FeaturesResponse? = null
     private var cachedAccessToken: String? = null
     private var cachedDeviceId: String? = null
+    private var currentSyncMarker: UUID? = null
+
+    override fun getCurrentSyncMarker(): UUID? = currentSyncMarker
 
     override suspend fun connectWebSocket(
         handleFrames: suspend (DefaultClientWebSocketSession) -> Unit
     ) {
         logger.info("Connecting to the webSocket, waiting for events")
 
+        currentSyncMarker = UUID.randomUUID()
         val token = loginUser()
+
+        val url = StringBuilder().apply {
+            append("/$API_VERSION/events")
+            append("?client=$cachedDeviceId")
+            append("&access_token=$token")
+            append("&sync_marker=$currentSyncMarker")
+        }.toString()
+
         httpClient.wss(
             host = IsolatedKoinContext.getApiHost()?.replace("https://", "")
                 ?.replace("-https", "-ssl"),
-            path = "/$API_VERSION/events?client=$cachedDeviceId&access_token=$token"
+            path = url
         ) {
             handleFrames(this)
         }

lib/src/main/kotlin/com/wire/integrations/jvm/client/BackendClientImpl.kt

@@ -39,6 +39,7 @@ import io.ktor.client.request.get
 import io.ktor.client.request.header
 import io.ktor.client.request.post
 import io.ktor.http.HttpHeaders
+import java.util.UUID
 import org.slf4j.LoggerFactory
 
 /**
@@ -47,6 +48,8 @@ import org.slf4j.LoggerFactory
 internal class BackendClientImpl(private val httpClient: HttpClient) : BackendClient {
     private val logger = LoggerFactory.getLogger(this::class.java)
 
+    override fun getCurrentSyncMarker(): UUID? = null
+
     override suspend fun connectWebSocket(
         handleFrames: suspend (DefaultClientWebSocketSession) -> Unit
     ) {

lib/src/main/kotlin/com/wire/integrations/jvm/model/http/ConsumableNotificationResponse.kt

@@ -30,15 +30,15 @@ sealed class ConsumableNotificationResponse {
         @SerialName("data") val data: EventDataDTO
     ) : ConsumableNotificationResponse()
 
-    @Serializable
-    @SerialName("message_count")
-    data class MessageCount(
-        @SerialName("data") val data: NotificationCount
-    ) : ConsumableNotificationResponse()
-
     @Serializable
     @SerialName("notifications_missed")
     data object MissedNotification : ConsumableNotificationResponse()
+
+    @Serializable
+    @SerialName("synchronization")
+    data class SynchronizationNotification(
+        @SerialName("data") val data: SynchronizationDataDTO
+    ) : ConsumableNotificationResponse()
 }
 
 @Serializable
@@ -50,7 +50,9 @@ data class EventDataDTO(
 )
 
 @Serializable
-data class NotificationCount(
-    @SerialName("count")
-    val count: ULong
+data class SynchronizationDataDTO(
+    @SerialName("delivery_tag")
+    val deliveryTag: ULong?,
+    @SerialName("marker_id")
+    val markerId: String
 )

lib/src/main/kotlin/com/wire/integrations/jvm/service/WireTeamEventsListener.kt

@@ -89,15 +89,27 @@ internal class WireTeamEventsListener internal constructor(
                 }
             }
 
-            is ConsumableNotificationResponse.MessageCount -> {
-                logger.info("Websocket back online, ${notification.data.count} events to fetch")
-            }
-
             is ConsumableNotificationResponse.MissedNotification -> {
                 logger.warn("App was offline for too long, missed some notifications")
                 val ackRequest = EventAcknowledgeRequest.notificationMissedAck()
                 ackEvent(ackRequest, session)
             }
+
+            is ConsumableNotificationResponse.SynchronizationNotification -> {
+                notification.data.deliveryTag?.let { deliveryTag ->
+                    val ackRequest = EventAcknowledgeRequest.basicAck(deliveryTag)
+                    ackEvent(ackRequest, session)
+                }
+                val currentSyncMarker = backendClient.getCurrentSyncMarker()
+                if (notification.data.markerId == currentSyncMarker.toString()) {
+                    logger.info("Notifications are up to date since last sync marker.")
+                } else {
+                    logger.debug(
+                        "Skipping sync marker [${notification.data.markerId}], " +
+                            "as it is not valid for this session."
+                    )
+                }
+            }
         }
     }