Call expire after read DaR messages

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

@@ -12,8 +12,10 @@ import org.session.libsession.utilities.Document;
 import org.session.libsession.utilities.IdentityKeyMismatch;
 import org.session.libsession.utilities.IdentityKeyMismatchList;
 import org.session.libsignal.crypto.IdentityKey;
+import org.session.libsignal.protos.SignalServiceProtos;
 import org.session.libsignal.utilities.JsonUtil;
 import org.session.libsignal.utilities.Log;
+import org.thoughtcrime.securesms.conversation.disappearingmessages.ExpiryType;
 import org.thoughtcrime.securesms.database.helpers.SQLCipherOpenHelper;
 import org.thoughtcrime.securesms.database.model.MessageRecord;
 import org.thoughtcrime.securesms.util.SqlUtil;
@@ -273,6 +275,16 @@ public abstract class MessagingDatabase extends Database implements MmsSmsColumn
     public ExpirationInfo getExpirationInfo() {
       return expirationInfo;
     }
+
+    public ExpiryType guessExpiryType() {
+      long expireStarted = expirationInfo.expireStarted;
+      long expiresIn = expirationInfo.expiresIn;
+      long timestamp = syncMessageId.timetamp;
+
+      if (timestamp == expireStarted) return ExpiryType.AFTER_SEND;
+      if (expiresIn > 0) return ExpiryType.AFTER_READ;
+      return ExpiryType.NONE;
+    }
   }
 
   public static class InsertResult {

app/src/main/java/org/thoughtcrime/securesms/notifications/MarkReadReceiver.kt

@@ -6,8 +6,12 @@ import android.content.Context
 import android.content.Intent
 import android.os.AsyncTask
 import androidx.core.app.NotificationManagerCompat
-import com.annimon.stream.Collectors
+import kotlinx.coroutines.Dispatchers
-import com.annimon.stream.Stream
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.async
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
 import nl.komponents.kovenant.task
 import org.session.libsession.messaging.MessagingModuleConfiguration.Companion.shared
 import org.session.libsession.messaging.messages.control.ReadReceipt
@@ -20,9 +24,9 @@ import org.session.libsession.utilities.associateByNotNull
 import org.session.libsession.utilities.recipients.Recipient
 import org.session.libsignal.utilities.Log
 import org.thoughtcrime.securesms.ApplicationContext
+import org.thoughtcrime.securesms.conversation.disappearingmessages.ExpiryType
 import org.thoughtcrime.securesms.database.MessagingDatabase.ExpirationInfo
 import org.thoughtcrime.securesms.database.MessagingDatabase.MarkedMessageInfo
-import org.thoughtcrime.securesms.database.MessagingDatabase.SyncMessageId
 import org.thoughtcrime.securesms.dependencies.DatabaseComponent
 import org.thoughtcrime.securesms.util.SessionMetaProtocol.shouldSendReadReceipt
 
@@ -49,48 +53,85 @@ class MarkReadReceiver : BroadcastReceiver() {
     }
 
     companion object {
-        private val TAG = MarkReadReceiver::class.java.getSimpleName()
+        private val TAG = MarkReadReceiver::class.java.simpleName
         const val CLEAR_ACTION = "network.loki.securesms.notifications.CLEAR"
         const val THREAD_IDS_EXTRA = "thread_ids"
         const val NOTIFICATION_ID_EXTRA = "notification_id"
 
         @JvmStatic
-        fun process(context: Context, markedReadMessages: List<MarkedMessageInfo>) {
+        fun process(
+            context: Context,
+            markedReadMessages: List<MarkedMessageInfo>
+        ) {
             if (markedReadMessages.isEmpty()) return
 
-            val loki = DatabaseComponent.get(context).lokiMessageDatabase()
+            sendReadReceipts(context, markedReadMessages)
-
-            task {
-                val hashToInfo = markedReadMessages.associateByNotNull {
-                    it.expirationInfo.run { loki.getMessageServerHash(id, isMms) }
-                }
-                if (hashToInfo.isEmpty()) return@task
-
-                @Suppress("UNCHECKED_CAST")
-                val expiries = SnodeAPI.getExpiries(hashToInfo.keys.toList(), TextSecurePreferences.getLocalNumber(context)!!)
-                    .get()["expiries"] as Map<String, Long>
-
-                hashToInfo.forEach { (hash, info) -> expiries[hash]?.let { scheduleDeletion(context, info.expirationInfo, it - info.expirationInfo.expireStarted) } }
-            } fail {
-                Log.e(TAG, "process() disappear after read failed", it)
-            }
 
             markedReadMessages.forEach { scheduleDeletion(context, it.expirationInfo) }
 
-            if (!isReadReceiptsEnabled(context)) return
+            getHashToMessage(context, markedReadMessages)?.let {
+                fetchUpdatedExpiriesAndScheduleDeletion(context, it)
+                shortenExpiryOfDisappearingAfterRead(context, it)
+            }
+        }
 
-            markedReadMessages.map { it.syncMessageId }
+        private fun getHashToMessage(
-                .filter { shouldSendReadReceipt(Recipient.from(context, it.address, false)) }
+            context: Context,
-                .groupBy { it.address }
+            markedReadMessages: List<MarkedMessageInfo>
-                .forEach { (address, messages) ->
+        ): Map<String, MarkedMessageInfo>? {
-                    messages.map { it.timetamp }
+            val loki = DatabaseComponent.get(context).lokiMessageDatabase()
-                        .let(::ReadReceipt)
+
-                        .apply { sentTimestamp = nowWithOffset }
+            return markedReadMessages
-                        .let { send(it, address) }
+                .associateByNotNull { it.expirationInfo.run { loki.getMessageServerHash(id, isMms) } }
+                .takeIf { it.isNotEmpty() }
+        }
+
+        private fun shortenExpiryOfDisappearingAfterRead(
+            context: Context,
+            hashToMessage: Map<String, MarkedMessageInfo>
+        ) {
+            hashToMessage.filterValues { it.guessExpiryType() == ExpiryType.AFTER_READ }
+                .entries
+                .groupBy(
+                    keySelector =  { it.value.expirationInfo.expiresIn },
+                    valueTransform = { it.key }
+                ).forEach { (expiresIn, hashes) ->
+                    SnodeAPI.alterTtl(
+                        messageHashes = hashes,
+                        newExpiry = nowWithOffset + expiresIn,
+                        publicKey = TextSecurePreferences.getLocalNumber(context)!!,
+                        shorten = true
+                    )
                 }
         }
 
-        fun scheduleDeletion(
+        private fun sendReadReceipts(
+            context: Context,
+            markedReadMessages: List<MarkedMessageInfo>
+        ) {
+            if (isReadReceiptsEnabled(context)) {
+                markedReadMessages.map { it.syncMessageId }
+                    .filter { shouldSendReadReceipt(Recipient.from(context, it.address, false)) }
+                    .groupBy { it.address }
+                    .forEach { (address, messages) ->
+                        messages.map { it.timetamp }
+                            .let(::ReadReceipt)
+                            .apply { sentTimestamp = nowWithOffset }
+                            .let { send(it, address) }
+                    }
+            }
+        }
+
+        private fun fetchUpdatedExpiriesAndScheduleDeletion(
+            context: Context,
+            hashToMessage: Map<String, MarkedMessageInfo>
+        ) {
+            @Suppress("UNCHECKED_CAST")
+            val expiries = SnodeAPI.getExpiries(hashToMessage.keys.toList(), TextSecurePreferences.getLocalNumber(context)!!).get()["expiries"] as Map<String, Long>
+            hashToMessage.forEach { (hash, info) -> expiries[hash]?.let { scheduleDeletion(context, info.expirationInfo, it - info.expirationInfo.expireStarted) } }
+        }
+
+        private fun scheduleDeletion(
             context: Context?,
             expirationInfo: ExpirationInfo,
             expiresIn: Long = expirationInfo.expiresIn
@@ -98,13 +139,10 @@ class MarkReadReceiver : BroadcastReceiver() {
             android.util.Log.d(TAG, "scheduleDeletion() called with: expirationInfo = $expirationInfo, expiresIn = $expiresIn")
 
             if (expiresIn > 0 && expirationInfo.expireStarted <= 0) {
-                val expirationManager =
+                if (expirationInfo.isMms) DatabaseComponent.get(context!!).mmsDatabase().markExpireStarted(expirationInfo.id)
-                    ApplicationContext.getInstance(context).expiringMessageManager
+                else DatabaseComponent.get(context!!).smsDatabase().markExpireStarted(expirationInfo.id)
-                if (expirationInfo.isMms) DatabaseComponent.get(context!!).mmsDatabase()
+
-                    .markExpireStarted(expirationInfo.id) else DatabaseComponent.get(
+                ApplicationContext.getInstance(context).expiringMessageManager.scheduleDeletion(
-                    context!!
-                ).smsDatabase().markExpireStarted(expirationInfo.id)
-                expirationManager.scheduleDeletion(
                     expirationInfo.id,
                     expirationInfo.isMms,
                     expiresIn