Group keys and message response improvement

libsession/src/main/java/org/session/libsession/messaging/groups/RemoveGroupMemberHandler.kt

@@ -143,7 +143,7 @@ class RemoveGroupMemberHandler @Inject constructor(
             pendingRemovals to (calls as List<SnodeAPI.SnodeBatchRequestInfo>)
         }
 
-        if (batchCalls.isEmpty()) {
+        if (pendingRemovals.isEmpty() || batchCalls.isEmpty()) {
             return
         }
 

libsession/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/ClosedGroupPoller.kt

@@ -18,10 +18,12 @@ import org.session.libsession.messaging.jobs.MessageReceiveParameters
 import org.session.libsession.messaging.messages.Destination
 import org.session.libsession.snode.RawResponse
 import org.session.libsession.snode.SnodeAPI
+import org.session.libsession.snode.SnodeClock
 import org.session.libsession.snode.model.RetrieveMessageResponse
 import org.session.libsession.snode.utilities.await
 import org.session.libsession.utilities.ConfigFactoryProtocol
 import org.session.libsession.utilities.ConfigMessage
+import org.session.libsession.utilities.getClosedGroup
 import org.session.libsignal.database.LokiAPIDatabaseProtocol
 import org.session.libsignal.utilities.AccountId
 import org.session.libsignal.utilities.Log
@@ -38,6 +40,7 @@ class ClosedGroupPoller(
     private val groupManagerV2: GroupManagerV2,
     private val storage: StorageProtocol,
     private val lokiApiDatabase: LokiAPIDatabaseProtocol,
+    private val clock: SnodeClock,
 ) {
     companion object {
         private const val POLL_INTERVAL = 3_000L
@@ -112,9 +115,7 @@ class ClosedGroupPoller(
             }
         }
 
-        val adminKey = requireNotNull(configFactoryProtocol.withUserConfigs {
-            it.userGroups.getClosedGroup(closedGroupSessionId.hexString)
-        }) {
+        val adminKey = requireNotNull(configFactoryProtocol.getClosedGroup(closedGroupSessionId)) {
             "Group doesn't exist"
         }.adminKey
 
@@ -135,7 +136,7 @@ class ClosedGroupPoller(
                     maxSize = null,
                 ),
                 RetrieveMessageResponse::class.java
-            )
+            ).messages.filterNotNull()
         }
 
         if (configHashesToExtends.isNotEmpty() && adminKey != null) {
@@ -146,7 +147,7 @@ class ClosedGroupPoller(
                     SnodeAPI.buildAuthenticatedAlterTtlBatchRequest(
                         messageHashes = configHashesToExtends.toList(),
                         auth = groupAuth,
-                        newExpiry = SnodeAPI.nowWithOffset + 14.days.inWholeMilliseconds,
+                        newExpiry = clock.currentTimeMills() + 14.days.inWholeMilliseconds,
                         extend = true
                     ),
                 )
@@ -191,7 +192,7 @@ class ClosedGroupPoller(
                         maxSize = null,
                     ),
                     responseType = RetrieveMessageResponse::class.java
-                )
+                ).messages.filterNotNull()
             }
         }
 
@@ -238,23 +239,27 @@ class ClosedGroupPoller(
     }
 
     private fun RetrieveMessageResponse.Message.toConfigMessage(): ConfigMessage {
-        return ConfigMessage(hash, data, timestamp)
+        return ConfigMessage(hash, data, timestamp ?: clock.currentTimeMills())
     }
 
-    private fun saveLastMessageHash(snode: Snode, body: RetrieveMessageResponse, namespace: Int) {
-        if (body.messages.isNotEmpty()) {
+    private fun saveLastMessageHash(
+        snode: Snode,
+        messages: List<RetrieveMessageResponse.Message>,
+        namespace: Int
+    ) {
+        if (messages.isNotEmpty()) {
             lokiApiDatabase.setLastMessageHashValue(
                 snode = snode,
                 publicKey = closedGroupSessionId.hexString,
-                newValue = body.messages.last().hash,
+                newValue = messages.last().hash,
                 namespace = namespace
             )
         }
     }
 
-    private suspend fun handleRevoked(body: RetrieveMessageResponse) {
-        body.messages.forEach { msg ->
-            val decoded = configFactoryProtocol.maybeDecryptForUser(
+    private suspend fun handleRevoked(messages: List<RetrieveMessageResponse.Message>) {
+        messages.forEach { msg ->
+            val decoded = configFactoryProtocol.decryptForUser(
                 msg.data,
                 Sodium.KICKED_DOMAIN,
                 closedGroupSessionId,
@@ -284,26 +289,26 @@ class ClosedGroupPoller(
     }
 
     private fun handleGroupConfigMessages(
-        keysResponse: RetrieveMessageResponse,
-        infoResponse: RetrieveMessageResponse,
-        membersResponse: RetrieveMessageResponse
+        keysResponse: List<RetrieveMessageResponse.Message>,
+        infoResponse: List<RetrieveMessageResponse.Message>,
+        membersResponse: List<RetrieveMessageResponse.Message>
     ) {
-        if (keysResponse.messages.isEmpty() && infoResponse.messages.isEmpty() && membersResponse.messages.isEmpty()) {
+        if (keysResponse.isEmpty() && infoResponse.isEmpty() && membersResponse.isEmpty()) {
             return
         }
 
         Log.d(
             TAG, "Handling group config messages(" +
-                    "info = ${infoResponse.messages.size}, " +
-                    "keys = ${keysResponse.messages.size}, " +
-                    "members = ${membersResponse.messages.size})"
+                    "info = ${infoResponse.size}, " +
+                    "keys = ${keysResponse.size}, " +
+                    "members = ${membersResponse.size})"
         )
 
         configFactoryProtocol.mergeGroupConfigMessages(
             groupId = closedGroupSessionId,
-            keys = keysResponse.messages.map { it.toConfigMessage() },
-            info = infoResponse.messages.map { it.toConfigMessage() },
-            members = membersResponse.messages.map { it.toConfigMessage() },
+            keys = keysResponse.map { it.toConfigMessage() },
+            info = infoResponse.map { it.toConfigMessage() },
+            members = membersResponse.map { it.toConfigMessage() },
         )
     }
 
@@ -314,6 +319,7 @@ class ClosedGroupPoller(
                 snode = snode,
                 publicKey = closedGroupSessionId.hexString,
                 decrypt = it.groupKeys::decrypt,
+                namespace = Namespace.CLOSED_GROUP_MESSAGES(),
             )
         }
 
@@ -331,7 +337,7 @@ class ClosedGroupPoller(
         }
 
         if (messages.isNotEmpty()) {
-            Log.d(TAG, "namespace for messages rx count: ${messages.size}")
+            Log.d(TAG, "Received and handled ${messages.size} group messages")
         }
     }
 }

libsession/src/main/java/org/session/libsession/snode/SnodeAPI.kt

@@ -48,7 +48,6 @@ import org.session.libsignal.utilities.Snode
 import org.session.libsignal.utilities.prettifiedDescription
 import org.session.libsignal.utilities.retryIfNeeded
 import org.session.libsignal.utilities.retryWithUniformInterval
-import java.util.Date
 import java.util.Locale
 import kotlin.collections.component1
 import kotlin.collections.component2
@@ -989,7 +988,7 @@ object SnodeAPI {
                     }
                     else Pair(MessageWrapper.unwrap(data), rawMessageAsJSON["hash"] as? String)
                 } catch (e: Exception) {
-                    Log.d("Loki", "Failed to unwrap data for message: ${rawMessage.prettifiedDescription()}.")
+                    Log.d("Loki", "Failed to unwrap data for message: ${rawMessage.prettifiedDescription()}.", e)
                     null
                 }
             } else {

libsession/src/main/java/org/session/libsession/snode/model/MessageResponses.kt

@@ -1,7 +1,11 @@
 package org.session.libsession.snode.model
 
+import android.util.Base64
 import com.fasterxml.jackson.annotation.JsonCreator
 import com.fasterxml.jackson.annotation.JsonProperty
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize
+import com.fasterxml.jackson.databind.util.StdConverter
 
 data class StoreMessageResponse @JsonCreator constructor(
     @JsonProperty("hash") val hash: String,
@@ -9,12 +13,29 @@ data class StoreMessageResponse @JsonCreator constructor(
 )
 
 class RetrieveMessageResponse @JsonCreator constructor(
-    @JsonProperty("messages") val messages: List<Message>,
+    @JsonProperty("messages")
+    // Apply converter to the element so that if one of the message fails to deserialize, it will
+    // be a null value instead of failing the whole list.
+    @JsonDeserialize(contentConverter = RetrieveMessageConverter::class)
+    val messages: List<Message?>,
 ) {
-    class Message @JsonCreator constructor(
-        @JsonProperty("hash") val hash: String,
-        @JsonProperty("t") val timestamp: Long,
-        // Jackson is able to deserialize byte arrays from base64 strings
-        @JsonProperty("data") val data: ByteArray,
+    class Message(
+        val hash: String,
+        val timestamp: Long?,
+        val data: ByteArray,
     )
+}
+
+internal class RetrieveMessageConverter : StdConverter<JsonNode, RetrieveMessageResponse.Message?>() {
+    override fun convert(value: JsonNode?): RetrieveMessageResponse.Message? {
+        value ?: return null
+
+        val hash = value.get("hash")?.asText()?.takeIf { it.isNotEmpty() } ?: return null
+        val timestamp = value.get("t")?.asLong()?.takeIf { it > 0 }
+        val data = runCatching {
+            Base64.decode(value.get("data")?.asText().orEmpty(), Base64.DEFAULT)
+        }.getOrNull() ?: return null
+
+        return RetrieveMessageResponse.Message(hash, timestamp, data)
+    }
 }

libsession/src/main/java/org/session/libsession/utilities/ConfigFactoryProtocol.kt

@@ -3,7 +3,6 @@ package org.session.libsession.utilities
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.first
-import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.withTimeoutOrNull
 import network.loki.messenger.libsession_util.MutableConfig
@@ -48,9 +47,9 @@ interface ConfigFactoryProtocol {
     fun getGroupAuth(groupId: AccountId): SwarmAuth?
     fun removeGroup(groupId: AccountId)
 
-    fun maybeDecryptForUser(encoded: ByteArray,
-                            domain: String,
-                            closedGroupSessionId: AccountId): ByteArray?
+    fun decryptForUser(encoded: ByteArray,
+                       domain: String,
+                       closedGroupSessionId: AccountId): ByteArray?
 
     fun mergeGroupConfigMessages(
         groupId: AccountId,