chore: Streaming images for pdf transcript

apps/meteor/server/services/upload/service.ts

@@ -1,9 +1,15 @@
+import fs from 'fs';
+import type Stream from 'stream';
+
+import type { IUploadDetails } from '@rocket.chat/apps-engine/definition/uploads/IUploadDetails';
 import { ServiceClassInternal } from '@rocket.chat/core-services';
 import type { ISendFileLivechatMessageParams, ISendFileMessageParams, IUploadFileParams, IUploadService } from '@rocket.chat/core-services';
 import type { IUpload, IUser, FilesAndAttachments, IMessage } from '@rocket.chat/core-typings';
 import { isFileAttachment } from '@rocket.chat/core-typings';
 import { Logger } from '@rocket.chat/logger';
 import { Uploads } from '@rocket.chat/models';
+import { Random } from '@rocket.chat/random';
+import sharp from 'sharp';
 
 import { canAccessRoomIdAsync } from '../../../app/authorization/server/functions/canAccessRoom';
 import { canDeleteMessageAsync } from '../../../app/authorization/server/functions/canDeleteMessage';
@@ -13,6 +19,7 @@ import { updateMessage } from '../../../app/lib/server/functions/updateMessage';
 import { sendFileLivechatMessage } from '../../../app/livechat/server/methods/sendFileLivechatMessage';
 import { NOTIFICATION_ATTACHMENT_COLOR } from '../../../lib/constants';
 import { i18n } from '../../lib/i18n';
+import { UploadFS } from '../../ufs';
 
 const logger = new Logger('UploadService');
 
@@ -133,4 +140,71 @@ export class UploadService extends ServiceClassInternal implements IUploadServic
 
 		await updateMessage(editedMessage, user, msg);
 	}
+
+	async streamUploadedFile({
+		file,
+		imageResizeOpts,
+	}: {
+		file: IUpload;
+		imageResizeOpts?: { width: number; height: number };
+	}): Promise<Stream.Readable> {
+		const stream = await FileUpload.getStore('Uploads')._store.getReadStream(file._id, file);
+		if (!stream) {
+			throw new Error('error-file-not-found');
+		}
+
+		if (file?.type?.includes('image') && imageResizeOpts) {
+			const { width, height } = imageResizeOpts;
+
+			const transformer = sharp().resize({ width, height, fit: 'contain' });
+
+			stream.on('error', (err) => transformer.destroy(err));
+
+			return stream.pipe(transformer);
+		}
+
+		return stream;
+	}
+
+	async uploadFileFromStream({
+		streamParam,
+		details,
+	}: {
+		streamParam: Stream.Readable;
+		details: Omit<IUploadDetails, 'size'>;
+	}): Promise<IUpload> {
+		const resolver = Promise.withResolvers<IUpload>();
+
+		const tempFilePath = UploadFS.getTempFilePath(Random.id());
+
+		const writeStream = fs.createWriteStream(tempFilePath);
+		streamParam.pipe(writeStream);
+
+		const cleanup = (err: unknown) => {
+			fs.promises.unlink(tempFilePath).catch(() => undefined);
+			resolver.reject(err);
+		};
+
+		writeStream.on('finish', async () => {
+			FileUpload.getStore('Uploads')
+				.insert(
+					{
+						...details,
+						size: writeStream.bytesWritten,
+					},
+					tempFilePath,
+				)
+				.then(resolver.resolve)
+				.catch(cleanup);
+		});
+
+		streamParam.on('error', async (err) => {
+			writeStream.destroy();
+			cleanup(err);
+		});
+
+		writeStream.on('error', cleanup);
+
+		return resolver.promise;
+	}
 }

ee/packages/federation-matrix/src/events/message.ts

@@ -44,9 +44,9 @@ async function handleMediaMessage(
 
 	const fileRefId = await MatrixMediaService.downloadAndStoreRemoteFile(url, matrixRoomId, {
 		name: messageBody,
-		size: fileInfo?.size,
-		type: mimeType,
-		roomId: room._id,
+		size: fileInfo?.size || 0,
+		type: mimeType || 'application/octet-stream',
+		rid: room._id,
 		userId: user._id,
 	});
 

ee/packages/federation-matrix/src/services/MatrixMediaService.ts

@@ -1,3 +1,4 @@
+import type { IUploadDetails } from '@rocket.chat/apps-engine/definition/uploads/IUploadDetails';
 import { Upload } from '@rocket.chat/core-services';
 import type { IUpload } from '@rocket.chat/core-typings';
 import { federationSDK } from '@rocket.chat/federation-sdk';
@@ -79,18 +80,7 @@ export class MatrixMediaService {
 		}
 	}
 
-	static async downloadAndStoreRemoteFile(
-		mxcUri: string,
-		matrixRoomId: string,
-		metadata: {
-			name: string;
-			size?: number;
-			type?: string;
-			messageId?: string;
-			roomId?: string;
-			userId?: string;
-		},
-	): Promise<string> {
+	static async downloadAndStoreRemoteFile(mxcUri: string, matrixRoomId: string, metadata: IUploadDetails): Promise<string> {
 		try {
 			const parts = this.parseMXCUri(mxcUri);
 			if (!parts) {
@@ -113,11 +103,8 @@ export class MatrixMediaService {
 				userId: metadata.userId || 'federation',
 				buffer,
 				details: {
-					name: metadata.name || 'unnamed',
+					...metadata,
 					size: buffer.length,
-					type: metadata.type || 'application/octet-stream',
-					rid: metadata.roomId,
-					userId: metadata.userId || 'federation',
 				},
 			});
 

ee/packages/network-broker/src/NetworkBroker.ts

@@ -1,3 +1,5 @@
+import Stream from 'stream';
+
 import { asyncLocalStorage } from '@rocket.chat/core-services';
 import type { CallingOptions, IBroker, IBrokerNode, IServiceMetrics, IServiceClass, EventSignatures } from '@rocket.chat/core-services';
 import { injectCurrentContext, tracerSpan } from '@rocket.chat/tracing';
@@ -39,8 +41,14 @@ export class NetworkBroker implements IBroker {
 
 		const context = asyncLocalStorage.getStore();
 
+		const stream = data?.[0]?.streamParam;
+		const isStreamingCall = !!stream;
+
 		if (context?.ctx?.call) {
-			return context.ctx.call(method, data, options);
+			return context.ctx.call(method, isStreamingCall ? stream : data, {
+				...options,
+				...(isStreamingCall && { meta: { ...((options as any)?.meta || {}), details: data[0].details } }),
+			});
 		}
 
 		const services: { name: string }[] = await this.broker.call('$node.services', {
@@ -51,10 +59,11 @@ export class NetworkBroker implements IBroker {
 			return new Error('method-not-available');
 		}
 
-		return this.broker.call(method, data, {
+		return this.broker.call(method, isStreamingCall ? stream : data, {
 			...options,
 			meta: {
 				optl: injectCurrentContext(),
+				...(isStreamingCall && { details: data?.[0]?.details }),
 			},
 		});
 	}
@@ -134,9 +143,12 @@ export class NetworkBroker implements IBroker {
 				continue;
 			}
 
-			service.actions[method] = async (ctx: Context<[], { optl?: unknown }>): Promise<any> => {
+			service.actions[method] = async (ctx: Context<[], { optl?: unknown; details?: unknown }>): Promise<any> => {
+				const isStreamingCall = Stream.isReadable(ctx.params as any);
+				const params = isStreamingCall ? [{ streamParam: ctx.params, details: ctx.meta?.details }] : ctx.params;
+
 				return tracerSpan(
-					`action ${name}:${method}`,
+					`action ${name}:${method}${isStreamingCall ? ' (streaming)' : ''}`,
 					{},
 					() => {
 						return asyncLocalStorage.run(
@@ -147,7 +159,7 @@ export class NetworkBroker implements IBroker {
 								broker: this,
 								ctx,
 							},
-							() => serviceInstance[method](...ctx.params),
+							() => serviceInstance[method](...params),
 						);
 					},
 					ctx.meta?.optl,

ee/packages/omnichannel-services/src/OmnichannelTranscript.ts

@@ -1,4 +1,4 @@
-import type { Readable } from 'stream';
+import { Readable } from 'stream';
 
 import {
 	ServiceClass,
@@ -241,8 +241,9 @@ export class OmnichannelTranscript extends ServiceClass implements IOmnichannelT
 				}
 
 				try {
-					const fileBuffer = await uploadService.getFileBuffer({ file: uploadedFile });
-					files.push({ name: file.name, buffer: fileBuffer, extension: uploadedFile.extension });
+					const stream = await uploadService.streamUploadedFile({ file: uploadedFile, imageResizeOpts: { width: 400, height: 240 } });
+
+					files.push({ name: file.name, buffer: await streamToBuffer(stream), extension: uploadedFile.extension });
 				} catch (err: unknown) {
 					this.log.error({ msg: 'Failed to fetch file buffer', err });
 					// Push empty buffer so parser processes this as "unsupported file"
@@ -363,13 +364,11 @@ export class OmnichannelTranscript extends ServiceClass implements IOmnichannelT
 		const transcriptText = i18n.t('Transcript');
 
 		const stream = await this.worker.renderToStream({ data, i18n });
-		const outBuff = await streamToBuffer(stream as Readable);
 
 		try {
 			const { rid } = await roomService.createDirectMessage({ to: details.userId, from: 'rocket.cat' });
 			const [rocketCatFile, transcriptFile] = await this.uploadFiles({
-				details,
-				buffer: outBuff,
+				streamParam: Readable.from(stream),
 				roomIds: [rid, details.rid],
 				data,
 				transcriptText,
@@ -406,23 +405,20 @@ export class OmnichannelTranscript extends ServiceClass implements IOmnichannelT
 	}
 
 	private async uploadFiles({
-		details,
-		buffer,
+		streamParam,
 		roomIds,
 		data,
 		transcriptText,
 	}: {
-		details: WorkDetailsWithSource;
-		buffer: Buffer;
+		streamParam: Readable;
 		roomIds: string[];
 		data: Pick<WorkerData, 'siteName' | 'visitor'>;
 		transcriptText: string;
 	}): Promise<IUpload[]> {
 		return Promise.all(
 			roomIds.map((roomId) => {
-				return uploadService.uploadFile({
-					userId: details.userId,
-					buffer,
+				return uploadService.uploadFileFromStream({
+					streamParam,
 					details: {
 						// transcript_{company-name}_{date}_{hour}.pdf
 						name: `${transcriptText}_${data.siteName}_${new Intl.DateTimeFormat('en-US').format(new Date()).replace(/\//g, '-')}_${
@@ -432,7 +428,6 @@ export class OmnichannelTranscript extends ServiceClass implements IOmnichannelT
 						rid: roomId,
 						// Rocket.cat is the goat
 						userId: 'rocket.cat',
-						size: buffer.length,
 					},
 				});
 			}),

packages/core-services/src/types/IUploadService.ts

@@ -1,10 +1,12 @@
+import type Stream from 'stream';
+
 import type { IUploadDetails } from '@rocket.chat/apps-engine/definition/uploads/IUploadDetails';
 import type { IMessage, IUpload, IUser, FilesAndAttachments } from '@rocket.chat/core-typings';
 
 export interface IUploadFileParams {
 	userId: string;
 	buffer: Buffer;
-	details: Partial<IUploadDetails>;
+	details: IUploadDetails;
 }
 export interface ISendFileMessageParams {
 	roomId: string;
@@ -29,4 +31,12 @@ export interface IUploadService {
 	parseFileIntoMessageAttachments(file: Partial<IUpload>, roomId: string, user: IUser): Promise<FilesAndAttachments>;
 	canDeleteFile(user: IUser, file: IUpload, msg: IMessage | null): Promise<boolean>;
 	deleteFile(user: IUser, fileId: IUpload['_id'], msg: IMessage | null): Promise<{ deletedFiles: IUpload['_id'][] }>;
+	streamUploadedFile({
+		file,
+		imageResizeOpts,
+	}: {
+		file: IUpload;
+		imageResizeOpts?: { width: number; height: number };
+	}): Promise<Stream.Readable>;
+	uploadFileFromStream({ streamParam, details }: { streamParam: Stream.Readable; details: Omit<IUploadDetails, 'size'> }): Promise<IUpload>;
 }