diff --git a/app/actions/websocket/posts.ts b/app/actions/websocket/posts.ts index 710f190fc..b5ae15186 100644 --- a/app/actions/websocket/posts.ts +++ b/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); diff --git a/app/store/ephemeral_store.ts b/app/store/ephemeral_store.ts index 4a1f7857f..3591120c4 100644 --- a/app/store/ephemeral_store.ts +++ b/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} = {}; + private websocketEditingPost: {[serverUrl: string]: {[id: string]: {post: Post; timeout: NodeJS.Timeout} | undefined} | undefined} = {}; + private websocketRemovingPost: {[serverUrl: string]: Set | 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); diff --git a/patches/@nozbe+watermelondb+0.25.5.patch b/patches/@nozbe+watermelondb+0.25.5.patch index 7a5708d45..514f2af36 100644 --- a/patches/@nozbe+watermelondb+0.25.5.patch +++ b/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() {