merge uploads

.changeset/spotty-steaks-notice.md

@@ -0,0 +1,6 @@
+---
+'@rocket.chat/apps-engine': patch
+'@rocket.chat/meteor': patch
+---
+
+Fixes an issue that caused a spike in memory usage when apps handled the IPreFileUpload event

apps/meteor/app/api/server/definition.ts

@@ -1,3 +1,5 @@
+import type { IncomingMessage } from 'http';
+
 import type { IUser, LicenseModule } from '@rocket.chat/core-typings';
 import type { Logger } from '@rocket.chat/logger';
 import type { Method, MethodOf, OperationParams, OperationResult, PathPattern, UrlParams } from '@rocket.chat/rest-typings';
@@ -184,6 +186,7 @@ export type ActionThis<TMethod extends Method, TPathPattern extends PathPattern,
 				: // TODO remove the extra (optionals) params when all the endpoints that use these are typed correctly
 					Partial<OperationParams<TMethod, TPathPattern>>;
 	readonly request: Request;
+	readonly incoming: IncomingMessage;
 
 	readonly queryOperations: TOptions extends { queryOperations: infer T } ? T : never;
 	readonly queryFields: TOptions extends { queryFields: infer T } ? T : never;

apps/meteor/app/api/server/lib/MultipartUploadHandler.ts

@@ -0,0 +1,203 @@
+import fs from 'fs';
+import { IncomingMessage } from 'http';
+import type { Stream, Transform } from 'stream';
+import { Readable } from 'stream';
+import { pipeline } from 'stream/promises';
+
+import { MeteorError } from '@rocket.chat/core-services';
+import { Random } from '@rocket.chat/random';
+import busboy, { type BusboyConfig } from 'busboy';
+import ExifTransformer from 'exif-be-gone';
+
+import { UploadFS } from '../../../../server/ufs';
+import { getMimeType } from '../../../utils/lib/mimeTypes';
+
+export type ParsedUpload = {
+	tempFilePath: string;
+	filename: string;
+	mimetype: string;
+	size: number;
+	fieldname: string;
+};
+
+export type ParseOptions = {
+	field: string;
+	maxSize?: number;
+	allowedMimeTypes?: string[];
+	transforms?: Transform[]; // Optional transform pipeline (e.g., EXIF stripping)
+	fileOptional?: boolean;
+};
+
+export class MultipartUploadHandler {
+	static transforms = {
+		stripExif(): Transform {
+			return new ExifTransformer();
+		},
+	};
+
+	static async cleanup(tempFilePath: string): Promise<void> {
+		try {
+			await fs.promises.unlink(tempFilePath);
+		} catch (error: any) {
+			console.warn(`[UploadService] Failed to cleanup temp file: ${tempFilePath}`, error);
+		}
+	}
+
+	static async stripExifFromFile(tempFilePath: string): Promise<number> {
+		const strippedPath = `${tempFilePath}.stripped`;
+
+		try {
+			const writeStream = fs.createWriteStream(strippedPath);
+
+			await pipeline(fs.createReadStream(tempFilePath), new ExifTransformer(), writeStream);
+
+			await fs.promises.rename(strippedPath, tempFilePath);
+
+			return writeStream.bytesWritten;
+		} catch (error) {
+			void this.cleanup(strippedPath);
+
+			throw error;
+		}
+	}
+
+	static async parseRequest(
+		request: IncomingMessage | Request,
+		options: ParseOptions,
+	): Promise<{ file: ParsedUpload | null; fields: Record<string, string> }> {
+		const limits: BusboyConfig['limits'] = { files: 1 };
+
+		if (options.maxSize && options.maxSize > 0) {
+			// We add an extra byte to the configured limit so we don't fail the upload
+			// of a file that is EXACTLY maxSize
+			limits.fileSize = options.maxSize + 1;
+		}
+
+		const headers =
+			request instanceof IncomingMessage ? (request.headers as Record<string, string>) : Object.fromEntries(request.headers.entries());
+
+		const bb = busboy({
+			headers,
+			defParamCharset: 'utf8',
+			limits,
+		});
+
+		const fields: Record<string, string> = {};
+		let parsedFile: ParsedUpload | null = null;
+		let busboyFinished = false;
+		let writeStreamFinished = options.fileOptional === true;
+
+		const { promise, resolve, reject } = Promise.withResolvers<{
+			file: ParsedUpload | null;
+			fields: Record<string, string>;
+		}>();
+
+		const tryResolve = () => {
+			if (busboyFinished && writeStreamFinished) {
+				if (!parsedFile && !options.fileOptional) {
+					return reject(new MeteorError('error-no-file', 'No file uploaded'));
+				}
+				resolve({ file: parsedFile, fields });
+			}
+		};
+
+		bb.on('field', (fieldname: string, value: string) => {
+			fields[fieldname] = value;
+		});
+
+		bb.on('file', (fieldname, file, info) => {
+			const { filename, mimeType } = info;
+
+			writeStreamFinished = false;
+
+			if (options.field && fieldname !== options.field) {
+				file.resume();
+				return reject(new MeteorError('invalid-field'));
+			}
+
+			if (options.allowedMimeTypes && !options.allowedMimeTypes.includes(mimeType)) {
+				file.resume();
+				return reject(new MeteorError('error-invalid-file-type', `File type ${mimeType} not allowed`));
+			}
+
+			const fileId = Random.id();
+			const tempFilePath = UploadFS.getTempFilePath(fileId);
+
+			const writeStream = fs.createWriteStream(tempFilePath);
+
+			let currentStream: Stream = file;
+			if (options.transforms?.length) {
+				const fileDestroyer = file.destroy.bind(file);
+				for (const transform of options.transforms) {
+					transform.on('error', fileDestroyer);
+					currentStream = currentStream.pipe(transform);
+				}
+			}
+
+			currentStream.pipe(writeStream);
+
+			writeStream.on('finish', () => {
+				if (file.truncated) {
+					void this.cleanup(tempFilePath);
+					return reject(new MeteorError('error-file-too-large', 'File size exceeds the allowed limit'));
+				}
+
+				parsedFile = {
+					tempFilePath,
+					filename,
+					mimetype: getMimeType(mimeType, filename),
+					size: writeStream.bytesWritten,
+					fieldname,
+				};
+				writeStreamFinished = true;
+				tryResolve();
+			});
+
+			writeStream.on('error', (err) => {
+				file.destroy();
+				void this.cleanup(tempFilePath);
+				reject(new MeteorError('error-file-upload', err.message));
+			});
+
+			file.on('error', (err) => {
+				writeStream.destroy();
+				void this.cleanup(tempFilePath);
+				reject(new MeteorError('error-file-upload', err.message));
+			});
+		});
+
+		bb.on('finish', () => {
+			busboyFinished = true;
+			tryResolve();
+		});
+
+		bb.on('error', (err: any) => {
+			reject(new MeteorError('error-upload-failed', err.message));
+		});
+
+		bb.on('filesLimit', () => {
+			reject(new MeteorError('error-too-many-files', 'Too many files in upload'));
+		});
+
+		bb.on('partsLimit', () => {
+			reject(new MeteorError('error-too-many-parts', 'Too many parts in upload'));
+		});
+
+		bb.on('fieldsLimit', () => {
+			reject(new MeteorError('error-too-many-fields', 'Too many fields in upload'));
+		});
+
+		if (request instanceof IncomingMessage) {
+			request.pipe(bb);
+		} else {
+			if (!request.body) {
+				return Promise.reject(new MeteorError('error-no-body', 'Request has no body'));
+			}
+
+			const nodeStream = Readable.fromWeb(request.body as any);
+			nodeStream.pipe(bb);
+		}
+
+		return promise;
+	}
+}

apps/meteor/app/api/server/middlewares/logger.ts

@@ -10,10 +10,15 @@ export const loggerMiddleware =
 
 		let payload = {};
 
-		try {
-			payload = await c.req.raw.clone().json();
-			// eslint-disable-next-line no-empty
-		} catch {}
+		// We don't want to consume the request body stream for multipart requests
+		if (!c.req.header('content-type')?.includes('multipart/form-data')) {
+			try {
+				payload = await c.req.raw.clone().json();
+				// eslint-disable-next-line no-empty
+			} catch {}
+		} else {
+			payload = '[multipart/form-data]';
+		}
 
 		const log = logger.logger.child({
 			method: c.req.method,

apps/meteor/app/api/server/router.ts

@@ -1,16 +1,18 @@
-/* eslint-disable @typescript-eslint/naming-convention */
+import type { IncomingMessage } from 'node:http';
+
 import type { ResponseSchema } from '@rocket.chat/http-router';
 import { Router } from '@rocket.chat/http-router';
-import type { Context as HonoContext } from 'hono';
+import type { Context } from 'hono';
 
 import type { TypedOptions } from './definition';
 
-declare module 'hono' {
-	interface ContextVariableMap {
-		'route': string;
+type HonoContext = Context<{
+	Bindings: { incoming: IncomingMessage };
+	Variables: {
+		'remoteAddress': string;
 		'bodyParams-override'?: Record<string, any>;
-	}
-}
+	};
+}>;
 
 export type APIActionContext = {
 	requestIp: string;
@@ -21,6 +23,7 @@ export type APIActionContext = {
 	path: string;
 	response: any;
 	route: string;
+	incoming: IncomingMessage;
 };
 
 export type APIActionHandler = (this: APIActionContext, request: Request) => Promise<ResponseSchema<TypedOptions>>;
@@ -39,9 +42,10 @@ export class RocketChatAPIRouter<
 				request: req,
 				extra: { bodyParamsOverride: c.var['bodyParams-override'] || {} },
 			});
+
 			const request = req.raw.clone();
 
-			const context = {
+			const context: APIActionContext = {
 				requestIp: c.get('remoteAddress'),
 				urlParams: req.param(),
 				queryParams,
@@ -50,7 +54,8 @@ export class RocketChatAPIRouter<
 				path: req.path,
 				response: res,
 				route: req.routePath,
-			} as APIActionContext;
+				incoming: c.env.incoming,
+			};
 
 			return action.apply(context, [request]);
 		};

apps/meteor/app/api/server/v1/rooms.ts

@@ -1,4 +1,4 @@
-import { FederationMatrix, Media, MeteorError, Team } from '@rocket.chat/core-services';
+import { FederationMatrix, MeteorError, Team } from '@rocket.chat/core-services';
 import type { IRoom, IUpload } from '@rocket.chat/core-typings';
 import { isPrivateRoom, isPublicRoom } from '@rocket.chat/core-typings';
 import { Messages, Rooms, Users, Uploads, Subscriptions } from '@rocket.chat/models';
@@ -55,7 +55,7 @@ import { API } from '../api';
 import { composeRoomWithLastMessage } from '../helpers/composeRoomWithLastMessage';
 import { getPaginationItems } from '../helpers/getPaginationItems';
 import { getUserFromParams } from '../helpers/getUserFromParams';
-import { getUploadFormData } from '../lib/getUploadFormData';
+import { MultipartUploadHandler } from '../lib/MultipartUploadHandler';
 import {
 	findAdminRoom,
 	findAdminRooms,
@@ -197,24 +197,18 @@ API.v1.addRoute(
 				return API.v1.forbidden();
 			}
 
-			const file = await getUploadFormData(
-				{
-					request: this.request,
-				},
-				{ field: 'file', sizeLimit: settings.get<number>('FileUpload_MaxFileSize') },
-			);
+			const { file, fields } = await MultipartUploadHandler.parseRequest(this.incoming, {
+				field: 'file',
+				maxSize: settings.get<number>('FileUpload_MaxFileSize'),
+			});
 
 			if (!file) {
-				throw new Meteor.Error('invalid-field');
+				throw new Meteor.Error('error-no-file-uploaded', 'No file was uploaded');
 			}
 
-			let { fileBuffer } = file;
-
 			const expiresAt = new Date();
 			expiresAt.setHours(expiresAt.getHours() + 24);
 
-			const { fields } = file;
-
 			let content;
 
 			if (fields.content) {
@@ -228,23 +222,17 @@ API.v1.addRoute(
 
 			const details = {
 				name: file.filename,
-				size: fileBuffer.length,
+				size: file.size,
 				type: file.mimetype,
 				rid: this.urlParams.rid,
 				userId: this.userId,
 				content,
 				expiresAt,
 			};
 
-			const stripExif = settings.get('Message_Attachments_Strip_Exif');
-			if (stripExif) {
-				// No need to check mime. Library will ignore any files without exif/xmp tags (like BMP, ico, PDF, etc)
-				fileBuffer = await Media.stripExifFromBuffer(fileBuffer);
-				details.size = fileBuffer.length;
-			}
-
+			// TODO: In the future, we should isolate file receival from storage and post-processing.
 			const fileStore = FileUpload.getStore('Uploads');
-			const uploadedFile = await fileStore.insert(details, fileBuffer);
+			const uploadedFile = await fileStore.insert(details, file.tempFilePath);
 
 			uploadedFile.path = FileUpload.getPath(`${uploadedFile._id}/${encodeURI(uploadedFile.name || '')}`);