Merge pull request #443 from RyanRory/multi_device_refactor

The Refactor: Multi Device & Closed Group

app/src/main/java/org/thoughtcrime/securesms/attachments/DatabaseAttachmentProvider.kt

@@ -8,6 +8,8 @@ import org.session.libsession.messaging.sending_receiving.attachments.*
 import org.session.libsession.messaging.threads.Address
 import org.session.libsignal.libsignal.util.guava.Optional
 import org.session.libsignal.service.api.messages.SignalServiceAttachment
+import org.session.libsignal.service.api.messages.SignalServiceAttachmentPointer
+import org.session.libsignal.service.api.messages.SignalServiceAttachmentStream
 import org.thoughtcrime.securesms.database.Database
 import org.thoughtcrime.securesms.database.DatabaseFactory
 import org.thoughtcrime.securesms.database.helpers.SQLCipherOpenHelper
@@ -32,6 +34,18 @@ class DatabaseAttachmentProvider(context: Context, helper: SQLCipherOpenHelper)
         return databaseAttachment.toAttachmentPointer()
     }
 
+    override fun getSignalAttachmentStream(attachmentId: Long): SignalServiceAttachmentStream? {
+        val attachmentDatabase = DatabaseFactory.getAttachmentDatabase(context)
+        val databaseAttachment = attachmentDatabase.getAttachment(AttachmentId(attachmentId, 0)) ?: return null
+        return databaseAttachment.toSignalAttachmentStream(context)
+    }
+
+    override fun getSignalAttachmentPointer(attachmentId: Long): SignalServiceAttachmentPointer? {
+        val attachmentDatabase = DatabaseFactory.getAttachmentDatabase(context)
+        val databaseAttachment = attachmentDatabase.getAttachment(AttachmentId(attachmentId, 0)) ?: return null
+        return databaseAttachment.toSignalAttachmentPointer()
+    }
+
     override fun setAttachmentState(attachmentState: AttachmentState, attachmentId: Long, messageID: Long) {
         val attachmentDatabase = DatabaseFactory.getAttachmentDatabase(context)
         attachmentDatabase.setTransferState(messageID, AttachmentId(attachmentId, 0), attachmentState.value)
@@ -103,6 +117,17 @@ fun DatabaseAttachment.toAttachmentStream(context: Context): SessionServiceAttac
     return attachmentStream
 }
 
+fun DatabaseAttachment.toSignalAttachmentPointer(): SignalServiceAttachmentPointer {
+    return SignalServiceAttachmentPointer(attachmentId.rowId, contentType, key?.toByteArray(), Optional.fromNullable(size.toInt()), Optional.absent(), width, height, Optional.fromNullable(digest), Optional.fromNullable(fileName), isVoiceNote, Optional.fromNullable(caption), url)
+}
+
+fun DatabaseAttachment.toSignalAttachmentStream(context: Context): SignalServiceAttachmentStream {
+    val stream = PartAuthority.getAttachmentStream(context, this.dataUri!!)
+    val listener = SignalServiceAttachment.ProgressListener { total: Long, progress: Long -> EventBus.getDefault().postSticky(PartProgressEvent(this, total, progress))}
+
+    return SignalServiceAttachmentStream(stream, this.contentType, this.size, Optional.fromNullable(this.fileName), this.isVoiceNote, Optional.absent(), this.width, this.height, Optional.fromNullable(this.caption), listener)
+}
+
 fun DatabaseAttachment.shouldHaveImageSize(): Boolean {
     return (MediaUtil.isVideo(this) || MediaUtil.isImage(this) || MediaUtil.isGif(this));
 }

app/src/main/java/org/thoughtcrime/securesms/database/GroupDatabase.java

@@ -136,7 +136,7 @@ public class GroupDatabase extends Database implements LokiOpenGroupDatabaseProt
         return cursor.getString(cursor.getColumnIndexOrThrow(GROUP_ID));
       } else {
         String groupId = GroupUtil.getEncodedMMSGroupID(allocateGroupId());
-        create(groupId, null, members, null, null, admins);
+        create(groupId, null, members, null, null, admins, System.currentTimeMillis());
         return groupId;
       }
     } finally {
@@ -196,7 +196,7 @@ public class GroupDatabase extends Database implements LokiOpenGroupDatabaseProt
   }
 
   public long create(@NonNull String groupId, @Nullable String title, @NonNull List<Address> members,
-                     @Nullable SignalServiceAttachmentPointer avatar, @Nullable String relay, @Nullable List<Address> admins)
+                     @Nullable SignalServiceAttachmentPointer avatar, @Nullable String relay, @Nullable List<Address> admins, @NonNull Long formationTimestamp)
   {
     Collections.sort(members);
 
@@ -214,7 +214,7 @@ public class GroupDatabase extends Database implements LokiOpenGroupDatabaseProt
     }
 
     contentValues.put(AVATAR_RELAY, relay);
-    contentValues.put(TIMESTAMP, System.currentTimeMillis());
+    contentValues.put(TIMESTAMP, formationTimestamp);
     contentValues.put(ACTIVE, 1);
     contentValues.put(MMS, GroupUtil.isMmsGroup(groupId));
 

app/src/main/java/org/thoughtcrime/securesms/database/MmsDatabase.java

@@ -905,6 +905,9 @@ public class MmsDatabase extends MessagingDatabase {
   public Optional<InsertResult> insertSecureDecryptedMessageOutbox(OutgoingMediaMessage retrieved, long threadId, long serverTimestamp)
           throws MmsException
   {
+    if (threadId == -1) {
+      threadId = DatabaseFactory.getThreadDatabase(context).getOrCreateThreadIdFor(retrieved.getRecipient());
+    }
     long messageId = insertMessageOutbox(retrieved, threadId, false, null, serverTimestamp);
     if (messageId == -1) {
       return Optional.absent();

app/src/main/java/org/thoughtcrime/securesms/database/MmsSmsDatabase.java

@@ -78,6 +78,14 @@ public class MmsSmsDatabase extends Database {
     super(context, databaseHelper);
   }
 
+  public @Nullable MessageRecord getMessageForTimestamp(long timestamp) {
+    MmsSmsDatabase db = DatabaseFactory.getMmsSmsDatabase(context);
+    try (Cursor cursor = queryTables(PROJECTION, MmsSmsColumns.NORMALIZED_DATE_SENT + " = " + timestamp, null, null)) {
+      MmsSmsDatabase.Reader reader = db.readerFor(cursor);
+      return reader.getNext();
+    }
+  }
+
   public @Nullable MessageRecord getMessageFor(long messageId) {
     MmsSmsDatabase db = DatabaseFactory.getMmsSmsDatabase(context);
 

app/src/main/java/org/thoughtcrime/securesms/database/SmsDatabase.java

@@ -687,6 +687,9 @@ public class SmsDatabase extends MessagingDatabase {
   }
 
   public Optional<InsertResult> insertMessageOutbox(long threadId, OutgoingTextMessage message, long serverTimestamp) {
+    if (threadId == -1) {
+      threadId = DatabaseFactory.getThreadDatabase(context).getOrCreateThreadIdFor(message.getRecipient());
+    }
     long messageId = insertMessageOutbox(threadId, message, false, serverTimestamp, null);
     if (messageId == -1) {
       return Optional.absent();

app/src/main/java/org/thoughtcrime/securesms/database/Storage.kt

@@ -12,6 +12,8 @@ import org.session.libsession.messaging.messages.visible.Attachment
 import org.session.libsession.messaging.messages.visible.VisibleMessage
 import org.session.libsession.messaging.opengroups.OpenGroup
 import org.session.libsession.messaging.sending_receiving.attachments.AttachmentId
+import org.session.libsession.messaging.sending_receiving.attachments.PointerAttachment
+import org.session.libsession.messaging.sending_receiving.attachments.SessionServiceAttachment
 import org.session.libsession.messaging.sending_receiving.linkpreview.LinkPreview
 import org.session.libsession.messaging.sending_receiving.quotes.QuoteModel
 import org.session.libsession.messaging.threads.Address
@@ -27,17 +29,21 @@ import org.session.libsignal.service.api.messages.SignalServiceAttachmentPointer
 import org.session.libsignal.service.api.messages.SignalServiceGroup
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 import org.session.libsignal.service.loki.api.opengroups.PublicChat
+import org.session.libsignal.utilities.logging.Log
 import org.thoughtcrime.securesms.crypto.IdentityKeyUtil
 import org.thoughtcrime.securesms.database.helpers.SQLCipherOpenHelper
 import org.thoughtcrime.securesms.loki.database.LokiThreadDatabase
+import org.thoughtcrime.securesms.loki.protocol.SessionMetaProtocol
 import org.thoughtcrime.securesms.loki.utilities.OpenGroupUtilities
 import org.thoughtcrime.securesms.loki.utilities.get
 import org.thoughtcrime.securesms.loki.utilities.getString
 import org.thoughtcrime.securesms.mms.IncomingMediaMessage
 import org.thoughtcrime.securesms.mms.OutgoingGroupMediaMessage
+import org.thoughtcrime.securesms.mms.OutgoingMediaMessage
 import org.thoughtcrime.securesms.mms.PartAuthority
 import org.thoughtcrime.securesms.sms.IncomingGroupMessage
 import org.thoughtcrime.securesms.sms.IncomingTextMessage
+import org.thoughtcrime.securesms.sms.OutgoingTextMessage
 
 class Storage(context: Context, helper: SQLCipherOpenHelper) : Database(context, helper), StorageProtocol {
     override fun getUserPublicKey(): String? {
@@ -90,9 +96,8 @@ class Storage(context: Context, helper: SQLCipherOpenHelper) : Database(context,
 
     override fun persist(message: VisibleMessage, quotes: QuoteModel?, linkPreview: List<LinkPreview?>, groupPublicKey: String?, openGroupID: String?): Long? {
         var messageID: Long? = null
-        val address = Address.fromSerialized(message.sender!!)
+        val senderAddress = Address.fromSerialized(message.sender!!)
-        val recipient = Recipient.from(context, address, false)
+        val senderRecipient = Recipient.from(context, senderAddress, false)
-        val body: Optional<String> = if (message.text != null) Optional.of(message.text) else Optional.absent()
         var group: Optional<SignalServiceGroup> = Optional.absent()
         if (openGroupID != null) {
             group = Optional.of(SignalServiceGroup(openGroupID.toByteArray(), SignalServiceGroup.GroupType.PUBLIC_CHAT))
@@ -100,17 +105,39 @@ class Storage(context: Context, helper: SQLCipherOpenHelper) : Database(context,
             group = Optional.of(SignalServiceGroup(groupPublicKey.toByteArray(), SignalServiceGroup.GroupType.SIGNAL))
         }
         if (message.isMediaMessage()) {
-            val attachments: Optional<List<SignalServiceAttachment>> = Optional.absent() // TODO figure out how to get SignalServiceAttachment with attachmentID
             val quote: Optional<QuoteModel> = if (quotes != null) Optional.of(quotes) else Optional.absent()
             val linkPreviews: Optional<List<LinkPreview>> = if (linkPreview.isEmpty()) Optional.absent() else Optional.of(linkPreview.mapNotNull { it!! })
-            val mediaMessage = IncomingMediaMessage(address, message.receivedTimestamp!!, -1, recipient.expireMessages * 1000L, false, false, body, group, attachments, quote, Optional.absent(), linkPreviews, Optional.absent())
             val mmsDatabase = DatabaseFactory.getMmsDatabase(context)
             mmsDatabase.beginTransaction()
-            val insertResult: Optional<MessagingDatabase.InsertResult>
+            val insertResult = if (message.sender == getUserPublicKey()) {
-            if (group.isPresent) {
+                val targetAddress = if (message.syncTarget != null) {
-                insertResult = mmsDatabase.insertSecureDecryptedMessageInbox(mediaMessage, message.threadID ?: -1, message.sentTimestamp!!);
+                    Address.fromSerialized(message.syncTarget!!)
+                } else {
+                    if (group.isPresent) {
+                        Address.fromSerialized(GroupUtil.getEncodedId(group.get()))
+                    } else {
+                        Log.d("Loki", "Cannot handle message from self.")
+                        return null
+                    }
+                }
+                val attachments = message.attachmentIDs.mapNotNull {
+                    DatabaseFactory.getAttachmentProvider(context).getSignalAttachmentPointer(it)
+                }.mapNotNull {
+                    PointerAttachment.forPointer(Optional.of(it)).orNull()
+                }
+                val mediaMessage = OutgoingMediaMessage.from(message, Recipient.from(context, targetAddress, false), attachments, quote.orNull(), linkPreviews.orNull())
+                mmsDatabase.insertSecureDecryptedMessageOutbox(mediaMessage, message.threadID ?: -1, message.sentTimestamp!!)
             } else {
-                insertResult = mmsDatabase.insertSecureDecryptedMessageInbox(mediaMessage, message.threadID ?: -1)
+                // It seems like we have replaced SignalServiceAttachment with SessionServiceAttachment
+                val attachments: Optional<List<SignalServiceAttachment>> = Optional.of(message.attachmentIDs.mapNotNull {
+                    DatabaseFactory.getAttachmentProvider(context).getSignalAttachmentPointer(it)
+                })
+                val mediaMessage = IncomingMediaMessage.from(message, senderAddress, senderRecipient.expireMessages * 1000L, group, attachments, quote, linkPreviews)
+                if (group.isPresent) {
+                    mmsDatabase.insertSecureDecryptedMessageInbox(mediaMessage, message.threadID ?: -1, message.sentTimestamp!!)
+                } else {
+                    mmsDatabase.insertSecureDecryptedMessageInbox(mediaMessage, message.threadID ?: -1)
+                }
             }
             if (insertResult.isPresent) {
                 mmsDatabase.setTransactionSuccessful()
@@ -118,9 +145,28 @@ class Storage(context: Context, helper: SQLCipherOpenHelper) : Database(context,
             }
             mmsDatabase.endTransaction()
         } else {
-            val textMessage = IncomingTextMessage(address, 1, message.receivedTimestamp!!, body.get(), group, recipient.expireMessages * 1000L, false)
             val smsDatabase = DatabaseFactory.getSmsDatabase(context)
-            val insertResult = smsDatabase.insertMessageInbox(textMessage)
+            val insertResult = if (message.sender == getUserPublicKey()) {
+                val targetAddress = if (message.syncTarget != null) {
+                    Address.fromSerialized(message.syncTarget!!)
+                } else {
+                    if (group.isPresent) {
+                        Address.fromSerialized(GroupUtil.getEncodedId(group.get()))
+                    } else {
+                        Log.d("Loki", "Cannot handle message from self.")
+                        return null
+                    }
+                }
+                val textMessage = OutgoingTextMessage.from(message, Recipient.from(context, targetAddress, false))
+                smsDatabase.insertMessageOutbox(message.threadID ?: -1, textMessage, message.sentTimestamp!!)
+            } else {
+                val textMessage = IncomingTextMessage.from(message, senderAddress, group, senderRecipient.expireMessages * 1000L)
+                if (group.isPresent) {
+                    smsDatabase.insertMessageInbox(textMessage, message.sentTimestamp!!)
+                } else {
+                    smsDatabase.insertMessageInbox(textMessage)
+                }
+            }
             if (insertResult.isPresent) {
                 messageID = insertResult.get().messageId
             }
@@ -129,7 +175,6 @@ class Storage(context: Context, helper: SQLCipherOpenHelper) : Database(context,
     }
 
     // JOBS
-
     override fun persistJob(job: Job) {
         DatabaseFactory.getSessionJobDatabase(context).persistJob(job)
     }
@@ -235,6 +280,15 @@ class Storage(context: Context, helper: SQLCipherOpenHelper) : Database(context,
         DatabaseFactory.getLokiAPIDatabase(context).removeLastDeletionServerID(group, server)
     }
 
+    override fun isMessageDuplicated(timestamp: Long, sender: String): Boolean {
+        val database = DatabaseFactory.getMmsSmsDatabase(context)
+        return if (sender.isEmpty()) {
+            database.getMessageForTimestamp(timestamp) != null
+        } else {
+            database.getMessageFor(timestamp, sender) != null
+        }
+    }
+
     override fun setUserCount(group: Long, server: String, newValue: Int) {
         DatabaseFactory.getLokiAPIDatabase(context).setUserCount(group, server, newValue)
     }
@@ -256,13 +310,17 @@ class Storage(context: Context, helper: SQLCipherOpenHelper) : Database(context,
     }
 
     override fun getReceivedMessageTimestamps(): Set<Long> {
-        TODO("Not yet implemented")
+        return SessionMetaProtocol.getTimestamps()
     }
 
     override fun addReceivedMessageTimestamp(timestamp: Long) {
-        TODO("Not yet implemented")
+        SessionMetaProtocol.addTimestamp(timestamp)
     }
 
+//    override fun removeReceivedMessageTimestamps(timestamps: Set<Long>) {
+//        TODO("Not yet implemented")
+//    }
+
     override fun getMessageIdInDatabase(timestamp: Long, author: String): Long? {
         val database = DatabaseFactory.getMmsSmsDatabase(context)
         val address = Address.fromSerialized(author)
@@ -314,8 +372,8 @@ class Storage(context: Context, helper: SQLCipherOpenHelper) : Database(context,
         return if (group.isPresent) { group.get() } else null
     }
 
-    override fun createGroup(groupId: String, title: String?, members: List<Address>, avatar: SignalServiceAttachmentPointer?, relay: String?, admins: List<Address>) {
+    override fun createGroup(groupId: String, title: String?, members: List<Address>, avatar: SignalServiceAttachmentPointer?, relay: String?, admins: List<Address>, formationTimestamp: Long) {
-        DatabaseFactory.getGroupDatabase(context).create(groupId, title, members, avatar, relay, admins)
+        DatabaseFactory.getGroupDatabase(context).create(groupId, title, members, avatar, relay, admins, formationTimestamp)
     }
 
     override fun setActive(groupID: String, value: Boolean) {

app/src/main/java/org/thoughtcrime/securesms/groups/GroupManager.java

@@ -77,7 +77,7 @@ public class GroupManager {
     String masterPublicKey = masterPublicKeyOrNull != null ? masterPublicKeyOrNull : TextSecurePreferences.getLocalNumber(context);
 
     memberAddresses.add(Address.Companion.fromSerialized(masterPublicKey));
-    groupDatabase.create(groupId, name, new LinkedList<>(memberAddresses), null, null, new LinkedList<>(adminAddresses));
+    groupDatabase.create(groupId, name, new LinkedList<>(memberAddresses), null, null, new LinkedList<>(adminAddresses), System.currentTimeMillis());
 
     groupDatabase.updateProfilePicture(groupId, avatarBytes);
     DatabaseFactory.getRecipientDatabase(context).setProfileSharing(groupRecipient, true);
@@ -104,7 +104,7 @@ public class GroupManager {
     final Set<Address>  memberAddresses = new HashSet<>();
 
     memberAddresses.add(Address.Companion.fromSerialized(Objects.requireNonNull(TextSecurePreferences.getLocalNumber(context))));
-    groupDatabase.create(groupId, name, new LinkedList<>(memberAddresses), null, null, new LinkedList<>());
+    groupDatabase.create(groupId, name, new LinkedList<>(memberAddresses), null, null, new LinkedList<>(), System.currentTimeMillis());
 
     groupDatabase.updateProfilePicture(groupId, avatarBytes);
 

app/src/main/java/org/thoughtcrime/securesms/groups/GroupMessageProcessor.java

@@ -69,7 +69,7 @@ public class GroupMessageProcessor {
     if (record.isPresent() && group.getType() == Type.UPDATE) {
       return handleGroupUpdate(context, content, group, record.get(), outgoing);
     } else if (!record.isPresent() && group.getType() == Type.UPDATE) {
-      return handleGroupCreate(context, content, group, outgoing);
+      return handleGroupCreate(context, content, group, outgoing, message.getTimestamp());
     } else if (record.isPresent() && group.getType() == Type.QUIT) {
       return handleGroupLeave(context, content, group, record.get(), outgoing);
     } else if (record.isPresent() && group.getType() == Type.REQUEST_INFO) {
@@ -83,7 +83,8 @@ public class GroupMessageProcessor {
   private static @Nullable Long handleGroupCreate(@NonNull Context context,
                                                   @NonNull SignalServiceContent content,
                                                   @NonNull SignalServiceGroup group,
-                                                  boolean outgoing)
+                                                  boolean outgoing,
+                                                  Long formationTimestamp)
   {
     GroupDatabase        database = DatabaseFactory.getGroupDatabase(context);
     String               id       = GroupUtil.getEncodedId(group);
@@ -108,7 +109,7 @@ public class GroupMessageProcessor {
     }
 
     database.create(id, group.getName().orNull(), members,
-                    avatar != null && avatar.isPointer() ? avatar.asPointer() : null, null, admins);
+                    avatar != null && avatar.isPointer() ? avatar.asPointer() : null, null, admins, formationTimestamp);
 
     if (group.getMembers().isPresent()) {
       ClosedGroupsProtocol.establishSessionsWithMembersIfNeeded(context, group.getMembers().get());

app/src/main/java/org/thoughtcrime/securesms/loki/protocol/ClosedGroupsProtocolV2.kt

@@ -69,7 +69,7 @@ object ClosedGroupsProtocolV2 {
             val admins = setOf( userPublicKey )
             val adminsAsData = admins.map { Hex.fromStringCondensed(it) }
             DatabaseFactory.getGroupDatabase(context).create(groupID, name, LinkedList(members.map { Address.fromSerialized(it) }),
-                    null, null, LinkedList(admins.map { Address.fromSerialized(it!!) }))
+                    null, null, LinkedList(admins.map { Address.fromSerialized(it!!) }), System.currentTimeMillis())
             DatabaseFactory.getRecipientDatabase(context).setProfileSharing(Recipient.from(context, Address.fromSerialized(groupID), false), true)
             // Send a closed group update message to all members individually
             // Add the group to the user's set of public keys to poll for
@@ -463,7 +463,7 @@ object ClosedGroupsProtocolV2 {
             groupDB.updateMembers(groupID, members.map { Address.fromSerialized(it) })
         } else {
             groupDB.create(groupID, name, LinkedList(members.map { Address.fromSerialized(it) }),
-                    null, null, LinkedList(admins.map { Address.fromSerialized(it) }))
+                    null, null, LinkedList(admins.map { Address.fromSerialized(it) }), sentTimestamp)
         }
         DatabaseFactory.getRecipientDatabase(context).setProfileSharing(Recipient.from(context, Address.fromSerialized(groupID), false), true)
         // Add the group to the user's set of public keys to poll for
@@ -716,7 +716,7 @@ object ClosedGroupsProtocolV2 {
                                    senderPublicKey: String): Boolean  {
         val oldMembers = group.members.map { it.serialize() }
         // Check that the message isn't from before the group was created
-        if (group.createdAt > sentTimestamp) {
+        if (group.formationTimestamp > sentTimestamp) {
             Log.d("Loki", "Ignoring closed group update from before thread was created.")
             return false
         }

app/src/main/java/org/thoughtcrime/securesms/loki/protocol/MultiDeviceProtocol.kt

@@ -19,6 +19,7 @@ import java.util.*
 
 object MultiDeviceProtocol {
 
+    // TODO: refactor this to use new message sending job
     @JvmStatic
     fun syncConfigurationIfNeeded(context: Context) {
         val userPublicKey = TextSecurePreferences.getLocalNumber(context) ?: return
@@ -41,6 +42,7 @@ object MultiDeviceProtocol {
         }
     }
 
+    // TODO: refactor this to use new message sending job
     fun forceSyncConfigurationNowIfNeeded(context: Context) {
         val userPublicKey = TextSecurePreferences.getLocalNumber(context) ?: return
         val configurationMessage = ConfigurationMessage.getCurrent()
@@ -58,6 +60,7 @@ object MultiDeviceProtocol {
         }
     }
 
+    // TODO: remove this after we migrate to new message receiving pipeline
     @JvmStatic
     fun handleConfigurationMessage(context: Context, content: SignalServiceProtos.Content, senderPublicKey: String, timestamp: Long) {
         if (TextSecurePreferences.getConfigurationMessageSynced(context)) return

app/src/main/java/org/thoughtcrime/securesms/loki/protocol/SessionMetaProtocol.kt

@@ -21,6 +21,14 @@ object SessionMetaProtocol {
         timestamps.remove(timestamp)
     }
 
+    fun getTimestamps(): Set<Long> {
+        return timestamps
+    }
+
+    fun addTimestamp(timestamp: Long) {
+        timestamps.add(timestamp)
+    }
+
     @JvmStatic
     fun shouldIgnoreMessage(timestamp: Long): Boolean {
         val shouldIgnoreMessage = timestamps.contains(timestamp)

app/src/main/java/org/thoughtcrime/securesms/mms/IncomingMediaMessage.java

@@ -1,15 +1,18 @@
 package org.thoughtcrime.securesms.mms;
 
+import org.session.libsession.messaging.messages.visible.VisibleMessage;
 import org.session.libsession.messaging.sending_receiving.attachments.Attachment;
 import org.session.libsession.messaging.sending_receiving.attachments.PointerAttachment;
 import org.session.libsession.messaging.sending_receiving.sharecontacts.Contact;
 import org.session.libsession.messaging.threads.Address;
 import org.session.libsession.messaging.sending_receiving.linkpreview.LinkPreview;
 import org.session.libsession.messaging.sending_receiving.quotes.QuoteModel;
+import org.session.libsession.messaging.threads.recipients.Recipient;
 import org.session.libsession.utilities.GroupUtil;
 import org.session.libsignal.libsignal.util.guava.Optional;
 import org.session.libsignal.service.api.messages.SignalServiceAttachment;
 import org.session.libsignal.service.api.messages.SignalServiceGroup;
+import org.thoughtcrime.securesms.ApplicationContext;
 
 import java.util.Collections;
 import java.util.LinkedList;
@@ -92,6 +95,18 @@ public class IncomingMediaMessage {
     }
   }
 
+  public static IncomingMediaMessage from(VisibleMessage message,
+                                          Address from,
+                                          long expiresIn,
+                                          Optional<SignalServiceGroup> group,
+                                          Optional<List<SignalServiceAttachment>> attachments,
+                                          Optional<QuoteModel> quote,
+                                          Optional<List<LinkPreview>> linkPreviews)
+  {
+    return new IncomingMediaMessage(from, message.getReceivedTimestamp(), -1, expiresIn, false,
+            false, Optional.fromNullable(message.getText()), group, attachments, quote, Optional.absent(), linkPreviews, Optional.absent());
+  }
+
   public int getSubscriptionId() {
     return subscriptionId;
   }

app/src/main/java/org/thoughtcrime/securesms/mms/OutgoingMediaMessage.java

@@ -4,6 +4,8 @@ import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;
 import android.text.TextUtils;
 
+import org.session.libsession.messaging.messages.visible.VisibleMessage;
+import org.thoughtcrime.securesms.database.ThreadDatabase;
 import org.thoughtcrime.securesms.database.documents.IdentityKeyMismatch;
 import org.thoughtcrime.securesms.database.documents.NetworkFailure;
 import org.session.libsession.messaging.sending_receiving.attachments.Attachment;
@@ -12,6 +14,7 @@ import org.session.libsession.messaging.sending_receiving.linkpreview.LinkPrevie
 import org.session.libsession.messaging.sending_receiving.quotes.QuoteModel;
 import org.session.libsession.messaging.threads.recipients.Recipient;
 
+import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 
@@ -86,6 +89,17 @@ public class OutgoingMediaMessage {
     this.linkPreviews.addAll(that.linkPreviews);
   }
 
+  public static OutgoingMediaMessage from(VisibleMessage message,
+                                          Recipient recipient,
+                                          List<Attachment> attachments,
+                                          @Nullable QuoteModel outgoingQuote,
+                                          @NonNull List<LinkPreview> linkPreviews)
+  {
+    return new OutgoingMediaMessage(recipient, message.getText(), attachments, message.getSentTimestamp(), -1,
+            recipient.getExpireMessages() * 1000, ThreadDatabase.DistributionTypes.DEFAULT, outgoingQuote, Collections.emptyList(),
+            linkPreviews, Collections.emptyList(), Collections.emptyList());
+  }
+
   public Recipient getRecipient() {
     return recipient;
   }

app/src/main/java/org/thoughtcrime/securesms/sms/IncomingTextMessage.java

@@ -7,6 +7,7 @@ import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;
 import android.telephony.SmsMessage;
 
+import org.session.libsession.messaging.messages.visible.VisibleMessage;
 import org.session.libsession.messaging.threads.Address;
 import org.session.libsession.utilities.GroupUtil;
 import org.session.libsignal.libsignal.util.guava.Optional;
@@ -155,6 +156,14 @@ public class IncomingTextMessage implements Parcelable {
     this.unidentified         = false;
   }
 
+  public static IncomingTextMessage from(VisibleMessage message,
+                                         Address sender,
+                                         Optional<SignalServiceGroup> group,
+                                         long expiresInMillis)
+  {
+    return new IncomingTextMessage(sender, 1, message.getReceivedTimestamp(), message.getText(), group, expiresInMillis, false);
+  }
+
   public int getSubscriptionId() {
     return subscriptionId;
   }

app/src/main/java/org/thoughtcrime/securesms/sms/OutgoingTextMessage.java

@@ -1,5 +1,6 @@
 package org.thoughtcrime.securesms.sms;
 
+import org.session.libsession.messaging.messages.visible.VisibleMessage;
 import org.thoughtcrime.securesms.database.model.SmsMessageRecord;
 import org.session.libsession.messaging.threads.recipients.Recipient;
 
@@ -28,6 +29,10 @@ public class OutgoingTextMessage {
     this.message        = body;
   }
 
+  public static OutgoingTextMessage from(VisibleMessage message, Recipient recipient) {
+    return new OutgoingTextMessage(recipient, message.getText(), recipient.getExpireMessages() * 1000, -1);
+  }
+
   public long getExpiresIn() {
     return expiresIn;
   }

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

@@ -6,6 +6,7 @@ import org.session.libsession.messaging.sending_receiving.attachments.SessionSer
 import org.session.libsession.messaging.sending_receiving.attachments.SessionServiceAttachmentStream
 import org.session.libsession.messaging.threads.Address
 import org.session.libsignal.service.api.messages.SignalServiceAttachmentPointer
+import org.session.libsignal.service.api.messages.SignalServiceAttachmentStream
 import java.io.InputStream
 
 interface MessageDataProvider {
@@ -14,9 +15,11 @@ interface MessageDataProvider {
     fun deleteMessage(messageID: Long)
 
     fun getAttachmentStream(attachmentId: Long): SessionServiceAttachmentStream?
-
     fun getAttachmentPointer(attachmentId: Long): SessionServiceAttachmentPointer?
 
+    fun getSignalAttachmentStream(attachmentId: Long): SignalServiceAttachmentStream?
+    fun getSignalAttachmentPointer(attachmentId: Long): SignalServiceAttachmentPointer?
+
     fun setAttachmentState(attachmentState: AttachmentState, attachmentId: Long, messageID: Long)
 
     fun insertAttachment(messageId: Long, attachmentId: Long, stream : InputStream)

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

@@ -86,8 +86,10 @@ interface StorageProtocol {
     fun removeLastDeletionServerID(group: Long, server: String)
 
     // Message Handling
+    fun isMessageDuplicated(timestamp: Long, sender: String): Boolean
     fun getReceivedMessageTimestamps(): Set<Long>
     fun addReceivedMessageTimestamp(timestamp: Long)
+//    fun removeReceivedMessageTimestamps(timestamps: Set<Long>)
     // Returns the IDs of the saved attachments.
     fun persistAttachments(messageId: Long, attachments: List<Attachment>): List<Long>
 
@@ -99,7 +101,7 @@ interface StorageProtocol {
 
     // Closed Groups
     fun getGroup(groupID: String): GroupRecord?
-    fun createGroup(groupID: String, title: String?, members: List<Address>, avatar: SignalServiceAttachmentPointer?, relay: String?, admins: List<Address>)
+    fun createGroup(groupID: String, title: String?, members: List<Address>, avatar: SignalServiceAttachmentPointer?, relay: String?, admins: List<Address>, formationTimestamp: Long)
     fun setActive(groupID: String, value: Boolean)
     fun removeMember(groupID: String, member: Address)
     fun updateMembers(groupID: String, members: List<Address>)

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/messages/visible/VisibleMessage.kt

@@ -10,6 +10,7 @@ import org.session.libsignal.service.internal.push.SignalServiceProtos
 
 class VisibleMessage : Message()  {
 
+    var syncTarget: String? = null
     var text: String? = null
     var attachmentIDs = ArrayList<Long>()
     var quote: Quote? = null
@@ -17,12 +18,15 @@ class VisibleMessage : Message()  {
     var contact: Contact? = null
     var profile: Profile? = null
 
+    override val isSelfSendValid: Boolean = true
+
     companion object {
         const val TAG = "VisibleMessage"
 
         fun fromProto(proto: SignalServiceProtos.Content): VisibleMessage? {
             val dataMessage = proto.dataMessage ?: return null
             val result = VisibleMessage()
+            result.syncTarget = dataMessage.syncTarget
             result.text = dataMessage.body
             // Attachments are handled in MessageReceiver
             val quoteProto = dataMessage.quote
@@ -103,6 +107,10 @@ class VisibleMessage : Message()  {
         }
         val attachmentProtos = attachments.mapNotNull { it.toProto() }
         dataMessage.addAllAttachments(attachmentProtos)
+        // Sync target
+        if (syncTarget != null) {
+            dataMessage.syncTarget = syncTarget
+        }
         // TODO Contact
         // Build
         try {

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.isMessageDuplicated(envelope.timestamp, envelope.source) && !isRetry) throw Error.DuplicateMessage
         storage.addReceivedMessageTimestamp(envelope.timestamp)
         // Decrypt the contents
         val ciphertext = envelope.content ?: throw Error.NoData
@@ -60,7 +63,7 @@ object MessageReceiver {
         } else {
             when (envelope.type) {
                 SignalServiceProtos.Envelope.Type.UNIDENTIFIED_SENDER -> {
-                    val userX25519KeyPair = MessagingConfiguration.shared.storage.getUserX25519KeyPair() ?: throw Error.NoUserX25519KeyPair
+                    val userX25519KeyPair = MessagingConfiguration.shared.storage.getUserX25519KeyPair()
                     val decryptionResult = MessageReceiverDecryption.decryptWithSessionProtocol(ciphertext.toByteArray(), userX25519KeyPair)
                     plaintext = decryptionResult.first
                     sender = decryptionResult.second
@@ -89,16 +92,28 @@ object MessageReceiver {
                             }
                         }
                     }
-                    decrypt()
                     groupPublicKey = envelope.source
+                    decrypt()
+//                    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 +121,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

@@ -103,18 +103,21 @@ 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) {
         if (allClosedGroupPublicKeys.contains(closeGroup.publicKey)) continue
-        handleNewClosedGroup(message.sender!!, closeGroup.publicKey, closeGroup.name, closeGroup.encryptionKeyPair, closeGroup.members, closeGroup.admins)
+        handleNewClosedGroup(message.sender!!, closeGroup.publicKey, closeGroup.name, closeGroup.encryptionKeyPair, closeGroup.members, closeGroup.admins, message.sentTimestamp!!)
     }
     val allOpenGroups = storage.getAllOpenGroups().map { it.value.server }
     for (openGroup in message.openGroups) {
         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 +158,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 +212,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)
     }
 }
 
@@ -217,11 +221,11 @@ private fun MessageReceiver.handleNewClosedGroup(message: ClosedGroupControlMess
     val groupPublicKey = kind.publicKey.toByteArray().toHexString()
     val members = kind.members.map { it.toByteArray().toHexString() }
     val admins = kind.admins.map { it.toByteArray().toHexString() }
-    handleNewClosedGroup(message.sender!!, groupPublicKey, kind.name, kind.encryptionKeyPair, members, admins)
+    handleNewClosedGroup(message.sender!!, groupPublicKey, kind.name, kind.encryptionKeyPair, members, admins, message.sentTimestamp!!)
 }
 
 // Parameter @sender:String is just for inserting incoming info message
-private fun handleNewClosedGroup(sender: String, groupPublicKey: String, name: String, encryptionKeyPair: ECKeyPair, members: List<String>, admins: List<String>) {
+private fun handleNewClosedGroup(sender: String, groupPublicKey: String, name: String, encryptionKeyPair: ECKeyPair, members: List<String>, admins: List<String>, formationTimestamp: Long) {
     val context = MessagingConfiguration.shared.context
     val storage = MessagingConfiguration.shared.storage
     // Create the group
@@ -232,15 +236,15 @@ private fun handleNewClosedGroup(sender: String, groupPublicKey: String, name: S
         storage.updateMembers(groupID, members.map { Address.fromSerialized(it) })
     } else {
         storage.createGroup(groupID, name, LinkedList(members.map { Address.fromSerialized(it) }),
-                            null, null, LinkedList(admins.map { Address.fromSerialized(it) }))
+                            null, null, LinkedList(admins.map { Address.fromSerialized(it) }), formationTimestamp)
+        // 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 +304,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 +313,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 +324,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")
 }
@@ -368,12 +377,16 @@ private fun MessageReceiver.handleClosedGroupMembersAdded(message: ClosedGroupCo
     val members = group.members.map { it.serialize() }
     val admins = group.admins.map { it.serialize() }
 
-    // Users that are part of this remove update
     val updateMembers = kind.members.map { it.toByteArray().toHexString() }
-    // newMembers to save is old members minus removed members
     val newMembers = members + updateMembers
     storage.updateMembers(groupID, newMembers.map { Address.fromSerialized(it) })
-
+    // Send the latest encryption key pair to the added members if the current user is the admin of the group
+    val isCurrentUserAdmin = admins.contains(storage.getUserPublicKey()!!)
+    if (isCurrentUserAdmin) {
+        for (member in updateMembers) {
+            MessageSender.sendLatestEncryptionKeyPair(member, groupPublicKey)
+        }
+    }
     storage.insertIncomingInfoMessage(context, senderPublicKey, groupID, SignalServiceProtos.GroupContext.Type.UPDATE, SignalServiceGroup.Type.UPDATE, name, members, admins)
 }
 
@@ -431,8 +444,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,12 +474,31 @@ 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 }
+    MessageSender.sendLatestEncryptionKeyPair(senderPublicKey, groupPublicKey)
+}
+
 private fun isValidGroupUpdate(group: GroupRecord,
                                sentTimestamp: Long,
                                senderPublicKey: String): Boolean  {
     val oldMembers = group.members.map { it.serialize() }
     // Check that the message isn't from before the group was created
-    if (group.createdAt > sentTimestamp) {
+    if (group.formationTimestamp > sentTimestamp) {
         android.util.Log.d("Loki", "Ignoring closed group update from before thread was created.")
         return false
     }

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

@@ -2,32 +2,30 @@
 
 package org.session.libsession.messaging.sending_receiving
 
-import android.util.Log
 import com.google.protobuf.ByteString
 import nl.komponents.kovenant.Promise
 import nl.komponents.kovenant.deferred
 
 import org.session.libsession.messaging.MessagingConfiguration
 import org.session.libsession.messaging.messages.control.ClosedGroupControlMessage
-import org.session.libsession.messaging.messages.control.ClosedGroupUpdate
 import org.session.libsession.messaging.sending_receiving.notifications.PushNotificationAPI
 import org.session.libsession.messaging.sending_receiving.MessageSender.Error
 import org.session.libsession.messaging.threads.Address
-import org.session.libsession.messaging.threads.recipients.Recipient
 import org.session.libsession.utilities.GroupUtil
-import org.session.libsession.utilities.TextSecurePreferences
 import org.session.libsignal.utilities.Hex
 
 import org.session.libsignal.libsignal.ecc.Curve
+import org.session.libsignal.libsignal.ecc.ECKeyPair
+import org.session.libsignal.libsignal.util.guava.Optional
 import org.session.libsignal.service.internal.push.SignalServiceProtos
-import org.session.libsignal.service.loki.protocol.closedgroups.ClosedGroupRatchetCollectionType
-import org.session.libsignal.service.loki.protocol.closedgroups.ClosedGroupSenderKey
-import org.session.libsignal.service.loki.protocol.closedgroups.SharedSenderKeysImplementation
-import org.session.libsignal.service.loki.utilities.hexEncodedPrivateKey
 import org.session.libsignal.service.loki.utilities.hexEncodedPublicKey
 import org.session.libsignal.service.loki.utilities.removing05PrefixIfNeeded
 import org.session.libsignal.utilities.ThreadUtils
+import org.session.libsignal.utilities.logging.Log
 import java.util.*
+import java.util.concurrent.ConcurrentHashMap
+
+private val pendingKeyPair = ConcurrentHashMap<String, Optional<ECKeyPair>>()
 
 fun MessageSender.createClosedGroup(name: String, members: Collection<String>): Promise<String, Exception> {
     val deferred = deferred<String, Exception>()
@@ -46,12 +44,11 @@ fun MessageSender.createClosedGroup(name: String, members: Collection<String>):
         val admins = setOf( userPublicKey )
         val adminsAsData = admins.map { ByteString.copyFrom(Hex.fromStringCondensed(it)) }
         storage.createGroup(groupID, name, LinkedList(members.map { Address.fromSerialized(it) }),
-                null, null, LinkedList(admins.map { Address.fromSerialized(it) }))
+                null, null, LinkedList(admins.map { Address.fromSerialized(it) }), System.currentTimeMillis())
         storage.setProfileSharing(Address.fromSerialized(groupID), true)
         // 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 +157,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,151 +209,6 @@ fun MessageSender.v2_leave(groupPublicKey: String) {
     storage.insertOutgoingInfoMessage(context, groupID, infoType, name, updatedMembers, admins, threadID)
 }
 
-fun MessageSender.update(groupPublicKey: String, members: Collection<String>, name: String): Promise<Unit, Exception> {
-    val deferred = deferred<Unit, Exception>()
-    val context = MessagingConfiguration.shared.context
-    val storage = MessagingConfiguration.shared.storage
-    val userPublicKey = storage.getUserPublicKey()!!
-    val sskDatabase = MessagingConfiguration.shared.sskDatabase
-    val groupID = GroupUtil.getEncodedClosedGroupID(GroupUtil.getEncodedClosedGroupID(Hex.fromStringCondensed(groupPublicKey)).toByteArray()) // double encoded
-    val group = storage.getGroup(groupID)
-    if (group == null) {
-        Log.d("Loki", "Can't update nonexistent closed group.")
-        deferred.reject(Error.NoThread)
-        return deferred.promise
-    }
-    val oldMembers = group.members.map { it.serialize() }.toSet()
-    val newMembers = members.minus(oldMembers)
-    val membersAsData = members.map { Hex.fromStringCondensed(it) }
-    val admins = group.admins.map { it.serialize() }
-    val adminsAsData = admins.map { Hex.fromStringCondensed(it) }
-    val groupPrivateKey = sskDatabase.getClosedGroupPrivateKey(groupPublicKey)
-    if (groupPrivateKey == null) {
-        Log.d("Loki", "Couldn't get private key for closed group.")
-        deferred.reject(Error.NoPrivateKey)
-        return deferred.promise
-    }
-    val wasAnyUserRemoved = members.toSet().intersect(oldMembers) != oldMembers.toSet()
-    val removedMembers = oldMembers.minus(members)
-    val isUserLeaving = removedMembers.contains(userPublicKey)
-    val newSenderKeys: List<ClosedGroupSenderKey>
-    if (wasAnyUserRemoved) {
-        if (isUserLeaving && removedMembers.count() != 1) {
-            Log.d("Loki", "Can't remove self and others simultaneously.")
-            deferred.reject(Error.InvalidClosedGroupUpdate)
-            return deferred.promise
-        }
-        // Send the update to the existing members using established channels (don't include new ratchets as everyone should regenerate new ratchets individually)
-        val promises = oldMembers.map { member ->
-            val closedGroupUpdateKind = ClosedGroupUpdate.Kind.Info(Hex.fromStringCondensed(groupPublicKey),
-                    name, setOf(), membersAsData, adminsAsData)
-            val closedGroupUpdate = ClosedGroupUpdate()
-            closedGroupUpdate.kind = closedGroupUpdateKind
-            val address = Address.fromSerialized(member)
-            MessageSender.sendNonDurably(closedGroupUpdate, address).get()
-        }
-
-        val allOldRatchets = sskDatabase.getAllClosedGroupRatchets(groupPublicKey, ClosedGroupRatchetCollectionType.Current)
-        for (pair in allOldRatchets) {
-            val senderPublicKey = pair.first
-            val ratchet = pair.second
-            val collection = ClosedGroupRatchetCollectionType.Old
-            sskDatabase.setClosedGroupRatchet(groupPublicKey, senderPublicKey, ratchet, collection)
-        }
-        // Delete all ratchets (it's important that this happens * after * sending out the update)
-        sskDatabase.removeAllClosedGroupRatchets(groupPublicKey, ClosedGroupRatchetCollectionType.Current)
-        // Remove the group from the user's set of public keys to poll for if the user is leaving. Otherwise generate a new ratchet and
-        // send it out to all members (minus the removed ones) using established channels.
-        if (isUserLeaving) {
-            sskDatabase.removeClosedGroupPrivateKey(groupPublicKey)
-            storage.setActive(groupID, false)
-            storage.removeMember(groupID, Address.fromSerialized(userPublicKey))
-            // Notify the PN server
-            PushNotificationAPI.performOperation(PushNotificationAPI.ClosedGroupOperation.Unsubscribe, groupPublicKey, userPublicKey)
-        } else {
-            // Send closed group update messages to any new members using established channels
-            for (member in newMembers) {
-                val closedGroupUpdateKind = ClosedGroupUpdate.Kind.New(Hex.fromStringCondensed(groupPublicKey), name,
-                        Hex.fromStringCondensed(groupPrivateKey), listOf(), membersAsData, adminsAsData)
-                val closedGroupUpdate = ClosedGroupUpdate()
-                closedGroupUpdate.kind = closedGroupUpdateKind
-                val address = Address.fromSerialized(member)
-                MessageSender.sendNonDurably(closedGroupUpdate, address)
-            }
-            // Send out the user's new ratchet to all members (minus the removed ones) using established channels
-            val userRatchet = SharedSenderKeysImplementation.shared.generateRatchet(groupPublicKey, userPublicKey)
-            val userSenderKey = ClosedGroupSenderKey(Hex.fromStringCondensed(userRatchet.chainKey), userRatchet.keyIndex, Hex.fromStringCondensed(userPublicKey))
-            for (member in members) {
-                if (member == userPublicKey) { continue }
-                val closedGroupUpdateKind = ClosedGroupUpdate.Kind.SenderKey(Hex.fromStringCondensed(groupPublicKey), userSenderKey)
-                val closedGroupUpdate = ClosedGroupUpdate()
-                closedGroupUpdate.kind = closedGroupUpdateKind
-                val address = Address.fromSerialized(member)
-                MessageSender.sendNonDurably(closedGroupUpdate, address)
-            }
-        }
-    } else if (newMembers.isNotEmpty()) {
-        // Generate ratchets for any new members
-        newSenderKeys = newMembers.map { publicKey ->
-            val ratchet = SharedSenderKeysImplementation.shared.generateRatchet(groupPublicKey, publicKey)
-            ClosedGroupSenderKey(Hex.fromStringCondensed(ratchet.chainKey), ratchet.keyIndex, Hex.fromStringCondensed(publicKey))
-        }
-        // Send a closed group update message to the existing members with the new members' ratchets (this message is aimed at the group)
-        val closedGroupUpdateKind = ClosedGroupUpdate.Kind.Info(Hex.fromStringCondensed(groupPublicKey), name,
-                newSenderKeys, membersAsData, adminsAsData)
-        val closedGroupUpdate = ClosedGroupUpdate()
-        closedGroupUpdate.kind = closedGroupUpdateKind
-        val address = Address.fromSerialized(groupID)
-        MessageSender.send(closedGroupUpdate, address)
-        // Send closed group update messages to the new members using established channels
-        var allSenderKeys = sskDatabase.getAllClosedGroupSenderKeys(groupPublicKey, ClosedGroupRatchetCollectionType.Current)
-        allSenderKeys = allSenderKeys.union(newSenderKeys)
-        for (member in newMembers) {
-            val closedGroupUpdateKind = ClosedGroupUpdate.Kind.New(Hex.fromStringCondensed(groupPublicKey), name,
-                    Hex.fromStringCondensed(groupPrivateKey), allSenderKeys, membersAsData, adminsAsData)
-            val closedGroupUpdate = ClosedGroupUpdate()
-            closedGroupUpdate.kind = closedGroupUpdateKind
-            val address = Address.fromSerialized(member)
-            MessageSender.send(closedGroupUpdate, address)
-        }
-    } else {
-        val allSenderKeys = sskDatabase.getAllClosedGroupSenderKeys(groupPublicKey, ClosedGroupRatchetCollectionType.Current)
-        val closedGroupUpdateKind = ClosedGroupUpdate.Kind.Info(Hex.fromStringCondensed(groupPublicKey), name,
-                allSenderKeys, membersAsData, adminsAsData)
-        val closedGroupUpdate = ClosedGroupUpdate()
-        closedGroupUpdate.kind = closedGroupUpdateKind
-        val address = Address.fromSerialized(groupID)
-        MessageSender.send(closedGroupUpdate, address)
-    }
-    // Update the group
-    storage.updateTitle(groupID, name)
-    if (!isUserLeaving) {
-        // The call below sets isActive to true, so if the user is leaving we have to use groupDB.remove(...) instead
-        storage.updateMembers(groupID, members.map { Address.fromSerialized(it) })
-    }
-    // Notify the user
-    val infoType = if (isUserLeaving) SignalServiceProtos.GroupContext.Type.QUIT else SignalServiceProtos.GroupContext.Type.UPDATE
-    val threadID = storage.getOrCreateThreadIdFor(Address.fromSerialized(groupID))
-    storage.insertOutgoingInfoMessage(context, groupID, infoType, name, members, admins, threadID)
-    deferred.resolve(Unit)
-    return deferred.promise
-}
-
-fun MessageSender.leave(groupPublicKey: String) {
-    val storage = MessagingConfiguration.shared.storage
-    val userPublicKey = storage.getUserPublicKey()!!
-    val groupID = GroupUtil.getEncodedClosedGroupID(GroupUtil.getEncodedClosedGroupID(Hex.fromStringCondensed(groupPublicKey)).toByteArray()) // double encoded
-    val group = storage.getGroup(groupID)
-    if (group == null) {
-        Log.d("Loki", "Can't leave nonexistent closed group.")
-        return
-    }
-    val name = group.title
-    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
@@ -372,6 +224,11 @@ fun MessageSender.generateAndSendNewEncryptionKeyPair(groupPublicKey: String, ta
     }
     // Generate the new encryption key pair
     val newKeyPair = Curve.generateKeyPair()
+    // replace call will not succeed if no value already set
+    pendingKeyPair.putIfAbsent(groupPublicKey,Optional.absent())
+    do {
+        // make sure we set the pendingKeyPair or wait until it is not null
+    } while (!pendingKeyPair.replace(groupPublicKey,Optional.absent(),Optional.fromNullable(newKeyPair)))
     // Distribute it
     val proto = SignalServiceProtos.KeyPair.newBuilder()
     proto.publicKey = ByteString.copyFrom(newKeyPair.publicKey.serialize().removing05PrefixIfNeeded())
@@ -381,10 +238,54 @@ fun MessageSender.generateAndSendNewEncryptionKeyPair(groupPublicKey: String, ta
         val ciphertext = MessageSenderEncryption.encryptWithSessionProtocol(plaintext, publicKey)
         ClosedGroupControlMessage.KeyPairWrapper(publicKey, ByteString.copyFrom(ciphertext))
     }
-    val kind = ClosedGroupControlMessage.Kind.EncryptionKeyPair(wrappers)
+    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)
+        pendingKeyPair[groupPublicKey] = Optional.absent()
     }
 }
+
+/// Note: Shouldn't currently be in use.
+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))
+}
+
+fun MessageSender.sendLatestEncryptionKeyPair(publicKey: String, groupPublicKey: String) {
+    val storage = MessagingConfiguration.shared.storage
+    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
+    val group = storage.getGroup(groupID) ?: run {
+        Log.d("Loki", "Can't send encryption key pair for nonexistent closed group.")
+        throw Error.NoThread
+    }
+    val members = group.members.map { it.serialize() }
+    if (!members.contains(publicKey)) {
+        Log.d("Loki", "Refusing to send latest encryption key pair to non-member.")
+        return
+    }
+    // Get the latest encryption key pair
+    val encryptionKeyPair = pendingKeyPair[groupPublicKey]?.orNull()
+            ?: 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, publicKey)
+    Log.d("Loki", "Sending latest encryption key pair to: $publicKey.")
+    val wrapper = ClosedGroupControlMessage.KeyPairWrapper(publicKey, ByteString.copyFrom(ciphertext))
+    val kind = ClosedGroupControlMessage.Kind.EncryptionKeyPair(ByteString.copyFrom(Hex.fromStringCondensed(groupPublicKey)), listOf(wrapper))
+    val closedGroupControlMessage = ClosedGroupControlMessage(kind)
+    MessageSender.send(closedGroupControlMessage, Address.fromSerialized(publicKey))
+}

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

libsession/src/main/java/org/session/libsession/messaging/threads/GroupRecord.kt

@@ -9,7 +9,7 @@ class GroupRecord(
         val encodedId: String, val title: String, members: String?, val avatar: ByteArray?,
         val avatarId: Long?, val avatarKey: ByteArray?, val avatarContentType: String?,
         val relay: String?, val isActive: Boolean, val avatarDigest: ByteArray?, val isMms: Boolean,
-        val url: String?, admins: String?, val createdAt: Long
+        val url: String?, admins: String?, val formationTimestamp: Long
 ) {
     var members: List<Address> = LinkedList<Address>()
     var admins: List<Address> = LinkedList<Address>()