Sync disappear after read with other devices

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

@@ -816,13 +816,7 @@ public class ThreadDatabase extends Database {
     MmsSmsDatabase mmsSmsDatabase = DatabaseComponent.get(context).mmsSmsDatabase();
     if (mmsSmsDatabase.getConversationCount(threadId) <= 0 && !force) return false;
     List<MarkedMessageInfo> messages = setRead(threadId, lastSeenTime);
-    if (isGroupRecipient) {
+    MarkReadReceiver.process(context, messages);
-      for (MarkedMessageInfo message: messages) {
-        MarkReadReceiver.scheduleDeletion(context, message.getExpirationInfo());
-      }
-    } else {
-      MarkReadReceiver.process(context, messages);
-    }
     ApplicationContext.getInstance(context).messageNotifier.updateNotification(context, threadId);
     return setLastSeen(threadId, lastSeenTime);
   }

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

@@ -1,100 +0,0 @@
-package org.thoughtcrime.securesms.notifications;
-
-import android.annotation.SuppressLint;
-import android.content.BroadcastReceiver;
-import android.content.Context;
-import android.content.Intent;
-import android.os.AsyncTask;
-
-import androidx.annotation.NonNull;
-import androidx.core.app.NotificationManagerCompat;
-
-import com.annimon.stream.Collectors;
-import com.annimon.stream.Stream;
-
-import org.session.libsession.database.StorageProtocol;
-import org.session.libsession.messaging.MessagingModuleConfiguration;
-import org.session.libsession.messaging.messages.control.ReadReceipt;
-import org.session.libsession.messaging.sending_receiving.MessageSender;
-import org.session.libsession.snode.SnodeAPI;
-import org.session.libsession.utilities.Address;
-import org.session.libsession.utilities.TextSecurePreferences;
-import org.session.libsession.utilities.recipients.Recipient;
-import org.session.libsignal.utilities.Log;
-import org.thoughtcrime.securesms.ApplicationContext;
-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.service.ExpiringMessageManager;
-import org.thoughtcrime.securesms.util.SessionMetaProtocol;
-
-import java.util.List;
-import java.util.Map;
-
-public class MarkReadReceiver extends BroadcastReceiver {
-
-  private static final String TAG                   = MarkReadReceiver.class.getSimpleName();
-  public static final  String CLEAR_ACTION          = "network.loki.securesms.notifications.CLEAR";
-  public static final  String THREAD_IDS_EXTRA      = "thread_ids";
-  public static final  String NOTIFICATION_ID_EXTRA = "notification_id";
-
-  @SuppressLint("StaticFieldLeak")
-  @Override
-  public void onReceive(final Context context, Intent intent) {
-    if (!CLEAR_ACTION.equals(intent.getAction()))
-      return;
-
-    final long[] threadIds = intent.getLongArrayExtra(THREAD_IDS_EXTRA);
-
-    if (threadIds != null) {
-      NotificationManagerCompat.from(context).cancel(intent.getIntExtra(NOTIFICATION_ID_EXTRA, -1));
-
-      new AsyncTask<Void, Void, Void>() {
-        @Override
-        protected Void doInBackground(Void... params) {
-          long currentTime = SnodeAPI.getNowWithOffset();
-          for (long threadId : threadIds) {
-            Log.i(TAG, "Marking as read: " + threadId);
-            StorageProtocol storage = MessagingModuleConfiguration.getShared().getStorage();
-            storage.markConversationAsRead(threadId,currentTime, true);
-          }
-          return null;
-        }
-      }.executeOnExecutor(AsyncTask.THREAD_POOL_EXECUTOR);
-    }
-  }
-
-  public static void process(@NonNull Context context, @NonNull List<MarkedMessageInfo> markedReadMessages) {
-    if (markedReadMessages.isEmpty()) return;
-
-    for (MarkedMessageInfo messageInfo : markedReadMessages) {
-      scheduleDeletion(context, messageInfo.getExpirationInfo());
-    }
-
-    if (!TextSecurePreferences.isReadReceiptsEnabled(context)) return;
-
-    Map<Address, List<SyncMessageId>> addressMap = Stream.of(markedReadMessages)
-                                                         .map(MarkedMessageInfo::getSyncMessageId)
-                                                         .collect(Collectors.groupingBy(SyncMessageId::getAddress));
-
-    for (Address address : addressMap.keySet()) {
-      List<Long> timestamps = Stream.of(addressMap.get(address)).map(SyncMessageId::getTimetamp).toList();
-      if (!SessionMetaProtocol.shouldSendReadReceipt(Recipient.from(context, address, false))) { continue; }
-      ReadReceipt readReceipt = new ReadReceipt(timestamps);
-      readReceipt.setSentTimestamp(SnodeAPI.getNowWithOffset());
-      MessageSender.send(readReceipt, address);
-    }
-  }
-
-  public static void scheduleDeletion(Context context, ExpirationInfo expirationInfo) {
-    if (expirationInfo.getExpiresIn() > 0 && expirationInfo.getExpireStarted() <= 0) {
-      ExpiringMessageManager expirationManager = ApplicationContext.getInstance(context).getExpiringMessageManager();
-
-      if (expirationInfo.isMms()) DatabaseComponent.get(context).mmsDatabase().markExpireStarted(expirationInfo.getId());
-      else                        DatabaseComponent.get(context).smsDatabase().markExpireStarted(expirationInfo.getId());
-
-      expirationManager.scheduleDeletion(expirationInfo.getId(), expirationInfo.isMms(), expirationInfo.getExpiresIn());
-    }
-  }
-}

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

@@ -0,0 +1,120 @@
+package org.thoughtcrime.securesms.notifications
+
+import android.annotation.SuppressLint
+import android.content.BroadcastReceiver
+import android.content.Context
+import android.content.Intent
+import android.os.AsyncTask
+import androidx.core.app.NotificationManagerCompat
+import com.annimon.stream.Collectors
+import com.annimon.stream.Stream
+import nl.komponents.kovenant.task
+import org.session.libsession.messaging.MessagingModuleConfiguration.Companion.shared
+import org.session.libsession.messaging.messages.control.ReadReceipt
+import org.session.libsession.messaging.sending_receiving.MessageSender.send
+import org.session.libsession.snode.SnodeAPI
+import org.session.libsession.snode.SnodeAPI.nowWithOffset
+import org.session.libsession.utilities.TextSecurePreferences
+import org.session.libsession.utilities.TextSecurePreferences.Companion.isReadReceiptsEnabled
+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.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
+
+class MarkReadReceiver : BroadcastReceiver() {
+    @SuppressLint("StaticFieldLeak")
+    override fun onReceive(context: Context, intent: Intent) {
+        if (CLEAR_ACTION != intent.action) return
+        val threadIds = intent.getLongArrayExtra(THREAD_IDS_EXTRA)
+        if (threadIds != null) {
+            NotificationManagerCompat.from(context)
+                .cancel(intent.getIntExtra(NOTIFICATION_ID_EXTRA, -1))
+            object : AsyncTask<Void?, Void?, Void?>() {
+                override fun doInBackground(vararg params: Void?): Void? {
+                    val currentTime = nowWithOffset
+                    for (threadId in threadIds) {
+                        Log.i(TAG, "Marking as read: $threadId")
+                        val storage = shared.storage
+                        storage.markConversationAsRead(threadId, currentTime, true)
+                    }
+                    return null
+                }
+            }.executeOnExecutor(AsyncTask.THREAD_POOL_EXECUTOR)
+        }
+    }
+
+    companion object {
+        private val TAG = MarkReadReceiver::class.java.getSimpleName()
+        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>) {
+
+            val loki = DatabaseComponent.get(context).lokiMessageDatabase()
+
+            task {
+                val hashToInfo = markedReadMessages.associateByNotNull { loki.getMessageServerHash(it.expirationInfo.id) }
+
+                if (hashToInfo.isEmpty()) return@task
+
+                @Suppress("UNCHECKED_CAST")
+                val hashToExpiry = SnodeAPI.getExpiries(hashToInfo.keys.toList(), TextSecurePreferences.getLocalNumber(context)!!)
+                    .get()["expiries"] as Map<String, Long>
+
+                hashToInfo.forEach { (hash, info) -> hashToExpiry[hash]?.let { scheduleDeletion(context, info.expirationInfo, it - info.expirationInfo.expireStarted) } }
+            } fail {
+                Log.e(TAG, "process() disappear after read failed", it)
+            }
+
+            if (markedReadMessages.isEmpty()) return
+            for (messageInfo in markedReadMessages) {
+                scheduleDeletion(context, messageInfo.expirationInfo)
+            }
+            if (!isReadReceiptsEnabled(context)) return
+
+            val addressMap = Stream.of(markedReadMessages)
+                .map { it.syncMessageId }
+                .collect(Collectors.groupingBy { it.address } )
+
+            for (address in addressMap.keys) {
+                val timestamps = addressMap[address]!!.map { obj: SyncMessageId -> obj.timetamp }
+                if (!shouldSendReadReceipt(Recipient.from(context, address, false))) {
+                    continue
+                }
+
+                ReadReceipt(timestamps)
+                    .apply { sentTimestamp = nowWithOffset }
+                    .let { send(it, address) }
+            }
+        }
+
+        fun scheduleDeletion(
+            context: Context?,
+            expirationInfo: ExpirationInfo,
+            expiresIn: Long = expirationInfo.expiresIn
+        ) {
+            android.util.Log.d(TAG, "scheduleDeletion() called with: expirationInfo = $expirationInfo, expiresIn = $expiresIn")
+
+            if (expiresIn > 0 && expirationInfo.expireStarted <= 0) {
+                val expirationManager =
+                    ApplicationContext.getInstance(context).expiringMessageManager
+                if (expirationInfo.isMms) DatabaseComponent.get(context!!).mmsDatabase()
+                    .markExpireStarted(expirationInfo.id) else DatabaseComponent.get(
+                    context!!
+                ).smsDatabase().markExpireStarted(expirationInfo.id)
+                expirationManager.scheduleDeletion(
+                    expirationInfo.id,
+                    expirationInfo.isMms,
+                    expiresIn
+                )
+            }
+        }
+    }
+}

libsession/src/main/java/org/session/libsession/messaging/messages/control/ReadReceipt.kt

@@ -25,7 +25,7 @@ class ReadReceipt() : ControlMessage() {
         }
     }
 
-    internal constructor(timestamps: List<Long>?) : this() {
+    constructor(timestamps: List<Long>?) : this() {
         this.timestamps = timestamps
     }
 

libsession/src/main/java/org/session/libsession/snode/SnodeAPI.kt

@@ -533,14 +533,10 @@ object SnodeAPI {
 
     fun getExpiries(messageHashes: List<String>, publicKey: String) : RawResponsePromise {
         val userEd25519KeyPair = MessagingModuleConfiguration.shared.getUserED25519KeyPair() ?: return Promise.ofFail(NullPointerException("No user key pair"))
+        val hashes = messageHashes.takeIf { it.size != 1 } ?: (messageHashes + "") // TODO remove this when bug is fixed on nodes.
         return retryIfNeeded(maxRetryCount) {
             val timestamp = System.currentTimeMillis() + clockOffset
-            val params = mutableMapOf(
+            val signData = "${Snode.Method.GetExpiries.rawValue}$timestamp${hashes.joinToString(separator = "")}".toByteArray()
-                "pubkey" to publicKey,
-                "messages" to messageHashes,
-                "timestamp" to timestamp
-            )
-            val signData = "${Snode.Method.GetExpiries.rawValue}$timestamp${messageHashes.joinToString(separator = "")}".toByteArray()
 
             val ed25519PublicKey = userEd25519KeyPair.publicKey.asHexString
             val signature = ByteArray(Sign.BYTES)
@@ -555,9 +551,14 @@ object SnodeAPI {
                 Log.e("Loki", "Signing data failed with user secret key", e)
                 return@retryIfNeeded Promise.ofFail(e)
             }
-            params["pubkey_ed25519"] = ed25519PublicKey
+            val params = mapOf(
-            params["signature"] = Base64.encodeBytes(signature)
+                "pubkey" to publicKey,
-            getSingleTargetSnode(publicKey).bind { snode ->
+                "messages" to hashes,
+                "timestamp" to timestamp,
+                "pubkey_ed25519" to ed25519PublicKey,
+                "signature" to Base64.encodeBytes(signature)
+            )
+            getSingleTargetSnode(publicKey) bind { snode ->
                 invoke(Snode.Method.GetExpiries, snode, params, publicKey)
             }
         }

libsession/src/main/java/org/session/libsession/utilities/Util.kt

@@ -365,4 +365,19 @@ object Util {
         val digitGroups = (Math.log10(sizeBytes.toDouble()) / Math.log10(1024.0)).toInt()
         return DecimalFormat("#,##0.#").format(sizeBytes / Math.pow(1024.0, digitGroups.toDouble())) + " " + units[digitGroups]
     }
-}
+}
+
+fun <T, K: Any> Iterable<T>.associateByNotNull(
+    keySelector: (T) -> K?
+) = associateByNotNull(keySelector) { it }
+
+fun <T, K: Any, V: Any> Iterable<T>.associateByNotNull(
+    keySelector: (T) -> K?,
+    valueTransform: (T) -> V?,
+): Map<K, V> = buildMap {
+    for (item in this@associateByNotNull) {
+        val key = keySelector(item) ?: continue
+        val value = valueTransform(item) ?: continue
+        this[key] = value
+    }
+}