Fix case of lost messages: Do not delete events from the last forward chunk

changelog.d/5878.bugfix

@@ -0,0 +1 @@
+Fix a case where paginating forwards would lose messages

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/database/helper/ChunkEntityHelper.kt

@@ -39,6 +39,21 @@ import org.matrix.android.sdk.internal.extensions.assertIsManaged
 import org.matrix.android.sdk.internal.session.room.timeline.PaginationDirection
 import timber.log.Timber
 
+internal fun ChunkEntity.moveEventsFrom(chunkToMerge: ChunkEntity, direction: PaginationDirection) {
+    assertIsManaged()
+    val localRealm = this.realm
+    val eventsToMerge = if (direction == PaginationDirection.FORWARDS) {
+        chunkToMerge.timelineEvents.sort(TimelineEventEntityFields.DISPLAY_INDEX, Sort.ASCENDING)
+    } else {
+        chunkToMerge.timelineEvents.sort(TimelineEventEntityFields.DISPLAY_INDEX, Sort.DESCENDING)
+    }
+    eventsToMerge.forEach {
+        if (addTimelineEventFromMove(localRealm, it, direction)) {
+            chunkToMerge.timelineEvents.remove(it)
+        }
+    }
+}
+
 internal fun ChunkEntity.merge(roomId: String, chunkToMerge: ChunkEntity, direction: PaginationDirection) {
     assertIsManaged()
     val localRealm = this.realm
@@ -165,6 +180,17 @@ private fun ChunkEntity.addTimelineEventFromMerge(realm: Realm, timelineEventEnt
     timelineEvents.add(copied)
 }
 
+private fun ChunkEntity.addTimelineEventFromMove(realm: Realm, event: TimelineEventEntity, direction: PaginationDirection): Boolean {
+    val eventId = event.eventId
+    if (timelineEvents.find(eventId) != null) {
+        return false
+    }
+    event.displayIndex = nextDisplayIndex(direction)
+    handleThreadSummary(realm, eventId, event)
+    timelineEvents.add(event)
+    return true
+}
+
 /**
  * Upon copy of the timeline events we should update the latestMessage TimelineEventEntity with the new one.
  */

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/timeline/TokenChunkEventPersistor.kt

@@ -180,6 +180,8 @@ internal class TokenChunkEventPersistor @Inject constructor(
             }
         }
         val optimizedThreadSummaryMap = hashMapOf<String, EventEntity>()
+        var hasNewEvents = false
+        var existingChunkToLink: ChunkEntity? = null
         run processTimelineEvents@{
             eventList.forEach { event ->
                 if (event.eventId == null || event.senderId == null) {
@@ -194,18 +196,36 @@ internal class TokenChunkEventPersistor @Inject constructor(
                 // If it exists, we want to stop here, just link the prevChunk
                 val existingChunk = existingTimelineEvent?.chunk?.firstOrNull()
                 if (existingChunk != null) {
+                    // If we haven't found a single new event yet, we don't want to link in the pagination direction, as that might cause a
+                    // timeline loop if the other chunk is in the other direction.
+                    if (!hasNewEvents) {
+                        Timber.i("Skip adding event $eventId, already exists")
+                        // Only skip this event, but still process other events.
+                        // Remember this chunk, since in case we don't find any new events, we still want to link this in pagination direction
+                        // e.g. in order to link a chunk to the /sync chunk
+                        if (existingChunkToLink == null) {
+                            existingChunkToLink = existingChunk
+                        }
+                        return@forEach
+                    }
                     when (direction) {
                         PaginationDirection.BACKWARDS -> {
                             if (currentChunk.nextChunk == existingChunk) {
-                                Timber.w("Avoid double link, shouldn't happen in an ideal world")
+                                Timber.i("Avoid double link")
+                                // Do not stop processing here: even though this event already exists in an already linked chunk,
+                                // we still may have new events to add
+                                return@forEach
                             } else {
                                 currentChunk.prevChunk = existingChunk
                                 existingChunk.nextChunk = currentChunk
                             }
                         }
                         PaginationDirection.FORWARDS  -> {
                             if (currentChunk.prevChunk == existingChunk) {
-                                Timber.w("Avoid double link, shouldn't happen in an ideal world")
+                                Timber.i("Avoid double link")
+                                // Do not stop processing here: even though this event already exists in an already linked chunk,
+                                // we still may have new events to add
+                                return@forEach
                             } else {
                                 currentChunk.nextChunk = existingChunk
                                 existingChunk.prevChunk = currentChunk
@@ -215,6 +235,10 @@ internal class TokenChunkEventPersistor @Inject constructor(
                     // Stop processing here
                     return@processTimelineEvents
                 }
+
+                // existingChunk == null => this is a new event we haven't seen before
+                hasNewEvents = true
+
                 val ageLocalTs = event.unsignedData?.age?.let { now - it }
                 var eventEntity = event.toEntity(roomId, SendState.SYNCED, ageLocalTs).copyToRealmOrIgnore(realm, EventInsertType.PAGINATION)
                 if (event.type == EventType.STATE_ROOM_MEMBER && event.stateKey != null) {
@@ -243,6 +267,27 @@ internal class TokenChunkEventPersistor @Inject constructor(
                 }
             }
         }
+        val existingChunk = existingChunkToLink
+        if (!hasNewEvents && existingChunk != null) {
+            when (direction) {
+                PaginationDirection.BACKWARDS -> {
+                    if (currentChunk.nextChunk == existingChunk) {
+                        Timber.w("Avoid double link, shouldn't happen in an ideal world")
+                    } else {
+                        currentChunk.prevChunk = existingChunk
+                        existingChunk.nextChunk = currentChunk
+                    }
+                }
+                PaginationDirection.FORWARDS  -> {
+                    if (currentChunk.prevChunk == existingChunk) {
+                        Timber.w("Avoid double link, shouldn't happen in an ideal world")
+                    } else {
+                        currentChunk.nextChunk = existingChunk
+                        existingChunk.prevChunk = currentChunk
+                    }
+                }
+            }
+        }
         if (currentChunk.isValid) {
             RoomEntity.where(realm, roomId).findFirst()?.addIfNecessary(currentChunk)
         }

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/sync/handler/room/RoomSyncHandler.kt

@@ -43,6 +43,7 @@ import org.matrix.android.sdk.internal.crypto.DefaultCryptoService
 import org.matrix.android.sdk.internal.database.helper.addIfNecessary
 import org.matrix.android.sdk.internal.database.helper.addTimelineEvent
 import org.matrix.android.sdk.internal.database.helper.createOrUpdate
+import org.matrix.android.sdk.internal.database.helper.moveEventsFrom
 import org.matrix.android.sdk.internal.database.helper.updateThreadSummaryIfNeeded
 import org.matrix.android.sdk.internal.database.mapper.asDomain
 import org.matrix.android.sdk.internal.database.mapper.toEntity
@@ -368,6 +369,15 @@ internal class RoomSyncHandler @Inject constructor(
                                      aggregator: SyncResponsePostTreatmentAggregator): ChunkEntity {
         val lastChunk = ChunkEntity.findLastForwardChunkOfRoom(realm, roomEntity.roomId)
         if (isLimited && lastChunk != null) {
+            Timber.i("Deleting last forward chunk (${lastChunk.identifier()})")
+            // Add events that oldPrev may have dropped since they were already in lastChunk
+            val oldPrev = lastChunk.prevChunk
+            if (oldPrev != null && oldPrev.nextToken != lastChunk.prevToken) {
+                // If the tokens mismatch, this means we have chained them due to duplicated events.
+                // In this case, we need to make sure to re-add possibly dropped events (which would have
+                // been duplicates otherwise)
+                oldPrev.moveEventsFrom(lastChunk, PaginationDirection.FORWARDS)
+            }
             lastChunk.deleteOnCascade(deleteStateEvents = false, canDeleteRoot = true)
         }
         val chunkEntity = if (!isLimited && lastChunk != null) {