Ensure story media is only uploaded once.

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

@@ -7,6 +7,7 @@ import android.os.Build;
 import androidx.annotation.GuardedBy;
 import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;
+import androidx.annotation.VisibleForTesting;
 import androidx.annotation.WorkerThread;
 
 import org.signal.core.util.ThreadUtil;
@@ -459,7 +460,8 @@ public class JobManager implements ConstraintObserver.Notifier {
     private final JobManager jobManager;
     private final List<List<Job>> jobs;
 
-    private Chain(@NonNull JobManager jobManager, @NonNull List<? extends Job> jobs) {
+    @VisibleForTesting
+    public Chain(@NonNull JobManager jobManager, @NonNull List<? extends Job> jobs) {
       this.jobManager = jobManager;
       this.jobs       = new LinkedList<>();
 
@@ -489,7 +491,8 @@ public class JobManager implements ConstraintObserver.Notifier {
       enqueue();
     }
 
-    private List<List<Job>> getJobListChain() {
+    @VisibleForTesting
+    public List<List<Job>> getJobListChain() {
       return jobs;
     }
   }

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

@@ -87,10 +87,12 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 public class MessageSender {
 
@@ -135,13 +137,15 @@ public class MessageSender {
   public static void sendStories(@NonNull final Context context,
                                  @NonNull final List<OutgoingSecureMediaMessage> messages,
                                  @Nullable final String metricId,
-                                 @Nullable final SmsDatabase.InsertListener insertListener)
+                                 @Nullable final SmsDatabase.InsertListener insertListener
+                                 /** TODO [alex] -- Preupload set if preuploads were of valid length -- */)
   {
     Log.i(TAG, "Sending story messages to " + messages.size() + " targets.");
     ThreadDatabase        threadDatabase  = SignalDatabase.threads();
     MessageDatabase       database        = SignalDatabase.mms();
     List<Long>            messageIds      = new ArrayList<>(messages.size());
     List<Long>            threads         = new ArrayList<>(messages.size());
+    UploadDependencyGraph dependencyGraph = UploadDependencyGraph.EMPTY;
 
     try {
       database.beginTransaction();
@@ -172,6 +176,36 @@ public class MessageSender {
         }
       }
 
+      dependencyGraph = UploadDependencyGraph.create(
+          messages,
+          ApplicationDependencies.getJobManager(),
+          attachment -> {
+            try {
+              return SignalDatabase.attachments().insertAttachmentForPreUpload(attachment);
+            } catch (MmsException e) {
+              Log.e(TAG, e);
+              throw new IllegalStateException(e);
+            }
+          }
+      );
+
+      for (int i = 0; i < messageIds.size(); i++) {
+        long                             messageId = messageIds.get(i);
+        OutgoingSecureMediaMessage       message   = messages.get(i);
+        List<UploadDependencyGraph.Node> nodes     = dependencyGraph.getDependencyMap().get(message);
+
+        if (nodes == null || nodes.isEmpty()) {
+          Log.d(TAG, "No attachments for given message. Skipping.");
+          continue;
+        }
+
+        List<AttachmentId> attachmentIds = nodes.stream().map(UploadDependencyGraph.Node::getAttachmentId).collect(Collectors.toList());
+        SignalDatabase.attachments().updateMessageId(attachmentIds, messageId, true);
+        for (final AttachmentId attachmentId : attachmentIds) {
+          SignalDatabase.attachments().updateAttachmentCaption(attachmentId, message.getBody());
+        }
+      }
+
       database.setTransactionSuccessful();
     } catch (MmsException e) {
       Log.w(TAG, e);
@@ -179,12 +213,25 @@ public class MessageSender {
       database.endTransaction();
     }
 
+    List<JobManager.Chain> chains = dependencyGraph.consumeDeferredQueue();
+    for (final JobManager.Chain chain : chains) {
+      chain.enqueue();
+    }
+
     for (int i = 0; i < messageIds.size(); i++) {
       long                             messageId    = messageIds.get(i);
       OutgoingSecureMediaMessage       message      = messages.get(i);
       Recipient                        recipient    = message.getRecipient();
+      List<UploadDependencyGraph.Node> dependencies = dependencyGraph.getDependencyMap().get(message);
 
-      sendMediaMessage(context, recipient, false, messageId, Collections.emptyList());
+      List<String> jobDependencyIds = (dependencies != null) ? dependencies.stream().map(UploadDependencyGraph.Node::getJobId).collect(Collectors.toList())
+                                                             : Collections.emptyList();
+
+      sendMediaMessage(context,
+                       recipient,
+                       false,
+                       messageId,
+                       jobDependencyIds);
     }
 
     onMessageSent();

app/src/main/java/org/thoughtcrime/securesms/sms/UploadDependencyGraph.kt

@@ -0,0 +1,208 @@
+package org.thoughtcrime.securesms.sms
+
+import androidx.annotation.WorkerThread
+import org.thoughtcrime.securesms.attachments.Attachment
+import org.thoughtcrime.securesms.attachments.AttachmentId
+import org.thoughtcrime.securesms.attachments.DatabaseAttachment
+import org.thoughtcrime.securesms.attachments.UriAttachment
+import org.thoughtcrime.securesms.database.AttachmentDatabase
+import org.thoughtcrime.securesms.jobmanager.Job
+import org.thoughtcrime.securesms.jobmanager.JobManager
+import org.thoughtcrime.securesms.jobs.AttachmentCompressionJob
+import org.thoughtcrime.securesms.jobs.AttachmentCopyJob
+import org.thoughtcrime.securesms.jobs.AttachmentUploadJob
+import org.thoughtcrime.securesms.jobs.ResumableUploadSpecJob
+import org.thoughtcrime.securesms.mms.OutgoingMediaMessage
+
+/**
+ * Helper alias for working with JobIds.
+ */
+private typealias JobId = String
+
+/**
+ * Represents message send dependencies on attachments. Allows for the consumption of the job queue
+ * in a way in which repeated access will return an empty list.
+ *
+ * @param dependencyMap Maps an OutgoingMediaMessage to all of the Attachments it depends on.
+ * @param deferredJobQueue A list of job chains that can be executed on the job manager when ready (outside of a database transaction).
+ */
+class UploadDependencyGraph private constructor(
+  val dependencyMap: Map<OutgoingMediaMessage, List<Node>>,
+  private val deferredJobQueue: List<JobManager.Chain>
+) {
+
+  /**
+   * Contains the dependency job id as well as the attachment the job is working on.
+   */
+  data class Node(
+    val jobId: JobId,
+    val attachmentId: AttachmentId
+  )
+
+  /**
+   * A generic attachment key which is unique given the attachment AND it's transform properties.
+   */
+  private data class AttachmentKey<A : Attachment>(
+    val attachment: A,
+    private val transformProperties: AttachmentDatabase.TransformProperties = attachment.transformProperties
+  )
+
+  private var hasConsumedJobQueue = false
+
+  /**
+   * Returns the list of chains exactly once.
+   */
+  fun consumeDeferredQueue(): List<JobManager.Chain> {
+    if (hasConsumedJobQueue) {
+      return emptyList()
+    }
+
+    synchronized(this) {
+      if (hasConsumedJobQueue) {
+        return emptyList()
+      }
+
+      hasConsumedJobQueue = true
+      return deferredJobQueue
+    }
+  }
+
+  companion object {
+
+    @JvmField
+    val EMPTY = UploadDependencyGraph(emptyMap(), emptyList())
+
+    /**
+     * Allows representation of a unique database attachment by its internal id and its transform properties.
+     */
+    private fun DatabaseAttachment.asDatabaseAttachmentKey(): AttachmentKey<DatabaseAttachment> {
+      return AttachmentKey(this, this.transformProperties)
+    }
+
+    /**
+     * Allows representation of a unique URI attachment by its internal Uri and its transform properties.
+     */
+    private fun UriAttachment.asUriAttachmentKey(): AttachmentKey<UriAttachment> {
+      return AttachmentKey(this, transformProperties)
+    }
+
+    /**
+     * Given a list of outgoing media messages, give me a mapping of those messages to their dependent attachments and set of deferred
+     * job chains that can be executed to upload and copy the required jobs.
+     *
+     * This should be run within a database transaction, but does not enforce on itself. There is no direct access here to the database,
+     * instead that is isolated within the passed parameters.
+     *
+     * @param messages The list of outgoing messages
+     * @param jobManager The JobManager instance
+     * @param insertAttachmentForPreUpload A method which will create a new database row for a given attachment.
+     */
+    @JvmStatic
+    @WorkerThread
+    fun create(
+      messages: List<OutgoingMediaMessage>,
+      jobManager: JobManager,
+      insertAttachmentForPreUpload: (Attachment) -> DatabaseAttachment
+    ): UploadDependencyGraph {
+      return buildDependencyGraph(buildAttachmentMap(messages, insertAttachmentForPreUpload), jobManager, insertAttachmentForPreUpload)
+    }
+
+    /**
+     * Produce a mapping of AttachmentKey{DatabaseAttachment,TransformProperties} -> Set<OutgoingMediaMessage>
+     * This map represents which messages require a specific attachment.
+     */
+    private fun buildAttachmentMap(messages: List<OutgoingMediaMessage>, insertAttachmentForPreUpload: (Attachment) -> DatabaseAttachment): Map<AttachmentKey<DatabaseAttachment>, Set<OutgoingMediaMessage>> {
+      val attachmentMap = mutableMapOf<AttachmentKey<DatabaseAttachment>, Set<OutgoingMediaMessage>>()
+      val preUploadCache = mutableMapOf<AttachmentKey<UriAttachment>, DatabaseAttachment>()
+
+      for (message in messages) {
+        val attachmentList: List<Attachment> = message.attachments +
+          message.linkPreviews.mapNotNull { it.thumbnail.orElse(null) } +
+          message.sharedContacts.mapNotNull { it.avatar?.attachment }
+
+        val uniqueAttachments: Set<AttachmentKey<Attachment>> = attachmentList.map { AttachmentKey(it, it.transformProperties) }.toSet()
+
+        for (attachmentKey in uniqueAttachments) {
+          when (val attachment = attachmentKey.attachment) {
+            is DatabaseAttachment -> {
+              val messageIdList: Set<OutgoingMediaMessage> = attachmentMap.getOrDefault(attachment.asDatabaseAttachmentKey(), emptySet())
+              attachmentMap[attachment.asDatabaseAttachmentKey()] = messageIdList + message
+            }
+            is UriAttachment -> {
+              val dbAttachmentKey: AttachmentKey<DatabaseAttachment> = preUploadCache.getOrPut(attachment.asUriAttachmentKey()) { insertAttachmentForPreUpload(attachment) }.asDatabaseAttachmentKey()
+              val messageIdList: Set<OutgoingMediaMessage> = attachmentMap.getOrDefault(dbAttachmentKey, emptySet())
+              attachmentMap[dbAttachmentKey] = messageIdList + message
+            }
+            else -> {
+              error("Unsupported attachment subclass - ${attachment::class.java}")
+            }
+          }
+        }
+      }
+
+      return attachmentMap
+    }
+
+    /**
+     * Builds out the [UploadDependencyGraph] which collects dependency information for a given set of messages.
+     * Each attachment will be uploaded exactly once and copied N times, where N is the number of messages in its set, minus 1 (the upload)
+     * The resulting object contains a list of jobs that a subsequent send job can depend on, as well as a list of Chains which can be
+     * enqueued to perform uploading. Since a send job can depend on multiple chains, it's cleaner to give back a mapping of
+     * [OutgoingMediaMessage] -> [List<Node>] instead of forcing the caller to try to weave new jobs into the original chains.
+     *
+     * Each chain consists of:
+     *  1. Compression job
+     *  1. Resumable upload spec job
+     *  1. Attachment upload job
+     *  1. O to 1 copy jobs
+     */
+    private fun buildDependencyGraph(
+      attachmentIdToOutgoingMessagesMap: Map<AttachmentKey<DatabaseAttachment>, Set<OutgoingMediaMessage>>,
+      jobManager: JobManager,
+      insertAttachmentForPreUpload: (Attachment) -> DatabaseAttachment
+    ): UploadDependencyGraph {
+      val resultMap = mutableMapOf<OutgoingMediaMessage, List<Node>>()
+      val jobQueue = mutableListOf<JobManager.Chain>()
+
+      for ((attachmentKey, messages) in attachmentIdToOutgoingMessagesMap) {
+        val (uploadJobId, uploadChain) = createAttachmentUploadChain(jobManager, attachmentKey.attachment)
+        val uploadMessage: OutgoingMediaMessage = messages.first()
+        val copyMessages: List<OutgoingMediaMessage> = messages.drop(1)
+
+        val uploadMessageDependencies: List<Node> = resultMap.getOrDefault(uploadMessage, emptyList())
+        resultMap[uploadMessage] = uploadMessageDependencies + Node(uploadJobId, attachmentKey.attachment.attachmentId)
+
+        if (copyMessages.isNotEmpty()) {
+          val copyAttachments: Map<OutgoingMediaMessage, AttachmentId> = copyMessages.associateWith { insertAttachmentForPreUpload(attachmentKey.attachment).attachmentId }
+          val copyJob = AttachmentCopyJob(attachmentKey.attachment.attachmentId, copyAttachments.values.toList())
+
+          copyAttachments.forEach { (message, attachmentId) ->
+            val copyMessageDependencies: List<Node> = resultMap.getOrDefault(message, emptyList())
+            resultMap[message] = copyMessageDependencies + Node(copyJob.id, attachmentId)
+          }
+
+          uploadChain.then(copyJob)
+        }
+
+        jobQueue.add(uploadChain)
+      }
+
+      return UploadDependencyGraph(resultMap, jobQueue)
+    }
+
+    /**
+     * Creates the minimum necessary upload chain for the given attachment. This includes compression, grabbing the resumable upload spec,
+     * and the upload job itself.
+     */
+    private fun createAttachmentUploadChain(jobManager: JobManager, databaseAttachment: DatabaseAttachment): Pair<JobId, JobManager.Chain> {
+      val compressionJob: Job = AttachmentCompressionJob.fromAttachment(databaseAttachment, false, -1)
+      val resumableUploadSpecJob: Job = ResumableUploadSpecJob()
+      val uploadJob: Job = AttachmentUploadJob(databaseAttachment.attachmentId)
+
+      return uploadJob.id to jobManager
+        .startChain(compressionJob)
+        .then(resumableUploadSpecJob)
+        .then(uploadJob)
+    }
+  }
+}

app/src/test/java/org/thoughtcrime/securesms/sms/UploadDependencyGraphTest.kt

@@ -0,0 +1,240 @@
+package org.thoughtcrime.securesms.sms
+
+import android.app.Application
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.mockito.kotlin.any
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.whenever
+import org.robolectric.RobolectricTestRunner
+import org.robolectric.annotation.Config
+import org.thoughtcrime.securesms.attachments.Attachment
+import org.thoughtcrime.securesms.attachments.AttachmentId
+import org.thoughtcrime.securesms.attachments.DatabaseAttachment
+import org.thoughtcrime.securesms.database.AttachmentDatabase
+import org.thoughtcrime.securesms.jobmanager.Job
+import org.thoughtcrime.securesms.jobmanager.JobManager
+import org.thoughtcrime.securesms.jobs.AttachmentCompressionJob
+import org.thoughtcrime.securesms.jobs.AttachmentCopyJob
+import org.thoughtcrime.securesms.jobs.AttachmentUploadJob
+import org.thoughtcrime.securesms.jobs.ResumableUploadSpecJob
+import org.thoughtcrime.securesms.mms.SentMediaQuality
+import org.thoughtcrime.securesms.testutil.OutgoingMediaMessageBuilder
+import org.thoughtcrime.securesms.testutil.OutgoingMediaMessageBuilder.secure
+import org.thoughtcrime.securesms.testutil.UriAttachmentBuilder
+import org.thoughtcrime.securesms.util.JsonUtils
+import org.thoughtcrime.securesms.util.MediaUtil
+import java.util.concurrent.atomic.AtomicLong
+
+/**
+ * Requires Robolectric due to usage of Uri
+ */
+@RunWith(RobolectricTestRunner::class)
+@Config(application = Application::class)
+class UploadDependencyGraphTest {
+
+  private val jobManager: JobManager = mock()
+
+  private var uniqueLong = AtomicLong(0)
+
+  @Before
+  fun setUp() {
+    whenever(jobManager.startChain(any<Job>())).then {
+      JobManager.Chain(jobManager, listOf(it.getArgument(0)))
+    }
+  }
+
+  @Test
+  fun `Given a list of Uri attachments and a list of Messages, when I get the dependencyMap, then I expect a times m results`() {
+    // GIVEN
+    val uriAttachments = (1..5).map { UriAttachmentBuilder.build(id = uniqueLong.getAndIncrement(), contentType = MediaUtil.IMAGE_JPEG) }
+    val messages = (1..5).map { OutgoingMediaMessageBuilder.create(attachments = uriAttachments).secure() }
+    val testSubject = UploadDependencyGraph.create(messages, jobManager) { getAttachmentForPreUpload(uniqueLong.getAndIncrement(), it) }
+
+    // WHEN
+    val result = testSubject.dependencyMap
+
+    // THEN
+    assertEquals(5, result.size)
+    result.values.forEach { assertEquals(5, it.size) }
+  }
+
+  @Test
+  fun `Given a list of Uri attachments and a list of Messages, when I consumeDeferredQueue, then I expect one upload chain and one copy job for each attachment`() {
+    // GIVEN
+    val uriAttachments = (1..5).map { UriAttachmentBuilder.build(id = uniqueLong.getAndIncrement(), contentType = MediaUtil.IMAGE_JPEG) }
+    val messages = (1..5).map { OutgoingMediaMessageBuilder.create(attachments = uriAttachments).secure() }
+    val testSubject = UploadDependencyGraph.create(messages, jobManager) { getAttachmentForPreUpload(uniqueLong.getAndIncrement(), it) }
+
+    // WHEN
+    val deferredQueue = testSubject.consumeDeferredQueue()
+
+    // THEN
+    deferredQueue.forEach { assertValidJobChain(it, 4) }
+  }
+
+  @Test
+  fun `Given a list of Uri attachments with same id but different transforms and a list of Messages, when I consumeDeferredQueue, then I expect one upload chain for each attachment and 5 copy jobs`() {
+    // GIVEN
+    val uriAttachments = (1..5).map {
+      val increment = uniqueLong.getAndIncrement()
+      UriAttachmentBuilder.build(
+        id = 10,
+        contentType = MediaUtil.IMAGE_JPEG,
+        transformProperties = AttachmentDatabase.TransformProperties(false, true, increment, increment + 1, SentMediaQuality.STANDARD.code)
+      )
+    }
+
+    val messages = (1..5).map { OutgoingMediaMessageBuilder.create(attachments = uriAttachments).secure() }
+    val testSubject = UploadDependencyGraph.create(messages, jobManager) { getAttachmentForPreUpload(uniqueLong.getAndIncrement(), it) }
+
+    // WHEN
+    val deferredQueue = testSubject.consumeDeferredQueue()
+
+    // THEN
+    deferredQueue.forEach { assertValidJobChain(it, 4) }
+  }
+
+  @Test
+  fun `Given a single Uri attachment with same id and a list of Messages, when I consumeDeferredQueue, then I expect one upload chain and one copy job`() {
+    // GIVEN
+    val uriAttachments = (1..5).map {
+      UriAttachmentBuilder.build(
+        id = 10,
+        contentType = MediaUtil.IMAGE_JPEG
+      )
+    }
+
+    val messages = (1..5).map { OutgoingMediaMessageBuilder.create(attachments = uriAttachments).secure() }
+    val testSubject = UploadDependencyGraph.create(messages, jobManager) { getAttachmentForPreUpload(uniqueLong.getAndIncrement(), it) }
+
+    // WHEN
+    val deferredQueue = testSubject.consumeDeferredQueue()
+
+    // THEN
+    assertEquals(1, deferredQueue.size)
+    deferredQueue.forEach { assertValidJobChain(it, 4) }
+  }
+
+  @Test
+  fun `Given three Uri attachments with same id and two share transform properties and a list of Messages, when I executeDeferredQueue, then I expect two chains`() {
+    // GIVEN
+    val uriAttachments = (1..3).map {
+      UriAttachmentBuilder.build(
+        id = 10,
+        contentType = MediaUtil.IMAGE_JPEG,
+        transformProperties = if (it != 1) AttachmentDatabase.TransformProperties(false, true, 1, 2, SentMediaQuality.STANDARD.code) else null
+      )
+    }
+
+    val messages = (1..8).map { OutgoingMediaMessageBuilder.create(attachments = uriAttachments).secure() }
+    val testSubject = UploadDependencyGraph.create(messages, jobManager) { getAttachmentForPreUpload(uniqueLong.getAndIncrement(), it) }
+
+    // WHEN
+    val deferredQueue = testSubject.consumeDeferredQueue()
+
+    // THEN
+    assertEquals(2, deferredQueue.size)
+    deferredQueue.forEach { assertValidJobChain(it, 7) }
+  }
+
+  @Test
+  fun `Given a list of Database attachments and a list of Messages, when I get the dependency map, then I expect a times m results`() {
+    // GIVEN
+    val databaseAttachments = (1..5).map {
+      val id = uniqueLong.getAndIncrement()
+      val uriAttachment = UriAttachmentBuilder.build(id = uniqueLong.getAndIncrement(), contentType = MediaUtil.IMAGE_JPEG)
+      getAttachmentForPreUpload(id, uriAttachment)
+    }
+
+    val messages = (1..5).map { OutgoingMediaMessageBuilder.create(attachments = databaseAttachments).secure() }
+    val testSubject = UploadDependencyGraph.create(messages, jobManager) { getAttachmentForPreUpload(uniqueLong.getAndIncrement(), it) }
+
+    // WHEN
+    val result = testSubject.dependencyMap
+
+    // THEN
+    assertEquals(5, result.size)
+    result.values.forEach { assertEquals(5, it.size) }
+  }
+
+  @Test
+  fun `Given a list of messages with unique ids, when I consumeDeferredQueue, then I expect no copy jobs`() {
+    // GIVEN
+    val attachment1 = UriAttachmentBuilder.build(uniqueLong.getAndIncrement(), contentType = MediaUtil.IMAGE_JPEG)
+    val attachment2 = UriAttachmentBuilder.build(uniqueLong.getAndIncrement(), contentType = MediaUtil.IMAGE_JPEG)
+    val message1 = OutgoingMediaMessageBuilder.create(attachments = listOf(attachment1))
+    val message2 = OutgoingMediaMessageBuilder.create(attachments = listOf(attachment2))
+    val testSubject = UploadDependencyGraph.create(listOf(message1, message2), jobManager) { getAttachmentForPreUpload(uniqueLong.getAndIncrement(), it) }
+
+    // WHEN
+    val result = testSubject.consumeDeferredQueue()
+
+    // THEN
+    assertEquals(2, result.size)
+    result.forEach {
+      assertValidJobChain(it, 0)
+    }
+  }
+
+  private fun assertValidJobChain(chain: JobManager.Chain, expectedCopyDestinationCount: Int) {
+    val steps: List<List<Job>> = chain.jobListChain
+
+    assertTrue(steps.all { it.size == 1 })
+    assertTrue(steps[0][0] is AttachmentCompressionJob)
+    assertTrue(steps[1][0] is ResumableUploadSpecJob)
+    assertTrue(steps[2][0] is AttachmentUploadJob)
+
+    if (expectedCopyDestinationCount > 0) {
+      assertTrue(steps[3][0] is AttachmentCopyJob)
+
+      val uploadData = steps[2][0].serialize()
+      val copyData = steps[3][0].serialize()
+
+      val uploadAttachmentId = AttachmentId(uploadData.getLong("row_id"), uploadData.getLong("unique_id"))
+      val copySourceAttachmentId = JsonUtils.fromJson(copyData.getString("source_id"), AttachmentId::class.java)
+
+      assertEquals(uploadAttachmentId, copySourceAttachmentId)
+
+      val copyDestinations = copyData.getStringArray("destination_ids")
+      assertEquals(expectedCopyDestinationCount, copyDestinations.size)
+    } else {
+      assertEquals(3, steps.size)
+    }
+  }
+
+  private fun getAttachmentForPreUpload(id: Long, attachment: Attachment): DatabaseAttachment {
+    return DatabaseAttachment(
+      AttachmentId(id, id),
+      AttachmentDatabase.PREUPLOAD_MESSAGE_ID,
+      false,
+      false,
+      attachment.contentType,
+      AttachmentDatabase.TRANSFER_PROGRESS_PENDING,
+      attachment.size,
+      attachment.fileName,
+      attachment.cdnNumber,
+      attachment.location,
+      attachment.key,
+      attachment.relay,
+      attachment.digest,
+      attachment.fastPreflightId,
+      attachment.isVoiceNote,
+      attachment.isBorderless,
+      attachment.isVideoGif,
+      attachment.width,
+      attachment.height,
+      attachment.isQuote,
+      attachment.caption,
+      attachment.sticker,
+      attachment.blurHash,
+      attachment.audioHash,
+      attachment.transformProperties,
+      0,
+      attachment.uploadTimestamp
+    )
+  }
+}

app/src/test/java/org/thoughtcrime/securesms/testutil/OutgoingMediaMessageBuilder.kt

@@ -0,0 +1,62 @@
+package org.thoughtcrime.securesms.testutil
+
+import org.thoughtcrime.securesms.attachments.Attachment
+import org.thoughtcrime.securesms.contactshare.Contact
+import org.thoughtcrime.securesms.database.ThreadDatabase
+import org.thoughtcrime.securesms.database.documents.IdentityKeyMismatch
+import org.thoughtcrime.securesms.database.documents.NetworkFailure
+import org.thoughtcrime.securesms.database.model.Mention
+import org.thoughtcrime.securesms.database.model.ParentStoryId
+import org.thoughtcrime.securesms.database.model.StoryType
+import org.thoughtcrime.securesms.database.model.databaseprotos.GiftBadge
+import org.thoughtcrime.securesms.linkpreview.LinkPreview
+import org.thoughtcrime.securesms.mms.OutgoingMediaMessage
+import org.thoughtcrime.securesms.mms.OutgoingSecureMediaMessage
+import org.thoughtcrime.securesms.mms.QuoteModel
+import org.thoughtcrime.securesms.recipients.Recipient
+
+object OutgoingMediaMessageBuilder {
+  fun create(
+    recipient: Recipient = Recipient.UNKNOWN,
+    message: String = "",
+    attachments: List<Attachment> = emptyList(),
+    sentTimeMillis: Long = System.currentTimeMillis(),
+    subscriptionId: Int = -1,
+    expiresIn: Long = -1,
+    viewOnce: Boolean = false,
+    distributionType: Int = ThreadDatabase.DistributionTypes.DEFAULT,
+    storyType: StoryType = StoryType.NONE,
+    parentStoryId: ParentStoryId? = null,
+    isStoryReaction: Boolean = false,
+    quoteModel: QuoteModel? = null,
+    contacts: List<Contact> = emptyList(),
+    linkPreviews: List<LinkPreview> = emptyList(),
+    mentions: List<Mention> = emptyList(),
+    networkFailures: Set<NetworkFailure> = emptySet(),
+    identityKeyMismatches: Set<IdentityKeyMismatch> = emptySet(),
+    giftBadge: GiftBadge? = null
+  ): OutgoingMediaMessage {
+    return OutgoingMediaMessage(
+      recipient,
+      message,
+      attachments,
+      sentTimeMillis,
+      subscriptionId,
+      expiresIn,
+      viewOnce,
+      distributionType,
+      storyType,
+      parentStoryId,
+      isStoryReaction,
+      quoteModel,
+      contacts,
+      linkPreviews,
+      mentions,
+      networkFailures,
+      identityKeyMismatches,
+      giftBadge
+    )
+  }
+
+  fun OutgoingMediaMessage.secure(): OutgoingSecureMediaMessage = OutgoingSecureMediaMessage(this)
+}

app/src/test/java/org/thoughtcrime/securesms/testutil/UriAttachmentBuilder.kt

@@ -0,0 +1,45 @@
+package org.thoughtcrime.securesms.testutil
+
+import android.net.Uri
+import org.thoughtcrime.securesms.attachments.UriAttachment
+import org.thoughtcrime.securesms.audio.AudioHash
+import org.thoughtcrime.securesms.blurhash.BlurHash
+import org.thoughtcrime.securesms.database.AttachmentDatabase
+import org.thoughtcrime.securesms.stickers.StickerLocator
+
+object UriAttachmentBuilder {
+  fun build(
+    id: Long,
+    uri: Uri = Uri.parse("content://$id"),
+    contentType: String,
+    transferState: Int = AttachmentDatabase.TRANSFER_PROGRESS_PENDING,
+    size: Long = 0L,
+    fileName: String = "file$id",
+    voiceNote: Boolean = false,
+    borderless: Boolean = false,
+    videoGif: Boolean = false,
+    quote: Boolean = false,
+    caption: String? = null,
+    stickerLocator: StickerLocator? = null,
+    blurHash: BlurHash? = null,
+    audioHash: AudioHash? = null,
+    transformProperties: AttachmentDatabase.TransformProperties? = null
+  ): UriAttachment {
+    return UriAttachment(
+      uri,
+      contentType,
+      transferState,
+      size,
+      fileName,
+      voiceNote,
+      borderless,
+      videoGif,
+      quote,
+      caption,
+      stickerLocator,
+      blurHash,
+      audioHash,
+      transformProperties
+    )
+  }
+}