Implement multi device & closed group protocol in refactored sending & receiving pipeline

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

@@ -8,7 +8,7 @@ import org.session.libsignal.service.api.crypto.AttachmentCipherInputStream
 import java.io.File
 import java.io.FileInputStream
 
-class AttachmentDownloadJob(val attachmentID: Long, val tsIncomingMessageID: Long): Job {
+class AttachmentDownloadJob(val attachmentID: Long, val databaseMessageID: Long): Job {
 
     override var delegate: JobDelegate? = null
     override var id: String? = null
@@ -34,17 +34,17 @@ class AttachmentDownloadJob(val attachmentID: Long, val tsIncomingMessageID: Lon
     override fun execute() {
         val messageDataProvider = MessagingConfiguration.shared.messageDataProvider
         val attachmentStream = messageDataProvider.getAttachmentStream(attachmentID) ?: return handleFailure(Error.NoAttachment)
-        messageDataProvider.setAttachmentState(AttachmentState.STARTED, attachmentID, this.tsIncomingMessageID)
+        messageDataProvider.setAttachmentState(AttachmentState.STARTED, attachmentID, this.databaseMessageID)
         val tempFile = createTempFile()
         val handleFailure: (java.lang.Exception) -> Unit = { exception ->
             tempFile.delete()
             if(exception is Error && exception == Error.NoAttachment) {
-                MessagingConfiguration.shared.messageDataProvider.setAttachmentState(AttachmentState.FAILED, attachmentID, tsIncomingMessageID)
+                MessagingConfiguration.shared.messageDataProvider.setAttachmentState(AttachmentState.FAILED, attachmentID, databaseMessageID)
                 this.handlePermanentFailure(exception)
             } else if (exception is DotNetAPI.Error && exception == DotNetAPI.Error.ParsingFailed) {
                 // No need to retry if the response is invalid. Most likely this means we (incorrectly)
                 // got a "Cannot GET ..." error from the file server.
-                MessagingConfiguration.shared.messageDataProvider.setAttachmentState(AttachmentState.FAILED, attachmentID, tsIncomingMessageID)
+                MessagingConfiguration.shared.messageDataProvider.setAttachmentState(AttachmentState.FAILED, attachmentID, databaseMessageID)
                 this.handlePermanentFailure(exception)
             } else {
                 this.handleFailure(exception)
@@ -62,7 +62,7 @@ class AttachmentDownloadJob(val attachmentID: Long, val tsIncomingMessageID: Lon
         var stream = if (!attachmentStream.digest.isPresent || attachmentStream.key == null) FileInputStream(tempFile)
             else AttachmentCipherInputStream.createForAttachment(tempFile, attachmentStream.length.or(0).toLong(), attachmentStream.key?.toByteArray(), attachmentStream?.digest.get())
 
-        messageDataProvider.insertAttachment(tsIncomingMessageID, attachmentID, stream)
+        messageDataProvider.insertAttachment(databaseMessageID, attachmentID, stream)
 
         tempFile.delete()
 
@@ -90,7 +90,7 @@ class AttachmentDownloadJob(val attachmentID: Long, val tsIncomingMessageID: Lon
 
     override fun serialize(): Data {
         return Data.Builder().putLong(KEY_ATTACHMENT_ID, attachmentID)
-                .putLong(KEY_TS_INCOMING_MESSAGE_ID, tsIncomingMessageID)
+                .putLong(KEY_TS_INCOMING_MESSAGE_ID, databaseMessageID)
                 .build();
     }
 

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

@@ -76,14 +76,13 @@ class JobQueue : JobDelegate {
 
     private fun getRetryInterval(job: Job): Long {
         // Arbitrary backoff factor...
-        // try  1 delay:  0ms
+        // try  1 delay: 0.5s
-        // try  2 delay:  190ms
+        // try  2 delay: 1s
         // ...
-        // try  5 delay:  1300ms
+        // try  5 delay: 16s
         // ...
-        // try 11 delay: 61310ms
+        // try 11 delay: 512s
-        val backoffFactor = 1.9
+        val maxBackoff = (10 * 60).toDouble() // 10 minutes
-        val maxBackoff = (60 * 60 * 1000).toDouble()
+        return (1000 * 0.25 * min(maxBackoff, (2.0).pow(job.failureCount))).roundToLong()
-        return (100 * min(maxBackoff, backoffFactor.pow(job.failureCount))).roundToLong()
     }
 }

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

@@ -32,17 +32,21 @@ class MessageReceiveJob(val data: ByteArray, val isBackgroundPoll: Boolean, val
     fun executeAsync(): Promise<Unit, Exception> {
         val deferred = deferred<Unit, Exception>()
         try {
-            val (message, proto) = MessageReceiver.parse(this.data, this.openGroupMessageServerID)
+            val isRetry: Boolean = failureCount != 0
+            val (message, proto) = MessageReceiver.parse(this.data, this.openGroupMessageServerID, isRetry)
             MessageReceiver.handle(message, proto, this.openGroupID)
             this.handleSuccess()
             deferred.resolve(Unit)
         } catch (e: Exception) {
             Log.d(TAG, "Couldn't receive message due to error: $e.")
             val error = e as? MessageReceiver.Error
-            error?.let {
+            if (error != null && !error.isRetryable) {
-                if (!error.isRetryable) this.handlePermanentFailure(error)
+                Log.d("Loki", "Message receive job permanently failed due to error: $error.")
+                this.handlePermanentFailure(error)
+            } else {
+                Log.d("Loki", "Couldn't receive message due to error: $e.")
+                this.handleFailure(e)
             }
-            this.handleFailure(e)
             deferred.resolve(Unit) // The promise is just used to keep track of when we're done
         }
         return deferred.promise

libsession/src/main/java/org/session/libsession/messaging/messages/control/ClosedGroupControlMessage.kt

@@ -18,24 +18,24 @@ class ClosedGroupControlMessage() : ControlMessage() {
         }
     }
 
-    override val isSelfSendValid: Boolean = run {
+    override val isSelfSendValid: Boolean = true
-        when(kind) {
-            is Kind.New -> false
-            else -> true
-        }
-    }
 
     var kind: Kind? = null
 
     // Kind enum
     sealed class Kind {
         class New(val publicKey: ByteString, val name: String, val encryptionKeyPair: ECKeyPair, val members: List<ByteString>, val admins: List<ByteString>) : Kind()
-        class Update(val name: String, val members: List<ByteString>) : Kind() //deprecated
+        /// - Note: Deprecated in favor of more explicit group updates.
-        class EncryptionKeyPair(val wrappers: Collection<KeyPairWrapper>) : Kind()
+        class Update(val name: String, val members: List<ByteString>) : Kind()
+        /// An encryption key pair encrypted for each member individually.
+        ///
+        /// - Note: `publicKey` is only set when an encryption key pair is sent in a one-to-one context (i.e. not in a group).
+        class EncryptionKeyPair(val publicKey: ByteString?, val wrappers: Collection<KeyPairWrapper>) : Kind()
         class NameChange(val name: String) : Kind()
         class MembersAdded(val members: List<ByteString>) : Kind()
         class MembersRemoved( val members: List<ByteString>) : Kind()
         object MemberLeft : Kind()
+        object EncryptionKeyPairRequest: Kind()
 
         val description: String = run {
             when(this) {
@@ -46,6 +46,7 @@ class ClosedGroupControlMessage() : ControlMessage() {
                 is MembersAdded -> "membersAdded"
                 is MembersRemoved -> "membersRemoved"
                 MemberLeft -> "memberLeft"
+                EncryptionKeyPairRequest -> "encryptionKeyPairRequest"
             }
         }
     }
@@ -54,43 +55,45 @@ class ClosedGroupControlMessage() : ControlMessage() {
         const val TAG = "ClosedGroupControlMessage"
 
         fun fromProto(proto: SignalServiceProtos.Content): ClosedGroupControlMessage? {
-            val closedGroupUpdateProto = proto.dataMessage?.closedGroupUpdateV2 ?: return null
+            val closedGroupControlMessageProto = proto.dataMessage?.closedGroupUpdateV2 ?: return null
             val kind: Kind
-            when(closedGroupUpdateProto.type) {
+            when(closedGroupControlMessageProto.type) {
                 SignalServiceProtos.ClosedGroupUpdateV2.Type.NEW -> {
-                    val publicKey = closedGroupUpdateProto.publicKey ?: return null
+                    val publicKey = closedGroupControlMessageProto.publicKey ?: return null
-                    val name = closedGroupUpdateProto.name ?: return null
+                    val name = closedGroupControlMessageProto.name ?: return null
-                    val encryptionKeyPairAsProto = closedGroupUpdateProto.encryptionKeyPair ?: return null
+                    val encryptionKeyPairAsProto = closedGroupControlMessageProto.encryptionKeyPair ?: return null
 
                     try {
                         val encryptionKeyPair = ECKeyPair(DjbECPublicKey(encryptionKeyPairAsProto.publicKey.toByteArray()), DjbECPrivateKey(encryptionKeyPairAsProto.privateKey.toByteArray()))
-                        kind = Kind.New(publicKey, name, encryptionKeyPair, closedGroupUpdateProto.membersList, closedGroupUpdateProto.adminsList)
+                        kind = Kind.New(publicKey, name, encryptionKeyPair, closedGroupControlMessageProto.membersList, closedGroupControlMessageProto.adminsList)
                     } catch (e: Exception) {
                         Log.w(TAG, "Couldn't parse key pair")
                         return null
                     }
                 }
                 SignalServiceProtos.ClosedGroupUpdateV2.Type.UPDATE -> {
-                    val name = closedGroupUpdateProto.name ?: return null
+                    val name = closedGroupControlMessageProto.name ?: return null
-                    kind = Kind.Update(name, closedGroupUpdateProto.membersList)
+                    kind = Kind.Update(name, closedGroupControlMessageProto.membersList)
                 }
                 SignalServiceProtos.ClosedGroupUpdateV2.Type.ENCRYPTION_KEY_PAIR -> {
-                    val wrappers = closedGroupUpdateProto.wrappersList.mapNotNull { KeyPairWrapper.fromProto(it) }
+                    val publicKey = closedGroupControlMessageProto.publicKey
-                    kind = Kind.EncryptionKeyPair(wrappers)
+                    val wrappers = closedGroupControlMessageProto.wrappersList.mapNotNull { KeyPairWrapper.fromProto(it) }
+                    kind = Kind.EncryptionKeyPair(publicKey, wrappers)
                 }
                 SignalServiceProtos.ClosedGroupUpdateV2.Type.NAME_CHANGE -> {
-                    val name = closedGroupUpdateProto.name ?: return null
+                    val name = closedGroupControlMessageProto.name ?: return null
                     kind = Kind.NameChange(name)
                 }
                 SignalServiceProtos.ClosedGroupUpdateV2.Type.MEMBERS_ADDED -> {
-                    kind = Kind.MembersAdded(closedGroupUpdateProto.membersList)
+                    kind = Kind.MembersAdded(closedGroupControlMessageProto.membersList)
                 }
                 SignalServiceProtos.ClosedGroupUpdateV2.Type.MEMBERS_REMOVED -> {
-                    kind = Kind.MembersRemoved(closedGroupUpdateProto.membersList)
+                    kind = Kind.MembersRemoved(closedGroupControlMessageProto.membersList)
                 }
                 SignalServiceProtos.ClosedGroupUpdateV2.Type.MEMBER_LEFT -> {
                     kind = Kind.MemberLeft
                 }
+                //TODO: SignalServiceProtos.ClosedGroupUpdateV2.Type.ENCRYPTION_KEY_PAIR_REQUEST
             }
             return ClosedGroupControlMessage(kind)
         }
@@ -116,6 +119,7 @@ class ClosedGroupControlMessage() : ControlMessage() {
             is Kind.MembersAdded -> kind.members.isNotEmpty()
             is Kind.MembersRemoved -> kind.members.isNotEmpty()
             is Kind.MemberLeft -> true
+            is Kind.EncryptionKeyPairRequest -> true
         }
     }
 
@@ -126,53 +130,57 @@ class ClosedGroupControlMessage() : ControlMessage() {
             return null
         }
         try {
-            val closedGroupUpdate: SignalServiceProtos.ClosedGroupUpdateV2.Builder = SignalServiceProtos.ClosedGroupUpdateV2.newBuilder()
+            val closedGroupControlMessage: SignalServiceProtos.ClosedGroupUpdateV2.Builder = SignalServiceProtos.ClosedGroupUpdateV2.newBuilder()
             when (kind) {
                 is Kind.New -> {
-                    closedGroupUpdate.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.NEW
+                    closedGroupControlMessage.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.NEW
-                    closedGroupUpdate.publicKey = kind.publicKey
+                    closedGroupControlMessage.publicKey = kind.publicKey
-                    closedGroupUpdate.name = kind.name
+                    closedGroupControlMessage.name = kind.name
                     val encryptionKeyPairAsProto = SignalServiceProtos.KeyPair.newBuilder()
                     encryptionKeyPairAsProto.publicKey = ByteString.copyFrom(kind.encryptionKeyPair.publicKey.serialize())
                     encryptionKeyPairAsProto.privateKey = ByteString.copyFrom(kind.encryptionKeyPair.privateKey.serialize())
 
                     try {
-                        closedGroupUpdate.encryptionKeyPair = encryptionKeyPairAsProto.build()
+                        closedGroupControlMessage.encryptionKeyPair = encryptionKeyPairAsProto.build()
                     } catch (e: Exception) {
                         Log.w(TAG, "Couldn't construct closed group update proto from: $this")
                         return null
                     }
-                    closedGroupUpdate.addAllMembers(kind.members)
+                    closedGroupControlMessage.addAllMembers(kind.members)
-                    closedGroupUpdate.addAllAdmins(kind.admins)
+                    closedGroupControlMessage.addAllAdmins(kind.admins)
                 }
                 is Kind.Update -> {
-                    closedGroupUpdate.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.UPDATE
+                    closedGroupControlMessage.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.UPDATE
-                    closedGroupUpdate.name = kind.name
+                    closedGroupControlMessage.name = kind.name
-                    closedGroupUpdate.addAllMembers(kind.members)
+                    closedGroupControlMessage.addAllMembers(kind.members)
                 }
                 is Kind.EncryptionKeyPair -> {
-                    closedGroupUpdate.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.ENCRYPTION_KEY_PAIR
+                    closedGroupControlMessage.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.ENCRYPTION_KEY_PAIR
-                    closedGroupUpdate.addAllWrappers(kind.wrappers.map { it.toProto() })
+                    closedGroupControlMessage.publicKey = kind.publicKey
+                    closedGroupControlMessage.addAllWrappers(kind.wrappers.map { it.toProto() })
                 }
                 is Kind.NameChange -> {
-                    closedGroupUpdate.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.NAME_CHANGE
+                    closedGroupControlMessage.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.NAME_CHANGE
-                    closedGroupUpdate.name = kind.name
+                    closedGroupControlMessage.name = kind.name
                 }
                 is Kind.MembersAdded -> {
-                    closedGroupUpdate.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.MEMBERS_ADDED
+                    closedGroupControlMessage.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.MEMBERS_ADDED
-                    closedGroupUpdate.addAllMembers(kind.members)
+                    closedGroupControlMessage.addAllMembers(kind.members)
                 }
                 is Kind.MembersRemoved -> {
-                    closedGroupUpdate.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.MEMBERS_REMOVED
+                    closedGroupControlMessage.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.MEMBERS_REMOVED
-                    closedGroupUpdate.addAllMembers(kind.members)
+                    closedGroupControlMessage.addAllMembers(kind.members)
                 }
                 is Kind.MemberLeft -> {
-                    closedGroupUpdate.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.MEMBER_LEFT
+                    closedGroupControlMessage.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.MEMBER_LEFT
+                }
+                is Kind.EncryptionKeyPairRequest -> {
+                    // TODO: closedGroupControlMessage.type = SignalServiceProtos.ClosedGroupUpdateV2.Type.ENCRYPTION_KEY_PAIR_REQUEST
                 }
             }
             val contentProto = SignalServiceProtos.Content.newBuilder()
             val dataMessageProto = SignalServiceProtos.DataMessage.newBuilder()
-            dataMessageProto.closedGroupUpdateV2 = closedGroupUpdate.build()
+            dataMessageProto.closedGroupUpdateV2 = closedGroupControlMessage.build()
             // Group context
             contentProto.dataMessage = dataMessageProto.build()
             return contentProto.build()

libsession/src/main/java/org/session/libsession/messaging/sending_receiving/MessageReceiver.kt

@@ -2,17 +2,15 @@ package org.session.libsession.messaging.sending_receiving
 
 import org.session.libsession.messaging.MessagingConfiguration
 import org.session.libsession.messaging.messages.Message
-import org.session.libsession.messaging.messages.control.ClosedGroupUpdate
+import org.session.libsession.messaging.messages.control.*
-import org.session.libsession.messaging.messages.control.ExpirationTimerUpdate
-import org.session.libsession.messaging.messages.control.ReadReceipt
-import org.session.libsession.messaging.messages.control.TypingIndicator
 import org.session.libsession.messaging.messages.visible.VisibleMessage
-import org.session.libsignal.libsignal.ecc.ECKeyPair
 
 import org.session.libsignal.service.internal.push.SignalServiceProtos
-import java.lang.Error
 
 object MessageReceiver {
+
+    private val lastEncryptionKeyPairRequest = mutableMapOf<String, Long>()
+
     internal sealed class Error(val description: String) : Exception() {
         object DuplicateMessage: Error("Duplicate message.")
         object InvalidMessage: Error("Invalid message.")
@@ -20,6 +18,7 @@ object MessageReceiver {
         object UnknownEnvelopeType: Error("Unknown envelope type.")
         object NoUserX25519KeyPair: Error("Couldn't find user X25519 key pair.")
         object NoUserED25519KeyPair: Error("Couldn't find user ED25519 key pair.")
+        object InvalidSignature: Error("Invalid message signature.")
         object NoData: Error("Received an empty envelope.")
         object SenderBlocked: Error("Received a message from a blocked user.")
         object NoThread: Error("Couldn't find thread for message.")
@@ -28,12 +27,13 @@ object MessageReceiver {
         // Shared sender keys
         object InvalidGroupPublicKey: Error("Invalid group public key.")
         object NoGroupKeyPair: Error("Missing group key pair.")
-        object SharedSecretGenerationFailed: Error("Couldn't generate a shared secret.")
 
         internal val isRetryable: Boolean = when (this) {
+            is DuplicateMessage -> false
             is InvalidMessage -> false
             is UnknownMessage -> false
             is UnknownEnvelopeType -> false
+            is InvalidSignature -> false
             is NoData -> false
             is SenderBlocked -> false
             is SelfSend -> false
@@ -41,13 +41,16 @@ object MessageReceiver {
         }
     }
 
-    internal fun parse(data: ByteArray, openGroupServerID: Long?): Pair<Message, SignalServiceProtos.Content> {
+    internal fun parse(data: ByteArray, openGroupServerID: Long?, isRetry: Boolean = false): Pair<Message, SignalServiceProtos.Content> {
         val storage = MessagingConfiguration.shared.storage
         val userPublicKey = storage.getUserPublicKey()
         val isOpenGroupMessage = openGroupServerID != null
         // Parse the envelope
         val envelope = SignalServiceProtos.Envelope.parseFrom(data)
-        if (storage.getReceivedMessageTimestamps().contains(envelope.timestamp)) throw Error.DuplicateMessage
+        // If the message failed to process the first time around we retry it later (if the error is retryable). In this case the timestamp
+        // will already be in the database but we don't want to treat the message as a duplicate. The isRetry flag is a simple workaround
+        // for this issue.
+        if (storage.getReceivedMessageTimestamps().contains(envelope.timestamp) && !isRetry) throw Error.DuplicateMessage
         storage.addReceivedMessageTimestamp(envelope.timestamp)
         // Decrypt the contents
         val ciphertext = envelope.content ?: throw Error.NoData
@@ -89,16 +92,27 @@ object MessageReceiver {
                             }
                         }
                     }
-                    decrypt()
                     groupPublicKey = envelope.source
+                    try {
+                        decrypt()
+                    } catch(error: Exception) {
+                        val now = System.currentTimeMillis()
+                        var shouldRequestEncryptionKeyPair = true
+                        lastEncryptionKeyPairRequest[groupPublicKey!!]?.let {
+                            shouldRequestEncryptionKeyPair = now - it > 30 * 1000
+                        }
+                        if (shouldRequestEncryptionKeyPair) {
+                            MessageSender.requestEncryptionKeyPair(groupPublicKey)
+                            lastEncryptionKeyPairRequest[groupPublicKey] = now
+                        }
+                        throw error
+                    }
                 }
                 else -> throw Error.UnknownEnvelopeType
             }
         }
         // Don't process the envelope any further if the sender is blocked
         if (isBlock(sender!!)) throw Error.SenderBlocked
-        // Ignore self sends
-        if (sender == userPublicKey) throw Error.SelfSend
         // Parse the proto
         val proto = SignalServiceProtos.Content.parseFrom(plaintext)
         // Parse the message
@@ -106,17 +120,24 @@ object MessageReceiver {
                                TypingIndicator.fromProto(proto) ?:
                                ClosedGroupUpdate.fromProto(proto) ?:
                                ExpirationTimerUpdate.fromProto(proto) ?:
+                               ConfigurationMessage.fromProto(proto) ?:
                                VisibleMessage.fromProto(proto) ?: throw Error.UnknownMessage
+        // Ignore self sends if needed
+        if (!message.isSelfSendValid && sender == userPublicKey) throw Error.SelfSend
+        // Guard against control messages in open groups
         if (isOpenGroupMessage && message !is VisibleMessage) throw Error.InvalidMessage
+        // Finish parsing
         message.sender = sender
         message.recipient = userPublicKey
         message.sentTimestamp = envelope.timestamp
         message.receivedTimestamp = System.currentTimeMillis()
         message.groupPublicKey = groupPublicKey
         message.openGroupServerMessageID = openGroupServerID
+        // Validate
         var isValid = message.isValid()
         if (message is VisibleMessage && !isValid && proto.dataMessage.attachmentsCount == 0) { isValid = true }
         if (!isValid) { throw Error.InvalidMessage }
+        // Return
         return Pair(message, proto)
     }
 }

libsession/src/main/java/org/session/libsession/messaging/sending_receiving/MessageReceiverHandler.kt

@@ -1,6 +1,7 @@
 package org.session.libsession.messaging.sending_receiving
 
 import android.text.TextUtils
+import com.google.protobuf.ByteString
 import org.session.libsession.messaging.MessagingConfiguration
 import org.session.libsession.messaging.jobs.AttachmentDownloadJob
 import org.session.libsession.messaging.jobs.JobQueue
@@ -27,6 +28,7 @@ import org.session.libsignal.service.api.messages.SignalServiceGroup
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 import org.session.libsignal.service.loki.utilities.removing05PrefixIfNeeded
 import org.session.libsignal.service.loki.utilities.toHexString
+import org.session.libsignal.utilities.Hex
 import java.security.MessageDigest
 import java.util.*
 import kotlin.collections.ArrayList
@@ -103,7 +105,9 @@ fun MessageReceiver.disableExpirationTimer(message: ExpirationTimerUpdate, proto
 }
 
 private fun MessageReceiver.handleConfigurationMessage(message: ConfigurationMessage) {
+    val context = MessagingConfiguration.shared.context
     val storage = MessagingConfiguration.shared.storage
+    if (TextSecurePreferences.getConfigurationMessageSynced(context)) return
     if (message.sender != storage.getUserPublicKey()) return
     val allClosedGroupPublicKeys = storage.getAllClosedGroupPublicKeys()
     for (closeGroup in message.closedGroups) {
@@ -115,6 +119,7 @@ private fun MessageReceiver.handleConfigurationMessage(message: ConfigurationMes
         if (allOpenGroups.contains(openGroup)) continue
         storage.addOpenGroup(openGroup, 1)
     }
+    TextSecurePreferences.setConfigurationMessageSynced(context, true)
 }
 
 fun MessageReceiver.handleVisibleMessage(message: VisibleMessage, proto: SignalServiceProtos.Content, openGroupID: String?) {
@@ -155,7 +160,7 @@ fun MessageReceiver.handleVisibleMessage(message: VisibleMessage, proto: SignalS
         }
     }
     // Get or create thread
-    val threadID = storage.getOrCreateThreadIdFor(message.sender!!, message.groupPublicKey, openGroupID)
+    val threadID = storage.getOrCreateThreadIdFor(message.syncTarget ?: message.sender!!, message.groupPublicKey, openGroupID)
     // Parse quote if needed
     var quoteModel: QuoteModel? = null
     if (message.quote != null && proto.dataMessage.hasQuote()) {
@@ -209,6 +214,7 @@ private fun MessageReceiver.handleClosedGroupControlMessage(message: ClosedGroup
         is ClosedGroupControlMessage.Kind.MembersAdded -> handleClosedGroupMembersAdded(message)
         is ClosedGroupControlMessage.Kind.MembersRemoved -> handleClosedGroupMembersRemoved(message)
         ClosedGroupControlMessage.Kind.MemberLeft -> handleClosedGroupMemberLeft(message)
+        ClosedGroupControlMessage.Kind.EncryptionKeyPairRequest -> handleClosedGroupEncryptionKeyPairRequest(message)
     }
 }
 
@@ -233,14 +239,14 @@ private fun handleNewClosedGroup(sender: String, groupPublicKey: String, name: S
     } else {
         storage.createGroup(groupID, name, LinkedList(members.map { Address.fromSerialized(it) }),
                             null, null, LinkedList(admins.map { Address.fromSerialized(it) }))
+        // Notify the user
+        storage.insertIncomingInfoMessage(context, sender, groupID, SignalServiceProtos.GroupContext.Type.UPDATE, SignalServiceGroup.Type.UPDATE, name, members, admins)
     }
     storage.setProfileSharing(Address.fromSerialized(groupID), true)
     // Add the group to the user's set of public keys to poll for
     storage.addClosedGroupPublicKey(groupPublicKey)
     // Store the encryption key pair
     storage.addClosedGroupEncryptionKeyPair(encryptionKeyPair, groupPublicKey)
-    // Notify the user
-    storage.insertIncomingInfoMessage(context, sender, groupID, SignalServiceProtos.GroupContext.Type.UPDATE, SignalServiceGroup.Type.UPDATE, name, members, admins)
     // Notify the PN server
     PushNotificationAPI.performOperation(PushNotificationAPI.ClosedGroupOperation.Subscribe, groupPublicKey, storage.getUserPublicKey()!!)
 }
@@ -300,7 +306,7 @@ private fun MessageReceiver.handleClosedGroupEncryptionKeyPair(message: ClosedGr
     val storage = MessagingConfiguration.shared.storage
     val senderPublicKey = message.sender ?: return
     val kind = message.kind!! as? ClosedGroupControlMessage.Kind.EncryptionKeyPair ?: return
-    val groupPublicKey = message.groupPublicKey ?: return
+    val groupPublicKey = kind.publicKey?.toByteArray()?.toHexString() ?: message.groupPublicKey ?: return
     val userPublicKey = storage.getUserPublicKey()!!
     val userKeyPair = storage.getUserX25519KeyPair()
     // Unwrap the message
@@ -309,8 +315,8 @@ private fun MessageReceiver.handleClosedGroupEncryptionKeyPair(message: ClosedGr
         Log.d("Loki", "Ignoring closed group info message for nonexistent group.")
         return
     }
-    if (!group.admins.map { it.toString() }.contains(senderPublicKey)) {
+    if (!group.members.map { it.toString() }.contains(senderPublicKey)) {
-        android.util.Log.d("Loki", "Ignoring closed group encryption key pair from non-admin.")
+        android.util.Log.d("Loki", "Ignoring closed group encryption key pair from non-member.")
         return
     }
     // Find our wrapper and decrypt it if possible
@@ -320,7 +326,12 @@ private fun MessageReceiver.handleClosedGroupEncryptionKeyPair(message: ClosedGr
     // Parse it
     val proto = SignalServiceProtos.KeyPair.parseFrom(plaintext)
     val keyPair = ECKeyPair(DjbECPublicKey(proto.publicKey.toByteArray().removing05PrefixIfNeeded()), DjbECPrivateKey(proto.privateKey.toByteArray()))
-    // Store it
+    // Store it if needed
+    val closedGroupEncryptionKeyPairs = storage.getClosedGroupEncryptionKeyPairs(groupPublicKey)
+    if (closedGroupEncryptionKeyPairs.contains(keyPair)) {
+        Log.d("Loki", "Ignoring duplicate closed group encryption key pair.")
+        return
+    }
     storage.addClosedGroupEncryptionKeyPair(keyPair, groupPublicKey)
     Log.d("Loki", "Received a new closed group encryption key pair")
 }
@@ -431,8 +442,7 @@ private fun MessageReceiver.handleClosedGroupMemberLeft(message: ClosedGroupCont
     val storage = MessagingConfiguration.shared.storage
     val senderPublicKey = message.sender ?: return
     val userPublicKey = storage.getUserPublicKey()!!
-    if (senderPublicKey == userPublicKey) { return } // Check the user leaving isn't us, will already be handled
+    if (message.kind!! !is ClosedGroupControlMessage.Kind.MemberLeft) return
-    val kind = message.kind!! as? ClosedGroupControlMessage.Kind.MembersAdded ?: return
     val groupPublicKey = message.groupPublicKey ?: return
     val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
     val group = storage.getGroup(groupID) ?: run {
@@ -462,6 +472,37 @@ private fun MessageReceiver.handleClosedGroupMemberLeft(message: ClosedGroupCont
     storage.insertIncomingInfoMessage(context, senderPublicKey, groupID, SignalServiceProtos.GroupContext.Type.QUIT, SignalServiceGroup.Type.QUIT, name, members, admins)
 }
 
+private fun MessageReceiver.handleClosedGroupEncryptionKeyPairRequest(message: ClosedGroupControlMessage) {
+    val storage = MessagingConfiguration.shared.storage
+    val senderPublicKey = message.sender ?: return
+    val userPublicKey = storage.getUserPublicKey()!!
+    if (message.kind!! !is ClosedGroupControlMessage.Kind.EncryptionKeyPairRequest) return
+    if (senderPublicKey == userPublicKey) {
+        Log.d("Loki", "Ignoring invalid closed group update.")
+        return
+    }
+    val groupPublicKey = message.groupPublicKey ?: return
+    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
+    val group = storage.getGroup(groupID) ?: run {
+        Log.d("Loki", "Ignoring closed group info message for nonexistent group.")
+        return
+    }
+    if (!isValidGroupUpdate(group, message.sentTimestamp!!, senderPublicKey)) { return }
+    // Get the latest encryption key pair
+    val encryptionKeyPair = storage.getLatestClosedGroupEncryptionKeyPair(groupPublicKey) ?: return
+    // Send it
+    val proto = SignalServiceProtos.KeyPair.newBuilder()
+    proto.publicKey = ByteString.copyFrom(encryptionKeyPair.publicKey.serialize())
+    proto.privateKey = ByteString.copyFrom(encryptionKeyPair.privateKey.serialize())
+    val plaintext = proto.build().toByteArray()
+    val ciphertext = MessageSenderEncryption.encryptWithSessionProtocol(plaintext, senderPublicKey)
+    Log.d("Loki", "Responding to closed group encryption key pair request from: $senderPublicKey.")
+    val wrapper = ClosedGroupControlMessage.KeyPairWrapper(senderPublicKey, ByteString.copyFrom(ciphertext))
+    val kind = ClosedGroupControlMessage.Kind.EncryptionKeyPair(ByteString.copyFrom(Hex.fromStringCondensed(groupPublicKey)), listOf(wrapper))
+    val closedGroupControlMessage = ClosedGroupControlMessage(kind)
+    MessageSender.send(closedGroupControlMessage, Address.fromSerialized(senderPublicKey))
+}
+
 private fun isValidGroupUpdate(group: GroupRecord,
                                sentTimestamp: Long,
                                senderPublicKey: String): Boolean  {

libsession/src/main/java/org/session/libsession/messaging/sending_receiving/MessageSender.kt

@@ -8,7 +8,9 @@ import org.session.libsession.messaging.jobs.JobQueue
 import org.session.libsession.messaging.jobs.NotifyPNServerJob
 import org.session.libsession.messaging.messages.Destination
 import org.session.libsession.messaging.messages.Message
+import org.session.libsession.messaging.messages.control.ClosedGroupControlMessage
 import org.session.libsession.messaging.messages.control.ClosedGroupUpdate
+import org.session.libsession.messaging.messages.control.ConfigurationMessage
 import org.session.libsession.messaging.messages.visible.Attachment
 import org.session.libsession.messaging.messages.visible.Profile
 import org.session.libsession.messaging.messages.visible.VisibleMessage
@@ -86,7 +88,7 @@ object MessageSender {
     }
 
     // One-on-One Chats & Closed Groups
-    fun sendToSnodeDestination(destination: Destination, message: Message): Promise<Unit, Exception> {
+    fun sendToSnodeDestination(destination: Destination, message: Message, isSyncMessage: Boolean = false): Promise<Unit, Exception> {
         val deferred = deferred<Unit, Exception>()
         val promise = deferred.promise
         val storage = MessagingConfiguration.shared.storage
@@ -113,8 +115,13 @@ object MessageSender {
             }
             // Validate the message
             if (!message.isValid()) { throw Error.InvalidMessage }
-            // Stop here if this is a self-send
+            // Stop here if this is a self-send, unless it's:
-            if (isSelfSend) {
+            // • a configuration message
+            // • a sync message
+            // • a closed group control message of type `new`
+            var isNewClosedGroupControlMessage = false
+            if (message is ClosedGroupControlMessage && message.kind is ClosedGroupControlMessage.Kind.New) isNewClosedGroupControlMessage = true
+            if (isSelfSend && message !is ConfigurationMessage && !isSyncMessage && !isNewClosedGroupControlMessage) {
                 handleSuccessfulMessageSend(message, destination)
                 deferred.resolve(Unit)
                 return promise
@@ -183,8 +190,8 @@ object MessageSender {
                         if (destination is Destination.Contact && message is VisibleMessage && !isSelfSend) {
                             //TODO Notify user for message sent
                         }
-                        handleSuccessfulMessageSend(message, destination)
+                        handleSuccessfulMessageSend(message, destination, isSyncMessage)
-                        var shouldNotify = (message is VisibleMessage)
+                        var shouldNotify = (message is VisibleMessage && !isSyncMessage)
                         if (message is ClosedGroupUpdate && message.kind is ClosedGroupUpdate.Kind.New) {
                             shouldNotify = true
                         }
@@ -261,15 +268,28 @@ object MessageSender {
     }
 
     // Result Handling
-    fun handleSuccessfulMessageSend(message: Message, destination: Destination) {
+    fun handleSuccessfulMessageSend(message: Message, destination: Destination, isSyncMessage: Boolean = false) {
         val storage = MessagingConfiguration.shared.storage
         val messageId = storage.getMessageIdInDatabase(message.sentTimestamp!!, message.sender!!) ?: return
+        // Ignore future self-sends
+        storage.addReceivedMessageTimestamp(message.sentTimestamp!!)
+        // Track the open group server message ID
         if (message.openGroupServerMessageID != null) {
             storage.setOpenGroupServerMessageID(messageId, message.openGroupServerMessageID!!)
         }
+        // Mark the message as sent
         storage.markAsSent(messageId)
         storage.markUnidentified(messageId)
+        // Start the disappearing messages timer if needed
         SSKEnvironment.shared.messageExpirationManager.startAnyExpiration(messageId)
+        // Sync the message if:
+        // • it's a visible message
+        // • the destination was a contact
+        // • we didn't sync it already
+        val userPublicKey = storage.getUserPublicKey()!!
+        if (destination is Destination.Contact && !isSyncMessage && message is VisibleMessage) {
+            sendToSnodeDestination(Destination.Contact(userPublicKey), message, true).get()
+        }
     }
 
     fun handleFailedMessageSend(message: Message, error: Exception) {

libsession/src/main/java/org/session/libsession/messaging/sending_receiving/MessageSenderClosedGroup.kt

@@ -51,7 +51,6 @@ fun MessageSender.createClosedGroup(name: String, members: Collection<String>):
         // Send a closed group update message to all members individually
         val closedGroupUpdateKind = ClosedGroupControlMessage.Kind.New(ByteString.copyFrom(Hex.fromStringCondensed(groupPublicKey)), name, encryptionKeyPair, membersAsData, adminsAsData)
         for (member in members) {
-            if (member == userPublicKey) { continue }
             val closedGroupControlMessage = ClosedGroupControlMessage(closedGroupUpdateKind)
             sendNonDurably(closedGroupControlMessage, Address.fromSerialized(groupID)).get()
         }
@@ -160,7 +159,7 @@ fun MessageSender.removeMembers(groupPublicKey: String, membersToRemove: List<St
         Log.d("Loki", "Can't remove members from nonexistent closed group.")
         throw Error.NoThread
     }
-    if (membersToRemove.isEmpty()) {
+    if (membersToRemove.isEmpty() || membersToRemove.contains(userPublicKey)) {
         Log.d("Loki", "Invalid closed group update.")
         throw Error.InvalidClosedGroupUpdate
     }
@@ -212,6 +211,53 @@ fun MessageSender.v2_leave(groupPublicKey: String) {
     storage.insertOutgoingInfoMessage(context, groupID, infoType, name, updatedMembers, admins, threadID)
 }
 
+fun MessageSender.generateAndSendNewEncryptionKeyPair(groupPublicKey: String, targetMembers: Collection<String>) {
+    // Prepare
+    val storage = MessagingConfiguration.shared.storage
+    val userPublicKey = storage.getUserPublicKey()!!
+    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
+    val group = storage.getGroup(groupID) ?: run {
+        Log.d("Loki", "Can't update nonexistent closed group.")
+        throw Error.NoThread
+    }
+    if (!group.admins.map { it.toString() }.contains(userPublicKey)) {
+        Log.d("Loki", "Can't distribute new encryption key pair as non-admin.")
+        throw Error.InvalidClosedGroupUpdate
+    }
+    // Generate the new encryption key pair
+    val newKeyPair = Curve.generateKeyPair()
+    // Distribute it
+    val proto = SignalServiceProtos.KeyPair.newBuilder()
+    proto.publicKey = ByteString.copyFrom(newKeyPair.publicKey.serialize().removing05PrefixIfNeeded())
+    proto.privateKey = ByteString.copyFrom(newKeyPair.privateKey.serialize())
+    val plaintext = proto.build().toByteArray()
+    val wrappers = targetMembers.map { publicKey ->
+        val ciphertext = MessageSenderEncryption.encryptWithSessionProtocol(plaintext, publicKey)
+        ClosedGroupControlMessage.KeyPairWrapper(publicKey, ByteString.copyFrom(ciphertext))
+    }
+    val kind = ClosedGroupControlMessage.Kind.EncryptionKeyPair(null, wrappers)
+    val closedGroupControlMessage = ClosedGroupControlMessage(kind)
+    sendNonDurably(closedGroupControlMessage, Address.fromSerialized(groupID)).success {
+        // Store it * after * having sent out the message to the group
+        storage.addClosedGroupEncryptionKeyPair(newKeyPair, groupPublicKey)
+    }
+}
+
+fun MessageSender.requestEncryptionKeyPair(groupPublicKey: String) {
+    val storage = MessagingConfiguration.shared.storage
+    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
+    val group = storage.getGroup(groupID) ?: run {
+        Log.d("Loki", "Can't request encryption key pair for nonexistent closed group.")
+        throw Error.NoThread
+    }
+    val members = group.members.map { it.serialize() }.toSet()
+    if (!members.contains(storage.getUserPublicKey()!!)) return
+    // Send the request to the group
+    val closedGroupControlMessage = ClosedGroupControlMessage(ClosedGroupControlMessage.Kind.EncryptionKeyPairRequest)
+    send(closedGroupControlMessage, Address.fromSerialized(groupID))
+}
+
+/// - Note: Deprecated.
 fun MessageSender.update(groupPublicKey: String, members: Collection<String>, name: String): Promise<Unit, Exception> {
     val deferred = deferred<Unit, Exception>()
     val context = MessagingConfiguration.shared.context
@@ -342,6 +388,7 @@ fun MessageSender.update(groupPublicKey: String, members: Collection<String>, na
     return deferred.promise
 }
 
+/// - Note: Deprecated.
 fun MessageSender.leave(groupPublicKey: String) {
     val storage = MessagingConfiguration.shared.storage
     val userPublicKey = storage.getUserPublicKey()!!
@@ -355,36 +402,4 @@ fun MessageSender.leave(groupPublicKey: String) {
     val oldMembers = group.members.map { it.serialize() }.toSet()
     val newMembers = oldMembers.minus(userPublicKey)
     return update(groupPublicKey, newMembers, name).get()
-}
-
-fun MessageSender.generateAndSendNewEncryptionKeyPair(groupPublicKey: String, targetMembers: Collection<String>) {
-    // Prepare
-    val storage = MessagingConfiguration.shared.storage
-    val userPublicKey = storage.getUserPublicKey()!!
-    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
-    val group = storage.getGroup(groupID) ?: run {
-        Log.d("Loki", "Can't update nonexistent closed group.")
-        throw Error.NoThread
-    }
-    if (!group.admins.map { it.toString() }.contains(userPublicKey)) {
-        Log.d("Loki", "Can't distribute new encryption key pair as non-admin.")
-        throw Error.InvalidClosedGroupUpdate
-    }
-    // Generate the new encryption key pair
-    val newKeyPair = Curve.generateKeyPair()
-    // Distribute it
-    val proto = SignalServiceProtos.KeyPair.newBuilder()
-    proto.publicKey = ByteString.copyFrom(newKeyPair.publicKey.serialize().removing05PrefixIfNeeded())
-    proto.privateKey = ByteString.copyFrom(newKeyPair.privateKey.serialize())
-    val plaintext = proto.build().toByteArray()
-    val wrappers = targetMembers.map { publicKey ->
-        val ciphertext = MessageSenderEncryption.encryptWithSessionProtocol(plaintext, publicKey)
-        ClosedGroupControlMessage.KeyPairWrapper(publicKey, ByteString.copyFrom(ciphertext))
-    }
-    val kind = ClosedGroupControlMessage.Kind.EncryptionKeyPair(wrappers)
-    val closedGroupControlMessage = ClosedGroupControlMessage(kind)
-    sendNonDurably(closedGroupControlMessage, Address.fromSerialized(groupID)).success {
-        // Store it * after * having sent out the message to the group
-        storage.addClosedGroupEncryptionKeyPair(newKeyPair, groupPublicKey)
-    }
 }

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

@@ -184,21 +184,13 @@ class OpenGroupPoller(private val openGroup: OpenGroup) {
                 groupProto.setType(GroupContext.Type.DELIVER)
                 groupProto.setName(openGroup.displayName)
                 dataMessageProto.setGroup(groupProto.build())
+                // Sync target
+                if (wasSentByCurrentUser) {
+                    dataMessageProto.setSyncTarget(openGroup.id)
+                }
                 // Content
                 val content = Content.newBuilder()
-                if (!wasSentByCurrentUser) { // Incoming message
+                content.setDataMessage(dataMessageProto.build())
-                    content.setDataMessage(dataMessageProto.build())
-                } else { // Outgoing message
-                    // FIXME: This needs to be updated as we removed sync message handling
-                    val syncMessageSentBuilder = SyncMessage.Sent.newBuilder()
-                    syncMessageSentBuilder.setMessage(dataMessageProto)
-                    syncMessageSentBuilder.setDestination(userHexEncodedPublicKey)
-                    syncMessageSentBuilder.setTimestamp(message.timestamp)
-                    val syncMessageSent = syncMessageSentBuilder.build()
-                    val syncMessageBuilder = SyncMessage.newBuilder()
-                    syncMessageBuilder.setSent(syncMessageSent)
-                    content.setSyncMessage(syncMessageBuilder.build())
-                }
                 // Envelope
                 val builder = Envelope.newBuilder()
                 builder.type = Envelope.Type.UNIDENTIFIED_SENDER