Setup backupV2 infrastructure and testing.

Co-authored-by: Clark Chen <clark@signal.org>

app/src/main/java/org/thoughtcrime/securesms/backup/v2/BackupRepository.kt

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2
+
+import org.signal.core.util.EventTimer
+import org.signal.core.util.logging.Log
+import org.signal.core.util.withinTransaction
+import org.thoughtcrime.securesms.backup.v2.database.ChatItemImportInserter
+import org.thoughtcrime.securesms.backup.v2.database.clearAllDataForBackupRestore
+import org.thoughtcrime.securesms.backup.v2.processor.AccountDataProcessor
+import org.thoughtcrime.securesms.backup.v2.processor.ChatBackupProcessor
+import org.thoughtcrime.securesms.backup.v2.processor.ChatItemBackupProcessor
+import org.thoughtcrime.securesms.backup.v2.processor.RecipientBackupProcessor
+import org.thoughtcrime.securesms.backup.v2.stream.BackupExportStream
+import org.thoughtcrime.securesms.backup.v2.stream.PlainTextBackupExportStream
+import org.thoughtcrime.securesms.backup.v2.stream.PlainTextBackupImportStream
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.keyvalue.SignalStore
+import org.thoughtcrime.securesms.recipients.RecipientId
+import java.io.ByteArrayInputStream
+import java.io.ByteArrayOutputStream
+
+object BackupRepository {
+
+  private val TAG = Log.tag(BackupRepository::class.java)
+
+  fun export(): ByteArray {
+    val eventTimer = EventTimer()
+
+    val outputStream = ByteArrayOutputStream()
+    val writer: BackupExportStream = PlainTextBackupExportStream(outputStream)
+
+    // Note: Without a transaction, we may export inconsistent state. But because we have a transaction,
+    // writes from other threads are blocked. This is something to think more about.
+    SignalDatabase.rawDatabase.withinTransaction {
+      AccountDataProcessor.export {
+        writer.write(it)
+        eventTimer.emit("account")
+      }
+
+      RecipientBackupProcessor.export {
+        writer.write(it)
+        eventTimer.emit("recipient")
+      }
+
+      ChatBackupProcessor.export { frame ->
+        writer.write(frame)
+        eventTimer.emit("thread")
+      }
+
+      ChatItemBackupProcessor.export { frame ->
+        writer.write(frame)
+        eventTimer.emit("message")
+      }
+    }
+
+    Log.d(TAG, "export() ${eventTimer.stop().summary}")
+
+    return outputStream.toByteArray()
+  }
+
+  fun import(data: ByteArray) {
+    val eventTimer = EventTimer()
+
+    val stream = ByteArrayInputStream(data)
+    val frameReader = PlainTextBackupImportStream(stream)
+
+    // Note: Without a transaction, bad imports could lead to lost data. But because we have a transaction,
+    // writes from other threads are blocked. This is something to think more about.
+    SignalDatabase.rawDatabase.withinTransaction {
+      SignalStore.clearAllDataForBackupRestore()
+      SignalDatabase.recipients.clearAllDataForBackupRestore()
+      SignalDatabase.distributionLists.clearAllDataForBackupRestore()
+      SignalDatabase.threads.clearAllDataForBackupRestore()
+      SignalDatabase.messages.clearAllDataForBackupRestore()
+
+      val backupState = BackupState()
+      val chatItemInserter: ChatItemImportInserter = ChatItemBackupProcessor.beginImport(backupState)
+
+      for (frame in frameReader) {
+        when {
+          frame.account != null -> {
+            AccountDataProcessor.import(frame.account)
+            eventTimer.emit("account")
+          }
+
+          frame.recipient != null -> {
+            RecipientBackupProcessor.import(frame.recipient, backupState)
+            eventTimer.emit("recipient")
+          }
+
+          frame.chat != null -> {
+            ChatBackupProcessor.import(frame.chat, backupState)
+            eventTimer.emit("chat")
+          }
+
+          frame.chatItem != null -> {
+            chatItemInserter.insert(frame.chatItem)
+            eventTimer.emit("chatItem")
+            // TODO if there's stuff in the stream after chatItems, we need to flush the inserter before going to the next phase
+          }
+
+          else -> Log.w(TAG, "Unrecognized frame")
+        }
+      }
+
+      if (chatItemInserter.flush()) {
+        eventTimer.emit("chatItem")
+      }
+
+      backupState.chatIdToLocalThreadId.values.forEach {
+        SignalDatabase.threads.update(it, unarchive = false, allowDeletion = false)
+      }
+    }
+
+    Log.d(TAG, "import() ${eventTimer.stop().summary}")
+  }
+}
+
+class BackupState {
+  val backupToLocalRecipientId = HashMap<Long, RecipientId>()
+  val chatIdToLocalThreadId = HashMap<Long, Long>()
+  val chatIdToLocalRecipientId = HashMap<Long, RecipientId>()
+  val chatIdToBackupRecipientId = HashMap<Long, Long>()
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/database/ChatItemExportIterator.kt

@@ -0,0 +1,349 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.database
+
+import android.database.Cursor
+import org.signal.core.util.logging.Log
+import org.signal.core.util.requireBlob
+import org.signal.core.util.requireBoolean
+import org.signal.core.util.requireInt
+import org.signal.core.util.requireLong
+import org.signal.core.util.requireString
+import org.thoughtcrime.securesms.backup.v2.proto.ChatItem
+import org.thoughtcrime.securesms.backup.v2.proto.Quote
+import org.thoughtcrime.securesms.backup.v2.proto.Reaction
+import org.thoughtcrime.securesms.backup.v2.proto.RemoteDeletedMessage
+import org.thoughtcrime.securesms.backup.v2.proto.SendStatus
+import org.thoughtcrime.securesms.backup.v2.proto.StandardMessage
+import org.thoughtcrime.securesms.backup.v2.proto.Text
+import org.thoughtcrime.securesms.database.GroupReceiptTable
+import org.thoughtcrime.securesms.database.MessageTable
+import org.thoughtcrime.securesms.database.MessageTypes
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.database.documents.IdentityKeyMismatchSet
+import org.thoughtcrime.securesms.database.documents.NetworkFailureSet
+import org.thoughtcrime.securesms.database.model.ReactionRecord
+import org.thoughtcrime.securesms.database.model.databaseprotos.BodyRangeList
+import org.thoughtcrime.securesms.mms.QuoteModel
+import org.thoughtcrime.securesms.util.JsonUtils
+import java.io.Closeable
+import java.io.IOException
+import java.util.LinkedList
+import java.util.Queue
+import org.thoughtcrime.securesms.backup.v2.proto.BodyRange as BackupBodyRange
+
+/**
+ * An iterator for chat items with a clever performance twist: rather than do the extra queries one at a time (for reactions,
+ * attachments, etc), this will populate items in batches, doing bulk lookups to improve throughput. We keep these in a buffer
+ * and only do more queries when the buffer is empty.
+ *
+ * All of this complexity is hidden from the user -- they just get a normal iterator interface.
+ */
+class ChatItemExportIterator(private val cursor: Cursor, private val batchSize: Int) : Iterator<ChatItem>, Closeable {
+
+  companion object {
+    private val TAG = Log.tag(ChatItemExportIterator::class.java)
+
+    const val COLUMN_BASE_TYPE = "base_type"
+  }
+
+  /**
+   * A queue of already-parsed ChatItems. Processing in batches means that we read ahead in the cursor and put
+   * the pending items here.
+   */
+  private val buffer: Queue<ChatItem> = LinkedList()
+
+  override fun hasNext(): Boolean {
+    return buffer.isNotEmpty() || (cursor.count > 0 && !cursor.isLast && !cursor.isAfterLast)
+  }
+
+  override fun next(): ChatItem {
+    if (buffer.isNotEmpty()) {
+      return buffer.remove()
+    }
+
+    val records: LinkedHashMap<Long, BackupMessageRecord> = linkedMapOf()
+
+    for (i in 0 until batchSize) {
+      if (cursor.moveToNext()) {
+        val record = cursor.toBackupMessageRecord()
+        records[record.id] = record
+      } else {
+        break
+      }
+    }
+
+    val reactionsById: Map<Long, List<ReactionRecord>> = SignalDatabase.reactions.getReactionsForMessages(records.keys)
+    val groupReceiptsById: Map<Long, List<GroupReceiptTable.GroupReceiptInfo>> = SignalDatabase.groupReceipts.getGroupReceiptInfoForMessages(records.keys)
+
+    for ((id, record) in records) {
+      val builder = record.toBasicChatItemBuilder(groupReceiptsById[id])
+
+      when {
+        record.remoteDeleted -> builder.remoteDeletedMessage = RemoteDeletedMessage()
+        else -> builder.standardMessage = record.toTextMessage(reactionsById[id])
+      }
+
+      buffer += builder.build()
+    }
+
+    return if (buffer.isNotEmpty()) {
+      buffer.remove()
+    } else {
+      throw NoSuchElementException()
+    }
+  }
+
+  override fun close() {
+    cursor.close()
+  }
+
+  private fun BackupMessageRecord.toBasicChatItemBuilder(groupReceipts: List<GroupReceiptTable.GroupReceiptInfo>?): ChatItem.Builder {
+    val record = this
+
+    return ChatItem.Builder().apply {
+      chatId = record.threadId
+      authorId = record.fromRecipientId
+      dateSent = record.dateSent
+      dateReceived = record.dateReceived
+      expireStart = if (record.expireStarted > 0) record.expireStarted else null
+      expiresIn = if (record.expiresIn > 0) record.expiresIn else null
+      revisions = emptyList()
+      sms = !MessageTypes.isSecureType(record.type)
+
+      if (MessageTypes.isOutgoingMessageType(record.type)) {
+        outgoing = ChatItem.OutgoingMessageDetails(
+          sendStatus = record.toBackupSendStatus(groupReceipts)
+        )
+      } else {
+        incoming = ChatItem.IncomingMessageDetails(
+          dateServerSent = record.dateServer,
+          sealedSender = record.sealedSender,
+          read = record.read
+        )
+      }
+    }
+  }
+
+  private fun BackupMessageRecord.toTextMessage(reactionRecords: List<ReactionRecord>?): StandardMessage {
+    return StandardMessage(
+      quote = this.toQuote(),
+      text = Text(
+        body = this.body!!,
+        bodyRanges = this.bodyRanges?.toBackupBodyRanges() ?: emptyList()
+      ),
+      linkPreview = null,
+      longText = null,
+      reactions = reactionRecords.toBackupReactions()
+    )
+  }
+
+  private fun BackupMessageRecord.toQuote(): Quote? {
+    return if (this.quoteTargetSentTimestamp > 0) {
+      // TODO Attachments!
+      val type = QuoteModel.Type.fromCode(this.quoteType)
+      Quote(
+        targetSentTimestamp = this.quoteTargetSentTimestamp,
+        authorId = this.quoteAuthor,
+        text = this.quoteBody,
+        originalMessageMissing = this.quoteMissing,
+        bodyRanges = this.quoteBodyRanges?.toBackupBodyRanges() ?: emptyList(),
+        type = when (type) {
+          QuoteModel.Type.NORMAL -> Quote.Type.NORMAL
+          QuoteModel.Type.GIFT_BADGE -> Quote.Type.GIFTBADGE
+        }
+      )
+    } else {
+      null
+    }
+  }
+
+  private fun ByteArray.toBackupBodyRanges(): List<BackupBodyRange> {
+    val decoded: BodyRangeList = try {
+      BodyRangeList.ADAPTER.decode(this)
+    } catch (e: IOException) {
+      Log.w(TAG, "Failed to decode BodyRangeList!")
+      return emptyList()
+    }
+
+    return decoded.ranges.map {
+      BackupBodyRange(
+        start = it.start,
+        length = it.length,
+        mentionAci = it.mentionUuid,
+        style = it.style?.toBackupBodyRangeStyle()
+      )
+    }
+  }
+
+  private fun BodyRangeList.BodyRange.Style.toBackupBodyRangeStyle(): BackupBodyRange.Style {
+    return when (this) {
+      BodyRangeList.BodyRange.Style.BOLD -> BackupBodyRange.Style.BOLD
+      BodyRangeList.BodyRange.Style.ITALIC -> BackupBodyRange.Style.ITALIC
+      BodyRangeList.BodyRange.Style.STRIKETHROUGH -> BackupBodyRange.Style.STRIKETHROUGH
+      BodyRangeList.BodyRange.Style.MONOSPACE -> BackupBodyRange.Style.MONOSPACE
+      BodyRangeList.BodyRange.Style.SPOILER -> BackupBodyRange.Style.SPOILER
+    }
+  }
+
+  private fun List<ReactionRecord>?.toBackupReactions(): List<Reaction> {
+    return this
+      ?.map {
+        Reaction(
+          emoji = it.emoji,
+          authorId = it.author.toLong(),
+          sentTimestamp = it.dateSent,
+          receivedTimestamp = it.dateReceived
+        )
+      } ?: emptyList()
+  }
+
+  private fun BackupMessageRecord.toBackupSendStatus(groupReceipts: List<GroupReceiptTable.GroupReceiptInfo>?): List<SendStatus> {
+    if (!MessageTypes.isOutgoingMessageType(this.type)) {
+      return emptyList()
+    }
+
+    if (!groupReceipts.isNullOrEmpty()) {
+      return groupReceipts.toBackupSendStatus(this.networkFailureRecipientIds, this.identityMismatchRecipientIds)
+    }
+
+    val status: SendStatus.Status = when {
+      this.viewedReceiptCount > 0 -> SendStatus.Status.VIEWED
+      this.readReceiptCount > 0 -> SendStatus.Status.READ
+      this.deliveryReceiptCount > 0 -> SendStatus.Status.DELIVERED
+      this.baseType == MessageTypes.BASE_SENT_TYPE -> SendStatus.Status.SENT
+      MessageTypes.isFailedMessageType(this.type) -> SendStatus.Status.FAILED
+      else -> SendStatus.Status.PENDING
+    }
+
+    return listOf(
+      SendStatus(
+        recipientId = this.toRecipientId,
+        deliveryStatus = status,
+        timestamp = this.receiptTimestamp,
+        sealedSender = this.sealedSender,
+        networkFailure = this.networkFailureRecipientIds.contains(this.toRecipientId),
+        identityKeyMismatch = this.identityMismatchRecipientIds.contains(this.toRecipientId)
+      )
+    )
+  }
+
+  private fun List<GroupReceiptTable.GroupReceiptInfo>.toBackupSendStatus(networkFailureRecipientIds: Set<Long>, identityMismatchRecipientIds: Set<Long>): List<SendStatus> {
+    return this.map {
+      SendStatus(
+        recipientId = it.recipientId.toLong(),
+        deliveryStatus = it.status.toBackupDeliveryStatus(),
+        sealedSender = it.isUnidentified,
+        timestamp = it.timestamp,
+        networkFailure = networkFailureRecipientIds.contains(it.recipientId.toLong()),
+        identityKeyMismatch = identityMismatchRecipientIds.contains(it.recipientId.toLong())
+      )
+    }
+  }
+
+  private fun Int.toBackupDeliveryStatus(): SendStatus.Status {
+    return when (this) {
+      GroupReceiptTable.STATUS_UNDELIVERED -> SendStatus.Status.PENDING
+      GroupReceiptTable.STATUS_DELIVERED -> SendStatus.Status.DELIVERED
+      GroupReceiptTable.STATUS_READ -> SendStatus.Status.READ
+      GroupReceiptTable.STATUS_VIEWED -> SendStatus.Status.VIEWED
+      GroupReceiptTable.STATUS_SKIPPED -> SendStatus.Status.SKIPPED
+      else -> SendStatus.Status.SKIPPED
+    }
+  }
+
+  private fun String?.parseNetworkFailures(): Set<Long> {
+    if (this.isNullOrBlank()) {
+      return emptySet()
+    }
+
+    return try {
+      JsonUtils.fromJson(this, NetworkFailureSet::class.java).items.map { it.recipientId.toLong() }.toSet()
+    } catch (e: IOException) {
+      emptySet()
+    }
+  }
+
+  private fun String?.parseIdentityMismatches(): Set<Long> {
+    if (this.isNullOrBlank()) {
+      return emptySet()
+    }
+
+    return try {
+      JsonUtils.fromJson(this, IdentityKeyMismatchSet::class.java).items.map { it.recipientId.toLong() }.toSet()
+    } catch (e: IOException) {
+      emptySet()
+    }
+  }
+
+  private fun Cursor.toBackupMessageRecord(): BackupMessageRecord {
+    return BackupMessageRecord(
+      id = this.requireLong(MessageTable.ID),
+      dateSent = this.requireLong(MessageTable.DATE_SENT),
+      dateReceived = this.requireLong(MessageTable.DATE_RECEIVED),
+      dateServer = this.requireLong(MessageTable.DATE_SERVER),
+      type = this.requireLong(MessageTable.TYPE),
+      threadId = this.requireLong(MessageTable.THREAD_ID),
+      body = this.requireString(MessageTable.BODY),
+      bodyRanges = this.requireBlob(MessageTable.MESSAGE_RANGES),
+      fromRecipientId = this.requireLong(MessageTable.FROM_RECIPIENT_ID),
+      toRecipientId = this.requireLong(MessageTable.TO_RECIPIENT_ID),
+      expiresIn = this.requireLong(MessageTable.EXPIRES_IN),
+      expireStarted = this.requireLong(MessageTable.EXPIRE_STARTED),
+      remoteDeleted = this.requireBoolean(MessageTable.REMOTE_DELETED),
+      sealedSender = this.requireBoolean(MessageTable.UNIDENTIFIED),
+      quoteTargetSentTimestamp = this.requireLong(MessageTable.QUOTE_ID),
+      quoteAuthor = this.requireLong(MessageTable.QUOTE_AUTHOR),
+      quoteBody = this.requireString(MessageTable.QUOTE_BODY),
+      quoteMissing = this.requireBoolean(MessageTable.QUOTE_MISSING),
+      quoteBodyRanges = this.requireBlob(MessageTable.QUOTE_BODY_RANGES),
+      quoteType = this.requireInt(MessageTable.QUOTE_TYPE),
+      originalMessageId = this.requireLong(MessageTable.ORIGINAL_MESSAGE_ID),
+      latestRevisionId = this.requireLong(MessageTable.LATEST_REVISION_ID),
+      deliveryReceiptCount = this.requireInt(MessageTable.DELIVERY_RECEIPT_COUNT),
+      viewedReceiptCount = this.requireInt(MessageTable.VIEWED_RECEIPT_COUNT),
+      readReceiptCount = this.requireInt(MessageTable.READ_RECEIPT_COUNT),
+      read = this.requireBoolean(MessageTable.READ),
+      receiptTimestamp = this.requireLong(MessageTable.RECEIPT_TIMESTAMP),
+      networkFailureRecipientIds = this.requireString(MessageTable.NETWORK_FAILURES).parseNetworkFailures(),
+      identityMismatchRecipientIds = this.requireString(MessageTable.MISMATCHED_IDENTITIES).parseIdentityMismatches(),
+      baseType = this.requireLong(COLUMN_BASE_TYPE)
+    )
+  }
+
+  private class BackupMessageRecord(
+    val id: Long,
+    val dateSent: Long,
+    val dateReceived: Long,
+    val dateServer: Long,
+    val type: Long,
+    val threadId: Long,
+    val body: String?,
+    val bodyRanges: ByteArray?,
+    val fromRecipientId: Long,
+    val toRecipientId: Long,
+    val expiresIn: Long,
+    val expireStarted: Long,
+    val remoteDeleted: Boolean,
+    val sealedSender: Boolean,
+    val quoteTargetSentTimestamp: Long,
+    val quoteAuthor: Long,
+    val quoteBody: String?,
+    val quoteMissing: Boolean,
+    val quoteBodyRanges: ByteArray?,
+    val quoteType: Int,
+    val originalMessageId: Long,
+    val latestRevisionId: Long,
+    val deliveryReceiptCount: Int,
+    val readReceiptCount: Int,
+    val viewedReceiptCount: Int,
+    val receiptTimestamp: Long,
+    val read: Boolean,
+    val networkFailureRecipientIds: Set<Long>,
+    val identityMismatchRecipientIds: Set<Long>,
+    val baseType: Long
+  )
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/database/ChatItemImportInserter.kt

@@ -0,0 +1,402 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.database
+
+import android.content.ContentValues
+import androidx.core.content.contentValuesOf
+import org.signal.core.util.SqlUtil
+import org.signal.core.util.logging.Log
+import org.signal.core.util.toInt
+import org.thoughtcrime.securesms.backup.v2.BackupState
+import org.thoughtcrime.securesms.backup.v2.proto.BodyRange
+import org.thoughtcrime.securesms.backup.v2.proto.ChatItem
+import org.thoughtcrime.securesms.backup.v2.proto.Quote
+import org.thoughtcrime.securesms.backup.v2.proto.Reaction
+import org.thoughtcrime.securesms.backup.v2.proto.SendStatus
+import org.thoughtcrime.securesms.backup.v2.proto.StandardMessage
+import org.thoughtcrime.securesms.database.GroupReceiptTable
+import org.thoughtcrime.securesms.database.MessageTable
+import org.thoughtcrime.securesms.database.MessageTypes
+import org.thoughtcrime.securesms.database.ReactionTable
+import org.thoughtcrime.securesms.database.SQLiteDatabase
+import org.thoughtcrime.securesms.database.documents.IdentityKeyMismatch
+import org.thoughtcrime.securesms.database.documents.IdentityKeyMismatchSet
+import org.thoughtcrime.securesms.database.documents.NetworkFailure
+import org.thoughtcrime.securesms.database.documents.NetworkFailureSet
+import org.thoughtcrime.securesms.database.model.databaseprotos.BodyRangeList
+import org.thoughtcrime.securesms.mms.QuoteModel
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.recipients.RecipientId
+import org.thoughtcrime.securesms.util.JsonUtils
+
+/**
+ * An object that will ingest all fo the [ChatItem]s you want to write, buffer them until hitting a specified batch size, and then batch insert them
+ * for fast throughput.
+ */
+class ChatItemImportInserter(
+  private val db: SQLiteDatabase,
+  private val backupState: BackupState,
+  private val batchSize: Int
+) {
+  companion object {
+    private val TAG = Log.tag(ChatItemImportInserter::class.java)
+
+    private val MESSAGE_COLUMNS = arrayOf(
+      MessageTable.DATE_SENT,
+      MessageTable.DATE_RECEIVED,
+      MessageTable.DATE_SERVER,
+      MessageTable.TYPE,
+      MessageTable.THREAD_ID,
+      MessageTable.READ,
+      MessageTable.BODY,
+      MessageTable.FROM_RECIPIENT_ID,
+      MessageTable.TO_RECIPIENT_ID,
+      MessageTable.DELIVERY_RECEIPT_COUNT,
+      MessageTable.READ_RECEIPT_COUNT,
+      MessageTable.VIEWED_RECEIPT_COUNT,
+      MessageTable.MISMATCHED_IDENTITIES,
+      MessageTable.EXPIRES_IN,
+      MessageTable.EXPIRE_STARTED,
+      MessageTable.UNIDENTIFIED,
+      MessageTable.REMOTE_DELETED,
+      MessageTable.REMOTE_DELETED,
+      MessageTable.NETWORK_FAILURES,
+      MessageTable.QUOTE_ID,
+      MessageTable.QUOTE_AUTHOR,
+      MessageTable.QUOTE_BODY,
+      MessageTable.QUOTE_MISSING,
+      MessageTable.QUOTE_BODY_RANGES,
+      MessageTable.QUOTE_TYPE,
+      MessageTable.SHARED_CONTACTS,
+      MessageTable.LINK_PREVIEWS,
+      MessageTable.MESSAGE_RANGES,
+      MessageTable.VIEW_ONCE
+    )
+
+    private val REACTION_COLUMNS = arrayOf(
+      ReactionTable.MESSAGE_ID,
+      ReactionTable.AUTHOR_ID,
+      ReactionTable.EMOJI,
+      ReactionTable.DATE_SENT,
+      ReactionTable.DATE_RECEIVED
+    )
+
+    private val GROUP_RECEIPT_COLUMNS = arrayOf(
+      GroupReceiptTable.MMS_ID,
+      GroupReceiptTable.RECIPIENT_ID,
+      GroupReceiptTable.STATUS,
+      GroupReceiptTable.TIMESTAMP,
+      GroupReceiptTable.UNIDENTIFIED
+    )
+  }
+
+  private val selfId = Recipient.self().id
+  private val buffer: Buffer = Buffer()
+  private var messageId: Long = SqlUtil.getNextAutoIncrementId(db, MessageTable.TABLE_NAME)
+
+  /**
+   * Indicate that you want to insert the [ChatItem] into the database.
+   * If this item causes the buffer to hit the batch size, then a batch of items will actually be inserted.
+   */
+  fun insert(chatItem: ChatItem) {
+    val fromLocalRecipientId: RecipientId? = backupState.backupToLocalRecipientId[chatItem.authorId]
+    if (fromLocalRecipientId == null) {
+      Log.w(TAG, "[insert] Could not find a local recipient for backup recipient ID ${chatItem.authorId}! Skipping.")
+      return
+    }
+
+    val chatLocalRecipientId: RecipientId? = backupState.chatIdToLocalRecipientId[chatItem.chatId]
+    if (chatLocalRecipientId == null) {
+      Log.w(TAG, "[insert] Could not find a local recipient for chatId ${chatItem.chatId}! Skipping.")
+      return
+    }
+
+    val localThreadId: Long? = backupState.chatIdToLocalThreadId[chatItem.chatId]
+    if (localThreadId == null) {
+      Log.w(TAG, "[insert] Could not find a local threadId for backup chatId ${chatItem.chatId}! Skipping.")
+      return
+    }
+
+    val chatBackupRecipientId: Long? = backupState.chatIdToBackupRecipientId[chatItem.chatId]
+    if (chatBackupRecipientId == null) {
+      Log.w(TAG, "[insert] Could not find a backup recipientId for backup chatId ${chatItem.chatId}! Skipping.")
+      return
+    }
+
+    buffer.messages += chatItem.toMessageContentValues(fromLocalRecipientId, chatLocalRecipientId, localThreadId)
+    buffer.reactions += chatItem.toReactionContentValues(messageId)
+    buffer.groupReceipts += chatItem.toGroupReceiptContentValues(messageId, chatBackupRecipientId)
+
+    messageId++
+
+    if (buffer.size >= batchSize) {
+      flush()
+    }
+  }
+
+  /** Returns true if something was written to the db, otherwise false. */
+  fun flush(): Boolean {
+    if (buffer.size == 0) {
+      return false
+    }
+
+    SqlUtil.buildBulkInsert(MessageTable.TABLE_NAME, MESSAGE_COLUMNS, buffer.messages).forEach {
+      db.execSQL(it.where, it.whereArgs)
+    }
+
+    SqlUtil.buildBulkInsert(ReactionTable.TABLE_NAME, REACTION_COLUMNS, buffer.reactions).forEach {
+      db.execSQL(it.where, it.whereArgs)
+    }
+
+    SqlUtil.buildBulkInsert(GroupReceiptTable.TABLE_NAME, GROUP_RECEIPT_COLUMNS, buffer.groupReceipts).forEach {
+      db.execSQL(it.where, it.whereArgs)
+    }
+
+    messageId = SqlUtil.getNextAutoIncrementId(db, MessageTable.TABLE_NAME)
+
+    return true
+  }
+
+  private fun ChatItem.toMessageContentValues(fromRecipientId: RecipientId, chatRecipientId: RecipientId, threadId: Long): ContentValues {
+    val contentValues = ContentValues()
+
+    contentValues.put(MessageTable.TYPE, this.getMessageType())
+    contentValues.put(MessageTable.DATE_SENT, this.dateSent)
+    contentValues.put(MessageTable.DATE_SERVER, this.incoming?.dateServerSent ?: -1)
+    contentValues.put(MessageTable.FROM_RECIPIENT_ID, fromRecipientId.serialize())
+    contentValues.put(MessageTable.TO_RECIPIENT_ID, (if (this.outgoing != null) chatRecipientId else selfId).serialize())
+    contentValues.put(MessageTable.THREAD_ID, threadId)
+    contentValues.put(MessageTable.DATE_RECEIVED, this.dateReceived)
+    contentValues.put(MessageTable.RECEIPT_TIMESTAMP, this.outgoing?.sendStatus?.maxOf { it.timestamp } ?: 0)
+    contentValues.putNull(MessageTable.LATEST_REVISION_ID)
+    contentValues.putNull(MessageTable.ORIGINAL_MESSAGE_ID)
+    contentValues.put(MessageTable.REVISION_NUMBER, 0)
+    contentValues.put(MessageTable.EXPIRES_IN, this.expiresIn ?: 0)
+    contentValues.put(MessageTable.EXPIRE_STARTED, this.expireStart ?: 0)
+
+    if (this.outgoing != null) {
+      val viewReceiptCount = this.outgoing.sendStatus.count { it.deliveryStatus == SendStatus.Status.VIEWED }
+      val readReceiptCount = Integer.max(viewReceiptCount, this.outgoing.sendStatus.count { it.deliveryStatus == SendStatus.Status.READ })
+      val deliveryReceiptCount = Integer.max(readReceiptCount, this.outgoing.sendStatus.count { it.deliveryStatus == SendStatus.Status.DELIVERED })
+
+      contentValues.put(MessageTable.VIEWED_RECEIPT_COUNT, viewReceiptCount)
+      contentValues.put(MessageTable.READ_RECEIPT_COUNT, readReceiptCount)
+      contentValues.put(MessageTable.DELIVERY_RECEIPT_COUNT, deliveryReceiptCount)
+      contentValues.put(MessageTable.UNIDENTIFIED, this.outgoing.sendStatus.count { it.sealedSender })
+      contentValues.put(MessageTable.READ, 1)
+
+      contentValues.addNetworkFailures(this, backupState)
+      contentValues.addIdentityKeyMismatches(this, backupState)
+    } else {
+      contentValues.put(MessageTable.VIEWED_RECEIPT_COUNT, 0)
+      contentValues.put(MessageTable.READ_RECEIPT_COUNT, 0)
+      contentValues.put(MessageTable.DELIVERY_RECEIPT_COUNT, 0)
+      contentValues.put(MessageTable.UNIDENTIFIED, this.incoming?.sealedSender?.toInt() ?: 0)
+      contentValues.put(MessageTable.READ, this.incoming?.read?.toInt() ?: 0)
+    }
+
+    contentValues.put(MessageTable.QUOTE_ID, 0)
+    contentValues.put(MessageTable.QUOTE_AUTHOR, 0)
+    contentValues.put(MessageTable.QUOTE_MISSING, 0)
+    contentValues.put(MessageTable.QUOTE_TYPE, 0)
+    contentValues.put(MessageTable.VIEW_ONCE, 0)
+    contentValues.put(MessageTable.REMOTE_DELETED, 0)
+
+    when {
+      this.standardMessage != null -> contentValues.addStandardMessage(this.standardMessage)
+      this.remoteDeletedMessage != null -> contentValues.put(MessageTable.REMOTE_DELETED, 1)
+    }
+
+    return contentValues
+  }
+
+  private fun ChatItem.toReactionContentValues(messageId: Long): List<ContentValues> {
+    val reactions: List<Reaction> = when {
+      this.standardMessage != null -> this.standardMessage.reactions
+      this.contactMessage != null -> this.contactMessage.reactions
+      this.voiceMessage != null -> this.voiceMessage.reactions
+      this.stickerMessage != null -> this.stickerMessage.reactions
+      else -> emptyList()
+    }
+
+    return reactions
+      .mapNotNull {
+        val authorId: Long? = backupState.backupToLocalRecipientId[it.authorId]?.toLong()
+
+        if (authorId != null) {
+          contentValuesOf(
+            ReactionTable.MESSAGE_ID to messageId,
+            ReactionTable.AUTHOR_ID to authorId,
+            ReactionTable.DATE_SENT to it.sentTimestamp,
+            ReactionTable.DATE_RECEIVED to it.receivedTimestamp,
+            ReactionTable.EMOJI to it.emoji
+          )
+        } else {
+          Log.w(TAG, "[Reaction] Could not find a local recipient for backup recipient ID ${it.authorId}! Skipping.")
+          null
+        }
+      }
+  }
+
+  private fun ChatItem.toGroupReceiptContentValues(messageId: Long, chatBackupRecipientId: Long): List<ContentValues> {
+    if (this.outgoing == null) {
+      return emptyList()
+    }
+
+    // TODO This seems like an indirect/bad way to detect if this is a 1:1 or group convo
+    if (this.outgoing.sendStatus.size == 1 && this.outgoing.sendStatus[0].recipientId == chatBackupRecipientId) {
+      return emptyList()
+    }
+
+    return this.outgoing.sendStatus.mapNotNull { sendStatus ->
+      val recipientId = backupState.backupToLocalRecipientId[sendStatus.recipientId]
+
+      if (recipientId != null) {
+        contentValuesOf(
+          GroupReceiptTable.MMS_ID to messageId,
+          GroupReceiptTable.RECIPIENT_ID to recipientId.serialize(),
+          GroupReceiptTable.STATUS to sendStatus.deliveryStatus.toLocalSendStatus(),
+          GroupReceiptTable.TIMESTAMP to sendStatus.timestamp,
+          GroupReceiptTable.UNIDENTIFIED to sendStatus.sealedSender
+        )
+      } else {
+        Log.w(TAG, "[GroupReceipts] Could not find a local recipient for backup recipient ID ${sendStatus.recipientId}! Skipping.")
+        null
+      }
+    }
+  }
+
+  private fun ChatItem.getMessageType(): Long {
+    var type: Long = if (this.outgoing != null) {
+      if (this.outgoing.sendStatus.count { it.identityKeyMismatch } > 0) {
+        MessageTypes.BASE_SENT_FAILED_TYPE
+      } else if (this.outgoing.sendStatus.count { it.networkFailure } > 0) {
+        MessageTypes.BASE_SENDING_TYPE
+      } else {
+        MessageTypes.BASE_SENT_TYPE
+      }
+    } else {
+      MessageTypes.BASE_INBOX_TYPE
+    }
+
+    if (!this.sms) {
+      type = type or MessageTypes.SECURE_MESSAGE_BIT or MessageTypes.PUSH_MESSAGE_BIT
+    }
+
+    return type
+  }
+
+  private fun ContentValues.addStandardMessage(standardMessage: StandardMessage) {
+    if (standardMessage.text != null) {
+      this.put(MessageTable.BODY, standardMessage.text.body)
+
+      if (standardMessage.text.bodyRanges.isNotEmpty()) {
+        this.put(MessageTable.MESSAGE_RANGES, standardMessage.text.bodyRanges.toLocalBodyRanges()?.encode() as ByteArray?)
+      }
+    }
+
+    if (standardMessage.quote != null) {
+      this.addQuote(standardMessage.quote)
+    }
+  }
+
+  private fun ContentValues.addQuote(quote: Quote) {
+    this.put(MessageTable.QUOTE_ID, quote.targetSentTimestamp)
+    this.put(MessageTable.QUOTE_AUTHOR, backupState.backupToLocalRecipientId[quote.authorId]!!.serialize())
+    this.put(MessageTable.QUOTE_BODY, quote.text)
+    this.put(MessageTable.QUOTE_TYPE, quote.type.toLocalQuoteType())
+    this.put(MessageTable.QUOTE_BODY_RANGES, quote.bodyRanges.toLocalBodyRanges()?.encode())
+    // TODO quote attachments
+    this.put(MessageTable.QUOTE_MISSING, quote.originalMessageMissing.toInt())
+  }
+
+  private fun Quote.Type.toLocalQuoteType(): Int {
+    return when (this) {
+      Quote.Type.UNKNOWN -> QuoteModel.Type.NORMAL.code
+      Quote.Type.NORMAL -> QuoteModel.Type.NORMAL.code
+      Quote.Type.GIFTBADGE -> QuoteModel.Type.GIFT_BADGE.code
+    }
+  }
+
+  private fun ContentValues.addNetworkFailures(chatItem: ChatItem, backupState: BackupState) {
+    if (chatItem.outgoing == null) {
+      return
+    }
+
+    val networkFailures = chatItem.outgoing.sendStatus
+      .filter { status -> status.networkFailure }
+      .mapNotNull { status -> backupState.backupToLocalRecipientId[status.recipientId] }
+      .map { recipientId -> NetworkFailure(recipientId) }
+      .toSet()
+
+    if (networkFailures.isNotEmpty()) {
+      this.put(MessageTable.NETWORK_FAILURES, JsonUtils.toJson(NetworkFailureSet(networkFailures)))
+    }
+  }
+
+  private fun ContentValues.addIdentityKeyMismatches(chatItem: ChatItem, backupState: BackupState) {
+    if (chatItem.outgoing == null) {
+      return
+    }
+
+    val mismatches = chatItem.outgoing.sendStatus
+      .filter { status -> status.identityKeyMismatch }
+      .mapNotNull { status -> backupState.backupToLocalRecipientId[status.recipientId] }
+      .map { recipientId -> IdentityKeyMismatch(recipientId, null) } // TODO We probably want the actual identity key in this status situation?
+      .toSet()
+
+    if (mismatches.isNotEmpty()) {
+      this.put(MessageTable.MISMATCHED_IDENTITIES, JsonUtils.toJson(IdentityKeyMismatchSet(mismatches)))
+    }
+  }
+
+  private fun List<BodyRange>.toLocalBodyRanges(): BodyRangeList? {
+    if (this.isEmpty()) {
+      return null
+    }
+
+    return BodyRangeList(
+      ranges = this.map { bodyRange ->
+        BodyRangeList.BodyRange(
+          mentionUuid = bodyRange.mentionAci,
+          style = bodyRange.style?.let {
+            when (bodyRange.style) {
+              BodyRange.Style.BOLD -> BodyRangeList.BodyRange.Style.BOLD
+              BodyRange.Style.ITALIC -> BodyRangeList.BodyRange.Style.ITALIC
+              BodyRange.Style.MONOSPACE -> BodyRangeList.BodyRange.Style.MONOSPACE
+              BodyRange.Style.SPOILER -> BodyRangeList.BodyRange.Style.SPOILER
+              BodyRange.Style.STRIKETHROUGH -> BodyRangeList.BodyRange.Style.STRIKETHROUGH
+              else -> null
+            }
+          },
+          start = bodyRange.start ?: 0,
+          length = bodyRange.length ?: 0
+        )
+      }
+    )
+  }
+
+  private fun SendStatus.Status.toLocalSendStatus(): Int {
+    return when (this) {
+      SendStatus.Status.FAILED -> GroupReceiptTable.STATUS_UNKNOWN
+      SendStatus.Status.PENDING -> GroupReceiptTable.STATUS_UNDELIVERED
+      SendStatus.Status.SENT -> GroupReceiptTable.STATUS_UNDELIVERED
+      SendStatus.Status.DELIVERED -> GroupReceiptTable.STATUS_DELIVERED
+      SendStatus.Status.READ -> GroupReceiptTable.STATUS_READ
+      SendStatus.Status.VIEWED -> GroupReceiptTable.STATUS_VIEWED
+      SendStatus.Status.SKIPPED -> GroupReceiptTable.STATUS_SKIPPED
+    }
+  }
+
+  private class Buffer(
+    val messages: MutableList<ContentValues> = mutableListOf(),
+    val reactions: MutableList<ContentValues> = mutableListOf(),
+    val groupReceipts: MutableList<ContentValues> = mutableListOf()
+  ) {
+    val size: Int
+      get() = listOf(messages.size, reactions.size, groupReceipts.size).max()
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/database/DistributionListTablesBackupExtensions.kt

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.database
+
+import okio.ByteString.Companion.toByteString
+import org.signal.core.util.CursorUtil
+import org.signal.core.util.delete
+import org.signal.core.util.logging.Log
+import org.signal.core.util.readToList
+import org.signal.core.util.requireLong
+import org.signal.core.util.requireNonNullString
+import org.signal.core.util.requireObject
+import org.signal.core.util.select
+import org.thoughtcrime.securesms.backup.v2.BackupState
+import org.thoughtcrime.securesms.database.DistributionListTables
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.database.model.DistributionListId
+import org.thoughtcrime.securesms.database.model.DistributionListPrivacyMode
+import org.thoughtcrime.securesms.database.model.DistributionListRecord
+import org.thoughtcrime.securesms.recipients.RecipientId
+import org.whispersystems.signalservice.api.push.DistributionId
+import org.whispersystems.signalservice.api.util.UuidUtil
+import org.whispersystems.signalservice.api.util.toByteArray
+import org.thoughtcrime.securesms.backup.v2.proto.DistributionList as BackupDistributionList
+
+private val TAG = Log.tag(DistributionListTables::class.java)
+
+fun DistributionListTables.getAllForBackup(): List<BackupRecipient> {
+  val records = readableDatabase
+    .select()
+    .from(DistributionListTables.ListTable.TABLE_NAME)
+    .run()
+    .readToList { cursor ->
+      val id: DistributionListId = DistributionListId.from(cursor.requireLong(DistributionListTables.ListTable.ID))
+      val privacyMode: DistributionListPrivacyMode = cursor.requireObject(DistributionListTables.ListTable.PRIVACY_MODE, DistributionListPrivacyMode.Serializer)
+
+      DistributionListRecord(
+        id = id,
+        name = cursor.requireNonNullString(DistributionListTables.ListTable.NAME),
+        distributionId = DistributionId.from(cursor.requireNonNullString(DistributionListTables.ListTable.DISTRIBUTION_ID)),
+        allowsReplies = CursorUtil.requireBoolean(cursor, DistributionListTables.ListTable.ALLOWS_REPLIES),
+        rawMembers = getRawMembers(id, privacyMode),
+        members = getMembers(id),
+        deletedAtTimestamp = 0L,
+        isUnknown = CursorUtil.requireBoolean(cursor, DistributionListTables.ListTable.IS_UNKNOWN),
+        privacyMode = privacyMode
+      )
+    }
+
+  return records
+    .map { record ->
+      BackupRecipient(
+        distributionList = BackupDistributionList(
+          name = record.name,
+          distributionId = record.distributionId.asUuid().toByteArray().toByteString(),
+          allowReplies = record.allowsReplies,
+          deletionTimestamp = record.deletedAtTimestamp,
+          isUnknown = record.isUnknown,
+          privacyMode = record.privacyMode.toBackupPrivacyMode(),
+          memberRecipientIds = record.members.map { it.toLong() }
+        )
+      )
+    }
+}
+
+fun DistributionListTables.restoreFromBackup(dlist: BackupDistributionList, backupState: BackupState): RecipientId {
+  val members: List<RecipientId> = dlist.memberRecipientIds
+    .mapNotNull { backupState.backupToLocalRecipientId[it] }
+
+  if (members.size != dlist.memberRecipientIds.size) {
+    Log.w(TAG, "Couldn't find some member recipients! Missing backup recipientIds: ${dlist.memberRecipientIds.toSet() - members.toSet()}")
+  }
+
+  val dlistId = this.createList(
+    name = dlist.name,
+    members = members,
+    distributionId = DistributionId.from(UuidUtil.fromByteString(dlist.distributionId)),
+    allowsReplies = dlist.allowReplies,
+    deletionTimestamp = dlist.deletionTimestamp,
+    storageId = null,
+    isUnknown = dlist.isUnknown,
+    privacyMode = dlist.privacyMode.toLocalPrivacyMode()
+  )!!
+
+  return SignalDatabase.distributionLists.getRecipientId(dlistId)!!
+}
+
+fun DistributionListTables.clearAllDataForBackupRestore() {
+  writableDatabase
+    .delete(DistributionListTables.ListTable.TABLE_NAME)
+    .run()
+
+  writableDatabase
+    .delete(DistributionListTables.MembershipTable.TABLE_NAME)
+    .run()
+}
+
+private fun DistributionListPrivacyMode.toBackupPrivacyMode(): BackupDistributionList.PrivacyMode {
+  return when (this) {
+    DistributionListPrivacyMode.ONLY_WITH -> BackupDistributionList.PrivacyMode.ONLY_WITH
+    DistributionListPrivacyMode.ALL -> BackupDistributionList.PrivacyMode.ALL
+    DistributionListPrivacyMode.ALL_EXCEPT -> BackupDistributionList.PrivacyMode.ALL_EXCEPT
+  }
+}
+
+private fun BackupDistributionList.PrivacyMode.toLocalPrivacyMode(): DistributionListPrivacyMode {
+  return when (this) {
+    BackupDistributionList.PrivacyMode.ONLY_WITH -> DistributionListPrivacyMode.ONLY_WITH
+    BackupDistributionList.PrivacyMode.ALL -> DistributionListPrivacyMode.ALL
+    BackupDistributionList.PrivacyMode.ALL_EXCEPT -> DistributionListPrivacyMode.ALL_EXCEPT
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/database/MessageTableBackupExtensions.kt

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.database
+
+import org.signal.core.util.SqlUtil
+import org.signal.core.util.logging.Log
+import org.signal.core.util.select
+import org.thoughtcrime.securesms.backup.v2.BackupState
+import org.thoughtcrime.securesms.database.MessageTable
+import org.thoughtcrime.securesms.database.MessageTypes
+
+private val TAG = Log.tag(MessageTable::class.java)
+private const val BASE_TYPE = "base_type"
+
+fun MessageTable.getMessagesForBackup(): ChatItemExportIterator {
+  val cursor = readableDatabase
+    .select(
+      MessageTable.ID,
+      MessageTable.DATE_SENT,
+      MessageTable.DATE_RECEIVED,
+      MessageTable.DATE_SERVER,
+      MessageTable.TYPE,
+      MessageTable.THREAD_ID,
+      MessageTable.BODY,
+      MessageTable.MESSAGE_RANGES,
+      MessageTable.FROM_RECIPIENT_ID,
+      MessageTable.TO_RECIPIENT_ID,
+      MessageTable.EXPIRES_IN,
+      MessageTable.EXPIRE_STARTED,
+      MessageTable.REMOTE_DELETED,
+      MessageTable.UNIDENTIFIED,
+      MessageTable.QUOTE_ID,
+      MessageTable.QUOTE_AUTHOR,
+      MessageTable.QUOTE_BODY,
+      MessageTable.QUOTE_MISSING,
+      MessageTable.QUOTE_BODY_RANGES,
+      MessageTable.QUOTE_TYPE,
+      MessageTable.ORIGINAL_MESSAGE_ID,
+      MessageTable.LATEST_REVISION_ID,
+      MessageTable.DELIVERY_RECEIPT_COUNT,
+      MessageTable.READ_RECEIPT_COUNT,
+      MessageTable.VIEWED_RECEIPT_COUNT,
+      MessageTable.RECEIPT_TIMESTAMP,
+      MessageTable.READ,
+      MessageTable.NETWORK_FAILURES,
+      MessageTable.MISMATCHED_IDENTITIES,
+      "${MessageTable.TYPE} & ${MessageTypes.BASE_TYPE_MASK} AS ${ChatItemExportIterator.COLUMN_BASE_TYPE}"
+    )
+    .from(MessageTable.TABLE_NAME)
+    .where(
+      """
+      $BASE_TYPE IN (
+        ${MessageTypes.BASE_INBOX_TYPE},
+        ${MessageTypes.BASE_OUTBOX_TYPE},
+        ${MessageTypes.BASE_SENT_TYPE},
+        ${MessageTypes.BASE_SENDING_TYPE},
+        ${MessageTypes.BASE_SENT_FAILED_TYPE}
+      )
+      """
+    )
+    .orderBy("${MessageTable.DATE_RECEIVED} ASC")
+    .run()
+
+  return ChatItemExportIterator(cursor, 100)
+}
+
+fun MessageTable.createChatItemInserter(backupState: BackupState): ChatItemImportInserter {
+  return ChatItemImportInserter(writableDatabase, backupState, 100)
+}
+
+fun MessageTable.clearAllDataForBackupRestore() {
+  writableDatabase.delete(MessageTable.TABLE_NAME, null, null)
+  SqlUtil.resetAutoIncrementValue(writableDatabase, MessageTable.TABLE_NAME)
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/database/RecipientTableBackupExtensions.kt

@@ -0,0 +1,330 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.database
+
+import android.content.ContentValues
+import android.database.Cursor
+import okio.ByteString.Companion.toByteString
+import org.signal.core.util.Base64
+import org.signal.core.util.SqlUtil
+import org.signal.core.util.delete
+import org.signal.core.util.logging.Log
+import org.signal.core.util.nullIfBlank
+import org.signal.core.util.requireBoolean
+import org.signal.core.util.requireInt
+import org.signal.core.util.requireLong
+import org.signal.core.util.requireNonNullBlob
+import org.signal.core.util.requireString
+import org.signal.core.util.select
+import org.signal.core.util.toInt
+import org.signal.core.util.update
+import org.signal.libsignal.zkgroup.InvalidInputException
+import org.thoughtcrime.securesms.backup.v2.BackupState
+import org.thoughtcrime.securesms.backup.v2.proto.AccountData
+import org.thoughtcrime.securesms.backup.v2.proto.Contact
+import org.thoughtcrime.securesms.backup.v2.proto.Group
+import org.thoughtcrime.securesms.backup.v2.proto.Self
+import org.thoughtcrime.securesms.database.GroupTable
+import org.thoughtcrime.securesms.database.RecipientTable
+import org.thoughtcrime.securesms.database.RecipientTableCursorUtil
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.database.model.databaseprotos.RecipientExtras
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
+import org.thoughtcrime.securesms.phonenumbers.PhoneNumberFormatter
+import org.thoughtcrime.securesms.profiles.ProfileName
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.recipients.RecipientId
+import org.whispersystems.signalservice.api.push.ServiceId.ACI
+import org.whispersystems.signalservice.api.push.ServiceId.PNI
+import java.io.Closeable
+
+typealias BackupRecipient = org.thoughtcrime.securesms.backup.v2.proto.Recipient
+typealias BackupGroup = Group
+
+/**
+ * Fetches all individual contacts for backups and returns the result as an iterator.
+ * It's important to note that the iterator still needs to be closed after it's used.
+ * It's recommended to use `.use` or a try-with-resources pattern.
+ */
+fun RecipientTable.getContactsForBackup(selfId: Long): BackupContactIterator {
+  val cursor = readableDatabase
+    .select(
+      RecipientTable.ID,
+      RecipientTable.ACI_COLUMN,
+      RecipientTable.PNI_COLUMN,
+      RecipientTable.USERNAME,
+      RecipientTable.E164,
+      RecipientTable.BLOCKED,
+      RecipientTable.HIDDEN,
+      RecipientTable.REGISTERED,
+      RecipientTable.UNREGISTERED_TIMESTAMP,
+      RecipientTable.PROFILE_KEY,
+      RecipientTable.PROFILE_SHARING,
+      RecipientTable.PROFILE_GIVEN_NAME,
+      RecipientTable.PROFILE_FAMILY_NAME,
+      RecipientTable.PROFILE_JOINED_NAME,
+      RecipientTable.MUTE_UNTIL,
+      RecipientTable.EXTRAS
+    )
+    .from(RecipientTable.TABLE_NAME)
+    .where(
+      """
+      ${RecipientTable.TYPE} = ? AND (
+        ${RecipientTable.ACI_COLUMN} NOT NULL OR
+        ${RecipientTable.PNI_COLUMN} NOT NULL OR
+        ${RecipientTable.E164} NOT NULL
+      )
+      """,
+      RecipientTable.RecipientType.INDIVIDUAL.id
+    )
+    .run()
+
+  return BackupContactIterator(cursor, selfId)
+}
+
+fun RecipientTable.getGroupsForBackup(): BackupGroupIterator {
+  val cursor = readableDatabase
+    .select(
+      "${RecipientTable.TABLE_NAME}.${RecipientTable.ID}",
+      "${RecipientTable.TABLE_NAME}.${RecipientTable.BLOCKED}",
+      "${RecipientTable.TABLE_NAME}.${RecipientTable.PROFILE_SHARING}",
+      "${RecipientTable.TABLE_NAME}.${RecipientTable.MUTE_UNTIL}",
+      "${RecipientTable.TABLE_NAME}.${RecipientTable.EXTRAS}",
+      "${GroupTable.TABLE_NAME}.${GroupTable.V2_MASTER_KEY}",
+      "${GroupTable.TABLE_NAME}.${GroupTable.SHOW_AS_STORY_STATE}"
+    )
+    .from(
+      """
+      ${RecipientTable.TABLE_NAME} 
+        INNER JOIN ${GroupTable.TABLE_NAME} ON ${RecipientTable.TABLE_NAME}.${RecipientTable.ID} = ${GroupTable.TABLE_NAME}.${GroupTable.RECIPIENT_ID}
+      """
+    )
+    .run()
+
+  return BackupGroupIterator(cursor)
+}
+
+/**
+ * Takes a [BackupRecipient] and writes it into the database.
+ */
+fun RecipientTable.restoreRecipientFromBackup(recipient: BackupRecipient, backupState: BackupState): RecipientId? {
+  // TODO Need to handle groups
+  // TODO Also, should we move this when statement up to mimic the export? Kinda weird that this calls distributionListTable functions
+  return when {
+    recipient.contact != null -> restoreContactFromBackup(recipient.contact)
+    recipient.distributionList != null -> SignalDatabase.distributionLists.restoreFromBackup(recipient.distributionList, backupState)
+    recipient.self != null -> Recipient.self().id
+    else -> {
+      Log.w(TAG, "Unrecognized recipient type!")
+      null
+    }
+  }
+}
+
+/**
+ * Given [AccountData], this will insert the necessary data for the local user into the [RecipientTable].
+ */
+fun RecipientTable.restoreSelfFromBackup(accountData: AccountData) {
+  val self = Recipient.trustedPush(ACI.parseOrThrow(accountData.aci.toByteArray()), PNI.parseOrNull(accountData.pni.toByteArray()), accountData.e164.toString())
+
+  val values = ContentValues().apply {
+    put(RecipientTable.PROFILE_GIVEN_NAME, accountData.givenName.nullIfBlank())
+    put(RecipientTable.PROFILE_FAMILY_NAME, accountData.familyName.nullIfBlank())
+    put(RecipientTable.PROFILE_JOINED_NAME, ProfileName.fromParts(accountData.givenName, accountData.familyName).toString().nullIfBlank())
+    put(RecipientTable.PROFILE_AVATAR, accountData.avatarUrlPath.nullIfBlank())
+    put(RecipientTable.REGISTERED, RecipientTable.RegisteredState.REGISTERED.id)
+    put(RecipientTable.PROFILE_SHARING, true)
+    put(RecipientTable.UNREGISTERED_TIMESTAMP, 0)
+    put(RecipientTable.EXTRAS, RecipientExtras().encode())
+
+    try {
+      put(RecipientTable.PROFILE_KEY, Base64.encodeWithPadding(accountData.profileKey.toByteArray()).nullIfBlank())
+    } catch (e: InvalidInputException) {
+      Log.w(TAG, "Missing profile key during restore")
+    }
+
+    put(RecipientTable.USERNAME, accountData.username)
+  }
+
+  writableDatabase
+    .update(RecipientTable.TABLE_NAME)
+    .values(values)
+    .where("${RecipientTable.ID} = ?", self.id)
+    .run()
+}
+
+fun RecipientTable.clearAllDataForBackupRestore() {
+  writableDatabase.delete(RecipientTable.TABLE_NAME).run()
+  SqlUtil.resetAutoIncrementValue(writableDatabase, RecipientTable.TABLE_NAME)
+
+  RecipientId.clearCache()
+  ApplicationDependencies.getRecipientCache().clear()
+  ApplicationDependencies.getRecipientCache().clearSelf()
+}
+
+private fun RecipientTable.restoreContactFromBackup(contact: Contact): RecipientId {
+  val id = getAndPossiblyMergePnpVerified(
+    aci = ACI.parseOrNull(contact.aci?.toByteArray()),
+    pni = PNI.parseOrNull(contact.pni?.toByteArray()),
+    e164 = contact.formattedE164
+  )
+
+  val profileKey = contact.profileKey?.toByteArray()
+
+  writableDatabase
+    .update(RecipientTable.TABLE_NAME)
+    .values(
+      RecipientTable.BLOCKED to contact.blocked,
+      RecipientTable.HIDDEN to contact.hidden,
+      RecipientTable.PROFILE_FAMILY_NAME to contact.profileFamilyName.nullIfBlank(),
+      RecipientTable.PROFILE_GIVEN_NAME to contact.profileGivenName.nullIfBlank(),
+      RecipientTable.PROFILE_JOINED_NAME to contact.profileJoinedName.nullIfBlank(),
+      RecipientTable.PROFILE_KEY to if (profileKey == null) null else Base64.encodeWithPadding(profileKey),
+      RecipientTable.PROFILE_SHARING to contact.profileSharing.toInt(),
+      RecipientTable.REGISTERED to contact.registered.toLocalRegisteredState().id,
+      RecipientTable.USERNAME to contact.username,
+      RecipientTable.UNREGISTERED_TIMESTAMP to contact.unregisteredTimestamp,
+      RecipientTable.EXTRAS to contact.toLocalExtras().encode()
+    )
+    .where("${RecipientTable.ID} = ?", id)
+    .run()
+
+  return id
+}
+
+private fun Contact.toLocalExtras(): RecipientExtras {
+  return RecipientExtras(
+    hideStory = this.hideStory
+  )
+}
+
+/**
+ * Provides a nice iterable interface over a [RecipientTable] cursor, converting rows to [BackupRecipient]s.
+ * Important: Because this is backed by a cursor, you must close it. It's recommended to use `.use()` or try-with-resources.
+ */
+class BackupContactIterator(private val cursor: Cursor, private val selfId: Long) : Iterator<BackupRecipient>, Closeable {
+  override fun hasNext(): Boolean {
+    return cursor.count > 0 && !cursor.isLast
+  }
+
+  override fun next(): BackupRecipient {
+    if (!cursor.moveToNext()) {
+      throw NoSuchElementException()
+    }
+
+    val id = cursor.requireLong(RecipientTable.ID)
+    if (id == selfId) {
+      return BackupRecipient(
+        id = id,
+        self = Self()
+      )
+    }
+
+    val aci = ACI.parseOrNull(cursor.requireString(RecipientTable.ACI_COLUMN))
+    val pni = PNI.parseOrNull(cursor.requireString(RecipientTable.PNI_COLUMN))
+    val registeredState = RecipientTable.RegisteredState.fromId(cursor.requireInt(RecipientTable.REGISTERED))
+    val profileKey = cursor.requireString(RecipientTable.PROFILE_KEY)
+    val extras = RecipientTableCursorUtil.getExtras(cursor)
+
+    return BackupRecipient(
+      id = id,
+      contact = Contact(
+        aci = aci?.toByteArray()?.toByteString(),
+        pni = pni?.toByteArray()?.toByteString(),
+        username = cursor.requireString(RecipientTable.USERNAME),
+        e164 = cursor.requireString(RecipientTable.E164)?.e164ToLong(),
+        blocked = cursor.requireBoolean(RecipientTable.BLOCKED),
+        hidden = cursor.requireBoolean(RecipientTable.HIDDEN),
+        registered = registeredState.toContactRegisteredState(),
+        unregisteredTimestamp = cursor.requireLong(RecipientTable.UNREGISTERED_TIMESTAMP),
+        profileKey = if (profileKey != null) Base64.decode(profileKey).toByteString() else null,
+        profileSharing = cursor.requireBoolean(RecipientTable.PROFILE_SHARING),
+        profileGivenName = cursor.requireString(RecipientTable.PROFILE_GIVEN_NAME).nullIfBlank(),
+        profileFamilyName = cursor.requireString(RecipientTable.PROFILE_FAMILY_NAME).nullIfBlank(),
+        profileJoinedName = cursor.requireString(RecipientTable.PROFILE_JOINED_NAME).nullIfBlank(),
+        hideStory = extras?.hideStory() ?: false
+      )
+    )
+  }
+
+  override fun close() {
+    cursor.close()
+  }
+}
+
+/**
+ * Provides a nice iterable interface over a [RecipientTable] cursor, converting rows to [BackupRecipient]s.
+ * Important: Because this is backed by a cursor, you must close it. It's recommended to use `.use()` or try-with-resources.
+ */
+class BackupGroupIterator(private val cursor: Cursor) : Iterator<BackupRecipient>, Closeable {
+  override fun hasNext(): Boolean {
+    return cursor.count > 0 && !cursor.isLast
+  }
+
+  override fun next(): BackupRecipient {
+    if (!cursor.moveToNext()) {
+      throw NoSuchElementException()
+    }
+
+    val extras = RecipientTableCursorUtil.getExtras(cursor)
+    val showAsStoryState: GroupTable.ShowAsStoryState = GroupTable.ShowAsStoryState.deserialize(cursor.requireInt(GroupTable.SHOW_AS_STORY_STATE))
+
+    return BackupRecipient(
+      id = cursor.requireLong(RecipientTable.ID),
+      group = BackupGroup(
+        masterKey = cursor.requireNonNullBlob(GroupTable.V2_MASTER_KEY).toByteString(),
+        whitelisted = cursor.requireBoolean(RecipientTable.PROFILE_SHARING),
+        hideStory = extras?.hideStory() ?: false,
+        storySendMode = showAsStoryState.toGroupStorySendMode()
+      )
+    )
+  }
+
+  override fun close() {
+    cursor.close()
+  }
+}
+
+private fun String.e164ToLong(): Long? {
+  val fixed = if (this.startsWith("+")) {
+    this.substring(1)
+  } else {
+    this
+  }
+
+  return fixed.toLongOrNull()
+}
+
+private fun RecipientTable.RegisteredState.toContactRegisteredState(): Contact.Registered {
+  return when (this) {
+    RecipientTable.RegisteredState.REGISTERED -> Contact.Registered.REGISTERED
+    RecipientTable.RegisteredState.NOT_REGISTERED -> Contact.Registered.NOT_REGISTERED
+    RecipientTable.RegisteredState.UNKNOWN -> Contact.Registered.UNKNOWN
+  }
+}
+
+private fun Contact.Registered.toLocalRegisteredState(): RecipientTable.RegisteredState {
+  return when (this) {
+    Contact.Registered.REGISTERED -> RecipientTable.RegisteredState.REGISTERED
+    Contact.Registered.NOT_REGISTERED -> RecipientTable.RegisteredState.NOT_REGISTERED
+    Contact.Registered.UNKNOWN -> RecipientTable.RegisteredState.UNKNOWN
+  }
+}
+
+private fun GroupTable.ShowAsStoryState.toGroupStorySendMode(): Group.StorySendMode {
+  return when (this) {
+    GroupTable.ShowAsStoryState.ALWAYS -> Group.StorySendMode.ENABLED
+    GroupTable.ShowAsStoryState.NEVER -> Group.StorySendMode.DISABLED
+    GroupTable.ShowAsStoryState.IF_ACTIVE -> Group.StorySendMode.DEFAULT
+  }
+}
+
+private val Contact.formattedE164: String?
+  get() {
+    return e164?.let {
+      PhoneNumberFormatter.get(ApplicationDependencies.getApplication()).format(e164.toString())
+    }
+  }

app/src/main/java/org/thoughtcrime/securesms/backup/v2/database/ThreadTableBackupExtensions.kt

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.database
+
+import android.database.Cursor
+import org.signal.core.util.SqlUtil
+import org.signal.core.util.requireBoolean
+import org.signal.core.util.requireLong
+import org.signal.core.util.select
+import org.thoughtcrime.securesms.backup.v2.proto.Chat
+import org.thoughtcrime.securesms.database.ThreadTable
+import java.io.Closeable
+
+fun ThreadTable.getThreadsForBackup(): ChatIterator {
+  val cursor = readableDatabase
+    .select(
+      ThreadTable.ID,
+      ThreadTable.RECIPIENT_ID,
+      ThreadTable.ARCHIVED,
+      ThreadTable.PINNED,
+      ThreadTable.EXPIRES_IN
+    )
+    .from(ThreadTable.TABLE_NAME)
+    .run()
+
+  return ChatIterator(cursor)
+}
+
+fun ThreadTable.clearAllDataForBackupRestore() {
+  writableDatabase.delete(ThreadTable.TABLE_NAME, null, null)
+  SqlUtil.resetAutoIncrementValue(writableDatabase, ThreadTable.TABLE_NAME)
+  clearCache()
+}
+
+class ChatIterator(private val cursor: Cursor) : Iterator<Chat>, Closeable {
+  override fun hasNext(): Boolean {
+    return cursor.count > 0 && !cursor.isLast
+  }
+
+  override fun next(): Chat {
+    if (!cursor.moveToNext()) {
+      throw NoSuchElementException()
+    }
+
+    return Chat(
+      id = cursor.requireLong(ThreadTable.ID),
+      recipientId = cursor.requireLong(ThreadTable.RECIPIENT_ID),
+      archived = cursor.requireBoolean(ThreadTable.ARCHIVED),
+      pinned = cursor.requireBoolean(ThreadTable.PINNED),
+      expirationTimer = cursor.requireLong(ThreadTable.EXPIRES_IN)
+    )
+  }
+
+  override fun close() {
+    cursor.close()
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/processor/AccountDataProcessor.kt

@@ -0,0 +1,176 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.processor
+
+import okio.ByteString.Companion.EMPTY
+import okio.ByteString.Companion.toByteString
+import org.thoughtcrime.securesms.backup.v2.database.restoreSelfFromBackup
+import org.thoughtcrime.securesms.backup.v2.proto.AccountData
+import org.thoughtcrime.securesms.backup.v2.proto.Frame
+import org.thoughtcrime.securesms.backup.v2.stream.BackupFrameEmitter
+import org.thoughtcrime.securesms.components.settings.app.usernamelinks.UsernameQrCodeColorScheme
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.database.SignalDatabase.Companion.recipients
+import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
+import org.thoughtcrime.securesms.jobs.RetrieveProfileAvatarJob
+import org.thoughtcrime.securesms.keyvalue.PhoneNumberPrivacyValues
+import org.thoughtcrime.securesms.keyvalue.SignalStore
+import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.subscription.Subscriber
+import org.thoughtcrime.securesms.util.ProfileUtil
+import org.thoughtcrime.securesms.util.TextSecurePreferences
+import org.whispersystems.signalservice.api.push.UsernameLinkComponents
+import org.whispersystems.signalservice.api.storage.StorageRecordProtoUtil.defaultAccountRecord
+import org.whispersystems.signalservice.api.subscriptions.SubscriberId
+import org.whispersystems.signalservice.api.util.UuidUtil
+
+object AccountDataProcessor {
+
+  fun export(emitter: BackupFrameEmitter) {
+    val context = ApplicationDependencies.getApplication()
+
+    val self = Recipient.self().fresh()
+    val record = recipients.getRecordForSync(self.id)
+
+    val pniIdentityKey = SignalStore.account().pniIdentityKey
+    val aciIdentityKey = SignalStore.account().aciIdentityKey
+
+    val subscriber: Subscriber? = SignalStore.donationsValues().getSubscriber()
+
+    emitter.emit(
+      Frame(
+        account = AccountData(
+          aci = SignalStore.account().aci!!.toByteString(),
+          pni = SignalStore.account().pni!!.toByteString(),
+          e164 = SignalStore.account().e164!!.toLong(),
+          pniIdentityPrivateKey = pniIdentityKey.privateKey.serialize().toByteString(),
+          pniIdentityPublicKey = pniIdentityKey.publicKey.serialize().toByteString(),
+          aciIdentityPrivateKey = aciIdentityKey.privateKey.serialize().toByteString(),
+          aciIdentityPublicKey = aciIdentityKey.publicKey.serialize().toByteString(),
+          profileKey = self.profileKey?.toByteString() ?: EMPTY,
+          givenName = self.profileName.givenName,
+          familyName = self.profileName.familyName,
+          avatarUrlPath = self.profileAvatar ?: "",
+          subscriptionManuallyCancelled = SignalStore.donationsValues().isUserManuallyCancelled(),
+          username = SignalStore.account().username,
+          subscriberId = subscriber?.subscriberId?.bytes?.toByteString() ?: defaultAccountRecord.subscriberId,
+          subscriberCurrencyCode = subscriber?.currencyCode ?: defaultAccountRecord.subscriberCurrencyCode,
+          accountSettings = AccountData.AccountSettings(
+            storyViewReceiptsEnabled = SignalStore.storyValues().viewedReceiptsEnabled,
+            hasReadOnboardingStory = SignalStore.storyValues().userHasViewedOnboardingStory || SignalStore.storyValues().userHasReadOnboardingStory,
+            noteToSelfArchived = record != null && record.syncExtras.isArchived,
+            noteToSelfMarkedUnread = record != null && record.syncExtras.isForcedUnread,
+            typingIndicators = TextSecurePreferences.isTypingIndicatorsEnabled(context),
+            readReceipts = TextSecurePreferences.isReadReceiptsEnabled(context),
+            sealedSenderIndicators = TextSecurePreferences.isShowUnidentifiedDeliveryIndicatorsEnabled(context),
+            linkPreviews = SignalStore.settings().isLinkPreviewsEnabled,
+            unlistedPhoneNumber = SignalStore.phoneNumberPrivacy().phoneNumberListingMode.isUnlisted,
+            phoneNumberSharingMode = SignalStore.phoneNumberPrivacy().phoneNumberSharingMode.toBackupPhoneNumberSharingMode(),
+            preferContactAvatars = SignalStore.settings().isPreferSystemContactPhotos,
+            universalExpireTimer = SignalStore.settings().universalExpireTimer,
+            preferredReactionEmoji = SignalStore.emojiValues().reactions,
+            storiesDisabled = SignalStore.storyValues().isFeatureDisabled,
+            hasViewedOnboardingStory = SignalStore.storyValues().userHasViewedOnboardingStory,
+            hasSetMyStoriesPrivacy = SignalStore.storyValues().userHasBeenNotifiedAboutStories,
+            keepMutedChatsArchived = SignalStore.settings().shouldKeepMutedChatsArchived(),
+            displayBadgesOnProfile = SignalStore.donationsValues().getDisplayBadgesOnProfile(),
+            hasSeenGroupStoryEducationSheet = SignalStore.storyValues().userHasSeenGroupStoryEducationSheet
+          )
+        )
+      )
+    )
+  }
+
+  fun import(accountData: AccountData) {
+    SignalStore.account().restoreAciIdentityKeyFromBackup(accountData.aciIdentityPublicKey.toByteArray(), accountData.aciIdentityPrivateKey.toByteArray())
+    SignalStore.account().restorePniIdentityKeyFromBackup(accountData.pniIdentityPublicKey.toByteArray(), accountData.pniIdentityPrivateKey.toByteArray())
+
+    recipients.restoreSelfFromBackup(accountData)
+
+    SignalStore.account().setRegistered(true)
+
+    val context = ApplicationDependencies.getApplication()
+
+    val settings = accountData.accountSettings
+
+    if (settings != null) {
+      TextSecurePreferences.setReadReceiptsEnabled(context, settings.readReceipts)
+      TextSecurePreferences.setTypingIndicatorsEnabled(context, settings.typingIndicators)
+      TextSecurePreferences.setShowUnidentifiedDeliveryIndicatorsEnabled(context, settings.sealedSenderIndicators)
+      SignalStore.settings().isLinkPreviewsEnabled = settings.linkPreviews
+      SignalStore.phoneNumberPrivacy().phoneNumberListingMode = if (settings.unlistedPhoneNumber) PhoneNumberPrivacyValues.PhoneNumberListingMode.UNLISTED else PhoneNumberPrivacyValues.PhoneNumberListingMode.LISTED
+      SignalStore.phoneNumberPrivacy().phoneNumberSharingMode = settings.phoneNumberSharingMode.toLocalPhoneNumberMode()
+      SignalStore.settings().isPreferSystemContactPhotos = settings.preferContactAvatars
+      SignalStore.settings().universalExpireTimer = settings.universalExpireTimer
+      SignalStore.emojiValues().reactions = settings.preferredReactionEmoji
+      SignalStore.donationsValues().setDisplayBadgesOnProfile(settings.displayBadgesOnProfile)
+      SignalStore.settings().setKeepMutedChatsArchived(settings.keepMutedChatsArchived)
+      SignalStore.storyValues().userHasBeenNotifiedAboutStories = settings.hasSetMyStoriesPrivacy
+      SignalStore.storyValues().userHasViewedOnboardingStory = settings.hasViewedOnboardingStory
+      SignalStore.storyValues().isFeatureDisabled = settings.storiesDisabled
+      SignalStore.storyValues().userHasReadOnboardingStory = settings.hasReadOnboardingStory
+      SignalStore.storyValues().userHasSeenGroupStoryEducationSheet = settings.hasSeenGroupStoryEducationSheet
+      SignalStore.storyValues().viewedReceiptsEnabled = settings.storyViewReceiptsEnabled ?: settings.readReceipts
+
+      if (accountData.subscriptionManuallyCancelled) {
+        SignalStore.donationsValues().updateLocalStateForManualCancellation()
+      } else {
+        SignalStore.donationsValues().clearUserManuallyCancelled()
+      }
+
+      if (accountData.subscriberId.size > 0) {
+        val subscriber = Subscriber(SubscriberId.fromBytes(accountData.subscriberId.toByteArray()), accountData.subscriberCurrencyCode)
+        SignalStore.donationsValues().setSubscriber(subscriber)
+      }
+
+      if (accountData.avatarUrlPath.isNotEmpty()) {
+        ApplicationDependencies.getJobManager().add(RetrieveProfileAvatarJob(Recipient.self().fresh(), accountData.avatarUrlPath))
+      }
+
+      if (accountData.usernameLink != null) {
+        SignalStore.account().usernameLink = UsernameLinkComponents(
+          accountData.usernameLink.entropy.toByteArray(),
+          UuidUtil.parseOrThrow(accountData.usernameLink.serverId.toByteArray())
+        )
+        SignalStore.misc().usernameQrCodeColorScheme = accountData.usernameLink.color.toLocalUsernameColor()
+      }
+    }
+
+    SignalDatabase.runPostSuccessfulTransaction { ProfileUtil.handleSelfProfileKeyChange() }
+
+    Recipient.self().live().refresh()
+  }
+
+  private fun PhoneNumberPrivacyValues.PhoneNumberSharingMode.toBackupPhoneNumberSharingMode(): AccountData.PhoneNumberSharingMode {
+    return when (this) {
+      PhoneNumberPrivacyValues.PhoneNumberSharingMode.DEFAULT -> AccountData.PhoneNumberSharingMode.EVERYBODY
+      PhoneNumberPrivacyValues.PhoneNumberSharingMode.EVERYBODY -> AccountData.PhoneNumberSharingMode.EVERYBODY
+      PhoneNumberPrivacyValues.PhoneNumberSharingMode.NOBODY -> AccountData.PhoneNumberSharingMode.NOBODY
+    }
+  }
+
+  private fun AccountData.PhoneNumberSharingMode.toLocalPhoneNumberMode(): PhoneNumberPrivacyValues.PhoneNumberSharingMode {
+    return when (this) {
+      AccountData.PhoneNumberSharingMode.UNKNOWN -> PhoneNumberPrivacyValues.PhoneNumberSharingMode.EVERYBODY
+      AccountData.PhoneNumberSharingMode.EVERYBODY -> PhoneNumberPrivacyValues.PhoneNumberSharingMode.EVERYBODY
+      AccountData.PhoneNumberSharingMode.NOBODY -> PhoneNumberPrivacyValues.PhoneNumberSharingMode.NOBODY
+    }
+  }
+
+  private fun AccountData.UsernameLink.Color?.toLocalUsernameColor(): UsernameQrCodeColorScheme {
+    return when (this) {
+      AccountData.UsernameLink.Color.BLUE -> UsernameQrCodeColorScheme.Blue
+      AccountData.UsernameLink.Color.WHITE -> UsernameQrCodeColorScheme.White
+      AccountData.UsernameLink.Color.GREY -> UsernameQrCodeColorScheme.Grey
+      AccountData.UsernameLink.Color.OLIVE -> UsernameQrCodeColorScheme.Tan
+      AccountData.UsernameLink.Color.GREEN -> UsernameQrCodeColorScheme.Green
+      AccountData.UsernameLink.Color.ORANGE -> UsernameQrCodeColorScheme.Orange
+      AccountData.UsernameLink.Color.PINK -> UsernameQrCodeColorScheme.Pink
+      AccountData.UsernameLink.Color.PURPLE -> UsernameQrCodeColorScheme.Purple
+      else -> UsernameQrCodeColorScheme.Blue
+    }
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/processor/ChatBackupProcessor.kt

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.processor
+
+import org.signal.core.util.logging.Log
+import org.thoughtcrime.securesms.backup.v2.BackupState
+import org.thoughtcrime.securesms.backup.v2.database.getThreadsForBackup
+import org.thoughtcrime.securesms.backup.v2.proto.Chat
+import org.thoughtcrime.securesms.backup.v2.proto.Frame
+import org.thoughtcrime.securesms.backup.v2.stream.BackupFrameEmitter
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.recipients.RecipientId
+import java.util.Collections
+
+object ChatBackupProcessor {
+  val TAG = Log.tag(ChatBackupProcessor::class.java)
+
+  fun export(emitter: BackupFrameEmitter) {
+    SignalDatabase.threads.getThreadsForBackup().use { reader ->
+      for (chat in reader) {
+        emitter.emit(Frame(chat = chat))
+      }
+    }
+  }
+
+  fun import(chat: Chat, backupState: BackupState) {
+    // TODO Perf can be improved here by doing a single insert instead of insert + multiple updates
+
+    val recipientId: RecipientId? = backupState.backupToLocalRecipientId[chat.recipientId]
+
+    if (recipientId != null) {
+      val threadId = SignalDatabase.threads.getOrCreateThreadIdFor(org.thoughtcrime.securesms.recipients.Recipient.resolved(recipientId))
+
+      if (chat.archived) {
+        SignalDatabase.threads.archiveConversation(threadId)
+      }
+
+      if (chat.pinned) {
+        SignalDatabase.threads.pinConversations(Collections.singleton(threadId))
+      }
+
+      backupState.chatIdToLocalRecipientId[chat.id] = recipientId
+      backupState.chatIdToLocalThreadId[chat.id] = threadId
+      backupState.chatIdToBackupRecipientId[chat.id] = chat.recipientId
+    } else {
+      Log.w(TAG, "Recipient doesnt exist with id $recipientId")
+    }
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/processor/ChatItemBackupProcessor.kt

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.processor
+
+import org.signal.core.util.logging.Log
+import org.thoughtcrime.securesms.backup.v2.BackupState
+import org.thoughtcrime.securesms.backup.v2.database.ChatItemImportInserter
+import org.thoughtcrime.securesms.backup.v2.database.createChatItemInserter
+import org.thoughtcrime.securesms.backup.v2.database.getMessagesForBackup
+import org.thoughtcrime.securesms.backup.v2.proto.Frame
+import org.thoughtcrime.securesms.backup.v2.stream.BackupFrameEmitter
+import org.thoughtcrime.securesms.database.SignalDatabase
+
+object ChatItemBackupProcessor {
+  val TAG = Log.tag(ChatItemBackupProcessor::class.java)
+
+  fun export(emitter: BackupFrameEmitter) {
+    SignalDatabase.messages.getMessagesForBackup().use { chatItems ->
+      for (chatItem in chatItems) {
+        emitter.emit(Frame(chatItem = chatItem))
+      }
+    }
+  }
+
+  fun beginImport(backupState: BackupState): ChatItemImportInserter {
+    return SignalDatabase.messages.createChatItemInserter(backupState)
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/processor/RecipientBackupProcessor.kt

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.processor
+
+import org.signal.core.util.logging.Log
+import org.thoughtcrime.securesms.backup.v2.BackupState
+import org.thoughtcrime.securesms.backup.v2.database.getAllForBackup
+import org.thoughtcrime.securesms.backup.v2.database.getContactsForBackup
+import org.thoughtcrime.securesms.backup.v2.database.getGroupsForBackup
+import org.thoughtcrime.securesms.backup.v2.database.restoreRecipientFromBackup
+import org.thoughtcrime.securesms.backup.v2.proto.Frame
+import org.thoughtcrime.securesms.backup.v2.stream.BackupFrameEmitter
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.recipients.Recipient
+
+typealias BackupRecipient = org.thoughtcrime.securesms.backup.v2.proto.Recipient
+
+object RecipientBackupProcessor {
+
+  val TAG = Log.tag(RecipientBackupProcessor::class.java)
+
+  fun export(emitter: BackupFrameEmitter) {
+    val selfId = Recipient.self().id.toLong()
+
+    SignalDatabase.recipients.getContactsForBackup(selfId).use { reader ->
+      for (backupRecipient in reader) {
+        emitter.emit(Frame(recipient = backupRecipient))
+      }
+    }
+
+    SignalDatabase.recipients.getGroupsForBackup().use { reader ->
+      for (backupRecipient in reader) {
+        emitter.emit(Frame(recipient = backupRecipient))
+      }
+    }
+
+    SignalDatabase.distributionLists.getAllForBackup().forEach {
+      emitter.emit(Frame(recipient = it))
+    }
+  }
+
+  fun import(recipient: BackupRecipient, backupState: BackupState) {
+    val newId = SignalDatabase.recipients.restoreRecipientFromBackup(recipient, backupState)
+    if (newId != null) {
+      backupState.backupToLocalRecipientId[recipient.id] = newId
+    }
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/stream/BackupExportStream.kt

@@ -0,0 +1,12 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.stream
+
+import org.thoughtcrime.securesms.backup.v2.proto.Frame
+
+interface BackupExportStream {
+  fun write(frame: Frame)
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/stream/BackupFrameEmitter.kt

@@ -0,0 +1,15 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.stream
+
+import org.thoughtcrime.securesms.backup.v2.proto.Frame
+
+/**
+ * An interface that lets sub-processors emit [Frame]s as they export data.
+ */
+fun interface BackupFrameEmitter {
+  fun emit(frame: Frame)
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/stream/BackupImportStream.kt

@@ -0,0 +1,12 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.stream
+
+import org.thoughtcrime.securesms.backup.v2.proto.Frame
+
+interface BackupImportStream {
+  fun read(): Frame?
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/stream/PlainTextBackupExportStream.kt

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.stream
+
+import org.signal.core.util.Conversions
+import org.thoughtcrime.securesms.backup.v2.proto.Frame
+import java.io.IOException
+import java.io.OutputStream
+
+/**
+ * Writes backup frames to the wrapped stream in plain text. Only for testing!
+ */
+class PlainTextBackupExportStream(private val outputStream: OutputStream) : BackupExportStream {
+
+  @Throws(IOException::class)
+  override fun write(frame: Frame) {
+    val frameBytes: ByteArray = frame.encode()
+    val lengthBytes: ByteArray = Conversions.intToByteArray(frameBytes.size)
+
+    outputStream.write(lengthBytes)
+    outputStream.write(frameBytes)
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/backup/v2/stream/PlainTextBackupImportStream.kt

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup.v2.stream
+
+import org.signal.core.util.Conversions
+import org.signal.core.util.readNBytesOrThrow
+import org.thoughtcrime.securesms.backup.v2.proto.Frame
+import java.io.EOFException
+import java.io.InputStream
+
+/**
+ * Reads a plaintext backup import stream one frame at a time.
+ */
+class PlainTextBackupImportStream(val inputStream: InputStream) : BackupImportStream, Iterator<Frame> {
+
+  var next: Frame? = null
+
+  init {
+    next = read()
+  }
+
+  override fun hasNext(): Boolean {
+    return next != null
+  }
+
+  override fun next(): Frame {
+    next?.let { out ->
+      next = read()
+      return out
+    } ?: throw NoSuchElementException()
+  }
+
+  override fun read(): Frame? {
+    try {
+      val lengthBytes: ByteArray = inputStream.readNBytesOrThrow(4)
+      val length = Conversions.byteArrayToInt(lengthBytes)
+
+      val frameBytes: ByteArray = inputStream.readNBytesOrThrow(length)
+      val frame: Frame = Frame.ADAPTER.decode(frameBytes)
+
+      return frame
+    } catch (e: EOFException) {
+      return null
+    }
+  }
+}