From dffb30de471a86ceae81fdd29bf51f03160653f3 Mon Sep 17 00:00:00 2001 From: Rahim Rahman Date: Sun, 1 Jun 2025 07:40:38 -0600 Subject: [PATCH] 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 | 128 ++++++++++++++-------- app/actions/remote/entry/common.ts | 2 +- app/actions/remote/post.auxiliary.test.ts | 106 ++++++++++++++++++ app/actions/remote/post.auxiliary.ts | 76 +++++++++++++ app/actions/remote/post.test.ts | 10 +- app/actions/remote/post.ts | 45 +++----- 6 files changed, 287 insertions(+), 80 deletions(-) create mode 100644 app/actions/remote/post.auxiliary.test.ts create mode 100644 app/actions/remote/post.auxiliary.ts diff --git a/app/actions/local/post.ts b/app/actions/local/post.ts index 5a9b159a7..803be3d9d 100644 --- a/app/actions/local/post.ts +++ b/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} the models + */ +export async function prepareModelsForChannelPosts( + serverUrl: string, + channelId: string, + posts: Post[], + actionType: string, + order: string[], + previousPostId: string, + authors: UserProfile[], + isCRTEnabled: boolean, +): Promise { + 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'); diff --git a/app/actions/remote/entry/common.ts b/app/actions/remote/entry/common.ts index 1fe0ccb79..04012b212 100644 --- a/app/actions/remote/entry/common.ts +++ b/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) { diff --git a/app/actions/remote/post.auxiliary.test.ts b/app/actions/remote/post.auxiliary.test.ts new file mode 100644 index 000000000..7c583d7b9 --- /dev/null +++ b/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); + }); +}); diff --git a/app/actions/remote/post.auxiliary.ts b/app/actions/remote/post.auxiliary.ts new file mode 100644 index 000000000..e4ba33bcb --- /dev/null +++ b/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 { + const prepareModelsPromises: Array> = []; + 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); + } +} diff --git a/app/actions/remote/post.test.ts b/app/actions/remote/post.test.ts index a4ee437d4..74120ac80 100644 --- a/app/actions/remote/post.test.ts +++ b/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 () => { diff --git a/app/actions/remote/post.ts b/app/actions/remote/post.ts index 4ed1be887..94193edd0 100644 --- a/app/actions/remote/post.ts +++ b/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 => { + excludeChannelId?: string, isCRTEnabled?: boolean, groupLabel?: RequestGroupLabel, +): Promise => { const teamIndexMap = new Map(); 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); - 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 {