feat(MM-64410): batch write all posts for all channels by team (#8892)

* feat(MM-64410): batch write all posts for all channels on team-by-team basis
* skipAuthor=false, so the authors will be fetched after post fetching
* renaming a file
* revert unplanned changes
* missing end line
* fix broken test
* Review comment. Using 20 posts

app/actions/local/post.ts

@@ -15,7 +15,7 @@ import {getPostIdsForCombinedUserActivityPost} from '@utils/post_list';
 
 import {updateLastPostAt, updateMyChannelLastFetchedAt} from './channel';
 
-import type {Q} from '@nozbe/watermelondb';
+import type {Model, Q} from '@nozbe/watermelondb';
 import type MyChannelModel from '@typings/database/models/servers/my_channel';
 import type PostModel from '@typings/database/models/servers/post';
 import type UserModel from '@typings/database/models/servers/user';
@@ -173,6 +173,81 @@ export async function markPostAsDeleted(serverUrl: string, post: Post, prepareRe
     }
 }
 
+/**
+ * Prepare the models related to the posts of a channel.
+ * @param {string} serverUrl the server URL for this channel
+ * @param {string} channelId the channel id
+ * @param {Post[]} posts the posts in the channel
+ * @param {string} actionType the action type for posts if it's `NEW`, `RECEIVED_IN_CHANNEL`, `RECEIVED_IN_THREAD`, etc.
+ * @param {string[]} order the order of the posts
+ * @param {string} previousPostId the previous post id
+ * @param {UserProfile[]} authors the authors of the posts
+ * @param {boolean} isCRTEnabled whether CRT is enabled for this channel
+ * @returns {Promise<Model[]>} the models
+ */
+export async function prepareModelsForChannelPosts(
+    serverUrl: string,
+    channelId: string,
+    posts: Post[],
+    actionType: string,
+    order: string[],
+    previousPostId: string,
+    authors: UserProfile[],
+    isCRTEnabled: boolean,
+): Promise<Model[]> {
+    const {operator} = DatabaseManager.getServerDatabaseAndOperator(serverUrl);
+    const models = [];
+
+    const postModels = await operator.handlePosts({
+        actionType,
+        order,
+        posts,
+        previousPostId,
+        prepareRecordsOnly: true,
+    });
+    models.push(...postModels);
+
+    if (authors.length) {
+        const userModels = await operator.handleUsers({users: authors, prepareRecordsOnly: true});
+        models.push(...userModels);
+    }
+
+    const lastFetchedAt = getLastFetchedAtFromPosts(posts);
+    let myChannelModel: MyChannelModel | undefined;
+    if (lastFetchedAt) {
+        const {member} = await updateMyChannelLastFetchedAt(serverUrl, channelId, lastFetchedAt, true);
+        myChannelModel = member;
+    }
+
+    let lastPostAt = 0;
+    for (const post of posts) {
+        const isCrtReply = isCRTEnabled && post.root_id !== '';
+        if (!isCrtReply) {
+            lastPostAt = post.create_at > lastPostAt ? post.create_at : lastPostAt;
+        }
+    }
+
+    if (lastPostAt) {
+        const {member} = await updateLastPostAt(serverUrl, channelId, lastPostAt, true);
+        if (member) {
+            myChannelModel = member;
+        }
+    }
+
+    if (myChannelModel) {
+        models.push(myChannelModel);
+    }
+
+    if (isCRTEnabled) {
+        const threadModels = await prepareThreadsFromReceivedPosts(operator, posts, false);
+        if (threadModels?.length) {
+            models.push(...threadModels);
+        }
+    }
+
+    return models;
+}
+
 export async function storePostsForChannel(
     serverUrl: string, channelId: string, posts: Post[], order: string[], previousPostId: string,
     actionType: string, authors: UserProfile[], prepareRecordsOnly = false,
@@ -182,53 +257,16 @@ export async function storePostsForChannel(
 
         const isCRTEnabled = await getIsCRTEnabled(database);
 
-        const models = [];
-        const postModels = await operator.handlePosts({
+        const models = await prepareModelsForChannelPosts(
+            serverUrl,
+            channelId,
+            posts,
             actionType,
             order,
-            posts,
             previousPostId,
-            prepareRecordsOnly: true,
-        });
-        models.push(...postModels);
-
-        if (authors.length) {
-            const userModels = await operator.handleUsers({users: authors, prepareRecordsOnly: true});
-            models.push(...userModels);
-        }
-
-        const lastFetchedAt = getLastFetchedAtFromPosts(posts);
-        let myChannelModel: MyChannelModel | undefined;
-        if (lastFetchedAt) {
-            const {member} = await updateMyChannelLastFetchedAt(serverUrl, channelId, lastFetchedAt, true);
-            myChannelModel = member;
-        }
-
-        let lastPostAt = 0;
-        for (const post of posts) {
-            const isCrtReply = isCRTEnabled && post.root_id !== '';
-            if (!isCrtReply) {
-                lastPostAt = post.create_at > lastPostAt ? post.create_at : lastPostAt;
-            }
-        }
-
-        if (lastPostAt) {
-            const {member} = await updateLastPostAt(serverUrl, channelId, lastPostAt, true);
-            if (member) {
-                myChannelModel = member;
-            }
-        }
-
-        if (myChannelModel) {
-            models.push(myChannelModel);
-        }
-
-        if (isCRTEnabled) {
-            const threadModels = await prepareThreadsFromReceivedPosts(operator, posts, false);
-            if (threadModels?.length) {
-                models.push(...threadModels);
-            }
-        }
+            authors,
+            isCRTEnabled,
+        );
 
         if (models.length && !prepareRecordsOnly) {
             await operator.batchRecords(models, 'storePostsForChannel');

app/actions/remote/entry/common.ts

@@ -295,7 +295,7 @@ export async function restDeferredAppEntryActions(
                 if (isCRTEnabled && initialTeamId) {
                     await syncTeamThreads(serverUrl, initialTeamId, {groupLabel: requestLabel});
                 }
-                fetchPostsForUnreadChannels(serverUrl, mySortedTeams, chData.channels, chData.memberships, initialChannelId, isCRTEnabled, false, requestLabel);
+                fetchPostsForUnreadChannels(serverUrl, mySortedTeams, chData.channels, chData.memberships, initialChannelId, isCRTEnabled, requestLabel);
             }
 
             if (myOtherSortedTeams.length) {

app/actions/remote/post.auxiliary.test.ts

@@ -0,0 +1,106 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+import DatabaseManager from '@database/manager';
+import {queryPostsById, queryPostsChunk} from '@queries/servers/post';
+import TestHelper from '@test/test_helper';
+
+import {fetchPostAuthors, fetchPostsForChannel} from './post';
+import {processChannelPostsByTeam} from './post.auxiliary';
+
+import type ServerDataOperator from '@database/operator/server_data_operator';
+
+jest.mock('./post');
+
+const user1 = {id: 'userid1', username: 'user1', email: 'user1@mattermost.com', roles: ''} as UserProfile;
+const user2 = {id: 'userid2', username: 'user2', email: 'user2@mattermost.com', roles: ''} as UserProfile;
+
+const post1 = TestHelper.fakePost({channel_id: 'channelid1', id: 'postid1', user_id: user1.id});
+const post2 = TestHelper.fakePost({channel_id: 'channelid1', id: 'postid2', user_id: user2.id});
+const post3 = TestHelper.fakePost({channel_id: 'channelid2', id: 'postid3', user_id: user2.id});
+
+describe('post.auxilary', () => {
+    let operator: ServerDataOperator;
+    let spyOnBatchRecords: jest.SpyInstance;
+    const serverUrl = 'baseHandler.test.com';
+    beforeEach(async () => {
+        jest.clearAllMocks();
+
+        await DatabaseManager.init([serverUrl]);
+        operator = DatabaseManager.serverDatabases[serverUrl]!.operator;
+        spyOnBatchRecords = jest.spyOn(operator, 'batchRecords');
+    });
+
+    afterEach(async () => {
+        DatabaseManager.destroyServerDatabase('baseHandler.test.com');
+    });
+
+    it('should batch record once for multiple posts', async () => {
+        (fetchPostsForChannel as jest.Mock).mockResolvedValue({
+            posts: Array.from({length: 20}, (_, index) => TestHelper.fakePost({channel_id: 'channelid1', id: `postid${index}`, user_id: user1.id})),
+        });
+        await processChannelPostsByTeam(serverUrl, ['channelid1']);
+        expect(spyOnBatchRecords).toHaveBeenCalledTimes(1);
+
+        const postsByChannel = await queryPostsChunk(operator.database, 'channelid1', 0, Date.now());
+        expect(postsByChannel.length).toBe(20);
+    });
+
+    it('should fetch post authors', async () => {
+        (fetchPostsForChannel as jest.Mock).mockResolvedValue({
+            posts: [post1, post2],
+        });
+        await processChannelPostsByTeam(serverUrl, ['channelid1']);
+
+        expect(fetchPostAuthors).toHaveBeenCalledWith(serverUrl, [post1, post2], false, undefined);
+    });
+
+    it('should not fetch post authors if skipAuthors is true', async () => {
+        (fetchPostsForChannel as jest.Mock).mockResolvedValue({
+            posts: [post1, post2],
+        });
+        await processChannelPostsByTeam(serverUrl, ['channelid1'], true);
+        expect(fetchPostAuthors).not.toHaveBeenCalled();
+    });
+
+    it('should not batch record if no channels are passed', async () => {
+        await processChannelPostsByTeam(serverUrl, []);
+        expect(spyOnBatchRecords).not.toHaveBeenCalled();
+    });
+
+    it('should not batch record if no posts fetched from server', async () => {
+        (fetchPostsForChannel as jest.Mock).mockResolvedValue({});
+        await processChannelPostsByTeam(serverUrl, ['channelid1']);
+        expect(spyOnBatchRecords).not.toHaveBeenCalled();
+    });
+
+    it('should not fetch authors if no posts returned', async () => {
+        (fetchPostsForChannel as jest.Mock).mockResolvedValue({});
+        await processChannelPostsByTeam(serverUrl, ['channelid1']);
+        expect(fetchPostAuthors).not.toHaveBeenCalled();
+    });
+
+    it('should still batch record even if some posts are returning errors', async () => {
+        (fetchPostsForChannel as jest.Mock).mockRejectedValueOnce(new Error('error'));
+        (fetchPostsForChannel as jest.Mock).mockResolvedValue({
+            posts: [post3],
+        });
+        await processChannelPostsByTeam(serverUrl, ['channelid1', 'channelid2']);
+        expect(spyOnBatchRecords).toHaveBeenCalledWith(
+            expect.arrayContaining([
+                expect.objectContaining({
+                    _raw: expect.objectContaining({
+                        channel_id: 'channelid2',
+                    }),
+                }),
+            ]),
+            'processTeamChannels',
+        );
+
+        const postsChannel1 = await queryPostsById(operator.database, ['postid1']);
+        expect(postsChannel1.length).toBe(0);
+
+        const postsChannel2 = await queryPostsById(operator.database, ['postid3']);
+        expect(postsChannel2.length).toBe(1);
+    });
+});

app/actions/remote/post.auxiliary.ts

@@ -0,0 +1,76 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+import {chunk} from 'lodash';
+
+import {prepareModelsForChannelPosts} from '@actions/local/post';
+import {ActionType} from '@constants';
+import DatabaseManager from '@database/manager';
+
+import {fetchPostAuthors, fetchPostsForChannel} from './post';
+
+import type {Model} from '@nozbe/watermelondb';
+
+/**
+ * processChannelPostsByTeam is to fetch posts for each channel in a team, and prepare the models for the posts.
+ * Then once all the models are prepared, write them into the database all at once.
+ * Writing posts channel by channel is time consuming, and has caused the UI to lock up significantly.
+ * @param serverUrl string the server URL
+ * @param channelIds string[]
+ * @param skipAuthors boolean
+ * @param groupLabel string
+ * @param isCRTEnabled boolean
+ */
+export async function processChannelPostsByTeam(
+    serverUrl: string,
+    channelIds: string[],
+    skipAuthors = false,
+    groupLabel?: RequestGroupLabel,
+    isCRTEnabled = false,
+): Promise<void> {
+    const prepareModelsPromises: Array<Promise<Model[]>> = [];
+    const allPosts: Post[] = [];
+
+    const chunks = chunk(channelIds, 10);
+    for await (const channelIdsChunk of chunks) {
+        const channelPromises = channelIdsChunk.map((channelId) =>
+
+            // hard-coding true for skipAuthors because we want to fetch authors
+            // and upsert later to avoid unique constraint errors
+            fetchPostsForChannel(serverUrl, channelId, true, true, groupLabel),
+        );
+        const chunkResults = await Promise.allSettled(channelPromises);
+
+        for (const [i, result] of chunkResults.entries()) {
+            if (result.status === 'fulfilled') {
+                const {posts, order, previousPostId, authors} = result.value;
+                if (posts?.length) {
+                    const channelId = channelIdsChunk[i];
+                    allPosts.push(...posts);
+                    prepareModelsPromises.push(
+                        prepareModelsForChannelPosts(
+                            serverUrl,
+                            channelId,
+                            posts,
+                            ActionType.POSTS.RECEIVED_IN_CHANNEL,
+                            order || [],
+                            previousPostId || '',
+                            authors || [],
+                            isCRTEnabled,
+                        ),
+                    );
+                }
+            }
+        }
+    }
+
+    if (prepareModelsPromises.length) {
+        const {operator} = DatabaseManager.getServerDatabaseAndOperator(serverUrl);
+
+        const models = await Promise.all(prepareModelsPromises);
+        operator.batchRecords(models.flat(), 'processTeamChannels');
+    }
+    if (!skipAuthors && allPosts.length) {
+        await fetchPostAuthors(serverUrl, allPosts, false, groupLabel);
+    }
+}

app/actions/remote/post.test.ts

@@ -32,6 +32,7 @@ import {
     fetchSavedPosts,
     fetchPinnedPosts,
 } from './post';
+import * as PostAuxilaryFunctions from './post.auxiliary';
 
 import type ServerDataOperator from '@database/operator/server_data_operator';
 
@@ -132,7 +133,6 @@ jest.mock('@actions/local/reactions', () => {
 });
 
 beforeAll(() => {
-    // eslint-disable-next-line
     // @ts-ignore
     NetworkManager.getClient = () => mockClient;
 });
@@ -578,13 +578,13 @@ describe('get posts', () => {
     });
 
     it('fetchPostsForUnreadChannels - base case', async () => {
+        const spyOnProcessChannelPostsByTeam = jest.spyOn(PostAuxilaryFunctions, 'processChannelPostsByTeam');
         await operator.handleSystem({systems: [{id: SYSTEM_IDENTIFIERS.CURRENT_USER_ID, value: user1.id}], prepareRecordsOnly: false});
         await operator.handleMyChannel({channels: [channel1], myChannels: [channelMember1], prepareRecordsOnly: false});
 
-        const result = await fetchPostsForUnreadChannels(serverUrl, [{id: teamId}] as Team[], [channel1, {...channel1, id: 'channelid2', total_msg_count: 10}], [{...channelMember1, msg_count: 5}, {...channelMember1, channel_id: 'channelid2', msg_count: 10}], 'testid');
-        expect(result).toBeDefined();
-        expect(result?.length).toBe(1); // Only returns the response for the channel with unread messages
-        expect(result?.[0].posts?.[0].channel_id).toBe(channel1.id);
+        await fetchPostsForUnreadChannels(serverUrl, [{id: teamId}] as Team[], [channel1, {...channel1, id: 'channelid2', total_msg_count: 10}], [{...channelMember1, msg_count: 5}, {...channelMember1, channel_id: 'channelid2', msg_count: 10}], 'testid');
+
+        expect(spyOnProcessChannelPostsByTeam).toHaveBeenCalledWith(serverUrl, ['channelid1'], false, undefined, undefined);
     });
 
     it('fetchPosts - handle database not found', async () => {

app/actions/remote/post.ts

@@ -4,8 +4,6 @@
 
 /* eslint-disable max-lines */
 
-import {chunk} from 'lodash';
-
 import {markChannelAsUnread, updateLastPostAt} from '@actions/local/channel';
 import {addPostAcknowledgement, removePost, removePostAcknowledgement, storePostsForChannel} from '@actions/local/post';
 import {addRecentReaction} from '@actions/local/reactions';
@@ -29,6 +27,7 @@ import {logDebug, logError} from '@utils/log';
 import {processPostsFetched} from '@utils/post';
 import {getPostIdsForCombinedUserActivityPost} from '@utils/post_list';
 
+import {processChannelPostsByTeam} from './post.auxiliary';
 import {forceLogoutIfNecessary} from './session';
 
 import type {Client} from '@client/rest';
@@ -336,8 +335,8 @@ export async function fetchPostsForChannel(serverUrl: string, channelId: string,
 
 export const fetchPostsForUnreadChannels = async (
     serverUrl: string, teams: Team[], channels: Channel[], memberships: ChannelMembership[],
-    excludeChannelId?: string, isCRTEnabled?: boolean, fetchOnly = false, groupLabel?: RequestGroupLabel,
-): Promise<PostsForChannel[]> => {
+    excludeChannelId?: string, isCRTEnabled?: boolean, groupLabel?: RequestGroupLabel,
+): Promise<void> => {
     const teamIndexMap = new Map<string, number>();
     teams.forEach((team, index) => teamIndexMap.set(team.id, index));
 
@@ -346,7 +345,7 @@ export const fetchPostsForUnreadChannels = async (
         channelsMap.set(channel.id, channel);
     }
 
-    const sortedUnreadchannelIds = memberships.filter((member) => {
+    const sortedUnreadchannelIdsByTeam = memberships.filter((member) => {
         const channel = channelsMap.get(member.channel_id);
         if (channel && !channel.delete_at && channel.id !== excludeChannelId) {
             const unreads = isCRTEnabled ? (channel.total_msg_count_root ?? 0) - (member.msg_count_root ?? 0) : channel.total_msg_count - member.msg_count;
@@ -383,33 +382,21 @@ export const fetchPostsForUnreadChannels = async (
             // If team index is the same, sort by last_viewed_at in descending order
             return b.last_viewed_at - a.last_viewed_at;
         }).
-        map((member) => member.channel_id);
-
-    const postsForChannel: PostsForChannel[] = [];
-    const posts: Post[] = [];
-
-    // process 10 unread channels at a time
-    const chunks = chunk(sortedUnreadchannelIds, 10);
-    for await (const channelIds of chunks) {
-        const promises = [];
-        for (const channelId of channelIds) {
-            promises.push(fetchPostsForChannel(serverUrl, channelId, fetchOnly, true, groupLabel));
-        }
-
-        const results = await Promise.all(promises);
-        postsForChannel.push(...results);
-        for (const r of results) {
-            if (r.posts?.length) {
-                posts.push(...r.posts);
+        reduce((acc, member) => {
+            const channel = channelsMap.get(member.channel_id);
+            const teamId = channel!.team_id || '';
+            if (!acc[teamId]) {
+                acc[teamId] = [];
             }
-        }
-    }
+            acc[teamId].push(channel!.id);
 
-    if (posts.length) {
-        fetchPostAuthors(serverUrl, posts, false, groupLabel);
-    }
+            return acc;
+        }, {} as Record<string, string[]>);
 
-    return postsForChannel;
+    for (const channelIds of Object.values(sortedUnreadchannelIdsByTeam)) {
+        /* eslint-disable-next-line no-await-in-loop */
+        await processChannelPostsByTeam(serverUrl, channelIds, false, groupLabel, isCRTEnabled);
+    }
 };
 
 export async function fetchPosts(serverUrl: string, channelId: string, page = 0, perPage = General.POST_CHUNK_SIZE, fetchOnly = false, groupLabel?: RequestGroupLabel): Promise<PostsRequest> {