Refactor community room info polling (#1473)

Author: SessionHero01

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

@@ -14,7 +14,6 @@ import org.session.libsession.messaging.messages.visible.Attachment
 import org.session.libsession.messaging.messages.visible.Profile
 import org.session.libsession.messaging.messages.visible.Reaction
 import org.session.libsession.messaging.messages.visible.VisibleMessage
-import org.session.libsession.messaging.open_groups.OpenGroup
 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
@@ -67,12 +66,8 @@ interface StorageProtocol {
     fun getServerCapabilities(server: String): List<String>?
 
     // Open Groups
-    fun getAllOpenGroups(): Map<Long, OpenGroup>
-    fun getOpenGroup(threadId: Long): OpenGroup?
-    fun getOpenGroup(address: Address): OpenGroup?
     suspend fun addOpenGroup(urlAsString: String)
     fun setOpenGroupServerMessageID(messageID: MessageId, serverID: Long, threadID: Long)
-    fun getOpenGroup(room: String, server: String): OpenGroup?
 
     // Open Group Public Keys
     fun getOpenGroupPublicKey(server: String): String?
@@ -82,7 +77,6 @@ interface StorageProtocol {
     fun updateProfilePicture(groupID: String, newValue: ByteArray)
     fun removeProfilePicture(groupID: String)
     fun hasDownloadedProfilePicture(groupID: String): Boolean
-    fun setUserCount(room: String, server: String, newValue: Int)
 
     // Last Message Server ID
     fun getLastMessageServerID(room: String, server: String): Long?
@@ -179,7 +173,14 @@ interface StorageProtocol {
     /**
      * Returns the ID of the `TSIncomingMessage` that was constructed.
      */
-    fun persist(message: VisibleMessage, quotes: QuoteModel?, linkPreview: List<LinkPreview?>, fromGroup: Address.GroupLike?, attachments: List<Attachment>, runThreadUpdate: Boolean): MessageId?
+    fun persist(
+        threadRecipient: Recipient,
+        message: VisibleMessage,
+        quotes: QuoteModel?,
+        linkPreview: List<LinkPreview?>,
+        attachments: List<Attachment>,
+        runThreadUpdate: Boolean
+    ): MessageId?
     fun markConversationAsRead(threadId: Long, lastSeenTime: Long, force: Boolean = false)
     fun markConversationAsUnread(threadId: Long)
     fun getLastSeen(threadId: Long): Long

app/src/main/java/org/session/libsession/messaging/jobs/AttachmentDownloadJob.kt

@@ -15,6 +15,7 @@ import org.session.libsession.messaging.sending_receiving.attachments.DatabaseAt
 import org.session.libsession.messaging.utilities.Data
 import org.session.libsession.snode.OnionRequestAPI
 import org.session.libsession.snode.utilities.await
+import org.session.libsession.utilities.Address
 import org.session.libsession.utilities.DecodedAudio
 import org.session.libsession.utilities.DownloadUtilities
 import org.session.libsession.utilities.InputStreamMediaDataSource
@@ -66,11 +67,12 @@ class AttachmentDownloadJob @AssistedInject constructor(
          * of whether the download is allowed, it does not check if the download has already taken
          * place.
          */
-        fun eligibleForDownload(threadID: Long,
-                                storage: StorageProtocol,
-                                recipientRepository: RecipientRepository,
-                                messageDataProvider: MessageDataProvider,
-                                mmsId: Long): Boolean {
+        fun eligibleForDownload(
+            threadID: Long,
+            storage: StorageProtocol,
+            messageDataProvider: MessageDataProvider,
+            mmsId: Long
+        ): Boolean {
             val threadRecipient = storage.getRecipientForThread(threadID) ?: return false
 
             // if we are the sender we are always eligible
@@ -138,14 +140,15 @@ class AttachmentDownloadJob @AssistedInject constructor(
         if (!eligibleForDownload(
                 threadID = threadID,
                 storage = storage,
-                recipientRepository = recipientRepository,
                 messageDataProvider = messageDataProvider,
                 mmsId = mmsMessageId
             )) {
             handleFailure(Error.NoSender, null)
             return
         }
 
+        val threadRecipient = storage.getRecipientForThread(threadID)
+
         var tempFile: File? = null
         var attachment: DatabaseAttachment? = null
 
@@ -157,15 +160,14 @@ class AttachmentDownloadJob @AssistedInject constructor(
                 return
             }
             messageDataProvider.setAttachmentState(AttachmentState.DOWNLOADING, attachment.attachmentId, this.mmsMessageId)
-            val openGroup = storage.getOpenGroup(threadID)
-            val downloadedData = if (openGroup == null) {
+            val downloadedData = if (threadRecipient?.address !is Address.Community) {
                 Log.d("AttachmentDownloadJob", "downloading normal attachment")
                 DownloadUtilities.downloadFromFileServer(attachment.url).body
             } else {
                 Log.d("AttachmentDownloadJob", "downloading open group attachment")
                 val url = attachment.url.toHttpUrlOrNull()!!
                 val fileID = url.pathSegments.last()
-                OpenGroupApi.download(fileID, openGroup.room, openGroup.server).await()
+                OpenGroupApi.download(fileID, room = threadRecipient.address.room, server = threadRecipient.address.serverUrl).await()
             }
 
             tempFile = createTempFile().also { file ->

app/src/main/java/org/session/libsession/messaging/jobs/AttachmentUploadJob.kt

@@ -18,6 +18,7 @@ import org.session.libsession.messaging.open_groups.OpenGroupApi
 import org.session.libsession.messaging.sending_receiving.MessageSender
 import org.session.libsession.messaging.utilities.Data
 import org.session.libsession.snode.utilities.await
+import org.session.libsession.utilities.Address
 import org.session.libsession.utilities.DecodedAudio
 import org.session.libsession.utilities.InputStreamMediaDataSource
 import org.session.libsession.utilities.UploadResult
@@ -30,6 +31,7 @@ import org.session.libsignal.streams.PlaintextOutputStreamFactory
 import org.session.libsignal.utilities.Log
 import org.session.libsignal.utilities.PushAttachmentData
 import org.session.libsignal.utilities.Util
+import org.thoughtcrime.securesms.database.ThreadDatabase
 
 class AttachmentUploadJob @AssistedInject constructor(
     @Assisted val attachmentID: Long,
@@ -39,6 +41,7 @@ class AttachmentUploadJob @AssistedInject constructor(
     private val storage: StorageProtocol,
     private val messageDataProvider: MessageDataProvider,
     private val messageSendJobFactory: MessageSendJob.Factory,
+    private val threadDatabase: ThreadDatabase,
 ) : Job {
     override var delegate: JobDelegate? = null
     override var id: String? = null
@@ -67,11 +70,13 @@ class AttachmentUploadJob @AssistedInject constructor(
         try {
             val attachment = messageDataProvider.getScaledSignalAttachmentStream(attachmentID)
                 ?: return handleFailure(dispatcherName, Error.NoAttachment)
-            val openGroup = storage.getOpenGroup(threadID.toLong())
 
-            if (openGroup != null) {
-                val keyAndResult = upload(attachment, openGroup.server, false) {
-                    OpenGroupApi.upload(it, openGroup.room, openGroup.server)
+            val threadAddress = threadDatabase.getRecipientForThreadId(threadID.toLong()) ?: return handlePermanentFailure(dispatcherName,
+                RuntimeException("Thread doesn't exist"))
+
+            if (threadAddress is Address.Community) {
+                val keyAndResult = upload(attachment, threadAddress.serverUrl, false) {
+                    OpenGroupApi.upload(it, threadAddress.room, threadAddress.serverUrl)
                 }
                 handleSuccess(dispatcherName, attachment, keyAndResult.first, keyAndResult.second)
             } else {

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

@@ -36,8 +36,8 @@ import org.session.libsession.utilities.Address.Companion.toAddress
 import org.session.libsession.utilities.ConfigFactoryProtocol
 import org.session.libsession.utilities.UserConfigType
 import org.session.libsignal.protos.UtilProtos
-import org.session.libsignal.utilities.AccountId
 import org.session.libsignal.utilities.Log
+import org.thoughtcrime.securesms.database.RecipientRepository
 import org.thoughtcrime.securesms.database.ThreadDatabase
 import org.thoughtcrime.securesms.database.model.MessageId
 import org.thoughtcrime.securesms.database.model.ReactionRecord
@@ -61,6 +61,7 @@ class BatchMessageReceiveJob @AssistedInject constructor(
     private val visibleMessageHandlerContextFactory: VisibleMessageHandlerContext.Factory,
     private val messageNotifier: MessageNotifier,
     private val threadDatabase: ThreadDatabase,
+    private val recipientRepository: RecipientRepository,
 ) : Job {
 
     override var delegate: JobDelegate? = null
@@ -102,7 +103,8 @@ class BatchMessageReceiveJob @AssistedInject constructor(
             visibleMessageHandlerContextFactory = visibleMessageHandlerContextFactory,
             messageNotifier = messageNotifier,
             fromCommunity = fromCommunity,
-            threadDatabase = threadDatabase
+            threadDatabase = threadDatabase,
+            recipientRepository = recipientRepository,
         )
     }
 
@@ -146,7 +148,7 @@ class BatchMessageReceiveJob @AssistedInject constructor(
     }
 
     suspend fun executeAsync(dispatcherName: String) {
-        val threadMap = mutableMapOf<Long, MutableList<ParsedMessage>>()
+        val threadMap = mutableMapOf<Long, Pair<Address.Conversable, MutableList<ParsedMessage>>>()
         val localUserPublicKey = storage.getUserPublicKey()
         val serverPublicKey = fromCommunity?.let { storage.getOpenGroupPublicKey(it.serverUrl) }
         val currentClosedGroups = storage.getAllActiveClosedGroupPublicKeys()
@@ -171,14 +173,14 @@ class BatchMessageReceiveJob @AssistedInject constructor(
                     fromCommunity != null -> fromCommunity
                     message.groupPublicKey != null -> message.groupPublicKey!!.toAddress()
                     else -> message.senderOrSync.toAddress()
-                }
+                } as Address.Conversable
 
                 val threadID = if (shouldCreateThread(parsedParams)) {
                     threadDatabase.getOrCreateThreadIdFor(threadAddress)
                 } else {
                     threadDatabase.getThreadIdIfExistsFor(threadAddress)
                 }
-                threadMap.getOrPut(threadID) { mutableListOf() } += parsedParams
+                threadMap.getOrPut(threadID) { threadAddress to mutableListOf() }.second += parsedParams
             } catch (e: Exception) {
                 when (e) {
                     is MessageReceiver.Error.DuplicateMessage, MessageReceiver.Error.SelfSend -> {
@@ -202,13 +204,14 @@ class BatchMessageReceiveJob @AssistedInject constructor(
         }
 
         // iterate over threads and persist them (persistence is the longest constant in the batch process operation)
-        suspend fun processMessages(threadId: Long, messages: List<ParsedMessage>) {
+        suspend fun processMessages(threadId: Long, threadAddress: Address.Conversable, messages: List<ParsedMessage>) {
             // The LinkedHashMap should preserve insertion order
             val messageIds = linkedMapOf<MessageId, Pair<Boolean, Boolean>>()
             val myLastSeen = storage.getLastSeen(threadId)
             var updatedLastSeen = myLastSeen.takeUnless { it == -1L } ?: 0
             val handlerContext = visibleMessageHandlerContextFactory.create(
                 threadId = threadId,
+                threadAddress,
             )
 
             val communityReactions = mutableMapOf<MessageId, MutableList<ReactionRecord>>()
@@ -262,8 +265,7 @@ class BatchMessageReceiveJob @AssistedInject constructor(
                             message = message,
                             proto = proto,
                             threadId = threadId,
-                            groupv2Id = parameters.closedGroup?.publicKey?.let(::AccountId),
-                            fromCommunity = fromCommunity,
+                            threadAddress = threadAddress
                         )
                     }
                 } catch (e: Exception) {
@@ -297,18 +299,23 @@ class BatchMessageReceiveJob @AssistedInject constructor(
 
         coroutineScope {
             val withoutDefault = threadMap.entries.filter { it.key != NO_THREAD_MAPPING }
-            val deferredThreadMap = withoutDefault.map { (threadId, messages) ->
+            val deferredThreadMap = withoutDefault.map { (threadId, data) ->
+                val (threadAddress, messages) = data
                 async(Dispatchers.Default) {
-                    processMessages(threadId, messages)
+                    processMessages(
+                        threadId = threadId,
+                        threadAddress = threadAddress,
+                        messages = messages
+                    )
                 }
             }
             // await all thread processing
             deferredThreadMap.awaitAll()
         }
 
-        val noThreadMessages = threadMap[NO_THREAD_MAPPING] ?: listOf()
-        if (noThreadMessages.isNotEmpty()) {
-            processMessages(NO_THREAD_MAPPING, noThreadMessages)
+        val noThreadMessages = threadMap[NO_THREAD_MAPPING]
+        if (noThreadMessages != null && noThreadMessages.second.isNotEmpty()) {
+            processMessages(NO_THREAD_MAPPING, noThreadMessages.first, noThreadMessages.second)
         }
 
         if (failures.isEmpty()) {

app/src/main/java/org/session/libsession/messaging/messages/Destination.kt

@@ -43,23 +43,19 @@ sealed class Destination {
                     LegacyClosedGroup(address.groupPublicKeyHex)
                 }
                 is Address.Community -> {
-                    val storage = MessagingModuleConfiguration.shared.storage
-                    val threadID = storage.getThreadId(address)!!
-                    storage.getOpenGroup(threadID)?.let {
-                        OpenGroup(roomToken = it.room, server = it.server, fileIds = fileIds)
-                    } ?: throw Exception("Missing open group for thread with ID: $threadID.")
+                    OpenGroup(roomToken = address.room, server = address.serverUrl, fileIds = fileIds)
                 }
                 is Address.CommunityBlindedId -> {
                     val serverPublicKey = MessagingModuleConfiguration.shared.configFactory
                         .withUserConfigs { configs ->
                             configs.userGroups.allCommunityInfo()
-                                .first { it.community.baseUrl == address.serverUrl.toString() }
+                                .first { it.community.baseUrl == address.serverUrl }
                                 .community
                                 .pubKeyHex
                         }
 
                     OpenGroupInbox(
-                        server = address.serverUrl.toString(),
+                        server = address.serverUrl,
                         serverPublicKey = serverPublicKey,
                         blindedPublicKey = address.blindedId.blindedId.hexString,
                     )

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

@@ -9,6 +9,7 @@ import org.session.libsession.messaging.open_groups.OpenGroup.Companion.toAddres
 import org.session.libsession.utilities.Address
 import org.session.libsession.utilities.GroupUtil
 
+@Deprecated("This class is no longer used except in migration. Use RoomInfo instead")
 @Serializable
 data class OpenGroup(
     val server: String,