Add Delete for Me sync support.

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

@@ -46,6 +46,7 @@ import org.signal.core.util.readToList
 import org.signal.core.util.readToSet
 import org.signal.core.util.readToSingleInt
 import org.signal.core.util.readToSingleLong
+import org.signal.core.util.readToSingleLongOrNull
 import org.signal.core.util.readToSingleObject
 import org.signal.core.util.requireBlob
 import org.signal.core.util.requireBoolean
@@ -449,6 +450,32 @@ open class MessageTable(context: Context?, databaseHelper: SignalDatabase) : Dat
         .joinToString(" OR ")
     }
 
+    /**
+     * A message that can be correctly identified with an author/sent timestamp across devices.
+     *
+     * Must be:
+     * - Incoming or sent outgoing
+     * - Secure or push
+     * - Not a group update
+     * - Not a key exchange message
+     * - Not an encryption message
+     * - Not a report spam message
+     * - Not a message rqeuest accepted message
+     * - Have a valid sent timestamp
+     * - Be a normal message or direct (1:1) story reply
+     */
+    private const val IS_ADDRESSABLE_CLAUSE = """
+      (($TYPE & ${MessageTypes.BASE_TYPE_MASK}) = ${MessageTypes.BASE_SENT_TYPE} OR ($TYPE & ${MessageTypes.BASE_TYPE_MASK}) = ${MessageTypes.BASE_INBOX_TYPE}) AND 
+      ($TYPE & (${MessageTypes.SECURE_MESSAGE_BIT} | ${MessageTypes.PUSH_MESSAGE_BIT})) != 0 AND 
+      ($TYPE & ${MessageTypes.GROUP_MASK}) = 0 AND 
+      ($TYPE & ${MessageTypes.KEY_EXCHANGE_MASK}) = 0 AND 
+      ($TYPE & ${MessageTypes.ENCRYPTION_MASK}) = 0 AND
+      ($TYPE & ${MessageTypes.SPECIAL_TYPES_MASK}) != ${MessageTypes.SPECIAL_TYPE_REPORTED_SPAM} AND
+      ($TYPE & ${MessageTypes.SPECIAL_TYPES_MASK}) != ${MessageTypes.SPECIAL_TYPE_MESSAGE_REQUEST_ACCEPTED} AND      
+      $DATE_SENT > 0 AND
+      $PARENT_STORY_ID <= 0
+    """
+
     @JvmStatic
     fun mmsReaderFor(cursor: Cursor): MmsReader {
       return MmsReader(cursor)
@@ -1722,6 +1749,33 @@ open class MessageTable(context: Context?, databaseHelper: SignalDatabase) : Dat
       .readToSingleLong(-1)
   }
 
+  fun getLatestReceivedAt(threadId: Long, messages: List<SyncMessageId>): Long? {
+    if (messages.isEmpty()) {
+      return null
+    }
+
+    val args: List<Array<String>> = messages.map { arrayOf(it.timetamp.toString(), it.recipientId.serialize(), threadId.toString()) }
+    val queries = SqlUtil.buildCustomCollectionQuery("$DATE_SENT = ? AND $FROM_RECIPIENT_ID = ? AND $THREAD_ID = ?", args)
+
+    var overallLatestReceivedAt: Long? = null
+    for (query in queries) {
+      val latestReceivedAt: Long? = readableDatabase
+        .select("MAX($DATE_RECEIVED)")
+        .from(TABLE_NAME)
+        .where(query.where, query.whereArgs)
+        .run()
+        .readToSingleLongOrNull()
+
+      if (overallLatestReceivedAt == null) {
+        overallLatestReceivedAt = latestReceivedAt
+      } else if (latestReceivedAt != null) {
+        overallLatestReceivedAt = max(overallLatestReceivedAt, latestReceivedAt)
+      }
+    }
+
+    return overallLatestReceivedAt
+  }
+
   fun getScheduledMessageCountForThread(threadId: Long): Int {
     return readableDatabase
       .select("COUNT(*)")
@@ -3200,7 +3254,8 @@ open class MessageTable(context: Context?, databaseHelper: SignalDatabase) : Dat
     return deleteMessage(messageId, threadId)
   }
 
-  private fun deleteMessage(messageId: Long, threadId: Long = getThreadIdForMessage(messageId), notify: Boolean = true, updateThread: Boolean = true): Boolean {
+  @VisibleForTesting
+  fun deleteMessage(messageId: Long, threadId: Long, notify: Boolean = true, updateThread: Boolean = true): Boolean {
     Log.d(TAG, "deleteMessage($messageId)")
 
     attachments.deleteAttachmentsForMessage(messageId)
@@ -3378,12 +3433,12 @@ open class MessageTable(context: Context?, databaseHelper: SignalDatabase) : Dat
 
     writableDatabase.withinTransaction { db ->
       SqlUtil.buildCollectionQuery(THREAD_ID, threadIds).forEach { query ->
-        db.select(ID)
+        db.select(ID, THREAD_ID)
           .from(TABLE_NAME)
           .where(query.where, query.whereArgs)
           .run()
           .forEach { cursor ->
-            deleteMessage(cursor.requireLong(ID), notify = false, updateThread = false)
+            deleteMessage(cursor.requireLong(ID), cursor.requireLong(THREAD_ID), notify = false, updateThread = false)
           }
       }
     }
@@ -3394,10 +3449,12 @@ open class MessageTable(context: Context?, databaseHelper: SignalDatabase) : Dat
     OptimizeMessageSearchIndexJob.enqueue()
   }
 
-  fun deleteMessagesInThreadBeforeDate(threadId: Long, date: Long): Int {
+  fun deleteMessagesInThreadBeforeDate(threadId: Long, date: Long, inclusive: Boolean): Int {
+    val condition = if (inclusive) "<=" else "<"
+
     return writableDatabase
       .delete(TABLE_NAME)
-      .where("$THREAD_ID = ? AND $DATE_RECEIVED < $date", threadId)
+      .where("$THREAD_ID = ? AND $DATE_RECEIVED $condition $date", threadId)
       .run()
   }
 
@@ -3423,6 +3480,48 @@ open class MessageTable(context: Context?, databaseHelper: SignalDatabase) : Dat
     }
   }
 
+  fun deleteMessages(messagesToDelete: List<MessageTable.SyncMessageId>): List<SyncMessageId> {
+    val threads = mutableSetOf<Long>()
+    val unhandled = mutableListOf<SyncMessageId>()
+
+    for (message in messagesToDelete) {
+      readableDatabase
+        .select(ID, THREAD_ID)
+        .from(TABLE_NAME)
+        .where("$DATE_SENT = ? AND $FROM_RECIPIENT_ID = ?", message.timetamp, message.recipientId)
+        .run()
+        .use {
+          if (it.moveToFirst()) {
+            val messageId = it.requireLong(ID)
+            val threadId = it.requireLong(THREAD_ID)
+
+            deleteMessage(
+              messageId = messageId,
+              threadId = threadId,
+              notify = false,
+              updateThread = false
+            )
+            threads += threadId
+          } else {
+            unhandled += message
+          }
+        }
+    }
+
+    threads
+      .forEach { threadId ->
+        SignalDatabase.threads.update(threadId, unarchive = false)
+        notifyConversationListeners(threadId)
+      }
+
+    notifyConversationListListeners()
+    notifyStickerListeners()
+    notifyStickerPackListeners()
+    OptimizeMessageSearchIndexJob.enqueue()
+
+    return unhandled
+  }
+
   private fun getMessagesInThreadAfterInclusive(threadId: Long, timestamp: Long, limit: Long): List<MessageRecord> {
     val where = "$TABLE_NAME.$THREAD_ID = ? AND $TABLE_NAME.$DATE_RECEIVED >= ? AND $TABLE_NAME.$SCHEDULED_DATE = -1 AND $TABLE_NAME.$LATEST_REVISION_ID IS NULL"
     val args = buildArgs(threadId, timestamp)
@@ -4863,6 +4962,48 @@ open class MessageTable(context: Context?, databaseHelper: SignalDatabase) : Dat
       .run()
   }
 
+  fun threadContainsAddressableMessages(threadId: Long): Boolean {
+    return readableDatabase
+      .exists(TABLE_NAME)
+      .where("$IS_ADDRESSABLE_CLAUSE AND $THREAD_ID = ?", threadId)
+      .run()
+  }
+
+  fun threadIsEmpty(threadId: Long): Boolean {
+    val hasMessages = readableDatabase
+      .exists(TABLE_NAME)
+      .where("$THREAD_ID = ?", threadId)
+      .run()
+
+    return !hasMessages
+  }
+
+  fun getMostRecentAddressableMessages(threadId: Long): Set<MessageRecord> {
+    return readableDatabase
+      .select(*MMS_PROJECTION)
+      .from(TABLE_NAME)
+      .where("$IS_ADDRESSABLE_CLAUSE AND $THREAD_ID = ?", threadId)
+      .orderBy("$DATE_RECEIVED DESC")
+      .limit(5)
+      .run()
+      .use {
+        MmsReader(it).toSet()
+      }
+  }
+
+  fun getAddressableMessagesBefore(threadId: Long, beforeTimestamp: Long): Set<MessageRecord> {
+    return readableDatabase
+      .select(*MMS_PROJECTION)
+      .from(TABLE_NAME)
+      .where("$IS_ADDRESSABLE_CLAUSE AND $THREAD_ID = ? AND $DATE_RECEIVED < ?", threadId, beforeTimestamp)
+      .orderBy("$DATE_RECEIVED DESC")
+      .limit(5)
+      .run()
+      .use {
+        MmsReader(it).toSet()
+      }
+  }
+
   protected enum class ReceiptType(val columnName: String, val groupStatus: Int) {
     READ(HAS_READ_RECEIPT, GroupReceiptTable.STATUS_READ),
     DELIVERY(HAS_DELIVERY_RECEIPT, GroupReceiptTable.STATUS_DELIVERED),

app/src/main/java/org/thoughtcrime/securesms/database/MessageTypes.java

@@ -90,6 +90,7 @@ public interface MessageTypes {
   long PUSH_MESSAGE_BIT   = 0x200000;
 
   // Group Message Information
+  long GROUP_MASK                  = 0xF0000;
   long GROUP_UPDATE_BIT            = 0x10000;
   // Note: Leave bit was previous QUIT bit for GV1, now also general member leave for GV2
   long GROUP_LEAVE_BIT             = 0x20000;

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

@@ -50,6 +50,7 @@ import org.thoughtcrime.securesms.database.model.serialize
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
 import org.thoughtcrime.securesms.groups.BadGroupIdException
 import org.thoughtcrime.securesms.groups.GroupId
+import org.thoughtcrime.securesms.jobs.MultiDeviceDeleteSendSyncJob
 import org.thoughtcrime.securesms.jobs.OptimizeMessageSearchIndexJob
 import org.thoughtcrime.securesms.keyvalue.SignalStore
 import org.thoughtcrime.securesms.mms.SlideDeck
@@ -61,6 +62,7 @@ import org.thoughtcrime.securesms.recipients.RecipientId
 import org.thoughtcrime.securesms.recipients.RecipientUtil
 import org.thoughtcrime.securesms.storage.StorageSyncHelper
 import org.thoughtcrime.securesms.util.ConversationUtil
+import org.thoughtcrime.securesms.util.FeatureFlags
 import org.thoughtcrime.securesms.util.JsonUtils
 import org.thoughtcrime.securesms.util.JsonUtils.SaneJSONObject
 import org.thoughtcrime.securesms.util.LRUCache
@@ -324,13 +326,23 @@ class ThreadTable(context: Context, databaseHelper: SignalDatabase) : DatabaseTa
       return
     }
 
+    val syncThreadTrimDeletes = SignalStore.settings().shouldSyncThreadTrimDeletes() && FeatureFlags.deleteSyncEnabled()
+    val threadTrimsToSync = mutableListOf<Pair<Long, Set<MessageRecord>>>()
+
     readableDatabase
       .select(ID)
       .from(TABLE_NAME)
       .run()
       .use { cursor ->
         while (cursor.moveToNext()) {
-          trimThreadInternal(cursor.requireLong(ID), length, trimBeforeDate)
+          trimThreadInternal(
+            threadId = cursor.requireLong(ID),
+            syncThreadTrimDeletes = syncThreadTrimDeletes,
+            length = length,
+            trimBeforeDate = trimBeforeDate
+          )?.also {
+            threadTrimsToSync += it
+          }
         }
       }
 
@@ -346,18 +358,29 @@ class ThreadTable(context: Context, databaseHelper: SignalDatabase) : DatabaseTa
       Log.i(TAG, "Trim all threads caused $deletes attachments to be deleted.")
     }
 
+    if (syncThreadTrimDeletes && threadTrimsToSync.isNotEmpty()) {
+      MultiDeviceDeleteSendSyncJob.enqueueThreadDeletes(threadTrimsToSync, isFullDelete = false)
+    }
+
     notifyAttachmentListeners()
     notifyStickerPackListeners()
     OptimizeMessageSearchIndexJob.enqueue()
   }
 
-  fun trimThread(threadId: Long, length: Int, trimBeforeDate: Long) {
+  fun trimThread(
+    threadId: Long,
+    syncThreadTrimDeletes: Boolean,
+    length: Int = NO_TRIM_MESSAGE_COUNT_SET,
+    trimBeforeDate: Long = NO_TRIM_BEFORE_DATE_SET,
+    inclusive: Boolean = false
+  ) {
     if (length == NO_TRIM_MESSAGE_COUNT_SET && trimBeforeDate == NO_TRIM_BEFORE_DATE_SET) {
       return
     }
 
+    var threadTrimToSync: Pair<Long, Set<MessageRecord>>? = null
     val deletes = writableDatabase.withinTransaction {
-      trimThreadInternal(threadId, length, trimBeforeDate)
+      threadTrimToSync = trimThreadInternal(threadId, syncThreadTrimDeletes, length, trimBeforeDate, inclusive)
       messages.deleteAbandonedMessages()
       attachments.trimAllAbandonedAttachments()
       groupReceipts.deleteAbandonedRows()
@@ -369,14 +392,24 @@ class ThreadTable(context: Context, databaseHelper: SignalDatabase) : DatabaseTa
       Log.i(TAG, "Trim thread $threadId caused $deletes attachments to be deleted.")
     }
 
+    if (syncThreadTrimDeletes && threadTrimToSync != null) {
+      MultiDeviceDeleteSendSyncJob.enqueueThreadDeletes(listOf(threadTrimToSync!!), isFullDelete = false)
+    }
+
     notifyAttachmentListeners()
     notifyStickerPackListeners()
     OptimizeMessageSearchIndexJob.enqueue()
   }
 
-  private fun trimThreadInternal(threadId: Long, length: Int, trimBeforeDate: Long) {
+  private fun trimThreadInternal(
+    threadId: Long,
+    syncThreadTrimDeletes: Boolean,
+    length: Int,
+    trimBeforeDate: Long,
+    inclusive: Boolean = false
+  ): Pair<Long, Set<MessageRecord>>? {
     if (length == NO_TRIM_MESSAGE_COUNT_SET && trimBeforeDate == NO_TRIM_BEFORE_DATE_SET) {
-      return
+      return null
     }
 
     val finalTrimBeforeDate = if (length != NO_TRIM_MESSAGE_COUNT_SET && length > 0) {
@@ -393,19 +426,29 @@ class ThreadTable(context: Context, databaseHelper: SignalDatabase) : DatabaseTa
     }
 
     if (finalTrimBeforeDate != NO_TRIM_BEFORE_DATE_SET) {
-      Log.i(TAG, "Trimming thread: $threadId before: $finalTrimBeforeDate")
+      Log.i(TAG, "Trimming thread: $threadId before: $finalTrimBeforeDate inclusive: $inclusive")
+
+      val addressableMessages: Set<MessageRecord> = if (syncThreadTrimDeletes) messages.getAddressableMessagesBefore(threadId, finalTrimBeforeDate) else emptySet()
+      val deletes = messages.deleteMessagesInThreadBeforeDate(threadId, finalTrimBeforeDate, inclusive)
 
-      val deletes = messages.deleteMessagesInThreadBeforeDate(threadId, finalTrimBeforeDate)
       if (deletes > 0) {
         Log.i(TAG, "Trimming deleted $deletes messages thread: $threadId")
         setLastScrolled(threadId, 0)
-        update(threadId, false)
+        val threadDeleted = update(threadId, false)
         notifyConversationListeners(threadId)
         SignalDatabase.calls.updateCallEventDeletionTimestamps()
+
+        return if (syncThreadTrimDeletes && (threadDeleted || addressableMessages.isNotEmpty())) {
+          threadId to addressableMessages
+        } else {
+          null
+        }
       } else {
         Log.i(TAG, "Trimming deleted no messages thread: $threadId")
       }
     }
+
+    return null
   }
 
   fun setAllThreadsRead(): List<MarkedMessageInfo> {
@@ -1068,10 +1111,30 @@ class ThreadTable(context: Context, databaseHelper: SignalDatabase) : DatabaseTa
       }
   }
 
-  fun deleteConversation(threadId: Long) {
+  fun deleteConversationIfContainsOnlyLocal(threadId: Long): Boolean {
+    return writableDatabase.withinTransaction {
+      val containsAddressable = messages.threadContainsAddressableMessages(threadId)
+      val isEmpty = messages.threadIsEmpty(threadId)
+
+      if (containsAddressable || isEmpty) {
+        false
+      } else {
+        deleteConversation(threadId, syncThreadDeletes = false)
+        true
+      }
+    }
+  }
+
+  @JvmOverloads
+  fun deleteConversation(threadId: Long, syncThreadDeletes: Boolean = true) {
     val recipientIdForThreadId = getRecipientIdForThreadId(threadId)
 
+    var addressableMessages: Set<MessageRecord> = emptySet()
     writableDatabase.withinTransaction { db ->
+      if (syncThreadDeletes && FeatureFlags.deleteSyncEnabled()) {
+        addressableMessages = messages.getMostRecentAddressableMessages(threadId)
+      }
+
       messages.deleteThread(threadId)
       drafts.clearDrafts(threadId)
       db.deactivateThread(threadId)
@@ -1080,6 +1143,10 @@ class ThreadTable(context: Context, databaseHelper: SignalDatabase) : DatabaseTa
       }
     }
 
+    if (syncThreadDeletes) {
+      MultiDeviceDeleteSendSyncJob.enqueueThreadDeletes(listOf(threadId to addressableMessages), isFullDelete = true)
+    }
+
     notifyConversationListListeners()
     notifyConversationListeners(threadId)
     ApplicationDependencies.getDatabaseObserver().notifyConversationDeleteListeners(threadId)
@@ -1089,12 +1156,20 @@ class ThreadTable(context: Context, databaseHelper: SignalDatabase) : DatabaseTa
   fun deleteConversations(selectedConversations: Set<Long>) {
     val recipientIds = getRecipientIdsForThreadIds(selectedConversations)
 
+    val addressableMessages = mutableListOf<Pair<Long, Set<MessageRecord>>>()
+
     val queries: List<SqlUtil.Query> = SqlUtil.buildCollectionQuery(ID, selectedConversations)
     writableDatabase.withinTransaction { db ->
       for (query in queries) {
         db.deactivateThread(query)
       }
 
+      if (FeatureFlags.deleteSyncEnabled()) {
+        for (threadId in selectedConversations) {
+          addressableMessages += threadId to messages.getMostRecentAddressableMessages(threadId)
+        }
+      }
+
       messages.deleteAbandonedMessages()
       attachments.trimAllAbandonedAttachments()
       groupReceipts.deleteAbandonedRows()
@@ -1108,6 +1183,8 @@ class ThreadTable(context: Context, databaseHelper: SignalDatabase) : DatabaseTa
       }
     }
 
+    MultiDeviceDeleteSendSyncJob.enqueueThreadDeletes(addressableMessages, isFullDelete = true)
+
     notifyConversationListListeners()
     notifyConversationListeners(selectedConversations)
     ApplicationDependencies.getDatabaseObserver().notifyConversationDeleteListeners(selectedConversations)

app/src/main/java/org/thoughtcrime/securesms/database/model/MessageRecord.java

@@ -72,6 +72,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
+import java.util.Objects;
 import java.util.Set;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -707,7 +708,7 @@ public abstract class MessageRecord extends DisplayRecord {
   }
 
   public int hashCode() {
-    return (int)getId();
+    return Objects.hash(id, isMms());
   }
 
   public int getSubscriptionId() {