chore: auto-start pending file downloads with stall timeout

.changeset/deep-ways-poke.md

@@ -0,0 +1,7 @@
+---
+'@rocket.chat/core-typings': patch
+'@rocket.chat/models': patch
+'@rocket.chat/meteor': patch
+---
+
+Fixes the Slack importer storing shared files as raw URLs in the message body. Imported file messages now stay hidden until "Download Pending Files" button fetches them, then display as native attachments with image previews. Failed downloads (e.g. invalidated export links) are no longer silently saved as the file's content — they are counted as errors and can be retried.

apps/meteor/app/file-upload/server/lib/FileUpload.ts

@@ -5,7 +5,7 @@ import { unlink, rename, writeFile } from 'node:fs/promises';
 import type * as http from 'node:http';
 import type * as https from 'node:https';
 import stream from 'node:stream';
-import { finished } from 'node:stream/promises';
+import { pipeline } from 'node:stream/promises';
 import URL from 'node:url';
 import { isArrayBufferView } from 'node:util/types';
 
@@ -903,13 +903,16 @@ export class FileUploadClass {
 			} else if (isArrayBufferView(content)) {
 				await fs.promises.writeFile(tmpFile, content);
 			} else if (content instanceof stream.Readable) {
-				await finished(content.pipe(fs.createWriteStream(tmpFile)), { cleanup: true });
+				await pipeline(content, fs.createWriteStream(tmpFile));
 			} else {
 				throw new Error('Invalid file type');
 			}
 
-			return ufsComplete(fileId, this.name, { session: options?.session });
+			return await ufsComplete(fileId, this.name, { session: options?.session });
 		} catch (e) {
+			// The record is created before the content is written; drop it so a failed
+			// stream doesn't leave an orphaned upload record and tmp file behind.
+			await this.store.removeById(fileId, { session: options?.session }).catch(() => undefined);
 			throw e;
 		}
 	}

apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts

@@ -1,22 +1,20 @@
 import http from 'node:http';
 import https from 'node:https';
+import type { Readable } from 'node:stream';
 
 import { api } from '@rocket.chat/core-services';
-import type { IImport, MessageAttachment, IUpload, IImporterShortSelection } from '@rocket.chat/core-typings';
-import { Messages } from '@rocket.chat/models';
+import type { MessageAttachment, IUpload, IImporterShortSelection, IMessageWithPendingFileImport } from '@rocket.chat/core-typings';
+import { Messages, Users } from '@rocket.chat/models';
 import { Random } from '@rocket.chat/random';
 
 import { FileUpload } from '../../file-upload/server';
+import { parseFileIntoMessageAttachments } from '../../file-upload/server/methods/sendFileMessage';
 import { Importer, ProgressStep } from '../../importer/server';
-import type { ConverterOptions } from '../../importer/server/classes/ImportDataConverter';
 import type { ImporterProgress } from '../../importer/server/classes/ImporterProgress';
-import type { ImporterInfo } from '../../importer/server/definitions/ImporterInfo';
 
-export class PendingFileImporter extends Importer {
-	constructor(info: ImporterInfo, importRecord: IImport, converterOptions: ConverterOptions = {}) {
-		super(info, importRecord, converterOptions);
-	}
+const LEASE_MS = 2 * 60 * 1000;
 
+export class PendingFileImporter extends Importer {
 	async prepareFileCount() {
 		this.logger.debug('start preparing import operation');
 		await super.updateProgress(ProgressStep.PREPARING_STARTED);
@@ -37,141 +35,120 @@ export class PendingFileImporter extends Importer {
 		this.reportProgress();
 
 		setImmediate(() => {
-			void this.startImport({});
+			this.startImport({}).catch(() => undefined);
 		});
 
 		return fileCount;
 	}
 
-	override async startImport(importSelection: IImporterShortSelection): Promise<ImporterProgress> {
-		const downloadedFileIds: string[] = [];
-		const maxFileCount = 10;
-		const maxFileSize = 1024 * 1024 * 500;
+	override async startImport(_importSelection: IImporterShortSelection): Promise<ImporterProgress> {
+		const importedRoomIds = new Set<string>();
+		const inFlightFileIds = new Set<string>();
+		const skipMessageIds = new Set<string>();
 
-		let count = 0;
-		let currentSize = 0;
-		let nextSize = 0;
+		const processMessage = async (message: IMessageWithPendingFileImport) => {
+			const { _importFile } = message;
+			const url = _importFile.downloadUrl;
 
-		const waitForFiles = async () => {
-			if (count + 1 < maxFileCount && currentSize + nextSize < maxFileSize) {
+			if (!url?.startsWith('http')) {
+				skipMessageIds.add(message._id);
+				await this.addCountError(1);
 				return;
 			}
 
-			return new Promise<void>((resolve) => {
-				const handler = setInterval(() => {
-					if (count + 1 >= maxFileCount) {
-						return;
-					}
-
-					if (currentSize + nextSize >= maxFileSize && count > 0) {
-						return;
-					}
-
-					clearInterval(handler);
-					resolve();
-				}, 1000);
-			});
-		};
-
-		const completeFile = async (details: { size: number }) => {
-			await this.addCountCompleted(1);
-			count--;
-			currentSize -= details.size;
+			if (inFlightFileIds.has(_importFile.id)) {
+				skipMessageIds.add(message._id);
+				return;
+			}
+			inFlightFileIds.add(_importFile.id);
+
+			const details: { message_id: string; name: string; size: number; userId: string; rid: string; type?: string } = {
+				message_id: `${message._id}-file-${_importFile.id}`,
+				name: _importFile.name || Random.id(),
+				size: _importFile.size || 0,
+				userId: message.u._id,
+				rid: message.rid,
+			};
+
+			const renewal = setInterval(() => void Messages.renewPendingFileImportLease(message._id, LEASE_MS), LEASE_MS / 4);
+
+			try {
+				const fileStream = await this.downloadFile(url, details);
+
+				// Bypass the fileStore filters
+				const file = await FileUpload.getStore('Uploads')._doInsert(details, fileStream);
+
+				const rocketChatUrl = FileUpload.getPath(`${file._id}/${encodeURI(file.name || '')}`);
+				const user = await Users.findOneById(message.u._id);
+				const attachment = user
+					? (await parseFileIntoMessageAttachments(file, message.rid, user)).attachments[0]
+					: this.getMessageAttachment(file, rocketChatUrl);
+
+				const result = await Messages.setImportFileRocketChatAttachment(_importFile.id, rocketChatUrl, attachment);
+				await this.addCountCompleted('modifiedCount' in result ? result.modifiedCount : 1);
+				importedRoomIds.add(message.rid);
+			} catch (err) {
+				this.logger.error({ msg: 'Failed to download pending file', url: url.split('?')[0], err });
+				skipMessageIds.add(message._id);
+				inFlightFileIds.delete(_importFile.id);
+				await this.addCountError(1);
+			} finally {
+				clearInterval(renewal);
+			}
 		};
 
-		const logError = this.logger.error.bind(this.logger);
-
-		try {
-			const pendingFileMessageList = Messages.findAllImportedMessagesWithFilesToDownload();
-			const importedRoomIds = new Set<string>();
-			for await (const message of pendingFileMessageList) {
-				try {
-					const { _importFile } = message;
-
-					if (!_importFile || _importFile.downloaded || downloadedFileIds.includes(_importFile.id)) {
-						await this.addCountCompleted(1);
-						continue;
-					}
-
-					const url = _importFile.downloadUrl;
-					if (!url?.startsWith('http')) {
-						await this.addCountCompleted(1);
-						continue;
-					}
-
-					const details: { message_id: string; name: string; size: number; userId: string; rid: string; type?: string } = {
-						message_id: `${message._id}-file-${_importFile.id}`,
-						name: _importFile.name || Random.id(),
-						size: _importFile.size || 0,
-						userId: message.u._id,
-						rid: message.rid,
-					};
-
-					const requestModule = /https/i.test(url) ? https : http;
-					const fileStore = FileUpload.getStore('Uploads');
-
-					nextSize = details.size;
-					await waitForFiles();
-					count++;
-					currentSize += nextSize;
-					downloadedFileIds.push(_importFile.id);
-
-					requestModule.get(url, (res) => {
-						const contentType = res.headers['content-type'];
-						if (!details.type && contentType) {
-							details.type = contentType;
-						}
-
-						const rawData: Uint8Array[] = [];
-						res.on('data', (chunk) => {
-							rawData.push(chunk);
-
-							// Update progress more often on large files
-							this.reportProgress();
-						});
-						res.on('error', async (err) => {
-							await completeFile(details);
-							logError({ err });
-						});
-
-						res.on('end', async () => {
-							try {
-								// Bypass the fileStore filters
-								const file = await fileStore._doInsert(details, Buffer.concat(rawData));
-
-								const url = FileUpload.getPath(`${file._id}/${encodeURI(file.name || '')}`);
-								const attachment = this.getMessageAttachment(file, url);
-
-								await Messages.setImportFileRocketChatAttachment(_importFile.id, url, attachment);
-								await completeFile(details);
-								importedRoomIds.add(message.rid);
-							} catch (err) {
-								await completeFile(details);
-								logError({ err });
-							}
-						});
-					});
-				} catch (err) {
-					this.logger.error({ err });
+		const worker = async () => {
+			for (;;) {
+				const message = await Messages.findOneAndClaimPendingFileImport(LEASE_MS, Array.from(skipMessageIds));
+				if (!message) {
+					return;
 				}
-			}
 
-			void api.broadcast('notify.importedMessages', { roomIds: Array.from(importedRoomIds) });
-		} catch (error) {
-			// If the cursor expired, restart the method
-			if (this.isCursorNotFoundError(error)) {
-				this.logger.info('CursorNotFound');
-				return this.startImport(importSelection);
+				await processMessage(message);
 			}
+		};
 
+		const results = await Promise.allSettled(Array.from({ length: 10 }, () => worker()));
+
+		const crashes = results.filter((result): result is PromiseRejectedResult => result.status === 'rejected');
+		if (crashes.length > 0) {
+			crashes.forEach((crash) => this.logger.error({ msg: 'Pending file worker crashed', err: crash.reason }));
 			await super.updateProgress(ProgressStep.ERROR);
-			throw error;
+			throw crashes[0].reason;
 		}
 
+		void api.broadcast('notify.importedMessages', { roomIds: Array.from(importedRoomIds) });
+
 		await super.updateProgress(ProgressStep.DONE);
 		return this.getProgress();
 	}
 
+	private downloadFile(url: string, details: { type?: string }): Promise<Readable> {
+		return new Promise((resolve, reject) => {
+			const requestModule = /https/i.test(url) ? https : http;
+
+			const request = requestModule.get(url, (res) => {
+				if (!res.statusCode || res.statusCode < 200 || res.statusCode >= 300) {
+					// Error responses (e.g. a redirect to a login page) would otherwise be saved as the file's content.
+					res.on('error', () => undefined); // already rejected; just avoid crashing while draining
+					res.resume(); // discard the body and free the socket
+					reject(new Error(`Unexpected response status ${res.statusCode}`));
+					return;
+				}
+
+				if (!details.type) {
+					details.type = res.headers['content-type'] || 'application/octet-stream';
+				}
+
+				resolve(res);
+			});
+
+			request.on('error', reject);
+
+			request.setTimeout(60 * 1000, () => request.destroy(new Error('Download stalled')));
+		});
+	}
+
 	getMessageAttachment(file: IUpload, url: string): MessageAttachment {
 		if (file.type) {
 			if (/^image\/.+/.test(file.type)) {

apps/meteor/app/importer-slack/server/SlackImporter.ts

@@ -1,9 +1,11 @@
+import { Import } from '@rocket.chat/core-services';
 import type { IImportUser, IImportMessage, IImportPendingFile } from '@rocket.chat/core-typings';
 import { Messages, Settings, ImportData } from '@rocket.chat/models';
 import type { IZipEntry } from 'adm-zip';
 
-import { Importer, ProgressStep, ImporterWebsocket } from '../../importer/server';
+import { Importer, ProgressStep, ImporterWebsocket, Importers } from '../../importer/server';
 import type { ImporterProgress } from '../../importer/server/classes/ImporterProgress';
+import { PendingFileImporter } from '../../importer-pending-files/server/PendingFileImporter';
 import { notifyOnSettingChanged } from '../../lib/server/lib/notifyListener';
 import { MentionsParser } from '../../mentions/lib/MentionsParser';
 import { settings } from '../../settings/server';
@@ -119,6 +121,21 @@ type SlackAttachment = {
 export class SlackImporter extends Importer {
 	private _useUpsert = false;
 
+	protected override async onImportComplete(): Promise<void> {
+		const pendingFilesImporter = Importers.get('pending-files');
+		if (!pendingFilesImporter) {
+			return;
+		}
+
+		if ((await Messages.countAllImportedMessagesWithFilesToDownload()) === 0) {
+			return;
+		}
+
+		const operation = await Import.newOperation(this.importRecord.user, pendingFilesImporter.name, pendingFilesImporter.key);
+		const instance = new PendingFileImporter(pendingFilesImporter, operation);
+		await instance.prepareFileCount();
+	}
+
 	async prepareChannelsFile(entry: IZipEntry): Promise<number> {
 		await super.updateProgress(ProgressStep.PREPARING_CHANNELS);
 		const data = (JSON.parse(entry.getData().toString()) as SlackChannel[]).filter(
@@ -496,7 +513,8 @@ export class SlackImporter extends Importer {
 						_id: fileId,
 						rid: newMessage.rid,
 						ts: newMessage.ts,
-						msg: message.file.url_private_download || '',
+						msg: '',
+						_hidden: true,
 						_importFile: this.convertSlackFileToPendingFile(message.file),
 						u: {
 							_id: newMessage.u._id,
@@ -568,7 +586,8 @@ export class SlackImporter extends Importer {
 						_id: fileId,
 						rid: slackChannelId,
 						ts: newMessage.ts,
-						msg: file.url_private_download || '',
+						msg: '',
+						_hidden: true,
 						_importFile: this.convertSlackFileToPendingFile(file),
 						u: {
 							_id: this._replaceSlackUserId(message.user),

apps/meteor/app/importer/server/classes/Importer.ts

@@ -226,6 +226,14 @@ export class Importer {
 				await this.applySettingValues(this.oldSettings);
 			}
 
+			if (this.progress.step === ProgressStep.DONE) {
+				try {
+					await this.onImportComplete?.();
+				} catch (err) {
+					this.logger.error({ msg: 'Failed to run the post-import step', err });
+				}
+			}
+
 			const timeTook = Date.now() - started;
 			this.logger.log({ msg: 'Import completed', durationMs: timeTook });
 		});
@@ -268,6 +276,8 @@ export class Importer {
 		return this.progress;
 	}
 
+	protected onImportComplete?(): Promise<void>;
+
 	/**
 	 * Updates the progress step of this importer.
 	 * It also changes some internal settings at various stages of the import.

apps/meteor/app/importer/server/classes/converters/MessageConverter.ts

@@ -104,6 +104,7 @@ export class MessageConverter extends RecordConverter<IImportMessageRecord> {
 			mentions,
 			channels,
 			_importFile: data._importFile,
+			_hidden: data._hidden,
 			url: data.url,
 			attachments: data.attachments,
 			bot: data.bot,

packages/core-typings/src/IMessage/IMessage.ts

@@ -403,6 +403,7 @@ export type IMessageWithPendingFileImport = IMessage & {
 		original: Record<string, any>;
 		rocketChatUrl?: string;
 		downloaded?: boolean;
+		lockedUntil?: Date;
 	};
 };
 

packages/core-typings/src/import/IImportMessage.ts

@@ -48,4 +48,5 @@ export interface IImportMessage {
 
 	url?: string;
 	_importFile?: IImportPendingFile;
+	_hidden?: boolean;
 }