Merge branch 'dev' of https://github.com/loki-project/session-android into open-group-invitations

app/build.gradle

@@ -8,7 +8,7 @@ buildscript {
         jcenter()
     }
     dependencies {
-        classpath 'com.android.tools.build:gradle:4.1.2'
+        classpath 'com.android.tools.build:gradle:4.1.3'
         classpath files('libs/gradle-witness.jar')
         classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlin_version"
         classpath "org.jetbrains.kotlin:kotlin-serialization:$kotlin_version"
@@ -158,8 +158,8 @@ dependencies {
     testImplementation 'org.robolectric:shadows-multidex:4.2'
 }
 
-def canonicalVersionCode = 158
+def canonicalVersionCode = 162
-def canonicalVersionName = "1.10.1"
+def canonicalVersionName = "1.10.3"
 
 def postFixSize = 10
 def abiPostFix = ['armeabi-v7a' : 1,

app/src/main/java/org/thoughtcrime/securesms/ApplicationContext.java

@@ -327,7 +327,6 @@ public class ApplicationContext extends MultiDexApplication implements Dependenc
                 .setJobStorage(new FastJobStorage(DatabaseFactory.getJobDatabase(this)))
                 .setDependencyInjector(this)
                 .build());
-        JobQueue.getShared().resumePendingJobs();
     }
 
     private void initializeDependencyInjection() {
@@ -455,7 +454,6 @@ public class ApplicationContext extends MultiDexApplication implements Dependenc
             poller.setUserPublicKey(userPublicKey);
             return;
         }
-        LokiAPIDatabase apiDB = DatabaseFactory.getLokiAPIDatabase(this);
         poller = new Poller();
         closedGroupPoller = new ClosedGroupPoller();
     }

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

@@ -193,8 +193,8 @@ class Storage(context: Context, helper: SQLCipherOpenHelper) : Database(context,
         DatabaseFactory.getSessionJobDatabase(context).markJobAsSucceeded(jobId)
     }
 
-    override fun markJobAsFailed(jobId: String) {
+    override fun markJobAsFailedPermanently(jobId: String) {
-        DatabaseFactory.getSessionJobDatabase(context).markJobAsFailed(jobId)
+        DatabaseFactory.getSessionJobDatabase(context).markJobAsFailedPermanently(jobId)
     }
 
     override fun getAllPendingJobs(type: String): Map<String, Job?> {
@@ -261,7 +261,7 @@ class Storage(context: Context, helper: SQLCipherOpenHelper) : Database(context,
         val database = databaseHelper.readableDatabase
         return database.get(LokiThreadDatabase.publicChatTable, "${LokiThreadDatabase.threadID} = ?", arrayOf(threadId)) { cursor ->
             val publicChatAsJson = cursor.getString(LokiThreadDatabase.publicChat)
-            OpenGroupV2.fromJson(publicChatAsJson)
+            OpenGroupV2.fromJSON(publicChatAsJson)
         }
     }
 
@@ -585,7 +585,7 @@ class Storage(context: Context, helper: SQLCipherOpenHelper) : Database(context,
         val database = DatabaseFactory.getThreadDatabase(context)
         if (!openGroupID.isNullOrEmpty()) {
             val recipient = Recipient.from(context, Address.fromSerialized(GroupUtil.getEncodedOpenGroupID(openGroupID.toByteArray())), false)
-            return database.getOrCreateThreadIdFor(recipient)
+            return database.getThreadIdIfExistsFor(recipient)
         } else if (!groupPublicKey.isNullOrEmpty()) {
             val recipient = Recipient.from(context, Address.fromSerialized(GroupUtil.doubleEncodeGroupID(groupPublicKey)), false)
             return database.getOrCreateThreadIdFor(recipient)

app/src/main/java/org/thoughtcrime/securesms/jobmanager/Job.java

@@ -5,7 +5,7 @@ import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;
 import androidx.annotation.WorkerThread;
 
-import org.session.libsession.messaging.jobs.Data;
+import org.session.libsession.messaging.utilities.Data;
 import org.session.libsignal.utilities.logging.Log;
 
 import java.util.LinkedList;

app/src/main/java/org/thoughtcrime/securesms/jobmanager/JobController.java

@@ -7,7 +7,7 @@ import androidx.annotation.WorkerThread;
 
 import com.annimon.stream.Stream;
 
-import org.session.libsession.messaging.jobs.Data;
+import org.session.libsession.messaging.utilities.Data;
 import org.thoughtcrime.securesms.jobmanager.persistence.ConstraintSpec;
 import org.thoughtcrime.securesms.jobmanager.persistence.DependencySpec;
 import org.thoughtcrime.securesms.jobmanager.persistence.FullSpec;

app/src/main/java/org/thoughtcrime/securesms/jobmanager/JobInstantiator.java

@@ -2,7 +2,7 @@ package org.thoughtcrime.securesms.jobmanager;
 
 import androidx.annotation.NonNull;
 
-import org.session.libsession.messaging.jobs.Data;
+import org.session.libsession.messaging.utilities.Data;
 
 import java.util.HashMap;
 import java.util.Map;

app/src/main/java/org/thoughtcrime/securesms/jobmanager/JobManager.java

@@ -5,7 +5,7 @@ import android.content.Intent;
 import android.os.Build;
 import androidx.annotation.NonNull;
 
-import org.session.libsession.messaging.jobs.Data;
+import org.session.libsession.messaging.utilities.Data;
 import org.thoughtcrime.securesms.jobmanager.impl.DefaultExecutorFactory;
 import org.thoughtcrime.securesms.jobmanager.impl.JsonDataSerializer;
 import org.thoughtcrime.securesms.jobmanager.persistence.JobStorage;

app/src/main/java/org/thoughtcrime/securesms/jobmanager/impl/JsonDataSerializer.java

@@ -2,7 +2,7 @@ package org.thoughtcrime.securesms.jobmanager.impl;
 
 import androidx.annotation.NonNull;
 
-import org.session.libsession.messaging.jobs.Data;
+import org.session.libsession.messaging.utilities.Data;
 import org.session.libsignal.utilities.logging.Log;
 import org.session.libsignal.utilities.JsonUtil;
 

app/src/main/java/org/thoughtcrime/securesms/jobs/AvatarDownloadJob.java

@@ -3,7 +3,7 @@ package org.thoughtcrime.securesms.jobs;
 import android.graphics.Bitmap;
 import androidx.annotation.NonNull;
 
-import org.session.libsession.messaging.jobs.Data;
+import org.session.libsession.messaging.utilities.Data;
 import org.session.libsession.utilities.DownloadUtilities;
 import org.session.libsignal.service.api.crypto.AttachmentCipherInputStream;
 import org.thoughtcrime.securesms.database.DatabaseFactory;

app/src/main/java/org/thoughtcrime/securesms/jobs/LocalBackupJob.java

@@ -2,7 +2,7 @@ package org.thoughtcrime.securesms.jobs;
 
 import androidx.annotation.NonNull;
 
-import org.session.libsession.messaging.jobs.Data;
+import org.session.libsession.messaging.utilities.Data;
 import org.session.libsignal.utilities.externalstorage.NoExternalStorageException;
 import org.thoughtcrime.securesms.jobmanager.Job;
 import org.session.libsignal.utilities.logging.Log;

app/src/main/java/org/thoughtcrime/securesms/jobs/RetrieveProfileAvatarJob.java

@@ -7,7 +7,7 @@ import android.text.TextUtils;
 import androidx.annotation.NonNull;
 
 import org.session.libsession.messaging.avatars.AvatarHelper;
-import org.session.libsession.messaging.jobs.Data;
+import org.session.libsession.messaging.utilities.Data;
 import org.session.libsession.messaging.threads.Address;
 import org.session.libsession.messaging.threads.recipients.Recipient;
 import org.session.libsession.utilities.DownloadUtilities;

app/src/main/java/org/thoughtcrime/securesms/jobs/TrimThreadJob.java

@@ -18,7 +18,7 @@ package org.thoughtcrime.securesms.jobs;
 
 import androidx.annotation.NonNull;
 
-import org.session.libsession.messaging.jobs.Data;
+import org.session.libsession.messaging.utilities.Data;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
 import org.thoughtcrime.securesms.jobmanager.Job;
 import org.session.libsignal.utilities.logging.Log;

app/src/main/java/org/thoughtcrime/securesms/jobs/UpdateApkJob.java

@@ -13,7 +13,7 @@ import androidx.annotation.Nullable;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-import org.session.libsession.messaging.jobs.Data;
+import org.session.libsession.messaging.utilities.Data;
 import org.thoughtcrime.securesms.jobmanager.Job;
 import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraint;
 import org.session.libsignal.utilities.logging.Log;

app/src/main/java/org/thoughtcrime/securesms/loki/activities/JoinPublicChatActivity.kt

@@ -195,7 +195,7 @@ class EnterChatURLFragment : Fragment() {
             chip.chipIcon = drawable
             chip.text = defaultGroup.name
             chip.setOnClickListener {
-                (requireActivity() as JoinPublicChatActivity).joinPublicChatIfPossible(defaultGroup.toJoinUrl())
+                (requireActivity() as JoinPublicChatActivity).joinPublicChatIfPossible(defaultGroup.joinURL)
             }
             defaultRoomsGridLayout.addView(chip)
         }

app/src/main/java/org/thoughtcrime/securesms/loki/api/BackgroundPollWorker.kt

@@ -8,7 +8,6 @@ import nl.komponents.kovenant.Promise
 import nl.komponents.kovenant.all
 import nl.komponents.kovenant.functional.map
 import org.session.libsession.messaging.jobs.MessageReceiveJob
-import org.session.libsession.messaging.open_groups.OpenGroup
 import org.session.libsession.messaging.open_groups.OpenGroupV2
 import org.session.libsession.messaging.sending_receiving.pollers.ClosedGroupPoller
 import org.session.libsession.messaging.sending_receiving.pollers.OpenGroupPoller
@@ -17,7 +16,6 @@ import org.session.libsession.snode.SnodeAPI
 import org.session.libsession.utilities.TextSecurePreferences
 import org.session.libsignal.utilities.logging.Log
 import org.thoughtcrime.securesms.database.DatabaseFactory
-import java.io.IOException
 import java.util.concurrent.TimeUnit
 
 class BackgroundPollWorker(val context: Context, params: WorkerParameters) : Worker(context, params) {
@@ -25,45 +23,23 @@ class BackgroundPollWorker(val context: Context, params: WorkerParameters) : Wor
     companion object {
         const val TAG = "BackgroundPollWorker"
 
-        private const val RETRY_ATTEMPTS = 3
-
-        @JvmStatic
-        fun scheduleInstant(context: Context) {
-            val workRequest = OneTimeWorkRequestBuilder<BackgroundPollWorker>()
-                    .setConstraints(Constraints.Builder()
-                            .setRequiredNetworkType(NetworkType.CONNECTED)
-                            .build()
-                    )
-                    .build()
-
-            WorkManager
-                    .getInstance(context)
-                    .enqueue(workRequest)
-        }
-
         @JvmStatic
         fun schedulePeriodic(context: Context) {
             Log.v(TAG, "Scheduling periodic work.")
-            val workRequest = PeriodicWorkRequestBuilder<BackgroundPollWorker>(15, TimeUnit.MINUTES)
+            val builder = PeriodicWorkRequestBuilder<BackgroundPollWorker>(5, TimeUnit.MINUTES)
-                    .setConstraints(Constraints.Builder()
+            builder.setConstraints(Constraints.Builder().setRequiredNetworkType(NetworkType.CONNECTED).build())
-                            .setRequiredNetworkType(NetworkType.CONNECTED)
+            val workRequest = builder.build()
-                            .build()
+            WorkManager.getInstance(context).enqueueUniquePeriodicWork(
-                    )
+                TAG,
-                    .build()
+                ExistingPeriodicWorkPolicy.REPLACE,
-
+                workRequest
-            WorkManager
+            )
-                    .getInstance(context)
-                    .enqueueUniquePeriodicWork(
-                            TAG,
-                            ExistingPeriodicWorkPolicy.KEEP,
-                            workRequest
-                    )
         }
     }
 
     override fun doWork(): Result {
         if (TextSecurePreferences.getLocalNumber(context) == null) {
-            Log.v(TAG, "Background poll is canceled due to the Session user is not set up yet.")
+            Log.v(TAG, "User not registered yet.")
             return Result.failure()
         }
 
@@ -71,43 +47,41 @@ class BackgroundPollWorker(val context: Context, params: WorkerParameters) : Wor
             Log.v(TAG, "Performing background poll.")
             val promises = mutableListOf<Promise<Unit, Exception>>()
 
-            // Private chats
+            // DMs
             val userPublicKey = TextSecurePreferences.getLocalNumber(context)!!
-            val privateChatsPromise = SnodeAPI.getMessages(userPublicKey).map { envelopes ->
+            val dmsPromise = SnodeAPI.getMessages(userPublicKey).map { envelopes ->
                 envelopes.map { envelope ->
+                    // FIXME: Using a job here seems like a bad idea...
                     MessageReceiveJob(envelope.toByteArray(), false).executeAsync()
                 }
             }
-            promises.addAll(privateChatsPromise.get())
+            promises.addAll(dmsPromise.get())
 
             // Closed groups
             promises.addAll(ClosedGroupPoller().pollOnce())
 
             // Open Groups
-            val openGroups = DatabaseFactory.getLokiThreadDatabase(context).getAllPublicChats().map { (_,chat)->
+            val openGroups = DatabaseFactory.getLokiThreadDatabase(context).getAllPublicChats().values
-                OpenGroup(chat.channel, chat.server, chat.displayName, chat.isDeletable)
-            }
             for (openGroup in openGroups) {
                 val poller = OpenGroupPoller(openGroup)
                 promises.add(poller.pollForNewMessages())
             }
 
-            val openGroupsV2 = DatabaseFactory.getLokiThreadDatabase(context).getAllV2OpenGroups().values.groupBy(OpenGroupV2::server)
+            val v2OpenGroups = DatabaseFactory.getLokiThreadDatabase(context).getAllV2OpenGroups().values.groupBy(OpenGroupV2::server)
 
-            openGroupsV2.values.map { groups ->
+            v2OpenGroups.values.map { groups ->
                 OpenGroupV2Poller(groups)
             }.forEach { poller ->
-                promises.add(poller.compactPoll(true).map{ /*Unit*/ })
+                promises.add(poller.compactPoll(true).map { })
             }
 
-            // Wait till all the promises get resolved
+            // Wait until all the promises are resolved
             all(promises).get()
 
             return Result.success()
         } catch (exception: Exception) {
-            Log.v(TAG, "Background poll failed due to error: ${exception.message}.", exception)
+            Log.e(TAG, "Background poll failed due to error: ${exception.message}.", exception)
-
+            return Result.retry()
-            return if (runAttemptCount < RETRY_ATTEMPTS) Result.retry() else Result.failure()
         }
     }
 
@@ -116,8 +90,7 @@ class BackgroundPollWorker(val context: Context, params: WorkerParameters) : Wor
         override fun onReceive(context: Context, intent: Intent) {
             if (intent.action == Intent.ACTION_BOOT_COMPLETED) {
                 Log.v(TAG, "Boot broadcast caught.")
-                BackgroundPollWorker.scheduleInstant(context)
+                schedulePeriodic(context)
-                BackgroundPollWorker.schedulePeriodic(context)
             }
         }
     }

app/src/main/java/org/thoughtcrime/securesms/loki/api/PrepareAttachmentAudioExtrasJob.kt

@@ -5,7 +5,7 @@ import android.os.Build
 import org.session.libsignal.utilities.logging.Log
 import androidx.annotation.RequiresApi
 import org.greenrobot.eventbus.EventBus
-import org.session.libsession.messaging.jobs.Data
+import org.session.libsession.messaging.utilities.Data
 import org.session.libsession.messaging.sending_receiving.attachments.Attachment
 import org.session.libsession.messaging.sending_receiving.attachments.AttachmentId
 import org.session.libsession.messaging.sending_receiving.attachments.DatabaseAttachmentAudioExtras

app/src/main/java/org/thoughtcrime/securesms/loki/api/PublicChatManager.kt

@@ -31,7 +31,7 @@ class PublicChatManager(private val context: Context) {
     refreshChatsAndPollers()
     for ((threadID, _) in chats) {
       val poller = pollers[threadID]
-      areAllCaughtUp = if (poller != null) areAllCaughtUp && poller.isCaughtUp else true
+      areAllCaughtUp = if (poller != null) areAllCaughtUp && poller.isCaughtUp else areAllCaughtUp
     }
     return areAllCaughtUp
   }
@@ -42,6 +42,9 @@ class PublicChatManager(private val context: Context) {
       val poller = pollers[threadID] ?: OpenGroupPoller(chat, executorService)
       poller.isCaughtUp = false
     }
+    for ((_,poller) in v2Pollers) {
+      poller.isCaughtUp = false
+    }
   }
 
   public fun startPollersIfNeeded() {

app/src/main/java/org/thoughtcrime/securesms/loki/api/SessionProtocolImpl.kt

@@ -23,7 +23,6 @@ class SessionProtocolImpl(private val context: Context) : SessionProtocol {
     override fun decrypt(ciphertext: ByteArray, x25519KeyPair: ECKeyPair): Pair<ByteArray, String> {
         val recipientX25519PrivateKey = x25519KeyPair.privateKey.serialize()
         val recipientX25519PublicKey = Hex.fromStringCondensed(x25519KeyPair.hexEncodedPublicKey.removing05PrefixIfNeeded())
-        Log.d("Test", "recipientX25519PublicKey: $recipientX25519PublicKey")
         val signatureSize = Sign.BYTES
         val ed25519PublicKeySize = Sign.PUBLICKEYBYTES
 

app/src/main/java/org/thoughtcrime/securesms/loki/database/LokiThreadDatabase.kt

@@ -68,7 +68,7 @@ class LokiThreadDatabase(context: Context, helper: SQLCipherOpenHelper) : Databa
             while (cursor != null && cursor.moveToNext()) {
                 val threadID = cursor.getLong(threadID)
                 val string = cursor.getString(publicChat)
-                val openGroup = OpenGroupV2.fromJson(string)
+                val openGroup = OpenGroupV2.fromJSON(string)
                 if (openGroup != null) result[threadID] = openGroup
             }
         } catch (e: Exception) {
@@ -100,7 +100,7 @@ class LokiThreadDatabase(context: Context, helper: SQLCipherOpenHelper) : Databa
         val database = databaseHelper.readableDatabase
         return database.get(publicChatTable, "${Companion.threadID} = ?", arrayOf(threadID.toString())) { cursor ->
             val json = cursor.getString(publicChat)
-            OpenGroupV2.fromJson(json)
+            OpenGroupV2.fromJSON(json)
         }
     }
 

app/src/main/java/org/thoughtcrime/securesms/loki/database/SessionJobDatabase.kt

@@ -4,6 +4,7 @@ import android.content.ContentValues
 import android.content.Context
 import net.sqlcipher.Cursor
 import org.session.libsession.messaging.jobs.*
+import org.session.libsession.messaging.utilities.Data
 import org.session.libsignal.utilities.logging.Log
 import org.thoughtcrime.securesms.database.Database
 import org.thoughtcrime.securesms.database.helpers.SQLCipherOpenHelper
@@ -18,53 +19,55 @@ class SessionJobDatabase(context: Context, helper: SQLCipherOpenHelper) : Databa
         const val jobType = "job_type"
         const val failureCount = "failure_count"
         const val serializedData = "serialized_data"
-        @JvmStatic val createSessionJobTableCommand = "CREATE TABLE $sessionJobTable ($jobID INTEGER PRIMARY KEY, $jobType STRING, $failureCount INTEGER DEFAULT 0, $serializedData TEXT);"
+        @JvmStatic val createSessionJobTableCommand
+            = "CREATE TABLE $sessionJobTable ($jobID INTEGER PRIMARY KEY, $jobType STRING, $failureCount INTEGER DEFAULT 0, $serializedData TEXT);"
     }
 
     fun persistJob(job: Job) {
         val database = databaseHelper.writableDatabase
         val contentValues = ContentValues(4)
-        contentValues.put(jobID, job.id)
+        contentValues.put(jobID, job.id!!)
         contentValues.put(jobType, job.getFactoryKey())
         contentValues.put(failureCount, job.failureCount)
         contentValues.put(serializedData, SessionJobHelper.dataSerializer.serialize(job.serialize()))
-        database.insertOrUpdate(sessionJobTable, contentValues, "$jobID = ?", arrayOf(jobID))
+        database.insertOrUpdate(sessionJobTable, contentValues, "$jobID = ?", arrayOf( job.id!! ))
     }
 
-    fun markJobAsSucceeded(jobId: String) {
+    fun markJobAsSucceeded(jobID: String) {
-        databaseHelper.writableDatabase.delete(sessionJobTable, "$jobID = ?", arrayOf(jobId))
+        databaseHelper.writableDatabase.delete(sessionJobTable, "${Companion.jobID} = ?", arrayOf( jobID ))
     }
 
-    fun markJobAsFailed(jobId: String) {
+    fun markJobAsFailedPermanently(jobID: String) {
-        databaseHelper.writableDatabase.delete(sessionJobTable, "$jobID = ?", arrayOf(jobId))
+        databaseHelper.writableDatabase.delete(sessionJobTable, "${Companion.jobID} = ?", arrayOf( jobID ))
     }
 
     fun getAllPendingJobs(type: String): Map<String, Job?> {
         val database = databaseHelper.readableDatabase
-        return database.getAll(sessionJobTable, "$jobType = ?", arrayOf(type)) { cursor ->
+        return database.getAll(sessionJobTable, "$jobType = ?", arrayOf( type )) { cursor ->
-            val jobId = cursor.getString(jobID)
+            val jobID = cursor.getString(jobID)
             try {
-                jobId to jobFromCursor(cursor)
+                jobID to jobFromCursor(cursor)
             } catch (e: Exception) {
-                Log.e("Loki", "Error serializing Job of type $type",e)
+                Log.e("Loki", "Error deserializing job of type: $type.", e)
-                jobId to null
+                jobID to null
             }
         }.toMap()
     }
 
     fun getAttachmentUploadJob(attachmentID: Long): AttachmentUploadJob? {
         val database = databaseHelper.readableDatabase
-        var result = mutableListOf<AttachmentUploadJob>()
+        val result = mutableListOf<AttachmentUploadJob>()
-        database.getAll(sessionJobTable, "$jobType = ?", arrayOf(AttachmentUploadJob.KEY)) { cursor ->
+        database.getAll(sessionJobTable, "$jobType = ?", arrayOf( AttachmentUploadJob.KEY )) { cursor ->
-            result.add(jobFromCursor(cursor) as AttachmentUploadJob)
+            val job = jobFromCursor(cursor) as AttachmentUploadJob?
+            if (job != null) { result.add(job) }
         }
         return result.firstOrNull { job -> job.attachmentID == attachmentID }
     }
 
     fun getMessageSendJob(messageSendJobID: String): MessageSendJob? {
         val database = databaseHelper.readableDatabase
-        return database.get(sessionJobTable, "$jobID = ? AND $jobType = ?", arrayOf(messageSendJobID, MessageSendJob.KEY)) { cursor ->
+        return database.get(sessionJobTable, "$jobID = ? AND $jobType = ?", arrayOf( messageSendJobID, MessageSendJob.KEY )) { cursor ->
-            jobFromCursor(cursor) as MessageSendJob
+            jobFromCursor(cursor) as MessageSendJob?
         }
     }
 
@@ -72,8 +75,8 @@ class SessionJobDatabase(context: Context, helper: SQLCipherOpenHelper) : Databa
         val database = databaseHelper.readableDatabase
         var cursor: android.database.Cursor? = null
         try {
-            cursor = database.rawQuery("SELECT * FROM $sessionJobTable WHERE $jobID = ?", arrayOf(job.id))
+            cursor = database.rawQuery("SELECT * FROM $sessionJobTable WHERE $jobID = ?", arrayOf( job.id!! ))
-            return cursor != null && cursor.moveToFirst()
+            return cursor == null || !cursor.moveToFirst()
         } catch (e: Exception) {
             // Do nothing
         }  finally {
@@ -82,10 +85,10 @@ class SessionJobDatabase(context: Context, helper: SQLCipherOpenHelper) : Databa
         return false
     }
 
-    private fun jobFromCursor(cursor: Cursor): Job {
+    private fun jobFromCursor(cursor: Cursor): Job? {
         val type = cursor.getString(jobType)
         val data = SessionJobHelper.dataSerializer.deserialize(cursor.getString(serializedData))
-        val job = SessionJobHelper.sessionJobInstantiator.instantiate(type, data)
+        val job = SessionJobHelper.sessionJobInstantiator.instantiate(type, data) ?: return null
         job.id = cursor.getString(jobID)
         job.failureCount = cursor.getInt(failureCount)
         return job

app/src/test/java/org/thoughtcrime/securesms/jobmanager/impl/JsonDataSerializerTest.java

@@ -1,7 +1,7 @@
 package org.thoughtcrime.securesms.jobmanager.impl;
 
 import org.junit.Test;
-import org.session.libsession.messaging.jobs.Data;
+import org.session.libsession.messaging.utilities.Data;
 import org.session.libsession.utilities.Util;
 
 import java.io.IOException;

build.gradle

@@ -6,7 +6,7 @@ buildscript {
         jcenter()
     }
     dependencies {
-        classpath 'com.android.tools.build:gradle:4.1.2'
+        classpath 'com.android.tools.build:gradle:4.1.3'
         classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlinVersion"
         classpath "com.google.gms:google-services:4.3.4"
         classpath files('libs/gradle-witness.jar')

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

@@ -8,8 +8,8 @@ class MessagingModuleConfiguration(
     val context: Context,
     val storage: StorageProtocol,
     val messageDataProvider: MessageDataProvider,
-    val sessionProtocol: SessionProtocol)
+    val sessionProtocol: SessionProtocol
-{
+) {
 
     companion object {
         lateinit var shared: MessagingModuleConfiguration

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

@@ -45,7 +45,7 @@ interface StorageProtocol {
     // Jobs
     fun persistJob(job: Job)
     fun markJobAsSucceeded(jobId: String)
-    fun markJobAsFailed(jobId: String)
+    fun markJobAsFailedPermanently(jobId: String)
     fun getAllPendingJobs(type: String): Map<String,Job?>
     fun getAttachmentUploadJob(attachmentID: Long): AttachmentUploadJob?
     fun getMessageSendJob(messageSendJobID: String): MessageSendJob?

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

@@ -1,106 +1,91 @@
 package org.session.libsession.messaging.file_server
 
 import nl.komponents.kovenant.Promise
-import nl.komponents.kovenant.functional.bind
 import nl.komponents.kovenant.functional.map
 import okhttp3.Headers
 import okhttp3.HttpUrl
 import okhttp3.MediaType
 import okhttp3.RequestBody
-import org.session.libsession.messaging.MessagingModuleConfiguration
 import org.session.libsession.messaging.open_groups.OpenGroupAPIV2
 import org.session.libsession.snode.OnionRequestAPI
 import org.session.libsignal.service.loki.HTTP
-import org.session.libsignal.service.loki.utilities.retryIfNeeded
 import org.session.libsignal.utilities.Base64
 import org.session.libsignal.utilities.JsonUtil
 import org.session.libsignal.utilities.logging.Log
 
 object FileServerAPIV2 {
 
-    const val DEFAULT_SERVER = "http://88.99.175.227"
     private const val DEFAULT_SERVER_PUBLIC_KEY = "7cb31905b55cd5580c686911debf672577b3fb0bff81df4ce2d5c4cb3a7aaa69"
+    const val DEFAULT_SERVER = "http://88.99.175.227"
 
-    sealed class Error : Exception() {
+    sealed class Error(message: String) : Exception(message) {
-        object PARSING_FAILED : Error()
+        object ParsingFailed : Error("Invalid response.")
-        object INVALID_URL : Error()
+        object InvalidURL : Error("Invalid URL.")
-
-        fun errorDescription() = when (this) {
-            PARSING_FAILED -> "Invalid response."
-            INVALID_URL -> "Invalid URL."
-        }
-
     }
 
     data class Request(
-            val verb: HTTP.Verb,
+        val verb: HTTP.Verb,
-            val endpoint: String,
+        val endpoint: String,
-            val queryParameters: Map<String, String> = mapOf(),
+        val queryParameters: Map<String, String> = mapOf(),
-            val parameters: Any? = null,
+        val parameters: Any? = null,
-            val headers: Map<String, String> = mapOf(),
+        val headers: Map<String, String> = mapOf(),
-            // Always `true` under normal circumstances. You might want to disable
+        /**
-            // this when running over Lokinet.
+         * Always `true` under normal circumstances. You might want to disable
-            val useOnionRouting: Boolean = true
+         * this when running over Lokinet.
+         */
+        val useOnionRouting: Boolean = true
     )
 
     private fun createBody(parameters: Any?): RequestBody? {
         if (parameters == null) return null
-
         val parametersAsJSON = JsonUtil.toJson(parameters)
         return RequestBody.create(MediaType.get("application/json"), parametersAsJSON)
     }
 
     private fun send(request: Request): Promise<Map<*, *>, Exception> {
-        val parsed = HttpUrl.parse(DEFAULT_SERVER) ?: return Promise.ofFail(OpenGroupAPIV2.Error.INVALID_URL)
+        val url = HttpUrl.parse(DEFAULT_SERVER) ?: return Promise.ofFail(OpenGroupAPIV2.Error.InvalidURL)
         val urlBuilder = HttpUrl.Builder()
-                .scheme(parsed.scheme())
+            .scheme(url.scheme())
-                .host(parsed.host())
+            .host(url.host())
-                .port(parsed.port())
+            .port(url.port())
-                .addPathSegments(request.endpoint)
+            .addPathSegments(request.endpoint)
-
         if (request.verb == HTTP.Verb.GET) {
             for ((key, value) in request.queryParameters) {
                 urlBuilder.addQueryParameter(key, value)
             }
         }
-
         val requestBuilder = okhttp3.Request.Builder()
-                .url(urlBuilder.build())
+            .url(urlBuilder.build())
-                .headers(Headers.of(request.headers))
+            .headers(Headers.of(request.headers))
         when (request.verb) {
             HTTP.Verb.GET -> requestBuilder.get()
             HTTP.Verb.PUT -> requestBuilder.put(createBody(request.parameters)!!)
             HTTP.Verb.POST -> requestBuilder.post(createBody(request.parameters)!!)
             HTTP.Verb.DELETE -> requestBuilder.delete(createBody(request.parameters))
         }
-
         if (request.useOnionRouting) {
-            return OnionRequestAPI.sendOnionRequest(requestBuilder.build(), DEFAULT_SERVER, DEFAULT_SERVER_PUBLIC_KEY)
+            return OnionRequestAPI.sendOnionRequest(requestBuilder.build(), DEFAULT_SERVER, DEFAULT_SERVER_PUBLIC_KEY).fail { e ->
-                    .fail { e ->
+                Log.e("Loki", "File server request failed.", e)
-                        Log.e("Loki", "FileServerV2 failed with error",e)
+            }
-                    }
         } else {
             return Promise.ofFail(IllegalStateException("It's currently not allowed to send non onion routed requests."))
         }
-
     }
 
-    // region Sending
     fun upload(file: ByteArray): Promise<Long, Exception> {
         val base64EncodedFile = Base64.encodeBytes(file)
-        val parameters = mapOf("file" to base64EncodedFile)
+        val parameters = mapOf( "file" to base64EncodedFile )
         val request = Request(verb = HTTP.Verb.POST, endpoint = "files", parameters = parameters)
         return send(request).map { json ->
-            json["result"] as? Long ?: throw OpenGroupAPIV2.Error.PARSING_FAILED
+            json["result"] as? Long ?: throw OpenGroupAPIV2.Error.ParsingFailed
         }
     }
 
     fun download(file: Long): Promise<ByteArray, Exception> {
         val request = Request(verb = HTTP.Verb.GET, endpoint = "files/$file")
         return send(request).map { json ->
-            val base64EncodedFile = json["result"] as? String ?: throw Error.PARSING_FAILED
+            val base64EncodedFile = json["result"] as? String ?: throw Error.ParsingFailed
-            Base64.decode(base64EncodedFile) ?: throw Error.PARSING_FAILED
+            Base64.decode(base64EncodedFile) ?: throw Error.ParsingFailed
         }
     }
-
 }

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

@@ -3,9 +3,9 @@ package org.session.libsession.messaging.jobs
 import okhttp3.HttpUrl
 import org.session.libsession.messaging.MessagingModuleConfiguration
 import org.session.libsession.messaging.file_server.FileServerAPI
-import org.session.libsession.messaging.file_server.FileServerAPIV2
 import org.session.libsession.messaging.open_groups.OpenGroupAPIV2
 import org.session.libsession.messaging.sending_receiving.attachments.AttachmentState
+import org.session.libsession.messaging.utilities.Data
 import org.session.libsession.messaging.utilities.DotNetAPI
 import org.session.libsession.utilities.DownloadUtilities
 import org.session.libsignal.service.api.crypto.AttachmentCipherInputStream
@@ -31,8 +31,8 @@ class AttachmentDownloadJob(val attachmentID: Long, val databaseMessageID: Long)
         val KEY: String = "AttachmentDownloadJob"
 
         // Keys used for database storage
-        private val KEY_ATTACHMENT_ID = "attachment_id"
+        private val ATTACHMENT_ID_KEY = "attachment_id"
-        private val KEY_TS_INCOMING_MESSAGE_ID = "tsIncoming_message_id"
+        private val TS_INCOMING_MESSAGE_ID_KEY = "tsIncoming_message_id"
     }
 
     override fun execute() {
@@ -52,18 +52,19 @@ class AttachmentDownloadJob(val attachmentID: Long, val databaseMessageID: Long)
         try {
             val messageDataProvider = MessagingModuleConfiguration.shared.messageDataProvider
             val attachment = messageDataProvider.getDatabaseAttachment(attachmentID)
-                    ?: return handleFailure(Error.NoAttachment)
+                ?: return handleFailure(Error.NoAttachment)
             messageDataProvider.setAttachmentState(AttachmentState.STARTED, attachmentID, this.databaseMessageID)
             val tempFile = createTempFile()
-
             val threadId = MessagingModuleConfiguration.shared.storage.getThreadIdForMms(databaseMessageID)
             val openGroupV2 = MessagingModuleConfiguration.shared.storage.getV2OpenGroup(threadId.toString())
-
             val stream = if (openGroupV2 == null) {
                 DownloadUtilities.downloadFile(tempFile, attachment.url, FileServerAPI.maxFileSize, null)
                 // Assume we're retrieving an attachment for an open group server if the digest is not set
-                if (attachment.digest?.size ?: 0 == 0 || attachment.key.isNullOrEmpty()) FileInputStream(tempFile)
+                if (attachment.digest?.size ?: 0 == 0 || attachment.key.isNullOrEmpty()) {
-                else AttachmentCipherInputStream.createForAttachment(tempFile, attachment.size, Base64.decode(attachment.key), attachment.digest)
+                    FileInputStream(tempFile)
+                } else {
+                    AttachmentCipherInputStream.createForAttachment(tempFile, attachment.size, Base64.decode(attachment.key), attachment.digest)
+                }
             } else {
                 val url = HttpUrl.parse(attachment.url)!!
                 val fileId = url.pathSegments().last()
@@ -100,8 +101,9 @@ class AttachmentDownloadJob(val attachmentID: Long, val databaseMessageID: Long)
     }
 
     override fun serialize(): Data {
-        return Data.Builder().putLong(KEY_ATTACHMENT_ID, attachmentID)
+        return Data.Builder()
-            .putLong(KEY_TS_INCOMING_MESSAGE_ID, databaseMessageID)
+            .putLong(ATTACHMENT_ID_KEY, attachmentID)
+            .putLong(TS_INCOMING_MESSAGE_ID_KEY, databaseMessageID)
             .build();
     }
 
@@ -110,8 +112,9 @@ class AttachmentDownloadJob(val attachmentID: Long, val databaseMessageID: Long)
     }
 
     class Factory : Job.Factory<AttachmentDownloadJob> {
+
         override fun create(data: Data): AttachmentDownloadJob {
-            return AttachmentDownloadJob(data.getLong(KEY_ATTACHMENT_ID), data.getLong(KEY_TS_INCOMING_MESSAGE_ID))
+            return AttachmentDownloadJob(data.getLong(ATTACHMENT_ID_KEY), data.getLong(TS_INCOMING_MESSAGE_ID_KEY))
         }
     }
 }

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

@@ -8,6 +8,7 @@ import org.session.libsession.messaging.file_server.FileServerAPI
 import org.session.libsession.messaging.messages.Message
 import org.session.libsession.messaging.open_groups.OpenGroupAPIV2
 import org.session.libsession.messaging.sending_receiving.MessageSender
+import org.session.libsession.messaging.utilities.Data
 import org.session.libsession.messaging.utilities.DotNetAPI
 import org.session.libsignal.service.api.crypto.AttachmentCipherOutputStream
 import org.session.libsignal.service.api.messages.SignalServiceAttachmentStream
@@ -30,44 +31,39 @@ class AttachmentUploadJob(val attachmentID: Long, val threadID: String, val mess
 
     // Settings
     override val maxFailureCount: Int = 20
+
     companion object {
         val TAG = AttachmentUploadJob::class.simpleName
         val KEY: String = "AttachmentUploadJob"
 
         // Keys used for database storage
-        private val KEY_ATTACHMENT_ID = "attachment_id"
+        private val ATTACHMENT_ID_KEY = "attachment_id"
-        private val KEY_THREAD_ID = "thread_id"
+        private val THREAD_ID_KEY = "thread_id"
-        private val KEY_MESSAGE = "message"
+        private val MESSAGE_KEY = "message"
-        private val KEY_MESSAGE_SEND_JOB_ID = "message_send_job_id"
+        private val MESSAGE_SEND_JOB_ID_KEY = "message_send_job_id"
     }
 
     override fun execute() {
         try {
             val attachment = MessagingModuleConfiguration.shared.messageDataProvider.getScaledSignalAttachmentStream(attachmentID)
                 ?: return handleFailure(Error.NoAttachment)
-
             val usePadding = false
             val openGroupV2 = MessagingModuleConfiguration.shared.storage.getV2OpenGroup(threadID)
             val openGroup = MessagingModuleConfiguration.shared.storage.getOpenGroup(threadID)
-            val server = openGroup?.let {
+            val server = openGroupV2?.server ?: openGroup?.server ?: FileServerAPI.shared.server
-                it.server
-            } ?: openGroupV2?.let {
-                it.server
-            } ?: FileServerAPI.shared.server
             val shouldEncrypt = (openGroup == null && openGroupV2 == null) // Encrypt if this isn't an open group
-
             val attachmentKey = Util.getSecretBytes(64)
             val paddedLength = if (usePadding) PaddingInputStream.getPaddedSize(attachment.length) else attachment.length
             val dataStream = if (usePadding) PaddingInputStream(attachment.inputStream, attachment.length) else attachment.inputStream
             val ciphertextLength = if (shouldEncrypt) AttachmentCipherOutputStream.getCiphertextLength(paddedLength) else attachment.length
-
             val outputStreamFactory = if (shouldEncrypt) AttachmentCipherOutputStreamFactory(attachmentKey) else PlaintextOutputStreamFactory()
             val attachmentData = PushAttachmentData(attachment.contentType, dataStream, ciphertextLength, outputStreamFactory, attachment.listener)
-
+            val uploadResult = if (openGroupV2 != null) {
-            val uploadResult = if (openGroupV2 == null) FileServerAPI.shared.uploadAttachment(server, attachmentData) else {
                 val dataBytes = attachmentData.data.readBytes()
                 val result = OpenGroupAPIV2.upload(dataBytes, openGroupV2.room, openGroupV2.server).get()
                 DotNetAPI.UploadResult(result, "${openGroupV2.server}/files/$result", byteArrayOf())
+            } else {
+                FileServerAPI.shared.uploadAttachment(server, attachmentData)
             }
             handleSuccess(attachment, attachmentKey, uploadResult)
         } catch (e: java.lang.Exception) {
@@ -82,7 +78,7 @@ class AttachmentUploadJob(val attachmentID: Long, val threadID: String, val mess
     }
 
     private fun handleSuccess(attachment: SignalServiceAttachmentStream, attachmentKey: ByteArray, uploadResult: DotNetAPI.UploadResult) {
-        Log.w(TAG, "Attachment uploaded successfully.")
+        Log.d(TAG, "Attachment uploaded successfully.")
         delegate?.handleJobSucceeded(this)
         MessagingModuleConfiguration.shared.messageDataProvider.updateAttachmentAfterUploadSucceeded(attachmentID, attachment, attachmentKey, uploadResult)
         MessagingModuleConfiguration.shared.storage.resumeMessageSendJobIfNeeded(messageSendJobID)
@@ -108,7 +104,7 @@ class AttachmentUploadJob(val attachmentID: Long, val threadID: String, val mess
         val messageSendJob = storage.getMessageSendJob(messageSendJobID)
         MessageSender.handleFailedMessageSend(this.message, e)
         if (messageSendJob != null) {
-            storage.markJobAsFailed(messageSendJobID)
+            storage.markJobAsFailedPermanently(messageSendJobID)
         }
     }
 
@@ -119,10 +115,11 @@ class AttachmentUploadJob(val attachmentID: Long, val threadID: String, val mess
         val output = Output(serializedMessage)
         kryo.writeObject(output, message)
         output.close()
-        return Data.Builder().putLong(KEY_ATTACHMENT_ID, attachmentID)
+        return Data.Builder()
-            .putString(KEY_THREAD_ID, threadID)
+            .putLong(ATTACHMENT_ID_KEY, attachmentID)
-            .putByteArray(KEY_MESSAGE, serializedMessage)
+            .putString(THREAD_ID_KEY, threadID)
-            .putString(KEY_MESSAGE_SEND_JOB_ID, messageSendJobID)
+            .putByteArray(MESSAGE_KEY, serializedMessage)
+            .putString(MESSAGE_SEND_JOB_ID_KEY, messageSendJobID)
             .build();
     }
 
@@ -133,12 +130,18 @@ class AttachmentUploadJob(val attachmentID: Long, val threadID: String, val mess
     class Factory: Job.Factory<AttachmentUploadJob> {
 
         override fun create(data: Data): AttachmentUploadJob {
-            val serializedMessage = data.getByteArray(KEY_MESSAGE)
+            val serializedMessage = data.getByteArray(MESSAGE_KEY)
             val kryo = Kryo()
+            kryo.isRegistrationRequired = false
             val input = Input(serializedMessage)
             val message: Message = kryo.readObject(input, Message::class.java)
             input.close()
-            return AttachmentUploadJob(data.getLong(KEY_ATTACHMENT_ID), data.getString(KEY_THREAD_ID)!!, message, data.getString(KEY_MESSAGE_SEND_JOB_ID)!!)
+            return AttachmentUploadJob(
+                data.getLong(ATTACHMENT_ID_KEY),
+                data.getString(THREAD_ID_KEY)!!,
+                message,
+                data.getString(MESSAGE_SEND_JOB_ID_KEY)!!
+            )
         }
     }
 }

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

@@ -1,5 +1,7 @@
 package org.session.libsession.messaging.jobs
 
+import org.session.libsession.messaging.utilities.Data
+
 interface Job {
     var delegate: JobDelegate?
     var id: String?
@@ -8,21 +10,21 @@ interface Job {
     val maxFailureCount: Int
 
     companion object {
+
         // Keys used for database storage
-        private val KEY_ID = "id"
+        private val ID_KEY = "id"
-        private val KEY_FAILURE_COUNT = "failure_count"
+        private val FAILURE_COUNT_KEY = "failure_count"
+        internal const val MAX_BUFFER_SIZE = 1_000_000 // bytes
     }
 
     fun execute()
 
     fun serialize(): Data
 
-    /**
-     * Returns the key that can be used to find the relevant factory needed to create your job.
-     */
     fun getFactoryKey(): String
 
     interface Factory<T : Job> {
-        fun create(data: Data): T
+
+        fun create(data: Data): T?
     }
 }

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

@@ -5,6 +5,7 @@ import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
 import org.session.libsession.messaging.MessagingModuleConfiguration
 import org.session.libsignal.utilities.logging.Log
+import java.lang.IllegalStateException
 import java.util.*
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.Executors
@@ -17,44 +18,58 @@ import kotlin.math.roundToLong
 class JobQueue : JobDelegate {
     private var hasResumedPendingJobs = false // Just for debugging
     private val jobTimestampMap = ConcurrentHashMap<Long, AtomicInteger>()
-    private val dispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
+    private val rxDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
-    private val multiDispatcher = Executors.newFixedThreadPool(2).asCoroutineDispatcher()
+    private val txDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
+    private val attachmentDispatcher = Executors.newFixedThreadPool(2).asCoroutineDispatcher()
     private val scope = GlobalScope + SupervisorJob()
     private val queue = Channel<Job>(UNLIMITED)
 
     val timer = Timer()
 
+    private fun CoroutineScope.processWithDispatcher(channel: Channel<Job>, dispatcher: CoroutineDispatcher) = launch(dispatcher) {
+        for (job in channel) {
+            if (!isActive) break
+            job.delegate = this@JobQueue
+            job.execute()
+        }
+    }
+
     init {
         // Process jobs
-        scope.launch(dispatcher) {
+        scope.launch {
+            val rxQueue = Channel<Job>(capacity = 1024)
+            val txQueue = Channel<Job>(capacity = 1024)
+            val attachmentQueue = Channel<Job>(capacity = 1024)
+
+            val receiveJob = processWithDispatcher(rxQueue, rxDispatcher)
+            val txJob = processWithDispatcher(txQueue, txDispatcher)
+            val attachmentJob = processWithDispatcher(attachmentQueue, attachmentDispatcher)
+
             while (isActive) {
-                queue.receive().let { job ->
+                for (job in queue) {
-                    if (job.canExecuteParallel()) {
+                    when (job) {
-                        launch(multiDispatcher) {
+                        is NotifyPNServerJob, is AttachmentUploadJob, is MessageSendJob -> txQueue.send(job)
-                            job.delegate = this@JobQueue
+                        is AttachmentDownloadJob -> attachmentQueue.send(job)
-                            job.execute()
+                        is MessageReceiveJob -> rxQueue.send(job)
-                        }
+                        else -> throw IllegalStateException("Unexpected job type.")
-                    } else {
-                        job.delegate = this@JobQueue
-                        job.execute()
                     }
                 }
             }
+
+            // The job has been cancelled
+            receiveJob.cancel()
+            txJob.cancel()
+            attachmentJob.cancel()
+
         }
     }
 
     companion object {
+
         @JvmStatic
         val shared: JobQueue by lazy { JobQueue() }
     }
 
-    private fun Job.canExecuteParallel(): Boolean {
-        return this.javaClass in arrayOf(
-                AttachmentUploadJob::class.java,
-                AttachmentDownloadJob::class.java
-        )
-    }
-
     fun add(job: Job) {
         addWithoutExecuting(job)
         queue.offer(job) // offer always called on unlimited capacity
@@ -68,7 +83,6 @@ class JobQueue : JobDelegate {
         val currentTime = System.currentTimeMillis()
         jobTimestampMap.putIfAbsent(currentTime, AtomicInteger())
         job.id = currentTime.toString() + jobTimestampMap[currentTime]!!.getAndIncrement().toString()
-
         MessagingModuleConfiguration.shared.storage.persistJob(job)
     }
 
@@ -78,25 +92,26 @@ class JobQueue : JobDelegate {
             return
         }
         hasResumedPendingJobs = true
-        val allJobTypes = listOf(AttachmentUploadJob.KEY,
+        val allJobTypes = listOf(
-                AttachmentDownloadJob.KEY,
+            AttachmentUploadJob.KEY,
-                MessageReceiveJob.KEY,
+            AttachmentDownloadJob.KEY,
-                MessageSendJob.KEY,
+            MessageReceiveJob.KEY,
-                NotifyPNServerJob.KEY
+            MessageSendJob.KEY,
+            NotifyPNServerJob.KEY
         )
         allJobTypes.forEach { type ->
             val allPendingJobs = MessagingModuleConfiguration.shared.storage.getAllPendingJobs(type)
             val pendingJobs = mutableListOf<Job>()
             for ((id, job) in allPendingJobs) {
                 if (job == null) {
-                    // job failed to serialize, remove it from the DB
+                    // Job failed to deserialize, remove it from the DB
                     handleJobFailedPermanently(id)
                 } else {
                     pendingJobs.add(job)
                 }
             }
             pendingJobs.sortedBy { it.id }.forEach { job ->
-                Log.i("Jobs", "Resuming pending job of type: ${job::class.simpleName}.")
+                Log.i("Loki", "Resuming pending job of type: ${job::class.simpleName}.")
                 queue.offer(job) // Offer always called on unlimited capacity
             }
         }
@@ -108,17 +123,31 @@ class JobQueue : JobDelegate {
     }
 
     override fun handleJobFailed(job: Job, error: Exception) {
-        job.failureCount += 1
+        // Canceled
         val storage = MessagingModuleConfiguration.shared.storage
-        if (storage.isJobCanceled(job)) { return Log.i("Jobs", "${job::class.simpleName} canceled.")}
+        if (storage.isJobCanceled(job)) {
-        if (job.failureCount == job.maxFailureCount) {
+            return Log.i("Loki", "${job::class.simpleName} canceled.")
+        }
+        // Message send jobs waiting for the attachment to upload
+        if (job is MessageSendJob && error is MessageSendJob.AwaitingAttachmentUploadException) {
+            val retryInterval: Long = 1000 * 4
+            Log.i("Loki", "Message send job waiting for attachment upload to finish.")
+            timer.schedule(delay = retryInterval) {
+                Log.i("Loki", "Retrying ${job::class.simpleName}.")
+                queue.offer(job)
+            }
+            return
+        }
+        // Regular job failure
+        job.failureCount += 1
+        if (job.failureCount >= job.maxFailureCount) {
             handleJobFailedPermanently(job, error)
         } else {
             storage.persistJob(job)
             val retryInterval = getRetryInterval(job)
-            Log.i("Jobs", "${job::class.simpleName} failed; scheduling retry (failure count is ${job.failureCount}).")
+            Log.i("Loki", "${job::class.simpleName} failed; scheduling retry (failure count is ${job.failureCount}).")
             timer.schedule(delay = retryInterval) {
-                Log.i("Jobs", "Retrying ${job::class.simpleName}.")
+                Log.i("Loki", "Retrying ${job::class.simpleName}.")
                 queue.offer(job)
             }
         }
@@ -131,7 +160,7 @@ class JobQueue : JobDelegate {
 
     private fun handleJobFailedPermanently(jobId: String) {
         val storage = MessagingModuleConfiguration.shared.storage
-        storage.markJobAsFailed(jobId)
+        storage.markJobAsFailedPermanently(jobId)
     }
 
     private fun getRetryInterval(job: Job): Long {

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

@@ -4,6 +4,7 @@ import nl.komponents.kovenant.Promise
 import nl.komponents.kovenant.deferred
 import org.session.libsession.messaging.sending_receiving.MessageReceiver
 import org.session.libsession.messaging.sending_receiving.handle
+import org.session.libsession.messaging.utilities.Data
 import org.session.libsignal.utilities.logging.Log
 
 class MessageReceiveJob(val data: ByteArray, val isBackgroundPoll: Boolean, val openGroupMessageServerID: Long? = null, val openGroupID: String? = null) : Job {
@@ -11,7 +12,6 @@ class MessageReceiveJob(val data: ByteArray, val isBackgroundPoll: Boolean, val
     override var id: String? = null
     override var failureCount: Int = 0
 
-    // Settings
     override val maxFailureCount: Int = 10
     companion object {
         val TAG = MessageReceiveJob::class.simpleName
@@ -20,10 +20,11 @@ class MessageReceiveJob(val data: ByteArray, val isBackgroundPoll: Boolean, val
         private val RECEIVE_LOCK = Object()
 
         // Keys used for database storage
-        private val KEY_DATA = "data"
+        private val DATA_KEY = "data"
-        private val KEY_IS_BACKGROUND_POLL = "is_background_poll"
+        // FIXME: We probably shouldn't be using this job when background polling
-        private val KEY_OPEN_GROUP_MESSAGE_SERVER_ID = "openGroupMessageServerID"
+        private val IS_BACKGROUND_POLL_KEY = "is_background_poll"
-        private val KEY_OPEN_GROUP_ID = "open_group_id"
+        private val OPEN_GROUP_MESSAGE_SERVER_ID_KEY = "openGroupMessageServerID"
+        private val OPEN_GROUP_ID_KEY = "open_group_id"
     }
 
     override fun execute() {
@@ -35,19 +36,18 @@ class MessageReceiveJob(val data: ByteArray, val isBackgroundPoll: Boolean, val
         try {
             val isRetry: Boolean = failureCount != 0
             val (message, proto) = MessageReceiver.parse(this.data, this.openGroupMessageServerID, isRetry)
-            synchronized(RECEIVE_LOCK) {
+            synchronized(RECEIVE_LOCK) { // FIXME: Do we need this?
                 MessageReceiver.handle(message, proto, this.openGroupID)
             }
             this.handleSuccess()
             deferred.resolve(Unit)
         } catch (e: Exception) {
-            Log.e(TAG, "Couldn't receive message due to error", e)
+            Log.e(TAG, "Couldn't receive message.", e)
-            val error = e as? MessageReceiver.Error
+            if (e is MessageReceiver.Error && !e.isRetryable) {
-            if (error != null && !error.isRetryable) {
+                Log.e("Loki", "Message receive job permanently failed.", e)
-                Log.e("Loki", "Message receive job permanently failed due to error", e)
+                this.handlePermanentFailure(e)
-                this.handlePermanentFailure(error)
             } else {
-                Log.e("Loki", "Couldn't receive message due to error", e)
+                Log.e("Loki", "Couldn't receive message.", e)
                 this.handleFailure(e)
             }
             deferred.resolve(Unit) // The promise is just used to keep track of when we're done
@@ -68,10 +68,10 @@ class MessageReceiveJob(val data: ByteArray, val isBackgroundPoll: Boolean, val
     }
 
     override fun serialize(): Data {
-        val builder = Data.Builder().putByteArray(KEY_DATA, data)
+        val builder = Data.Builder().putByteArray(DATA_KEY, data)
-                .putBoolean(KEY_IS_BACKGROUND_POLL, isBackgroundPoll)
+            .putBoolean(IS_BACKGROUND_POLL_KEY, isBackgroundPoll)
-        openGroupMessageServerID?.let { builder.putLong(KEY_OPEN_GROUP_MESSAGE_SERVER_ID, openGroupMessageServerID) }
+        openGroupMessageServerID?.let { builder.putLong(OPEN_GROUP_MESSAGE_SERVER_ID_KEY, it) }
-        openGroupID?.let { builder.putString(KEY_OPEN_GROUP_ID, openGroupID) }
+        openGroupID?.let { builder.putString(OPEN_GROUP_ID_KEY, it) }
         return builder.build();
     }
 
@@ -82,7 +82,12 @@ class MessageReceiveJob(val data: ByteArray, val isBackgroundPoll: Boolean, val
     class Factory: Job.Factory<MessageReceiveJob> {
 
         override fun create(data: Data): MessageReceiveJob {
-            return MessageReceiveJob(data.getByteArray(KEY_DATA), data.getBoolean(KEY_IS_BACKGROUND_POLL), data.getLong(KEY_OPEN_GROUP_MESSAGE_SERVER_ID), data.getString(KEY_OPEN_GROUP_ID))
+            return MessageReceiveJob(
+                data.getByteArray(DATA_KEY),
+                data.getBoolean(IS_BACKGROUND_POLL_KEY),
+                data.getLong(OPEN_GROUP_MESSAGE_SERVER_ID_KEY),
+                data.getString(OPEN_GROUP_ID_KEY)
+            )
         }
     }
 }

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

@@ -4,33 +4,38 @@ import com.esotericsoftware.kryo.Kryo
 import com.esotericsoftware.kryo.io.Input
 import com.esotericsoftware.kryo.io.Output
 import org.session.libsession.messaging.MessagingModuleConfiguration
+import org.session.libsession.messaging.jobs.Job.Companion.MAX_BUFFER_SIZE
 import org.session.libsession.messaging.messages.Destination
 import org.session.libsession.messaging.messages.Message
 import org.session.libsession.messaging.messages.visible.VisibleMessage
 import org.session.libsession.messaging.sending_receiving.MessageSender
+import org.session.libsession.messaging.utilities.Data
 import org.session.libsignal.utilities.logging.Log
 
 class MessageSendJob(val message: Message, val destination: Destination) : Job {
+
+    object AwaitingAttachmentUploadException : Exception("Awaiting attachment upload.")
+
     override var delegate: JobDelegate? = null
     override var id: String? = null
     override var failureCount: Int = 0
 
-    // Settings
     override val maxFailureCount: Int = 10
+
     companion object {
         val TAG = MessageSendJob::class.simpleName
         val KEY: String = "MessageSendJob"
 
         // Keys used for database storage
-        private val KEY_MESSAGE = "message"
+        private val MESSAGE_KEY = "message"
-        private val KEY_DESTINATION = "destination"
+        private val DESTINATION_KEY = "destination"
     }
 
     override fun execute() {
         val messageDataProvider = MessagingModuleConfiguration.shared.messageDataProvider
         val message = message as? VisibleMessage
-        message?.let {
+        if (message != null) {
-            if(!messageDataProvider.isOutgoingMessage(message.sentTimestamp!!)) return // The message has been deleted
+            if (!messageDataProvider.isOutgoingMessage(message.sentTimestamp!!)) return // The message has been deleted
             val attachmentIDs = mutableListOf<Long>()
             attachmentIDs.addAll(message.attachmentIDs)
             message.quote?.let { it.attachmentID?.let { attachmentID -> attachmentIDs.add(attachmentID) } }
@@ -45,15 +50,17 @@ class MessageSendJob(val message: Message, val destination: Destination) : Job {
                     JobQueue.shared.add(job)
                 }
             }
-            if (attachmentsToUpload.isNotEmpty()) return // Wait for all attachments to upload before continuing
+            if (attachmentsToUpload.isNotEmpty()) {
+                this.handleFailure(AwaitingAttachmentUploadException)
+                return
+            } // Wait for all attachments to upload before continuing
         }
         MessageSender.send(this.message, this.destination).success {
             this.handleSuccess()
         }.fail { exception ->
             Log.e(TAG, "Couldn't send message due to error: $exception.")
-            val e = exception as? MessageSender.Error
+            if (exception is MessageSender.Error) {
-            e?.let {
+                if (!exception.isRetryable) { this.handlePermanentFailure(exception) }
-                if (!e.isRetryable) this.handlePermanentFailure(e)
             }
             this.handleFailure(exception)
         }
@@ -70,8 +77,10 @@ class MessageSendJob(val message: Message, val destination: Destination) : Job {
     private fun handleFailure(error: Exception) {
         Log.w(TAG, "Failed to send $message::class.simpleName.")
         val message = message as? VisibleMessage
-        message?.let {
+        if (message != null) {
-            if(!MessagingModuleConfiguration.shared.messageDataProvider.isOutgoingMessage(message.sentTimestamp!!)) return // The message has been deleted
+            if (!MessagingModuleConfiguration.shared.messageDataProvider.isOutgoingMessage(message.sentTimestamp!!)) {
+                return // The message has been deleted
+            }
         }
         delegate?.handleJobFailed(this, error)
     }
@@ -79,35 +88,55 @@ class MessageSendJob(val message: Message, val destination: Destination) : Job {
     override fun serialize(): Data {
         val kryo = Kryo()
         kryo.isRegistrationRequired = false
-        val output = Output(ByteArray(4096), 10_000_000)
+        val output = Output(ByteArray(4096), MAX_BUFFER_SIZE)
+        // Message
         kryo.writeClassAndObject(output, message)
         output.close()
         val serializedMessage = output.toBytes()
         output.clear()
+        // Destination
         kryo.writeClassAndObject(output, destination)
         output.close()
         val serializedDestination = output.toBytes()
-        return Data.Builder().putByteArray(KEY_MESSAGE, serializedMessage)
+        output.clear()
-            .putByteArray(KEY_DESTINATION, serializedDestination)
+        // Serialize
-            .build();
+        return Data.Builder()
+            .putByteArray(MESSAGE_KEY, serializedMessage)
+            .putByteArray(DESTINATION_KEY, serializedDestination)
+            .build()
     }
 
     override fun getFactoryKey(): String {
         return KEY
     }
 
-    class Factory: Job.Factory<MessageSendJob> {
+    class Factory : Job.Factory<MessageSendJob> {
 
-        override fun create(data: Data): MessageSendJob {
+        override fun create(data: Data): MessageSendJob? {
-            val serializedMessage = data.getByteArray(KEY_MESSAGE)
+            val serializedMessage = data.getByteArray(MESSAGE_KEY)
-            val serializedDestination = data.getByteArray(KEY_DESTINATION)
+            val serializedDestination = data.getByteArray(DESTINATION_KEY)
             val kryo = Kryo()
-            var input = Input(serializedMessage)
+            // Message
-            val message = kryo.readClassAndObject(input) as Message
+            val messageInput = Input(serializedMessage)
-            input.close()
+            val message: Message
-            input = Input(serializedDestination)
+            try {
-            val destination = kryo.readClassAndObject(input) as Destination
+                message = kryo.readClassAndObject(messageInput) as Message
-            input.close()
+            } catch (e: Exception) {
+                Log.e("Loki", "Couldn't deserialize message send job.", e)
+                return null
+            }
+            messageInput.close()
+            // Destination
+            val destinationInput = Input(serializedDestination)
+            val destination: Destination
+            try {
+                destination = kryo.readClassAndObject(destinationInput) as Destination
+            } catch (e: Exception) {
+                Log.e("Loki", "Couldn't deserialize message send job.", e)
+                return null
+            }
+            destinationInput.close()
+            // Return
             return MessageSendJob(message, destination)
         }
     }

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

@@ -9,6 +9,7 @@ import okhttp3.Request
 import okhttp3.RequestBody
 
 import org.session.libsession.messaging.sending_receiving.notifications.PushNotificationAPI
+import org.session.libsession.messaging.utilities.Data
 import org.session.libsession.snode.SnodeMessage
 import org.session.libsession.snode.OnionRequestAPI
 
@@ -21,16 +22,14 @@ class NotifyPNServerJob(val message: SnodeMessage) : Job {
     override var id: String? = null
     override var failureCount: Int = 0
 
-    // Settings
     override val maxFailureCount: Int = 20
     companion object {
         val KEY: String = "NotifyPNServerJob"
 
         // Keys used for database storage
-        private val KEY_MESSAGE = "message"
+        private val MESSAGE_KEY = "message"
     }
 
-    // Running
     override fun execute() {
         val server = PushNotificationAPI.server
         val parameters = mapOf( "data" to message.data, "send_to" to message.recipient )
@@ -41,10 +40,10 @@ class NotifyPNServerJob(val message: SnodeMessage) : Job {
             OnionRequestAPI.sendOnionRequest(request.build(), server, PushNotificationAPI.serverPublicKey, "/loki/v2/lsrpc").map { json ->
                 val code = json["code"] as? Int
                 if (code == null || code == 0) {
-                    Log.d("Loki", "[Loki] Couldn't notify PN server due to error: ${json["message"] as? String ?: "null"}.")
+                    Log.d("Loki", "Couldn't notify PN server due to error: ${json["message"] as? String ?: "null"}.")
                 }
             }.fail { exception ->
-                Log.d("Loki", "[Loki] Couldn't notify PN server due to error: $exception.")
+                Log.d("Loki", "Couldn't notify PN server due to error: $exception.")
             }
         }.success {
             handleSuccess()
@@ -68,18 +67,19 @@ class NotifyPNServerJob(val message: SnodeMessage) : Job {
         val output = Output(serializedMessage)
         kryo.writeObject(output, message)
         output.close()
-        return Data.Builder().putByteArray(KEY_MESSAGE, serializedMessage).build();
+        return Data.Builder().putByteArray(MESSAGE_KEY, serializedMessage).build();
     }
 
     override fun getFactoryKey(): String {
         return KEY
     }
 
-    class Factory: Job.Factory<NotifyPNServerJob> {
+    class Factory : Job.Factory<NotifyPNServerJob> {
 
         override fun create(data: Data): NotifyPNServerJob {
-            val serializedMessage = data.getByteArray(KEY_MESSAGE)
+            val serializedMessage = data.getByteArray(MESSAGE_KEY)
             val kryo = Kryo()
+            kryo.isRegistrationRequired = false
             val input = Input(serializedMessage)
             val message: SnodeMessage = kryo.readObject(input, SnodeMessage::class.java)
             input.close()

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

@@ -1,12 +1,14 @@
 package org.session.libsession.messaging.jobs
 
+import org.session.libsession.messaging.utilities.Data
+
 class SessionJobInstantiator(private val jobFactories: Map<String, Job.Factory<out Job>>) {
 
-    fun instantiate(jobFactoryKey: String, data: Data): Job {
+    fun instantiate(jobFactoryKey: String, data: Data): Job? {
         if (jobFactories.containsKey(jobFactoryKey)) {
-            return jobFactories[jobFactoryKey]?.create(data) ?: throw IllegalStateException("Tried to instantiate a job with key '$jobFactoryKey', but no matching factory was found.")
+            return jobFactories[jobFactoryKey]?.create(data)
         } else {
-            throw IllegalStateException("Tried to instantiate a job with key '$jobFactoryKey', but no matching factory was found.")
+            return null
         }
     }
 }

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

@@ -3,6 +3,7 @@ package org.session.libsession.messaging.jobs
 class SessionJobManagerFactories {
 
     companion object {
+
         fun getSessionJobFactories(): Map<String, Job.Factory<out Job>> {
             return mapOf(
                 AttachmentDownloadJob.KEY to AttachmentDownloadJob.Factory(),

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

@@ -7,9 +7,6 @@ import org.session.libsession.messaging.threads.Address
 import org.session.libsession.utilities.GroupUtil
 import org.session.libsignal.service.loki.utilities.toHexString
 
-typealias OpenGroupModel = OpenGroup
-typealias OpenGroupV2Model = OpenGroupV2
-
 sealed class Destination {
 
     class Contact(var publicKey: String) : Destination() {
@@ -21,11 +18,12 @@ sealed class Destination {
     class OpenGroup(var channel: Long, var server: String) : Destination() {
         internal constructor(): this(0, "")
     }
-    class OpenGroupV2(var room: String, var server: String): Destination() {
+    class OpenGroupV2(var room: String, var server: String) : Destination() {
         internal constructor(): this("", "")
     }
 
     companion object {
+
         fun from(address: Address): Destination {
             return when {
                 address.isContact -> {
@@ -39,10 +37,12 @@ sealed class Destination {
                 address.isOpenGroup -> {
                     val storage = MessagingModuleConfiguration.shared.storage
                     val threadID = storage.getThreadID(address.contactIdentifier())!!
-                    when (val openGroup = storage.getOpenGroup(threadID) ?: storage.getV2OpenGroup(threadID)) {
+                    when (val openGroup = storage.getV2OpenGroup(threadID) ?: storage.getOpenGroup(threadID)) {
-                        is OpenGroupModel -> OpenGroup(openGroup.channel, openGroup.server)
+                        is org.session.libsession.messaging.open_groups.OpenGroup
-                        is OpenGroupV2Model -> OpenGroupV2(openGroup.room, openGroup.server)
+                            -> Destination.OpenGroup(openGroup.channel, openGroup.server)
-                        else -> throw Exception("Invalid OpenGroup $openGroup")
+                        is org.session.libsession.messaging.open_groups.OpenGroupV2
+                            -> Destination.OpenGroupV2(openGroup.room, openGroup.server)
+                        else -> throw Exception("Missing open group for thread with ID: $threadID.")
                     }
                 }
                 else -> {

libsession/src/main/java/org/session/libsession/messaging/messages/Message.kt

@@ -18,12 +18,10 @@ abstract class Message {
     open val isSelfSendValid: Boolean = false
 
     open fun isValid(): Boolean {
-        sentTimestamp?.let {
+        val sentTimestamp = sentTimestamp
-            if (it <= 0) return false
+        if (sentTimestamp != null && sentTimestamp <= 0) { return false }
-        }
+        val receivedTimestamp = receivedTimestamp
-        receivedTimestamp?.let {
+        if (receivedTimestamp != null && receivedTimestamp <= 0) { return false }
-            if (it <= 0) return false
-        }
         return sender != null && recipient != null
     }
 

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

@@ -16,9 +16,10 @@ import org.session.libsignal.utilities.Hex
 import org.session.libsignal.utilities.logging.Log
 
 class ClosedGroupControlMessage() : ControlMessage() {
+    var kind: Kind? = null
 
-    override val ttl: Long = run {
+    override val ttl: Long get() {
-        when (kind) {
+        return when (kind) {
             is Kind.EncryptionKeyPair -> 14 * 24 * 60 * 60 * 1000
             else -> 14 * 24 * 60 * 60 * 1000
         }
@@ -26,31 +27,46 @@ class ClosedGroupControlMessage() : ControlMessage() {
 
     override val isSelfSendValid: Boolean = true
 
-    var kind: Kind? = null
+    override fun isValid(): Boolean {
+        val kind = kind
+        if (!super.isValid() || kind == null) return false
+        return when (kind) {
+            is Kind.New -> {
+                !kind.publicKey.isEmpty && kind.name.isNotEmpty() && kind.encryptionKeyPair?.publicKey != null
+                    && kind.encryptionKeyPair?.privateKey != null && kind.members.isNotEmpty() && kind.admins.isNotEmpty()
+            }
+            is Kind.EncryptionKeyPair -> true
+            is Kind.NameChange -> kind.name.isNotEmpty()
+            is Kind.MembersAdded -> kind.members.isNotEmpty()
+            is Kind.MembersRemoved -> kind.members.isNotEmpty()
+            is Kind.MemberLeft -> true
+        }
+    }
 
     sealed class Kind {
         class New(var publicKey: ByteString, var name: String, var encryptionKeyPair: ECKeyPair?, var members: List<ByteString>, var admins: List<ByteString>) : Kind() {
-            internal constructor(): this(ByteString.EMPTY, "", null, listOf(), listOf())
+            internal constructor() : this(ByteString.EMPTY, "", null, listOf(), listOf())
         }
-        /// An encryption key pair encrypted for each member individually.
+        /** 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).
+         * **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(var publicKey: ByteString?, var wrappers: Collection<KeyPairWrapper>) : Kind() {
-            internal constructor(): this(null, listOf())
+            internal constructor() : this(null, listOf())
         }
         class NameChange(var name: String) : Kind() {
-            internal constructor(): this("")
+            internal constructor() : this("")
         }
         class MembersAdded(var members: List<ByteString>) : Kind() {
-            internal constructor(): this(listOf())
+            internal constructor() : this(listOf())
         }
         class MembersRemoved(var members: List<ByteString>) : Kind() {
-            internal constructor(): this(listOf())
+            internal constructor() : this(listOf())
         }
         class MemberLeft() : Kind()
 
         val description: String =
-            when(this) {
+            when (this) {
                 is New -> "new"
                 is EncryptionKeyPair -> "encryptionKeyPair"
                 is NameChange -> "nameChange"
@@ -65,18 +81,19 @@ class ClosedGroupControlMessage() : ControlMessage() {
 
         fun fromProto(proto: SignalServiceProtos.Content): ClosedGroupControlMessage? {
             if (!proto.hasDataMessage() || !proto.dataMessage.hasClosedGroupControlMessage()) return null
-            val closedGroupControlMessageProto = proto.dataMessage?.closedGroupControlMessage!!
+            val closedGroupControlMessageProto = proto.dataMessage!!.closedGroupControlMessage!!
             val kind: Kind
-            when (closedGroupControlMessageProto.type) {
+            when (closedGroupControlMessageProto.type!!) {
                 DataMessage.ClosedGroupControlMessage.Type.NEW -> {
                     val publicKey = closedGroupControlMessageProto.publicKey ?: return null
                     val name = closedGroupControlMessageProto.name ?: return null
                     val encryptionKeyPairAsProto = closedGroupControlMessageProto.encryptionKeyPair ?: return null
                     try {
-                        val encryptionKeyPair = ECKeyPair(DjbECPublicKey(encryptionKeyPairAsProto.publicKey.toByteArray()), DjbECPrivateKey(encryptionKeyPairAsProto.privateKey.toByteArray()))
+                        val encryptionKeyPair = ECKeyPair(DjbECPublicKey(encryptionKeyPairAsProto.publicKey.toByteArray()),
+                            DjbECPrivateKey(encryptionKeyPairAsProto.privateKey.toByteArray()))
                         kind = Kind.New(publicKey, name, encryptionKeyPair, closedGroupControlMessageProto.membersList, closedGroupControlMessageProto.adminsList)
                     } catch (e: Exception) {
-                        Log.w(TAG, "Couldn't parse key pair")
+                        Log.w(TAG, "Couldn't parse key pair from proto: $encryptionKeyPairAsProto.")
                         return null
                     }
                 }
@@ -107,26 +124,10 @@ class ClosedGroupControlMessage() : ControlMessage() {
         this.kind = kind
     }
 
-    override fun isValid(): Boolean {
-        if (!super.isValid()) return false
-        val kind = kind ?: return false
-        return when(kind) {
-            is Kind.New -> {
-                !kind.publicKey.isEmpty && kind.name.isNotEmpty() && kind.encryptionKeyPair!!.publicKey != null
-                    && kind.encryptionKeyPair!!.privateKey != null && kind.members.isNotEmpty() && kind.admins.isNotEmpty()
-            }
-            is Kind.EncryptionKeyPair -> true
-            is Kind.NameChange -> kind.name.isNotEmpty()
-            is Kind.MembersAdded -> kind.members.isNotEmpty()
-            is Kind.MembersRemoved -> kind.members.isNotEmpty()
-            is Kind.MemberLeft -> true
-        }
-    }
-
     override fun toProto(): SignalServiceProtos.Content? {
         val kind = kind
         if (kind == null) {
-            Log.w(TAG, "Couldn't construct closed group update proto from: $this")
+            Log.w(TAG, "Couldn't construct closed group control message proto from: $this.")
             return null
         }
         try {
@@ -176,7 +177,7 @@ class ClosedGroupControlMessage() : ControlMessage() {
             contentProto.dataMessage = dataMessageProto.build()
             return contentProto.build()
         } catch (e: Exception) {
-            Log.w(TAG, "Couldn't construct closed group update proto from: $this")
+            Log.w(TAG, "Couldn't construct closed group control message proto from: $this.")
             return null
         }
     }
@@ -188,6 +189,7 @@ class ClosedGroupControlMessage() : ControlMessage() {
         }
 
         companion object {
+
             fun fromProto(proto: DataMessage.ClosedGroupControlMessage.KeyPairWrapper): KeyPairWrapper {
                 return KeyPairWrapper(proto.publicKey.toByteArray().toHexString(), proto.encryptedKeyPair)
             }
@@ -199,7 +201,6 @@ class ClosedGroupControlMessage() : ControlMessage() {
             val result = DataMessage.ClosedGroupControlMessage.KeyPairWrapper.newBuilder()
             result.publicKey = ByteString.copyFrom(Hex.fromStringCondensed(publicKey))
             result.encryptedKeyPair = encryptedKeyPair
-
             return try {
                 result.build()
             } catch (e: Exception) {

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

@@ -14,12 +14,15 @@ import org.session.libsignal.service.loki.utilities.removing05PrefixIfNeeded
 import org.session.libsignal.service.loki.utilities.toHexString
 import org.session.libsignal.utilities.Hex
 
-class ConfigurationMessage(var closedGroups: List<ClosedGroup>, var openGroups: List<String>, var contacts: List<Contact>, var displayName: String, var profilePicture: String?, var profileKey: ByteArray): ControlMessage() {
+class ConfigurationMessage(var closedGroups: List<ClosedGroup>, var openGroups: List<String>, var contacts: List<Contact>,
+    var displayName: String, var profilePicture: String?, var profileKey: ByteArray) : ControlMessage() {
+
+    override val isSelfSendValid: Boolean = true
 
     class ClosedGroup(var publicKey: String, var name: String, var encryptionKeyPair: ECKeyPair?, var members: List<String>, var admins: List<String>) {
         val isValid: Boolean get() = members.isNotEmpty() && admins.isNotEmpty()
 
-        internal constructor(): this("", "", null, listOf(), listOf())
+        internal constructor() : this("", "", null, listOf(), listOf())
 
         override fun toString(): String {
             return name
@@ -56,7 +59,7 @@ class ConfigurationMessage(var closedGroups: List<ClosedGroup>, var openGroups:
 
     class Contact(var publicKey: String, var name: String, var profilePicture: String?, var profileKey: ByteArray?) {
 
-        internal constructor(): this("", "", null, null)
+        internal constructor() : this("", "", null, null)
 
         companion object {
 
@@ -66,8 +69,7 @@ class ConfigurationMessage(var closedGroups: List<ClosedGroup>, var openGroups:
                 val name = proto.name
                 val profilePicture = if (proto.hasProfilePicture()) proto.profilePicture else null
                 val profileKey = if (proto.hasProfileKey()) proto.profileKey.toByteArray() else null
-
+                return Contact(publicKey, name, profilePicture, profileKey)
-                return Contact(publicKey,name,profilePicture,profileKey)
             }
         }
 
@@ -79,18 +81,18 @@ class ConfigurationMessage(var closedGroups: List<ClosedGroup>, var openGroups:
             } catch (e: Exception) {
                 return null
             }
-            if (!this.profilePicture.isNullOrEmpty()) {
+            val profilePicture = profilePicture
-                result.profilePicture = this.profilePicture
+            if (!profilePicture.isNullOrEmpty()) {
+                result.profilePicture = profilePicture
             }
-            if (this.profileKey != null) {
+            val profileKey = profileKey
-                result.profileKey = ByteString.copyFrom(this.profileKey)
+            if (profileKey != null) {
+                result.profileKey = ByteString.copyFrom(profileKey)
             }
             return result.build()
         }
     }
 
-    override val isSelfSendValid: Boolean = true
-
     companion object {
 
         fun getCurrent(contacts: List<Contact>): ConfigurationMessage? {
@@ -103,24 +105,22 @@ class ConfigurationMessage(var closedGroups: List<ClosedGroup>, var openGroups:
             val profilePicture = TextSecurePreferences.getProfilePictureURL(context)
             val profileKey = ProfileKeyUtil.getProfileKey(context)
             val groups = storage.getAllGroups()
-            for (groupRecord in groups) {
+            for (group in groups) {
-                if (groupRecord.isClosedGroup) {
+                if (group.isClosedGroup) {
-                    if (!groupRecord.members.contains(Address.fromSerialized(storage.getUserPublicKey()!!))) continue
+                    if (!group.members.contains(Address.fromSerialized(storage.getUserPublicKey()!!))) continue
-                    val groupPublicKey = GroupUtil.doubleDecodeGroupID(groupRecord.encodedId).toHexString()
+                    val groupPublicKey = GroupUtil.doubleDecodeGroupID(group.encodedId).toHexString()
                     val encryptionKeyPair = storage.getLatestClosedGroupEncryptionKeyPair(groupPublicKey) ?: continue
-                    val closedGroup = ClosedGroup(groupPublicKey, groupRecord.title, encryptionKeyPair, groupRecord.members.map { it.serialize() }, groupRecord.admins.map { it.serialize() })
+                    val closedGroup = ClosedGroup(groupPublicKey, group.title, encryptionKeyPair, group.members.map { it.serialize() }, group.admins.map { it.serialize() })
                     closedGroups.add(closedGroup)
                 }
-                if (groupRecord.isOpenGroup) {
+                if (group.isOpenGroup) {
-                    val threadID = storage.getThreadID(groupRecord.encodedId) ?: continue
+                    val threadID = storage.getThreadID(group.encodedId) ?: continue
                     val openGroup = storage.getOpenGroup(threadID)
                     val openGroupV2 = storage.getV2OpenGroup(threadID)
-
+                    val shareUrl = openGroup?.server ?: openGroupV2?.joinURL ?: continue
-                    val shareUrl = openGroup?.server ?: openGroupV2?.toJoinUrl() ?: continue
                     openGroups.add(shareUrl)
                 }
             }
-
             return ConfigurationMessage(closedGroups, openGroups, contacts, displayName, profilePicture, profileKey)
         }
 
@@ -145,6 +145,7 @@ class ConfigurationMessage(var closedGroups: List<ClosedGroup>, var openGroups:
         configurationProto.addAllOpenGroups(openGroups)
         configurationProto.addAllContacts(this.contacts.mapNotNull { it.toProto() })
         configurationProto.displayName = displayName
+        val profilePicture = profilePicture
         if (!profilePicture.isNullOrEmpty()) {
             configurationProto.profilePicture = profilePicture
         }
@@ -157,10 +158,10 @@ class ConfigurationMessage(var closedGroups: List<ClosedGroup>, var openGroups:
     override fun toString(): String {
         return """ 
             ConfigurationMessage(
-                closedGroups: ${(closedGroups)}
+                closedGroups: ${(closedGroups)},
-                openGroups: ${(openGroups)}
+                openGroups: ${(openGroups)},
-                displayName: $displayName
+                displayName: $displayName,
-                profilePicture: $profilePicture
+                profilePicture: $profilePicture,
                 profileKey: $profileKey
             )
         """.trimIndent()

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

@@ -2,5 +2,4 @@ package org.session.libsession.messaging.messages.control
 
 import org.session.libsession.messaging.messages.Message
 
-abstract class ControlMessage : Message() {
+abstract class ControlMessage : Message()
-}

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

@@ -3,7 +3,7 @@ package org.session.libsession.messaging.messages.control
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 import org.session.libsignal.utilities.logging.Log
 
-class DataExtractionNotification(): ControlMessage() {
+class DataExtractionNotification() : ControlMessage() {
     var kind: Kind? = null
 
     sealed class Kind {
@@ -39,8 +39,8 @@ class DataExtractionNotification(): ControlMessage() {
     }
 
     override fun isValid(): Boolean {
-        if (!super.isValid()) return false
+        val kind = kind
-        val kind = kind ?: return false
+        if (!super.isValid() || kind == null) return false
         return when(kind) {
             is Kind.Screenshot -> true
             is Kind.MediaSaved -> kind.timestamp > 0

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

@@ -6,13 +6,20 @@ import org.session.libsignal.utilities.logging.Log
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 
 class ExpirationTimerUpdate() : ControlMessage() {
-    /// In the case of a sync message, the public key of the person the message was targeted at.
+    /** In the case of a sync message, the public key of the person the message was targeted at.
-    /// - Note: `nil` if this isn't a sync message.
+     *
+     * **Note:** `nil` if this isn't a sync message.
+     */
     var syncTarget: String? = null
     var duration: Int? = 0
 
     override val isSelfSendValid: Boolean = true
 
+    override fun isValid(): Boolean {
+        if (!super.isValid()) return false
+        return duration != null
+    }
+
     companion object {
         const val TAG = "ExpirationTimerUpdate"
 
@@ -26,19 +33,14 @@ class ExpirationTimerUpdate() : ControlMessage() {
         }
     }
 
-    internal constructor(syncTarget: String?, duration: Int) : this() {
-        this.syncTarget = syncTarget
-        this.duration = duration
-    }
-
     internal constructor(duration: Int) : this() {
         this.syncTarget = null
         this.duration = duration
     }
 
-    override fun isValid(): Boolean {
+    internal constructor(syncTarget: String, duration: Int) : this() {
-        if (!super.isValid()) return false
+        this.syncTarget = syncTarget
-        return duration != null
+        this.duration = duration
     }
 
     override fun toProto(): SignalServiceProtos.Content? {

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

@@ -6,6 +6,13 @@ import org.session.libsignal.utilities.logging.Log
 class ReadReceipt() : ControlMessage() {
     var timestamps: List<Long>? = null
 
+    override fun isValid(): Boolean {
+        if (!super.isValid()) return false
+        val timestamps = timestamps ?: return false
+        if (timestamps.isNotEmpty()) { return true }
+        return false
+    }
+
     companion object {
         const val TAG = "ReadReceipt"
 
@@ -22,13 +29,6 @@ class ReadReceipt() : ControlMessage() {
         this.timestamps = timestamps
     }
 
-    override fun isValid(): Boolean {
-        if (!super.isValid()) return false
-        val timestamps = timestamps ?: return false
-        if (timestamps.isNotEmpty()) { return true }
-        return false
-    }
-
     override fun toProto(): SignalServiceProtos.Content? {
         val timestamps = timestamps
         if (timestamps == null) {

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

@@ -4,9 +4,15 @@ import org.session.libsignal.service.internal.push.SignalServiceProtos
 import org.session.libsignal.utilities.logging.Log
 
 class TypingIndicator() : ControlMessage() {
-    override val ttl: Long = 30 * 1000
     var kind: Kind? = null
 
+    override val ttl: Long = 20 * 1000
+
+    override fun isValid(): Boolean {
+        if (!super.isValid()) return false
+        return kind != null
+    }
+
     companion object {
         const val TAG = "TypingIndicator"
 
@@ -41,11 +47,6 @@ class TypingIndicator() : ControlMessage() {
         this.kind = kind
     }
 
-    override fun isValid(): Boolean {
-        if (!super.isValid()) return false
-        return kind != null
-    }
-
     override fun toProto(): SignalServiceProtos.Content? {
         val timestamp = sentTimestamp
         val kind = kind

libsession/src/main/java/org/session/libsession/messaging/messages/visible/LinkPreview.kt

@@ -10,6 +10,10 @@ class LinkPreview() {
     var url: String? = null
     var attachmentID: Long? = 0
 
+    fun isValid(): Boolean {
+        return (title != null && url != null && attachmentID != null)
+    }
+
     companion object {
         const val TAG = "LinkPreview"
 
@@ -20,11 +24,8 @@ class LinkPreview() {
         }
 
         fun from(signalLinkPreview: SignalLinkPreiview?): LinkPreview? {
-            return if (signalLinkPreview == null) {
+            if (signalLinkPreview == null) { return null }
-                null
+            return LinkPreview(signalLinkPreview.title, signalLinkPreview.url, signalLinkPreview.attachmentId?.rowId)
-            } else {
-                LinkPreview(signalLinkPreview.title, signalLinkPreview.url, signalLinkPreview.attachmentId?.rowId)
-            }
         }
     }
 
@@ -34,10 +35,6 @@ class LinkPreview() {
         this.attachmentID = attachmentID
     }
 
-    fun isValid(): Boolean {
-        return (title != null && url != null && attachmentID != null)
-    }
-
     fun toProto(): SignalServiceProtos.DataMessage.Preview? {
         val url = url
         if (url == null) {
@@ -46,10 +43,10 @@ class LinkPreview() {
         }
         val linkPreviewProto = SignalServiceProtos.DataMessage.Preview.newBuilder()
         linkPreviewProto.url = url
-        title?.let { linkPreviewProto.title = title }
+        title?.let { linkPreviewProto.title = it }
-        val attachmentID = attachmentID
+        val database = MessagingModuleConfiguration.shared.messageDataProvider
         attachmentID?.let {
-            MessagingModuleConfiguration.shared.messageDataProvider.getSignalAttachmentPointer(attachmentID)?.let {
+            database.getSignalAttachmentPointer(it)?.let {
                 val attachmentProto = Attachment.createAttachmentPointer(it)
                 linkPreviewProto.image = attachmentProto
             }

libsession/src/main/java/org/session/libsession/messaging/messages/visible/Profile.kt

@@ -17,12 +17,11 @@ class Profile() {
             val displayName = profileProto.displayName ?: return null
             val profileKey = proto.profileKey
             val profilePictureURL = profileProto.profilePicture
-            profileKey?.let {
+            if (profileKey != null && profilePictureURL != null) {
-                profilePictureURL?.let {
+                return Profile(displayName, profileKey.toByteArray(), profilePictureURL)
-                    return Profile(displayName = displayName, profileKey = profileKey.toByteArray(), profilePictureURL = profilePictureURL)
+            } else {
-                }
+                return Profile(displayName)
             }
-            return Profile(displayName)
         }
     }
 
@@ -35,16 +34,14 @@ class Profile() {
     fun toProto(): SignalServiceProtos.DataMessage? {
         val displayName = displayName
         if (displayName == null) {
-            Log.w(TAG, "Couldn't construct link preview proto from: $this")
+            Log.w(TAG, "Couldn't construct profile proto from: $this")
             return null
         }
         val dataMessageProto = SignalServiceProtos.DataMessage.newBuilder()
         val profileProto = SignalServiceProtos.DataMessage.LokiProfile.newBuilder()
         profileProto.displayName = displayName
-        val profileKey = profileKey
+        profileKey?.let { dataMessageProto.profileKey = ByteString.copyFrom(it) }
-        profileKey?.let { dataMessageProto.profileKey = ByteString.copyFrom(profileKey) }
+        profilePictureURL?.let { profileProto.profilePicture = it }
-        val profilePictureURL = profilePictureURL
-        profilePictureURL?.let { profileProto.profilePicture = profilePictureURL }
         // Build
         try {
             dataMessageProto.profile = profileProto.build()

libsession/src/main/java/org/session/libsession/messaging/messages/visible/Quote.kt

@@ -13,6 +13,10 @@ class Quote() {
     var text: String? = null
     var attachmentID: Long? = null
 
+    fun isValid(): Boolean {
+        return (timestamp != null && publicKey != null)
+    }
+
     companion object {
         const val TAG = "Quote"
 
@@ -24,12 +28,9 @@ class Quote() {
         }
 
         fun from(signalQuote: SignalQuote?): Quote? {
-            return if (signalQuote == null) {
+            if (signalQuote == null) { return null }
-                null
+            val attachmentID = (signalQuote.attachments?.firstOrNull() as? DatabaseAttachment)?.attachmentId?.rowId
-            } else {
+            return Quote(signalQuote.id, signalQuote.author.serialize(), signalQuote.text, attachmentID)
-                val attachmentID = (signalQuote.attachments?.firstOrNull() as? DatabaseAttachment)?.attachmentId?.rowId
-                Quote(signalQuote.id, signalQuote.author.serialize(), signalQuote.text, attachmentID)
-            }
         }
     }
 
@@ -40,10 +41,6 @@ class Quote() {
         this.attachmentID = attachmentID
     }
 
-    fun isValid(): Boolean {
-        return (timestamp != null && publicKey != null)
-    }
-
     fun toProto(): SignalServiceProtos.DataMessage.Quote? {
         val timestamp = timestamp
         val publicKey = publicKey
@@ -54,7 +51,7 @@ class Quote() {
         val quoteProto = SignalServiceProtos.DataMessage.Quote.newBuilder()
         quoteProto.id = timestamp
         quoteProto.author = publicKey
-        text?.let { quoteProto.text = text }
+        text?.let { quoteProto.text = it }
         addAttachmentsIfNeeded(quoteProto)
         // Build
         try {
@@ -66,23 +63,23 @@ class Quote() {
     }
 
     private fun addAttachmentsIfNeeded(quoteProto: SignalServiceProtos.DataMessage.Quote.Builder) {
-        if (attachmentID == null) return
+        val attachmentID = attachmentID ?: return
-        val attachment = MessagingModuleConfiguration.shared.messageDataProvider.getSignalAttachmentPointer(attachmentID!!)
+        val database = MessagingModuleConfiguration.shared.messageDataProvider
-        if (attachment == null) {
+        val pointer = database.getSignalAttachmentPointer(attachmentID)
+        if (pointer == null) {
             Log.w(TAG, "Ignoring invalid attachment for quoted message.")
             return
         }
-        if (attachment.url.isNullOrEmpty()) {
+        if (pointer.url.isNullOrEmpty()) {
             if (BuildConfig.DEBUG) {
-                //TODO equivalent to iOS's preconditionFailure
+                Log.w(TAG,"Sending a message before all associated attachments have been uploaded.")
-                Log.d(TAG,"Sending a message before all associated attachments have been uploaded.")
                 return
             }
         }
         val quotedAttachmentProto = SignalServiceProtos.DataMessage.Quote.QuotedAttachment.newBuilder()
-        quotedAttachmentProto.contentType = attachment.contentType
+        quotedAttachmentProto.contentType = pointer.contentType
-        if (attachment.fileName.isPresent) quotedAttachmentProto.fileName = attachment.fileName.get()
+        if (pointer.fileName.isPresent) { quotedAttachmentProto.fileName = pointer.fileName.get() }
-        quotedAttachmentProto.thumbnail = Attachment.createAttachmentPointer(attachment)
+        quotedAttachmentProto.thumbnail = Attachment.createAttachmentPointer(pointer)
         try {
             quoteProto.addAttachments(quotedAttachmentProto.build())
         } catch (e: Exception) {

libsession/src/main/java/org/session/libsession/messaging/messages/visible/VisibleMessage.kt

@@ -12,6 +12,10 @@ import org.session.libsignal.utilities.logging.Log
 import org.session.libsession.messaging.sending_receiving.attachments.Attachment as SignalAttachment
 
 class VisibleMessage : Message()  {
+    /** In the case of a sync message, the public key of the person the message was targeted at.
+     *
+     * **Note:** `nil` if this isn't a sync message.
+     */
     var syncTarget: String? = null
     var text: String? = null
     val attachmentIDs: MutableList<Long> = mutableListOf()
@@ -22,51 +26,7 @@ class VisibleMessage : Message()  {
 
     override val isSelfSendValid: Boolean = true
 
-    companion object {
+    // region Validation
-        const val TAG = "VisibleMessage"
-
-        fun fromProto(proto: SignalServiceProtos.Content): VisibleMessage? {
-            val dataMessage = if (proto.hasDataMessage()) proto.dataMessage else return null
-            val result = VisibleMessage()
-            if (dataMessage.hasSyncTarget()) {
-                result.syncTarget = dataMessage.syncTarget
-            }
-            result.text = dataMessage.body
-            // Attachments are handled in MessageReceiver
-            val quoteProto = if (dataMessage.hasQuote()) dataMessage.quote else null
-            quoteProto?.let {
-                val quote = Quote.fromProto(quoteProto)
-                quote?.let { result.quote = quote }
-            }
-            val linkPreviewProto = dataMessage.previewList.firstOrNull()
-            linkPreviewProto?.let {
-                val linkPreview = LinkPreview.fromProto(linkPreviewProto)
-                linkPreview?.let { result.linkPreview = linkPreview }
-            }
-            val openGroupInvitationProto = if (dataMessage.hasOpenGroupInvitation()) dataMessage.openGroupInvitation else null
-            openGroupInvitationProto?.let {
-                val openGroupInvitation = OpenGroupInvitation.fromProto(openGroupInvitationProto)
-                openGroupInvitation?.let { result.openGroupInvitation = openGroupInvitation}
-            }
-            // TODO Contact
-            val profile = Profile.fromProto(dataMessage)
-            profile?.let { result.profile = profile }
-            return  result
-        }
-    }
-
-    fun addSignalAttachments(signalAttachments: List<SignalAttachment>) {
-        val attachmentIDs = signalAttachments.map {
-            val databaseAttachment = it as DatabaseAttachment
-            databaseAttachment.attachmentId.rowId
-        }
-        this.attachmentIDs.addAll(attachmentIDs)
-    }
-
-    fun isMediaMessage(): Boolean {
-        return attachmentIDs.isNotEmpty() || quote != null || linkPreview != null
-    }
-
     override fun isValid(): Boolean {
         if (!super.isValid()) return false
         if (attachmentIDs.isNotEmpty()) return true
@@ -75,61 +35,94 @@ class VisibleMessage : Message()  {
         if (text.isNotEmpty()) return true
         return false
     }
+    // endregion
+
+    // region Proto Conversion
+    companion object {
+        const val TAG = "VisibleMessage"
+
+        fun fromProto(proto: SignalServiceProtos.Content): VisibleMessage? {
+            val dataMessage = proto.dataMessage ?: return null
+            val result = VisibleMessage()
+            if (dataMessage.hasSyncTarget()) { result.syncTarget = dataMessage.syncTarget }
+            result.text = dataMessage.body
+            // Attachments are handled in MessageReceiver
+            val quoteProto = if (dataMessage.hasQuote()) dataMessage.quote else null
+            if (quoteProto != null) {
+                val quote = Quote.fromProto(quoteProto)
+                result.quote = quote
+            }
+            val linkPreviewProto = dataMessage.previewList.firstOrNull()
+            if (linkPreviewProto != null) {
+                val linkPreview = LinkPreview.fromProto(linkPreviewProto)
+                result.linkPreview = linkPreview
+            }
+            val openGroupInvitationProto = if (dataMessage.hasOpenGroupInvitation()) dataMessage.openGroupInvitation else null
+            openGroupInvitationProto?.let {
+                val openGroupInvitation = OpenGroupInvitation.fromProto(openGroupInvitationProto)
+                openGroupInvitation?.let { result.openGroupInvitation = openGroupInvitation}
+            }
+            // TODO Contact
+            val profile = Profile.fromProto(dataMessage)
+            if (profile != null) { result.profile = profile }
+            return  result
+        }
+    }
 
     override fun toProto(): SignalServiceProtos.Content? {
         val proto = SignalServiceProtos.Content.newBuilder()
         val dataMessage: SignalServiceProtos.DataMessage.Builder
         // Profile
-        val profile = profile
+        val profileProto = profile?.let { it.toProto() }
-        val profileProto = profile?.toProto()
         if (profileProto != null) {
             dataMessage = profileProto.toBuilder()
         } else {
             dataMessage = SignalServiceProtos.DataMessage.newBuilder()
         }
         // Text
-        text?.let { dataMessage.body = text }
+        if (text != null) { dataMessage.body = text }
         // Quote
-        quote?.let {
+        val quoteProto = quote?.let { it.toProto() }
-            val quoteProto = it.toProto()
+        if (quoteProto != null) {
-            if (quoteProto != null) dataMessage.quote = quoteProto
+            dataMessage.quote = quoteProto
         }
-        //Link preview
+        // Link preview
-        linkPreview?.let {
+        val linkPreviewProto = linkPreview?.let { it.toProto() }
-            val linkPreviewProto = it.toProto()
+        if (linkPreviewProto != null) {
-            linkPreviewProto?.let {
+            dataMessage.addAllPreview(listOf(linkPreviewProto))
-                dataMessage.addAllPreview(listOf(linkPreviewProto))
-            }
         }
         //Open group invitation
         openGroupInvitation?.let {
             val openGroupInvitationProto = it.toProto()
             if (openGroupInvitationProto != null) dataMessage.openGroupInvitation = openGroupInvitationProto
         }
-        //Attachments
+        // Attachments
-        val attachments = attachmentIDs.mapNotNull { MessagingModuleConfiguration.shared.messageDataProvider.getSignalAttachmentPointer(it) }
+        val database = MessagingModuleConfiguration.shared.messageDataProvider
-        if (!attachments.all { !it.url.isNullOrEmpty() }) {
+        val attachments = attachmentIDs.mapNotNull { database.getSignalAttachmentPointer(it) }
+        if (attachments.any { it.url.isNullOrEmpty() }) {
             if (BuildConfig.DEBUG) {
-                //TODO equivalent to iOS's preconditionFailure
+                Log.w(TAG, "Sending a message before all associated attachments have been uploaded.")
-                Log.d(TAG, "Sending a message before all associated attachments have been uploaded.")
             }
         }
-        val attachmentPointers = attachments.mapNotNull { Attachment.createAttachmentPointer(it) }
+        val pointers = attachments.mapNotNull { Attachment.createAttachmentPointer(it) }
-        dataMessage.addAllAttachments(attachmentPointers)
+        dataMessage.addAllAttachments(pointers)
-        // TODO Contact
+        // TODO: Contact
         // Expiration timer
         // TODO: We * want * expiration timer updates to be explicit. But currently Android will disable the expiration timer for a conversation
-        // if it receives a message without the current expiration timer value attached to it...
+        //       if it receives a message without the current expiration timer value attached to it...
         val storage = MessagingModuleConfiguration.shared.storage
         val context = MessagingModuleConfiguration.shared.context
-        val expiration = if (storage.isClosedGroup(recipient!!)) Recipient.from(context, Address.fromSerialized(GroupUtil.doubleEncodeGroupID(recipient!!)), false).expireMessages
+        val expiration = if (storage.isClosedGroup(recipient!!)) {
-                         else Recipient.from(context, Address.fromSerialized(recipient!!), false).expireMessages
+            Recipient.from(context, Address.fromSerialized(GroupUtil.doubleEncodeGroupID(recipient!!)), false).expireMessages
+        } else {
+            Recipient.from(context, Address.fromSerialized(recipient!!), false).expireMessages
+        }
         dataMessage.expireTimer = expiration
         // Group context
         if (storage.isClosedGroup(recipient!!)) {
             try {
                 setGroupContext(dataMessage)
-            } catch(e: Exception) {
+            } catch (e: Exception) {
                 Log.w(TAG, "Couldn't construct visible message proto from: $this")
                 return null
             }
@@ -147,4 +140,17 @@ class VisibleMessage : Message()  {
             return null
         }
     }
+    // endregion
+
+    fun addSignalAttachments(signalAttachments: List<SignalAttachment>) {
+        val attachmentIDs = signalAttachments.map {
+            val databaseAttachment = it as DatabaseAttachment
+            databaseAttachment.attachmentId.rowId
+        }
+        this.attachmentIDs.addAll(attachmentIDs)
+    }
+
+    fun isMediaMessage(): Boolean {
+        return attachmentIDs.isNotEmpty() || quote != null || linkPreview != null
+    }
 }

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

@@ -5,7 +5,6 @@ import com.fasterxml.jackson.databind.annotation.JsonNaming
 import com.fasterxml.jackson.databind.type.TypeFactory
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.MutableSharedFlow
-import nl.komponents.kovenant.Kovenant
 import nl.komponents.kovenant.Promise
 import nl.komponents.kovenant.functional.bind
 import nl.komponents.kovenant.functional.map
@@ -14,7 +13,6 @@ import okhttp3.HttpUrl
 import okhttp3.MediaType
 import okhttp3.RequestBody
 import org.session.libsession.messaging.MessagingModuleConfiguration
-import org.session.libsession.messaging.open_groups.OpenGroupAPIV2.Error
 import org.session.libsession.snode.OnionRequestAPI
 import org.session.libsession.utilities.AESGCM
 import org.session.libsignal.service.loki.HTTP
@@ -29,108 +27,83 @@ import org.whispersystems.curve25519.Curve25519
 import java.util.*
 
 object OpenGroupAPIV2 {
-
     private val moderators: HashMap<String, Set<String>> = hashMapOf() // Server URL to (channel ID to set of moderator IDs)
-    const val DEFAULT_SERVER = "http://116.203.70.33"
+    private val curve = Curve25519.getInstance(Curve25519.BEST)
-    private const val DEFAULT_SERVER_PUBLIC_KEY = "a03c383cf63c3c4efe67acc52112a6dd734b3a946b9545f488aaa93da7991238"
-
     val defaultRooms = MutableSharedFlow<List<DefaultGroup>>(replay = 1)
 
-    private val curve = Curve25519.getInstance(Curve25519.BEST)
+    private const val DEFAULT_SERVER_PUBLIC_KEY = "a03c383cf63c3c4efe67acc52112a6dd734b3a946b9545f488aaa93da7991238"
-
+    const val DEFAULT_SERVER = "http://116.203.70.33"
-    sealed class Error : Exception() {
-        object GENERIC : Error()
-        object PARSING_FAILED : Error()
-        object DECRYPTION_FAILED : Error()
-        object SIGNING_FAILED : Error()
-        object INVALID_URL : Error()
-        object NO_PUBLIC_KEY : Error()
-
-        fun errorDescription() = when (this) {
-            Error.GENERIC -> "An error occurred."
-            Error.PARSING_FAILED -> "Invalid response."
-            Error.DECRYPTION_FAILED -> "Couldn't decrypt response."
-            Error.SIGNING_FAILED -> "Couldn't sign message."
-            Error.INVALID_URL -> "Invalid URL."
-            Error.NO_PUBLIC_KEY -> "Couldn't find server public key."
-        }
 
+    sealed class Error(message: String) : Exception(message) {
+        object Generic : Error("An error occurred.")
+        object ParsingFailed : Error("Invalid response.")
+        object DecryptionFailed : Error("Couldn't decrypt response.")
+        object SigningFailed : Error("Couldn't sign message.")
+        object InvalidURL : Error("Invalid URL.")
+        object NoPublicKey : Error("Couldn't find server public key.")
     }
 
-    data class DefaultGroup(val id: String,
+    data class DefaultGroup(val id: String, val name: String, val image: ByteArray?) {
-                            val name: String,
+
-                            val image: ByteArray?) {
+        val joinURL: String get() = "$DEFAULT_SERVER/$id?public_key=$DEFAULT_SERVER_PUBLIC_KEY"
-        fun toJoinUrl(): String = "$DEFAULT_SERVER/$id?public_key=$DEFAULT_SERVER_PUBLIC_KEY"
     }
 
-    data class Info(
+    data class Info(val id: String, val name: String, val imageID: String?)
-            val id: String,
-            val name: String,
-            val imageID: String?
-    )
 
     @JsonNaming(PropertyNamingStrategy.SnakeCaseStrategy::class)
-    data class CompactPollRequest(val roomId: String,
+    data class CompactPollRequest(val roomID: String, val authToken: String, val fromDeletionServerID: Long?, val fromMessageServerID: Long?)
-                                  val authToken: String,
+    data class CompactPollResult(val messages: List<OpenGroupMessageV2>, val deletions: List<Long>, val moderators: List<String>)
-                                  val fromDeletionServerId: Long?,
-                                  val fromMessageServerId: Long?
-    )
-
-    data class CompactPollResult(val messages: List<OpenGroupMessageV2>,
-                                 val deletions: List<Long>,
-                                 val moderators: List<String>
-    )
 
     @JsonNaming(PropertyNamingStrategy.SnakeCaseStrategy::class)
-    data class MessageDeletion @JvmOverloads constructor(val id: Long = 0,
+    data class MessageDeletion
-                                                         val deletedMessageId: Long = 0
+        @JvmOverloads constructor(val id: Long = 0, val deletedMessageId: Long = 0
     ) {
+
         companion object {
             val EMPTY = MessageDeletion()
         }
     }
 
     data class Request(
-            val verb: HTTP.Verb,
+        val verb: HTTP.Verb,
-            val room: String?,
+        val room: String?,
-            val server: String,
+        val server: String,
-            val endpoint: String,
+        val endpoint: String,
-            val queryParameters: Map<String, String> = mapOf(),
+        val queryParameters: Map<String, String> = mapOf(),
-            val parameters: Any? = null,
+        val parameters: Any? = null,
-            val headers: Map<String, String> = mapOf(),
+        val headers: Map<String, String> = mapOf(),
-            val isAuthRequired: Boolean = true,
+        val isAuthRequired: Boolean = true,
-            // Always `true` under normal circumstances. You might want to disable
+        /**
-            // this when running over Lokinet.
+         * Always `true` under normal circumstances. You might want to disable
-            val useOnionRouting: Boolean = true
+         * this when running over Lokinet.
+         */
+        val useOnionRouting: Boolean = true
     )
 
     private fun createBody(parameters: Any?): RequestBody? {
         if (parameters == null) return null
-
         val parametersAsJSON = JsonUtil.toJson(parameters)
         return RequestBody.create(MediaType.get("application/json"), parametersAsJSON)
     }
 
     private fun send(request: Request, isJsonRequired: Boolean = true): Promise<Map<*, *>, Exception> {
-        val parsed = HttpUrl.parse(request.server) ?: return Promise.ofFail(Error.INVALID_URL)
+        val url = HttpUrl.parse(request.server) ?: return Promise.ofFail(Error.InvalidURL)
         val urlBuilder = HttpUrl.Builder()
-                .scheme(parsed.scheme())
+            .scheme(url.scheme())
-                .host(parsed.host())
+            .host(url.host())
-                .port(parsed.port())
+            .port(url.port())
-                .addPathSegments(request.endpoint)
+            .addPathSegments(request.endpoint)
-
         if (request.verb == GET) {
             for ((key, value) in request.queryParameters) {
                 urlBuilder.addQueryParameter(key, value)
             }
         }
-
         fun execute(token: String?): Promise<Map<*, *>, Exception> {
             val requestBuilder = okhttp3.Request.Builder()
-                    .url(urlBuilder.build())
+                .url(urlBuilder.build())
-                    .headers(Headers.of(request.headers))
+                .headers(Headers.of(request.headers))
             if (request.isAuthRequired) {
-                if (token.isNullOrEmpty()) throw IllegalStateException("No auth token for request")
+                if (token.isNullOrEmpty()) throw IllegalStateException("No auth token for request.")
                 requestBuilder.header("Authorization", token)
             }
             when (request.verb) {
@@ -139,25 +112,25 @@ object OpenGroupAPIV2 {
                 POST -> requestBuilder.post(createBody(request.parameters)!!)
                 DELETE -> requestBuilder.delete(createBody(request.parameters))
             }
-
             if (!request.room.isNullOrEmpty()) {
                 requestBuilder.header("Room", request.room)
             }
-
             if (request.useOnionRouting) {
                 val publicKey = MessagingModuleConfiguration.shared.storage.getOpenGroupPublicKey(request.server)
-                        ?: return Promise.ofFail(Error.NO_PUBLIC_KEY)
+                    ?: return Promise.ofFail(Error.NoPublicKey)
-                return OnionRequestAPI.sendOnionRequest(requestBuilder.build(), request.server, publicKey, isJSONRequired = isJsonRequired)
+                return OnionRequestAPI.sendOnionRequest(requestBuilder.build(), request.server, publicKey, isJSONRequired = isJsonRequired).fail { e ->
-                        .fail { e ->
+                    // A 401 means that we didn't provide a (valid) auth token for a route that required one. We use this as an
-                            if (e is OnionRequestAPI.HTTPRequestFailedAtDestinationException && e.statusCode == 401) {
+                    // indication that the token we're using has expired. Note that a 403 has a different meaning; it means that
-                                val storage = MessagingModuleConfiguration.shared.storage
+                    // we provided a valid token but it doesn't have a high enough permission level for the route in question.
-                                if (request.room != null) {
+                    if (e is OnionRequestAPI.HTTPRequestFailedAtDestinationException && e.statusCode == 401) {
-                                    storage.removeAuthToken("${request.server}.${request.room}")
+                        val storage = MessagingModuleConfiguration.shared.storage
-                                } else {
+                        if (request.room != null) {
-                                    storage.removeAuthToken(request.server)
+                            storage.removeAuthToken("${request.server}.${request.room}")
-                                }
+                        } else {
-                            }
+                            storage.removeAuthToken(request.server)
                         }
+                    }
+                }
             } else {
                 return Promise.ofFail(IllegalStateException("It's currently not allowed to send non onion routed requests."))
             }
@@ -172,52 +145,51 @@ object OpenGroupAPIV2 {
     fun downloadOpenGroupProfilePicture(roomID: String, server: String): Promise<ByteArray, Exception> {
         val request = Request(verb = GET, room = roomID, server = server, endpoint = "rooms/$roomID/image", isAuthRequired = false)
         return send(request).map { json ->
-            val result = json["result"] as? String ?: throw Error.PARSING_FAILED
+            val result = json["result"] as? String ?: throw Error.ParsingFailed
             decode(result)
         }
     }
 
+    // region Authorization
     fun getAuthToken(room: String, server: String): Promise<String, Exception> {
         val storage = MessagingModuleConfiguration.shared.storage
         return storage.getAuthToken(room, server)?.let {
             Promise.of(it)
         } ?: run {
             requestNewAuthToken(room, server)
-                    .bind { claimAuthToken(it, room, server) }
+                .bind { claimAuthToken(it, room, server) }
-                    .success { authToken ->
+                .success { authToken ->
-                        storage.setAuthToken(room, server, authToken)
+                    storage.setAuthToken(room, server, authToken)
-                    }
+                }
         }
     }
 
     fun requestNewAuthToken(room: String, server: String): Promise<String, Exception> {
         val (publicKey, privateKey) = MessagingModuleConfiguration.shared.storage.getUserKeyPair()
-                ?: return Promise.ofFail(Error.GENERIC)
+            ?: return Promise.ofFail(Error.Generic)
-        val queryParameters = mutableMapOf("public_key" to publicKey)
+        val queryParameters = mutableMapOf( "public_key" to publicKey )
         val request = Request(GET, room, server, "auth_token_challenge", queryParameters, isAuthRequired = false, parameters = null)
         return send(request).map { json ->
-            val challenge = json["challenge"] as? Map<*, *> ?: throw Error.PARSING_FAILED
+            val challenge = json["challenge"] as? Map<*, *> ?: throw Error.ParsingFailed
-            val base64EncodedCiphertext = challenge["ciphertext"] as? String
+            val base64EncodedCiphertext = challenge["ciphertext"] as? String ?: throw Error.ParsingFailed
-                    ?: throw Error.PARSING_FAILED
+            val base64EncodedEphemeralPublicKey = challenge["ephemeral_public_key"] as? String ?: throw Error.ParsingFailed
-            val base64EncodedEphemeralPublicKey = challenge["ephemeral_public_key"] as? String
-                    ?: throw Error.PARSING_FAILED
             val ciphertext = decode(base64EncodedCiphertext)
             val ephemeralPublicKey = decode(base64EncodedEphemeralPublicKey)
             val symmetricKey = AESGCM.generateSymmetricKey(ephemeralPublicKey, privateKey)
             val tokenAsData = try {
                 AESGCM.decrypt(ciphertext, symmetricKey)
             } catch (e: Exception) {
-                throw Error.DECRYPTION_FAILED
+                throw Error.DecryptionFailed
             }
             tokenAsData.toHexString()
         }
     }
 
     fun claimAuthToken(authToken: String, room: String, server: String): Promise<String, Exception> {
-        val parameters = mapOf("public_key" to MessagingModuleConfiguration.shared.storage.getUserPublicKey()!!)
+        val parameters = mapOf( "public_key" to MessagingModuleConfiguration.shared.storage.getUserPublicKey()!! )
-        val headers = mapOf("Authorization" to authToken)
+        val headers = mapOf( "Authorization" to authToken )
         val request = Request(verb = POST, room = room, server = server, endpoint = "claim_auth_token",
-                parameters = parameters, headers = headers, isAuthRequired = false)
+            parameters = parameters, headers = headers, isAuthRequired = false)
         return send(request).map { authToken }
     }
 
@@ -227,33 +199,36 @@ object OpenGroupAPIV2 {
             MessagingModuleConfiguration.shared.storage.removeAuthToken(room, server)
         }
     }
+    // endregion
 
-    // region Sending
+    // region Upload/Download
     fun upload(file: ByteArray, room: String, server: String): Promise<Long, Exception> {
         val base64EncodedFile = encodeBytes(file)
-        val parameters = mapOf("file" to base64EncodedFile)
+        val parameters = mapOf( "file" to base64EncodedFile )
         val request = Request(verb = POST, room = room, server = server, endpoint = "files", parameters = parameters)
         return send(request).map { json ->
-            json["result"] as? Long ?: throw Error.PARSING_FAILED
+            json["result"] as? Long ?: throw Error.ParsingFailed
         }
     }
 
     fun download(file: Long, room: String, server: String): Promise<ByteArray, Exception> {
         val request = Request(verb = GET, room = room, server = server, endpoint = "files/$file")
         return send(request).map { json ->
-            val base64EncodedFile = json["result"] as? String ?: throw Error.PARSING_FAILED
+            val base64EncodedFile = json["result"] as? String ?: throw Error.ParsingFailed
-            decode(base64EncodedFile) ?: throw Error.PARSING_FAILED
+            decode(base64EncodedFile) ?: throw Error.ParsingFailed
         }
     }
+    // endregion
 
+    // region Sending
     fun send(message: OpenGroupMessageV2, room: String, server: String): Promise<OpenGroupMessageV2, Exception> {
-        val signedMessage = message.sign() ?: return Promise.ofFail(Error.SIGNING_FAILED)
+        val signedMessage = message.sign() ?: return Promise.ofFail(Error.SigningFailed)
         val jsonMessage = signedMessage.toJSON()
         val request = Request(verb = POST, room = room, server = server, endpoint = "messages", parameters = jsonMessage)
         return send(request).map { json ->
             @Suppress("UNCHECKED_CAST") val rawMessage = json["message"] as? Map<String, Any>
-                    ?: throw Error.PARSING_FAILED
+                    ?: throw Error.ParsingFailed
-            OpenGroupMessageV2.fromJSON(rawMessage) ?: throw Error.PARSING_FAILED
+            OpenGroupMessageV2.fromJSON(rawMessage) ?: throw Error.ParsingFailed
         }
     }
     // endregion
@@ -266,37 +241,42 @@ object OpenGroupAPIV2 {
             queryParameters += "from_server_id" to lastId.toString()
         }
         val request = Request(verb = GET, room = room, server = server, endpoint = "messages", queryParameters = queryParameters)
-        return send(request).map { jsonList ->
+        return send(request).map { json ->
-            @Suppress("UNCHECKED_CAST") val rawMessages = jsonList["messages"] as? List<Map<String, Any>>
+            @Suppress("UNCHECKED_CAST") val rawMessages = json["messages"] as? List<Map<String, Any>>
-                    ?: throw Error.PARSING_FAILED
+                ?: throw Error.ParsingFailed
-            val lastMessageServerId = storage.getLastMessageServerId(room, server) ?: 0
+            parseMessages(room, server, rawMessages)
-
-            var currentMax = lastMessageServerId
-            val messages = rawMessages.mapNotNull { json ->
-                try {
-                    val message = OpenGroupMessageV2.fromJSON(json) ?: return@mapNotNull null
-                    if (message.serverID == null || message.sender.isNullOrEmpty()) return@mapNotNull null
-                    val sender = message.sender
-                    val data = decode(message.base64EncodedData)
-                    val signature = decode(message.base64EncodedSignature)
-                    val publicKey = Hex.fromStringCondensed(sender.removing05PrefixIfNeeded())
-                    val isValid = curve.verifySignature(publicKey, data, signature)
-                    if (!isValid) {
-                        Log.d("Loki", "Ignoring message with invalid signature")
-                        return@mapNotNull null
-                    }
-                    if (message.serverID > lastMessageServerId) {
-                        currentMax = message.serverID
-                    }
-                    message
-                } catch (e: Exception) {
-                    null
-                }
-            }
-            storage.setLastMessageServerId(room, server, currentMax)
-            messages
         }
     }
+
+    private fun parseMessages(room: String, server: String, rawMessages: List<Map<*, *>>): List<OpenGroupMessageV2> {
+        val storage = MessagingModuleConfiguration.shared.storage
+        val lastMessageServerID = storage.getLastMessageServerId(room, server) ?: 0
+        var currentLastMessageServerID = lastMessageServerID
+        val messages = rawMessages.mapNotNull { json ->
+            json as Map<String, Any>
+            try {
+                val message = OpenGroupMessageV2.fromJSON(json) ?: return@mapNotNull null
+                if (message.serverID == null || message.sender.isNullOrEmpty()) return@mapNotNull null
+                val sender = message.sender
+                val data = decode(message.base64EncodedData)
+                val signature = decode(message.base64EncodedSignature)
+                val publicKey = Hex.fromStringCondensed(sender.removing05PrefixIfNeeded())
+                val isValid = curve.verifySignature(publicKey, data, signature)
+                if (!isValid) {
+                    Log.d("Loki", "Ignoring message with invalid signature.")
+                    return@mapNotNull null
+                }
+                if (message.serverID > lastMessageServerID) {
+                    currentLastMessageServerID = message.serverID
+                }
+                message
+            } catch (e: Exception) {
+                null
+            }
+        }
+        storage.setLastMessageServerId(room, server, currentLastMessageServerID)
+        return messages
+    }
     // endregion
 
     // region Message Deletion
@@ -304,7 +284,7 @@ object OpenGroupAPIV2 {
     fun deleteMessage(serverID: Long, room: String, server: String): Promise<Unit, Exception> {
         val request = Request(verb = DELETE, room = room, server = server, endpoint = "messages/$serverID")
         return send(request).map {
-            Log.d("Loki", "Deleted server message")
+            Log.d("Loki", "Message deletion successful.")
         }
     }
 
@@ -318,7 +298,7 @@ object OpenGroupAPIV2 {
         return send(request).map { json ->
             val type = TypeFactory.defaultInstance().constructCollectionType(List::class.java, MessageDeletion::class.java)
             val idsAsString = JsonUtil.toJson(json["ids"])
-            val serverIDs = JsonUtil.fromJson<List<MessageDeletion>>(idsAsString, type) ?: throw Error.PARSING_FAILED
+            val serverIDs = JsonUtil.fromJson<List<MessageDeletion>>(idsAsString, type) ?: throw Error.ParsingFailed
             val lastMessageServerId = storage.getLastDeletionServerId(room, server) ?: 0
             val serverID = serverIDs.maxByOrNull {it.id } ?: MessageDeletion.EMPTY
             if (serverID.id > lastMessageServerId) {
@@ -338,7 +318,7 @@ object OpenGroupAPIV2 {
         val request = Request(verb = GET, room = room, server = server, endpoint = "moderators")
         return send(request).map { json ->
             @Suppress("UNCHECKED_CAST") val moderatorsJson = json["moderators"] as? List<String>
-                    ?: throw Error.PARSING_FAILED
+                ?: throw Error.ParsingFailed
             val id = "$server.$room"
             handleModerators(id, moderatorsJson)
             moderatorsJson
@@ -347,90 +327,77 @@ object OpenGroupAPIV2 {
 
     @JvmStatic
     fun ban(publicKey: String, room: String, server: String): Promise<Unit, Exception> {
-        val parameters = mapOf("public_key" to publicKey)
+        val parameters = mapOf( "public_key" to publicKey )
         val request = Request(verb = POST, room = room, server = server, endpoint = "block_list", parameters = parameters)
         return send(request).map {
-            Log.d("Loki", "Banned user $publicKey from $server.$room")
+            Log.d("Loki", "Banned user: $publicKey from: $server.$room.")
         }
     }
 
     fun unban(publicKey: String, room: String, server: String): Promise<Unit, Exception> {
         val request = Request(verb = DELETE, room = room, server = server, endpoint = "block_list/$publicKey")
         return send(request).map {
-            Log.d("Loki", "Unbanned user $publicKey from $server.$room")
+            Log.d("Loki", "Unbanned user: $publicKey from: $server.$room")
         }
     }
 
     @JvmStatic
     fun isUserModerator(publicKey: String, room: String, server: String): Boolean =
-            moderators["$server.$room"]?.contains(publicKey) ?: false
+        moderators["$server.$room"]?.contains(publicKey) ?: false
     // endregion
 
     // region General
     @Suppress("UNCHECKED_CAST")
     fun getCompactPoll(rooms: List<String>, server: String): Promise<Map<String, CompactPollResult>, Exception> {
-        val requestAuths = rooms.associateWith { room -> getAuthToken(room, server) }
+        val authTokenRequests = rooms.associateWith { room -> getAuthToken(room, server) }
         val storage = MessagingModuleConfiguration.shared.storage
         val requests = rooms.mapNotNull { room ->
             val authToken = try {
-                requestAuths[room]?.get()
+                authTokenRequests[room]?.get()
             } catch (e: Exception) {
-                Log.e("Loki", "Failed to get auth token for $room", e)
+                Log.e("Loki", "Failed to get auth token for $room.", e)
                 null
             } ?: return@mapNotNull null
-
+            CompactPollRequest(
-            CompactPollRequest(roomId = room,
+                roomID = room,
-                    authToken = authToken,
+                authToken = authToken,
-                    fromDeletionServerId = storage.getLastDeletionServerId(room, server),
+                fromDeletionServerID = storage.getLastDeletionServerId(room, server),
-                    fromMessageServerId = storage.getLastMessageServerId(room, server)
+                fromMessageServerID = storage.getLastMessageServerId(room, server)
             )
         }
-        val request = Request(verb = POST, room = null, server = server, endpoint = "compact_poll", isAuthRequired = false, parameters = mapOf("requests" to requests))
+        val request = Request(verb = POST, room = null, server = server, endpoint = "compact_poll", isAuthRequired = false, parameters = mapOf( "requests" to requests ))
-        // build a request for all rooms
         return send(request = request).map { json ->
-            val results = json["results"] as? List<*> ?: throw Error.PARSING_FAILED
+            val results = json["results"] as? List<*> ?: throw Error.ParsingFailed
-
+            results.mapNotNull { json ->
-            results.mapNotNull { roomJson ->
+                if (json !is Map<*,*>) return@mapNotNull null
-                if (roomJson !is Map<*,*>) return@mapNotNull null
+                val roomID = json["room_id"] as? String ?: return@mapNotNull null
-                val roomId = roomJson["room_id"] as? String ?: return@mapNotNull null
+                // A 401 means that we didn't provide a (valid) auth token for a route that required one. We use this as an
-
+                // indication that the token we're using has expired. Note that a 403 has a different meaning; it means that
-                // check the status was fine
+                // we provided a valid token but it doesn't have a high enough permission level for the route in question.
-                val statusCode = roomJson["status_code"] as? Int ?: return@mapNotNull null
+                val statusCode = json["status_code"] as? Int ?: return@mapNotNull null
                 if (statusCode == 401) {
                     // delete auth token and return null
-                    storage.removeAuthToken(roomId, server)
+                    storage.removeAuthToken(roomID, server)
                 }
-
+                // Moderators
-                // check and store mods
+                val moderators = json["moderators"] as? List<String> ?: return@mapNotNull null
-                val moderators = roomJson["moderators"] as? List<String> ?: return@mapNotNull null
+                handleModerators("$server.$roomID", moderators)
-                handleModerators("$server.$roomId", moderators)
+                // Deletions
-
-                // get deletions
                 val type = TypeFactory.defaultInstance().constructCollectionType(List::class.java, MessageDeletion::class.java)
-                val idsAsString = JsonUtil.toJson(roomJson["deletions"])
+                val idsAsString = JsonUtil.toJson(json["deletions"])
-                val deletedServerIDs = JsonUtil.fromJson<List<MessageDeletion>>(idsAsString, type) ?: throw Error.PARSING_FAILED
+                val deletedServerIDs = JsonUtil.fromJson<List<MessageDeletion>>(idsAsString, type) ?: throw Error.ParsingFailed
-                val lastDeletionServerId = storage.getLastDeletionServerId(roomId, server) ?: 0
+                val lastDeletionServerID = storage.getLastDeletionServerId(roomID, server) ?: 0
-                val serverID = deletedServerIDs.maxByOrNull {it.id } ?: MessageDeletion.EMPTY
+                val serverID = deletedServerIDs.maxByOrNull { it.id } ?: MessageDeletion.EMPTY
-                if (serverID.id > lastDeletionServerId) {
+                if (serverID.id > lastDeletionServerID) {
-                    storage.setLastDeletionServerId(roomId, server, serverID.id)
+                    storage.setLastDeletionServerId(roomID, server, serverID.id)
                 }
-
+                // Messages
-                // get messages
+                val rawMessages = json["messages"] as? List<Map<String, Any>> ?: return@mapNotNull null
-                val rawMessages = roomJson["messages"] as? List<Map<String, Any>> ?: return@mapNotNull null // parsing failed
+                val messages = parseMessages(roomID, server, rawMessages)
-
+                roomID to CompactPollResult(
-                val lastMessageServerId = storage.getLastMessageServerId(roomId, server) ?: 0
+                    messages = messages,
-                var currentMax = lastMessageServerId
+                    deletions = deletedServerIDs.map { it.deletedMessageId },
-                val messages = rawMessages.mapNotNull { rawMessage ->
+                    moderators = moderators
-                    val message = OpenGroupMessageV2.fromJSON(rawMessage)?.apply {
-                        currentMax = maxOf(currentMax,this.serverID ?: 0)
-                    }
-                    message
-                }
-                storage.setLastMessageServerId(roomId, server, currentMax)
-                roomId to CompactPollResult(
-                        messages = messages,
-                        deletions = deletedServerIDs.map { it.deletedMessageId },
-                        moderators = moderators
                 )
             }.toMap()
         }
@@ -443,7 +410,7 @@ object OpenGroupAPIV2 {
             val earlyGroups = groups.map { group ->
                 DefaultGroup(group.id, group.name, null)
             }
-            // see if we have any cached rooms, and if they already have images, don't overwrite with early non-image results
+            // See if we have any cached rooms, and if they already have images don't overwrite them with early non-image results
             defaultRooms.replayCache.firstOrNull()?.let { replayed ->
                 if (replayed.none { it.image?.isNotEmpty() == true}) {
                     defaultRooms.tryEmit(earlyGroups)
@@ -452,12 +419,11 @@ object OpenGroupAPIV2 {
             val images = groups.map { group ->
                 group.id to downloadOpenGroupProfilePicture(group.id, DEFAULT_SERVER)
             }.toMap()
-
             groups.map { group ->
                 val image = try {
                     images[group.id]!!.get()
                 } catch (e: Exception) {
-                    // no image or image failed to download
+                    // No image or image failed to download
                     null
                 }
                 DefaultGroup(group.id, group.name, image)
@@ -470,9 +436,9 @@ object OpenGroupAPIV2 {
     fun getInfo(room: String, server: String): Promise<Info, Exception> {
         val request = Request(verb = GET, room = null, server = server, endpoint = "rooms/$room", isAuthRequired = false)
         return send(request).map { json ->
-            val rawRoom = json["room"] as? Map<*, *> ?: throw Error.PARSING_FAILED
+            val rawRoom = json["room"] as? Map<*, *> ?: throw Error.ParsingFailed
-            val id = rawRoom["id"] as? String ?: throw Error.PARSING_FAILED
+            val id = rawRoom["id"] as? String ?: throw Error.ParsingFailed
-            val name = rawRoom["name"] as? String ?: throw Error.PARSING_FAILED
+            val name = rawRoom["name"] as? String ?: throw Error.ParsingFailed
             val imageID = rawRoom["image_id"] as? String
             Info(id = id, name = name, imageID = imageID)
         }
@@ -481,13 +447,13 @@ object OpenGroupAPIV2 {
     fun getAllRooms(server: String): Promise<List<Info>, Exception> {
         val request = Request(verb = GET, room = null, server = server, endpoint = "rooms", isAuthRequired = false)
         return send(request).map { json ->
-            val rawRooms = json["rooms"] as? List<Map<*, *>> ?: throw Error.PARSING_FAILED
+            val rawRooms = json["rooms"] as? List<Map<*, *>> ?: throw Error.ParsingFailed
             rawRooms.mapNotNull {
                 val roomJson = it as? Map<*, *> ?: return@mapNotNull null
                 val id = roomJson["id"] as? String ?: return@mapNotNull null
                 val name = roomJson["name"] as? String ?: return@mapNotNull null
-                val imageId = roomJson["image_id"] as? String
+                val imageID = roomJson["image_id"] as? String
-                Info(id, name, imageId)
+                Info(id, name, imageID)
             }
         }
     }
@@ -495,12 +461,11 @@ object OpenGroupAPIV2 {
     fun getMemberCount(room: String, server: String): Promise<Int, Exception> {
         val request = Request(verb = GET, room = room, server = server, endpoint = "member_count")
         return send(request).map { json ->
-            val memberCount = json["member_count"] as? Int ?: throw Error.PARSING_FAILED
+            val memberCount = json["member_count"] as? Int ?: throw Error.ParsingFailed
             val storage = MessagingModuleConfiguration.shared.storage
             storage.setUserCount(room, server, memberCount)
             memberCount
         }
     }
     // endregion
-
 }

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

@@ -9,14 +9,18 @@ import org.session.libsignal.utilities.logging.Log
 import org.whispersystems.curve25519.Curve25519
 
 data class OpenGroupMessageV2(
-        val serverID: Long? = null,
+    val serverID: Long? = null,
-        val sender: String?,
+    val sender: String?,
-        val sentTimestamp: Long,
+    val sentTimestamp: Long,
-        // The serialized protobuf in base64 encoding
+    /**
-        val base64EncodedData: String,
+     * The serialized protobuf in base64 encoding.
-        // When sending a message, the sender signs the serialized protobuf with their private key so that
+     */
-        // a receiving user can verify that the message wasn't tampered with.
+    val base64EncodedData: String,
-        val base64EncodedSignature: String? = null
+    /**
+     * When sending a message, the sender signs the serialized protobuf with their private key so that
+     * a receiving user can verify that the message wasn't tampered with.
+     */
+    val base64EncodedSignature: String? = null
 ) {
 
     companion object {
@@ -28,11 +32,12 @@ data class OpenGroupMessageV2(
             val serverID = json["server_id"] as? Int
             val sender = json["public_key"] as? String
             val base64EncodedSignature = json["signature"] as? String
-            return OpenGroupMessageV2(serverID = serverID?.toLong(),
+            return OpenGroupMessageV2(
-                    sender = sender,
+                serverID = serverID?.toLong(),
-                    sentTimestamp = sentTimestamp,
+                sender = sender,
-                    base64EncodedData = base64EncodedData,
+                sentTimestamp = sentTimestamp,
-                    base64EncodedSignature = base64EncodedSignature
+                base64EncodedData = base64EncodedData,
+                base64EncodedSignature = base64EncodedSignature
             )
         }
 
@@ -41,29 +46,26 @@ data class OpenGroupMessageV2(
     fun sign(): OpenGroupMessageV2? {
         if (base64EncodedData.isEmpty()) return null
         val (publicKey, privateKey) = MessagingModuleConfiguration.shared.storage.getUserKeyPair() ?: return null
-
+        if (sender != publicKey) return null
-        if (sender != publicKey) return null // only sign our own messages?
-
         val signature = try {
             curve.calculateSignature(privateKey, decode(base64EncodedData))
         } catch (e: Exception) {
-            Log.e("Loki", "Couldn't sign OpenGroupV2Message", e)
+            Log.w("Loki", "Couldn't sign open group message.", e)
             return null
         }
-
         return copy(base64EncodedSignature = Base64.encodeBytes(signature))
     }
 
     fun toJSON(): Map<String, Any> {
-        val jsonMap = mutableMapOf("data" to base64EncodedData, "timestamp" to sentTimestamp)
+        val json = mutableMapOf( "data" to base64EncodedData, "timestamp" to sentTimestamp )
-        serverID?.let { jsonMap["server_id"] = serverID }
+        serverID?.let { json["server_id"] = it }
-        sender?.let { jsonMap["public_key"] = sender }
+        sender?.let { json["public_key"] = it }
-        base64EncodedSignature?.let { jsonMap["signature"] = base64EncodedSignature }
+        base64EncodedSignature?.let { json["signature"] = it }
-        return jsonMap
+        return json
     }
 
-    fun toProto(): SignalServiceProtos.Content = decode(base64EncodedData).let(PushTransportDetails::getStrippedPaddingMessageBody).let { bytes ->
+    fun toProto(): SignalServiceProtos.Content {
-        SignalServiceProtos.Content.parseFrom(bytes)
+        val data = decode(base64EncodedData).let(PushTransportDetails::getStrippedPaddingMessageBody)
+        return SignalServiceProtos.Content.parseFrom(data)
     }
-
 }

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

@@ -1,51 +1,50 @@
 package org.session.libsession.messaging.open_groups
 
 import org.session.libsignal.utilities.JsonUtil
+import org.session.libsignal.utilities.logging.Log
 import java.util.*
 
 data class OpenGroupV2(
-        val server: String,
+    val server: String,
-        val room: String,
+    val room: String,
-        val id: String,
+    val id: String,
-        val name: String,
+    val name: String,
-        val publicKey: String
+    val publicKey: String
 ) {
 
     constructor(server: String, room: String, name: String, publicKey: String) : this(
-            server = server,
+        server = server,
-            room = room,
+        room = room,
-            id = "$server.$room",
+        id = "$server.$room",
-            name = name,
+        name = name,
-            publicKey = publicKey,
+        publicKey = publicKey,
     )
 
     companion object {
 
-        fun fromJson(jsonAsString: String): OpenGroupV2? {
+        fun fromJSON(jsonAsString: String): OpenGroupV2? {
             return try {
                 val json = JsonUtil.fromJson(jsonAsString)
                 if (!json.has("room")) return null
-
+                val room = json.get("room").asText().toLowerCase(Locale.US)
-                val room = json.get("room").asText().toLowerCase(Locale.getDefault())
+                val server = json.get("server").asText().toLowerCase(Locale.US)
-                val server = json.get("server").asText().toLowerCase(Locale.getDefault())
                 val displayName = json.get("displayName").asText()
                 val publicKey = json.get("publicKey").asText()
-
                 OpenGroupV2(server, room, displayName, publicKey)
             } catch (e: Exception) {
+                Log.w("Loki", "Couldn't parse open group from JSON: $jsonAsString.", e);
                 null
             }
         }
 
     }
 
-    fun toJoinUrl(): String = "$server/$room?public_key=$publicKey"
-
     fun toJson(): Map<String,String> = mapOf(
-            "room" to room,
+        "room" to room,
-            "server" to server,
+        "server" to server,
-            "displayName" to name,
+        "displayName" to name,
-            "publicKey" to publicKey,
+        "publicKey" to publicKey,
     )
 
+    val joinURL: String get() = "$server/$room?public_key=$publicKey"
 }

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

@@ -258,11 +258,11 @@ object MessageSender {
                     }
 
                     val proto = message.toProto()!!
-
+                    val plaintext = PushTransportDetails.getPaddedMessageBody(proto.toByteArray())
                     val openGroupMessage = OpenGroupMessageV2(
                             sender = message.sender,
                             sentTimestamp = message.sentTimestamp!!,
-                            base64EncodedData = Base64.encodeBytes(proto.toByteArray()),
+                            base64EncodedData = Base64.encodeBytes(plaintext),
                     )
 
                     OpenGroupAPIV2.send(openGroupMessage,room,server).success {

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

@@ -128,7 +128,7 @@ private fun handleConfigurationMessage(message: ConfigurationMessage) {
         handleNewClosedGroup(message.sender!!, message.sentTimestamp!!, closeGroup.publicKey, closeGroup.name, closeGroup.encryptionKeyPair!!, closeGroup.members, closeGroup.admins, message.sentTimestamp!!)
     }
     val allOpenGroups = storage.getAllOpenGroups().map { it.value.server }
-    val allV2OpenGroups = storage.getAllV2OpenGroups().map { it.value.toJoinUrl() }
+    val allV2OpenGroups = storage.getAllV2OpenGroups().map { it.value.joinURL }
     for (openGroup in message.openGroups) {
         if (allOpenGroups.contains(openGroup) || allV2OpenGroups.contains(openGroup)) continue
         storage.addOpenGroup(openGroup, 1)
@@ -162,7 +162,12 @@ fun MessageReceiver.handleVisibleMessage(message: VisibleMessage, proto: SignalS
 
     // Get or create thread
     val threadID = storage.getOrCreateThreadIdFor(message.syncTarget
-            ?: message.sender!!, message.groupPublicKey, openGroupID)
+        ?: message.sender!!, message.groupPublicKey, openGroupID)
+
+    if (threadID < 0) {
+        // thread doesn't exist, should only be reached in a case where we are processing open group messages for no longer existent thread
+        throw MessageReceiver.Error.NoThread
+    }
 
     val openGroup = threadID.let {
         storage.getOpenGroup(it.toString())
@@ -242,7 +247,7 @@ fun MessageReceiver.handleVisibleMessage(message: VisibleMessage, proto: SignalS
     }
     val openGroupServerID = message.openGroupServerMessageID
     if (openGroupServerID != null) {
-        storage.setOpenGroupServerMessageID(messageID, openGroupServerID, threadID, !(message.isMediaMessage() || attachments.isNotEmpty()))
+        storage.setOpenGroupServerMessageID(messageID, openGroupServerID, threadID, !message.isMediaMessage())
     }
     // Cancel any typing indicators if needed
     cancelTypingIndicatorsIfNeeded(message.sender!!)

libsession/src/main/java/org/session/libsession/messaging/utilities/Data.java

@@ -1,4 +1,4 @@
-package org.session.libsession.messaging.jobs;
+package org.session.libsession.messaging.utilities;
 
 import android.os.Parcelable;
 
@@ -12,11 +12,7 @@ import org.session.libsession.utilities.ParcelableUtil;
 import java.util.HashMap;
 import java.util.Map;
 
-// Introduce a dedicated Map<String, byte[]> field specifically for parcelable needs.
 public class Data {
-
-    public static final Data EMPTY = new Data.Builder().build();
-
     @JsonProperty private final Map<String, String>    strings;
     @JsonProperty private final Map<String, String[]>  stringArrays;
     @JsonProperty private final Map<String, Integer>   integers;
@@ -31,20 +27,23 @@ public class Data {
     @JsonProperty private final Map<String, boolean[]> booleanArrays;
     @JsonProperty private final Map<String, byte[]>    byteArrays;
 
-    public Data(@JsonProperty("strings")       @NonNull Map<String, String>    strings,
+    public static final Data EMPTY = new Data.Builder().build();
-                @JsonProperty("stringArrays")  @NonNull Map<String, String[]>  stringArrays,
+
-                @JsonProperty("integers")      @NonNull Map<String, Integer>   integers,
+    public Data(
-                @JsonProperty("integerArrays") @NonNull Map<String, int[]>     integerArrays,
+            @JsonProperty("strings")       @NonNull Map<String, String>    strings,
-                @JsonProperty("longs")         @NonNull Map<String, Long>      longs,
+            @JsonProperty("stringArrays")  @NonNull Map<String, String[]>  stringArrays,
-                @JsonProperty("longArrays")    @NonNull Map<String, long[]>    longArrays,
+            @JsonProperty("integers")      @NonNull Map<String, Integer>   integers,
-                @JsonProperty("floats")        @NonNull Map<String, Float>     floats,
+            @JsonProperty("integerArrays") @NonNull Map<String, int[]>     integerArrays,
-                @JsonProperty("floatArrays")   @NonNull Map<String, float[]>   floatArrays,
+            @JsonProperty("longs")         @NonNull Map<String, Long>      longs,
-                @JsonProperty("doubles")       @NonNull Map<String, Double>    doubles,
+            @JsonProperty("longArrays")    @NonNull Map<String, long[]>    longArrays,
-                @JsonProperty("doubleArrays")  @NonNull Map<String, double[]>  doubleArrays,
+            @JsonProperty("floats")        @NonNull Map<String, Float>     floats,
-                @JsonProperty("booleans")      @NonNull Map<String, Boolean>   booleans,
+            @JsonProperty("floatArrays")   @NonNull Map<String, float[]>   floatArrays,
-                @JsonProperty("booleanArrays") @NonNull Map<String, boolean[]> booleanArrays,
+            @JsonProperty("doubles")       @NonNull Map<String, Double>    doubles,
-                @JsonProperty("byteArrays")    @NonNull Map<String, byte[]>    byteArrays)
+            @JsonProperty("doubleArrays")  @NonNull Map<String, double[]>  doubleArrays,
-    {
+            @JsonProperty("booleans")      @NonNull Map<String, Boolean>   booleans,
+            @JsonProperty("booleanArrays") @NonNull Map<String, boolean[]> booleanArrays,
+            @JsonProperty("byteArrays")    @NonNull Map<String, byte[]>    byteArrays
+    ) {
         this.strings       = strings;
         this.stringArrays  = stringArrays;
         this.integers      = integers;
@@ -75,6 +74,7 @@ public class Data {
     }
 
 
+
     public boolean hasStringArray(@NonNull String key) {
         return stringArrays.containsKey(key);
     }
@@ -100,6 +100,7 @@ public class Data {
     }
 
 
+
     public boolean hasIntegerArray(@NonNull String key) {
         return integerArrays.containsKey(key);
     }
@@ -110,6 +111,7 @@ public class Data {
     }
 
 
+
     public boolean hasLong(@NonNull String key) {
         return longs.containsKey(key);
     }
@@ -125,6 +127,7 @@ public class Data {
     }
 
 
+
     public boolean hasLongArray(@NonNull String key) {
         return longArrays.containsKey(key);
     }
@@ -135,6 +138,7 @@ public class Data {
     }
 
 
+
     public boolean hasFloat(@NonNull String key) {
         return floats.containsKey(key);
     }
@@ -150,6 +154,7 @@ public class Data {
     }
 
 
+
     public boolean hasFloatArray(@NonNull String key) {
         return floatArrays.containsKey(key);
     }
@@ -160,6 +165,7 @@ public class Data {
     }
 
 
+
     public boolean hasDouble(@NonNull String key) {
         return doubles.containsKey(key);
     }
@@ -175,6 +181,7 @@ public class Data {
     }
 
 
+
     public boolean hasDoubleArray(@NonNull String key) {
         return floatArrays.containsKey(key);
     }
@@ -185,6 +192,7 @@ public class Data {
     }
 
 
+
     public boolean hasBoolean(@NonNull String key) {
         return booleans.containsKey(key);
     }
@@ -200,6 +208,7 @@ public class Data {
     }
 
 
+
     public boolean hasBooleanArray(@NonNull String key) {
         return booleanArrays.containsKey(key);
     }
@@ -209,6 +218,8 @@ public class Data {
         return booleanArrays.get(key);
     }
 
+
+
     public boolean hasByteArray(@NonNull String key) {
         return byteArrays.containsKey(key);
     }
@@ -218,6 +229,8 @@ public class Data {
         return byteArrays.get(key);
     }
 
+
+
     public boolean hasParcelable(@NonNull String key) {
         return byteArrays.containsKey(key);
     }
@@ -228,6 +241,8 @@ public class Data {
         return ParcelableUtil.unmarshall(bytes, creator);
     }
 
+
+
     private void throwIfAbsent(@NonNull Map map, @NonNull String key) {
         if (!map.containsKey(key)) {
             throw new IllegalStateException("Tried to retrieve a value with key '" + key + "', but it wasn't present.");
@@ -236,7 +251,6 @@ public class Data {
 
 
     public static class Builder {
-
         private final Map<String, String>    strings       = new HashMap<>();
         private final Map<String, String[]>  stringArrays  = new HashMap<>();
         private final Map<String, Integer>   integers      = new HashMap<>();
@@ -323,19 +337,21 @@ public class Data {
         }
 
         public Data build() {
-            return new Data(strings,
+            return new Data(
-                    stringArrays,
+                strings,
-                    integers,
+                stringArrays,
-                    integerArrays,
+                integers,
-                    longs,
+                integerArrays,
-                    longArrays,
+                longs,
-                    floats,
+                longArrays,
-                    floatArrays,
+                floats,
-                    doubles,
+                floatArrays,
-                    doubleArrays,
+                doubles,
-                    booleans,
+                doubleArrays,
-                    booleanArrays,
+                booleans,
-                    byteArrays);
+                booleanArrays,
+                byteArrays
+            );
         }
     }
 
@@ -343,5 +359,4 @@ public class Data {
         @NonNull String serialize(@NonNull Data data);
         @NonNull Data deserialize(@NonNull String serialized);
     }
-}
+}
-

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

@@ -53,11 +53,11 @@ object OnionRequestAPI {
     /**
      * The number of times a path can fail before it's replaced.
      */
-    private const val pathFailureThreshold = 1
+    private const val pathFailureThreshold = 3
     /**
      * The number of times a snode can fail before it's replaced.
      */
-    private const val snodeFailureThreshold = 1
+    private const val snodeFailureThreshold = 3
     /**
      * The number of guard snodes required to maintain `targetPathCount` paths.
      */
@@ -93,7 +93,7 @@ object OnionRequestAPI {
         ThreadUtils.queue { // No need to block the shared context for this
             val url = "${snode.address}:${snode.port}/get_stats/v1"
             try {
-                val json = HTTP.execute(HTTP.Verb.GET, url)
+                val json = HTTP.execute(HTTP.Verb.GET, url, 3)
                 val version = json["version"] as? String
                 if (version == null) { deferred.reject(Exception("Missing snode version.")); return@queue }
                 if (version >= "2.0.7") {
@@ -463,7 +463,6 @@ object OnionRequestAPI {
             "method" to request.method(),
             "headers" to headers
         )
-        url.isHttps
         val destination = Destination.Server(host, target, x25519PublicKey, url.scheme(), url.port())
         return sendOnionRequest(destination, payload, isJSONRequired).recover { exception ->
             Log.d("Loki", "Couldn't reach server: $urlAsString due to error: $exception.")

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

@@ -71,11 +71,11 @@ object OnionRequestEncryption {
                     }
                     is OnionRequestAPI.Destination.Server -> {
                         payload = mutableMapOf(
-                                "host" to rhs.host,
+                            "host" to rhs.host,
-                                "target" to rhs.target,
+                            "target" to rhs.target,
-                                "method" to "POST",
+                            "method" to "POST",
-                                "protocol" to rhs.scheme,
+                            "protocol" to rhs.scheme,
-                                "port" to rhs.port
+                            "port" to rhs.port
                         )
                     }
                 }

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

@@ -34,7 +34,7 @@ object SnodeAPI {
     // Settings
     private val maxRetryCount = 6
     private val minimumSnodePoolCount = 12
-    private val minimumSwarmSnodeCount = 2
+    private val minimumSwarmSnodeCount = 3
     // Use port 4433 if the API level can handle the network security configuration and enforce pinned certificates
     private val seedNodePort = if (Build.VERSION.SDK_INT < Build.VERSION_CODES.N) 443 else 4433
     private val seedNodePool by lazy {
@@ -44,7 +44,7 @@ object SnodeAPI {
             setOf( "https://storage.seed1.loki.network:$seedNodePort ", "https://storage.seed3.loki.network:$seedNodePort ", "https://public.loki.foundation:$seedNodePort" )
         }
     }
-    private val snodeFailureThreshold = 4
+    private val snodeFailureThreshold = 3
     private val targetSwarmSnodeCount = 2
     private val useOnionRequests = true
 
@@ -252,19 +252,20 @@ object SnodeAPI {
 
     private fun removeDuplicates(publicKey: String, rawMessages: List<*>): List<*> {
         val receivedMessageHashValues = database.getReceivedMessageHashValues(publicKey)?.toMutableSet() ?: mutableSetOf()
-        return rawMessages.filter { rawMessage ->
+        val result = rawMessages.filter { rawMessage ->
             val rawMessageAsJSON = rawMessage as? Map<*, *>
             val hashValue = rawMessageAsJSON?.get("hash") as? String
             if (hashValue != null) {
                 val isDuplicate = receivedMessageHashValues.contains(hashValue)
                 receivedMessageHashValues.add(hashValue)
-                database.setReceivedMessageHashValues(publicKey, receivedMessageHashValues)
                 !isDuplicate
             } else {
                 Log.d("Loki", "Missing hash value for message: ${rawMessage?.prettifiedDescription()}.")
                 false
             }
         }
+        database.setReceivedMessageHashValues(publicKey, receivedMessageHashValues)
+        return result
     }
 
     private fun parseEnvelopes(rawMessages: List<*>): List<SignalServiceProtos.Envelope> {
@@ -305,7 +306,7 @@ object SnodeAPI {
             }
         }
         when (statusCode) {
-            400, 500, 503 -> { // Usually indicates that the snode isn't up to date
+            400, 500, 502, 503 -> { // Usually indicates that the snode isn't up to date
                 handleBadSnode()
             }
             406 -> {
@@ -316,8 +317,20 @@ object SnodeAPI {
             421 -> {
                 // The snode isn't associated with the given public key anymore
                 if (publicKey != null) {
-                    Log.d("Loki", "Invalidating swarm for: $publicKey.")
+                    fun invalidateSwarm() {
-                    dropSnodeFromSwarmIfNeeded(snode, publicKey)
+                        Log.d("Loki", "Invalidating swarm for: $publicKey.")
+                        dropSnodeFromSwarmIfNeeded(snode, publicKey)
+                    }
+                    if (json != null) {
+                        val snodes = parseSnodes(json)
+                        if (snodes.isNotEmpty()) {
+                            database.setSwarm(publicKey, snodes.toSet())
+                        } else {
+                            invalidateSwarm()
+                        }
+                    } else {
+                        invalidateSwarm()
+                    }
                 } else {
                     Log.d("Loki", "Got a 421 without an associated public key.")
                 }

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

@@ -3,23 +3,33 @@ package org.session.libsession.snode
 import org.session.libsignal.service.loki.utilities.removing05PrefixIfNeeded
 
 data class SnodeMessage(
-    // The hex encoded public key of the recipient.
+    /**
+     * The hex encoded public key of the recipient.
+     */
     val recipient: String,
-    // The content of the message.
+    /**
+     * The content of the message.
+     */
     val data: String,
-    // The time to live for the message in milliseconds.
+    /**
+     * The time to live for the message in milliseconds.
+     */
     val ttl: Long,
-    // When the proof of work was calculated.
+    /**
+     * When the proof of work was calculated.
+     *
+     * **Note:** Expressed as milliseconds since 00:00:00 UTC on 1 January 1970.
+     */
     val timestamp: Long
 ) {
 
     internal fun toJSON(): Map<String, String> {
         return mapOf(
             "pubKey" to if (SnodeAPI.useTestnet) recipient.removing05PrefixIfNeeded() else recipient,
-                "data" to data,
+            "data" to data,
-                "ttl" to ttl.toString(),
+            "ttl" to ttl.toString(),
-                "timestamp" to timestamp.toString(),
+            "timestamp" to timestamp.toString(),
-                "nonce" to ""
+            "nonce" to ""
         )
     }
 }

libsession/src/main/java/org/session/libsession/snode/utilities/Random.kt

@@ -6,6 +6,7 @@ import java.security.SecureRandom
  * Uses `SecureRandom` to pick an element from this collection.
  */
 fun <T> Collection<T>.getRandomElementOrNull(): T? {
+    if (isEmpty()) return null
     val index = SecureRandom().nextInt(size) // SecureRandom() should be cryptographically secure
     return elementAtOrNull(index)
 }

libsignal/src/main/java/org/session/libsignal/service/loki/HTTP.kt

@@ -3,6 +3,7 @@ package org.session.libsignal.service.loki
 import okhttp3.*
 import org.session.libsignal.utilities.logging.Log
 import org.session.libsignal.utilities.JsonUtil
+import java.lang.IllegalStateException
 import java.security.SecureRandom
 import java.security.cert.X509Certificate
 import java.util.concurrent.TimeUnit
@@ -25,9 +26,7 @@ object HTTP {
 
             override fun checkClientTrusted(chain: Array<out X509Certificate>?, authorizationType: String?) { }
             override fun checkServerTrusted(chain: Array<out X509Certificate>?, authorizationType: String?) { }
-            override fun getAcceptedIssuers(): Array<X509Certificate> {
+            override fun getAcceptedIssuers(): Array<X509Certificate> { return arrayOf() }
-                return arrayOf()
-            }
         }
         val sslContext = SSLContext.getInstance("SSL")
         sslContext.init(null, arrayOf( trustManager ), SecureRandom())
@@ -40,7 +39,26 @@ object HTTP {
             .build()
     }
 
-    private const val timeout: Long = 20
+    private fun getDefaultConnection(timeout: Long): OkHttpClient {
+        // Snode to snode communication uses self-signed certificates but clients can safely ignore this
+        val trustManager = object : X509TrustManager {
+
+            override fun checkClientTrusted(chain: Array<out X509Certificate>?, authorizationType: String?) { }
+            override fun checkServerTrusted(chain: Array<out X509Certificate>?, authorizationType: String?) { }
+            override fun getAcceptedIssuers(): Array<X509Certificate> { return arrayOf() }
+        }
+        val sslContext = SSLContext.getInstance("SSL")
+        sslContext.init(null, arrayOf( trustManager ), SecureRandom())
+        return OkHttpClient().newBuilder()
+            .sslSocketFactory(sslContext.socketFactory, trustManager)
+            .hostnameVerifier { _, _ -> true }
+            .connectTimeout(timeout, TimeUnit.SECONDS)
+            .readTimeout(timeout, TimeUnit.SECONDS)
+            .writeTimeout(timeout, TimeUnit.SECONDS)
+            .build()
+    }
+
+    private const val timeout: Long = 10
 
     class HTTPRequestFailedException(val statusCode: Int, val json: Map<*, *>?)
         : kotlin.Exception("HTTP request failed with status code $statusCode.")
@@ -52,26 +70,26 @@ object HTTP {
     /**
      * Sync. Don't call from the main thread.
      */
-    fun execute(verb: Verb, url: String, useSeedNodeConnection: Boolean = false): Map<*, *> {
+    fun execute(verb: Verb, url: String, timeout: Long = HTTP.timeout, useSeedNodeConnection: Boolean = false): Map<*, *> {
-        return execute(verb = verb, url = url, body = null, useSeedNodeConnection = useSeedNodeConnection)
+        return execute(verb = verb, url = url, body = null, timeout = timeout, useSeedNodeConnection = useSeedNodeConnection)
     }
 
     /**
      * Sync. Don't call from the main thread.
      */
-    fun execute(verb: Verb, url: String, parameters: Map<String, Any>?, useSeedNodeConnection: Boolean = false): Map<*, *> {
+    fun execute(verb: Verb, url: String, parameters: Map<String, Any>?, timeout: Long = HTTP.timeout, useSeedNodeConnection: Boolean = false): Map<*, *> {
         if (parameters != null) {
             val body = JsonUtil.toJson(parameters).toByteArray()
-            return execute(verb = verb, url = url, body = body, useSeedNodeConnection = useSeedNodeConnection)
+            return execute(verb = verb, url = url, body = body, timeout = timeout, useSeedNodeConnection = useSeedNodeConnection)
         } else {
-            return execute(verb = verb, url = url, body = null, useSeedNodeConnection = useSeedNodeConnection)
+            return execute(verb = verb, url = url, body = null, timeout = timeout, useSeedNodeConnection = useSeedNodeConnection)
         }
     }
 
     /**
      * Sync. Don't call from the main thread.
      */
-    fun execute(verb: Verb, url: String, body: ByteArray?, useSeedNodeConnection: Boolean = false): Map<*, *> {
+    fun execute(verb: Verb, url: String, body: ByteArray?, timeout: Long = HTTP.timeout, useSeedNodeConnection: Boolean = false): Map<*, *> {
         val request = Request.Builder().url(url)
         when (verb) {
             Verb.GET -> request.get()
@@ -85,7 +103,15 @@ object HTTP {
         }
         lateinit var response: Response
         try {
-            val connection = if (useSeedNodeConnection) seedNodeConnection else defaultConnection
+            val connection: OkHttpClient
+            if (timeout != HTTP.timeout) { // Custom timeout
+                if (useSeedNodeConnection) {
+                    throw IllegalStateException("Setting a custom timeout is only allowed for requests to snodes.")
+                }
+                connection = getDefaultConnection(timeout)
+            } else {
+                connection = if (useSeedNodeConnection) seedNodeConnection else defaultConnection
+            }
             response = connection.newCall(request.build()).execute()
         } catch (exception: Exception) {
             Log.d("Loki", "${verb.rawValue} request to $url failed due to error: ${exception.localizedMessage}.")

libsignal/src/main/java/org/session/libsignal/service/loki/Random.kt

@@ -6,6 +6,7 @@ import java.security.SecureRandom
  * Uses `SecureRandom` to pick an element from this collection.
  */
 fun <T> Collection<T>.getRandomElementOrNull(): T? {
+    if (isEmpty()) return null
     val index = SecureRandom().nextInt(size) // SecureRandom() should be cryptographically secure
     return elementAtOrNull(index)
 }