Setup infra for better archive upload progress tracking.

app/src/main/java/org/thoughtcrime/securesms/backup/ArchiveUploadProgress.kt

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.backup
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.shareIn
+import org.signal.core.util.throttleLatest
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.keyvalue.SignalStore
+import org.thoughtcrime.securesms.keyvalue.protos.ArchiveUploadProgressState
+import kotlin.math.max
+import kotlin.time.Duration.Companion.milliseconds
+
+/**
+ * Tracks the progress of uploading your message archive and provides an observable stream of results.
+ */
+object ArchiveUploadProgress {
+
+  private val PROGRESS_NONE = ArchiveUploadProgressState(
+    state = ArchiveUploadProgressState.State.None
+  )
+
+  private val _progress: MutableSharedFlow<Unit> = MutableSharedFlow(replay = 1)
+
+  private var uploadProgress: ArchiveUploadProgressState = SignalStore.backup.archiveUploadState ?: PROGRESS_NONE
+
+  /**
+   * Observe this to get updates on the current upload progress.
+   */
+  val progress: Flow<ArchiveUploadProgressState> = _progress
+    .throttleLatest(500.milliseconds)
+    .map {
+      if (uploadProgress.state != ArchiveUploadProgressState.State.UploadingAttachments) {
+        return@map uploadProgress
+      }
+
+      val pendingCount = SignalDatabase.attachments.getPendingArchiveUploadCount()
+      if (pendingCount == uploadProgress.totalAttachments) {
+        return@map PROGRESS_NONE
+      }
+
+      // It's possible that new attachments may be pending upload after we start a backup.
+      // If we wanted the most accurate progress possible, we could maintain a new database flag that indicates whether an attachment has been flagged as part
+      // of the current upload batch. However, this gets us pretty close while keeping things simple and not having to juggle extra flags, with the caveat that
+      // the progress bar may occasionally be including media that is not actually referenced in the active backup file.
+      val totalCount = max(uploadProgress.totalAttachments, pendingCount)
+
+      ArchiveUploadProgressState(
+        state = ArchiveUploadProgressState.State.UploadingAttachments,
+        completedAttachments = totalCount - pendingCount,
+        totalAttachments = totalCount
+      )
+    }
+    .onEach {
+      updateState(it, notify = false)
+    }
+    .flowOn(Dispatchers.IO)
+    .shareIn(scope = CoroutineScope(Dispatchers.IO), started = SharingStarted.WhileSubscribed(), replay = 1)
+
+  val inProgress
+    get() = uploadProgress.state != ArchiveUploadProgressState.State.None
+
+  fun begin() {
+    updateState(
+      ArchiveUploadProgressState(
+        state = ArchiveUploadProgressState.State.BackingUpMessages
+      )
+    )
+  }
+
+  fun onMessageBackupCreated() {
+    updateState(
+      ArchiveUploadProgressState(
+        state = ArchiveUploadProgressState.State.UploadingMessages
+      )
+    )
+  }
+
+  fun onAttachmentsStarted(attachmentCount: Long) {
+    updateState(
+      ArchiveUploadProgressState(
+        state = ArchiveUploadProgressState.State.UploadingAttachments,
+        completedAttachments = 0,
+        totalAttachments = attachmentCount
+      )
+    )
+  }
+
+  fun onAttachmentFinished() {
+    _progress.tryEmit(Unit)
+  }
+
+  fun onMessageBackupFinishedEarly() {
+    updateState(PROGRESS_NONE)
+  }
+
+  private fun updateState(state: ArchiveUploadProgressState, notify: Boolean = true) {
+    uploadProgress = state
+    SignalStore.backup.archiveUploadState = state
+
+    if (notify) {
+      _progress.tryEmit(Unit)
+    }
+  }
+}

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

@@ -5,14 +5,6 @@
 
 package org.thoughtcrime.securesms.backup.v2
 
-class BackupV2Event(val type: Type, val count: Long, val estimatedTotalCount: Long) {
-  enum class Type {
-    PROGRESS_MESSAGES,
-    PROGRESS_ATTACHMENTS,
-    FINISHED
-  }
-}
-
 class LocalBackupV2Event(val type: Type, val count: Long = 0, val estimatedTotalCount: Long = 0) {
   enum class Type {
     PROGRESS_ACCOUNT,

app/src/main/java/org/thoughtcrime/securesms/components/settings/app/chats/backups/RemoteBackupsSettingsFragment.kt

@@ -49,9 +49,6 @@ import androidx.compose.ui.window.DialogProperties
 import androidx.fragment.app.setFragmentResultListener
 import androidx.navigation.fragment.findNavController
 import androidx.navigation.fragment.navArgs
-import org.greenrobot.eventbus.EventBus
-import org.greenrobot.eventbus.Subscribe
-import org.greenrobot.eventbus.ThreadMode
 import org.signal.core.ui.Buttons
 import org.signal.core.ui.Dialogs
 import org.signal.core.ui.Dividers
@@ -63,15 +60,15 @@ import org.signal.core.ui.Snackbars
 import org.signal.core.ui.Texts
 import org.signal.core.util.money.FiatMoney
 import org.thoughtcrime.securesms.R
+import org.thoughtcrime.securesms.backup.ArchiveUploadProgress
 import org.thoughtcrime.securesms.backup.v2.BackupFrequency
-import org.thoughtcrime.securesms.backup.v2.BackupV2Event
 import org.thoughtcrime.securesms.backup.v2.ui.subscription.MessageBackupsType
 import org.thoughtcrime.securesms.components.settings.app.chats.backups.type.BackupsTypeSettingsFragment
 import org.thoughtcrime.securesms.components.settings.app.subscription.MessageBackupsCheckoutLauncher.createBackupsCheckoutLauncher
 import org.thoughtcrime.securesms.compose.ComposeFragment
-import org.thoughtcrime.securesms.conversation.v2.registerForLifecycle
 import org.thoughtcrime.securesms.database.model.InAppPaymentSubscriberRecord
 import org.thoughtcrime.securesms.keyvalue.SignalStore
+import org.thoughtcrime.securesms.keyvalue.protos.ArchiveUploadProgressState
 import org.thoughtcrime.securesms.payments.FiatMoneyUtil
 import org.thoughtcrime.securesms.util.DateUtils
 import org.thoughtcrime.securesms.util.Util
@@ -96,6 +93,7 @@ class RemoteBackupsSettingsFragment : ComposeFragment() {
   @Composable
   override fun FragmentContent() {
     val state by viewModel.state.collectAsState()
+    val backupProgress by ArchiveUploadProgress.progress.collectAsState(initial = null)
     val callbacks = remember { Callbacks() }
 
     RemoteBackupsSettingsContent(
@@ -106,7 +104,7 @@ class RemoteBackupsSettingsFragment : ComposeFragment() {
       requestedDialog = state.dialog,
       requestedSnackbar = state.snackbar,
       contentCallbacks = callbacks,
-      backupProgress = state.backupProgress,
+      backupProgress = backupProgress,
       backupSize = state.backupSize
     )
   }
@@ -162,14 +160,8 @@ class RemoteBackupsSettingsFragment : ComposeFragment() {
     }
   }
 
-  @Subscribe(threadMode = ThreadMode.MAIN, sticky = true)
-  fun onEvent(backupEvent: BackupV2Event) {
-    viewModel.updateBackupProgress(backupEvent)
-  }
-
   override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
     super.onViewCreated(view, savedInstanceState)
-    EventBus.getDefault().registerForLifecycle(subscriber = this, lifecycleOwner = viewLifecycleOwner)
     checkoutLauncher = createBackupsCheckoutLauncher { backUpLater ->
       if (backUpLater) {
         viewModel.requestSnackbar(RemoteBackupsSettingsState.Snackbar.BACKUP_WILL_BE_CREATED_OVERNIGHT)
@@ -221,7 +213,7 @@ private fun RemoteBackupsSettingsContent(
   requestedDialog: RemoteBackupsSettingsState.Dialog,
   requestedSnackbar: RemoteBackupsSettingsState.Snackbar,
   contentCallbacks: ContentCallbacks,
-  backupProgress: BackupV2Event?,
+  backupProgress: ArchiveUploadProgressState?,
   backupSize: Long
 ) {
   val snackbarHostState = remember {
@@ -264,7 +256,7 @@ private fun RemoteBackupsSettingsContent(
           Texts.SectionHeader(text = stringResource(id = R.string.RemoteBackupsSettingsFragment__backup_details))
         }
 
-        if (backupProgress == null || backupProgress.type == BackupV2Event.Type.FINISHED) {
+        if (backupProgress == null || backupProgress.state == ArchiveUploadProgressState.State.None) {
           item {
             LastBackupRow(
               lastBackupTimestamp = lastBackupTimestamp,
@@ -273,7 +265,7 @@ private fun RemoteBackupsSettingsContent(
           }
         } else {
           item {
-            InProgressBackupRow(progress = backupProgress.count.toInt(), totalProgress = backupProgress.estimatedTotalCount.toInt())
+            InProgressBackupRow(progress = backupProgress.completedAttachments.toInt(), totalProgress = backupProgress.totalAttachments.toInt())
           }
         }
 

app/src/main/java/org/thoughtcrime/securesms/components/settings/app/chats/backups/RemoteBackupsSettingsState.kt

@@ -6,7 +6,6 @@
 package org.thoughtcrime.securesms.components.settings.app.chats.backups
 
 import org.thoughtcrime.securesms.backup.v2.BackupFrequency
-import org.thoughtcrime.securesms.backup.v2.BackupV2Event
 import org.thoughtcrime.securesms.backup.v2.ui.subscription.MessageBackupsType
 
 data class RemoteBackupsSettingsState(
@@ -16,8 +15,7 @@ data class RemoteBackupsSettingsState(
   val backupsFrequency: BackupFrequency = BackupFrequency.DAILY,
   val lastBackupTimestamp: Long = 0,
   val dialog: Dialog = Dialog.NONE,
-  val snackbar: Snackbar = Snackbar.NONE,
-  val backupProgress: BackupV2Event? = null
+  val snackbar: Snackbar = Snackbar.NONE
 ) {
   enum class Dialog {
     NONE,

app/src/main/java/org/thoughtcrime/securesms/components/settings/app/chats/backups/RemoteBackupsSettingsViewModel.kt

@@ -17,7 +17,6 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
 import org.thoughtcrime.securesms.backup.v2.BackupFrequency
 import org.thoughtcrime.securesms.backup.v2.BackupRepository
-import org.thoughtcrime.securesms.backup.v2.BackupV2Event
 import org.thoughtcrime.securesms.dependencies.AppDependencies
 import org.thoughtcrime.securesms.jobs.BackupMessagesJob
 import org.thoughtcrime.securesms.keyvalue.SignalStore
@@ -28,7 +27,7 @@ import kotlin.time.Duration.Companion.milliseconds
  * ViewModel for state management of RemoteBackupsSettingsFragment
  */
 class RemoteBackupsSettingsViewModel : ViewModel() {
-  private val internalState = MutableStateFlow(
+  private val _state = MutableStateFlow(
     RemoteBackupsSettingsState(
       messageBackupsType = null,
       lastBackupTimestamp = SignalStore.backup.lastBackupTime,
@@ -37,7 +36,7 @@ class RemoteBackupsSettingsViewModel : ViewModel() {
     )
   )
 
-  val state: StateFlow<RemoteBackupsSettingsState> = internalState
+  val state: StateFlow<RemoteBackupsSettingsState> = _state
 
   init {
     refresh()
@@ -45,22 +44,22 @@ class RemoteBackupsSettingsViewModel : ViewModel() {
 
   fun setCanBackUpUsingCellular(canBackUpUsingCellular: Boolean) {
     SignalStore.backup.backupWithCellular = canBackUpUsingCellular
-    internalState.update { it.copy(canBackUpUsingCellular = canBackUpUsingCellular) }
+    _state.update { it.copy(canBackUpUsingCellular = canBackUpUsingCellular) }
   }
 
   fun setBackupsFrequency(backupsFrequency: BackupFrequency) {
     SignalStore.backup.backupFrequency = backupsFrequency
-    internalState.update { it.copy(backupsFrequency = backupsFrequency) }
+    _state.update { it.copy(backupsFrequency = backupsFrequency) }
     MessageBackupListener.setNextBackupTimeToIntervalFromNow()
     MessageBackupListener.schedule(AppDependencies.application)
   }
 
   fun requestDialog(dialog: RemoteBackupsSettingsState.Dialog) {
-    internalState.update { it.copy(dialog = dialog) }
+    _state.update { it.copy(dialog = dialog) }
   }
 
   fun requestSnackbar(snackbar: RemoteBackupsSettingsState.Snackbar) {
-    internalState.update { it.copy(snackbar = snackbar) }
+    _state.update { it.copy(snackbar = snackbar) }
   }
 
   fun refresh() {
@@ -68,7 +67,7 @@ class RemoteBackupsSettingsViewModel : ViewModel() {
       val tier = SignalStore.backup.backupTier
       val backupType = if (tier != null) BackupRepository.getBackupsType(tier) else null
 
-      internalState.update {
+      _state.update {
         it.copy(
           messageBackupsType = backupType,
           lastBackupTimestamp = SignalStore.backup.lastBackupTime,
@@ -96,13 +95,8 @@ class RemoteBackupsSettingsViewModel : ViewModel() {
     }
   }
 
-  fun updateBackupProgress(backupEvent: BackupV2Event?) {
-    internalState.update { it.copy(backupProgress = backupEvent) }
-    refreshBackupState()
-  }
-
   private fun refreshBackupState() {
-    internalState.update {
+    _state.update {
       it.copy(
         lastBackupTimestamp = SignalStore.backup.lastBackupTime,
         backupSize = SignalStore.backup.totalBackupSize
@@ -111,8 +105,6 @@ class RemoteBackupsSettingsViewModel : ViewModel() {
   }
 
   fun onBackupNowClick() {
-    if (internalState.value.backupProgress == null || internalState.value.backupProgress?.type == BackupV2Event.Type.FINISHED) {
-      BackupMessagesJob.enqueue()
-    }
+    BackupMessagesJob.enqueue()
   }
 }

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

@@ -286,7 +286,8 @@ class AttachmentTable(
       "CREATE INDEX IF NOT EXISTS attachment_data_hash_start_index ON $TABLE_NAME ($DATA_HASH_START);",
       "CREATE INDEX IF NOT EXISTS attachment_data_hash_end_index ON $TABLE_NAME ($DATA_HASH_END);",
       "CREATE INDEX IF NOT EXISTS attachment_data_index ON $TABLE_NAME ($DATA_FILE);",
-      "CREATE INDEX IF NOT EXISTS attachment_archive_media_id_index ON $TABLE_NAME ($ARCHIVE_MEDIA_ID);"
+      "CREATE INDEX IF NOT EXISTS attachment_archive_media_id_index ON $TABLE_NAME ($ARCHIVE_MEDIA_ID);",
+      "CREATE INDEX IF NOT EXISTS attachment_archive_transfer_state ON $TABLE_NAME ($ARCHIVE_TRANSFER_STATE);"
     )
 
     @JvmStatic
@@ -629,6 +630,18 @@ class AttachmentTable(
     }
   }
 
+  /**
+   * Returns the number of attachments that are in pending upload states to the archive cdn.
+   */
+  fun getPendingArchiveUploadCount(): Long {
+    return readableDatabase
+      .count()
+      .from(TABLE_NAME)
+      .where("$ARCHIVE_TRANSFER_STATE IN (${ArchiveTransferState.UPLOAD_IN_PROGRESS.value}, ${ArchiveTransferState.COPY_PENDING.value})")
+      .run()
+      .readToSingleLong()
+  }
+
   fun deleteAttachmentsForMessage(mmsId: Long): Boolean {
     Log.d(TAG, "[deleteAttachmentsForMessage] mmsId: $mmsId")
 

app/src/main/java/org/thoughtcrime/securesms/database/helpers/SignalDatabaseMigrations.kt

@@ -106,6 +106,7 @@ import org.thoughtcrime.securesms.database.helpers.migration.V244_AttachmentRemo
 import org.thoughtcrime.securesms.database.helpers.migration.V245_DeletionTimestampOnCallLinks
 import org.thoughtcrime.securesms.database.helpers.migration.V246_DropThumbnailCdnFromAttachments
 import org.thoughtcrime.securesms.database.helpers.migration.V247_ClearUploadTimestamp
+import org.thoughtcrime.securesms.database.helpers.migration.V248_ArchiveTransferStateIndex
 
 /**
  * Contains all of the database migrations for [SignalDatabase]. Broken into a separate file for cleanliness.
@@ -213,10 +214,11 @@ object SignalDatabaseMigrations {
     244 to V244_AttachmentRemoteIv,
     245 to V245_DeletionTimestampOnCallLinks,
     246 to V246_DropThumbnailCdnFromAttachments,
-    247 to V247_ClearUploadTimestamp
+    247 to V247_ClearUploadTimestamp,
+    248 to V248_ArchiveTransferStateIndex
   )
 
-  const val DATABASE_VERSION = 247
+  const val DATABASE_VERSION = 248
 
   @JvmStatic
   fun migrate(context: Application, db: SQLiteDatabase, oldVersion: Int, newVersion: Int) {

app/src/main/java/org/thoughtcrime/securesms/database/helpers/migration/V248_ArchiveTransferStateIndex.kt

@@ -0,0 +1,19 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.database.helpers.migration
+
+import android.app.Application
+import net.zetetic.database.sqlcipher.SQLiteDatabase
+
+/**
+ * Adds an index to improve the perf of counting and filtering attachment rows by their transfer state.
+ */
+@Suppress("ClassName")
+object V248_ArchiveTransferStateIndex : SignalDatabaseMigration {
+  override fun migrate(context: Application, db: SQLiteDatabase, oldVersion: Int, newVersion: Int) {
+    db.execSQL("CREATE INDEX IF NOT EXISTS attachment_archive_transfer_state ON attachment (archive_transfer_state)")
+  }
+}

app/src/main/java/org/thoughtcrime/securesms/jobs/ArchiveAttachmentBackfillJob.kt

@@ -5,13 +5,11 @@
 
 package org.thoughtcrime.securesms.jobs
 
-import org.greenrobot.eventbus.EventBus
 import org.signal.core.util.logging.Log
-import org.thoughtcrime.securesms.backup.v2.BackupV2Event
+import org.thoughtcrime.securesms.backup.ArchiveUploadProgress
 import org.thoughtcrime.securesms.database.SignalDatabase
 import org.thoughtcrime.securesms.dependencies.AppDependencies
 import org.thoughtcrime.securesms.jobmanager.Job
-import org.thoughtcrime.securesms.keyvalue.SignalStore
 import kotlin.time.Duration.Companion.days
 
 /**
@@ -40,14 +38,11 @@ class ArchiveAttachmentBackfillJob private constructor(parameters: Parameters) :
 
   override fun run(): Result {
     val jobs = SignalDatabase.attachments.getAttachmentsThatNeedArchiveUpload()
-      .map { attachmentId -> UploadAttachmentToArchiveJob(attachmentId, forBackfill = true) }
+      .map { attachmentId -> UploadAttachmentToArchiveJob(attachmentId) }
 
     SignalDatabase.attachments.createKeyIvDigestForAttachmentsThatNeedArchiveUpload()
 
-    SignalStore.backup.totalAttachmentUploadCount = jobs.size.toLong()
-    SignalStore.backup.currentAttachmentUploadCount = 0
-
-    EventBus.getDefault().postSticky(BackupV2Event(BackupV2Event.Type.PROGRESS_ATTACHMENTS, count = 0, estimatedTotalCount = jobs.size.toLong()))
+    ArchiveUploadProgress.onAttachmentsStarted(jobs.size.toLong())
 
     Log.i(TAG, "Adding ${jobs.size} jobs to backfill attachments.")
     AppDependencies.jobManager.addAll(jobs)

app/src/main/java/org/thoughtcrime/securesms/jobs/BackupMessagesJob.kt

@@ -5,11 +5,10 @@
 
 package org.thoughtcrime.securesms.jobs
 
-import org.greenrobot.eventbus.EventBus
 import org.signal.core.util.Stopwatch
 import org.signal.core.util.logging.Log
+import org.thoughtcrime.securesms.backup.ArchiveUploadProgress
 import org.thoughtcrime.securesms.backup.v2.BackupRepository
-import org.thoughtcrime.securesms.backup.v2.BackupV2Event
 import org.thoughtcrime.securesms.database.SignalDatabase
 import org.thoughtcrime.securesms.dependencies.AppDependencies
 import org.thoughtcrime.securesms.jobmanager.Job
@@ -70,13 +69,15 @@ class BackupMessagesJob private constructor(parameters: Parameters) : Job(parame
     SignalDatabase.attachments.createKeyIvDigestForAttachmentsThatNeedArchiveUpload().takeIf { it > 0 }?.let { count -> Log.w(TAG, "Needed to create $count key/iv/digests.") }
     stopwatch.split("key-iv-digest")
 
-    EventBus.getDefault().postSticky(BackupV2Event(type = BackupV2Event.Type.PROGRESS_MESSAGES, count = 0, estimatedTotalCount = 0))
+    ArchiveUploadProgress.begin()
     val tempBackupFile = BlobProvider.getInstance().forNonAutoEncryptingSingleSessionOnDisk(AppDependencies.application)
 
     val outputStream = FileOutputStream(tempBackupFile)
     BackupRepository.export(outputStream = outputStream, append = { tempBackupFile.appendBytes(it) }, plaintext = false)
     stopwatch.split("export")
 
+    ArchiveUploadProgress.onMessageBackupCreated()
+
     FileInputStream(tempBackupFile).use {
       when (val result = BackupRepository.uploadBackupFile(it, tempBackupFile.length())) {
         is NetworkResult.Success -> Log.i(TAG, "Successfully uploaded backup file.")
@@ -95,7 +96,7 @@ class BackupMessagesJob private constructor(parameters: Parameters) : Job(parame
     SignalStore.backup.lastBackupTime = System.currentTimeMillis()
     SignalStore.backup.usedBackupMediaSpace = when (val result = BackupRepository.getRemoteBackupUsedSpace()) {
       is NetworkResult.Success -> result.result ?: 0
-      is NetworkResult.NetworkError -> SignalStore.backup.usedBackupMediaSpace // TODO enqueue a secondary job to fetch the latest number -- no need to fail this one
+      is NetworkResult.NetworkError -> SignalStore.backup.usedBackupMediaSpace // TODO [backup] enqueue a secondary job to fetch the latest number -- no need to fail this one
       is NetworkResult.StatusCodeError -> {
         Log.w(TAG, "Failed to get used space: ${result.code}")
         SignalStore.backup.usedBackupMediaSpace
@@ -110,7 +111,7 @@ class BackupMessagesJob private constructor(parameters: Parameters) : Job(parame
       AppDependencies.jobManager.add(ArchiveAttachmentBackfillJob())
     } else {
       Log.i(TAG, "No attachments need to be uploaded, we can finish.")
-      EventBus.getDefault().postSticky(BackupV2Event(BackupV2Event.Type.FINISHED, 0, 0))
+      ArchiveUploadProgress.onMessageBackupFinishedEarly()
     }
 
     return Result.success()

app/src/main/java/org/thoughtcrime/securesms/jobs/CopyAttachmentToArchiveJob.kt

@@ -1,12 +1,11 @@
 package org.thoughtcrime.securesms.jobs
 
-import org.greenrobot.eventbus.EventBus
 import org.signal.core.util.logging.Log
 import org.thoughtcrime.securesms.attachments.AttachmentId
 import org.thoughtcrime.securesms.attachments.Cdn
 import org.thoughtcrime.securesms.attachments.DatabaseAttachment
+import org.thoughtcrime.securesms.backup.ArchiveUploadProgress
 import org.thoughtcrime.securesms.backup.v2.BackupRepository
-import org.thoughtcrime.securesms.backup.v2.BackupV2Event
 import org.thoughtcrime.securesms.database.AttachmentTable
 import org.thoughtcrime.securesms.database.SignalDatabase
 import org.thoughtcrime.securesms.dependencies.AppDependencies
@@ -24,7 +23,7 @@ import java.util.concurrent.TimeUnit
  * Copies and re-encrypts attachments from the attachment cdn to the archive cdn.
  * If it's discovered that the attachment no longer exists on the attachment cdn, this job will schedule a re-upload via [UploadAttachmentToArchiveJob].
  */
-class CopyAttachmentToArchiveJob private constructor(private val attachmentId: AttachmentId, private val forBackfill: Boolean, parameters: Parameters) : Job(parameters) {
+class CopyAttachmentToArchiveJob private constructor(private val attachmentId: AttachmentId, parameters: Parameters) : Job(parameters) {
 
   companion object {
     private val TAG = Log.tag(CopyAttachmentToArchiveJob::class.java)
@@ -35,9 +34,8 @@ class CopyAttachmentToArchiveJob private constructor(private val attachmentId: A
     val ALLOWED_SOURCE_CDNS = setOf(Cdn.CDN_2, Cdn.CDN_3)
   }
 
-  constructor(attachmentId: AttachmentId, forBackfill: Boolean = false) : this(
+  constructor(attachmentId: AttachmentId) : this(
     attachmentId = attachmentId,
-    forBackfill = forBackfill,
     parameters = Parameters.Builder()
       .addConstraint(NetworkConstraint.KEY)
       .setLifespan(TimeUnit.DAYS.toMillis(1))
@@ -47,8 +45,7 @@ class CopyAttachmentToArchiveJob private constructor(private val attachmentId: A
   )
 
   override fun serialize(): ByteArray = CopyAttachmentToArchiveJobData(
-    attachmentId = attachmentId.id,
-    forBackfill = forBackfill
+    attachmentId = attachmentId.id
   ).encode()
 
   override fun getFactoryKey(): String = KEY
@@ -139,32 +136,14 @@ class CopyAttachmentToArchiveJob private constructor(private val attachmentId: A
       ArchiveThumbnailUploadJob.enqueueIfNecessary(attachmentId)
       SignalStore.backup.usedBackupMediaSpace += AttachmentCipherStreamUtil.getCiphertextLength(PaddingInputStream.getPaddedSize(attachment.size))
 
-      incrementBackfillProgressIfNecessary()
+      ArchiveUploadProgress.onAttachmentFinished()
     }
 
     return result
   }
 
   override fun onFailure() {
-    incrementBackfillProgressIfNecessary()
-  }
-
-  private fun incrementBackfillProgressIfNecessary() {
-    if (!forBackfill) {
-      return
-    }
-
-    if (SignalStore.backup.totalAttachmentUploadCount > 0) {
-      SignalStore.backup.currentAttachmentUploadCount++
-
-      if (SignalStore.backup.currentAttachmentUploadCount >= SignalStore.backup.totalAttachmentUploadCount) {
-        EventBus.getDefault().postSticky(BackupV2Event(BackupV2Event.Type.FINISHED, count = 0, estimatedTotalCount = 0))
-        SignalStore.backup.currentAttachmentUploadCount = 0
-        SignalStore.backup.totalAttachmentUploadCount = 0
-      } else {
-        EventBus.getDefault().postSticky(BackupV2Event(BackupV2Event.Type.PROGRESS_ATTACHMENTS, count = SignalStore.backup.currentAttachmentUploadCount, estimatedTotalCount = SignalStore.backup.totalAttachmentUploadCount))
-      }
-    }
+    ArchiveUploadProgress.onAttachmentFinished()
   }
 
   class Factory : Job.Factory<CopyAttachmentToArchiveJob> {
@@ -172,7 +151,6 @@ class CopyAttachmentToArchiveJob private constructor(private val attachmentId: A
       val jobData = CopyAttachmentToArchiveJobData.ADAPTER.decode(serializedData!!)
       return CopyAttachmentToArchiveJob(
         attachmentId = AttachmentId(jobData.attachmentId),
-        forBackfill = jobData.forBackfill,
         parameters = parameters
       )
     }

app/src/main/java/org/thoughtcrime/securesms/jobs/UploadAttachmentToArchiveJob.kt

@@ -33,7 +33,6 @@ import kotlin.time.Duration.Companion.days
 class UploadAttachmentToArchiveJob private constructor(
   private val attachmentId: AttachmentId,
   private var uploadSpec: ResumableUpload?,
-  private val forBackfill: Boolean,
   parameters: Parameters
 ) : Job(parameters) {
 
@@ -44,10 +43,9 @@ class UploadAttachmentToArchiveJob private constructor(
     fun buildQueueKey(attachmentId: AttachmentId) = "ArchiveAttachmentJobs_${attachmentId.id}"
   }
 
-  constructor(attachmentId: AttachmentId, forBackfill: Boolean = false) : this(
+  constructor(attachmentId: AttachmentId) : this(
     attachmentId = attachmentId,
     uploadSpec = null,
-    forBackfill = forBackfill,
     parameters = Parameters.Builder()
       .addConstraint(NetworkConstraint.KEY)
       .setLifespan(30.days.inWholeMilliseconds)
@@ -57,8 +55,7 @@ class UploadAttachmentToArchiveJob private constructor(
   )
 
   override fun serialize(): ByteArray = UploadAttachmentToArchiveJobData(
-    attachmentId = attachmentId.id,
-    forBackfill = forBackfill
+    attachmentId = attachmentId.id
   ).encode()
 
   override fun getFactoryKey(): String = KEY
@@ -97,7 +94,7 @@ class UploadAttachmentToArchiveJob private constructor(
 
     if (attachment.archiveTransferState == AttachmentTable.ArchiveTransferState.COPY_PENDING) {
       Log.i(TAG, "[$attachmentId] Already marked as pending transfer. Enqueueing a copy job just in case.")
-      AppDependencies.jobManager.add(CopyAttachmentToArchiveJob(attachment.attachmentId, forBackfill = forBackfill))
+      AppDependencies.jobManager.add(CopyAttachmentToArchiveJob(attachment.attachmentId))
       return Result.success()
     }
 
@@ -154,7 +151,7 @@ class UploadAttachmentToArchiveJob private constructor(
 
     SignalDatabase.attachments.finalizeAttachmentAfterUpload(attachment.attachmentId, uploadResult)
 
-    AppDependencies.jobManager.add(CopyAttachmentToArchiveJob(attachment.attachmentId, forBackfill = forBackfill))
+    AppDependencies.jobManager.add(CopyAttachmentToArchiveJob(attachment.attachmentId))
 
     return Result.success()
   }
@@ -207,7 +204,6 @@ class UploadAttachmentToArchiveJob private constructor(
       return UploadAttachmentToArchiveJob(
         attachmentId = AttachmentId(data.attachmentId),
         uploadSpec = data.uploadSpec,
-        forBackfill = data.forBackfill,
         parameters = parameters
       )
     }

app/src/main/java/org/thoughtcrime/securesms/keyvalue/BackupValues.kt

@@ -5,6 +5,7 @@ import org.signal.core.util.logging.Log
 import org.thoughtcrime.securesms.backup.RestoreState
 import org.thoughtcrime.securesms.backup.v2.BackupFrequency
 import org.thoughtcrime.securesms.backup.v2.MessageBackupTier
+import org.thoughtcrime.securesms.keyvalue.protos.ArchiveUploadProgressState
 import org.whispersystems.signalservice.api.archive.ArchiveServiceCredential
 import org.whispersystems.signalservice.api.archive.GetArchiveCdnCredentialsResponse
 import org.whispersystems.signalservice.internal.util.JsonUtil
@@ -37,8 +38,7 @@ class BackupValues(store: KeyValueStore) : SignalStoreValues(store) {
     private const val KEY_OPTIMIZE_STORAGE = "backup.optimizeStorage"
     private const val KEY_BACKUPS_INITIALIZED = "backup.initialized"
 
-    private const val KEY_TOTAL_ATTACHMENTS_UPLOAD_COUNT = "backup.totalAttachmentsUploadCount"
-    private const val KEY_CURRENT_ATTACHMENT_UPLOAD_COUNT = "backup.currentAttachmentUploadCount"
+    private const val KEY_ARCHIVE_UPLOAD_STATE = "backup.archiveUploadState"
 
     /**
      * Specifies whether remote backups are enabled on this device.
@@ -70,14 +70,9 @@ class BackupValues(store: KeyValueStore) : SignalStoreValues(store) {
   var backupTier: MessageBackupTier? by enumValue(KEY_BACKUP_TIER, null, MessageBackupTier.Serializer)
 
   /**
-   * When uploading attachments to the archive CDN, this tracks the total number of attachments that are pending upload.
+   * When uploading a backup, we store the progress state here so that I can remain across app restarts.
    */
-  var totalAttachmentUploadCount: Long by longValue(KEY_TOTAL_ATTACHMENTS_UPLOAD_COUNT, 0)
-
-  /**
-   * When uploading attachments to the archive CDN, this tracks the total number of attachments that have currently been uploaded.
-   */
-  var currentAttachmentUploadCount: Long by longValue(KEY_CURRENT_ATTACHMENT_UPLOAD_COUNT, 0)
+  var archiveUploadState: ArchiveUploadProgressState? by protoValue(KEY_ARCHIVE_UPLOAD_STATE, ArchiveUploadProgressState.ADAPTER)
 
   val totalBackupSize: Long get() = lastBackupProtoSize + usedBackupMediaSpace
 

app/src/main/protowire/JobData.proto

@@ -127,11 +127,9 @@ message RestoreAttachmentJobData {
 
 message CopyAttachmentToArchiveJobData {
   uint64 attachmentId = 1;
-  bool forBackfill = 2;
 }
 
 message UploadAttachmentToArchiveJobData {
   uint64 attachmentId = 1;
   ResumableUpload uploadSpec = 2;
-  bool forBackfill = 3;
 }

app/src/main/protowire/KeyValue.proto

@@ -14,4 +14,17 @@ option java_package = "org.thoughtcrime.securesms.keyvalue.protos";
 message LeastActiveLinkedDevice {
     string name = 1;
     uint64 lastActiveTimestamp = 2;
+}
+
+message ArchiveUploadProgressState {
+    enum State {
+        None = 0;
+        BackingUpMessages = 1;
+        UploadingMessages = 2;
+        UploadingAttachments = 3;
+    }
+
+    State state = 1;
+    uint64 completedAttachments = 2;
+    uint64 totalAttachments = 3;
 }