Fix out of order websocket events for posts (#7290)

app/actions/websocket/posts.ts

@@ -15,6 +15,7 @@ import {getChannelById, getMyChannel} from '@queries/servers/channel';
 import {getPostById} from '@queries/servers/post';
 import {getCurrentChannelId, getCurrentTeamId, getCurrentUserId} from '@queries/servers/system';
 import {getIsCRTEnabled} from '@queries/servers/thread';
+import EphemeralStore from '@store/ephemeral_store';
 import NavigationStore from '@store/navigation_store';
 import {isTablet} from '@utils/helpers';
 import {isFromWebhook, isSystemMessage, shouldIgnorePost} from '@utils/post';
@@ -162,6 +163,20 @@ export async function handleNewPostEvent(serverUrl: string, msg: WebSocketMessag
         actionType = ActionType.POSTS.RECEIVED_IN_THREAD;
     }
 
+    const outOfOrderWebsocketEvent = EphemeralStore.getLastPostWebsocketEvent(serverUrl, post.id);
+    if (outOfOrderWebsocketEvent?.deleted) {
+        for (const model of models) {
+            if (model._preparedState === 'update') {
+                model.cancelPrepareUpdate();
+            }
+        }
+        return;
+    }
+
+    if (outOfOrderWebsocketEvent?.post) {
+        post = outOfOrderWebsocketEvent.post;
+    }
+
     const postModels = await operator.handlePosts({
         actionType,
         order: [post.id],
@@ -191,7 +206,12 @@ export async function handlePostEdited(serverUrl: string, msg: WebSocketMessage)
     const {database} = operator;
 
     const oldPost = await getPostById(database, post.id);
-    if (oldPost && oldPost.isPinned !== post.is_pinned) {
+    if (!oldPost) {
+        EphemeralStore.addEditingPost(serverUrl, post);
+        return;
+    }
+
+    if (oldPost.isPinned !== post.is_pinned) {
         fetchChannelStats(serverUrl, post.channel_id);
     }
 
@@ -228,6 +248,12 @@ export async function handlePostDeleted(serverUrl: string, msg: WebSocketMessage
 
         const post: Post = JSON.parse(msg.data.post);
 
+        const oldPost = await getPostById(database, post.id);
+        if (!oldPost) {
+            EphemeralStore.addRemovingPost(serverUrl, post.id);
+            return;
+        }
+
         const models: Model[] = [];
 
         const {model: deleteModel} = await markPostAsDeleted(serverUrl, post, true);

app/store/ephemeral_store.ts

@@ -3,6 +3,10 @@
 
 import {BehaviorSubject} from 'rxjs';
 
+import {toMilliseconds} from '@utils/datetime';
+
+const TIME_TO_CLEAR_WEBSOCKET_ACTIONS = toMilliseconds({seconds: 30});
+
 class EphemeralStore {
     theme: Theme | undefined;
     creatingChannel = false;
@@ -11,6 +15,9 @@ class EphemeralStore {
     private pushProxyVerification: {[serverUrl: string]: string | undefined} = {};
     private canJoinOtherTeams: {[serverUrl: string]: BehaviorSubject<boolean>} = {};
 
+    private websocketEditingPost: {[serverUrl: string]: {[id: string]: {post: Post; timeout: NodeJS.Timeout} | undefined} | undefined} = {};
+    private websocketRemovingPost: {[serverUrl: string]: Set<string> | undefined} = {};
+
     // As of today, the server sends a duplicated event to add the user to the team.
     // If we do not handle this, this ends up showing some errors in the database, apart
     // of the extra computation time. We use this to track the events that are being handled
@@ -25,6 +32,66 @@ class EphemeralStore {
     private notificationTapped = false;
     private enablingCRT = false;
 
+    // Ephemeral control for out of order websocket events
+    addEditingPost = (serverUrl: string, post: Post) => {
+        if (this.websocketRemovingPost[serverUrl]?.has(post.id)) {
+            return;
+        }
+
+        const lastEdit = this.websocketEditingPost[serverUrl]?.[post.id];
+        if (lastEdit && post.edit_at < lastEdit.post.update_at) {
+            return;
+        }
+
+        if (!this.websocketEditingPost[serverUrl]) {
+            this.websocketEditingPost[serverUrl] = {};
+        }
+        const serverEditing = this.websocketEditingPost[serverUrl]!;
+
+        if (lastEdit?.timeout) {
+            clearTimeout(lastEdit.timeout);
+        }
+
+        const timeout = setTimeout(() => {
+            delete serverEditing[post.id];
+        }, TIME_TO_CLEAR_WEBSOCKET_ACTIONS);
+
+        serverEditing[post.id] = {post, timeout};
+    };
+
+    addRemovingPost = (serverUrl: string, postId: string) => {
+        if (this.websocketRemovingPost[serverUrl]?.has(postId)) {
+            return;
+        }
+
+        if (this.websocketEditingPost[serverUrl]?.[postId]) {
+            clearTimeout(this.websocketEditingPost[serverUrl]![postId]!.timeout);
+            delete this.websocketEditingPost[serverUrl]![postId];
+        }
+
+        if (!this.websocketRemovingPost[serverUrl]) {
+            this.websocketRemovingPost[serverUrl] = new Set();
+        }
+
+        setTimeout(() => {
+            this.websocketRemovingPost[serverUrl]?.delete(postId);
+        }, TIME_TO_CLEAR_WEBSOCKET_ACTIONS);
+
+        this.websocketRemovingPost[serverUrl]?.add(postId);
+    };
+
+    getLastPostWebsocketEvent = (serverUrl: string, postId: string) => {
+        if (this.websocketRemovingPost[serverUrl]?.has(postId)) {
+            return {deleted: true, post: undefined};
+        }
+
+        if (this.websocketEditingPost[serverUrl]?.[postId]) {
+            return {deleted: false, post: this.websocketEditingPost[serverUrl]![postId]!.post};
+        }
+
+        return undefined;
+    };
+
     // Ephemeral control when (un)archiving a channel locally
     addArchivingChannel = (channelId: string) => {
         this.archivingChannels.add(channelId);

patches/@nozbe+watermelondb+0.25.5.patch

@@ -38,7 +38,7 @@ index 96114ec..ecfe3c1 100644
  
    prepareDestroyPermanently(): this
 diff --git a/node_modules/@nozbe/watermelondb/Model/index.js b/node_modules/@nozbe/watermelondb/Model/index.js
-index b0e3a83..d7ead09 100644
+index b0e3a83..d5e730c 100644
 --- a/node_modules/@nozbe/watermelondb/Model/index.js
 +++ b/node_modules/@nozbe/watermelondb/Model/index.js
 @@ -81,7 +81,17 @@ var Model = /*#__PURE__*/function () {
@@ -68,13 +68,15 @@ index b0e3a83..d7ead09 100644
      (0, _ensureSync.default)(recordUpdater(this));
      this._isEditing = false;
      this._preparedState = 'update'; // TODO: `process.nextTick` doesn't work on React Native
-@@ -107,6 +116,19 @@ var Model = /*#__PURE__*/function () {
+@@ -107,6 +116,21 @@ var Model = /*#__PURE__*/function () {
      return this;
    };
  
 +  _proto.cancelPrepareUpdate = function cancelPrepareUpdate() {
++    var _this = this;
++
 +    if ('test' !== process.env.NODE_ENV && 'undefined' !== typeof process && process) {
-+      (0, _invariant.default)('update' !== _this._preparedState, "Cannot cancel an update on a model that has not been prepared in table " + this.table);
++      (0, _invariant.default)('update' === _this._preparedState, "Cannot cancel an update on a model that has not been prepared in table " + _this.table);
 +    }
 +    this.__changes = null;
 +    this._preparedState = null;
@@ -88,7 +90,7 @@ index b0e3a83..d7ead09 100644
    _proto.prepareMarkAsDeleted = function prepareMarkAsDeleted() {
      (0, _invariant.default)(!this._preparedState, "Cannot mark a record with pending changes as deleted");
  
-@@ -118,7 +140,10 @@ var Model = /*#__PURE__*/function () {
+@@ -118,7 +142,10 @@ var Model = /*#__PURE__*/function () {
    };
  
    _proto.prepareDestroyPermanently = function prepareDestroyPermanently() {