fix(message): key batched message responses by message id (#2254)

fluxer_api/src/api/channel/services/message/MessageResponseDataService.test.ts

@@ -4,7 +4,7 @@ import {MessageTypes} from '@fluxer/constants/src/ChannelConstants';
 import type {INatsConnectionManager} from '@pkgs/nats/src/INatsConnectionManager';
 import type {NatsConnection} from 'nats';
 import {describe, expect, it} from 'vitest';
-import {createChannelID, createMessageID, createUserID} from '../../../BrandedTypes';
+import {createChannelID, createGuildID, createMessageID, createUserID} from '../../../BrandedTypes';
 import {Message} from '../../../models/Message';
 import {MESSAGE_BUILD_BATCH_MAX_BYTES, MessageResponseDataService} from './MessageResponseDataService';
 
@@ -15,6 +15,7 @@ const ROUTER_SHARD_REQUEST_TIMEOUT_MS = 5000;
 class FakeConnectionManager implements INatsConnectionManager {
 	readonly payloads: Array<Record<string, unknown>> = [];
 	readonly timeouts: Array<number | undefined> = [];
+	readonly orphanedMessageIds = new Set<string>();
 
 	async connect(): Promise<void> {}
 
@@ -31,7 +32,9 @@ class FakeConnectionManager implements INatsConnectionManager {
 				this.payloads.push(payload);
 				this.timeouts.push(options?.timeout);
 				if (payload.op === 'BuildResponses') {
-					const messages = payload.messages as Array<{message_id: string}>;
+					const messages = (payload.messages as Array<{message_id: string}>).filter(
+						(message) => !this.orphanedMessageIds.has(message.message_id),
+					);
 					return {
 						data: encoder.encode(
 							JSON.stringify({
@@ -69,9 +72,9 @@ function fakeMessageResponse(messageId: string): Record<string, unknown> {
 	};
 }
 
-function makeMessage(messageId: bigint = 2n, content: string = ''): Message {
+function makeMessage(messageId: bigint = 2n, content: string = '', channelId: bigint = 1n): Message {
 	return new Message({
-		channel_id: createChannelID(1n),
+		channel_id: createChannelID(channelId),
 		bucket: 0,
 		message_id: createMessageID(messageId),
 		author_id: createUserID(3n),
@@ -115,6 +118,10 @@ function makeSizedMessage(index: number, totalBytes: number, baseBytes: number):
 	return makeMessage(BASE_MESSAGE_ID + BigInt(index), 'a'.repeat(totalBytes - baseBytes));
 }
 
+function messageId(index: number): string {
+	return (BASE_MESSAGE_ID + BigInt(index)).toString();
+}
+
 function batchSizes(connectionManager: FakeConnectionManager): Array<number> {
 	return connectionManager.payloads.map((payload) => (payload as {messages: Array<unknown>}).messages.length);
 }
@@ -221,6 +228,45 @@ describe('MessageResponseDataService', () => {
 		expect(responses.map((response) => response.id)).toEqual(messages.map((message) => message.id.toString()));
 	});
 
+	it('drops a message the service leaves out instead of shifting later responses', async () => {
+		const connectionManager = new FakeConnectionManager();
+		connectionManager.orphanedMessageIds.add(messageId(1));
+		const service = new MessageResponseDataService(connectionManager);
+		const messages = [0, 1, 2].map((index) => makeMessage(BASE_MESSAGE_ID + BigInt(index)));
+
+		const responses = await service.buildMessagesForChannels({
+			userId: VIEWER_ID,
+			messages,
+			channelById: new Map([['1', {guildId: null}]]),
+		});
+
+		expect(responses.map((response) => response.id)).toEqual([messageId(0), messageId(2)]);
+	});
+
+	it('keeps responses in input order when channels resolve to different guilds', async () => {
+		const connectionManager = new FakeConnectionManager();
+		connectionManager.orphanedMessageIds.add(messageId(2));
+		const service = new MessageResponseDataService(connectionManager);
+		const messages = [
+			makeMessage(BASE_MESSAGE_ID, '', 10n),
+			makeMessage(BASE_MESSAGE_ID + 1n, '', 20n),
+			makeMessage(BASE_MESSAGE_ID + 2n, '', 10n),
+			makeMessage(BASE_MESSAGE_ID + 3n, '', 20n),
+		];
+
+		const responses = await service.buildMessagesForChannels({
+			userId: VIEWER_ID,
+			messages,
+			channelById: new Map([
+				['10', {guildId: createGuildID(100n)}],
+				['20', {guildId: createGuildID(200n)}],
+			]),
+		});
+
+		expect(batchSizes(connectionManager)).toEqual([2, 2]);
+		expect(responses.map((response) => response.id)).toEqual([messageId(0), messageId(1), messageId(3)]);
+	});
+
 	it('keeps a full page of ordinary pins in a single request', async () => {
 		const connectionManager = new FakeConnectionManager();
 		const service = new MessageResponseDataService(connectionManager);

fluxer_api/src/api/channel/services/message/MessageResponseDataService.ts

@@ -260,7 +260,7 @@ export class MessageResponseDataService {
 		channelById: ReadonlyMap<string, Pick<Channel, 'guildId'>>;
 		includeReactions?: boolean;
 	}): Promise<Array<MessageResponse>> {
-		const responses = new Array<MessageResponse>(params.messages.length);
+		const responses = new Array<MessageResponse | undefined>(params.messages.length);
 		const groups = new Map<
 			string,
 			{
@@ -290,12 +290,13 @@ export class MessageResponseDataService {
 					access: group.access,
 					includeReactions: params.includeReactions,
 				});
-				for (const [mappedIndex, entry] of group.entries.entries()) {
-					responses[entry.index] = mapped[mappedIndex];
+				const mappedByMessageId = new Map(mapped.map((response) => [response.id, response] as const));
+				for (const entry of group.entries) {
+					responses[entry.index] = mappedByMessageId.get(entry.message.id.toString());
 				}
 			}),
 		);
-		return responses;
+		return responses.filter((response): response is MessageResponse => response !== undefined);
 	}
 
 	private async request(payload: Record<string, unknown>): Promise<MessageServiceResponse> {