perf(user): batch mention and saved message reads (#2139)

This commit is contained in:
Hampus authored and GitHub committed 2026-08-30 22:27:41 +02:00
1 parent c207918e90
commit 8491872908
5 files changed
+542 -45

No files matched your search

@@ -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<Message>;
}) {
const permissionChecks: Array<bigint> = [];
const authenticationCalls: Array<string> = [];
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()]);
});
});
@@ -105,6 +105,34 @@ export class MessageRetrievalService {
return repairedMessage;
}
async getMessagesByIds({
userId,
channelId,
messageIds,
}: {
userId: UserID;
channelId: ChannelID;
messageIds: Array<MessageID>;
}): Promise<Map<string, Message>> {
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,
@@ -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<typeof HarvestCreationResponseSchema>;
@@ -108,7 +109,12 @@ export class UserContentRequestService {
async listSavedMessages(params: SavedMessagesParams): Promise<SavedMessageEntryListResponse> {
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<void> {
@@ -154,15 +160,13 @@ export class UserContentRequestService {
return this.userContentService.streamHarvestDownload(params);
}
private async mapSavedMessageEntry(userId: UserID, entry: SavedMessageEntry): Promise<SavedMessageEntryResponse> {
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,
};
}
}
@@ -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<string>;
}
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<string, Error>(),
}: {
entries: Array<{channelId: ChannelID; messageId: MessageID}>;
readable: Array<{channelId: ChannelID; messageId: MessageID}>;
failures?: Map<string, Error>;
}) {
const batchCalls: Array<ChannelBatchCall> = [];
const deletedSavedMessageIds: Array<string> = [];
const readableByChannel = new Map<string, Map<string, Message>>();
for (const entry of readable) {
const key = entry.channelId.toString();
const messages = readableByChannel.get(key) ?? new Map<string, Message>();
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<MessageID>}) => {
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<string, Message>();
},
},
},
};
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<string, Error>([
[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<string, Error>([[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<string, Error>([[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']);
});
});
@@ -141,25 +141,56 @@ export class UserContentService {
}): Promise<Array<Message>> {
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<Map<string, Map<string, Message> | null>> {
const grouped = new Map<string, {channelId: ChannelID; messageIds: Array<MessageID>}>();
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<string, Map<string, Message> | 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<string, Map<string, Message> | 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<void> {
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<Array<SavedMessageEntry>> {
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<SavedMessageEntry> = [];
const staleMessageIds: Array<MessageID> = [];
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<typeof result> => 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));
}