[Gekidou MM-41093] CRT - WS Events, Actions, Queries, Thread Follow, Post Query (#6075)

* WS Events, Actions, Queries, Thread Follow, Post Query

* i18n changes

* Misc

* Only unread threads are marked as read

* Mark threads from WS even as visible in Global threads

* Merge fixes

* Update thread_post_list.tsx

* Merge fix

* Feedback fix

* Make teamId in handleThreads optional for unfollowed threads

* Removed unwated type and return

* Review changes

* Removing unused model

* Merge fix

* Misc fixes

* Following button query change

app/actions/local/post.ts

@@ -132,7 +132,7 @@ export const removePost = async (serverUrl: string, post: PostModel | Post) => {
     return {post};
 };
 
-export const markPostAsDeleted = async (serverUrl: string, post: Post) => {
+export const markPostAsDeleted = async (serverUrl: string, post: Post, prepareRecordsOnly = false) => {
     const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
     if (!operator) {
         return {error: `${serverUrl} database not found`};
@@ -140,41 +140,18 @@ export const markPostAsDeleted = async (serverUrl: string, post: Post) => {
 
     const dbPost = await getPostById(operator.database, post.id);
     if (!dbPost) {
-        return {};
+        return {error: 'Post not found'};
     }
 
-    dbPost.prepareUpdate((p) => {
+    const model = dbPost.prepareUpdate((p) => {
         p.deleteAt = Date.now();
         p.message = '';
         p.metadata = null;
         p.props = undefined;
     });
 
-    operator.batchRecords([dbPost]);
+    if (!prepareRecordsOnly) {
-
+        operator.batchRecords([dbPost]);
-    return {post: dbPost};
-};
-
-export const processPostsFetched = async (serverUrl: string, actionType: string, data: PostResponse, fetchOnly = false) => {
-    const order = data.order;
-    const posts = Object.values(data.posts) as Post[];
-    const previousPostId = data.prev_post_id;
-
-    if (!fetchOnly) {
-        const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
-        if (operator) {
-            await operator.handlePosts({
-                actionType,
-                order,
-                posts,
-                previousPostId,
-            });
-        }
     }
-
+    return {model};
-    return {
-        posts,
-        order,
-        previousPostId,
-    };
 };

app/actions/local/thread.ts

@@ -1,16 +1,19 @@
 // Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
 // See LICENSE.txt for license information.
 
-import {General, Screens} from '@constants';
+import {ActionType, General, Screens} from '@constants';
 import DatabaseManager from '@database/manager';
 import {getTranslations, t} from '@i18n';
 import {getChannelById} from '@queries/servers/channel';
 import {getPostById} from '@queries/servers/post';
+import {getIsCRTEnabled, getThreadById, prepareThreadsFromReceivedPosts, queryThreadsInTeam} from '@queries/servers/thread';
 import {getCurrentUser} from '@queries/servers/user';
 import {goToScreen} from '@screens/navigation';
 import EphemeralStore from '@store/ephemeral_store';
 import {changeOpacity} from '@utils/theme';
 
+import type Model from '@nozbe/watermelondb/Model';
+
 export const switchToThread = async (serverUrl: string, rootId: string) => {
     const database = DatabaseManager.serverDatabases[serverUrl]?.database;
     if (!database) {
@@ -37,6 +40,25 @@ export const switchToThread = async (serverUrl: string, rootId: string) => {
             return {error: 'Theme not found'};
         }
 
+        // Modal right buttons
+        const rightButtons = [];
+
+        const isCRTEnabled = await getIsCRTEnabled(database);
+        if (isCRTEnabled) {
+            // CRT: Add follow/following button
+            rightButtons.push({
+                id: 'thread-follow-button',
+                component: {
+                    id: post.id,
+                    name: Screens.THREAD_FOLLOW_BUTTON,
+                    passProps: {
+                        teamId: channel.teamId,
+                        threadId: post.id,
+                    },
+                },
+            });
+        }
+
         // Get translation by user locale
         const translations = getTranslations(user.locale);
 
@@ -62,6 +84,7 @@ export const switchToThread = async (serverUrl: string, rootId: string) => {
                     color: changeOpacity(theme.sidebarHeaderTextColor, 0.72),
                     text: subtitle,
                 },
+                rightButtons,
             },
         });
         return {};
@@ -69,3 +92,155 @@ export const switchToThread = async (serverUrl: string, rootId: string) => {
         return {error};
     }
 };
+
+// When new post arrives:
+// 1. If a reply, then update the reply_count, add user as the participant
+// 2. Else add the post as a thread
+export const createThreadFromNewPost = async (serverUrl: string, post: Post, prepareRecordsOnly = false) => {
+    const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
+    if (!operator) {
+        return {error: `${serverUrl} database not found`};
+    }
+
+    const models: Model[] = [];
+    if (post.root_id) {
+        // Update the thread data: `reply_count`
+        const {model: threadModel} = await updateThread(serverUrl, post.root_id, {reply_count: post.reply_count}, true);
+        if (threadModel) {
+            models.push(threadModel);
+        }
+
+        // Add user as a participant to the thread
+        const threadParticipantModels = await operator.handleThreadParticipants({
+            threadsParticipants: [{
+                thread_id: post.root_id,
+                participants: [{
+                    thread_id: post.root_id,
+                    id: post.user_id,
+                }],
+            }],
+            prepareRecordsOnly: true,
+            skipSync: true,
+        });
+        if (threadParticipantModels?.length) {
+            models.push(...threadParticipantModels);
+        }
+    } else { // If the post is a root post, then we need to add it to the thread table
+        const threadModels = await prepareThreadsFromReceivedPosts(operator, [post]);
+        if (threadModels?.length) {
+            models.push(...threadModels);
+        }
+    }
+
+    if (models.length && !prepareRecordsOnly) {
+        await operator.batchRecords(models);
+    }
+
+    return {models};
+};
+
+// On receiving threads, Along with the "threads" & "thread participants", extract and save "posts" & "users"
+export const processReceivedThreads = async (serverUrl: string, threads: Thread[], teamId: string, prepareRecordsOnly = false) => {
+    const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
+    if (!operator) {
+        return {error: `${serverUrl} database not found`};
+    }
+
+    const models: Model[] = [];
+
+    const posts: Post[] = [];
+    const users: UserProfile[] = [];
+
+    // Extract posts & users from the received threads
+    for (let i = 0; i < threads.length; i++) {
+        const {participants, post} = threads[i];
+        posts.push(post);
+        participants.forEach((participant) => users.push(participant));
+    }
+
+    const postModels = await operator.handlePosts({
+        actionType: ActionType.POSTS.RECEIVED_IN_CHANNEL,
+        order: [],
+        posts,
+        prepareRecordsOnly: true,
+    });
+
+    if (postModels.length) {
+        models.push(...postModels);
+    }
+
+    const threadModels = await operator.handleThreads({
+        threads,
+        teamId,
+        prepareRecordsOnly: true,
+    });
+
+    if (threadModels.length) {
+        models.push(...threadModels);
+    }
+
+    const userModels = await operator.handleUsers({
+        users,
+        prepareRecordsOnly: true,
+    });
+
+    if (userModels.length) {
+        models.push(...userModels);
+    }
+
+    if (models.length && !prepareRecordsOnly) {
+        await operator.batchRecords(models);
+    }
+    return {models};
+};
+
+export const markTeamThreadsAsRead = async (serverUrl: string, teamId: string, prepareRecordsOnly = false) => {
+    const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
+    if (!operator) {
+        return {error: `${serverUrl} database not found`};
+    }
+    try {
+        const {database} = operator;
+        const threads = await queryThreadsInTeam(database, teamId, true).fetch();
+        const models = threads.map((thread) => thread.prepareUpdate((record) => {
+            record.unreadMentions = 0;
+            record.unreadReplies = 0;
+            record.lastViewedAt = Date.now();
+        }));
+        if (!prepareRecordsOnly) {
+            await operator.batchRecords(models);
+        }
+        return {models};
+    } catch (error) {
+        return {error};
+    }
+};
+
+export const updateThread = async (serverUrl: string, threadId: string, updatedThread: Partial<Thread>, prepareRecordsOnly = false) => {
+    const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
+    if (!operator) {
+        return {error: `${serverUrl} database not found`};
+    }
+
+    try {
+        const {database} = operator;
+        const thread = await getThreadById(database, threadId);
+        if (thread) {
+            const model = thread.prepareUpdate((record) => {
+                record.isFollowing = updatedThread.is_following ?? record.isFollowing;
+                record.replyCount = updatedThread.reply_count ?? record.replyCount;
+
+                record.lastViewedAt = updatedThread.last_viewed_at ?? record.lastViewedAt;
+                record.unreadMentions = updatedThread.unread_mentions ?? record.unreadMentions;
+                record.unreadReplies = updatedThread.unread_replies ?? record.unreadReplies;
+            });
+            if (!prepareRecordsOnly) {
+                await operator.batchRecords([model]);
+            }
+            return {model};
+        }
+        return {error: 'Thread not found'};
+    } catch (error) {
+        return {error};
+    }
+};

app/actions/remote/post.ts

@@ -7,8 +7,9 @@
 import {DeviceEventEmitter} from 'react-native';
 
 import {markChannelAsUnread, updateLastPostAt} from '@actions/local/channel';
-import {processPostsFetched, removePost} from '@actions/local/post';
+import {removePost} from '@actions/local/post';
 import {addRecentReaction} from '@actions/local/reactions';
+import {createThreadFromNewPost} from '@actions/local/thread';
 import {ActionType, Events, General, Post, ServerErrors} from '@constants';
 import {MM_TABLES, SYSTEM_IDENTIFIERS} from '@constants/database';
 import DatabaseManager from '@database/manager';
@@ -20,8 +21,10 @@ import {prepareMissingChannelsForAllTeams, queryAllMyChannel} from '@queries/ser
 import {queryAllCustomEmojis} from '@queries/servers/custom_emoji';
 import {getPostById, getRecentPostsInChannel} from '@queries/servers/post';
 import {getCurrentUserId, getCurrentChannelId} from '@queries/servers/system';
+import {getIsCRTEnabled, prepareThreadsFromReceivedPosts} from '@queries/servers/thread';
 import {queryAllUsers} from '@queries/servers/user';
 import {getValidEmojis, matchEmoticons} from '@utils/emoji/helpers';
+import {processPostsFetched} from '@utils/post';
 import {getPostIdsForCombinedUserActivityPost} from '@utils/post_list';
 
 import {forceLogoutIfNecessary} from './session';
@@ -62,11 +65,13 @@ export const createPost = async (serverUrl: string, post: Partial<Post>, files:
         return {error};
     }
 
-    const currentUserId = await getCurrentUserId(operator.database);
+    const {database} = operator;
+
+    const currentUserId = await getCurrentUserId(database);
     const timestamp = Date.now();
     const pendingPostId = post.pending_post_id || `${currentUserId}:${timestamp}`;
 
-    const existing = await getPostById(operator.database, pendingPostId);
+    const existing = await getPostById(database, pendingPostId);
     if (existing && !existing.props.failed) {
         return {data: false};
     }
@@ -111,22 +116,33 @@ export const createPost = async (serverUrl: string, post: Partial<Post>, files:
         initialPostModels.push(...postModels);
     }
 
-    const customEmojis = await queryAllCustomEmojis(operator.database).fetch();
+    const customEmojis = await queryAllCustomEmojis(database).fetch();
     const emojisInMessage = matchEmoticons(newPost.message);
     const reactionModels = await addRecentReaction(serverUrl, getValidEmojis(emojisInMessage, customEmojis), true);
     if (!('error' in reactionModels) && reactionModels.length) {
         initialPostModels.push(...reactionModels);
     }
 
-    operator.batchRecords(initialPostModels);
+    await operator.batchRecords(initialPostModels);
+
+    const isCRTEnabled = await getIsCRTEnabled(database);
 
     try {
         const created = await client.createPost(newPost);
-        await operator.handlePosts({
+        const models: Model[] = await operator.handlePosts({
             actionType: ActionType.POSTS.RECEIVED_NEW,
             order: [created.id],
             posts: [created],
+            prepareRecordsOnly: true,
         });
+        if (isCRTEnabled) {
+            const {models: threadModels} = await createThreadFromNewPost(serverUrl, created, true);
+            if (threadModels?.length) {
+                models.push(...threadModels);
+            }
+        }
+        await operator.batchRecords(models);
+
         newPost = created;
     } catch (error: any) {
         const errorPost = {
@@ -147,11 +163,19 @@ export const createPost = async (serverUrl: string, post: Partial<Post>, files:
         ) {
             await removePost(serverUrl, databasePost);
         } else {
-            await operator.handlePosts({
+            const models: Model[] = await operator.handlePosts({
                 actionType: ActionType.POSTS.RECEIVED_NEW,
                 order: [errorPost.id],
                 posts: [errorPost],
+                prepareRecordsOnly: true,
             });
+            if (isCRTEnabled) {
+                const {models: threadModels} = await createThreadFromNewPost(serverUrl, errorPost, true);
+                if (threadModels?.length) {
+                    models.push(...threadModels);
+                }
+            }
+            await operator.batchRecords(models);
         }
     }
 
@@ -229,6 +253,14 @@ export const fetchPostsForChannel = async (serverUrl: string, channelId: string,
                 models.push(memberModel);
             }
 
+            const isCRTEnabled = await getIsCRTEnabled(operator.database);
+            if (isCRTEnabled) {
+                const threadModels = await prepareThreadsFromReceivedPosts(operator, data.posts);
+                if (threadModels?.length) {
+                    models.push(...threadModels);
+                }
+            }
+
             if (models.length) {
                 await operator.batchRecords(models);
             }
@@ -259,6 +291,10 @@ export const fetchPostsForUnreadChannels = async (serverUrl: string, channels: C
 };
 
 export const fetchPosts = async (serverUrl: string, channelId: string, page = 0, perPage = General.POST_CHUNK_SIZE, fetchOnly = false): Promise<PostsRequest> => {
+    const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
+    if (!operator) {
+        return {error: `${serverUrl} database not found`};
+    }
     let client: Client;
     try {
         client = NetworkManager.getClient(serverUrl);
@@ -267,8 +303,26 @@ export const fetchPosts = async (serverUrl: string, channelId: string, page = 0,
     }
 
     try {
-        const data = await client.getPosts(channelId, page, perPage);
+        const isCRTEnabled = await getIsCRTEnabled(operator.database);
-        return processPostsFetched(serverUrl, ActionType.POSTS.RECEIVED_IN_CHANNEL, data, fetchOnly);
+        const data = await client.getPosts(channelId, page, perPage, isCRTEnabled, isCRTEnabled);
+        const result = await processPostsFetched(data);
+        if (!fetchOnly) {
+            const models = await operator.handlePosts({
+                ...result,
+                actionType: ActionType.POSTS.RECEIVED_SINCE,
+                prepareRecordsOnly: true,
+            });
+            if (isCRTEnabled) {
+                const threadModels = await prepareThreadsFromReceivedPosts(operator, result.posts);
+                if (threadModels?.length) {
+                    models.push(...threadModels);
+                }
+            }
+            if (models.length) {
+                await operator.batchRecords(models);
+            }
+        }
+        return result;
     } catch (error) {
         forceLogoutIfNecessary(serverUrl, error as ClientErrorProps);
         return {error};
@@ -294,8 +348,9 @@ export const fetchPostsBefore = async (serverUrl: string, channelId: string, pos
         if (activeServerUrl === serverUrl) {
             DeviceEventEmitter.emit(Events.LOADING_CHANNEL_POSTS, true);
         }
-        const data = await client.getPostsBefore(channelId, postId, 0, perPage);
+        const isCRTEnabled = await getIsCRTEnabled(operator.database);
-        const result = await processPostsFetched(serverUrl, ActionType.POSTS.RECEIVED_BEFORE, data, true);
+        const data = await client.getPostsBefore(channelId, postId, 0, perPage, isCRTEnabled, isCRTEnabled);
+        const result = await processPostsFetched(data);
 
         if (activeServerUrl === serverUrl) {
             DeviceEventEmitter.emit(Events.LOADING_CHANNEL_POSTS, false);
@@ -317,6 +372,13 @@ export const fetchPostsBefore = async (serverUrl: string, channelId: string, pos
                     models.push(...userModels);
                 }
 
+                if (isCRTEnabled) {
+                    const threadModels = await prepareThreadsFromReceivedPosts(operator, result.posts);
+                    if (threadModels?.length) {
+                        models.push(...threadModels);
+                    }
+                }
+
                 await operator.batchRecords(models);
             } catch (error) {
                 // eslint-disable-next-line no-console
@@ -335,6 +397,11 @@ export const fetchPostsBefore = async (serverUrl: string, channelId: string, pos
 };
 
 export const fetchPostsSince = async (serverUrl: string, channelId: string, since: number, fetchOnly = false): Promise<PostsRequest> => {
+    const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
+    if (!operator) {
+        return {error: `${serverUrl} database not found`};
+    }
+
     let client: Client;
     try {
         client = NetworkManager.getClient(serverUrl);
@@ -343,8 +410,26 @@ export const fetchPostsSince = async (serverUrl: string, channelId: string, sinc
     }
 
     try {
-        const data = await client.getPostsSince(channelId, since);
+        const isCRTEnabled = await getIsCRTEnabled(operator.database);
-        return processPostsFetched(serverUrl, ActionType.POSTS.RECEIVED_SINCE, data, fetchOnly);
+        const data = await client.getPostsSince(channelId, since, isCRTEnabled, isCRTEnabled);
+        const result = await processPostsFetched(data);
+        if (!fetchOnly) {
+            const models = await operator.handlePosts({
+                ...result,
+                actionType: ActionType.POSTS.RECEIVED_SINCE,
+                prepareRecordsOnly: true,
+            });
+            if (isCRTEnabled) {
+                const threadModels = await prepareThreadsFromReceivedPosts(operator, result.posts);
+                if (threadModels?.length) {
+                    models.push(...threadModels);
+                }
+            }
+            if (models.length) {
+                await operator.batchRecords(models);
+            }
+        }
+        return result;
     } catch (error) {
         forceLogoutIfNecessary(serverUrl, error as ClientErrorProps);
         return {error};
@@ -423,6 +508,11 @@ export const fetchPostAuthors = async (serverUrl: string, posts: Post[], fetchOn
 };
 
 export const fetchPostThread = async (serverUrl: string, postId: string, fetchOnly = false): Promise<PostsRequest> => {
+    const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
+    if (!operator) {
+        return {error: `${serverUrl} database not found`};
+    }
+
     let client: Client;
     try {
         client = NetworkManager.getClient(serverUrl);
@@ -432,7 +522,25 @@ export const fetchPostThread = async (serverUrl: string, postId: string, fetchOn
 
     try {
         const data = await client.getPostThread(postId);
-        return processPostsFetched(serverUrl, ActionType.POSTS.RECEIVED_IN_THREAD, data, fetchOnly);
+        const result = processPostsFetched(data);
+        if (!fetchOnly) {
+            const models = await operator.handlePosts({
+                ...result,
+                actionType: ActionType.POSTS.RECEIVED_IN_THREAD,
+                prepareRecordsOnly: true,
+            });
+            const isCRTEnabled = await getIsCRTEnabled(operator.database);
+            if (isCRTEnabled) {
+                const threadModels = await prepareThreadsFromReceivedPosts(operator, result.posts);
+                if (threadModels?.length) {
+                    models.push(...threadModels);
+                }
+            }
+            if (models.length) {
+                await operator.batchRecords(models);
+            }
+        }
+        return result;
     } catch (error) {
         forceLogoutIfNecessary(serverUrl, error as ClientErrorProps);
         return {error};
@@ -468,7 +576,7 @@ export async function fetchPostsAround(serverUrl: string, channelId: string, pos
             order: [],
         };
 
-        const data = await processPostsFetched(serverUrl, ActionType.POSTS.RECEIVED_AROUND, preData, true);
+        const data = processPostsFetched(preData);
 
         let posts: Model[] = [];
         const models: Model[] = [];
@@ -494,6 +602,14 @@ export async function fetchPostsAround(serverUrl: string, channelId: string, pos
             });
 
             models.push(...posts);
+
+            const isCRTEnabled = await getIsCRTEnabled(operator.database);
+            if (isCRTEnabled) {
+                const threadModels = await prepareThreadsFromReceivedPosts(operator, data.posts);
+                if (threadModels?.length) {
+                    models.push(...threadModels);
+                }
+            }
             await operator.batchRecords(models);
         }
 
@@ -579,9 +695,8 @@ export async function fetchMissingChannelsFromPosts(serverUrl: string, posts: Po
                         }
                         return mdls;
                     });
-
+                    if (models.length) {
-                    if (models) {
+                        await operator.batchRecords(models);
-                        operator.batchRecords(models);
                     }
                 }
             }
@@ -630,6 +745,14 @@ export const fetchPostById = async (serverUrl: string, postId: string, fetchOnly
                 models.push(...users);
             }
 
+            const isCRTEnabled = await getIsCRTEnabled(operator.database);
+            if (isCRTEnabled) {
+                const threadModels = await prepareThreadsFromReceivedPosts(operator, [post]);
+                if (threadModels?.length) {
+                    models.push(...threadModels);
+                }
+            }
+
             await operator.batchRecords(models);
         }
 
@@ -833,6 +956,11 @@ export async function fetchSavedPosts(serverUrl: string, teamId?: string, channe
             }),
         );
 
+        const isCRTEnabled = await getIsCRTEnabled(operator.database);
+        if (isCRTEnabled) {
+            promises.push(prepareThreadsFromReceivedPosts(operator, postsArray));
+        }
+
         const modelArrays = await Promise.all(promises);
         const models = modelArrays.flatMap((mdls) => {
             if (!mdls || !mdls.length) {

app/actions/remote/search.ts

@@ -1,12 +1,12 @@
 // Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
 // See LICENSE.txt for license information.
 
-import {processPostsFetched} from '@actions/local/post';
 import {SYSTEM_IDENTIFIERS} from '@constants/database';
 import DatabaseManager from '@database/manager';
 import NetworkManager from '@init/network_manager';
 import {prepareMissingChannelsForAllTeams} from '@queries/servers/channel';
 import {getCurrentUser} from '@queries/servers/user';
+import {processPostsFetched} from '@utils/post';
 
 import {fetchPostAuthors, fetchMissingChannelsFromPosts} from './post';
 import {forceLogoutIfNecessary} from './session';
@@ -119,6 +119,12 @@ export async function fetchRecentMentions(serverUrl: string): Promise<PostSearch
 }
 
 export const searchPosts = async (serverUrl: string, params: PostSearchParams): Promise<PostSearchRequest> => {
+    const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
+
+    if (!operator) {
+        return {error: `${serverUrl} database not found`};
+    }
+
     let client: Client;
     try {
         client = NetworkManager.getClient(serverUrl);
@@ -134,5 +140,11 @@ export const searchPosts = async (serverUrl: string, params: PostSearchParams):
         return {error};
     }
 
-    return processPostsFetched(serverUrl, '', data, false);
+    const result = processPostsFetched(data);
+    await operator.handlePosts({
+        ...result,
+        actionType: '',
+    });
+
+    return result;
 };

app/actions/remote/thread.ts

@@ -1,13 +1,196 @@
 // Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
 // See LICENSE.txt for license information.
 
-import {switchToThread} from '@actions/local/thread';
+import {markTeamThreadsAsRead, processReceivedThreads, switchToThread, updateThread} from '@actions/local/thread';
 import {fetchPostThread} from '@actions/remote/post';
+import {General} from '@constants';
+import DatabaseManager from '@database/manager';
+import NetworkManager from '@init/network_manager';
+import {getChannelById} from '@queries/servers/channel';
+import {getPostById} from '@queries/servers/post';
+import {getCommonSystemValues} from '@queries/servers/system';
+import {getIsCRTEnabled, getThreadById} from '@queries/servers/thread';
+
+import {forceLogoutIfNecessary} from './session';
+
+type FetchThreadsRequest = {
+    error?: unknown;
+} | {
+    data: GetUserThreadsResponse;
+};
+
+type FetchThreadsOptions = {
+    before?: string;
+    after?: string;
+    perPage?: number;
+    deleted?: boolean;
+    unread?: boolean;
+    since?: number;
+}
 
 export const fetchAndSwitchToThread = async (serverUrl: string, rootId: string) => {
+    const database = DatabaseManager.serverDatabases[serverUrl]?.database;
+    if (!database) {
+        return {error: `${serverUrl} database not found`};
+    }
+
     // Load thread before we open to the thread modal
-    // https://mattermost.atlassian.net/browse/MM-42232
+    // @Todo: https://mattermost.atlassian.net/browse/MM-42232
     fetchPostThread(serverUrl, rootId);
 
+    // Mark thread as read
+    const isCRTEnabled = await getIsCRTEnabled(database);
+    if (isCRTEnabled) {
+        const post = await getPostById(database, rootId);
+        if (post) {
+            const thread = await getThreadById(database, rootId);
+            if (thread?.unreadReplies || thread?.unreadMentions) {
+                const channel = await getChannelById(database, post.channelId);
+                if (channel) {
+                    updateThreadRead(serverUrl, channel.teamId, thread.id, Date.now());
+                }
+            }
+        }
+    }
+
     switchToThread(serverUrl, rootId);
+
+    return {};
+};
+
+export const fetchThreads = async (
+    serverUrl: string,
+    teamId: string,
+    {
+        before,
+        after,
+        perPage = General.CRT_CHUNK_SIZE,
+        deleted = false,
+        unread = false,
+        since = 0,
+    }: FetchThreadsOptions = {
+        perPage: General.CRT_CHUNK_SIZE,
+        deleted: false,
+        unread: false,
+        since: 0,
+    },
+): Promise<FetchThreadsRequest> => {
+    const database = DatabaseManager.serverDatabases[serverUrl]?.database;
+    if (!database) {
+        return {error: `${serverUrl} database not found`};
+    }
+
+    let client;
+    try {
+        client = NetworkManager.getClient(serverUrl);
+    } catch (error) {
+        return {error};
+    }
+
+    try {
+        const {config} = await getCommonSystemValues(database);
+
+        const data = await client.getThreads('me', teamId, before, after, perPage, deleted, unread, since, config.Version);
+
+        const {threads} = data;
+
+        if (threads.length) {
+            // Mark all fetched threads as following
+            threads.forEach((thread: Thread) => {
+                thread.is_following = true;
+            });
+
+            await processReceivedThreads(serverUrl, threads, teamId);
+        }
+
+        return {data};
+    } catch (error) {
+        forceLogoutIfNecessary(serverUrl, error as ClientErrorProps);
+        return {error};
+    }
+};
+
+export const fetchThread = async (serverUrl: string, teamId: string, threadId: string, extended?: boolean) => {
+    let client;
+    try {
+        client = NetworkManager.getClient(serverUrl);
+    } catch (error) {
+        return {error};
+    }
+
+    try {
+        const thread = await client.getThread('me', teamId, threadId, extended);
+
+        await processReceivedThreads(serverUrl, [thread], teamId);
+
+        return {data: thread};
+    } catch (error) {
+        forceLogoutIfNecessary(serverUrl, error as ClientErrorProps);
+        return {error};
+    }
+};
+
+export const updateTeamThreadsAsRead = async (serverUrl: string, teamId: string) => {
+    let client;
+    try {
+        client = NetworkManager.getClient(serverUrl);
+    } catch (error) {
+        return {error};
+    }
+
+    try {
+        const data = await client.updateTeamThreadsAsRead('me', teamId);
+
+        // Update locally
+        await markTeamThreadsAsRead(serverUrl, teamId);
+
+        return {data};
+    } catch (error) {
+        forceLogoutIfNecessary(serverUrl, error as ClientErrorProps);
+        return {error};
+    }
+};
+
+export const updateThreadRead = async (serverUrl: string, teamId: string, threadId: string, timestamp: number) => {
+    let client;
+    try {
+        client = NetworkManager.getClient(serverUrl);
+    } catch (error) {
+        return {error};
+    }
+
+    try {
+        const data = await client.updateThreadRead('me', teamId, threadId, timestamp);
+
+        // Update locally
+        await updateThread(serverUrl, threadId, {
+            last_viewed_at: timestamp,
+        });
+
+        return {data};
+    } catch (error) {
+        forceLogoutIfNecessary(serverUrl, error as ClientErrorProps);
+        return {error};
+    }
+};
+
+export const updateThreadFollowing = async (serverUrl: string, teamId: string, threadId: string, state: boolean) => {
+    let client;
+    try {
+        client = NetworkManager.getClient(serverUrl);
+    } catch (error) {
+        return {error};
+    }
+
+    try {
+        const data = await client.updateThreadFollow('me', teamId, threadId, state);
+
+        // Update locally
+        await updateThread(serverUrl, threadId, {is_following: state});
+
+        return {data};
+    } catch (error) {
+        forceLogoutIfNecessary(serverUrl, error as ClientErrorProps);
+        return {error};
+    }
 };

app/actions/websocket/index.ts

@@ -25,6 +25,7 @@ import {handleAddCustomEmoji, handleReactionRemovedFromPostEvent, handleReaction
 import {handleUserRoleUpdatedEvent, handleTeamMemberRoleUpdatedEvent, handleRoleUpdatedEvent} from './roles';
 import {handleLicenseChangedEvent, handleConfigChangedEvent} from './system';
 import {handleLeaveTeamEvent, handleUserAddedToTeamEvent, handleUpdateTeamEvent} from './teams';
+import {handleThreadUpdatedEvent, handleThreadReadChangedEvent, handleThreadFollowChangedEvent} from './threads';
 import {handleUserUpdatedEvent, handleUserTypingEvent} from './users';
 
 // ESR: 5.37
@@ -325,17 +326,17 @@ export async function handleEvent(serverUrl: string, msg: WebSocketMessage) {
             break;
 
         case WebsocketEvents.THREAD_UPDATED:
+            handleThreadUpdatedEvent(serverUrl, msg);
             break;
 
-        // return dispatch(handleThreadUpdated(msg));
         case WebsocketEvents.THREAD_READ_CHANGED:
+            handleThreadReadChangedEvent(serverUrl, msg);
             break;
 
-        // return dispatch(handleThreadReadChanged(msg));
         case WebsocketEvents.THREAD_FOLLOW_CHANGED:
+            handleThreadFollowChangedEvent(serverUrl, msg);
             break;
 
-        // return dispatch(handleThreadFollowChanged(msg));
         case WebsocketEvents.APPS_FRAMEWORK_REFRESH_BINDINGS:
             break;
 

app/actions/websocket/posts.ts

@@ -6,13 +6,16 @@ import {DeviceEventEmitter} from 'react-native';
 
 import {storeMyChannelsForTeam, markChannelAsUnread, markChannelAsViewed, updateLastPostAt} from '@actions/local/channel';
 import {markPostAsDeleted} from '@actions/local/post';
+import {createThreadFromNewPost, updateThread} from '@actions/local/thread';
 import {fetchMyChannel, markChannelAsRead} from '@actions/remote/channel';
 import {fetchPostAuthors, fetchPostById} from '@actions/remote/post';
+import {fetchThread} from '@actions/remote/thread';
 import {ActionType, Events} from '@constants';
 import DatabaseManager from '@database/manager';
-import {getMyChannel} from '@queries/servers/channel';
+import {getChannelById, getMyChannel} from '@queries/servers/channel';
 import {getPostById} from '@queries/servers/post';
 import {getCurrentChannelId, getCurrentUserId} from '@queries/servers/system';
+import {getIsCRTEnabled} from '@queries/servers/thread';
 import {isFromWebhook, isSystemMessage, shouldIgnorePost} from '@utils/post';
 
 import type MyChannelModel from '@typings/database/models/servers/my_channel';
@@ -31,15 +34,17 @@ export async function handleNewPostEvent(serverUrl: string, msg: WebSocketMessag
         return;
     }
 
+    const {database} = operator;
+
     let post: Post;
     try {
         post = JSON.parse(msg.data.post);
     } catch {
         return;
     }
-    const currentUserId = await getCurrentUserId(operator.database);
+    const currentUserId = await getCurrentUserId(database);
 
-    const existing = await getPostById(operator.database, post.pending_post_id);
+    const existing = await getPostById(database, post.pending_post_id);
 
     if (existing) {
         return;
@@ -58,8 +63,16 @@ export async function handleNewPostEvent(serverUrl: string, msg: WebSocketMessag
         models.push(...postModels);
     }
 
+    const isCRTEnabled = await getIsCRTEnabled(database);
+    if (isCRTEnabled) {
+        const {models: threadModels} = await createThreadFromNewPost(serverUrl, post, true);
+        if (threadModels?.length) {
+            models.push(...threadModels);
+        }
+    }
+
     // Ensure the channel membership
-    let myChannel = await getMyChannel(operator.database, post.channel_id);
+    let myChannel = await getMyChannel(database, post.channel_id);
     if (myChannel) {
         const {member} = await updateLastPostAt(serverUrl, post.channel_id, post.create_at, false);
         if (member) {
@@ -77,7 +90,7 @@ export async function handleNewPostEvent(serverUrl: string, msg: WebSocketMessag
             return;
         }
 
-        myChannel = await getMyChannel(operator.database, post.channel_id);
+        myChannel = await getMyChannel(database, post.channel_id);
         if (!myChannel) {
             return;
         }
@@ -85,14 +98,14 @@ export async function handleNewPostEvent(serverUrl: string, msg: WebSocketMessag
 
     // If we don't have the root post for this post, fetch it from the server
     if (post.root_id) {
-        const rootPost = await getPostById(operator.database, post.root_id);
+        const rootPost = await getPostById(database, post.root_id);
 
         if (!rootPost) {
             fetchPostById(serverUrl, post.root_id);
         }
     }
 
-    const currentChannelId = await getCurrentChannelId(operator.database);
+    const currentChannelId = await getCurrentChannelId(database);
 
     if (post.channel_id === currentChannelId) {
         const data = {
@@ -112,11 +125,6 @@ export async function handleNewPostEvent(serverUrl: string, msg: WebSocketMessag
         }
     }
 
-    // TODO Thread related functionality: https://mattermost.atlassian.net/browse/MM-41084
-    //const viewingGlobalThreads = getViewingGlobalThreads(state);
-    // const collapsedThreadsEnabled = isCollapsedThreadsEnabled(state);
-    // actions.push(receivedNewPost(post, collapsedThreadsEnabled));
-
     if (!shouldIgnorePost(post)) {
         let markAsViewed = false;
         let markAsRead = false;
@@ -203,10 +211,44 @@ export async function handlePostEdited(serverUrl: string, msg: WebSocketMessage)
     }
 }
 
-export function handlePostDeleted(serverUrl: string, msg: WebSocketMessage) {
+export async function handlePostDeleted(serverUrl: string, msg: WebSocketMessage) {
+    const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
+    if (!operator) {
+        return;
+    }
     try {
-        const data: Post = JSON.parse(msg.data.post);
+        const {database} = operator;
-        markPostAsDeleted(serverUrl, data);
+
+        const post: Post = JSON.parse(msg.data.post);
+
+        const models: Model[] = [];
+
+        const {model: deleteModel} = await markPostAsDeleted(serverUrl, post, true);
+        if (deleteModel) {
+            models.push(deleteModel);
+        }
+
+        // update thread when a reply is deleted and CRT is enabled
+        if (post.root_id) {
+            const isCRTEnabled = await getIsCRTEnabled(database);
+            if (isCRTEnabled) {
+                // Update reply_count of the thread;
+                // Note: reply_count includes current deleted count, So subtract 1 from reply_count
+                const {model: threadModel} = await updateThread(serverUrl, post.root_id, {reply_count: post.reply_count - 1}, true);
+                if (threadModel) {
+                    models.push(threadModel);
+                }
+
+                const channel = await getChannelById(database, post.channel_id);
+                if (channel) {
+                    fetchThread(serverUrl, channel.teamId, post.root_id);
+                }
+            }
+        }
+
+        if (models.length) {
+            await operator.batchRecords(models);
+        }
     } catch {
         // Do nothing
     }

app/actions/websocket/threads.ts

@@ -0,0 +1,65 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+import {markTeamThreadsAsRead, processReceivedThreads, updateThread} from '@actions/local/thread';
+import DatabaseManager from '@database/manager';
+
+export async function handleThreadUpdatedEvent(serverUrl: string, msg: WebSocketMessage): Promise<void> {
+    try {
+        const thread = JSON.parse(msg.data.thread) as Thread;
+        const teamId = msg.broadcast.team_id;
+
+        // Mark it as following
+        thread.is_following = true;
+        processReceivedThreads(serverUrl, [thread], teamId);
+    } catch (error) {
+        // Do nothing
+    }
+}
+
+export async function handleThreadReadChangedEvent(serverUrl: string, msg: WebSocketMessage): Promise<void> {
+    const operator = DatabaseManager.serverDatabases[serverUrl].operator;
+    if (!operator) {
+        return;
+    }
+
+    try {
+        if (operator) {
+            const {thread_id, timestamp, unread_mentions, unread_replies} = msg.data as ThreadReadChangedData;
+            if (thread_id) {
+                await updateThread(serverUrl, thread_id, {
+                    last_viewed_at: timestamp,
+                    unread_mentions,
+                    unread_replies,
+                });
+            } else {
+                await markTeamThreadsAsRead(serverUrl, msg.broadcast.team_id);
+            }
+        }
+    } catch (error) {
+        // Do nothing
+    }
+}
+
+export async function handleThreadFollowChangedEvent(serverUrl: string, msg: WebSocketMessage): Promise<void> {
+    const operator = DatabaseManager.serverDatabases[serverUrl].operator;
+    if (!operator) {
+        return;
+    }
+
+    try {
+        if (operator) {
+            const {reply_count, state, thread_id} = msg.data as {
+                reply_count: number;
+                state: boolean;
+                thread_id: string;
+            };
+            await updateThread(serverUrl, thread_id, {
+                is_following: state,
+                reply_count,
+            });
+        }
+    } catch (error) {
+        // Do nothing
+    }
+}

app/client/rest/base.ts

@@ -193,6 +193,14 @@ export default class ClientBase {
         return `${this.urlVersion}/redirect_location`;
     }
 
+    getThreadsRoute(userId: string, teamId: string): string {
+        return `${this.getUserRoute(userId)}/teams/${teamId}/threads`;
+    }
+
+    getThreadRoute(userId: string, teamId: string, threadId: string): string {
+        return `${this.getThreadsRoute(userId, teamId)}/${threadId}`;
+    }
+
     getAppsProxyRoute() {
         return '/plugins/com.mattermost.apps';
     }

app/client/rest/index.ts

@@ -16,6 +16,7 @@ import ClientIntegrations, {ClientIntegrationsMix} from './integrations';
 import ClientPosts, {ClientPostsMix} from './posts';
 import ClientPreferences, {ClientPreferencesMix} from './preferences';
 import ClientTeams, {ClientTeamsMix} from './teams';
+import ClientThreads, {ClientThreadsMix} from './threads';
 import ClientTos, {ClientTosMix} from './tos';
 import ClientUsers, {ClientUsersMix} from './users';
 
@@ -33,6 +34,7 @@ interface Client extends ClientBase,
     ClientPostsMix,
     ClientPreferencesMix,
     ClientTeamsMix,
+    ClientThreadsMix,
     ClientTosMix,
     ClientUsersMix
 {}
@@ -49,6 +51,7 @@ class Client extends mix(ClientBase).with(
     ClientPosts,
     ClientPreferences,
     ClientTeams,
+    ClientThreads,
     ClientTos,
     ClientUsers,
 ) {

app/client/rest/posts.ts

@@ -11,11 +11,11 @@ export interface ClientPostsMix {
     getPost: (postId: string) => Promise<Post>;
     patchPost: (postPatch: Partial<Post> & {id: string}) => Promise<Post>;
     deletePost: (postId: string) => Promise<any>;
-    getPostThread: (postId: string) => Promise<any>;
+    getPostThread: (postId: string, collapsedThreads?: boolean, collapsedThreadsExtended?: boolean) => Promise<any>;
-    getPosts: (channelId: string, page?: number, perPage?: number) => Promise<PostResponse>;
+    getPosts: (channelId: string, page?: number, perPage?: number, collapsedThreads?: boolean, collapsedThreadsExtended?: boolean) => Promise<PostResponse>;
-    getPostsSince: (channelId: string, since: number) => Promise<PostResponse>;
+    getPostsSince: (channelId: string, since: number, collapsedThreads?: boolean, collapsedThreadsExtended?: boolean) => Promise<PostResponse>;
-    getPostsBefore: (channelId: string, postId: string, page?: number, perPage?: number) => Promise<PostResponse>;
+    getPostsBefore: (channelId: string, postId: string, page?: number, perPage?: number, collapsedThreads?: boolean, collapsedThreadsExtended?: boolean) => Promise<PostResponse>;
-    getPostsAfter: (channelId: string, postId: string, page?: number, perPage?: number) => Promise<PostResponse>;
+    getPostsAfter: (channelId: string, postId: string, page?: number, perPage?: number, collapsedThreads?: boolean, collapsedThreadsExtended?: boolean) => Promise<PostResponse>;
     getFileInfosForPost: (postId: string) => Promise<FileInfo[]>;
     getSavedPosts: (userId: string, channelId?: string, teamId?: string, page?: number, perPage?: number) => Promise<PostResponse>;
     getPinnedPosts: (channelId: string) => Promise<any>;
@@ -79,41 +79,41 @@ const ClientPosts = (superclass: any) => class extends superclass {
         );
     };
 
-    getPostThread = async (postId: string) => {
+    getPostThread = async (postId: string, collapsedThreads = false, collapsedThreadsExtended = false) => {
         return this.doFetch(
-            `${this.getPostRoute(postId)}/thread`,
+            `${this.getPostRoute(postId)}/thread${buildQueryString({collapsedThreads, collapsedThreadsExtended})}`,
             {method: 'get'},
         );
     };
 
-    getPosts = async (channelId: string, page = 0, perPage = PER_PAGE_DEFAULT) => {
+    getPosts = async (channelId: string, page = 0, perPage = PER_PAGE_DEFAULT, collapsedThreads = false, collapsedThreadsExtended = false) => {
         return this.doFetch(
-            `${this.getChannelRoute(channelId)}/posts${buildQueryString({page, per_page: perPage})}`,
+            `${this.getChannelRoute(channelId)}/posts${buildQueryString({page, per_page: perPage, collapsedThreads, collapsedThreadsExtended})}`,
             {method: 'get'},
         );
     };
 
-    getPostsSince = async (channelId: string, since: number) => {
+    getPostsSince = async (channelId: string, since: number, collapsedThreads = false, collapsedThreadsExtended = false) => {
         return this.doFetch(
-            `${this.getChannelRoute(channelId)}/posts${buildQueryString({since})}`,
+            `${this.getChannelRoute(channelId)}/posts${buildQueryString({since, collapsedThreads, collapsedThreadsExtended})}`,
             {method: 'get'},
         );
     };
 
-    getPostsBefore = async (channelId: string, postId: string, page = 0, perPage = PER_PAGE_DEFAULT) => {
+    getPostsBefore = async (channelId: string, postId: string, page = 0, perPage = PER_PAGE_DEFAULT, collapsedThreads = false, collapsedThreadsExtended = false) => {
         this.analytics.trackAPI('api_posts_get_before', {channel_id: channelId});
 
         return this.doFetch(
-            `${this.getChannelRoute(channelId)}/posts${buildQueryString({before: postId, page, per_page: perPage})}`,
+            `${this.getChannelRoute(channelId)}/posts${buildQueryString({before: postId, page, per_page: perPage, collapsedThreads, collapsedThreadsExtended})}`,
             {method: 'get'},
         );
     };
 
-    getPostsAfter = async (channelId: string, postId: string, page = 0, perPage = PER_PAGE_DEFAULT) => {
+    getPostsAfter = async (channelId: string, postId: string, page = 0, perPage = PER_PAGE_DEFAULT, collapsedThreads = false, collapsedThreadsExtended = false) => {
         this.analytics.trackAPI('api_posts_get_after', {channel_id: channelId});
 
         return this.doFetch(
-            `${this.getChannelRoute(channelId)}/posts${buildQueryString({after: postId, page, per_page: perPage})}`,
+            `${this.getChannelRoute(channelId)}/posts${buildQueryString({after: postId, page, per_page: perPage, collapsedThreads, collapsedThreadsExtended})}`,
             {method: 'get'},
         );
     };

app/client/rest/threads.ts

@@ -0,0 +1,69 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+import {buildQueryString, isMinimumServerVersion} from '@utils/helpers';
+
+import {PER_PAGE_DEFAULT} from './constants';
+
+export interface ClientThreadsMix {
+    getThreads: (userId: string, teamId: string, before?: string, after?: string, pageSize?: number, deleted?: boolean, unread?: boolean, since?: number, serverVersion?: string) => Promise<GetUserThreadsResponse>;
+    getThread: (userId: string, teamId: string, threadId: string, extended?: boolean) => Promise<any>;
+    updateTeamThreadsAsRead: (userId: string, teamId: string) => Promise<any>;
+    updateThreadRead: (userId: string, teamId: string, threadId: string, timestamp: number) => Promise<any>;
+    updateThreadFollow: (userId: string, teamId: string, threadId: string, state: boolean) => Promise<any>;
+}
+
+const ClientThreads = (superclass: any) => class extends superclass {
+    getThreads = async (userId: string, teamId: string, before = '', after = '', pageSize = PER_PAGE_DEFAULT, deleted = false, unread = false, since = 0, serverVersion = '') => {
+        const queryStringObj: Record<string, any> = {
+            extended: 'true',
+            before,
+            after,
+            deleted,
+            unread,
+            since,
+        };
+        if (serverVersion && isMinimumServerVersion(serverVersion, 6, 0)) {
+            queryStringObj.per_page = pageSize;
+        } else {
+            queryStringObj.pageSize = pageSize;
+        }
+        return this.doFetch(
+            `${this.getThreadsRoute(userId, teamId)}${buildQueryString(queryStringObj)}`,
+            {method: 'get'},
+        );
+    };
+
+    getThread = async (userId: string, teamId: string, threadId: string, extended = true) => {
+        return this.doFetch(
+            `${this.getThreadRoute(userId, teamId, threadId)}${buildQueryString({extended})}`,
+            {method: 'get'},
+        );
+    };
+
+    updateTeamThreadsAsRead = (userId: string, teamId: string) => {
+        const url = `${this.getThreadsRoute(userId, teamId)}/read`;
+        return this.doFetch(
+            url,
+            {method: 'put'},
+        );
+    };
+
+    updateThreadRead = (userId: string, teamId: string, threadId: string, timestamp: number) => {
+        const url = `${this.getThreadRoute(userId, teamId, threadId)}/read/${timestamp}`;
+        return this.doFetch(
+            url,
+            {method: 'put'},
+        );
+    };
+
+    updateThreadFollow = (userId: string, teamId: string, threadId: string, state: boolean) => {
+        const url = this.getThreadRoute(userId, teamId, threadId) + '/following';
+        return this.doFetch(
+            url,
+            {method: state ? 'put' : 'delete'},
+        );
+    };
+};
+
+export default ClientThreads;

app/constants/config.ts

@@ -0,0 +1,9 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+export default {
+    ALWAYS_ON: 'always_on',
+    DEFAULT_ON: 'default_on',
+    DISABLED: 'disabled',
+    TRUE: 'true',
+};

app/constants/general.ts

@@ -6,6 +6,7 @@ export default {
     POST_CHUNK_SIZE: 60,
     POST_AROUND_CHUNK_SIZE: 10,
     CHANNELS_CHUNK_SIZE: 50,
+    CRT_CHUNK_SIZE: 60,
     STATUS_INTERVAL: 60000,
     AUTOCOMPLETE_LIMIT_DEFAULT: 25,
     MENTION: 'mention',
@@ -30,9 +31,6 @@ export default {
     MIN_USERS_IN_GM: 3,
     MAX_GROUP_CHANNELS_FOR_PROFILES: 50,
     DEFAULT_AUTOLINKED_URL_SCHEMES: ['http', 'https', 'ftp', 'mailto', 'tel', 'mattermost'],
-    DISABLED: 'disabled',
-    DEFAULT_ON: 'default_on',
-    DEFAULT_OFF: 'default_off',
     PROFILE_CHUNK_SIZE: 100,
     SEARCH_TIMEOUT_MILLISECONDS: 100,
     AUTOCOMPLETE_SPLIT_CHARACTERS: ['.', '-', '_'],

app/constants/index.ts

@@ -4,6 +4,7 @@
 import ActionType from './action_type';
 import Apps from './apps';
 import Channel from './channel';
+import Config from './config';
 import {CustomStatusDuration} from './custom_status';
 import Database from './database';
 import DeepLink from './deep_linking';
@@ -30,6 +31,7 @@ import WebsocketEvents from './websocket';
 export {
     ActionType,
     Apps,
+    Config,
     CustomStatusDuration,
     Channel,
     Database,

app/constants/preferences.ts

@@ -11,6 +11,9 @@ const Preferences: Record<string, any> = {
     CATEGORY_FAVORITE_CHANNEL: 'favorite_channel',
     CATEGORY_AUTO_RESET_MANUAL_STATUS: 'auto_reset_manual_status',
     CATEGORY_NOTIFICATIONS: 'notifications',
+    COLLAPSED_REPLY_THREADS: 'collapsed_reply_threads',
+    COLLAPSED_REPLY_THREADS_OFF: 'off',
+    COLLAPSED_REPLY_THREADS_ON: 'on',
     COMMENTS: 'comments',
     COMMENTS_ANY: 'any',
     COMMENTS_ROOT: 'root',

app/constants/screens.ts

@@ -8,13 +8,13 @@ export const APP_FORM = 'AppForm';
 export const BOTTOM_SHEET = 'BottomSheet';
 export const BROWSE_CHANNELS = 'BrowseChannels';
 export const CHANNEL = 'Channel';
-export const CREATE_OR_EDIT_CHANNEL = 'CreateOrEditChannel';
 export const CHANNEL_ADD_PEOPLE = 'ChannelAddPeople';
 export const CHANNEL_DETAILS = 'ChannelDetails';
 export const CHANNEL_EDIT = 'ChannelEdit';
 export const CREATE_DIRECT_MESSAGE = 'CreateDirectMessage';
-export const CUSTOM_STATUS_CLEAR_AFTER = 'CustomStatusClearAfter';
+export const CREATE_OR_EDIT_CHANNEL = 'CreateOrEditChannel';
 export const CUSTOM_STATUS = 'CustomStatus';
+export const CUSTOM_STATUS_CLEAR_AFTER = 'CustomStatusClearAfter';
 export const EDIT_POST = 'EditPost';
 export const EDIT_PROFILE = 'EditProfile';
 export const EDIT_SERVER = 'EditServer';
@@ -26,6 +26,7 @@ export const IN_APP_NOTIFICATION = 'InAppNotification';
 export const LOGIN = 'Login';
 export const MENTIONS = 'Mentions';
 export const MFA = 'MFA';
+export const PARTICIPANTS_LIST = 'ParticipantsList';
 export const PERMALINK = 'Permalink';
 export const POST_OPTIONS = 'PostOptions';
 export const REACTIONS = 'Reactions';
@@ -35,6 +36,7 @@ export const SERVER = 'Server';
 export const SETTINGS_SIDEBAR = 'SettingsSidebar';
 export const SSO = 'SSO';
 export const THREAD = 'Thread';
+export const THREAD_FOLLOW_BUTTON = 'ThreadFollowButton';
 export const USER_PROFILE = 'UserProfile';
 
 export default {
@@ -63,6 +65,7 @@ export default {
     LOGIN,
     MENTIONS,
     MFA,
+    PARTICIPANTS_LIST,
     PERMALINK,
     POST_OPTIONS,
     REACTIONS,
@@ -72,6 +75,7 @@ export default {
     SETTINGS_SIDEBAR,
     SSO,
     THREAD,
+    THREAD_FOLLOW_BUTTON,
     USER_PROFILE,
 };
 

app/database/models/server/thread.ts

@@ -52,6 +52,9 @@ export default class ThreadModel extends Model implements ThreadModelInterface {
     /** unread_mentions : The number of mentions that have not been read by the user. */
     @field('unread_mentions') unreadMentions!: number;
 
+    /** viewed_at : The timestamp showing when the user's last opened this thread (this is used for the new line message indicator) */
+    @field('viewed_at') viewedAt!: number;
+
     /** participants : All the participants associated with this Thread */
     @children(THREAD_PARTICIPANT) participants!: Query<ThreadParticipantModel>;
 

app/database/operator/server_data_operator/handlers/thread.ts

@@ -24,7 +24,7 @@ const {
 } = Database.MM_TABLES.SERVER;
 
 export interface ThreadHandlerMix {
-    handleThreads: ({threads, prepareRecordsOnly}: HandleThreadsArgs) => Promise<Model[]>;
+    handleThreads: ({threads, teamId, prepareRecordsOnly}: HandleThreadsArgs) => Promise<Model[]>;
     handleThreadParticipants: ({threadsParticipants, prepareRecordsOnly}: HandleThreadParticipantsArgs) => Promise<ThreadParticipantModel[]>;
 }
 
@@ -79,11 +79,13 @@ const ThreadHandler = (superclass: any) => class extends superclass {
         const threadParticipants = (await this.handleThreadParticipants({threadsParticipants, prepareRecordsOnly: true})) as ThreadParticipantModel[];
         batch.push(...threadParticipants);
 
-        const threadsInTeam = await this.handleThreadInTeam({
+        if (teamId) {
-            threadsMap: {[teamId]: threads},
+            const threadsInTeam = await this.handleThreadInTeam({
-            prepareRecordsOnly: true,
+                threadsMap: {[teamId]: threads},
-        }) as ThreadInTeamModel[];
+                prepareRecordsOnly: true,
-        batch.push(...threadsInTeam);
+            }) as ThreadInTeamModel[];
+            batch.push(...threadsInTeam);
+        }
 
         if (batch.length && !prepareRecordsOnly) {
             await this.batchRecords(batch);
@@ -97,10 +99,11 @@ const ThreadHandler = (superclass: any) => class extends superclass {
      * @param {HandleThreadParticipantsArgs} handleThreadParticipants
      * @param {ParticipantsPerThread[]} handleThreadParticipants.threadsParticipants
      * @param {boolean} handleThreadParticipants.prepareRecordsOnly
+     * @param {boolean} handleThreadParticipants.skipSync
      * @throws DataOperatorException
      * @returns {Promise<Array<ThreadParticipantModel>>}
      */
-    handleThreadParticipants = async ({threadsParticipants, prepareRecordsOnly}: HandleThreadParticipantsArgs): Promise<ThreadParticipantModel[]> => {
+    handleThreadParticipants = async ({threadsParticipants, prepareRecordsOnly, skipSync = false}: HandleThreadParticipantsArgs): Promise<ThreadParticipantModel[]> => {
         const batchRecords: ThreadParticipantModel[] = [];
 
         if (!threadsParticipants.length) {
@@ -119,6 +122,7 @@ const ThreadHandler = (superclass: any) => class extends superclass {
                 database: this.database,
                 thread_id,
                 rawParticipants: rawValues,
+                skipSync,
             });
 
             if (createParticipants?.length) {

app/database/operator/server_data_operator/transformers/thread.ts

@@ -32,11 +32,12 @@ export const transformThreadRecord = ({action, database, value}: TransformerArgs
     const fieldsMapper = (thread: ThreadModel) => {
         thread._raw.id = isCreateAction ? (raw?.id ?? thread.id) : record.id;
         thread.lastReplyAt = raw.last_reply_at;
-        thread.lastViewedAt = raw.last_viewed_at;
+        thread.lastViewedAt = raw.last_viewed_at ?? record?.lastViewedAt ?? 0;
         thread.replyCount = raw.reply_count;
         thread.isFollowing = raw.is_following ?? record?.isFollowing;
-        thread.unreadReplies = raw.unread_replies;
+        thread.unreadReplies = raw.unread_replies ?? record?.lastViewedAt ?? 0;
-        thread.unreadMentions = raw.unread_mentions;
+        thread.unreadMentions = raw.unread_mentions ?? record?.lastViewedAt ?? 0;
+        thread.viewedAt = record?.viewedAt || 0;
     };
 
     return prepareBaseRecord({

app/database/operator/utils/thread.ts

@@ -4,6 +4,7 @@ import {Q} from '@nozbe/watermelondb';
 
 import {MM_TABLES} from '@constants/database';
 
+import type {Clause} from '@nozbe/watermelondb/QueryDescription';
 import type {RecordPair, SanitizeThreadParticipantsArgs} from '@typings/database/database';
 import type ThreadParticipantModel from '@typings/database/models/servers/thread_participant';
 
@@ -18,10 +19,20 @@ const {THREAD_PARTICIPANT} = MM_TABLES.SERVER;
  * @param {UserProfile[]} sanitizeThreadParticipants.rawParticipants
  * @returns {Promise<{createParticipants: ThreadParticipant[],  deleteParticipants: ThreadParticipantModel[]}>}
  */
-export const sanitizeThreadParticipants = async ({database, thread_id, rawParticipants}: SanitizeThreadParticipantsArgs) => {
+export const sanitizeThreadParticipants = async ({database, skipSync, thread_id, rawParticipants}: SanitizeThreadParticipantsArgs) => {
+    const clauses: Clause[] = [Q.where('thread_id', thread_id)];
+
+    // Check if we already have the participants
+    if (skipSync) {
+        clauses.push(
+            Q.where('user_id', Q.oneOf(
+                rawParticipants.map((participant) => participant.id),
+            )),
+        );
+    }
     const participants = (await database.collections.
         get(THREAD_PARTICIPANT).
-        query(Q.where('thread_id', thread_id)).
+        query(...clauses).
         fetch()) as ThreadParticipantModel[];
 
     // similarObjects: Contains objects that are in both the RawParticipant array and in the ThreadParticipant table
@@ -42,6 +53,10 @@ export const sanitizeThreadParticipants = async ({database, thread_id, rawPartic
         }
     }
 
+    if (skipSync) {
+        return {createParticipants, deleteParticipants: []};
+    }
+
     // finding out elements to delete using array subtract
     const deleteParticipants = participants.
         filter((participant) => !similarObjects.includes(participant)).

app/database/schema/server/table_schemas/thread.ts

@@ -16,5 +16,6 @@ export default tableSchema({
         {name: 'reply_count', type: 'number'},
         {name: 'unread_replies', type: 'number'},
         {name: 'unread_mentions', type: 'number'},
+        {name: 'viewed_at', type: 'number'},
     ],
 });

app/database/schema/server/test.ts

@@ -436,6 +436,7 @@ describe('*** Test schema for SERVER database ***', () => {
                         reply_count: {name: 'reply_count', type: 'number'},
                         unread_replies: {name: 'unread_replies', type: 'number'},
                         unread_mentions: {name: 'unread_mentions', type: 'number'},
+                        viewed_at: {name: 'viewed_at', type: 'number'},
                     },
                     columnArray: [
                         {name: 'last_reply_at', type: 'number'},
@@ -444,6 +445,7 @@ describe('*** Test schema for SERVER database ***', () => {
                         {name: 'reply_count', type: 'number'},
                         {name: 'unread_replies', type: 'number'},
                         {name: 'unread_mentions', type: 'number'},
+                        {name: 'viewed_at', type: 'number'},
                     ],
                 },
                 [THREAD_PARTICIPANT]: {

app/queries/servers/post.ts

@@ -142,7 +142,7 @@ export const queryPostsBetween = (database: Database, earliest: number, latest:
         andClauses.push(Q.where('user_id', userId));
     }
 
-    if (rootId) {
+    if (rootId != null) {
         andClauses.push(Q.where('root_id', rootId));
     }
 

app/queries/servers/thread.ts

@@ -0,0 +1,129 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+import {Database, Q, Query} from '@nozbe/watermelondb';
+import {combineLatest, of as of$} from 'rxjs';
+import {map, switchMap} from 'rxjs/operators';
+
+import {Preferences} from '@constants';
+import {MM_TABLES} from '@constants/database';
+import {isCRTEnabled} from '@utils/thread';
+
+import {queryPreferencesByCategoryAndName} from './preference';
+import {getConfig, observeConfig} from './system';
+
+import type ServerDataOperator from '@database/operator/server_data_operator';
+import type Model from '@nozbe/watermelondb/Model';
+import type ThreadModel from '@typings/database/models/servers/thread';
+
+const {SERVER: {CHANNEL, POST, THREAD}} = MM_TABLES;
+
+export const getIsCRTEnabled = async (database: Database): Promise<boolean> => {
+    const config = await getConfig(database);
+    const preferences = await queryPreferencesByCategoryAndName(database, Preferences.CATEGORY_DISPLAY_SETTINGS).fetch();
+    return isCRTEnabled(preferences, config);
+};
+
+export const getThreadById = async (database: Database, threadId: string) => {
+    try {
+        const thread = await database.get<ThreadModel>(THREAD).find(threadId);
+        return thread;
+    } catch {
+        return undefined;
+    }
+};
+
+export const observeIsCRTEnabled = (database: Database) => {
+    getConfig(database);
+    const config = observeConfig(database);
+    const preferences = queryPreferencesByCategoryAndName(database, Preferences.CATEGORY_DISPLAY_SETTINGS).observe();
+    return combineLatest([config, preferences]).pipe(
+        map(
+            ([cfg, prefs]) => isCRTEnabled(prefs, cfg),
+        ),
+    );
+};
+
+export const observeThreadById = (database: Database, threadId: string) => {
+    return database.get<ThreadModel>(THREAD).query(
+        Q.where('id', threadId),
+    ).observe().pipe(
+        switchMap((threads) => threads[0]?.observe() || of$(undefined)),
+    );
+};
+
+export const observeUnreadsAndMentionsInTeam = (database: Database, teamId: string) => {
+    return queryThreadsInTeam(database, teamId, true).observeWithColumns(['unread_replies', 'unread_mentions']).pipe(
+        switchMap((threads) => {
+            let unreads = 0;
+            let mentions = 0;
+            threads.forEach((thread) => {
+                unreads += thread.unreadReplies;
+                mentions += thread.unreadMentions;
+            });
+            return of$({unreads, mentions});
+        }),
+    );
+};
+
+// On receiving "posts", Save the "root posts" as "threads"
+export const prepareThreadsFromReceivedPosts = async (operator: ServerDataOperator, posts: Post[]) => {
+    const models: Model[] = [];
+    const threads: Thread[] = [];
+    posts.forEach((post: Post) => {
+        if (!post.root_id && post.type === '') {
+            threads.push({
+                id: post.id,
+                participants: post.participants,
+                reply_count: post.reply_count,
+                last_reply_at: post.last_reply_at,
+                is_following: post.is_following,
+            } as Thread);
+        }
+    });
+    if (threads.length) {
+        const threadModels = await operator.handleThreads({threads, prepareRecordsOnly: true});
+        if (threadModels.length) {
+            models.push(...threadModels);
+        }
+    }
+
+    return models;
+};
+
+export const queryThreadsInTeam = (database: Database, teamId: string, onlyUnreads?: boolean, hasReplies?: boolean, isFollowing?: boolean, sort?: boolean): Query<ThreadModel> => {
+    const query: Q.Clause[] = [
+        Q.experimentalNestedJoin(POST, CHANNEL),
+    ];
+
+    if (isFollowing) {
+        query.push(Q.where('is_following', true));
+    }
+
+    if (hasReplies) {
+        query.push(Q.where('reply_count', Q.gt(0)));
+    }
+
+    if (onlyUnreads) {
+        query.push(Q.where('unread_replies', Q.gt(0)));
+    }
+
+    if (sort) {
+        query.push(Q.sortBy('last_reply_at', Q.desc));
+    }
+
+    query.push(
+        Q.on(
+            POST,
+            Q.on(
+                CHANNEL,
+                Q.or(
+                    Q.where('team_id', teamId),
+                    Q.where('team_id', ''),
+                ),
+            ),
+        ),
+    );
+
+    return database.get<ThreadModel>(THREAD).query(...query);
+};

app/screens/channel/channel_post_list/index.ts

@@ -6,7 +6,7 @@ import {withDatabase} from '@nozbe/watermelondb/DatabaseProvider';
 import withObservables from '@nozbe/with-observables';
 import React from 'react';
 import {AppStateStatus} from 'react-native';
-import {of as of$} from 'rxjs';
+import {combineLatest, of as of$} from 'rxjs';
 import {switchMap} from 'rxjs/operators';
 
 import {Preferences} from '@constants';
@@ -15,6 +15,7 @@ import {observeMyChannel} from '@queries/servers/channel';
 import {queryPostsBetween, queryPostsInChannel} from '@queries/servers/post';
 import {queryPreferencesByCategoryAndName} from '@queries/servers/preference';
 import {observeConfigBooleanValue} from '@queries/servers/system';
+import {observeIsCRTEnabled} from '@queries/servers/thread';
 import {observeCurrentUser} from '@queries/servers/user';
 import {getTimezone} from '@utils/user';
 
@@ -25,21 +26,25 @@ import type {WithDatabaseArgs} from '@typings/database/database';
 const enhanced = withObservables(['channelId', 'forceQueryAfterAppState'], ({database, channelId}: {channelId: string; forceQueryAfterAppState: AppStateStatus} & WithDatabaseArgs) => {
     const currentUser = observeCurrentUser(database);
 
+    const isCRTEnabledObserver = observeIsCRTEnabled(database);
+    const postsInChannelObserver = queryPostsInChannel(database, channelId).observeWithColumns(['earliest', 'latest']);
+
     return {
         currentTimezone: currentUser.pipe((switchMap((user) => of$(getTimezone(user?.timezone || null))))),
         currentUsername: currentUser.pipe((switchMap((user) => of$(user?.username)))),
+        isCRTEnabled: isCRTEnabledObserver,
         isTimezoneEnabled: observeConfigBooleanValue(database, 'ExperimentalTimezone'),
         lastViewedAt: observeMyChannel(database, channelId).pipe(
             switchMap((myChannel) => of$(myChannel?.viewedAt)),
         ),
-        posts: queryPostsInChannel(database, channelId).observeWithColumns(['earliest', 'latest']).pipe(
+        posts: combineLatest([isCRTEnabledObserver, postsInChannelObserver]).pipe(
-            switchMap((postsInChannel) => {
+            switchMap(([isCRTEnabled, postsInChannel]) => {
                 if (!postsInChannel.length) {
                     return of$([]);
                 }
 
                 const {earliest, latest} = postsInChannel[0];
-                return queryPostsBetween(database, earliest, latest, Q.desc, '', channelId).observe();
+                return queryPostsBetween(database, earliest, latest, Q.desc, '', channelId, isCRTEnabled ? '' : undefined).observe();
             }),
         ),
         shouldShowJoinLeaveMessages: queryPreferencesByCategoryAndName(database, Preferences.CATEGORY_ADVANCED_SETTINGS, Preferences.ADVANCED_FILTER_JOIN_LEAVE).observe().pipe(

app/screens/index.tsx

@@ -145,6 +145,11 @@ Navigation.setLazyComponentRegistrator((screenName) => {
         case Screens.THREAD:
             screen = withServerDatabase(require('@screens/thread').default);
             break;
+        case Screens.THREAD_FOLLOW_BUTTON:
+            Navigation.registerComponent(Screens.THREAD_FOLLOW_BUTTON, () => withServerDatabase(
+                require('@screens/thread/thread_follow_button').default,
+            ));
+            break;
     }
 
     if (screen) {

app/screens/thread/thread.tsx

@@ -42,7 +42,6 @@ const Thread = ({rootPost}: ThreadProps) => {
                 <>
                     <View style={styles.flex}>
                         <ThreadPostList
-                            channelId={rootPost!.channelId}
                             forceQueryAfterAppState={appState}
                             nativeID={rootPost!.id}
                             rootPost={rootPost!}

app/screens/thread/thread_follow_button/index.ts

@@ -0,0 +1,23 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+import {withDatabase} from '@nozbe/watermelondb/DatabaseProvider';
+import withObservables from '@nozbe/with-observables';
+import {of as of$} from 'rxjs';
+import {switchMap} from 'rxjs/operators';
+
+import {observeThreadById} from '@queries/servers/thread';
+
+import ThreadFollowButton from './thread_follow_button';
+
+import type {WithDatabaseArgs} from '@typings/database/database';
+
+const enhanced = withObservables(['threadId'], ({threadId, database}: {threadId: string} & WithDatabaseArgs) => {
+    return {
+        isFollowing: observeThreadById(database, threadId).pipe(
+            switchMap((thread) => of$(thread?.isFollowing)),
+        ),
+    };
+});
+
+export default withDatabase(enhanced(ThreadFollowButton));

app/screens/thread/thread_follow_button/thread_follow_button.tsx

@@ -0,0 +1,86 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+import React from 'react';
+import {Platform, StyleSheet, TouchableOpacity, View} from 'react-native';
+
+import {updateThreadFollowing} from '@actions/remote/thread';
+import FormattedText from '@components/formatted_text';
+import {useServerUrl} from '@context/server';
+import {useTheme} from '@context/theme';
+import {preventDoubleTap} from '@utils/tap';
+import {changeOpacity, makeStyleSheetFromTheme} from '@utils/theme';
+import {typography} from '@utils/typography';
+
+type Props = {
+    isFollowing: boolean;
+    teamId: string;
+    threadId: string;
+};
+
+const getStyleSheet = makeStyleSheetFromTheme((theme: Theme) => {
+    return {
+        container: {
+            borderColor: theme.sidebarHeaderTextColor,
+            borderWidth: StyleSheet.hairlineWidth,
+            borderRadius: 4,
+            paddingVertical: 4.5,
+            paddingHorizontal: 10,
+            opacity: 0.72,
+            ...Platform.select({
+                android: {
+                    marginRight: 12,
+                },
+                ios: {
+                    right: -4,
+                },
+            }),
+        },
+        containerActive: {
+            backgroundColor: changeOpacity(theme.sidebarHeaderTextColor, 0.24),
+            borderColor: 'transparent',
+            opacity: 1,
+        },
+        text: {
+            color: theme.sidebarHeaderTextColor,
+            ...typography('Heading', 75, 'SemiBold'),
+        },
+    };
+});
+
+function ThreadFollow({isFollowing, teamId, threadId}: Props) {
+    const theme = useTheme();
+    const styles = getStyleSheet(theme);
+
+    const serverUrl = useServerUrl();
+
+    const onPress = preventDoubleTap(() => {
+        updateThreadFollowing(serverUrl, teamId, threadId, !isFollowing);
+    });
+
+    const containerStyle = [styles.container];
+    let followTextProps = {
+        id: 'threads.follow',
+        defaultMessage: 'Follow',
+    };
+    if (isFollowing) {
+        containerStyle.push(styles.containerActive);
+        followTextProps = {
+            id: 'threads.following',
+            defaultMessage: 'Following',
+        };
+    }
+
+    return (
+        <TouchableOpacity onPress={onPress}>
+            <View style={containerStyle}>
+                <FormattedText
+                    {...followTextProps}
+                    style={styles.text}
+                />
+            </View>
+        </TouchableOpacity>
+    );
+}
+
+export default ThreadFollow;

app/screens/thread/thread_post_list/index.ts

@@ -10,6 +10,7 @@ import {switchMap} from 'rxjs/operators';
 import {observeMyChannel} from '@queries/servers/channel';
 import {queryPostsChunk, queryPostsInThread} from '@queries/servers/post';
 import {observeConfigBooleanValue} from '@queries/servers/system';
+import {observeIsCRTEnabled} from '@queries/servers/thread';
 import {observeCurrentUser} from '@queries/servers/user';
 import {getTimezone} from '@utils/user';
 
@@ -19,19 +20,19 @@ import type {WithDatabaseArgs} from '@typings/database/database';
 import type PostModel from '@typings/database/models/servers/post';
 
 type Props = WithDatabaseArgs & {
-    channelId: string;
     forceQueryAfterAppState: AppStateStatus;
     rootPost: PostModel;
 };
 
-const enhanced = withObservables(['channelId', 'forceQueryAfterAppState', 'rootPost'], ({channelId, database, rootPost}: Props) => {
+const enhanced = withObservables(['forceQueryAfterAppState', 'rootPost'], ({database, rootPost}: Props) => {
     const currentUser = observeCurrentUser(database);
 
     return {
         currentTimezone: currentUser.pipe((switchMap((user) => of$(getTimezone(user?.timezone || null))))),
         currentUsername: currentUser.pipe((switchMap((user) => of$(user?.username || '')))),
+        isCRTEnabled: observeIsCRTEnabled(database),
         isTimezoneEnabled: observeConfigBooleanValue(database, 'ExperimentalTimezone'),
-        lastViewedAt: observeMyChannel(database, channelId).pipe(
+        lastViewedAt: observeMyChannel(database, rootPost.channelId).pipe(
             switchMap((myChannel) => of$(myChannel?.viewedAt)),
         ),
         posts: queryPostsInThread(database, rootPost.id, true, true).observeWithColumns(['earliest', 'latest']).pipe(
@@ -44,6 +45,9 @@ const enhanced = withObservables(['channelId', 'forceQueryAfterAppState', 'rootP
                 return queryPostsChunk(database, rootPost.id, earliest, latest, true).observe();
             }),
         ),
+        teamId: rootPost.channel.observe().pipe(
+            switchMap((channel) => of$(channel?.teamId)),
+        ),
     };
 });
 

app/screens/thread/thread_post_list/thread_post_list.tsx

@@ -1,25 +1,28 @@
 // Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
 // See LICENSE.txt for license information.
 
-import React, {useMemo} from 'react';
+import React, {useEffect, useMemo, useRef} from 'react';
 import {StyleSheet, View} from 'react-native';
 import {Edge, SafeAreaView} from 'react-native-safe-area-context';
 
+import {updateThreadRead} from '@actions/remote/thread';
 import PostList from '@components/post_list';
 import {Screens} from '@constants';
+import {useServerUrl} from '@context/server';
 import {useIsTablet} from '@hooks/device';
 
 import type PostModel from '@typings/database/models/servers/post';
 
 type Props = {
-    channelId: string;
     currentTimezone: string | null;
     currentUsername: string;
+    isCRTEnabled: boolean;
     isTimezoneEnabled: boolean;
     lastViewedAt: number;
     nativeID: string;
     posts: PostModel[];
     rootPost: PostModel;
+    teamId: string;
 }
 
 const edges: Edge[] = ['bottom'];
@@ -31,18 +34,28 @@ const styles = StyleSheet.create({
 });
 
 const ThreadPostList = ({
-    channelId, currentTimezone, currentUsername,
+    currentTimezone, currentUsername,
-    isTimezoneEnabled, lastViewedAt, nativeID, posts, rootPost,
+    isCRTEnabled, isTimezoneEnabled, lastViewedAt, nativeID, posts, rootPost, teamId,
 }: Props) => {
     const isTablet = useIsTablet();
+    const serverUrl = useServerUrl();
 
     const threadPosts = useMemo(() => {
         return [...posts, rootPost];
     }, [posts, rootPost]);
 
+    // If CRT is enabled, When new post arrives and thread modal is open, mark thread as read
+    const oldPostsCount = useRef<number>(posts.length);
+    useEffect(() => {
+        if (isCRTEnabled && oldPostsCount.current < posts.length) {
+            oldPostsCount.current = posts.length;
+            updateThreadRead(serverUrl, teamId, rootPost.id, Date.now());
+        }
+    }, [isCRTEnabled, posts, rootPost, serverUrl, teamId]);
+
     const postList = (
         <PostList
-            channelId={channelId}
+            channelId={rootPost.channelId}
             contentContainerStyle={styles.container}
             currentTimezone={currentTimezone}
             currentUsername={currentUsername}

app/utils/post/index.ts

@@ -71,3 +71,15 @@ export const sortPostsByNewest = (posts: PostModel[]) => {
         return -1;
     });
 };
+
+export const processPostsFetched = (data: PostResponse) => {
+    const order = data.order;
+    const posts = Object.values(data.posts) as Post[];
+    const previousPostId = data.prev_post_id;
+
+    return {
+        posts,
+        order,
+        previousPostId,
+    };
+};

app/utils/thread/index.ts

@@ -0,0 +1,26 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+import {Config, Preferences} from '@constants';
+import {getPreferenceValue} from '@helpers/api/preference';
+
+import type PreferenceModel from '@typings/database/models/servers/preference';
+
+export function isCRTEnabled(preferences: PreferenceModel[], config?: ClientConfig): boolean {
+    let preferenceDefault = Preferences.COLLAPSED_REPLY_THREADS_OFF;
+    const configValue = config?.CollapsedThreads;
+    if (configValue === Config.DEFAULT_ON) {
+        preferenceDefault = Preferences.COLLAPSED_REPLY_THREADS_ON;
+    }
+    const preference = getPreferenceValue(preferences, Preferences.CATEGORY_DISPLAY_SETTINGS, Preferences.COLLAPSED_REPLY_THREADS, preferenceDefault);
+
+    const isAllowed = (
+        config?.FeatureFlagCollapsedThreads === Config.TRUE &&
+        config?.CollapsedThreads !== Config.DISABLED
+    );
+
+    return isAllowed && (
+        preference === Preferences.COLLAPSED_REPLY_THREADS_ON ||
+        config?.CollapsedThreads === Config.ALWAYS_ON
+    );
+}

assets/base/i18n/en.json

@@ -485,6 +485,8 @@
   "thread.header.thread_in": "in {channelName}",
   "thread.noReplies": "No replies yet",
   "thread.repliesCount": "{repliesCount, number} {repliesCount, plural, one {reply} other {replies}}",
+  "threads.follow": "Follow",
+  "threads.following": "Following",
   "threads.followMessage": "Follow Message",
   "threads.followThread": "Follow Thread",
   "threads.unfollowMessage": "Unfollow Message",

types/api/channels.d.ts

@@ -55,7 +55,9 @@ type ChannelMembership = {
     roles: string;
     last_viewed_at: number;
     msg_count: number;
+    msg_count_root?: number;
     mention_count: number;
+    mention_count_root?: number;
     notify_props: Partial<ChannelNotifyProps>;
     last_post_at?: number;
     last_update_at: number;

types/api/threads.d.ts

@@ -18,3 +18,10 @@ type ThreadParticipant = {
     id: $ID<User>;
     thread_id: $ID<Thread>;
 };
+
+type GetUserThreadsResponse = {
+    threads: Thread[];
+    total: number;
+    total_unread_mentions: number;
+    total_unread_threads: number;
+};

types/api/websocket.d.ts

@@ -14,3 +14,10 @@ type WebSocketMessage = {
     broadcast: WebsocketBroadcast;
     seq: number;
 }
+
+type ThreadReadChangedData = {
+    thread_id: string;
+    timestamp: number;
+    unread_mentions: number;
+    unread_replies: number;
+};

types/database/database.d.ts

@@ -91,11 +91,12 @@ export type HandlePostsArgs = {
 export type HandleThreadsArgs = {
   threads: Thread[];
   prepareRecordsOnly?: boolean;
-  teamId: string;
+  teamId?: string;
 };
 
 export type HandleThreadParticipantsArgs = {
   prepareRecordsOnly: boolean;
+  skipSync?: boolean;
   threadsParticipants: ParticipantsPerThread[];
 };
 
@@ -113,6 +114,7 @@ export type SanitizeReactionsArgs = {
 
 export type SanitizeThreadParticipantsArgs = {
   database: Database;
+  skipSync?: boolean;
   thread_id: $ID<Thread>;
   rawParticipants: ThreadParticipant[];
 }

types/database/models/servers/thread.d.ts

@@ -35,6 +35,9 @@ export default class ThreadModel extends Model {
     /** unread_mentions : The number of mentions that are not read by the user. */
     unreadMentions: number;
 
+    /** viewed_at : The timestamp showing when the user's last opened this thread (this is used for the new line message indicator) */
+    viewedAt: number;
+
     /** participants: All the participants of the thread */
     participants: Query<ThreadParticipantsModel>;