fix: getting local user's ID if message is outgoing to check against that value in attachment download

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

@@ -119,6 +119,13 @@ class DatabaseAttachmentProvider(context: Context, helper: SQLCipherOpenHelper)
         ), threadId)
     }
 
+    override fun isMmsOutgoing(mmsMessageId: Long): Boolean {
+        val mmsDb = DatabaseFactory.getMmsDatabase(context)
+        return mmsDb.getMessage(mmsMessageId).use { cursor ->
+            mmsDb.readerFor(cursor).next
+        }.isOutgoing
+    }
+
     override fun isOutgoingMessage(timestamp: Long): Boolean {
         val smsDatabase = DatabaseFactory.getSmsDatabase(context)
         val mmsDatabase = DatabaseFactory.getMmsDatabase(context)

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

@@ -22,7 +22,8 @@ interface MessageDataProvider {
     fun setAttachmentState(attachmentState: AttachmentState, attachmentId: AttachmentId, messageID: Long)
     fun insertAttachment(messageId: Long, attachmentId: AttachmentId, stream : InputStream)
     fun updateAudioAttachmentDuration(attachmentId: AttachmentId, durationMs: Long, threadId: Long)
-    fun isOutgoingMessage(timestamp: Long): Boolean
+    fun isMmsOutgoing(mmsMessageId: Long): Boolean
+    fun isOutgoingMessage(mmsId: Long): Boolean
     fun handleSuccessfulAttachmentUpload(attachmentId: Long, attachmentStream: SignalServiceAttachmentStream, attachmentKey: ByteArray, uploadResult: UploadResult)
     fun handleFailedAttachmentUpload(attachmentId: Long)
     fun getMessageForQuote(timestamp: Long, author: Address): Pair<Long, Boolean>?

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

@@ -29,6 +29,7 @@ class AttachmentDownloadJob(val attachmentID: Long, val databaseMessageID: Long)
         object NoAttachment : Error("No such attachment.")
         object NoThread: Error("Thread no longer exists")
         object NoSender: Error("Thread recipient or sender does not exist")
+        object DuplicateData: Error("Attachment already downloaded")
     }
 
     // Settings
@@ -64,20 +65,24 @@ class AttachmentDownloadJob(val attachmentID: Long, val databaseMessageID: Long)
         }
 
         if (threadID < 0) {
-            Log.e("Loki", "Thread doesn't exist for database message ID $databaseMessageID")
             handleFailure(Error.NoThread, null)
             return
         }
 
         val threadRecipient = storage.getRecipientForThread(threadID)
-        val sender = messageDataProvider.getIndividualRecipientForMms(databaseMessageID)
+        val sender = if (messageDataProvider.isMmsOutgoing(databaseMessageID)) {
-        val contact = sender?.address?.let { storage.getContactWithSessionID(it.serialize()) }
+            storage.getUserPublicKey()
+        } else {
+            messageDataProvider.getIndividualRecipientForMms(databaseMessageID)?.address?.serialize()
+        }
+        val contact = sender?.let { storage.getContactWithSessionID(it) }
         if (threadRecipient == null || sender == null || contact == null) {
             handleFailure(Error.NoSender, null)
             return
         }
-        if (!threadRecipient.isGroupRecipient && !(contact.isTrusted || storage.getUserPublicKey() != sender.address.serialize())) {
+        if (!threadRecipient.isGroupRecipient && (!contact.isTrusted && storage.getUserPublicKey() != sender)) {
-            Log.e("Loki", "Thread isn't a group recipient, or contact isn't trusted or self-send")
+            // if we aren't receiving a group message, a message from ourselves (self-send) and the contact sending is not trusted:
+            // do not continue, but do not fail
             return
         }
 
@@ -86,7 +91,8 @@ class AttachmentDownloadJob(val attachmentID: Long, val databaseMessageID: Long)
             val attachment = messageDataProvider.getDatabaseAttachment(attachmentID)
                 ?: return handleFailure(Error.NoAttachment, null)
             if (attachment.hasData()) {
-                Log.d("Loki", "The attachment $attachmentID already has data")
+                handleFailure(Error.DuplicateData, attachment.attachmentId)
+                return
             }
             messageDataProvider.setAttachmentState(AttachmentState.STARTED, attachment.attachmentId, this.databaseMessageID)
             tempFile = createTempFile()

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

@@ -20,7 +20,7 @@ class JobQueue : JobDelegate {
     private val jobTimestampMap = ConcurrentHashMap<Long, AtomicInteger>()
     private val rxDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
     private val txDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
-    private val attachmentDispatcher = Executors.newFixedThreadPool(2).asCoroutineDispatcher()
+    private val attachmentDispatcher = Executors.newFixedThreadPool(4).asCoroutineDispatcher()
     private val scope = GlobalScope + SupervisorJob()
     private val queue = Channel<Job>(UNLIMITED)
     private val pendingJobIds = mutableSetOf<String>()