Merge pull request #1069 from mpretty-cyro/feature/performance-improvements

Performance improvements

libsession/src/main/java/org/session/libsession/database/MessageDataProvider.kt

@@ -20,7 +20,9 @@ interface MessageDataProvider {
      * @return pair of sms or mms table-specific ID and whether it is in SMS table
      */
     fun getMessageID(serverId: Long, threadId: Long): Pair<Long, Boolean>?
+    fun getMessageIDs(serverIDs: List<Long>, threadID: Long): Pair<List<Long>, List<Long>>
     fun deleteMessage(messageID: Long, isSms: Boolean)
+    fun deleteMessages(messageIDs: List<Long>, threadId: Long, isSms: Boolean)
     fun updateMessageAsDeleted(timestamp: Long, author: String)
     fun getServerHashForMessage(messageID: Long): String?
     fun getDatabaseAttachment(attachmentId: Long): DatabaseAttachment?

libsession/src/main/java/org/session/libsession/database/StorageProtocol.kt

@@ -16,6 +16,7 @@ import org.session.libsession.messaging.messages.visible.Reaction
 import org.session.libsession.messaging.messages.visible.VisibleMessage
 import org.session.libsession.messaging.open_groups.GroupMember
 import org.session.libsession.messaging.open_groups.OpenGroup
+import org.session.libsession.messaging.open_groups.OpenGroupApi
 import org.session.libsession.messaging.sending_receiving.attachments.AttachmentId
 import org.session.libsession.messaging.sending_receiving.attachments.DatabaseAttachment
 import org.session.libsession.messaging.sending_receiving.data_extraction.DataExtractionNotificationInfoMessage
@@ -66,7 +67,7 @@ interface StorageProtocol {
     fun getAllOpenGroups(): Map<Long, OpenGroup>
     fun updateOpenGroup(openGroup: OpenGroup)
     fun getOpenGroup(threadId: Long): OpenGroup?
-    fun addOpenGroup(urlAsString: String)
+    fun addOpenGroup(urlAsString: String): OpenGroupApi.RoomInfo?
     fun onOpenGroupAdded(server: String)
     fun hasBackgroundGroupAddJob(groupJoinUrl: String): Boolean
     fun setOpenGroupServerMessageID(messageID: Long, serverID: Long, threadID: Long, isSms: Boolean)
@@ -80,6 +81,7 @@ interface StorageProtocol {
     // Open Group Metadata
     fun updateTitle(groupID: String, newValue: String)
     fun updateProfilePicture(groupID: String, newValue: ByteArray)
+    fun hasDownloadedProfilePicture(groupID: String): Boolean
     fun setUserCount(room: String, server: String, newValue: Int)
 
     // Last Message Server ID

libsession/src/main/java/org/session/libsession/messaging/file_server/FileServerApi.kt

@@ -77,7 +77,11 @@ object FileServerApi {
             OnionRequestAPI.sendOnionRequest(requestBuilder.build(), server, serverPublicKey).map {
                 it.body ?: throw Error.ParsingFailed
             }.fail { e ->
-                Log.e("Loki", "File server request failed.", e)
+                when (e) {
+                    // No need for the stack trace for HTTP errors
+                    is HTTP.HTTPRequestFailedException -> Log.e("Loki", "File server request failed due to error: ${e.message}")
+                    else -> Log.e("Loki", "File server request failed", e)
+                }
             }
         } else {
             Promise.ofFail(IllegalStateException("It's currently not allowed to send non onion routed requests."))

libsession/src/main/java/org/session/libsession/messaging/jobs/BackgroundGroupAddJob.kt

@@ -41,15 +41,10 @@ class BackgroundGroupAddJob(val joinUrl: String): Job {
             }
             // get image
             storage.setOpenGroupPublicKey(openGroup.server, openGroup.serverPublicKey)
-            val (capabilities, info) = OpenGroupApi.getCapabilitiesAndRoomInfo(openGroup.room, openGroup.server, false).get()
-            storage.setServerCapabilities(openGroup.server, capabilities.capabilities)
-            val imageId = info.imageId
-            storage.addOpenGroup(openGroup.joinUrl())
+            val info = storage.addOpenGroup(openGroup.joinUrl())
+            val imageId = info?.imageId
             if (imageId != null) {
-                val bytes = OpenGroupApi.downloadOpenGroupProfilePicture(openGroup.server, openGroup.room, imageId).get()
-                val groupId = GroupUtil.getEncodedOpenGroupID("${openGroup.server}.${openGroup.room}".toByteArray())
-                storage.updateProfilePicture(groupId, bytes)
-                storage.updateTimestampUpdated(groupId, System.currentTimeMillis())
+                JobQueue.shared.add(GroupAvatarDownloadJob(openGroup.room, openGroup.server))
             }
             Log.d(KEY, "onOpenGroupAdded(${openGroup.server})")
             storage.onOpenGroupAdded(openGroup.server)

libsession/src/main/java/org/session/libsession/messaging/jobs/BatchMessageReceiveJob.kt

@@ -94,12 +94,23 @@ class BatchMessageReceiveJob(
                         threadMap[threadID]!! += parsedParams
                     }
                 } catch (e: Exception) {
-                    Log.e(TAG, "Couldn't receive message.", e)
-                    if (e is MessageReceiver.Error && !e.isRetryable) {
-                        Log.e(TAG, "Message failed permanently",e)
-                    } else {
-                        Log.e(TAG, "Message failed",e)
-                        failures += messageParameters
+                    when (e) {
+                        is MessageReceiver.Error.DuplicateMessage, MessageReceiver.Error.SelfSend -> {
+                            Log.i(TAG, "Couldn't receive message, failed with error: ${e.message}")
+                        }
+                        is MessageReceiver.Error -> {
+                            if (!e.isRetryable) {
+                                Log.e(TAG, "Couldn't receive message, failed permanently", e)
+                            }
+                            else {
+                                Log.e(TAG, "Couldn't receive message, failed", e)
+                                failures += messageParameters
+                            }
+                        }
+                        else -> {
+                            Log.e(TAG, "Couldn't receive message, failed", e)
+                            failures += messageParameters
+                        }
                     }
                 }
             }

libsession/src/main/java/org/session/libsession/messaging/jobs/GroupAvatarDownloadJob.kt

@@ -14,10 +14,9 @@ class GroupAvatarDownloadJob(val room: String, val server: String) : Job {
 
     override fun execute() {
         val storage = MessagingModuleConfiguration.shared.storage
+        val imageId = storage.getOpenGroup(room, server)?.imageId ?: return
         try {
-            val info = OpenGroupApi.getRoomInfo(room, server).get()
-            val imageId = info.imageId ?: return
-            val bytes = OpenGroupApi.downloadOpenGroupProfilePicture(server, info.token, imageId).get()
+            val bytes = OpenGroupApi.downloadOpenGroupProfilePicture(server, room, imageId).get()
             val groupId = GroupUtil.getEncodedOpenGroupID("$server.$room".toByteArray())
             storage.updateProfilePicture(groupId, bytes)
             storage.updateTimestampUpdated(groupId, System.currentTimeMillis())

libsession/src/main/java/org/session/libsession/messaging/jobs/JobQueue.kt

@@ -26,7 +26,7 @@ class JobQueue : JobDelegate {
     private val jobTimestampMap = ConcurrentHashMap<Long, AtomicInteger>()
     private val rxDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
     private val rxMediaDispatcher = Executors.newFixedThreadPool(4).asCoroutineDispatcher()
-    private val openGroupDispatcher = Executors.newCachedThreadPool().asCoroutineDispatcher()
+    private val openGroupDispatcher = Executors.newFixedThreadPool(8).asCoroutineDispatcher()
     private val txDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
     private val scope = CoroutineScope(Dispatchers.Default) + SupervisorJob()
     private val queue = Channel<Job>(UNLIMITED)

libsession/src/main/java/org/session/libsession/messaging/jobs/MessageSendJob.kt

@@ -11,6 +11,7 @@ import org.session.libsession.messaging.messages.visible.VisibleMessage
 import org.session.libsession.messaging.sending_receiving.MessageSender
 import org.session.libsession.messaging.utilities.Data
 import org.session.libsession.snode.OnionRequestAPI
+import org.session.libsignal.utilities.HTTP
 import org.session.libsignal.utilities.Log
 
 class MessageSendJob(val message: Message, val destination: Destination) : Job {
@@ -67,14 +68,25 @@ class MessageSendJob(val message: Message, val destination: Destination) : Job {
         val promise = MessageSender.send(this.message, this.destination).success {
             this.handleSuccess()
         }.fail { exception ->
-            Log.e(TAG, "Couldn't send message due to error: $exception.")
-            if (exception is MessageSender.Error) {
-                if (!exception.isRetryable) { this.handlePermanentFailure(exception) }
+            var logStacktrace = true
+
+            when (exception) {
+                // No need for the stack trace for HTTP errors
+                is HTTP.HTTPRequestFailedException -> {
+                    logStacktrace = false
+
+                    if (exception.statusCode == 429) { this.handlePermanentFailure(exception) }
+                    else { this.handleFailure(exception) }
+                }
+                is MessageSender.Error -> {
+                    if (!exception.isRetryable) { this.handlePermanentFailure(exception) }
+                    else { this.handleFailure(exception) }
+                }
+                else -> this.handleFailure(exception)
             }
-            if (exception is OnionRequestAPI.HTTPRequestFailedAtDestinationException && exception.statusCode == 429) {
-                this.handlePermanentFailure(exception)
-            }
-            this.handleFailure(exception)
+
+            if (logStacktrace) { Log.e(TAG, "Couldn't send message due to error", exception) }
+            else { Log.e(TAG, "Couldn't send message due to error: ${exception.message}") }
         }
         try {
             promise.get()

libsession/src/main/java/org/session/libsession/messaging/jobs/OpenGroupDeleteJob.kt

@@ -23,14 +23,27 @@ class OpenGroupDeleteJob(private val messageServerIds: LongArray, private val th
         val dataProvider = MessagingModuleConfiguration.shared.messageDataProvider
         val numberToDelete = messageServerIds.size
         Log.d(TAG, "Deleting $numberToDelete messages")
-        var numberDeleted = 0
-        messageServerIds.forEach { serverId ->
-            val (messageId, isSms) = dataProvider.getMessageID(serverId, threadId) ?: return@forEach
-            dataProvider.deleteMessage(messageId, isSms)
-            numberDeleted++
+
+        // FIXME: This entire process should probably run in a transaction (with the attachment deletion happening only if it succeeded)
+        try {
+            val messageIds = dataProvider.getMessageIDs(messageServerIds.toList(), threadId)
+
+            // Delete the SMS messages
+            if (messageIds.first.isNotEmpty()) {
+                dataProvider.deleteMessages(messageIds.first, threadId, true)
+            }
+
+            // Delete the MMS messages
+            if (messageIds.second.isNotEmpty()) {
+                dataProvider.deleteMessages(messageIds.second, threadId, false)
+            }
+
+            Log.d(TAG, "Deleted ${messageIds.first.size + messageIds.second.size} messages successfully")
+            delegate?.handleJobSucceeded(this)
+        }
+        catch (e: Exception) {
+            delegate?.handleJobFailed(this, e)
         }
-        Log.d(TAG, "Deleted $numberDeleted messages successfully")
-        delegate?.handleJobSucceeded(this)
     }
 
     override fun serialize(): Data = Data.Builder()

libsession/src/main/java/org/session/libsession/messaging/open_groups/OpenGroup.kt

@@ -11,15 +11,17 @@ data class OpenGroup(
     val id: String,
     val name: String,
     val publicKey: String,
+    val imageId: String?,
     val infoUpdates: Int,
 ) {
 
-    constructor(server: String, room: String, name: String, infoUpdates: Int, publicKey: String) : this(
+    constructor(server: String, room: String, publicKey: String, name: String, imageId: String?, infoUpdates: Int) : this(
         server = server,
         room = room,
         id = "$server.$room",
         name = name,
         publicKey = publicKey,
+        imageId = imageId,
         infoUpdates = infoUpdates,
     )
 
@@ -31,11 +33,12 @@ data class OpenGroup(
                 if (!json.has("room")) return null
                 val room = json.get("room").asText().toLowerCase(Locale.US)
                 val server = json.get("server").asText().toLowerCase(Locale.US)
-                val displayName = json.get("displayName").asText()
                 val publicKey = json.get("publicKey").asText()
+                val displayName = json.get("displayName").asText()
+                val imageId = json.get("imageId")?.asText()
                 val infoUpdates = json.get("infoUpdates")?.asText()?.toIntOrNull() ?: 0
                 val capabilities = json.get("capabilities")?.asText()?.split(",") ?: emptyList()
-                OpenGroup(server, room, displayName, infoUpdates, publicKey)
+                OpenGroup(server = server, room = room, name = displayName, publicKey = publicKey, imageId = imageId, infoUpdates = infoUpdates)
             } catch (e: Exception) {
                 Log.w("Loki", "Couldn't parse open group from JSON: $jsonAsString.", e);
                 null
@@ -53,11 +56,12 @@ data class OpenGroup(
         }
     }
 
-    fun toJson(): Map<String,String> = mapOf(
+    fun toJson(): Map<String,String?> = mapOf(
         "room" to room,
         "server" to server,
-        "displayName" to name,
         "publicKey" to publicKey,
+        "displayName" to name,
+        "imageId" to imageId,
         "infoUpdates" to infoUpdates.toString(),
     )
 

libsession/src/main/java/org/session/libsession/messaging/open_groups/OpenGroupApi.kt

@@ -91,7 +91,7 @@ object OpenGroupApi {
         val created: Long = 0,
         val activeUsers: Int = 0,
         val activeUsersCutoff: Int = 0,
-        val imageId: Long? = null,
+        val imageId: String? = null,
         val pinnedMessages: List<PinnedMessage> = emptyList(),
         val admin: Boolean = false,
         val globalAdmin: Boolean = false,
@@ -148,7 +148,7 @@ object OpenGroupApi {
     )
 
     enum class Capability {
-        BLIND, REACTIONS
+        SOGS, BLIND, REACTIONS
     }
 
     @JsonNaming(PropertyNamingStrategy.SnakeCaseStrategy::class)
@@ -337,7 +337,7 @@ object OpenGroupApi {
                     .plus(request.verb.rawValue.toByteArray())
                     .plus("/${request.endpoint.value}".toByteArray())
                     .plus(bodyHash)
-                if (serverCapabilities.contains(Capability.BLIND.name.lowercase())) {
+                if (serverCapabilities.isEmpty() || serverCapabilities.contains(Capability.BLIND.name.lowercase())) {
                     SodiumUtilities.blindedKeyPair(publicKey, ed25519KeyPair)?.let { keyPair ->
                         pubKey = SessionId(
                             IdPrefix.BLINDED,
@@ -383,7 +383,11 @@ object OpenGroupApi {
             }
             return if (request.useOnionRouting) {
                 OnionRequestAPI.sendOnionRequest(requestBuilder.build(), request.server, publicKey).fail { e ->
-                    Log.e("SOGS", "Failed onion request", e)
+                    when (e) {
+                        // No need for the stack trace for HTTP errors
+                        is HTTP.HTTPRequestFailedException -> Log.e("SOGS", "Failed onion request: ${e.message}")
+                        else -> Log.e("SOGS", "Failed onion request", e)
+                    }
                 }
             } else {
                 Promise.ofFail(IllegalStateException("It's currently not allowed to send non onion routed requests."))
@@ -395,13 +399,13 @@ object OpenGroupApi {
     fun downloadOpenGroupProfilePicture(
         server: String,
         roomID: String,
-        imageId: Long
+        imageId: String
     ): Promise<ByteArray, Exception> {
         val request = Request(
             verb = GET,
             room = roomID,
             server = server,
-            endpoint = Endpoint.RoomFileIndividual(roomID, imageId.toString())
+            endpoint = Endpoint.RoomFileIndividual(roomID, imageId)
         )
         return getResponseBody(request)
     }
@@ -794,16 +798,14 @@ object OpenGroupApi {
 
     private fun sequentialBatch(
         server: String,
-        requests: MutableList<BatchRequestInfo<*>>,
-        authRequired: Boolean = true
+        requests: MutableList<BatchRequestInfo<*>>
     ): Promise<List<BatchResponse<*>>, Exception> {
         val request = Request(
             verb = POST,
             room = null,
             server = server,
             endpoint = Endpoint.Sequence,
-            parameters = requests.map { it.request },
-            isAuthRequired = authRequired
+            parameters = requests.map { it.request }
         )
         return getBatchResponseJson(request, requests)
     }
@@ -912,8 +914,7 @@ object OpenGroupApi {
 
     fun getCapabilitiesAndRoomInfo(
         room: String,
-        server: String,
-        authRequired: Boolean = true
+        server: String
     ): Promise<Pair<Capabilities, RoomInfo>, Exception> {
         val requests = mutableListOf<BatchRequestInfo<*>>(
             BatchRequestInfo(
@@ -933,7 +934,7 @@ object OpenGroupApi {
                 responseType = object : TypeReference<RoomInfo>(){}
             )
         )
-        return sequentialBatch(server, requests, authRequired).map {
+        return sequentialBatch(server, requests).map {
             val capabilities = it.firstOrNull()?.body as? Capabilities ?: throw Error.ParsingFailed
             val roomInfo = it.lastOrNull()?.body as? RoomInfo ?: throw Error.ParsingFailed
             capabilities to roomInfo

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

@@ -59,7 +59,7 @@ class OpenGroupPoller(private val server: String, private val executorService: S
     fun poll(isPostCapabilitiesRetry: Boolean = false): Promise<Unit, Exception> {
         val storage = MessagingModuleConfiguration.shared.storage
         val rooms = storage.getAllOpenGroups().values.filter { it.server == server }.map { it.room }
-        rooms.forEach { downloadGroupAvatarIfNeeded(it) }
+
         return OpenGroupApi.poll(rooms, server).successBackground { responses ->
             responses.filterNot { it.body == null }.forEach { response ->
                 when (response.endpoint) {
@@ -117,15 +117,17 @@ class OpenGroupPoller(private val server: String, private val executorService: S
     ) {
         val storage = MessagingModuleConfiguration.shared.storage
         val groupId = "$server.$roomToken"
+        val dbGroupId = GroupUtil.getEncodedOpenGroupID(groupId.toByteArray())
 
         val existingOpenGroup = storage.getOpenGroup(roomToken, server)
         val publicKey = existingOpenGroup?.publicKey ?: return
         val openGroup = OpenGroup(
             server = server,
             room = pollInfo.token,
-            name = pollInfo.details?.name ?: "",
-            infoUpdates = pollInfo.details?.infoUpdates ?: 0,
+            name = if (pollInfo.details != null) { pollInfo.details.name } else { existingOpenGroup.name },
+            infoUpdates = if (pollInfo.details != null) { pollInfo.details.infoUpdates } else { existingOpenGroup.infoUpdates },
             publicKey = publicKey,
+            imageId = if (pollInfo.details != null) { pollInfo.details.imageId } else { existingOpenGroup.imageId }
         )
         // - Open Group changes
         storage.updateOpenGroup(openGroup)
@@ -155,6 +157,22 @@ class OpenGroupPoller(private val server: String, private val executorService: S
                 GroupMember(groupId, it, GroupMemberRole.HIDDEN_ADMIN)
             })
         }
+
+        if (
+            (
+                pollInfo.details != null &&
+                pollInfo.details.imageId != null && (
+                    pollInfo.details.imageId != existingOpenGroup.imageId ||
+                    !storage.hasDownloadedProfilePicture(dbGroupId)
+                )
+            ) || (
+                pollInfo.details == null &&
+                existingOpenGroup.imageId != null &&
+                !storage.hasDownloadedProfilePicture(dbGroupId)
+            )
+        ) {
+            JobQueue.shared.add(GroupAvatarDownloadJob(roomToken, server))
+        }
     }
 
     private fun handleMessages(
@@ -284,16 +302,4 @@ class OpenGroupPoller(private val server: String, private val executorService: S
             JobQueue.shared.add(deleteJob)
         }
     }
-
-    private fun downloadGroupAvatarIfNeeded(room: String) {
-        val storage = MessagingModuleConfiguration.shared.storage
-        if (storage.getGroupAvatarDownloadJob(server, room) != null) return
-        val groupId = GroupUtil.getEncodedOpenGroupID("$server.$room".toByteArray())
-        storage.getGroup(groupId)?.let {
-            if (System.currentTimeMillis() > it.updatedTimestamp + TimeUnit.DAYS.toMillis(7)) {
-                JobQueue.shared.add(GroupAvatarDownloadJob(room, server))
-            }
-        }
-    }
-
 }

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

@@ -78,8 +78,8 @@ object OnionRequestAPI {
     // endregion
 
     class HTTPRequestFailedBlindingRequiredException(statusCode: Int, json: Map<*, *>, destination: String): HTTPRequestFailedAtDestinationException(statusCode, json, destination)
-    open class HTTPRequestFailedAtDestinationException(val statusCode: Int, val json: Map<*, *>, val destination: String)
-        : Exception("HTTP request failed at destination ($destination) with status code $statusCode.")
+    open class HTTPRequestFailedAtDestinationException(statusCode: Int, json: Map<*, *>, val destination: String)
+        : HTTP.HTTPRequestFailedException(statusCode, json, "HTTP request failed at destination ($destination) with status code $statusCode.")
     class InsufficientSnodesException : Exception("Couldn't find enough snodes to build a path.")
 
     private data class OnionBuildingResult(

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

@@ -2,6 +2,7 @@ package org.session.libsession.utilities
 
 import okhttp3.HttpUrl
 import org.session.libsession.messaging.file_server.FileServerApi
+import org.session.libsignal.utilities.HTTP
 import org.session.libsignal.utilities.Log
 import java.io.*
 
@@ -40,7 +41,11 @@ object DownloadUtilities {
                 outputStream.write(it)
             }
         } catch (e: Exception) {
-            Log.e("Loki", "Couldn't download attachment.", e)
+            when (e) {
+                // No need for the stack trace for HTTP errors
+                is HTTP.HTTPRequestFailedException -> Log.e("Loki", "Couldn't download attachment due to error: ${e.message}")
+                else -> Log.e("Loki", "Couldn't download attachment", e)
+            }
             throw e
         }
     }