From 8491872908642c35eef0e5ab5c2fe8a8243e0168 Mon Sep 17 00:00:00 2001 From: Hampus Date: Sun, 30 Aug 2026 22:27:41 +0200 Subject: [PATCH] perf(user): batch mention and saved message reads (#2139) --- .../message/MessageRetrievalService.test.ts | 207 ++++++++++++++++ .../message/MessageRetrievalService.ts | 28 +++ .../services/UserContentRequestService.ts | 16 +- .../user/services/UserContentService.test.ts | 231 +++++++++++++++++- .../api/user/services/UserContentService.ts | 105 +++++--- 5 files changed, 542 insertions(+), 45 deletions(-) create mode 100644 fluxer_api/src/api/channel/services/message/MessageRetrievalService.test.ts diff --git a/fluxer_api/src/api/channel/services/message/MessageRetrievalService.test.ts b/fluxer_api/src/api/channel/services/message/MessageRetrievalService.test.ts new file mode 100644 index 000000000..ce25b91d2 --- /dev/null +++ b/fluxer_api/src/api/channel/services/message/MessageRetrievalService.test.ts @@ -0,0 +1,207 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {MessageTypes} from '@fluxer/constants/src/ChannelConstants'; +import {createSnowflakeFromTimestamp} from '@fluxer/snowflake/src/Snowflake'; +import {describe, expect, it} from 'vitest'; +import type {AttachmentDecayService} from '../../../attachment/AttachmentDecayService'; +import type {ChannelID, MessageID} from '../../../BrandedTypes'; +import {createChannelID, createMessageID, createUserID} from '../../../BrandedTypes'; +import type {UserCacheService} from '../../../infrastructure/UserCacheService'; +import {Message} from '../../../models/Message'; +import type {IUserRepository} from '../../../user/IUserRepository'; +import type {IChannelRepositoryAggregate} from '../../repositories/IChannelRepositoryAggregate'; +import type {AuthenticatedChannel} from '../AuthenticatedChannel'; +import type {MessageChannelAuthService} from './MessageChannelAuthService'; +import type {MessageProcessingService} from './MessageProcessingService'; +import {MessageRetrievalService} from './MessageRetrievalService'; +import type {MessageSearchService} from './MessageSearchService'; + +const CHANNEL_ID = createChannelID(10n); +const VIEWER_ID = createUserID(7n); +const CUTOFF = new Date('2026-01-10T00:00:00.000Z'); + +function makeMessageId(timestamp: Date, sequence: number): MessageID { + return createMessageID(createSnowflakeFromTimestamp(timestamp.getTime()) + BigInt(sequence)); +} + +function makeMessage(channelId: ChannelID, messageId: MessageID): Message { + return new Message({ + channel_id: channelId, + bucket: 0, + message_id: messageId, + author_id: createUserID(3n), + type: MessageTypes.DEFAULT, + webhook_id: null, + webhook_name: null, + webhook_avatar_hash: null, + content: '', + edited_timestamp: null, + pinned_timestamp: null, + flags: 0, + mention_everyone: false, + mention_users: null, + mention_roles: null, + mention_channels: null, + attachments: null, + embeds: null, + sticker_items: null, + message_reference: null, + message_snapshots: null, + call: null, + nsfw_emojis: null, + has_reaction: null, + version: 1, + }); +} + +function createRetrievalService({ + guilded, + canReadHistory, + messageHistoryCutoff = CUTOFF.toISOString(), + stored, +}: { + guilded: boolean; + canReadHistory: boolean; + messageHistoryCutoff?: string | null; + stored: Array; +}) { + const permissionChecks: Array = []; + const authenticationCalls: Array = []; + const storedById = new Map(stored.map((message) => [message.id.toString(), message] as const)); + const channel = {id: CHANNEL_ID, guildId: guilded ? 20n : null}; + const authChannel: AuthenticatedChannel = { + channel: channel as AuthenticatedChannel['channel'], + guild: guilded ? ({message_history_cutoff: messageHistoryCutoff} as AuthenticatedChannel['guild']) : null, + member: null, + hasPermission: async (permission: bigint) => { + permissionChecks.push(permission); + return canReadHistory; + }, + checkPermission: async () => {}, + }; + const channelRepository = { + messages: { + getMessage: async (_channelId: ChannelID, messageId: MessageID) => storedById.get(messageId.toString()) ?? null, + }, + } as unknown as IChannelRepositoryAggregate; + const channelAuthService = { + getChannelAuthenticated: async ({channelId}: {channelId: ChannelID}) => { + authenticationCalls.push(channelId.toString()); + return authChannel; + }, + } as unknown as MessageChannelAuthService; + const processingService = { + repairMentionsOnRead: async (message: Message) => message, + } as unknown as MessageProcessingService; + const attachmentDecayService = { + extendForAttachments: async () => {}, + } as unknown as AttachmentDecayService; + const service = new MessageRetrievalService( + channelRepository, + {} as unknown as UserCacheService, + channelAuthService, + processingService, + {} as unknown as MessageSearchService, + {} as unknown as IUserRepository, + attachmentDecayService, + ); + return {service, permissionChecks, authenticationCalls}; +} + +describe('MessageRetrievalService.getMessagesByIds', () => { + it('authenticates the channel and checks message history once for the whole batch', async () => { + const messages = [ + makeMessage(CHANNEL_ID, makeMessageId(new Date('2026-02-01T00:00:00.000Z'), 1)), + makeMessage(CHANNEL_ID, makeMessageId(new Date('2026-02-02T00:00:00.000Z'), 2)), + makeMessage(CHANNEL_ID, makeMessageId(new Date('2026-02-03T00:00:00.000Z'), 3)), + ]; + const {service, permissionChecks, authenticationCalls} = createRetrievalService({ + guilded: true, + canReadHistory: true, + stored: messages, + }); + + const result = await service.getMessagesByIds({ + userId: VIEWER_ID, + channelId: CHANNEL_ID, + messageIds: messages.map((message) => message.id), + }); + + expect(Array.from(result.keys())).toEqual(messages.map((message) => message.id.toString())); + expect(authenticationCalls).toEqual([CHANNEL_ID.toString()]); + expect(permissionChecks).toHaveLength(1); + }); + + it('drops messages older than the history cutoff when history is not readable', async () => { + const beforeCutoff = makeMessage(CHANNEL_ID, makeMessageId(new Date('2026-01-05T00:00:00.000Z'), 1)); + const afterCutoff = makeMessage(CHANNEL_ID, makeMessageId(new Date('2026-01-20T00:00:00.000Z'), 2)); + const {service, permissionChecks} = createRetrievalService({ + guilded: true, + canReadHistory: false, + stored: [beforeCutoff, afterCutoff], + }); + + const result = await service.getMessagesByIds({ + userId: VIEWER_ID, + channelId: CHANNEL_ID, + messageIds: [beforeCutoff.id, afterCutoff.id], + }); + + expect(Array.from(result.keys())).toEqual([afterCutoff.id.toString()]); + expect(permissionChecks).toHaveLength(1); + }); + + it('drops every message when history is not readable and no cutoff is configured', async () => { + const message = makeMessage(CHANNEL_ID, makeMessageId(new Date('2026-02-01T00:00:00.000Z'), 1)); + const {service} = createRetrievalService({ + guilded: true, + canReadHistory: false, + messageHistoryCutoff: null, + stored: [message], + }); + + const result = await service.getMessagesByIds({ + userId: VIEWER_ID, + channelId: CHANNEL_ID, + messageIds: [message.id], + }); + + expect(result.size).toBe(0); + }); + + it('skips the history permission check entirely for direct messages', async () => { + const message = makeMessage(CHANNEL_ID, makeMessageId(new Date('2026-02-01T00:00:00.000Z'), 1)); + const {service, permissionChecks} = createRetrievalService({ + guilded: false, + canReadHistory: false, + stored: [message], + }); + + const result = await service.getMessagesByIds({ + userId: VIEWER_ID, + channelId: CHANNEL_ID, + messageIds: [message.id], + }); + + expect(Array.from(result.keys())).toEqual([message.id.toString()]); + expect(permissionChecks).toEqual([]); + }); + + it('omits messages that no longer exist instead of failing the batch', async () => { + const present = makeMessage(CHANNEL_ID, makeMessageId(new Date('2026-02-01T00:00:00.000Z'), 1)); + const deleted = makeMessage(CHANNEL_ID, makeMessageId(new Date('2026-02-02T00:00:00.000Z'), 2)); + const {service} = createRetrievalService({ + guilded: true, + canReadHistory: true, + stored: [present], + }); + + const result = await service.getMessagesByIds({ + userId: VIEWER_ID, + channelId: CHANNEL_ID, + messageIds: [present.id, deleted.id], + }); + + expect(Array.from(result.keys())).toEqual([present.id.toString()]); + }); +}); diff --git a/fluxer_api/src/api/channel/services/message/MessageRetrievalService.ts b/fluxer_api/src/api/channel/services/message/MessageRetrievalService.ts index 2ea59bc44..91b1ea4af 100644 --- a/fluxer_api/src/api/channel/services/message/MessageRetrievalService.ts +++ b/fluxer_api/src/api/channel/services/message/MessageRetrievalService.ts @@ -105,6 +105,34 @@ export class MessageRetrievalService { return repairedMessage; } + async getMessagesByIds({ + userId, + channelId, + messageIds, + }: { + userId: UserID; + channelId: ChannelID; + messageIds: Array; + }): Promise> { + const authChannel = await this.channelAuthService.getChannelAuthenticated({userId, channelId}); + const canReadMessageHistory = + !authChannel.guild || (await authChannel.hasPermission(Permissions.READ_MESSAGE_HISTORY)); + const cutoff = authChannel.guild?.message_history_cutoff ?? null; + const readableIds = messageIds.filter( + (messageId) => canReadMessageHistory || (cutoff != null && this.isMessageAfterCutoff(messageId, cutoff)), + ); + const found = ( + await Promise.all( + readableIds.map((messageId) => this.channelRepository.messages.getMessage(channelId, messageId)), + ) + ).filter((message): message is Message => message != null); + const repairedMessages = await Promise.all( + found.map((message) => this.processingService.repairMentionsOnRead(message, authChannel.channel)), + ); + await this.extendAttachments(repairedMessages); + return new Map(repairedMessages.map((message) => [message.id.toString(), message] as const)); + } + async searchMessages({ userId, channelId, diff --git a/fluxer_api/src/api/user/services/UserContentRequestService.ts b/fluxer_api/src/api/user/services/UserContentRequestService.ts index 2facd1a17..a5ebdc161 100644 --- a/fluxer_api/src/api/user/services/UserContentRequestService.ts +++ b/fluxer_api/src/api/user/services/UserContentRequestService.ts @@ -1,6 +1,6 @@ // SPDX-License-Identifier: AGPL-3.0-or-later -import type {MessageListResponse} from '@fluxer/schema/src/domains/message/MessageResponseSchemas'; +import type {MessageListResponse, MessageResponse} from '@fluxer/schema/src/domains/message/MessageResponseSchemas'; import type { HarvestCreationResponseSchema, HarvestDownloadUrlResponse, @@ -16,6 +16,7 @@ import type {ChannelID, MessageID, UserID} from '../../BrandedTypes'; import type {IStorageService} from '../../infrastructure/IStorageService'; import type {UserCacheService} from '../../infrastructure/UserCacheService'; import type {RequestCache} from '../../middleware/RequestCacheMiddleware'; +import type {Message} from '../../models/Message'; import type {SavedMessageEntry, UserContentService} from './UserContentService'; type HarvestCreationResponse = z.infer; @@ -108,7 +109,12 @@ export class UserContentRequestService { async listSavedMessages(params: SavedMessagesParams): Promise { const entries = await this.userContentService.getSavedMessages({userId: params.userId, limit: params.limit}); - return Promise.all(entries.map((entry) => this.mapSavedMessageEntry(params.userId, entry))); + const messages = entries.map((entry) => entry.message).filter((message): message is Message => message != null); + const responses = await this.userContentService.buildMessageResponsesForUser(params.userId, messages); + const responseByMessageId = new Map(responses.map((response) => [response.id, response] as const)); + return entries.map((entry) => + this.mapSavedMessageEntry(entry, responseByMessageId.get(entry.messageId.toString()) ?? null), + ); } async saveMessage(params: SaveMessageParams): Promise { @@ -154,15 +160,13 @@ export class UserContentRequestService { return this.userContentService.streamHarvestDownload(params); } - private async mapSavedMessageEntry(userId: UserID, entry: SavedMessageEntry): Promise { + private mapSavedMessageEntry(entry: SavedMessageEntry, message: MessageResponse | null): SavedMessageEntryResponse { return { id: entry.messageId.toString(), channel_id: entry.channelId.toString(), message_id: entry.messageId.toString(), status: entry.status, - message: entry.message - ? (await this.userContentService.buildMessageResponsesForUser(userId, [entry.message]))[0] - : null, + message, }; } } diff --git a/fluxer_api/src/api/user/services/UserContentService.test.ts b/fluxer_api/src/api/user/services/UserContentService.test.ts index 2a69e58c8..21eb70b1a 100644 --- a/fluxer_api/src/api/user/services/UserContentService.test.ts +++ b/fluxer_api/src/api/user/services/UserContentService.test.ts @@ -1,11 +1,21 @@ // SPDX-License-Identifier: AGPL-3.0-or-later +import {MessageTypes} from '@fluxer/constants/src/ChannelConstants'; import {UnknownChannelError} from '@fluxer/errors/src/domains/channel/UnknownChannelError'; import {UnknownMessageError} from '@fluxer/errors/src/domains/channel/UnknownMessageError'; import {MissingPermissionsError} from '@fluxer/errors/src/domains/core/MissingPermissionsError'; import {UnknownGuildError} from '@fluxer/errors/src/domains/guild/UnknownGuildError'; import {describe, expect, it} from 'vitest'; -import {UserContentServiceTestHooks} from './UserContentService'; +import type {ApiContext} from '../../ApiContext'; +import type {ChannelID, MessageID, UserID} from '../../BrandedTypes'; +import {createChannelID, createMessageID, createUserID} from '../../BrandedTypes'; +import type {IChannelRepository} from '../../channel/IChannelRepository'; +import type {ChannelService} from '../../channel/services/ChannelService'; +import type {KVBulkMessageDeletionQueueService} from '../../infrastructure/KVBulkMessageDeletionQueueService'; +import type {UserCacheService} from '../../infrastructure/UserCacheService'; +import type {LimitConfigService} from '../../limits/LimitConfigService'; +import {Message} from '../../models/Message'; +import {UserContentService, UserContentServiceTestHooks} from './UserContentService'; const {isUnreachableEntityError} = UserContentServiceTestHooks; @@ -28,3 +38,222 @@ describe('isUnreachableEntityError', () => { expect(isUnreachableEntityError(null)).toBe(false); }); }); + +const VIEWER_ID = createUserID(7n); + +interface ChannelBatchCall { + channelId: string; + messageIds: Array; +} + +function makeMessage(channelId: ChannelID, messageId: MessageID): Message { + return new Message({ + channel_id: channelId, + bucket: 0, + message_id: messageId, + author_id: createUserID(3n), + type: MessageTypes.DEFAULT, + webhook_id: null, + webhook_name: null, + webhook_avatar_hash: null, + content: '', + edited_timestamp: null, + pinned_timestamp: null, + flags: 0, + mention_everyone: false, + mention_users: null, + mention_roles: null, + mention_channels: null, + attachments: null, + embeds: null, + sticker_items: null, + message_reference: null, + message_snapshots: null, + call: null, + nsfw_emojis: null, + has_reaction: null, + version: 1, + }); +} + +function createUserContentService({ + entries, + readable, + failures = new Map(), +}: { + entries: Array<{channelId: ChannelID; messageId: MessageID}>; + readable: Array<{channelId: ChannelID; messageId: MessageID}>; + failures?: Map; +}) { + const batchCalls: Array = []; + const deletedSavedMessageIds: Array = []; + const readableByChannel = new Map>(); + for (const entry of readable) { + const key = entry.channelId.toString(); + const messages = readableByChannel.get(key) ?? new Map(); + messages.set(entry.messageId.toString(), makeMessage(entry.channelId, entry.messageId)); + readableByChannel.set(key, messages); + } + const userRepository = { + listRecentMentions: async () => entries, + listSavedMessages: async () => entries, + deleteSavedMessage: async (_userId: UserID, messageId: MessageID) => { + deletedSavedMessageIds.push(messageId.toString()); + }, + }; + const channelService = { + messages: { + retrieval: { + getMessagesByIds: async ({channelId, messageIds}: {channelId: ChannelID; messageIds: Array}) => { + batchCalls.push({ + channelId: channelId.toString(), + messageIds: messageIds.map((messageId) => messageId.toString()), + }); + const failure = failures.get(channelId.toString()); + if (failure) throw failure; + return readableByChannel.get(channelId.toString()) ?? new Map(); + }, + }, + }, + }; + const service = new UserContentService( + {services: {users: userRepository, gateway: {}, worker: {}, snowflake: {}}} as unknown as ApiContext, + {} as unknown as UserCacheService, + channelService as unknown as ChannelService, + {} as unknown as IChannelRepository, + {} as unknown as KVBulkMessageDeletionQueueService, + {} as unknown as LimitConfigService, + ); + return {service, batchCalls, deletedSavedMessageIds}; +} + +const CHANNEL_A = createChannelID(100n); +const CHANNEL_B = createChannelID(200n); +const CHANNEL_C = createChannelID(300n); + +describe('getRecentMentions', () => { + it('authenticates each distinct channel once no matter how many mentions it holds', async () => { + const entries = [ + {channelId: CHANNEL_A, messageId: createMessageID(11n)}, + {channelId: CHANNEL_B, messageId: createMessageID(12n)}, + {channelId: CHANNEL_A, messageId: createMessageID(13n)}, + {channelId: CHANNEL_A, messageId: createMessageID(14n)}, + ]; + const {service, batchCalls} = createUserContentService({entries, readable: entries}); + + const messages = await service.getRecentMentions({ + userId: VIEWER_ID, + limit: 50, + everyone: true, + roles: true, + guilds: true, + }); + + expect(messages.map((message) => message.id.toString())).toEqual(['14', '13', '12', '11']); + expect(batchCalls).toEqual([ + {channelId: '100', messageIds: ['11', '13', '14']}, + {channelId: '200', messageIds: ['12']}, + ]); + }); + + it('drops only the mentions from the channels the viewer can no longer reach', async () => { + const entries = [ + {channelId: CHANNEL_A, messageId: createMessageID(11n)}, + {channelId: CHANNEL_B, messageId: createMessageID(12n)}, + {channelId: CHANNEL_C, messageId: createMessageID(13n)}, + {channelId: CHANNEL_B, messageId: createMessageID(14n)}, + ]; + const {service} = createUserContentService({ + entries, + readable: [entries[0], entries[2]], + failures: new Map([ + [CHANNEL_B.toString(), new MissingPermissionsError()], + [CHANNEL_C.toString(), new UnknownChannelError()], + ]), + }); + + const messages = await service.getRecentMentions({ + userId: VIEWER_ID, + limit: 50, + everyone: true, + roles: true, + guilds: true, + }); + + expect(messages.map((message) => message.id.toString())).toEqual(['11']); + }); + + it('drops a mention whose message the batch read did not return', async () => { + const entries = [ + {channelId: CHANNEL_A, messageId: createMessageID(11n)}, + {channelId: CHANNEL_A, messageId: createMessageID(12n)}, + ]; + const {service} = createUserContentService({entries, readable: [entries[1]]}); + + const messages = await service.getRecentMentions({ + userId: VIEWER_ID, + limit: 50, + everyone: true, + roles: true, + guilds: true, + }); + + expect(messages.map((message) => message.id.toString())).toEqual(['12']); + }); + + it('still lets an unexpected channel failure surface', async () => { + const entries = [{channelId: CHANNEL_A, messageId: createMessageID(11n)}]; + const {service} = createUserContentService({ + entries, + readable: [], + failures: new Map([[CHANNEL_A.toString(), new Error('database is on fire')]]), + }); + + await expect( + service.getRecentMentions({userId: VIEWER_ID, limit: 50, everyone: true, roles: true, guilds: true}), + ).rejects.toThrow('database is on fire'); + }); +}); + +describe('getSavedMessages', () => { + it('marks every entry of an unreachable channel as missing permissions without deleting it', async () => { + const entries = [ + {channelId: CHANNEL_A, messageId: createMessageID(11n)}, + {channelId: CHANNEL_B, messageId: createMessageID(12n)}, + {channelId: CHANNEL_B, messageId: createMessageID(13n)}, + ]; + const {service, batchCalls, deletedSavedMessageIds} = createUserContentService({ + entries, + readable: [entries[0]], + failures: new Map([[CHANNEL_B.toString(), new UnknownGuildError()]]), + }); + + const saved = await service.getSavedMessages({userId: VIEWER_ID, limit: 50}); + + expect( + saved.map((entry) => ({id: entry.messageId.toString(), status: entry.status, hasMessage: entry.message != null})), + ).toEqual([ + {id: '13', status: 'missing_permissions', hasMessage: false}, + {id: '12', status: 'missing_permissions', hasMessage: false}, + {id: '11', status: 'available', hasMessage: true}, + ]); + expect(deletedSavedMessageIds).toEqual([]); + expect(batchCalls).toEqual([ + {channelId: '100', messageIds: ['11']}, + {channelId: '200', messageIds: ['12', '13']}, + ]); + }); + + it('deletes and drops a saved message the batch read could not return', async () => { + const entries = [ + {channelId: CHANNEL_A, messageId: createMessageID(11n)}, + {channelId: CHANNEL_A, messageId: createMessageID(12n)}, + ]; + const {service, deletedSavedMessageIds} = createUserContentService({entries, readable: [entries[0]]}); + + const saved = await service.getSavedMessages({userId: VIEWER_ID, limit: 50}); + + expect(saved.map((entry) => entry.messageId.toString())).toEqual(['11']); + expect(deletedSavedMessageIds).toEqual(['12']); + }); +}); diff --git a/fluxer_api/src/api/user/services/UserContentService.ts b/fluxer_api/src/api/user/services/UserContentService.ts index bc90919c6..b41e2a039 100644 --- a/fluxer_api/src/api/user/services/UserContentService.ts +++ b/fluxer_api/src/api/user/services/UserContentService.ts @@ -141,25 +141,56 @@ export class UserContentService { }): Promise> { const {userId, limit, everyone, roles, guilds, before} = params; const mentions = await this.userRepository.listRecentMentions(userId, everyone, roles, guilds, limit, before); - const messagePromises = mentions.map(async (mention) => { - try { - return await this.channelService.messages.retrieval.getMessage({ - userId, - channelId: mention.channelId, - messageId: mention.messageId, - }); - } catch (error) { - if (error instanceof UnknownMessageError || isUnreachableEntityError(error)) { - return null; - } - throw error; - } - }); - const messageResults = await Promise.all(messagePromises); - const messages = messageResults.filter((message): message is Message => message != null); + const messagesByChannel = await this.readMessagesByChannel(userId, mentions); + const messages = mentions + .map((mention) => this.pickMessage(messagesByChannel, mention)) + .filter((message): message is Message => message != null); return messages.sort((a, b) => (b.id > a.id ? 1 : -1)); } + private async readMessagesByChannel( + userId: UserID, + entries: ReadonlyArray<{channelId: ChannelID; messageId: MessageID}>, + ): Promise | null>> { + const grouped = new Map}>(); + for (const entry of entries) { + const key = entry.channelId.toString(); + const group = grouped.get(key); + if (group) { + group.messageIds.push(entry.messageId); + } else { + grouped.set(key, {channelId: entry.channelId, messageIds: [entry.messageId]}); + } + } + const messagesByChannel = new Map | null>(); + await Promise.all( + Array.from(grouped, async ([key, group]) => { + try { + const messages = await this.channelService.messages.retrieval.getMessagesByIds({ + userId, + channelId: group.channelId, + messageIds: group.messageIds, + }); + messagesByChannel.set(key, messages); + } catch (error) { + if (isUnreachableEntityError(error)) { + messagesByChannel.set(key, null); + return; + } + throw error; + } + }), + ); + return messagesByChannel; + } + + private pickMessage( + messagesByChannel: ReadonlyMap | null>, + entry: {channelId: ChannelID; messageId: MessageID}, + ): Message | null { + return messagesByChannel.get(entry.channelId.toString())?.get(entry.messageId.toString()) ?? null; + } + async deleteRecentMention({userId, messageId}: {userId: UserID; messageId: MessageID}): Promise { const recentMention = await this.userRepository.getRecentMention(userId, messageId); if (!recentMention) return; @@ -181,35 +212,33 @@ export class UserContentService { async getSavedMessages({userId, limit}: {userId: UserID; limit: number}): Promise> { const savedMessages = await this.userRepository.listSavedMessages(userId, limit); - const messagePromises = savedMessages.map(async (savedMessage) => { - let message: Message | null = null; - let status: SavedMessageStatus = 'available'; - try { - message = await this.channelService.messages.retrieval.getMessage({ - userId, + const messagesByChannel = await this.readMessagesByChannel(userId, savedMessages); + const results: Array = []; + const staleMessageIds: Array = []; + for (const savedMessage of savedMessages) { + const channelMessages = messagesByChannel.get(savedMessage.channelId.toString()); + if (channelMessages === null) { + results.push({ channelId: savedMessage.channelId, messageId: savedMessage.messageId, + status: 'missing_permissions', + message: null, }); - } catch (error) { - if (error instanceof UnknownMessageError) { - await this.userRepository.deleteSavedMessage(userId, savedMessage.messageId); - return null; - } - if (isUnreachableEntityError(error)) { - status = 'missing_permissions'; - } else { - throw error; - } + continue; } - return { + const message = this.pickMessage(messagesByChannel, savedMessage); + if (!message) { + staleMessageIds.push(savedMessage.messageId); + continue; + } + results.push({ channelId: savedMessage.channelId, messageId: savedMessage.messageId, - status, + status: 'available', message, - }; - }); - const messageResults = await Promise.all(messagePromises); - const results = messageResults.filter((result): result is NonNullable => result != null); + }); + } + await Promise.all(staleMessageIds.map((messageId) => this.userRepository.deleteSavedMessage(userId, messageId))); return results.sort((a, b) => (b.messageId > a.messageId ? 1 : a.messageId > b.messageId ? -1 : 0)); }