Add kotlin/proto level message processing.

2026-07-13 09:17:49 +01:00 · 2023-03-30 11:45:13 -04:00
parent 28f27915c5
commit 2e45bd719a
43 changed files with 4505 additions and 84 deletions
@@ -0,0 +1,382 @@
+package org.thoughtcrime.securesms.messages
+
+import android.database.Cursor
+import android.util.Base64
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import org.junit.Before
+import org.junit.Rule
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.signal.core.util.ThreadUtil
+import org.signal.core.util.readToList
+import org.signal.core.util.select
+import org.signal.core.util.toSingleLine
+import org.signal.core.util.withinTransaction
+import org.thoughtcrime.securesms.database.AttachmentTable
+import org.thoughtcrime.securesms.database.MessageTable
+import org.thoughtcrime.securesms.database.MessageTypes
+import org.thoughtcrime.securesms.database.MessageTypes.isOutgoingMessageType
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.database.ThreadTable
+import org.thoughtcrime.securesms.keyvalue.SignalStore
+import org.thoughtcrime.securesms.testing.Entry
+import org.thoughtcrime.securesms.testing.InMemoryLogger
+import org.thoughtcrime.securesms.testing.MessageContentFuzzer
+import org.thoughtcrime.securesms.testing.SignalActivityRule
+import org.thoughtcrime.securesms.testing.assertIs
+import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
+import org.whispersystems.signalservice.api.messages.SignalServiceContent
+import org.whispersystems.signalservice.api.messages.SignalServiceMetadata
+import org.whispersystems.signalservice.api.push.SignalServiceAddress
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos
+import org.whispersystems.signalservice.internal.serialize.SignalServiceAddressProtobufSerializer
+import org.whispersystems.signalservice.internal.serialize.SignalServiceMetadataProtobufSerializer
+import org.whispersystems.signalservice.internal.serialize.protos.SignalServiceContentProto
+import java.util.Optional
+
+@RunWith(AndroidJUnit4::class)
+class MessageContentProcessorTestV2 {
+
+  companion object {
+    private val TAGS = listOf(MessageContentProcessor.TAG, MessageContentProcessorV2.TAG, AttachmentTable.TAG)
+
+    private val GENERALIZE_TAG = mapOf(
+      MessageContentProcessor.TAG to "MCP",
+      MessageContentProcessorV2.TAG to "MCP",
+      AttachmentTable.TAG to AttachmentTable.TAG
+    )
+
+    private val IGNORE_MESSAGE_COLUMNS = listOf(
+      MessageTable.DATE_RECEIVED,
+      MessageTable.NOTIFIED_TIMESTAMP,
+      MessageTable.REACTIONS_LAST_SEEN,
+      MessageTable.NOTIFIED
+    )
+
+    private val IGNORE_ATTACHMENT_COLUMNS = listOf(
+      AttachmentTable.UNIQUE_ID,
+      AttachmentTable.TRANSFER_FILE
+    )
+  }
+
+  @get:Rule
+  val harness = SignalActivityRule()
+
+  private lateinit var processorV1: MessageContentProcessor
+  private lateinit var processorV2: MessageContentProcessorV2
+  private lateinit var testResult: TestResults
+  private var envelopeTimestamp: Long = 0
+
+  @Before
+  fun setup() {
+    processorV1 = MessageContentProcessor(harness.context)
+    processorV2 = MessageContentProcessorV2(harness.context)
+    envelopeTimestamp = System.currentTimeMillis()
+    testResult = TestResults()
+  }
+
+  @Test
+  fun textMessage() {
+    var start = envelopeTimestamp
+
+    val messages: List<TestMessage> = (0 until 1).map {
+      start += 200
+      TestMessage(
+        envelope = MessageContentFuzzer.envelope(start),
+        content = MessageContentFuzzer.fuzzTextMessage(),
+        metadata = MessageContentFuzzer.fuzzMetadata(harness.others[0], harness.self.id),
+        serverDeliveredTimestamp = MessageContentFuzzer.fuzzServerDeliveredTimestamp(start)
+      )
+    }
+
+    testResult.runV2(messages)
+    testResult.runV1(messages)
+
+    testResult.assert()
+  }
+
+  @Test
+  fun mediaMessage() {
+    var start = envelopeTimestamp
+
+    val messages: List<TestMessage> = (0 until 10).map {
+      start += 200
+      TestMessage(
+        envelope = MessageContentFuzzer.envelope(start),
+        content = MessageContentFuzzer.fuzzTextMessage(),
+        metadata = MessageContentFuzzer.fuzzMetadata(harness.others[0], harness.self.id),
+        serverDeliveredTimestamp = MessageContentFuzzer.fuzzServerDeliveredTimestamp(start)
+      )
+    }
+
+    val moreMessages: List<TestMessage> = (0 until 10).map {
+      start += 200
+      TestMessage(
+        envelope = MessageContentFuzzer.envelope(start),
+        content = MessageContentFuzzer.fuzzMediaMessageWithBody(messages),
+        metadata = MessageContentFuzzer.fuzzMetadata(harness.others[0], harness.self.id),
+        serverDeliveredTimestamp = MessageContentFuzzer.fuzzServerDeliveredTimestamp(start)
+      )
+    }
+
+    val evenMoreMessages: List<TestMessage> = (0 until 10).map {
+      start += 200
+      TestMessage(
+        envelope = MessageContentFuzzer.envelope(start),
+        content = MessageContentFuzzer.fuzzMediaMessageNoContent(messages + moreMessages),
+        metadata = MessageContentFuzzer.fuzzMetadata(harness.others[0], harness.self.id),
+        serverDeliveredTimestamp = MessageContentFuzzer.fuzzServerDeliveredTimestamp(start)
+      )
+    }
+
+    val evenMoreMoreMessages: List<TestMessage> = (0 until 10).map {
+      start += 200
+      TestMessage(
+        envelope = MessageContentFuzzer.envelope(start),
+        content = MessageContentFuzzer.fuzzMediaMessageNoText(messages + moreMessages),
+        metadata = MessageContentFuzzer.fuzzMetadata(harness.others[0], harness.self.id),
+        serverDeliveredTimestamp = MessageContentFuzzer.fuzzServerDeliveredTimestamp(start)
+      )
+    }
+
+    testResult.runV2(messages + moreMessages + evenMoreMessages + evenMoreMoreMessages)
+    testResult.runV1(messages + moreMessages + evenMoreMessages + evenMoreMoreMessages)
+
+    testResult.assert()
+  }
+
+//  @Test
+//  fun fuzzIt() {
+//    MessageContentFuzzer.fuzzProto(SignalServiceProtos.DataMessage.Contact.Name::class)
+//    MessageContentFuzzer.fuzzProto(SignalServiceProtos.DataMessage.Contact.Avatar::class)
+//    MessageContentFuzzer.fuzzProto(SignalServiceProtos.DataMessage.Contact.Email::class)
+//  }
+
+  private inner class TestResults {
+
+    private lateinit var v1Logs: List<Entry>
+    private lateinit var v1Messages: List<List<Pair<String, String?>>>
+    private lateinit var v1Attachments: List<List<Pair<String, String?>>>
+    private lateinit var v2Logs: List<Entry>
+    private lateinit var v2Messages: List<List<Pair<String, String?>>>
+    private lateinit var v2Attachments: List<List<Pair<String, String?>>>
+
+    fun runV1(messages: List<TestMessage>) {
+      messages.forEach { (envelope, content, metadata, serverDeliveredTimestamp) ->
+        if (content.hasDataMessage()) {
+          processorV1.process(
+            MessageContentProcessor.MessageState.DECRYPTED_OK,
+            toSignalServiceContent(envelope, content, metadata, serverDeliveredTimestamp),
+            null,
+            envelope.timestamp,
+            -1
+          )
+          ThreadUtil.sleep(1)
+        }
+      }
+
+      v1Logs = harness.inMemoryLogger.logs()
+      harness.inMemoryLogger.clear()
+
+      v1Messages = dumpMessages()
+      v1Attachments = dumpAttachments()
+    }
+
+    fun runV2(messages: List<TestMessage>) {
+      messages.forEach { (envelope, content, metadata, serverDeliveredTimestamp) ->
+        if (content.hasDataMessage()) {
+          processorV2.process(
+            envelope,
+            content,
+            metadata,
+            serverDeliveredTimestamp,
+            false
+          )
+          ThreadUtil.sleep(1)
+        }
+      }
+
+      v2Logs = harness.inMemoryLogger.logs()
+      harness.inMemoryLogger.clear()
+
+      v2Messages = dumpMessages()
+      v2Attachments = dumpAttachments()
+
+      cleanup()
+    }
+
+    fun cleanup() {
+      SignalDatabase.rawDatabase.withinTransaction { db ->
+        SignalDatabase.threads.deleteAllConversations()
+        db.execSQL("DELETE FROM sqlite_sequence WHERE name = '${MessageTable.TABLE_NAME}'")
+        db.execSQL("DELETE FROM sqlite_sequence WHERE name = '${ThreadTable.TABLE_NAME}'")
+        db.execSQL("DELETE FROM sqlite_sequence WHERE name = '${AttachmentTable.TABLE_NAME}'")
+      }
+    }
+
+    fun assert() {
+      v2Logs.zip(v1Logs)
+        .forEach { (v2, v1) ->
+          GENERALIZE_TAG[v2.tag]!!.assertIs(GENERALIZE_TAG[v1.tag]!!)
+
+          if (v2.tag != AttachmentTable.TAG) {
+            if (v2.message?.startsWith("[") == true && v1.message?.startsWith("[") == false) {
+              v2.message!!.substring(v2.message!!.indexOf(']') + 2).assertIs(v1.message)
+            } else {
+              v2.message.assertIs(v1.message)
+            }
+          } else {
+            if (v2.message?.startsWith("Inserted attachment at ID: AttachmentId::") == true) {
+              v2.message!!
+                .substring(0, v2.message!!.indexOf(','))
+                .assertIs(
+                  v1.message!!
+                    .substring(0, v1.message!!.indexOf(','))
+                )
+            } else {
+              v2.message.assertIs(v1.message)
+            }
+          }
+          v2.throwable.assertIs(v1.throwable)
+        }
+
+      v2Messages.zip(v1Messages)
+        .forEach { (v2, v1) ->
+          v2.assertIs(v1)
+        }
+
+      v2Attachments.zip(v1Attachments)
+        .forEach { (v2, v1) ->
+          v2.assertIs(v1)
+        }
+    }
+
+    private fun InMemoryLogger.logs(): List<Entry> {
+      return entries()
+        .filter { TAGS.contains(it.tag) }
+    }
+
+    private fun dumpMessages(): List<List<Pair<String, String?>>> {
+      return dumpTable(MessageTable.TABLE_NAME)
+        .map { row ->
+          val newRow = row.toMutableList()
+          newRow.removeIf { IGNORE_MESSAGE_COLUMNS.contains(it.first) }
+          newRow
+        }
+    }
+
+    private fun dumpAttachments(): List<List<Pair<String, String?>>> {
+      return dumpTable(AttachmentTable.TABLE_NAME)
+        .map { row ->
+          val newRow = row.toMutableList()
+          newRow.removeIf { IGNORE_ATTACHMENT_COLUMNS.contains(it.first) }
+          newRow
+        }
+    }
+
+    private fun dumpTable(table: String): List<List<Pair<String, String?>>> {
+      return SignalDatabase.rawDatabase
+        .select()
+        .from(table)
+        .run()
+        .readToList { cursor ->
+          val map: List<Pair<String, String?>> = cursor.columnNames.map { column ->
+            val index = cursor.getColumnIndex(column)
+            var data: String? = when (cursor.getType(index)) {
+              Cursor.FIELD_TYPE_BLOB -> Base64.encodeToString(cursor.getBlob(index), 0)
+              else -> cursor.getString(index)
+            }
+            if (table == MessageTable.TABLE_NAME && column == "type") {
+              data = thing(cursor.getLong(index))
+            }
+
+            column to data
+          }
+          map
+        }
+    }
+  }
+
+  private fun toSignalServiceContent(envelope: SignalServiceProtos.Envelope, content: SignalServiceProtos.Content, metadata: EnvelopeMetadata, serverDeliveredTimestamp: Long): SignalServiceContent {
+    val localAddress = SignalServiceAddress(metadata.destinationServiceId, Optional.ofNullable(SignalStore.account().e164))
+    val signalServiceMetadata = SignalServiceMetadata(
+      SignalServiceAddress(metadata.sourceServiceId, Optional.ofNullable(metadata.sourceE164)),
+      metadata.sourceDeviceId,
+      envelope.timestamp,
+      envelope.serverTimestamp,
+      serverDeliveredTimestamp,
+      metadata.sealedSender,
+      envelope.serverGuid,
+      Optional.ofNullable(metadata.groupId),
+      metadata.destinationServiceId.toString()
+    )
+
+    val contentProto = SignalServiceContentProto.newBuilder()
+      .setLocalAddress(SignalServiceAddressProtobufSerializer.toProtobuf(localAddress))
+      .setMetadata(SignalServiceMetadataProtobufSerializer.toProtobuf(signalServiceMetadata))
+      .setContent(content)
+      .build()
+
+    return SignalServiceContent.createFromProto(contentProto)!!
+  }
+
+  fun thing(type: Long): String {
+    return """
+      isOutgoingMessageType:${isOutgoingMessageType(type)}
+      isForcedSms:${type and MessageTypes.MESSAGE_FORCE_SMS_BIT != 0L}
+      isDraftMessageType:${type and MessageTypes.BASE_TYPE_MASK == MessageTypes.BASE_DRAFT_TYPE}
+      isFailedMessageType:${type and MessageTypes.BASE_TYPE_MASK == MessageTypes.BASE_SENT_FAILED_TYPE}
+      isPendingMessageType:${type and MessageTypes.BASE_TYPE_MASK == MessageTypes.BASE_OUTBOX_TYPE || type and MessageTypes.BASE_TYPE_MASK == MessageTypes.BASE_SENDING_TYPE}
+      isSentType:${type and MessageTypes.BASE_TYPE_MASK == MessageTypes.BASE_SENT_TYPE}
+      isPendingSmsFallbackType:${type and MessageTypes.BASE_TYPE_MASK == MessageTypes.BASE_PENDING_INSECURE_SMS_FALLBACK || type and MessageTypes.BASE_TYPE_MASK == MessageTypes.BASE_PENDING_SECURE_SMS_FALLBACK}
+      isPendingSecureSmsFallbackType:${type and MessageTypes.BASE_TYPE_MASK == MessageTypes.BASE_PENDING_SECURE_SMS_FALLBACK}
+      isPendingInsecureSmsFallbackType:${type and MessageTypes.BASE_TYPE_MASK == MessageTypes.BASE_PENDING_INSECURE_SMS_FALLBACK}
+      isInboxType:${type and MessageTypes.BASE_TYPE_MASK == MessageTypes.BASE_INBOX_TYPE}
+      isJoinedType:${type and MessageTypes.BASE_TYPE_MASK == MessageTypes.JOINED_TYPE}
+      isUnsupportedMessageType:${type and MessageTypes.BASE_TYPE_MASK == MessageTypes.UNSUPPORTED_MESSAGE_TYPE}
+      isInvalidMessageType:${type and MessageTypes.BASE_TYPE_MASK == MessageTypes.INVALID_MESSAGE_TYPE}
+      isBadDecryptType:${type and MessageTypes.BASE_TYPE_MASK == MessageTypes.BAD_DECRYPT_TYPE}
+      isSecureType:${type and MessageTypes.SECURE_MESSAGE_BIT != 0L}
+      isPushType:${type and MessageTypes.PUSH_MESSAGE_BIT != 0L}
+      isEndSessionType:${type and MessageTypes.END_SESSION_BIT != 0L}
+      isKeyExchangeType:${type and MessageTypes.KEY_EXCHANGE_BIT != 0L}
+      isIdentityVerified:${type and MessageTypes.KEY_EXCHANGE_IDENTITY_VERIFIED_BIT != 0L}
+      isIdentityDefault:${type and MessageTypes.KEY_EXCHANGE_IDENTITY_DEFAULT_BIT != 0L}
+      isCorruptedKeyExchange:${type and MessageTypes.KEY_EXCHANGE_CORRUPTED_BIT != 0L}
+      isInvalidVersionKeyExchange:${type and MessageTypes.KEY_EXCHANGE_INVALID_VERSION_BIT != 0L}
+      isBundleKeyExchange:${type and MessageTypes.KEY_EXCHANGE_BUNDLE_BIT != 0L}
+      isContentBundleKeyExchange:${type and MessageTypes.KEY_EXCHANGE_CONTENT_FORMAT != 0L}
+      isIdentityUpdate:${type and MessageTypes.KEY_EXCHANGE_IDENTITY_UPDATE_BIT != 0L}
+      isRateLimited:${type and MessageTypes.MESSAGE_RATE_LIMITED_BIT != 0L}
+      isExpirationTimerUpdate:${type and MessageTypes.EXPIRATION_TIMER_UPDATE_BIT != 0L}
+      isIncomingAudioCall:${type == MessageTypes.INCOMING_AUDIO_CALL_TYPE}
+      isIncomingVideoCall:${type == MessageTypes.INCOMING_VIDEO_CALL_TYPE}
+      isOutgoingAudioCall:${type == MessageTypes.OUTGOING_AUDIO_CALL_TYPE}
+      isOutgoingVideoCall:${type == MessageTypes.OUTGOING_VIDEO_CALL_TYPE}
+      isMissedAudioCall:${type == MessageTypes.MISSED_AUDIO_CALL_TYPE}
+      isMissedVideoCall:${type == MessageTypes.MISSED_VIDEO_CALL_TYPE}
+      isGroupCall:${type == MessageTypes.GROUP_CALL_TYPE}
+      isGroupUpdate:${type and MessageTypes.GROUP_UPDATE_BIT != 0L}
+      isGroupV2:${type and MessageTypes.GROUP_V2_BIT != 0L}
+      isGroupQuit:${type and MessageTypes.GROUP_LEAVE_BIT != 0L && type and MessageTypes.GROUP_V2_BIT == 0L}
+      isChatSessionRefresh:${type and MessageTypes.ENCRYPTION_REMOTE_FAILED_BIT != 0L}
+      isDuplicateMessageType:${type and MessageTypes.ENCRYPTION_REMOTE_DUPLICATE_BIT != 0L}
+      isDecryptInProgressType:${type and 0x40000000 != 0L}
+      isNoRemoteSessionType:${type and MessageTypes.ENCRYPTION_REMOTE_NO_SESSION_BIT != 0L}
+      isLegacyType:${type and MessageTypes.ENCRYPTION_REMOTE_LEGACY_BIT != 0L || type and MessageTypes.ENCRYPTION_REMOTE_BIT != 0L}
+      isProfileChange:${type == MessageTypes.PROFILE_CHANGE_TYPE}
+      isGroupV1MigrationEvent:${type == MessageTypes.GV1_MIGRATION_TYPE}
+      isChangeNumber:${type == MessageTypes.CHANGE_NUMBER_TYPE}
+      isBoostRequest:${type == MessageTypes.BOOST_REQUEST_TYPE}
+      isThreadMerge:${type == MessageTypes.THREAD_MERGE_TYPE}
+      isSmsExport:${type == MessageTypes.SMS_EXPORT_TYPE}
+      isGroupV2LeaveOnly:${type and MessageTypes.GROUP_V2_LEAVE_BITS == MessageTypes.GROUP_V2_LEAVE_BITS}
+      isSpecialType:${type and MessageTypes.SPECIAL_TYPES_MASK != 0L}
+      isStoryReaction:${type and MessageTypes.SPECIAL_TYPES_MASK == MessageTypes.SPECIAL_TYPE_STORY_REACTION}
+      isGiftBadge:${type and MessageTypes.SPECIAL_TYPES_MASK == MessageTypes.SPECIAL_TYPE_GIFT_BADGE}
+      isPaymentsNotificaiton:${type and MessageTypes.SPECIAL_TYPES_MASK == MessageTypes.SPECIAL_TYPE_PAYMENTS_NOTIFICATION}
+      isRequestToActivatePayments:${type and MessageTypes.SPECIAL_TYPES_MASK == MessageTypes.SPECIAL_TYPE_PAYMENTS_ACTIVATE_REQUEST}
+      isPaymentsActivated:${type and MessageTypes.SPECIAL_TYPES_MASK == MessageTypes.SPECIAL_TYPE_PAYMENTS_ACTIVATED}
+    """.trimIndent().replace(Regex("is[A-Z][A-Za-z0-9]*:false\n?"), "").toSingleLine()
+  }
+}
@@ -0,0 +1,11 @@
+package org.thoughtcrime.securesms.messages
+
+import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos
+
+data class TestMessage(
+  val envelope: SignalServiceProtos.Envelope,
+  val content: SignalServiceProtos.Content,
+  val metadata: EnvelopeMetadata,
+  val serverDeliveredTimestamp: Long
+)
@@ -30,6 +30,16 @@ class InMemoryLogger : Log.Logger() {
    latch.await()
  }

+  fun clear() {
+    val latch = CountDownLatch(1)
+    executor.execute {
+      predicates.clear()
+      logEntries.clear()
+      latch.countDown()
+    }
+    latch.await()
+  }
+
  private fun add(entry: Entry) {
    executor.execute {
      logEntries += entry
@@ -0,0 +1,220 @@
+package org.thoughtcrime.securesms.testing
+
+import com.google.protobuf.ByteString
+import org.signal.core.util.logging.Log
+import org.thoughtcrime.securesms.database.model.toProtoByteString
+import org.thoughtcrime.securesms.messages.TestMessage
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.recipients.RecipientId
+import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
+import org.whispersystems.signalservice.api.util.UuidUtil
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.AttachmentPointer
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.Content
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.DataMessage
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.Envelope
+import java.util.UUID
+import kotlin.random.Random
+import kotlin.random.nextInt
+import kotlin.reflect.KClass
+import kotlin.reflect.KFunction
+import kotlin.reflect.full.declaredFunctions
+import kotlin.reflect.full.functions
+import kotlin.reflect.jvm.jvmErasure
+import kotlin.time.Duration.Companion.days
+
+object MessageContentFuzzer {
+
+  private val mediaTypes = listOf("image/png", "image/jpeg", "image/heic", "image/heif", "image/avif", "image/webp", "image/gif", "audio/aac", "audio/*", "video/mp4", "video/*", "text/x-vcard", "text/x-signal-plain", "application/x-signal-view-once", "*/*", "application/octet-stream")
+  private val emojis = listOf("😂", "❤️", "🔥", "😍", "👀", "🤔", "🙏", "👍", "🤷", "🥺")
+
+  private val random = Random(1)
+
+  fun envelope(timestamp: Long): Envelope {
+    return Envelope.newBuilder()
+      .setTimestamp(timestamp)
+      .setServerTimestamp(timestamp + 5)
+      .setServerGuidBytes(UuidUtil.toByteString(UUID.randomUUID()))
+      .build()
+  }
+
+  fun fuzzTextMessage(): Content {
+    return Content.newBuilder()
+      .setDataMessage(
+        DataMessage.newBuilder().run {
+          body = string()
+          if (random.nextBoolean()) {
+            expireTimer = random.nextInt(0..28.days.inWholeSeconds.toInt())
+          }
+          if (random.nextBoolean()) {
+            addBodyRanges(
+              SignalServiceProtos.BodyRange.newBuilder().run {
+                start = 0
+                length = 1
+                style = SignalServiceProtos.BodyRange.Style.BOLD
+                build()
+              }
+            )
+          }
+          build()
+        }
+      )
+      .build()
+  }
+
+  fun fuzzMediaMessageWithBody(quoteAble: List<TestMessage> = emptyList()): Content {
+    return Content.newBuilder()
+      .setDataMessage(
+        DataMessage.newBuilder().run {
+          if (random.nextBoolean()) {
+            body = string()
+          }
+
+          if (random.nextBoolean() && quoteAble.isNotEmpty()) {
+            body = string()
+            val quoted = quoteAble.random(random)
+            quote = DataMessage.Quote.newBuilder().run {
+              id = quoted.envelope.timestamp
+              authorUuid = quoted.metadata.sourceServiceId.toString()
+              text = quoted.content.dataMessage.body
+              addAllAttachments(quoted.content.dataMessage.attachmentsList)
+              addAllBodyRanges(quoted.content.dataMessage.bodyRangesList)
+              type = DataMessage.Quote.Type.NORMAL
+              build()
+            }
+          }
+
+          if (random.nextFloat() < 0.1 && quoteAble.isNotEmpty()) {
+            val quoted = quoteAble.random(random)
+            quote = DataMessage.Quote.newBuilder().run {
+              id = random.nextLong(quoted.envelope.timestamp - 1000000, quoted.envelope.timestamp)
+              authorUuid = quoted.metadata.sourceServiceId.toString()
+              text = quoted.content.dataMessage.body
+              build()
+            }
+          }
+
+          if (random.nextFloat() < 0.25) {
+            val total = random.nextInt(1, 2)
+            (0..total).forEach { _ -> addAttachments(attachmentPointer()) }
+          }
+
+          build()
+        }
+      )
+      .build()
+  }
+
+  fun fuzzMediaMessageNoContent(previousMessages: List<TestMessage> = emptyList()): Content {
+    return Content.newBuilder()
+      .setDataMessage(
+        DataMessage.newBuilder().run {
+          if (random.nextFloat() < 0.25) {
+            val reactTo = previousMessages.random(random)
+            reaction = DataMessage.Reaction.newBuilder().run {
+              emoji = emojis.random(random)
+              remove = false
+              targetAuthorUuid = reactTo.metadata.sourceServiceId.toString()
+              targetSentTimestamp = reactTo.envelope.timestamp
+              build()
+            }
+          }
+
+          build()
+        }
+      ).build()
+  }
+
+  fun fuzzMediaMessageNoText(previousMessages: List<TestMessage> = emptyList()): Content {
+    return Content.newBuilder()
+      .setDataMessage(
+        DataMessage.newBuilder().run {
+          if (random.nextFloat() < 0.9) {
+            sticker = DataMessage.Sticker.newBuilder().run {
+              packId = byteString(length = 24)
+              packKey = byteString(length = 128)
+              stickerId = random.nextInt()
+              data = attachmentPointer()
+              emoji = emojis.random(random)
+              build()
+            }
+          }
+          build()
+        }
+      ).build()
+  }
+
+  fun string(length: Int = 10, allowNullString: Boolean = false): String {
+    var string = ""
+
+    if (allowNullString && random.nextBoolean()) {
+      return string
+    }
+
+    for (i in 0 until length) {
+      string += random.nextInt(65..90).toChar()
+    }
+    return string
+  }
+
+  fun byteString(length: Int = 512): ByteString {
+    return random.nextBytes(512).toProtoByteString()
+  }
+
+  fun attachmentPointer(): SignalServiceProtos.AttachmentPointer {
+    return SignalServiceProtos.AttachmentPointer.newBuilder().run {
+      cdnKey = string()
+      contentType = mediaTypes.random(random)
+      key = byteString()
+      size = random.nextInt(1024 * 1024 * 50)
+      thumbnail = byteString()
+      digest = byteString()
+      fileName = string()
+      flags = 0
+      width = random.nextInt(until = 1024)
+      height = random.nextInt(until = 1024)
+      caption = string(allowNullString = true)
+      blurHash = string()
+      uploadTimestamp = random.nextLong()
+      cdnNumber = 1
+
+      build()
+    }
+  }
+
+  fun fuzzServerDeliveredTimestamp(envelopeTimestamp: Long): Long {
+    return envelopeTimestamp + 10
+  }
+
+  fun fuzzMetadata(source: RecipientId, destination: RecipientId): EnvelopeMetadata {
+    return EnvelopeMetadata(
+      sourceServiceId = Recipient.resolved(source).requireServiceId(),
+      sourceE164 = null,
+      sourceDeviceId = 1,
+      sealedSender = true,
+      groupId = null,
+      destinationServiceId = Recipient.resolved(destination).requireServiceId()
+    )
+  }
+
+  fun <T : Any> fuzzProto(protoClazz: KClass<T>) {
+    val newBuilder: Any = protoClazz.declaredFunctions.first { it.name == "newBuilder" }.call()!!
+
+    val setters: List<KFunction<*>> = newBuilder::class.functions.filter { it.name.startsWith("set") && !it.name.contains("Bytes") }
+
+    for (setter in setters) {
+      val type = setter.parameters[1].type.jvmErasure
+      when {
+        type == String::class -> setter.call(newBuilder, string())
+        type == Int::class -> setter.call(newBuilder, random.nextInt())
+        type == Long::class -> setter.call(newBuilder, random.nextLong())
+        type == AttachmentPointer::class -> setter.call(newBuilder, attachmentPointer())
+        type == Boolean::class -> setter.call(newBuilder, random.nextBoolean())
+//        type.superclasses.contains(EnumLite::class) ->
+        else -> Log.e("CODY", "WHAT!?!?!?! ${setter.parameters[1].type.jvmErasure}")
+      }
+    }
+
+    Log.e("CODY", newBuilder::class.functions.first { it.name == "build" }.call(newBuilder).toString())
+  }
+}
@@ -102,6 +102,8 @@ class SignalActivityRule(private val othersCount: Int = 4) : ExternalResource()
    RegistrationUtil.maybeMarkRegistrationComplete()
    SignalDatabase.recipients.setProfileName(Recipient.self().id, ProfileName.fromParts("Tester", "McTesterson"))

+    SignalStore.settings().isMessageNotificationsEnabled = false
+
    return Recipient.self()
  }

@@ -36,7 +36,7 @@ fun <T : Any?> T.assertIsNotNull() {
  assertThat(this, notNullValue())
 }

-infix fun <T : Any> T.assertIs(expected: T) {
+infix fun <T : Any?> T.assertIs(expected: T) {
  assertThat(this, `is`(expected))
 }

@@ -9,8 +9,11 @@ import org.thoughtcrime.securesms.blurhash.BlurHash;
 import org.thoughtcrime.securesms.database.AttachmentTable;
 import org.thoughtcrime.securesms.stickers.StickerLocator;
 import org.thoughtcrime.securesms.util.Base64;
+import org.whispersystems.signalservice.api.InvalidMessageStructureException;
 import org.whispersystems.signalservice.api.messages.SignalServiceAttachment;
 import org.whispersystems.signalservice.api.messages.SignalServiceDataMessage;
+import org.whispersystems.signalservice.api.util.AttachmentPointerUtil;
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos;

 import java.util.LinkedList;
 import java.util.List;
@@ -145,4 +148,33 @@ public class PointerAttachment extends Attachment {
                                             null,
                                             null));
  }
+
+  public static Optional<Attachment> forPointer(SignalServiceProtos.DataMessage.Quote.QuotedAttachment quotedAttachment) {
+    SignalServiceAttachment thumbnail;
+    try {
+      thumbnail = quotedAttachment.hasThumbnail() ? AttachmentPointerUtil.createSignalAttachmentPointer(quotedAttachment.getThumbnail()) : null;
+    } catch (InvalidMessageStructureException e) {
+      return Optional.empty();
+    }
+
+    return Optional.of(new PointerAttachment(quotedAttachment.getContentType(),
+                                             AttachmentTable.TRANSFER_PROGRESS_PENDING,
+                                             thumbnail != null ? thumbnail.asPointer().getSize().orElse(0) : 0,
+                                             quotedAttachment.getFileName(),
+                                             thumbnail != null ? thumbnail.asPointer().getCdnNumber() : 0,
+                                             thumbnail != null ? thumbnail.asPointer().getRemoteId().toString() : "0",
+                                             thumbnail != null && thumbnail.asPointer().getKey() != null ? Base64.encodeBytes(thumbnail.asPointer().getKey()) : null,
+                                             null,
+                                             thumbnail != null ? thumbnail.asPointer().getDigest().orElse(null) : null,
+                                             null,
+                                             false,
+                                             false,
+                                             false,
+                                             thumbnail != null ? thumbnail.asPointer().getWidth() : 0,
+                                             thumbnail != null ? thumbnail.asPointer().getHeight() : 0,
+                                             thumbnail != null ? thumbnail.asPointer().getUploadTimestamp() : 0,
+                                             thumbnail != null ? thumbnail.asPointer().getCaption().orElse(null) : null,
+                                             null,
+                                             null));
+  }
 }
@@ -62,12 +62,12 @@ public class TypingStatusRepository {
      ThreadUtil.cancelRunnableOnMain(timer);
    }

-    timer = () -> onTypingStopped(context, threadId, author, device, false);
+    timer = () -> onTypingStopped(threadId, author, device, false);
    ThreadUtil.runOnMainDelayed(timer, RECIPIENT_TYPING_TIMEOUT);
    timers.put(typist, timer);
  }

-  public synchronized void onTypingStopped(@NonNull Context context, long threadId, @NonNull Recipient author, int device, boolean isReplacedByIncomingMessage) {
+  public synchronized void onTypingStopped(long threadId, @NonNull Recipient author, int device, boolean isReplacedByIncomingMessage) {
    if (author.isSelf()) {
      return;
    }
@@ -2,9 +2,14 @@ package org.thoughtcrime.securesms.contactshare;

 import androidx.annotation.NonNull;

+import org.signal.core.util.logging.Log;
 import org.thoughtcrime.securesms.attachments.Attachment;
 import org.thoughtcrime.securesms.attachments.PointerAttachment;
+import org.whispersystems.signalservice.api.InvalidMessageStructureException;
+import org.whispersystems.signalservice.api.messages.SignalServiceAttachmentPointer;
 import org.whispersystems.signalservice.api.messages.shared.SharedContact;
+import org.whispersystems.signalservice.api.util.AttachmentPointerUtil;
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos;

 import java.util.ArrayList;
 import java.util.LinkedList;
@@ -19,6 +24,8 @@ import static org.thoughtcrime.securesms.contactshare.Contact.PostalAddress;

 public class ContactModelMapper {

+  private static final String TAG = Log.tag(ContactModelMapper.class);
+
  public static SharedContact.Builder localToRemoteBuilder(@NonNull Contact contact) {
    List<SharedContact.Phone>         phoneNumbers    = new ArrayList<>(contact.getPhoneNumbers().size());
    List<SharedContact.Email>         emails          = new ArrayList<>(contact.getEmails().size());
@@ -118,6 +125,57 @@ public class ContactModelMapper {
    return new Contact(name, sharedContact.getOrganization().orElse(null), phoneNumbers, emails, postalAddresses, avatar);
  }

+  public static Contact remoteToLocal(@NonNull SignalServiceProtos.DataMessage.Contact contact) {
+    Name name = new Name(contact.getName().getDisplayName(),
+                         contact.getName().getGivenName(),
+                         contact.getName().getFamilyName(),
+                         contact.getName().getPrefix(),
+                         contact.getName().getSuffix(),
+                         contact.getName().getMiddleName());
+
+    List<Phone> phoneNumbers = new ArrayList<>(contact.getNumberCount());
+    for (SignalServiceProtos.DataMessage.Contact.Phone phone : contact.getNumberList()) {
+      phoneNumbers.add(new Phone(phone.getValue(),
+                                 remoteToLocalType(phone.getType()),
+                                 phone.getLabel()));
+    }
+
+    List<Email> emails = new ArrayList<>(contact.getEmailCount());
+    for (SignalServiceProtos.DataMessage.Contact.Email email : contact.getEmailList()) {
+      emails.add(new Email(email.getValue(),
+                           remoteToLocalType(email.getType()),
+                           email.getLabel()));
+    }
+
+    List<PostalAddress> postalAddresses = new ArrayList<>(contact.getAddressCount());
+    for (SignalServiceProtos.DataMessage.Contact.PostalAddress postalAddress : contact.getAddressList()) {
+      postalAddresses.add(new PostalAddress(remoteToLocalType(postalAddress.getType()),
+                                            postalAddress.getLabel(),
+                                            postalAddress.getStreet(),
+                                            postalAddress.getPobox(),
+                                            postalAddress.getNeighborhood(),
+                                            postalAddress.getCity(),
+                                            postalAddress.getRegion(),
+                                            postalAddress.getPostcode(),
+                                            postalAddress.getCountry()));
+    }
+
+    Avatar avatar = null;
+    if (contact.hasAvatar()) {
+      try {
+        SignalServiceAttachmentPointer attachmentPointer = AttachmentPointerUtil.createSignalAttachmentPointer(contact.getAvatar().getAvatar());
+        Attachment                     attachment        = PointerAttachment.forPointer(Optional.of(attachmentPointer.asPointer())).get();
+        boolean                        isProfile         = contact.getAvatar().getIsProfile();
+
+        avatar = new Avatar(null, attachment, isProfile);
+      } catch (InvalidMessageStructureException e) {
+        Log.w(TAG, "Unable to create avatar for contact", e);
+      }
+    }
+
+    return new Contact(name, contact.getOrganization(), phoneNumbers, emails, postalAddresses, avatar);
+  }
+
  private static Phone.Type remoteToLocalType(SharedContact.Phone.Type type) {
    switch (type) {
      case HOME:   return Phone.Type.HOME;
@@ -127,6 +185,15 @@ public class ContactModelMapper {
    }
  }

+  private static Phone.Type remoteToLocalType(SignalServiceProtos.DataMessage.Contact.Phone.Type type) {
+    switch (type) {
+      case HOME:   return Phone.Type.HOME;
+      case MOBILE: return Phone.Type.MOBILE;
+      case WORK:   return Phone.Type.WORK;
+      default:     return Phone.Type.CUSTOM;
+    }
+  }
+
  private static Email.Type remoteToLocalType(SharedContact.Email.Type type) {
    switch (type) {
      case HOME:   return Email.Type.HOME;
@@ -136,6 +203,15 @@ public class ContactModelMapper {
    }
  }

+  private static Email.Type remoteToLocalType(SignalServiceProtos.DataMessage.Contact.Email.Type type) {
+    switch (type) {
+      case HOME:   return Email.Type.HOME;
+      case MOBILE: return Email.Type.MOBILE;
+      case WORK:   return Email.Type.WORK;
+      default:     return Email.Type.CUSTOM;
+    }
+  }
+
  private static PostalAddress.Type remoteToLocalType(SharedContact.PostalAddress.Type type) {
    switch (type) {
      case HOME:   return PostalAddress.Type.HOME;
@@ -144,6 +220,14 @@ public class ContactModelMapper {
    }
  }

+  private static PostalAddress.Type remoteToLocalType(SignalServiceProtos.DataMessage.Contact.PostalAddress.Type type) {
+    switch (type) {
+      case HOME:   return PostalAddress.Type.HOME;
+      case WORK:   return PostalAddress.Type.WORK;
+      default:     return PostalAddress.Type.CUSTOM;
+    }
+  }
+
  private static SharedContact.Phone.Type localToRemoteType(Phone.Type type) {
    switch (type) {
      case HOME:   return SharedContact.Phone.Type.HOME;
@@ -89,7 +89,7 @@ import java.util.stream.Collectors;

 public class AttachmentTable extends DatabaseTable {

-  private static final String TAG = Log.tag(AttachmentTable.class);
+  public  static final String TAG = Log.tag(AttachmentTable.class);

  public  static final String TABLE_NAME             = "part";
  public  static final String ROW_ID                 = "_id";
@@ -101,7 +101,7 @@ public class AttachmentTable extends DatabaseTable {
          static final String CONTENT_LOCATION       = "cl";
  public  static final String DATA                   = "_data";
          static final String TRANSFER_STATE         = "pending_push";
-  private static final String TRANSFER_FILE          = "transfer_file";
+  public  static final String TRANSFER_FILE          = "transfer_file";
  public  static final String SIZE                   = "data_size";
          static final String FILE_NAME              = "file_name";
  public  static final String UNIQUE_ID              = "unique_id";
@@ -139,6 +139,7 @@ import org.thoughtcrime.securesms.util.Util
 import org.thoughtcrime.securesms.util.isStory
 import org.whispersystems.signalservice.api.messages.multidevice.ReadMessage
 import org.whispersystems.signalservice.api.push.ServiceId
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.SyncMessage
 import java.io.Closeable
 import java.io.IOException
 import java.util.LinkedList
@@ -4342,6 +4343,12 @@ open class MessageTable(context: Context?, databaseHelper: SignalDatabase) : Dat
    return unhandled
  }

+  fun setTimestampReadFromSyncMessageProto(readMessages: List<SyncMessage.Read>, proposedExpireStarted: Long, threadToLatestRead: MutableMap<Long, Long>): Collection<SyncMessageId> {
+    val reads: List<ReadMessage> = readMessages.map { r -> ReadMessage(ServiceId.parseOrThrow(r.senderUuid), r.timestamp) }
+
+    return setTimestampReadFromSyncMessage(reads, proposedExpireStarted, threadToLatestRead)
+  }
+
  /**
   * Handles a synchronized read message.
   * @param messageId An id representing the author-timestamp pair of the message that was read on a linked device. Note that the author could be self when
@@ -5,7 +5,9 @@ import org.thoughtcrime.securesms.recipients.Recipient
 import org.thoughtcrime.securesms.recipients.RecipientId
 import org.whispersystems.signalservice.api.messages.SignalServiceStoryMessageRecipient
 import org.whispersystems.signalservice.api.push.DistributionId
+import org.whispersystems.signalservice.api.push.ServiceId
 import org.whispersystems.signalservice.api.push.SignalServiceAddress
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos

 /**
 * Represents a list of, or update to a list of, who can access a story through what
@@ -81,5 +83,17 @@ data class SentStorySyncManifest(

      return SentStorySyncManifest(entries)
    }
+
+    fun fromRecipientsSet(recipients: List<SignalServiceProtos.SyncMessage.Sent.StoryMessageRecipient>): SentStorySyncManifest {
+      val entries = recipients.toSet().map { recipient ->
+        Entry(
+          recipientId = RecipientId.from(ServiceId.parseOrThrow(recipient.destinationUuid)),
+          allowedToReply = recipient.isAllowedToReply,
+          distributionLists = recipient.distributionListIdsList.map { DistributionId.from(it) }
+        )
+      }
+
+      return SentStorySyncManifest(entries)
+    }
  }
 }
@@ -3,6 +3,8 @@ package org.thoughtcrime.securesms.groups;
 import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;

+import com.google.protobuf.ByteString;
+
 import org.signal.core.util.DatabaseId;
 import org.signal.core.util.Hex;
 import org.signal.libsignal.protocol.kdf.HKDFv3;
@@ -83,6 +85,10 @@ public abstract class GroupId implements DatabaseId {
                               .getGroupIdentifier());
  }

+  public static GroupId.Push push(ByteString bytes) throws BadGroupIdException {
+    return push(bytes.toByteArray());
+  }
+
  public static GroupId.Push push(byte[] bytes) throws BadGroupIdException {
    return bytes.length == V2_BYTE_LENGTH ? v2(bytes) : v1(bytes);
  }
@@ -57,7 +57,6 @@ import org.thoughtcrime.securesms.migrations.PniMigrationJob;
 import org.thoughtcrime.securesms.migrations.ProfileMigrationJob;
 import org.thoughtcrime.securesms.migrations.ProfileSharingUpdateMigrationJob;
 import org.thoughtcrime.securesms.migrations.RebuildMessageSearchIndexMigrationJob;
-import org.thoughtcrime.securesms.migrations.UpdateSmsJobsMigrationJob;
 import org.thoughtcrime.securesms.migrations.RecipientSearchMigrationJob;
 import org.thoughtcrime.securesms.migrations.RegistrationPinV2MigrationJob;
 import org.thoughtcrime.securesms.migrations.StickerAdditionMigrationJob;
@@ -71,6 +70,7 @@ import org.thoughtcrime.securesms.migrations.StoryReadStateMigrationJob;
 import org.thoughtcrime.securesms.migrations.StoryViewedReceiptsStateMigrationJob;
 import org.thoughtcrime.securesms.migrations.SyncDistributionListsMigrationJob;
 import org.thoughtcrime.securesms.migrations.TrimByLengthSettingsMigrationJob;
+import org.thoughtcrime.securesms.migrations.UpdateSmsJobsMigrationJob;
 import org.thoughtcrime.securesms.migrations.UserNotificationMigrationJob;
 import org.thoughtcrime.securesms.migrations.UuidMigrationJob;

@@ -162,10 +162,11 @@ public final class JobManagerFactories {
      put(PushNotificationReceiveJob.KEY,            new PushNotificationReceiveJob.Factory());
      put(PushProcessEarlyMessagesJob.KEY,           new PushProcessEarlyMessagesJob.Factory());
      put(PushProcessMessageJob.KEY,                 new PushProcessMessageJob.Factory());
+      put(PushProcessMessageJobV2.KEY,               new PushProcessMessageJobV2.Factory());
      put(ReactionSendJob.KEY,                       new ReactionSendJob.Factory());
      put(RebuildMessageSearchIndexJob.KEY,          new RebuildMessageSearchIndexJob.Factory());
      put(RefreshAttributesJob.KEY,                  new RefreshAttributesJob.Factory());
-      put(RefreshKbsCredentialsJob.KEY, new RefreshKbsCredentialsJob.Factory());
+      put(RefreshKbsCredentialsJob.KEY,              new RefreshKbsCredentialsJob.Factory());
      put(RefreshOwnProfileJob.KEY,                  new RefreshOwnProfileJob.Factory());
      put(RemoteConfigRefreshJob.KEY,                new RemoteConfigRefreshJob.Factory());
      put(RemoteDeleteSendJob.KEY,                   new RemoteDeleteSendJob.Factory());
@@ -1,14 +1,15 @@
 package org.thoughtcrime.securesms.jobs

 import org.signal.core.util.logging.Log
+import org.signal.core.util.orNull
 import org.thoughtcrime.securesms.database.SignalDatabase
 import org.thoughtcrime.securesms.database.model.ServiceMessageId
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
 import org.thoughtcrime.securesms.jobmanager.Job
 import org.thoughtcrime.securesms.messages.MessageContentProcessor
+import org.thoughtcrime.securesms.messages.MessageContentProcessorV2
+import org.thoughtcrime.securesms.util.EarlyMessageCacheEntry
 import org.whispersystems.signalservice.api.messages.SignalServiceContent
-import java.lang.Exception
-import java.util.Optional

 /**
 * A job that should be enqueued whenever we process a message that we think has arrived "early" (see [org.thoughtcrime.securesms.util.EarlyMessageCache]).
@@ -42,13 +43,19 @@ class PushProcessEarlyMessagesJob private constructor(parameters: Parameters) :
      Log.i(TAG, "There are ${earlyIds.size} items in the early message cache with matches.")

      for (id: ServiceMessageId in earlyIds) {
-        val contents: Optional<List<SignalServiceContent>> = ApplicationDependencies.getEarlyMessageCache().retrieve(id.sender, id.sentTimestamp)
+        val contents: List<SignalServiceContent>? = ApplicationDependencies.getEarlyMessageCache().retrieve(id.sender, id.sentTimestamp).orNull()
+        val earlyEntries: List<EarlyMessageCacheEntry>? = ApplicationDependencies.getEarlyMessageCache().retrieveV2(id.sender, id.sentTimestamp).orNull()

-        if (contents.isPresent) {
-          for (content: SignalServiceContent in contents.get()) {
+        if (contents != null) {
+          for (content: SignalServiceContent in contents) {
            Log.i(TAG, "[${id.sentTimestamp}] Processing early content for $id")
            MessageContentProcessor.create(context).processEarlyContent(MessageContentProcessor.MessageState.DECRYPTED_OK, content, null, id.sentTimestamp, -1)
          }
+        } else if (earlyEntries != null) {
+          for (entry in earlyEntries) {
+            Log.i(TAG, "[${id.sentTimestamp}] Processing early V2 content for $id")
+            MessageContentProcessorV2.create(context).process(entry.envelope, entry.content, entry.metadata, entry.serverDeliveredTimestamp, processingEarlyContent = true)
+          }
        } else {
          Log.w(TAG, "[${id.sentTimestamp}] Saw $id in the cache, but when we went to retrieve it, it was already gone.")
        }
@@ -0,0 +1,141 @@
+package org.thoughtcrime.securesms.jobs
+
+import androidx.annotation.WorkerThread
+import okio.ByteString
+import okio.ByteString.Companion.toByteString
+import org.signal.core.util.logging.Log
+import org.thoughtcrime.securesms.database.SignalDatabase.Companion.groups
+import org.thoughtcrime.securesms.groups.GroupChangeBusyException
+import org.thoughtcrime.securesms.jobmanager.Job
+import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraint
+import org.thoughtcrime.securesms.messages.MessageContentProcessorV2
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.groupId
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.recipients.RecipientId
+import org.thoughtcrime.securesms.util.GroupUtil
+import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
+import org.whispersystems.signalservice.api.crypto.protos.CompleteMessage
+import org.whispersystems.signalservice.api.groupsv2.NoCredentialForRedemptionTimeException
+import org.whispersystems.signalservice.api.push.ServiceId
+import org.whispersystems.signalservice.api.push.exceptions.PushNetworkException
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.Content
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.Envelope
+import java.io.IOException
+import java.util.concurrent.TimeUnit
+import org.whispersystems.signalservice.api.crypto.protos.EnvelopeMetadata as EnvelopeMetadataProto
+
+class PushProcessMessageJobV2 private constructor(
+  parameters: Parameters,
+  private val envelope: Envelope,
+  private val content: Content,
+  private val metadata: EnvelopeMetadata,
+  private val serverDeliveredTimestamp: Long
+) : BaseJob(parameters) {
+
+  @WorkerThread
+  constructor(
+    envelope: Envelope,
+    content: Content,
+    metadata: EnvelopeMetadata,
+    serverDeliveredTimestamp: Long
+  ) : this(createParameters(content, metadata), envelope.toBuilder().clearContent().build(), content, metadata, serverDeliveredTimestamp)
+
+  override fun shouldTrace() = true
+
+  override fun serialize(): ByteArray {
+    return CompleteMessage(
+      envelope = envelope.toByteArray().toByteString(),
+      content = content.toByteArray().toByteString(),
+      metadata = EnvelopeMetadataProto(
+        sourceServiceId = ByteString.of(*metadata.sourceServiceId.toByteArray()),
+        sourceE164 = metadata.sourceE164,
+        sourceDeviceId = metadata.sourceDeviceId,
+        sealedSender = metadata.sealedSender,
+        groupId = if (metadata.groupId != null) metadata.groupId!!.toByteString() else null,
+        destinationServiceId = ByteString.of(*metadata.destinationServiceId.toByteArray())
+      ),
+      serverDeliveredTimestamp = serverDeliveredTimestamp
+    ).encode()
+  }
+
+  override fun getFactoryKey(): String {
+    return KEY
+  }
+
+  public override fun onRun() {
+    val processor = MessageContentProcessorV2.create(context)
+    processor.process(envelope, content, metadata, serverDeliveredTimestamp)
+  }
+
+  public override fun onShouldRetry(e: Exception): Boolean {
+    return e is PushNetworkException ||
+      e is NoCredentialForRedemptionTimeException ||
+      e is GroupChangeBusyException
+  }
+
+  override fun onFailure() = Unit
+
+  class Factory : Job.Factory<PushProcessMessageJobV2?> {
+    override fun create(parameters: Parameters, data: ByteArray?): PushProcessMessageJobV2 {
+      return try {
+        val completeMessage = CompleteMessage.ADAPTER.decode(data!!)
+        PushProcessMessageJobV2(
+          parameters = parameters,
+          envelope = Envelope.parseFrom(completeMessage.envelope.toByteArray()),
+          content = Content.parseFrom(completeMessage.content.toByteArray()),
+          metadata = EnvelopeMetadata(
+            sourceServiceId = ServiceId.parseOrThrow(completeMessage.metadata.sourceServiceId.toByteArray()),
+            sourceE164 = completeMessage.metadata.sourceE164,
+            sourceDeviceId = completeMessage.metadata.sourceDeviceId,
+            sealedSender = completeMessage.metadata.sealedSender,
+            groupId = completeMessage.metadata.groupId?.toByteArray(),
+            destinationServiceId = ServiceId.parseOrThrow(completeMessage.metadata.destinationServiceId.toByteArray())
+          ),
+          serverDeliveredTimestamp = completeMessage.serverDeliveredTimestamp
+        )
+      } catch (e: IOException) {
+        throw AssertionError(e)
+      }
+    }
+  }
+
+  companion object {
+    const val KEY = "PushProcessMessageJobV2"
+    const val QUEUE_PREFIX = "__PUSH_PROCESS_JOB__"
+
+    private val TAG = Log.tag(PushProcessMessageJobV2::class.java)
+
+    private fun getQueueName(recipientId: RecipientId): String {
+      return QUEUE_PREFIX + recipientId.toQueueKey()
+    }
+
+    @WorkerThread
+    private fun createParameters(content: Content, metadata: EnvelopeMetadata): Parameters {
+      val queueName: String
+      val builder = Parameters.Builder().setMaxAttempts(Parameters.UNLIMITED)
+      val groupContext = GroupUtil.getGroupContextIfPresent(content)
+      val groupId = groupContext?.groupId
+
+      if (groupContext != null && groupId != null) {
+        queueName = getQueueName(Recipient.externalPossiblyMigratedGroup(groupId).id)
+
+        if (groupId.isV2) {
+          val localRevision = groups.getGroupV2Revision(groupId.requireV2())
+
+          if (groupContext.revision > localRevision || groups.getGroupV1ByExpectedV2(groupId.requireV2()).isPresent) {
+            Log.i(TAG, "Adding network constraint to group-related job.")
+            builder.addConstraint(NetworkConstraint.KEY).setLifespan(TimeUnit.DAYS.toMillis(30))
+          }
+        }
+      } else if (content.hasSyncMessage() && content.syncMessage.hasSent() && content.syncMessage.sent.hasDestinationUuid()) {
+        queueName = getQueueName(RecipientId.from(ServiceId.parseOrThrow(content.syncMessage.sent.destinationUuid)))
+      } else {
+        queueName = getQueueName(RecipientId.from(metadata.sourceServiceId))
+      }
+
+      builder.setQueue(queueName)
+
+      return builder.build()
+    }
+  }
+}
@@ -0,0 +1,162 @@
+package org.thoughtcrime.securesms.messages
+
+import org.signal.ringrtc.CallId
+import org.thoughtcrime.securesms.database.model.IdentityRecord
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
+import org.thoughtcrime.securesms.messages.MessageContentProcessorV2.Companion.log
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.recipients.RecipientId
+import org.thoughtcrime.securesms.ringrtc.RemotePeer
+import org.thoughtcrime.securesms.service.webrtc.WebRtcData.AnswerMetadata
+import org.thoughtcrime.securesms.service.webrtc.WebRtcData.CallMetadata
+import org.thoughtcrime.securesms.service.webrtc.WebRtcData.HangupMetadata
+import org.thoughtcrime.securesms.service.webrtc.WebRtcData.OfferMetadata
+import org.thoughtcrime.securesms.service.webrtc.WebRtcData.OpaqueMessageMetadata
+import org.thoughtcrime.securesms.service.webrtc.WebRtcData.ReceivedAnswerMetadata
+import org.thoughtcrime.securesms.service.webrtc.WebRtcData.ReceivedOfferMetadata
+import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
+import org.whispersystems.signalservice.api.messages.calls.HangupMessage
+import org.whispersystems.signalservice.api.messages.calls.OfferMessage
+import org.whispersystems.signalservice.api.push.ServiceId
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.CallMessage.Offer
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.CallMessage.Opaque
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.Envelope
+
+object CallMessageProcessor {
+  fun process(
+    senderRecipient: Recipient,
+    envelope: Envelope,
+    content: SignalServiceProtos.Content,
+    metadata: EnvelopeMetadata,
+    serverDeliveredTimestamp: Long
+  ) {
+    val callMessage = content.callMessage
+
+    when {
+      callMessage.hasOffer() -> handleCallOfferMessage(envelope, metadata, callMessage.offer, senderRecipient.id, serverDeliveredTimestamp, callMessage.multiRing)
+      callMessage.hasAnswer() -> handleCallAnswerMessage(envelope, metadata, callMessage.answer, senderRecipient.id, callMessage.multiRing)
+      callMessage.iceUpdateList.isNotEmpty() -> handleCallIceUpdateMessage(envelope, metadata, callMessage.iceUpdateList, senderRecipient.id)
+      callMessage.hasHangup() || callMessage.hasLegacyHangup() -> {
+        val hangup = if (callMessage.hasHangup()) callMessage.hangup else callMessage.legacyHangup
+        handleCallHangupMessage(envelope, metadata, hangup, senderRecipient.id, callMessage.hasLegacyHangup())
+      }
+      callMessage.hasBusy() -> handleCallBusyMessage(envelope, metadata, callMessage.busy, senderRecipient.id)
+      callMessage.hasOpaque() -> handleCallOpaqueMessage(envelope, metadata, callMessage.opaque, senderRecipient.requireServiceId(), serverDeliveredTimestamp)
+    }
+  }
+
+  private fun handleCallOfferMessage(envelope: Envelope, metadata: EnvelopeMetadata, offer: Offer, senderRecipientId: RecipientId, serverDeliveredTimestamp: Long, multiRing: Boolean) {
+    log(envelope.timestamp, "handleCallOfferMessage...")
+
+    val remotePeer = RemotePeer(senderRecipientId, CallId(offer.id))
+    val remoteIdentityKey = ApplicationDependencies.getProtocolStore().aci().identities().getIdentityRecord(senderRecipientId).map { (_, identityKey): IdentityRecord -> identityKey.serialize() }.get()
+
+    ApplicationDependencies.getSignalCallManager()
+      .receivedOffer(
+        CallMetadata(remotePeer, metadata.sourceDeviceId),
+        offer.toCallOfferMetadata(),
+        ReceivedOfferMetadata(
+          remoteIdentityKey,
+          envelope.serverTimestamp,
+          serverDeliveredTimestamp,
+          multiRing
+        )
+      )
+  }
+
+  private fun handleCallAnswerMessage(
+    envelope: Envelope,
+    metadata: EnvelopeMetadata,
+    answer: SignalServiceProtos.CallMessage.Answer,
+    senderRecipientId: RecipientId,
+    multiRing: Boolean
+  ) {
+    log(envelope.timestamp, "handleCallAnswerMessage...")
+
+    val remotePeer = RemotePeer(senderRecipientId, CallId(answer.id))
+    val remoteIdentityKey = ApplicationDependencies.getProtocolStore().aci().identities().getIdentityRecord(senderRecipientId).map { (_, identityKey): IdentityRecord -> identityKey.serialize() }.get()
+
+    ApplicationDependencies.getSignalCallManager()
+      .receivedAnswer(
+        CallMetadata(remotePeer, metadata.sourceDeviceId),
+        AnswerMetadata(if (answer.hasOpaque()) answer.opaque.toByteArray() else null, if (answer.hasSdp()) answer.sdp else null),
+        ReceivedAnswerMetadata(remoteIdentityKey, multiRing)
+      )
+  }
+
+  private fun handleCallIceUpdateMessage(
+    envelope: Envelope,
+    metadata: EnvelopeMetadata,
+    iceUpdateList: MutableList<SignalServiceProtos.CallMessage.IceUpdate>,
+    senderRecipientId: RecipientId
+  ) {
+    log(envelope.timestamp, "handleCallIceUpdateMessage... " + iceUpdateList.size)
+
+    val iceCandidates: MutableList<ByteArray> = ArrayList(iceUpdateList.size)
+    var callId: Long = -1
+
+    iceUpdateList
+      .filter { it.hasOpaque() }
+      .forEach { iceUpdate ->
+        iceCandidates += iceUpdate.opaque.toByteArray()
+        callId = iceUpdate.id
+      }
+
+    val remotePeer = RemotePeer(senderRecipientId, CallId(callId))
+    ApplicationDependencies.getSignalCallManager()
+      .receivedIceCandidates(
+        CallMetadata(remotePeer, metadata.sourceDeviceId),
+        iceCandidates
+      )
+  }
+
+  private fun handleCallHangupMessage(
+    envelope: Envelope,
+    metadata: EnvelopeMetadata,
+    hangup: SignalServiceProtos.CallMessage.Hangup,
+    senderRecipientId: RecipientId,
+    isLegacyHangup: Boolean
+  ) {
+    log(envelope.timestamp, "handleCallHangupMessage")
+
+    val remotePeer = RemotePeer(senderRecipientId, CallId(hangup.id))
+    ApplicationDependencies.getSignalCallManager()
+      .receivedCallHangup(
+        CallMetadata(remotePeer, metadata.sourceDeviceId),
+        HangupMetadata(HangupMessage.Type.fromProto(hangup.type), isLegacyHangup, hangup.deviceId)
+      )
+  }
+
+  private fun handleCallBusyMessage(envelope: Envelope, metadata: EnvelopeMetadata, busy: SignalServiceProtos.CallMessage.Busy, senderRecipientId: RecipientId) {
+    log(envelope.timestamp, "handleCallBusyMessage")
+
+    val remotePeer = RemotePeer(senderRecipientId, CallId(busy.id))
+    ApplicationDependencies.getSignalCallManager().receivedCallBusy(CallMetadata(remotePeer, metadata.sourceDeviceId))
+  }
+
+  private fun handleCallOpaqueMessage(envelope: Envelope, metadata: EnvelopeMetadata, opaque: Opaque, senderServiceId: ServiceId, serverDeliveredTimestamp: Long) {
+    log(envelope.timestamp.toString(), "handleCallOpaqueMessage")
+
+    var messageAgeSeconds: Long = 0
+    if (envelope.serverTimestamp in 1..serverDeliveredTimestamp) {
+      messageAgeSeconds = (serverDeliveredTimestamp - envelope.serverTimestamp) / 1000
+    }
+
+    ApplicationDependencies.getSignalCallManager()
+      .receivedOpaqueMessage(
+        OpaqueMessageMetadata(
+          senderServiceId.uuid(),
+          opaque.data.toByteArray(),
+          metadata.sourceDeviceId,
+          messageAgeSeconds
+        )
+      )
+  }
+
+  private fun Offer.toCallOfferMetadata(): OfferMetadata {
+    val sdp = if (hasSdp()) sdp else null
+    val opaque = if (hasOpaque()) opaque else null
+    return OfferMetadata(opaque?.toByteArray(), sdp, OfferMessage.Type.fromProto(type))
+  }
+}
@@ -28,6 +28,7 @@ import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraint
 import org.thoughtcrime.securesms.jobs.ForegroundServiceUtil.startWhenCapable
 import org.thoughtcrime.securesms.jobs.PushDecryptMessageJob
 import org.thoughtcrime.securesms.jobs.PushProcessMessageJob
+import org.thoughtcrime.securesms.jobs.PushProcessMessageJobV2
 import org.thoughtcrime.securesms.jobs.UnableToStartException
 import org.thoughtcrime.securesms.keyvalue.SignalStore
 import org.thoughtcrime.securesms.messages.MessageDecryptor.FollowUpOperation
@@ -36,17 +37,11 @@ import org.thoughtcrime.securesms.notifications.NotificationChannels
 import org.thoughtcrime.securesms.recipients.RecipientId
 import org.thoughtcrime.securesms.util.AppForegroundObserver
 import org.thoughtcrime.securesms.util.Util
-import org.whispersystems.signalservice.api.messages.SignalServiceContent
-import org.whispersystems.signalservice.api.messages.SignalServiceMetadata
 import org.whispersystems.signalservice.api.push.ServiceId
-import org.whispersystems.signalservice.api.push.SignalServiceAddress
 import org.whispersystems.signalservice.api.util.UuidUtil
 import org.whispersystems.signalservice.api.websocket.WebSocketConnectionState
 import org.whispersystems.signalservice.api.websocket.WebSocketUnavailableException
 import org.whispersystems.signalservice.internal.push.SignalServiceProtos
-import org.whispersystems.signalservice.internal.serialize.SignalServiceAddressProtobufSerializer
-import org.whispersystems.signalservice.internal.serialize.SignalServiceMetadataProtobufSerializer
-import org.whispersystems.signalservice.internal.serialize.protos.SignalServiceContentProto
 import java.util.*
 import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.TimeUnit
@@ -280,13 +275,7 @@ class IncomingMessageObserver(private val context: Application) {

    val extraJob: Job? = when (result) {
      is MessageDecryptor.Result.Success -> {
-        PushProcessMessageJob(
-          result.toMessageState(),
-          result.toSignalServiceContent(),
-          null,
-          -1,
-          result.envelope.timestamp
-        )
+        PushProcessMessageJobV2(result.envelope, result.content, result.metadata, result.serverDeliveredTimestamp)
      }

      is MessageDecryptor.Result.Error -> {
@@ -336,29 +325,6 @@ class IncomingMessageObserver(private val context: Application) {
    }
  }

-  private fun MessageDecryptor.Result.Success.toSignalServiceContent(): SignalServiceContent {
-    val localAddress = SignalServiceAddress(this.metadata.destinationServiceId, Optional.ofNullable(SignalStore.account().e164))
-    val metadata = SignalServiceMetadata(
-      SignalServiceAddress(this.metadata.sourceServiceId, Optional.ofNullable(this.metadata.sourceE164)),
-      this.metadata.sourceDeviceId,
-      this.envelope.timestamp,
-      this.envelope.serverTimestamp,
-      this.serverDeliveredTimestamp,
-      this.metadata.sealedSender,
-      this.envelope.serverGuid,
-      Optional.ofNullable(this.metadata.groupId),
-      this.metadata.destinationServiceId.toString()
-    )
-
-    val contentProto = SignalServiceContentProto.newBuilder()
-      .setLocalAddress(SignalServiceAddressProtobufSerializer.toProtobuf(localAddress))
-      .setMetadata(SignalServiceMetadataProtobufSerializer.toProtobuf(metadata))
-      .setContent(content)
-      .build()
-
-    return SignalServiceContent.createFromProto(contentProto)!!
-  }
-
  private fun MessageDecryptor.ErrorMetadata.toExceptionMetadata(): MessageContentProcessor.ExceptionMetadata {
    return MessageContentProcessor.ExceptionMetadata(
      this.sender,
@@ -2879,7 +2879,7 @@ public class MessageContentProcessor {
      ApplicationDependencies.getTypingStatusRepository().onTypingStarted(context,threadId, senderRecipient, content.getSenderDevice());
    } else {
      Log.d(TAG, "Typing stopped on thread " + threadId);
-      ApplicationDependencies.getTypingStatusRepository().onTypingStopped(context, threadId, senderRecipient, content.getSenderDevice(), false);
+      ApplicationDependencies.getTypingStatusRepository().onTypingStopped(threadId, senderRecipient, content.getSenderDevice(), false);
    }
  }

@@ -3257,11 +3257,11 @@ public class MessageContentProcessor {
  }

  private void notifyTypingStoppedFromIncomingMessage(@NonNull Recipient senderRecipient, @NonNull Recipient conversationRecipient, int device) {
-    long threadId = SignalDatabase.threads().getOrCreateThreadIdFor(conversationRecipient);
+    long threadId = SignalDatabase.threads().getThreadIdIfExistsFor(conversationRecipient.getId());

    if (threadId > 0 && TextSecurePreferences.isTypingIndicatorsEnabled(context)) {
      Log.d(TAG, "Typing stopped on thread " + threadId + " due to an incoming message.");
-      ApplicationDependencies.getTypingStatusRepository().onTypingStopped(context, threadId, senderRecipient, device, true);
+      ApplicationDependencies.getTypingStatusRepository().onTypingStopped(threadId, senderRecipient, device, true);
    }
  }

@@ -3403,13 +3403,11 @@ public class MessageContentProcessor {
    }
  }

-
-  @SuppressWarnings("WeakerAccess")
-  private static class StorageFailedException extends Exception {
+  static class StorageFailedException extends Exception {
    private final String sender;
    private final int    senderDevice;

-    private StorageFailedException(Exception e, String sender, int senderDevice) {
+    StorageFailedException(Exception e, String sender, int senderDevice) {
      super(e);
      this.sender       = sender;
      this.senderDevice = senderDevice;
@@ -0,0 +1,572 @@
+package org.thoughtcrime.securesms.messages
+
+import android.content.Context
+import org.signal.core.util.logging.Log
+import org.signal.core.util.orNull
+import org.signal.libsignal.protocol.SignalProtocolAddress
+import org.signal.libsignal.protocol.ecc.ECPublicKey
+import org.signal.libsignal.protocol.message.DecryptionErrorMessage
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.database.model.MessageLogEntry
+import org.thoughtcrime.securesms.database.model.MessageRecord
+import org.thoughtcrime.securesms.database.model.PendingRetryReceiptModel
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
+import org.thoughtcrime.securesms.groups.BadGroupIdException
+import org.thoughtcrime.securesms.groups.GroupChangeBusyException
+import org.thoughtcrime.securesms.groups.GroupId
+import org.thoughtcrime.securesms.groups.GroupManager
+import org.thoughtcrime.securesms.groups.GroupNotAMemberException
+import org.thoughtcrime.securesms.groups.GroupsV1MigrationUtil
+import org.thoughtcrime.securesms.jobs.NullMessageSendJob
+import org.thoughtcrime.securesms.jobs.ResendMessageJob
+import org.thoughtcrime.securesms.jobs.SenderKeyDistributionSendJob
+import org.thoughtcrime.securesms.keyvalue.SignalStore
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.groupMasterKey
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.hasDisallowedAnnouncementOnlyContent
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.hasGroupContext
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.hasSignedGroupChange
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.hasStarted
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.isExpirationUpdate
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.isMediaMessage
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.isValid
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.signedGroupChange
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.toDecryptionErrorMessage
+import org.thoughtcrime.securesms.notifications.v2.ConversationId
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.recipients.RecipientId
+import org.thoughtcrime.securesms.util.EarlyMessageCacheEntry
+import org.thoughtcrime.securesms.util.FeatureFlags
+import org.thoughtcrime.securesms.util.TextSecurePreferences
+import org.thoughtcrime.securesms.util.Util
+import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
+import org.whispersystems.signalservice.api.push.DistributionId
+import org.whispersystems.signalservice.api.push.ServiceId
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.Content
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.Envelope
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.TypingMessage
+import java.io.IOException
+
+class MessageContentProcessorV2(private val context: Context) {
+
+  companion object {
+    const val TAG = "MessageProcessorV2"
+
+    @JvmStatic
+    @JvmOverloads
+    fun create(context: Context = ApplicationDependencies.getApplication()): MessageContentProcessorV2 {
+      return MessageContentProcessorV2(context)
+    }
+
+    fun debug(message: String) {
+      Log.d(TAG, message)
+    }
+
+    fun log(message: String) {
+      Log.i(TAG, message)
+    }
+
+    fun log(timestamp: Long, message: String) {
+      log(timestamp.toString(), message)
+    }
+
+    fun log(extra: String, message: String) {
+      val extraLog = if (Util.isEmpty(extra)) "" else "[$extra] "
+      Log.i(TAG, extraLog + message)
+    }
+
+    fun warn(message: String) {
+      warn("", message, null)
+    }
+
+    fun warn(extra: String, message: String) {
+      warn(extra, message, null)
+    }
+
+    fun warn(timestamp: Long, message: String) {
+      warn(timestamp.toString(), message)
+    }
+
+    fun warn(timestamp: Long, message: String, t: Throwable?) {
+      warn(timestamp.toString(), message, t)
+    }
+
+    fun warn(message: String, t: Throwable?) {
+      warn("", message, t)
+    }
+
+    fun warn(extra: String, message: String, t: Throwable?) {
+      val extraLog = if (Util.isEmpty(extra)) "" else "[$extra] "
+      Log.w(TAG, extraLog + message, t)
+    }
+
+    fun formatSender(recipientId: RecipientId, serviceId: ServiceId, device: Int): String {
+      return "$recipientId ($serviceId.$device)"
+    }
+
+    @Throws(BadGroupIdException::class)
+    private fun getMessageDestination(content: Content, sender: Recipient): Recipient {
+      return if (content.hasStoryMessage() && content.storyMessage.group.isValid) {
+        getGroupRecipient(content.storyMessage.group, sender)
+      } else if (content.dataMessage.hasGroupContext) {
+        getGroupRecipient(content.dataMessage.groupV2, sender)
+      } else {
+        sender
+      }
+    }
+
+    private fun getGroupRecipient(groupContextV2: SignalServiceProtos.GroupContextV2?, senderRecipient: Recipient): Recipient {
+      return if (groupContextV2 != null) {
+        Recipient.externalPossiblyMigratedGroup(GroupId.v2(groupContextV2.groupMasterKey))
+      } else {
+        senderRecipient
+      }
+    }
+
+    @Throws(BadGroupIdException::class)
+    private fun shouldIgnore(content: Content, senderRecipient: Recipient, threadRecipient: Recipient): Boolean {
+      if (content.hasDataMessage()) {
+        val message = content.dataMessage
+        return if (threadRecipient.isGroup && threadRecipient.isBlocked) {
+          true
+        } else if (threadRecipient.isGroup) {
+          val groupId = if (message.hasGroupV2()) GroupId.v2(message.groupV2.groupMasterKey) else null
+          if (groupId != null && SignalDatabase.groups.isUnknownGroup(groupId)) {
+            return senderRecipient.isBlocked
+          }
+
+          val isTextMessage = message.hasBody()
+          val isMediaMessage = message.isMediaMessage
+          val isExpireMessage = message.isExpirationUpdate
+          val isGv2Update = message.hasSignedGroupChange
+          val isContentMessage = !isGv2Update && !isExpireMessage && (isTextMessage || isMediaMessage)
+          val isGroupActive = groupId != null && SignalDatabase.groups.isActive(groupId)
+
+          isContentMessage && !isGroupActive || senderRecipient.isBlocked && !isGv2Update
+        } else {
+          senderRecipient.isBlocked
+        }
+      } else if (content.hasCallMessage()) {
+        return senderRecipient.isBlocked
+      } else if (content.hasTypingMessage()) {
+        if (senderRecipient.isBlocked) {
+          return true
+        }
+
+        if (content.typingMessage.hasGroupId()) {
+          val groupId: GroupId = GroupId.push(content.typingMessage.groupId)
+          val groupRecipient = Recipient.externalPossiblyMigratedGroup(groupId)
+          return if (groupRecipient.isBlocked || !groupRecipient.isActiveGroup) {
+            true
+          } else {
+            val groupRecord = SignalDatabase.groups.getGroup(groupId)
+            groupRecord.isPresent && groupRecord.get().isAnnouncementGroup && !groupRecord.get().admins.contains(senderRecipient)
+          }
+        }
+      } else if (content.hasStoryMessage()) {
+        return if (threadRecipient.isGroup && threadRecipient.isBlocked) {
+          true
+        } else {
+          senderRecipient.isBlocked
+        }
+      }
+      return false
+    }
+
+    @Throws(BadGroupIdException::class)
+    private fun handlePendingRetry(pending: PendingRetryReceiptModel?, timestamp: Long, destination: Recipient): Long {
+      var receivedTime = System.currentTimeMillis()
+
+      if (pending != null) {
+        warn(timestamp, "Incoming message matches a pending retry we were expecting.")
+
+        val threadId = SignalDatabase.threads.getThreadIdFor(destination.id)
+        if (threadId != null) {
+          val (lastSeen) = SignalDatabase.threads.getConversationMetadata(threadId)
+          val visibleThread = ApplicationDependencies.getMessageNotifier().visibleThread.map(ConversationId::threadId).orElse(-1L)
+
+          if (threadId != visibleThread && lastSeen > 0 && lastSeen < pending.receivedTimestamp) {
+            receivedTime = pending.receivedTimestamp
+            warn(timestamp, "Thread has not been opened yet. Using received timestamp of $receivedTime")
+          } else {
+            warn(timestamp, "Thread was opened after receiving the original message. Using the current time for received time. (Last seen: " + lastSeen + ", ThreadVisible: " + (threadId == visibleThread) + ")")
+          }
+        } else {
+          warn(timestamp, "Could not find a thread for the pending message. Using current time for received time.")
+        }
+      }
+
+      return receivedTime
+    }
+
+    /**
+     * @return True if the content should be ignored, otherwise false.
+     */
+    @Throws(IOException::class, GroupChangeBusyException::class)
+    fun handleGv2PreProcessing(
+      context: Context,
+      timestamp: Long,
+      content: Content,
+      metadata: EnvelopeMetadata,
+      groupId: GroupId.V2,
+      groupV2: SignalServiceProtos.GroupContextV2,
+      senderRecipient: Recipient
+    ): Boolean {
+      val possibleGv1 = SignalDatabase.groups.getGroupV1ByExpectedV2(groupId)
+      if (possibleGv1.isPresent) {
+        GroupsV1MigrationUtil.performLocalMigration(context, possibleGv1.get().id.requireV1())
+      }
+
+      if (!updateGv2GroupFromServerOrP2PChange(context, timestamp, groupV2)) {
+        log(timestamp, "Ignoring GV2 message for group we are not currently in $groupId")
+        return true
+      }
+
+      val groupRecord = SignalDatabase.groups.getGroup(groupId)
+      if (groupRecord.isPresent && !groupRecord.get().members.contains(senderRecipient.id)) {
+        log(timestamp, "Ignoring GV2 message from member not in group $groupId. Sender: ${formatSender(senderRecipient.id, metadata.sourceServiceId, metadata.sourceDeviceId)}")
+        return true
+      }
+
+      if (groupRecord.isPresent && groupRecord.get().isAnnouncementGroup && !groupRecord.get().admins.contains(senderRecipient)) {
+        if (content.hasDataMessage()) {
+          if (content.dataMessage.hasDisallowedAnnouncementOnlyContent) {
+            Log.w(TAG, "Ignoring message from ${senderRecipient.id} because it has disallowed content, and they're not an admin in an announcement-only group.")
+            return true
+          }
+        } else if (content.hasTypingMessage()) {
+          Log.w(TAG, "Ignoring typing indicator from ${senderRecipient.id} because they're not an admin in an announcement-only group.")
+          return true
+        }
+      }
+
+      return false
+    }
+
+    @Throws(IOException::class, GroupChangeBusyException::class)
+    fun updateGv2GroupFromServerOrP2PChange(
+      context: Context,
+      timestamp: Long,
+      groupV2: SignalServiceProtos.GroupContextV2
+    ): Boolean {
+      return try {
+        val updatedTimestamp = if (groupV2.hasSignedGroupChange) timestamp else timestamp - 1
+        GroupManager.updateGroupFromServer(context, groupV2.groupMasterKey, groupV2.revision, updatedTimestamp, groupV2.signedGroupChange)
+        true
+      } catch (e: GroupNotAMemberException) {
+        warn(timestamp, "Ignoring message for a group we're not in")
+        false
+      }
+    }
+
+    private fun resetRecipientToPush(recipient: Recipient) {
+      if (recipient.isForceSmsSelection) {
+        SignalDatabase.recipients.setForceSmsSelection(recipient.id, false)
+      }
+    }
+  }
+
+  /**
+   * Given the details about a message decryption, this will insert the proper message content into
+   * the database.
+   *
+   * This is super-stateful, and it's recommended that this be run in a transaction so that no
+   * intermediate results are persisted to the database if the app were to crash.
+   *
+   * @param processingEarlyContent pass `true` to specifically target at early content. Using this method will *not*
+   * store or enqueue early content jobs if we detect this as being early, to avoid recursive scenarios.
+   */
+  @JvmOverloads
+  fun process(envelope: Envelope, content: Content, metadata: EnvelopeMetadata, serverDeliveredTimestamp: Long, processingEarlyContent: Boolean = false) {
+    val senderRecipient = Recipient.externalPush(metadata.sourceServiceId)
+
+    handleMessage(senderRecipient, envelope, content, metadata, serverDeliveredTimestamp, processingEarlyContent)
+
+    val earlyCacheEntries: List<EarlyMessageCacheEntry>? = ApplicationDependencies
+      .getEarlyMessageCache()
+      .retrieveV2(senderRecipient.id, envelope.timestamp)
+      .orNull()
+
+    if (!processingEarlyContent && earlyCacheEntries != null) {
+      log(envelope.timestamp, "Found " + earlyCacheEntries.size + " dependent item(s) that were retrieved earlier. Processing.")
+      for (entry in earlyCacheEntries) {
+        handleMessage(senderRecipient, entry.envelope, entry.content, entry.metadata, entry.serverDeliveredTimestamp, processingEarlyContent = true)
+      }
+    }
+  }
+
+  private fun handleMessage(
+    senderRecipient: Recipient,
+    envelope: Envelope,
+    content: Content,
+    metadata: EnvelopeMetadata,
+    serverDeliveredTimestamp: Long,
+    processingEarlyContent: Boolean
+  ) {
+    val threadRecipient = getMessageDestination(content, senderRecipient)
+
+    if (shouldIgnore(content, senderRecipient, threadRecipient)) {
+      log(envelope.timestamp, "Ignoring message.")
+      return
+    }
+
+    val pending: PendingRetryReceiptModel? = ApplicationDependencies.getPendingRetryReceiptCache().get(senderRecipient.id, envelope.timestamp)
+    val receivedTime: Long = handlePendingRetry(pending, envelope.timestamp, threadRecipient)
+
+    log(envelope.timestamp, "Beginning message processing. Sender: " + formatSender(senderRecipient.id, metadata.sourceServiceId, metadata.sourceDeviceId))
+
+    when {
+      content.hasDataMessage() -> {
+        DataMessageProcessor.process(
+          context,
+          senderRecipient,
+          threadRecipient,
+          envelope,
+          content,
+          metadata,
+          receivedTime,
+          if (processingEarlyContent) null else EarlyMessageCacheEntry(envelope, content, metadata, serverDeliveredTimestamp)
+        )
+      }
+      content.hasSyncMessage() -> {
+        TextSecurePreferences.setMultiDevice(context, true)
+
+        SyncMessageProcessor.process(
+          context,
+          senderRecipient,
+          envelope,
+          content,
+          metadata,
+          if (processingEarlyContent) null else EarlyMessageCacheEntry(envelope, content, metadata, serverDeliveredTimestamp)
+        )
+      }
+      content.hasCallMessage() -> {
+        log(envelope.timestamp, "Got call message...")
+
+        val message: SignalServiceProtos.CallMessage = content.callMessage
+        val destinationDeviceId: Int? = if (message.hasDestinationDeviceId()) message.destinationDeviceId else null
+
+        if (destinationDeviceId != null && destinationDeviceId != SignalStore.account().deviceId) {
+          log(envelope.timestamp, "Ignoring call message that is not for this device! intended: $destinationDeviceId, this: ${SignalStore.account().deviceId}")
+          return
+        }
+
+        CallMessageProcessor.process(senderRecipient, envelope, content, metadata, serverDeliveredTimestamp)
+      }
+      content.hasReceiptMessage() -> {
+        ReceiptMessageProcessor.process(
+          context,
+          senderRecipient,
+          envelope,
+          content,
+          metadata,
+          if (processingEarlyContent) null else EarlyMessageCacheEntry(envelope, content, metadata, serverDeliveredTimestamp)
+        )
+      }
+      content.hasTypingMessage() -> {
+        handleTypingMessage(envelope, metadata, content.typingMessage, senderRecipient)
+      }
+      content.hasStoryMessage() -> {
+        StoryMessageProcessor.process(
+          envelope,
+          content,
+          metadata,
+          senderRecipient,
+          threadRecipient
+        )
+      }
+      content.hasDecryptionErrorMessage() -> {
+        handleRetryReceipt(envelope, metadata, content.decryptionErrorMessage!!.toDecryptionErrorMessage(metadata), senderRecipient)
+      }
+      content.hasSenderKeyDistributionMessage() || content.hasPniSignatureMessage() -> {
+        // Already handled, here in order to prevent unrecognized message log
+      }
+      else -> {
+        warn(envelope.timestamp, "Got unrecognized message!")
+      }
+    }
+
+    resetRecipientToPush(senderRecipient)
+
+    if (pending != null) {
+      warn(envelope.timestamp, "Pending retry was processed. Deleting.")
+      ApplicationDependencies.getPendingRetryReceiptCache().delete(pending)
+    }
+  }
+
+  @Throws(BadGroupIdException::class)
+  private fun handleTypingMessage(
+    envelope: Envelope,
+    metadata: EnvelopeMetadata,
+    typingMessage: TypingMessage,
+    senderRecipient: Recipient
+  ) {
+    if (!TextSecurePreferences.isTypingIndicatorsEnabled(context)) {
+      return
+    }
+
+    val threadId: Long = if (typingMessage.hasGroupId()) {
+      val groupId = GroupId.push(typingMessage.groupId)
+      if (!SignalDatabase.groups.isCurrentMember(groupId, senderRecipient.id)) {
+        warn(envelope.timestamp, "Seen typing indicator for non-member " + senderRecipient.id)
+        return
+      }
+
+      val groupRecipient = Recipient.externalPossiblyMigratedGroup(groupId)
+      SignalDatabase.threads.getOrCreateThreadIdFor(groupRecipient)
+    } else {
+      SignalDatabase.threads.getOrCreateThreadIdFor(senderRecipient)
+    }
+
+    if (threadId <= 0) {
+      warn(envelope.timestamp, "Couldn't find a matching thread for a typing message.")
+      return
+    }
+
+    if (typingMessage.hasStarted) {
+      Log.d(TAG, "Typing started on thread $threadId")
+      ApplicationDependencies.getTypingStatusRepository().onTypingStarted(context, threadId, senderRecipient, metadata.sourceDeviceId)
+    } else {
+      Log.d(TAG, "Typing stopped on thread $threadId")
+      ApplicationDependencies.getTypingStatusRepository().onTypingStopped(threadId, senderRecipient, metadata.sourceDeviceId, false)
+    }
+  }
+
+  private fun handleRetryReceipt(envelope: Envelope, metadata: EnvelopeMetadata, decryptionErrorMessage: DecryptionErrorMessage, senderRecipient: Recipient) {
+    if (!FeatureFlags.retryReceipts()) {
+      warn(envelope.timestamp, "[RetryReceipt] Feature flag disabled, skipping retry receipt.")
+      return
+    }
+
+    if (decryptionErrorMessage.deviceId != SignalStore.account().deviceId) {
+      log(envelope.timestamp, "[RetryReceipt] Received a DecryptionErrorMessage targeting a linked device. Ignoring.")
+      return
+    }
+
+    val sentTimestamp = decryptionErrorMessage.timestamp
+    warn(envelope.timestamp, "[RetryReceipt] Received a retry receipt from ${formatSender(senderRecipient.id, metadata.sourceServiceId, metadata.sourceDeviceId)} for message with timestamp $sentTimestamp.")
+    if (!senderRecipient.hasServiceId()) {
+      warn(envelope.timestamp, "[RetryReceipt] Requester ${senderRecipient.id} somehow has no UUID! timestamp: $sentTimestamp")
+      return
+    }
+
+    val messageLogEntry = SignalDatabase.messageLog.getLogEntry(senderRecipient.id, metadata.sourceDeviceId, sentTimestamp)
+    if (decryptionErrorMessage.ratchetKey.isPresent) {
+      handleIndividualRetryReceipt(senderRecipient, messageLogEntry, envelope, metadata, decryptionErrorMessage)
+    } else {
+      handleSenderKeyRetryReceipt(senderRecipient, messageLogEntry, envelope, metadata, decryptionErrorMessage)
+    }
+  }
+
+  private fun handleSenderKeyRetryReceipt(
+    requester: Recipient,
+    messageLogEntry: MessageLogEntry?,
+    envelope: Envelope,
+    metadata: EnvelopeMetadata,
+    decryptionErrorMessage: DecryptionErrorMessage
+  ) {
+    val sentTimestamp = decryptionErrorMessage.timestamp
+    val relatedMessage = findRetryReceiptRelatedMessage(messageLogEntry, sentTimestamp)
+
+    if (relatedMessage == null) {
+      warn(envelope.timestamp, "[RetryReceipt-SK] The related message could not be found! There shouldn't be any sender key resends where we can't find the related message. Skipping.")
+      return
+    }
+
+    val threadRecipient = SignalDatabase.threads.getRecipientForThreadId(relatedMessage.threadId)
+    if (threadRecipient == null) {
+      warn(envelope.timestamp, "[RetryReceipt-SK] Could not find a thread recipient! Skipping.")
+      return
+    }
+
+    if (!threadRecipient.isPushV2Group && !threadRecipient.isDistributionList) {
+      warn(envelope.timestamp, "[RetryReceipt-SK] Thread recipient is not a V2 group or distribution list! Skipping.")
+      return
+    }
+
+    val distributionId: DistributionId?
+    val groupId: GroupId.V2?
+
+    if (threadRecipient.isGroup) {
+      groupId = threadRecipient.requireGroupId().requireV2()
+      distributionId = SignalDatabase.groups.getOrCreateDistributionId(groupId)
+    } else {
+      groupId = null
+      distributionId = SignalDatabase.distributionLists.getDistributionId(threadRecipient.id)
+    }
+
+    if (distributionId == null) {
+      Log.w(TAG, "[RetryReceipt-SK] Failed to find a distributionId! Skipping.")
+      return
+    }
+
+    val requesterAddress = SignalProtocolAddress(requester.requireServiceId().toString(), metadata.sourceDeviceId)
+    SignalDatabase.senderKeyShared.delete(distributionId, setOf(requesterAddress))
+
+    if (messageLogEntry != null) {
+      warn(envelope.timestamp, "[RetryReceipt-SK] Found MSL entry for ${requester.id} ($requesterAddress) with timestamp $sentTimestamp. Scheduling a resend.")
+      ApplicationDependencies.getJobManager().add(
+        ResendMessageJob(
+          messageLogEntry.recipientId,
+          messageLogEntry.dateSent,
+          messageLogEntry.content,
+          messageLogEntry.contentHint,
+          messageLogEntry.urgent,
+          groupId,
+          distributionId
+        )
+      )
+    } else {
+      warn(envelope.timestamp, "[RetryReceipt-SK] Unable to find MSL entry for ${requester.id} ($requesterAddress) with timestamp $sentTimestamp for ${if (groupId != null) "group $groupId" else "distribution list"}. Scheduling a job to send them the SenderKeyDistributionMessage. Membership will be checked there.")
+      ApplicationDependencies.getJobManager().add(SenderKeyDistributionSendJob(requester.id, threadRecipient.id))
+    }
+  }
+
+  private fun handleIndividualRetryReceipt(requester: Recipient, messageLogEntry: MessageLogEntry?, envelope: Envelope, metadata: EnvelopeMetadata, decryptionErrorMessage: DecryptionErrorMessage) {
+    var archivedSession = false
+
+    // TODO [pnp] Ignore retry receipts that have a PNI destinationUuid
+    if (decryptionErrorMessage.ratchetKey.isPresent &&
+      ratchetKeyMatches(requester, metadata.sourceDeviceId, decryptionErrorMessage.ratchetKey.get())
+    ) {
+      warn(envelope.timestamp, "[RetryReceipt-I] Ratchet key matches. Archiving the session.")
+      ApplicationDependencies.getProtocolStore().aci().sessions().archiveSession(requester.id, metadata.sourceDeviceId)
+      archivedSession = true
+    }
+
+    if (messageLogEntry != null) {
+      warn(envelope.timestamp, "[RetryReceipt-I] Found an entry in the MSL. Resending.")
+      ApplicationDependencies.getJobManager().add(
+        ResendMessageJob(
+          messageLogEntry.recipientId,
+          messageLogEntry.dateSent,
+          messageLogEntry.content,
+          messageLogEntry.contentHint,
+          messageLogEntry.urgent,
+          null,
+          null
+        )
+      )
+    } else if (archivedSession) {
+      warn(envelope.timestamp, "[RetryReceipt-I] Could not find an entry in the MSL, but we archived the session, so we're sending a null message to complete the reset.")
+      ApplicationDependencies.getJobManager().add(NullMessageSendJob(requester.id))
+    } else {
+      warn(envelope.timestamp, "[RetryReceipt-I] Could not find an entry in the MSL. Skipping.")
+    }
+  }
+
+  private fun findRetryReceiptRelatedMessage(messageLogEntry: MessageLogEntry?, sentTimestamp: Long): MessageRecord? {
+    return if (messageLogEntry != null && messageLogEntry.hasRelatedMessage) {
+      val id = messageLogEntry.relatedMessages[0].id
+      SignalDatabase.messages.getMessageRecordOrNull(id)
+    } else {
+      SignalDatabase.messages.getMessageFor(sentTimestamp, Recipient.self().id)
+    }
+  }
+
+  private fun ratchetKeyMatches(recipient: Recipient, deviceId: Int, ratchetKey: ECPublicKey): Boolean {
+    val address = recipient.resolve().requireServiceId().toProtocolAddress(deviceId)
+    val session = ApplicationDependencies.getProtocolStore().aci().loadSession(address)
+    return session.currentRatchetKeyMatches(ratchetKey)
+  }
+}
@@ -439,7 +439,7 @@ object MessageDecryptor {
    val followUpOperations: List<FollowUpOperation>

    /** Successfully decrypted the envelope content. The plaintext [Content] is available. */
-    class Success(
+    data class Success(
      override val envelope: Envelope,
      override val serverDeliveredTimestamp: Long,
      val content: Content,
@@ -0,0 +1,138 @@
+package org.thoughtcrime.securesms.messages
+
+import android.annotation.SuppressLint
+import android.content.Context
+import org.thoughtcrime.securesms.database.MessageTable.SyncMessageId
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
+import org.thoughtcrime.securesms.jobs.PushProcessEarlyMessagesJob
+import org.thoughtcrime.securesms.keyvalue.SignalStore
+import org.thoughtcrime.securesms.messages.MessageContentProcessorV2.Companion.log
+import org.thoughtcrime.securesms.messages.MessageContentProcessorV2.Companion.warn
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.recipients.RecipientId
+import org.thoughtcrime.securesms.util.EarlyMessageCacheEntry
+import org.thoughtcrime.securesms.util.TextSecurePreferences
+import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.Envelope
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.ReceiptMessage
+
+object ReceiptMessageProcessor {
+  fun process(context: Context, senderRecipient: Recipient, envelope: Envelope, content: SignalServiceProtos.Content, metadata: EnvelopeMetadata, earlyMessageCacheEntry: EarlyMessageCacheEntry?) {
+    val receiptMessage = content.receiptMessage
+
+    when (receiptMessage.type) {
+      ReceiptMessage.Type.DELIVERY -> handleDeliveryReceipt(envelope, metadata, receiptMessage, senderRecipient.id)
+      ReceiptMessage.Type.READ -> handleReadReceipt(context, senderRecipient.id, envelope, metadata, receiptMessage, earlyMessageCacheEntry)
+      ReceiptMessage.Type.VIEWED -> handleViewedReceipt(context, envelope, metadata, receiptMessage, senderRecipient.id, earlyMessageCacheEntry)
+      else -> warn(envelope.timestamp, "Unknown recipient message type ${receiptMessage.type}")
+    }
+  }
+
+  @SuppressLint("DefaultLocale")
+  private fun handleDeliveryReceipt(
+    envelope: Envelope,
+    metadata: EnvelopeMetadata,
+    deliveryReceipt: ReceiptMessage,
+    senderRecipientId: RecipientId
+  ) {
+    log(envelope.timestamp, "Processing delivery receipts. Sender: $senderRecipientId, Device: ${metadata.sourceDeviceId}, Timestamps: ${deliveryReceipt.timestampList.joinToString(", ")}")
+
+    val ids = deliveryReceipt.timestampList.map { SyncMessageId(senderRecipientId, it) }
+    val unhandled = SignalDatabase.messages.incrementDeliveryReceiptCounts(ids, envelope.timestamp)
+
+    for (id in unhandled) {
+      warn(envelope.timestamp, "[handleDeliveryReceipt] Could not find matching message! timestamp: ${id.timetamp}  author: ${id.recipientId}")
+      // Early delivery receipts are special-cased in the database methods
+    }
+
+    if (unhandled.isNotEmpty()) {
+      PushProcessEarlyMessagesJob.enqueue()
+    }
+
+    SignalDatabase.pendingPniSignatureMessages.acknowledgeReceipts(senderRecipientId, deliveryReceipt.timestampList, metadata.sourceDeviceId)
+    SignalDatabase.messageLog.deleteEntriesForRecipient(deliveryReceipt.timestampList, senderRecipientId, metadata.sourceDeviceId)
+  }
+
+  @SuppressLint("DefaultLocale")
+  private fun handleReadReceipt(
+    context: Context,
+    senderRecipientId: RecipientId,
+    envelope: Envelope,
+    metadata: EnvelopeMetadata,
+    readReceipt: ReceiptMessage,
+    earlyMessageCacheEntry: EarlyMessageCacheEntry?
+  ) {
+    if (!TextSecurePreferences.isReadReceiptsEnabled(context)) {
+      log(envelope.timestamp, "Ignoring read receipts for IDs: " + readReceipt.timestampList.joinToString(", "))
+      return
+    }
+
+    log(envelope.timestamp, "Processing read receipts. Sender: $senderRecipientId, Device: ${metadata.sourceDeviceId}, Timestamps: ${readReceipt.timestampList.joinToString(", ")}")
+
+    val ids = readReceipt.timestampList.map { SyncMessageId(senderRecipientId, it) }
+    val unhandled = SignalDatabase.messages.incrementReadReceiptCounts(ids, envelope.timestamp)
+
+    if (unhandled.isNotEmpty()) {
+      val selfId = Recipient.self().id
+
+      for (id in unhandled) {
+        warn(envelope.timestamp, "[handleReadReceipt] Could not find matching message! timestamp: ${id.timetamp}, author: ${id.recipientId} | Receipt, so associating with message from self ($selfId)")
+        if (earlyMessageCacheEntry != null) {
+          ApplicationDependencies.getEarlyMessageCache().store(selfId, id.timetamp, earlyMessageCacheEntry)
+        }
+      }
+    }
+
+    if (unhandled.isNotEmpty() && earlyMessageCacheEntry != null) {
+      PushProcessEarlyMessagesJob.enqueue()
+    }
+  }
+
+  private fun handleViewedReceipt(
+    context: Context,
+    envelope: Envelope,
+    metadata: EnvelopeMetadata,
+    viewedReceipt: ReceiptMessage,
+    senderRecipientId: RecipientId,
+    earlyMessageCacheEntry: EarlyMessageCacheEntry?
+  ) {
+    val readReceipts = TextSecurePreferences.isReadReceiptsEnabled(context)
+    val storyViewedReceipts = SignalStore.storyValues().viewedReceiptsEnabled
+
+    if (!readReceipts && !storyViewedReceipts) {
+      log(envelope.timestamp, "Ignoring viewed receipts for IDs: ${viewedReceipt.timestampList.joinToString(", ")}")
+      return
+    }
+
+    log(envelope.timestamp, "Processing viewed receipts. Sender: $senderRecipientId, Device: ${metadata.sourceDeviceId}, Only Stories: ${!readReceipts}, Timestamps: ${viewedReceipt.timestampList.joinToString(", ")}")
+
+    val ids = viewedReceipt.timestampList.map { SyncMessageId(senderRecipientId, it) }
+
+    val unhandled: Collection<SyncMessageId> = if (readReceipts && storyViewedReceipts) {
+      SignalDatabase.messages.incrementViewedReceiptCounts(ids, envelope.timestamp)
+    } else if (readReceipts) {
+      SignalDatabase.messages.incrementViewedNonStoryReceiptCounts(ids, envelope.timestamp)
+    } else {
+      SignalDatabase.messages.incrementViewedStoryReceiptCounts(ids, envelope.timestamp)
+    }
+
+    val handled: Set<SyncMessageId> = ids.toSet() - unhandled.toSet()
+    SignalDatabase.messages.updateViewedStories(handled)
+
+    if (unhandled.isNotEmpty()) {
+      val selfId = Recipient.self().id
+      for (id in unhandled) {
+        warn(envelope.timestamp, "[handleViewedReceipt] Could not find matching message! timestamp: ${id.timetamp}, author: ${id.recipientId} | Receipt so associating with message from self ($selfId)")
+        if (earlyMessageCacheEntry != null) {
+          ApplicationDependencies.getEarlyMessageCache().store(selfId, id.timetamp, earlyMessageCacheEntry)
+        }
+      }
+    }
+
+    if (unhandled.isNotEmpty() && earlyMessageCacheEntry != null) {
+      PushProcessEarlyMessagesJob.enqueue()
+    }
+  }
+}
@@ -0,0 +1,180 @@
+package org.thoughtcrime.securesms.messages
+
+import com.google.protobuf.ByteString
+import org.signal.core.util.orNull
+import org.signal.libsignal.protocol.message.DecryptionErrorMessage
+import org.signal.libsignal.zkgroup.groups.GroupMasterKey
+import org.thoughtcrime.securesms.attachments.Attachment
+import org.thoughtcrime.securesms.attachments.PointerAttachment
+import org.thoughtcrime.securesms.database.model.StoryType
+import org.thoughtcrime.securesms.groups.GroupId
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.toPointer
+import org.thoughtcrime.securesms.stickers.StickerLocator
+import org.thoughtcrime.securesms.util.MediaUtil
+import org.whispersystems.signalservice.api.InvalidMessageStructureException
+import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
+import org.whispersystems.signalservice.api.messages.SignalServiceAttachmentPointer
+import org.whispersystems.signalservice.api.payments.Money
+import org.whispersystems.signalservice.api.push.ServiceId
+import org.whispersystems.signalservice.api.util.AttachmentPointerUtil
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.AttachmentPointer
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.DataMessage
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.DataMessage.Payment
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.GroupContextV2
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.StoryMessage
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.SyncMessage.Sent
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.TypingMessage
+import java.util.Optional
+
+private val ByteString.isNotEmpty: Boolean
+  get() = !this.isEmpty
+
+object SignalServiceProtoUtil {
+
+  /** Contains some user data that affects the conversation  */
+  val DataMessage.hasRenderableContent: Boolean
+    get() {
+      return attachmentsList.isNotEmpty() ||
+        hasBody() ||
+        hasQuote() ||
+        contactList.isNotEmpty() ||
+        previewList.isNotEmpty() ||
+        bodyRangesList.isNotEmpty() ||
+        hasSticker() ||
+        hasReaction() ||
+        hasRemoteDelete
+    }
+
+  val DataMessage.hasDisallowedAnnouncementOnlyContent: Boolean
+    get() {
+      return hasBody() ||
+        attachmentsList.isNotEmpty() ||
+        hasQuote() ||
+        previewList.isNotEmpty() ||
+        bodyRangesList.isNotEmpty() ||
+        hasSticker()
+    }
+
+  val DataMessage.isExpirationUpdate: Boolean
+    get() = flags and DataMessage.Flags.EXPIRATION_TIMER_UPDATE_VALUE != 0
+
+  val DataMessage.hasRemoteDelete: Boolean
+    get() = hasDelete() && delete.hasTargetSentTimestamp()
+
+  val DataMessage.isGroupV2Update: Boolean
+    get() = !hasRenderableContent && hasSignedGroupChange
+
+  val DataMessage.hasGroupContext: Boolean
+    get() = hasGroupV2() && groupV2.hasMasterKey() && groupV2.masterKey.isNotEmpty
+
+  val DataMessage.hasSignedGroupChange: Boolean
+    get() = hasGroupContext && groupV2.hasSignedGroupChange
+
+  val DataMessage.isMediaMessage: Boolean
+    get() = attachmentsList.isNotEmpty() || hasQuote() || contactList.isNotEmpty() || hasSticker() || bodyRangesList.isNotEmpty()
+
+  val DataMessage.isEndSession: Boolean
+    get() = flags and DataMessage.Flags.END_SESSION_VALUE != 0
+
+  val DataMessage.isStoryReaction: Boolean
+    get() = hasReaction() && hasStoryContext()
+
+  val DataMessage.isPaymentActivationRequest: Boolean
+    get() = hasPayment() && payment.hasActivation() && payment.activation.type == Payment.Activation.Type.REQUEST
+
+  val DataMessage.isPaymentActivated: Boolean
+    get() = hasPayment() && payment.hasActivation() && payment.activation.type == Payment.Activation.Type.ACTIVATED
+
+  val DataMessage.isInvalid: Boolean
+    get() {
+      if (isViewOnce) {
+        val contentType = attachmentsList[0].contentType.lowercase()
+        return attachmentsList.size != 1 || !MediaUtil.isImageOrVideoType(contentType)
+      }
+      return false
+    }
+
+  val DataMessage.isEmptyGroupV2Message: Boolean
+    get() = hasGroupContext && !isGroupV2Update && !hasRenderableContent
+
+  val GroupContextV2.hasSignedGroupChange: Boolean
+    get() = hasGroupChange() && groupChange.isNotEmpty
+
+  val GroupContextV2.signedGroupChange: ByteArray
+    get() = groupChange.toByteArray()
+
+  val GroupContextV2.groupMasterKey: GroupMasterKey
+    get() = GroupMasterKey(masterKey.toByteArray())
+
+  val GroupContextV2?.isValid: Boolean
+    get() = this != null && masterKey.isNotEmpty
+
+  val GroupContextV2.groupId: GroupId.V2?
+    get() = if (isValid) GroupId.v2(groupMasterKey) else null
+
+  val StoryMessage.type: StoryType
+    get() {
+      return if (allowsReplies) {
+        if (hasTextAttachment()) {
+          StoryType.TEXT_STORY_WITH_REPLIES
+        } else {
+          StoryType.STORY_WITH_REPLIES
+        }
+      } else {
+        if (hasTextAttachment()) {
+          StoryType.TEXT_STORY_WITHOUT_REPLIES
+        } else {
+          StoryType.STORY_WITHOUT_REPLIES
+        }
+      }
+    }
+
+  fun Sent.isUnidentified(serviceId: ServiceId?): Boolean {
+    return serviceId != null && unidentifiedStatusList.firstOrNull { ServiceId.parseOrNull(it.destinationUuid) == serviceId }?.unidentified ?: false
+  }
+
+  val Sent.serviceIdsToUnidentifiedStatus: Map<ServiceId, Boolean>
+    get() {
+      return unidentifiedStatusList
+        .mapNotNull { status ->
+          val serviceId = ServiceId.parseOrNull(status.destinationUuid)
+          if (serviceId != null) {
+            serviceId to status.unidentified
+          } else {
+            null
+          }
+        }
+        .toMap()
+    }
+
+  val TypingMessage.hasStarted: Boolean
+    get() = hasAction() && action == TypingMessage.Action.STARTED
+
+  fun ByteString.toDecryptionErrorMessage(metadata: EnvelopeMetadata): DecryptionErrorMessage {
+    try {
+      return DecryptionErrorMessage(toByteArray())
+    } catch (e: InvalidMessageStructureException) {
+      throw InvalidMessageStructureException(e, metadata.sourceServiceId.toString(), metadata.sourceDeviceId)
+    }
+  }
+
+  fun List<AttachmentPointer>.toPointers(): List<Attachment> {
+    return mapNotNull { it.toPointer() }
+  }
+
+  fun AttachmentPointer.toPointer(stickerLocator: StickerLocator? = null): Attachment? {
+    return try {
+      PointerAttachment.forPointer(Optional.of(toSignalServiceAttachmentPointer()), stickerLocator).orNull()
+    } catch (e: InvalidMessageStructureException) {
+      null
+    }
+  }
+
+  fun AttachmentPointer.toSignalServiceAttachmentPointer(): SignalServiceAttachmentPointer {
+    return AttachmentPointerUtil.createSignalAttachmentPointer(this)
+  }
+
+  fun Long.toMobileCoinMoney(): Money {
+    return Money.picoMobileCoin(this)
+  }
+}
@@ -0,0 +1,155 @@
+package org.thoughtcrime.securesms.messages
+
+import android.graphics.Color
+import org.signal.core.util.orNull
+import org.thoughtcrime.securesms.database.MessageTable.InsertResult
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.database.model.StoryType
+import org.thoughtcrime.securesms.database.model.databaseprotos.ChatColor
+import org.thoughtcrime.securesms.database.model.databaseprotos.StoryTextPost
+import org.thoughtcrime.securesms.database.model.toBodyRangeList
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
+import org.thoughtcrime.securesms.messages.MessageContentProcessorV2.Companion.log
+import org.thoughtcrime.securesms.messages.MessageContentProcessorV2.Companion.warn
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.groupId
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.toPointer
+import org.thoughtcrime.securesms.mms.IncomingMediaMessage
+import org.thoughtcrime.securesms.mms.MmsException
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.stories.Stories
+import org.thoughtcrime.securesms.util.Base64
+import org.thoughtcrime.securesms.util.FeatureFlags
+import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos
+
+object StoryMessageProcessor {
+
+  fun process(envelope: SignalServiceProtos.Envelope, content: SignalServiceProtos.Content, metadata: EnvelopeMetadata, senderRecipient: Recipient, threadRecipient: Recipient) {
+    val storyMessage = content.storyMessage
+
+    log(envelope.timestamp, "Story message.")
+
+    if (threadRecipient.isInactiveGroup) {
+      warn(envelope.timestamp, "Dropping a group story from a group we're no longer in.")
+      return
+    }
+
+    if (threadRecipient.isGroup && !SignalDatabase.groups.isCurrentMember(threadRecipient.requireGroupId().requirePush(), senderRecipient.id)) {
+      warn(envelope.timestamp, "Dropping a group story from a user who's no longer a member.")
+      return
+    }
+
+    if (!threadRecipient.isGroup && !(senderRecipient.isProfileSharing || senderRecipient.isSystemContact)) {
+      warn(envelope.timestamp, "Dropping story from an untrusted source.")
+      return
+    }
+
+    val insertResult: InsertResult?
+
+    SignalDatabase.messages.beginTransaction()
+
+    try {
+      val storyType: StoryType = if (storyMessage.hasAllowsReplies() && storyMessage.allowsReplies) {
+        StoryType.withReplies(storyMessage.hasTextAttachment())
+      } else {
+        StoryType.withoutReplies(storyMessage.hasTextAttachment())
+      }
+
+      val mediaMessage = IncomingMediaMessage(
+        from = senderRecipient.id,
+        sentTimeMillis = envelope.timestamp,
+        serverTimeMillis = envelope.serverTimestamp,
+        receivedTimeMillis = System.currentTimeMillis(),
+        storyType = storyType,
+        isUnidentified = metadata.sealedSender,
+        body = serializeTextAttachment(storyMessage),
+        groupId = storyMessage.group.groupId,
+        attachments = if (storyMessage.hasFileAttachment()) listOfNotNull(storyMessage.fileAttachment.toPointer()) else emptyList(),
+        linkPreviews = DataMessageProcessor.getLinkPreviews(
+          previews = if (storyMessage.textAttachment.hasPreview()) listOf(storyMessage.textAttachment.preview) else emptyList(),
+          body = "",
+          isStoryEmbed = true
+        ),
+        serverGuid = envelope.serverGuid,
+        messageRanges = storyMessage.bodyRangesList.filterNot { it.hasMentionUuid() }.toBodyRangeList()
+      )
+
+      insertResult = SignalDatabase.messages.insertSecureDecryptedMessageInbox(mediaMessage, -1).orNull()
+      if (insertResult != null) {
+        SignalDatabase.messages.setTransactionSuccessful()
+      }
+    } catch (e: MmsException) {
+      throw MessageContentProcessor.StorageFailedException(e, metadata.sourceServiceId.toString(), metadata.sourceDeviceId)
+    } finally {
+      SignalDatabase.messages.endTransaction()
+    }
+
+    if (insertResult != null) {
+      Stories.enqueueNextStoriesForDownload(threadRecipient.id, false, FeatureFlags.storiesAutoDownloadMaximum())
+      ApplicationDependencies.getExpireStoriesManager().scheduleIfNecessary()
+    }
+  }
+
+  fun serializeTextAttachment(story: SignalServiceProtos.StoryMessage): String? {
+    if (!story.hasTextAttachment()) {
+      return null
+    }
+    val textAttachment = story.textAttachment
+    val builder = StoryTextPost.newBuilder()
+
+    if (textAttachment.hasText()) {
+      builder.body = textAttachment.text
+    }
+
+    if (textAttachment.hasTextStyle()) {
+      when (textAttachment.textStyle) {
+        SignalServiceProtos.TextAttachment.Style.DEFAULT -> builder.style = StoryTextPost.Style.DEFAULT
+        SignalServiceProtos.TextAttachment.Style.REGULAR -> builder.style = StoryTextPost.Style.REGULAR
+        SignalServiceProtos.TextAttachment.Style.BOLD -> builder.style = StoryTextPost.Style.BOLD
+        SignalServiceProtos.TextAttachment.Style.SERIF -> builder.style = StoryTextPost.Style.SERIF
+        SignalServiceProtos.TextAttachment.Style.SCRIPT -> builder.style = StoryTextPost.Style.SCRIPT
+        SignalServiceProtos.TextAttachment.Style.CONDENSED -> builder.style = StoryTextPost.Style.CONDENSED
+        else -> Unit
+      }
+    }
+
+    if (textAttachment.hasTextBackgroundColor()) {
+      builder.textBackgroundColor = textAttachment.textBackgroundColor
+    }
+
+    if (textAttachment.hasTextForegroundColor()) {
+      builder.textForegroundColor = textAttachment.textForegroundColor
+    }
+
+    val chatColorBuilder = ChatColor.newBuilder()
+
+    if (textAttachment.hasColor()) {
+      chatColorBuilder.setSingleColor(ChatColor.SingleColor.newBuilder().setColor(textAttachment.color))
+    } else if (textAttachment.hasGradient()) {
+      val gradient = textAttachment.gradient
+      val linearGradientBuilder = ChatColor.LinearGradient.newBuilder()
+      linearGradientBuilder.rotation = gradient.angle.toFloat()
+
+      if (gradient.positionsList.size > 1 && gradient.colorsList.size == gradient.positionsList.size) {
+        val positions = ArrayList(gradient.positionsList)
+        positions[0] = 0f
+        positions[positions.size - 1] = 1f
+        linearGradientBuilder.addAllColors(ArrayList(gradient.colorsList))
+        linearGradientBuilder.addAllPositions(positions)
+      } else if (gradient.colorsList.isNotEmpty()) {
+        warn("Incoming text story has color / position mismatch. Defaulting to start and end colors.")
+        linearGradientBuilder.addColors(gradient.colorsList[0])
+        linearGradientBuilder.addColors(gradient.colorsList[gradient.colorsList.size - 1])
+        linearGradientBuilder.addAllPositions(listOf(0f, 1f))
+      } else {
+        warn("Incoming text story did not have a valid linear gradient.")
+        linearGradientBuilder.addAllColors(listOf(Color.BLACK, Color.BLACK))
+        linearGradientBuilder.addAllPositions(listOf(0f, 1f))
+      }
+      chatColorBuilder.setLinearGradient(linearGradientBuilder)
+    }
+    builder.setBackground(chatColorBuilder)
+
+    return Base64.encodeBytes(builder.build().toByteArray())
+  }
+}
@@ -5,6 +5,7 @@ import org.thoughtcrime.securesms.database.model.Mention
 import org.thoughtcrime.securesms.database.model.databaseprotos.BodyRangeList
 import org.thoughtcrime.securesms.recipients.RecipientId
 import org.whispersystems.signalservice.api.messages.SignalServiceDataMessage
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos

 class QuoteModel(
  val id: Long,
@@ -47,6 +48,14 @@ class QuoteModel(
        }
        return NORMAL
      }
+
+      fun fromProto(type: SignalServiceProtos.DataMessage.Quote.Type): Type {
+        return if (type == SignalServiceProtos.DataMessage.Quote.Type.GIFT_BADGE) {
+          GIFT_BADGE
+        } else {
+          NORMAL
+        }
+      }
    }
  }
 }
@@ -11,6 +11,7 @@ import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;

 /**
 * Sometimes a message that is referencing another message can arrive out of order. In these cases,
@@ -19,7 +20,8 @@ import java.util.Optional;
 */
 public final class EarlyMessageCache {

-  private final LRUCache<ServiceMessageId, List<SignalServiceContent>> cache = new LRUCache<>(100);
+  private final LRUCache<ServiceMessageId, List<SignalServiceContent>>   cache   = new LRUCache<>(100);
+  private final LRUCache<ServiceMessageId, List<EarlyMessageCacheEntry>> cacheV2 = new LRUCache<>(100);

  /**
   * @param targetSender        The sender of the message this message depends on.
@@ -38,8 +40,25 @@ public final class EarlyMessageCache {
    cache.put(messageId, contentList);
  }

+  public synchronized void store(@NonNull RecipientId targetSender,
+                                 long targetSentTimestamp,
+                                 @NonNull EarlyMessageCacheEntry cacheEntry)
+  {
+    ServiceMessageId             messageId    = new ServiceMessageId(targetSender, targetSentTimestamp);
+    List<EarlyMessageCacheEntry> envelopeList = cacheV2.get(messageId);
+
+    if (envelopeList == null) {
+      envelopeList = new LinkedList<>();
+    }
+
+    envelopeList.add(cacheEntry);
+
+    cacheV2.put(messageId, envelopeList);
+  }
+
  /**
   * Returns and removes any content that is dependent on the provided message id.
+   *
   * @param sender        The sender of the message in question.
   * @param sentTimestamp The sent timestamp of the message in question.
   */
@@ -47,11 +66,17 @@ public final class EarlyMessageCache {
    return Optional.ofNullable(cache.remove(new ServiceMessageId(sender, sentTimestamp)));
  }

+  public synchronized Optional<List<EarlyMessageCacheEntry>> retrieveV2(@NonNull RecipientId sender, long sentTimestamp) {
+    return Optional.ofNullable(cacheV2.remove(new ServiceMessageId(sender, sentTimestamp)));
+  }
+
  /**
   * Returns a collection of all of the {@link ServiceMessageId}s referenced in the cache at the moment of inquiry.
   * Caution: There is no guarantee that this list will be relevant for any amount of time afterwards.
   */
  public synchronized @NonNull Collection<ServiceMessageId> getAllReferencedIds() {
-    return new HashSet<>(cache.keySet());
+    Set<ServiceMessageId> allIds = new HashSet<>(cache.keySet());
+    allIds.addAll(cacheV2.keySet());
+    return allIds;
  }
 }
@@ -0,0 +1,16 @@
+package org.thoughtcrime.securesms.util
+
+import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.Content
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos.Envelope
+
+/**
+ * The tuple of information needed to process a message. Used to in [EarlyMessageCache]
+ * to store potentially out-of-order messages.
+ */
+data class EarlyMessageCacheEntry(
+  val envelope: Envelope,
+  val content: Content,
+  val metadata: EnvelopeMetadata,
+  val serverDeliveredTimestamp: Long
+)
@@ -6,7 +6,6 @@ import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;
 import androidx.annotation.WorkerThread;

-
 import org.signal.core.util.StringUtil;
 import org.signal.core.util.logging.Log;
 import org.signal.libsignal.zkgroup.InvalidInputException;
@@ -16,12 +15,14 @@ import org.thoughtcrime.securesms.database.GroupTable;
 import org.thoughtcrime.securesms.database.SignalDatabase;
 import org.thoughtcrime.securesms.database.model.GroupRecord;
 import org.thoughtcrime.securesms.groups.GroupId;
+import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil;
 import org.thoughtcrime.securesms.mms.MessageGroupContext;
 import org.thoughtcrime.securesms.recipients.Recipient;
 import org.thoughtcrime.securesms.recipients.RecipientId;
 import org.whispersystems.signalservice.api.messages.SignalServiceContent;
 import org.whispersystems.signalservice.api.messages.SignalServiceDataMessage;
 import org.whispersystems.signalservice.api.messages.SignalServiceGroupV2;
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos;

 import java.io.IOException;
 import java.util.List;
@@ -55,6 +56,23 @@ public final class GroupUtil {
    }
  }

+
+  public static @Nullable SignalServiceProtos.GroupContextV2 getGroupContextIfPresent(@NonNull SignalServiceProtos.Content content) {
+    if (content.hasDataMessage() && SignalServiceProtoUtil.INSTANCE.getHasGroupContext(content.getDataMessage())) {
+      return content.getDataMessage().getGroupV2();
+    } else if (content.hasSyncMessage()                 &&
+               content.getSyncMessage().hasSent() &&
+               content.getSyncMessage().getSent().hasMessage() &&
+               SignalServiceProtoUtil.INSTANCE.getHasGroupContext(content.getSyncMessage().getSent().getMessage()))
+    {
+      return content.getSyncMessage().getSent().getMessage().getGroupV2();
+    } else if (content.hasStoryMessage() && SignalServiceProtoUtil.INSTANCE.isValid(content.getStoryMessage().getGroup())) {
+      return content.getStoryMessage().getGroup();
+    } else {
+      return null;
+    }
+  }
+
  /**
   * Result may be a v1 or v2 GroupId.
   */
@@ -7,8 +7,10 @@ import androidx.annotation.Nullable;
 import androidx.annotation.StringRes;

 import org.signal.core.util.concurrent.SignalExecutors;
+import org.signal.core.util.concurrent.SimpleTask;
 import org.signal.core.util.logging.Log;
 import org.signal.libsignal.protocol.IdentityKey;
+import org.signal.libsignal.protocol.InvalidKeyException;
 import org.signal.libsignal.protocol.SignalProtocolAddress;
 import org.signal.libsignal.protocol.state.SessionRecord;
 import org.signal.libsignal.protocol.state.SessionStore;
@@ -23,7 +25,6 @@ import org.thoughtcrime.securesms.database.SignalDatabase;
 import org.thoughtcrime.securesms.database.model.GroupRecord;
 import org.thoughtcrime.securesms.database.model.IdentityRecord;
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
-import org.thoughtcrime.securesms.keyvalue.SignalStore;
 import org.thoughtcrime.securesms.mms.MmsException;
 import org.thoughtcrime.securesms.mms.OutgoingMessage;
 import org.thoughtcrime.securesms.notifications.v2.ConversationId;
@@ -35,10 +36,11 @@ import org.thoughtcrime.securesms.sms.IncomingIdentityVerifiedMessage;
 import org.thoughtcrime.securesms.sms.IncomingTextMessage;
 import org.thoughtcrime.securesms.util.concurrent.ListenableFuture;
 import org.thoughtcrime.securesms.util.concurrent.SettableFuture;
-import org.signal.core.util.concurrent.SimpleTask;
 import org.whispersystems.signalservice.api.SignalSessionLock;
 import org.whispersystems.signalservice.api.messages.multidevice.VerifiedMessage;
+import org.whispersystems.signalservice.api.push.ServiceId;
 import org.whispersystems.signalservice.api.push.SignalServiceAddress;
+import org.whispersystems.signalservice.internal.push.SignalServiceProtos;

 import java.util.List;
 import java.util.Optional;
@@ -175,6 +177,28 @@ public final class IdentityUtil {
    }
  }

+  public static void processVerifiedMessage(Context context, SignalServiceProtos.Verified verified) throws InvalidKeyException {
+    SignalServiceAddress          destination = new SignalServiceAddress(ServiceId.parseOrThrow(verified.getDestinationUuid()));
+    IdentityKey                   identityKey = new IdentityKey(verified.getIdentityKey().toByteArray(), 0);
+    VerifiedMessage.VerifiedState state;
+
+    switch (verified.getState()) {
+      case DEFAULT:
+        state = VerifiedMessage.VerifiedState.DEFAULT;
+        break;
+      case VERIFIED:
+        state = VerifiedMessage.VerifiedState.VERIFIED;
+        break;
+      case UNVERIFIED:
+        state = VerifiedMessage.VerifiedState.UNVERIFIED;
+        break;
+      default:
+        throw new IllegalArgumentException();
+    }
+
+    processVerifiedMessage(context, new VerifiedMessage(destination, identityKey, state, System.currentTimeMillis()));
+  }
+
  public static void processVerifiedMessage(Context context, VerifiedMessage verifiedMessage) {
    try(SignalSessionLock.Lock unused = ReentrantSessionLock.INSTANCE.acquire()) {
      SignalIdentityKeyStore   identityStore  = ApplicationDependencies.getProtocolStore().aci().identities();
@@ -6,6 +6,7 @@ import com.annimon.stream.Stream;

 import org.thoughtcrime.securesms.database.model.MessageRecord;
 import org.thoughtcrime.securesms.recipients.Recipient;
+import org.thoughtcrime.securesms.recipients.RecipientId;

 import java.util.Collection;
 import java.util.concurrent.TimeUnit;
@@ -18,18 +19,23 @@ public final class RemoteDeleteUtil {
  private RemoteDeleteUtil() {}

  public static boolean isValidReceive(@NonNull MessageRecord targetMessage, @NonNull Recipient deleteSender, long deleteServerTimestamp) {
-    boolean isValidIncomingOutgoing = (deleteSender.isSelf() && targetMessage.isOutgoing()) ||
-                                      (!deleteSender.isSelf() && !targetMessage.isOutgoing());
+    return isValidReceive(targetMessage, deleteSender.getId(), deleteServerTimestamp);
+  }

-    boolean isValidSender = targetMessage.getIndividualRecipient().equals(deleteSender) ||
-                            deleteSender.isSelf() && targetMessage.isOutgoing();
+  public static boolean isValidReceive(@NonNull MessageRecord targetMessage, @NonNull RecipientId deleteSenderId, long deleteServerTimestamp) {
+    boolean selfIsDeleteSender = isSelf(deleteSenderId);

-    long messageTimestamp = deleteSender.isSelf() && targetMessage.isOutgoing() ? targetMessage.getDateSent()
-                                                                                : targetMessage.getServerTimestamp();
+    boolean isValidIncomingOutgoing = (selfIsDeleteSender && targetMessage.isOutgoing()) ||
+                                      (!selfIsDeleteSender && !targetMessage.isOutgoing());
+
+    boolean isValidSender = targetMessage.getIndividualRecipient().getId().equals(deleteSenderId) || selfIsDeleteSender && targetMessage.isOutgoing();
+
+    long messageTimestamp = selfIsDeleteSender && targetMessage.isOutgoing() ? targetMessage.getDateSent()
+                                                                                 : targetMessage.getServerTimestamp();

    return isValidIncomingOutgoing &&
-           isValidSender           &&
-           (((deleteServerTimestamp - messageTimestamp) < RECEIVE_THRESHOLD) || (deleteSender.isSelf() && targetMessage.isOutgoing()));
+           isValidSender &&
+           (((deleteServerTimestamp - messageTimestamp) < RECEIVE_THRESHOLD) || (selfIsDeleteSender && targetMessage.isOutgoing()));
  }

  public static boolean isValidSend(@NonNull Collection<MessageRecord> targetMessages, long currentTime) {
@@ -38,13 +44,17 @@ public final class RemoteDeleteUtil {
  }

  private static boolean isValidSend(MessageRecord message, long currentTime) {
-    return !message.isUpdate()                                                           &&
-           message.isOutgoing()                                                          &&
-           message.isPush()                                                              &&
+    return !message.isUpdate() &&
+           message.isOutgoing() &&
+           message.isPush() &&
           (!message.getRecipient().isGroup() || message.getRecipient().isActiveGroup()) &&
-           !message.isRemoteDelete()                                                     &&
-           !MessageRecordUtil.hasGiftBadge(message)                                      &&
-           !message.isPaymentNotification()                                              &&
+           !message.isRemoteDelete() &&
+           !MessageRecordUtil.hasGiftBadge(message) &&
+           !message.isPaymentNotification() &&
           (((currentTime - message.getDateSent()) < SEND_THRESHOLD) || message.getRecipient().isSelf());
  }
+
+  private static boolean isSelf(@NonNull RecipientId recipientId) {
+    return Recipient.isSelfSet() && Recipient.self().getId().equals(recipientId);
+  }
 }
@@ -17,3 +17,7 @@ fun <E> Optional<E>.isAbsent(): Boolean {
 fun <E : Any> E?.toOptional(): Optional<E> {
  return Optional.ofNullable(this)
 }
+
+fun <E> Optional<E>.orNull(): E? {
+  return orElse(null)
+}
@@ -115,7 +115,7 @@ dependencyResolutionManagement {
            alias('jackson-core').to('com.fasterxml.jackson.core:jackson-databind:2.9.9.2')
            alias('jackson-module-kotlin').to('com.fasterxml.jackson.module:jackson-module-kotlin:2.12.0')
            alias('square-okhttp3').to('com.squareup.okhttp3:okhttp:3.12.13')
-            alias('square-okio').to('com.squareup.okio:okio:2.2.2')
+            alias('square-okio').to('com.squareup.okio:okio:3.0.0')
            alias('square-leakcanary').to('com.squareup.leakcanary:leakcanary-android:2.7')
            alias('rxjava3-rxjava').to('io.reactivex.rxjava3:rxjava:3.0.13')
            alias('rxjava3-rxandroid').to('io.reactivex.rxjava3:rxandroid:3.0.0')
@@ -6,6 +6,7 @@ apply plugin: 'maven-publish'
 apply plugin: 'signing'
 apply plugin: 'idea'
 apply plugin: 'org.jlleitschuh.gradle.ktlint'
+apply plugin: 'com.squareup.wire'

 archivesBaseName = "signal-service-java"
 version          = lib_signal_service_version_number
@@ -30,6 +31,17 @@ configurations {
    ideaTestFixturesImplementation { extendsFrom testFixturesImplementation; canBeConsumed false; canBeResolved true }
 }

+ktlint {
+    // Use a newer version to resolve https://github.com/JLLeitschuh/ktlint-gradle/issues/507
+    version.set("0.47.1")
+
+    filter {
+        exclude { entry ->
+            entry.file.toString().contains("build/generated/source/wire")
+        }
+    }
+}
+
 dependencies {
    implementation libs.google.protobuf.javalite
    api libs.google.libphonenumber
@@ -75,6 +87,16 @@ protobuf {
    }
 }

+wire {
+    kotlin {
+        javaInterop = true
+    }
+
+    sourcePath {
+        srcDir 'src/main/protowire'
+    }
+}
+
 idea {
    module {
        generatedSourceDirs += file("${protobuf.generatedFilesBaseDir}/main/java")
@@ -24,7 +24,7 @@ class AccountAttributes @JsonCreator constructor(
  @JsonProperty val capabilities: Capabilities?,
  @JsonProperty val name: String?,
  @JsonProperty val pniRegistrationId: Int,
-  @JsonProperty val recoveryPassword: String?,
+  @JsonProperty val recoveryPassword: String?
 ) {
  constructor(
    signalingKey: String?,
@@ -1339,7 +1339,7 @@ import javax.annotation.Nullable;
    return new SignalServiceDataMessage.PaymentActivation(payment.getType());
  }

-  private static @Nullable List<SharedContact> createSharedContacts(SignalServiceProtos.DataMessage content) throws InvalidMessageStructureException {
+  public static @Nullable List<SharedContact> createSharedContacts(SignalServiceProtos.DataMessage content) throws InvalidMessageStructureException {
    if (content.getContactCount() <= 0) return null;

    List<SharedContact> results = new LinkedList<>();
@@ -11,6 +11,8 @@ import java.util.Objects;
 import java.util.UUID;
 import java.util.stream.Collectors;

+import javax.annotation.Nullable;
+
 /**
 * A wrapper around a UUID that represents an identifier for an account. Today, that is either an {@link ACI} or a {@link PNI}.
 * However, that doesn't mean every {@link ServiceId} is an <em>instance</em> of one of those classes. In reality, we often
@@ -41,7 +43,7 @@ public class ServiceId {
    return from(UuidUtil.parseOrThrow(raw));
  }

-  public static ServiceId parseOrNull(String raw) {
+  public static @Nullable ServiceId parseOrNull(String raw) {
    UUID uuid = UuidUtil.parseOrNull(raw);
    return uuid != null ? from(uuid) : null;
  }
@@ -5,6 +5,8 @@ import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;

+import javax.annotation.Nonnull;
+
 import okio.Buffer;
 import okio.BufferedSink;
 import okio.ByteString;
@@ -117,6 +119,11 @@ public class NowhereBufferedSink implements BufferedSink {
    return this;
  }

+  @Override
+  public BufferedSink write(@Nonnull ByteString byteString, int i, int i1) throws IOException {
+    return this;
+  }
+
  @Override
  public void write(Buffer source, long byteCount) throws IOException {

@@ -9,7 +9,7 @@ import com.fasterxml.jackson.annotation.JsonProperty
 data class RegistrationSessionMetadataResponse(
  val headers: RegistrationSessionMetadataHeaders,
  val body: RegistrationSessionMetadataJson,
-  val state: RegistrationSessionState?,
+  val state: RegistrationSessionState?
 )

 data class RegistrationSessionMetadataHeaders(
@@ -23,7 +23,7 @@ data class RegistrationSessionMetadataJson(
  @JsonProperty("nextVerificationAttempt") val nextVerificationAttempt: Int?,
  @JsonProperty("allowedToRequestCode") val allowedToRequestCode: Boolean,
  @JsonProperty("requestedInformation") val requestedInformation: List<String>,
-  @JsonProperty("verified") val verified: Boolean,
+  @JsonProperty("verified") val verified: Boolean
 ) {
  fun pushChallengedRequired(): Boolean {
    return requestedInformation.contains("pushChallenge")
@@ -35,5 +35,5 @@ data class RegistrationSessionMetadataJson(
 }

 data class RegistrationSessionState(
-  var pushChallengeTimedOut: Boolean,
+  var pushChallengeTimedOut: Boolean
 )
@@ -9,7 +9,7 @@ data class UpdateVerificationSessionRequestBody(
  @JsonProperty val pushToken: String?,
  @JsonProperty val pushChallenge: String?,
  @JsonProperty val mcc: String?,
-  @JsonProperty val mnc: String?,
+  @JsonProperty val mnc: String?
 ) {
  @JsonProperty
  val pushTokenType: String? = if (pushToken != null) "fcm" else null
@@ -8,7 +8,7 @@ data class VerificationSessionMetadataRequestBody(
  @JsonProperty val number: String,
  @JsonProperty val pushToken: String?,
  @JsonProperty val mcc: String?,
-  @JsonProperty val mnc: String?,
+  @JsonProperty val mnc: String?
 ) {
  @JsonProperty
  val pushTokenType: String? = if (pushToken != null) "fcm" else null
@@ -0,0 +1,27 @@
+/**
+ * Copyright (C) 2014-2016 Open Whisper Systems
+ *
+ * Licensed according to the LICENSE file in this repository.
+ */
+
+syntax = "proto2";
+
+package signalservice;
+
+option java_package = "org.whispersystems.signalservice.api.crypto.protos";
+
+message EnvelopeMetadata {
+  required bytes  sourceServiceId      = 1;
+  optional string sourceE164           = 2;
+  required int32  sourceDeviceId       = 3;
+  required bool   sealedSender         = 4;
+  optional bytes  groupId              = 5;
+  required bytes  destinationServiceId = 6;
+}
+
+message CompleteMessage {
+  required bytes            envelope                 = 1;
+  required bytes            content                  = 2;
+  required EnvelopeMetadata metadata                 = 3;
+  required int64            serverDeliveredTimestamp = 4;
+}