diff --git a/.changeset/deep-ways-poke.md b/.changeset/deep-ways-poke.md new file mode 100644 index 0000000000000..918b70379880e --- /dev/null +++ b/.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. diff --git a/apps/meteor/app/file-upload/server/lib/FileUpload.ts b/apps/meteor/app/file-upload/server/lib/FileUpload.ts index 3acc63bbb79cf..4b9a474c5afd7 100644 --- a/apps/meteor/app/file-upload/server/lib/FileUpload.ts +++ b/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; } } diff --git a/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts b/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts index 35b88c4267138..c3434420d10e2 100644 --- a/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts +++ b/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts @@ -1,22 +1,21 @@ -import http from 'node:http'; -import https from 'node:https'; +import { Transform, type Readable } from 'node:stream'; +import { pipeline } from 'node:stream/promises'; 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 { serverFetch } from '@rocket.chat/server-fetch'; 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'; +import { settings } from '../../settings/server'; -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 +36,128 @@ export class PendingFileImporter extends Importer { this.reportProgress(); setImmediate(() => { - void this.startImport({}); + this.startImport({}).catch(() => undefined); }); return fileCount; } - override async startImport(importSelection: IImporterShortSelection): Promise { - const downloadedFileIds: string[] = []; - const maxFileCount = 10; - const maxFileSize = 1024 * 1024 * 500; + override async startImport(_importSelection: IImporterShortSelection): Promise { + const importedRoomIds = new Set(); + const inFlightFileIds = new Set(); + const skipMessageIds = new Set(); - 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((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); + + let file: IUpload | undefined; + try { + const fileStream = await this.downloadFile(url, details); + + // Bypass the fileStore filters + 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 }); + if (file) { + await FileUpload.getStore('Uploads') + .deleteById(file._id) + .catch(() => undefined); + } + 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(); - 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 async downloadFile(url: string, details: { type?: string }): Promise { + const response = await serverFetch(url, { ignoreSsrfValidation: false, allowList: settings.get('SSRF_Allowlist') }); + + if (!response.ok) { + response.body.resume(); + throw new Error(`Unexpected response status ${response.status}`); + } + + if (!details.type) { + details.type = response.headers.get('content-type') || 'application/octet-stream'; + } + + // serverFetch only times out connecting; fail the download if the body goes idle for 60s. + // Per-chunk reset (not a total cap) so large files aren't cut off mid-download. + let idle: ReturnType; + const meter = new Transform({ + transform(chunk, _encoding, callback) { + idle.refresh(); + callback(null, chunk); + }, + }); + idle = setTimeout(() => meter.destroy(new Error('Download stalled')), 60 * 1000); + meter.on('close', () => clearTimeout(idle)); + + void pipeline(response.body, meter).catch(() => undefined); + return meter; + } + getMessageAttachment(file: IUpload, url: string): MessageAttachment { if (file.type) { if (/^image\/.+/.test(file.type)) { diff --git a/apps/meteor/app/importer-slack/server/SlackImporter.ts b/apps/meteor/app/importer-slack/server/SlackImporter.ts index d7ccfbdec6c49..c2b1511b7345d 100644 --- a/apps/meteor/app/importer-slack/server/SlackImporter.ts +++ b/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 { + 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 { 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), diff --git a/apps/meteor/app/importer/server/classes/Importer.ts b/apps/meteor/app/importer/server/classes/Importer.ts index c39da71ed2a31..c7a14778714f3 100644 --- a/apps/meteor/app/importer/server/classes/Importer.ts +++ b/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; + /** * Updates the progress step of this importer. * It also changes some internal settings at various stages of the import. diff --git a/apps/meteor/app/importer/server/classes/converters/MessageConverter.ts b/apps/meteor/app/importer/server/classes/converters/MessageConverter.ts index a8f785bdf17c7..7bf6a3e4e8746 100644 --- a/apps/meteor/app/importer/server/classes/converters/MessageConverter.ts +++ b/apps/meteor/app/importer/server/classes/converters/MessageConverter.ts @@ -104,6 +104,7 @@ export class MessageConverter extends RecordConverter { mentions, channels, _importFile: data._importFile, + _hidden: data._hidden, url: data.url, attachments: data.attachments, bot: data.bot, diff --git a/packages/core-typings/src/IMessage/IMessage.ts b/packages/core-typings/src/IMessage/IMessage.ts index a180c7d7528fa..335815d200ff1 100644 --- a/packages/core-typings/src/IMessage/IMessage.ts +++ b/packages/core-typings/src/IMessage/IMessage.ts @@ -403,6 +403,7 @@ export type IMessageWithPendingFileImport = IMessage & { original: Record; rocketChatUrl?: string; downloaded?: boolean; + lockedUntil?: Date; }; }; diff --git a/packages/core-typings/src/import/IImportMessage.ts b/packages/core-typings/src/import/IImportMessage.ts index 59dd231557096..0f4aa0d924685 100644 --- a/packages/core-typings/src/import/IImportMessage.ts +++ b/packages/core-typings/src/import/IImportMessage.ts @@ -48,4 +48,5 @@ export interface IImportMessage { url?: string; _importFile?: IImportPendingFile; + _hidden?: boolean; } diff --git a/packages/model-typings/src/models/IMessagesModel.ts b/packages/model-typings/src/models/IMessagesModel.ts index 0d16750f012c8..fd37184682fa6 100644 --- a/packages/model-typings/src/models/IMessagesModel.ts +++ b/packages/model-typings/src/models/IMessagesModel.ts @@ -281,7 +281,8 @@ export interface IMessagesModel extends IBaseModel { setAsReadById(_id: string): Promise; countThreads(): Promise; addThreadFollowerByThreadId(tmid: string, userId: string): Promise; - findAllImportedMessagesWithFilesToDownload(): FindCursor; + findOneAndClaimPendingFileImport(leaseMs: number, excludedMessageIds: IMessage['_id'][]): Promise; + renewPendingFileImportLease(messageId: IMessage['_id'], leaseMs: number): Promise; countAllImportedMessagesWithFilesToDownload(): Promise; findAgentLastMessageByVisitorLastMessageTs(roomId: string, visitorLastMessageTs: Date): Promise; removeThreadFollowerByThreadId(tmid: string, userId: string): Promise; diff --git a/packages/models/src/models/Messages.ts b/packages/models/src/models/Messages.ts index 0335b38c9d08e..073eaa7e1ef24 100644 --- a/packages/models/src/models/Messages.ts +++ b/packages/models/src/models/Messages.ts @@ -657,19 +657,18 @@ export class MessagesRaw extends BaseRaw implements IMessagesModel { rocketChatUrl: string, attachment: MessageAttachment, ): Promise { - const query = { - '_importFile.id': importFileId, - }; - - return this.updateMany(query, { - $set: { - '_importFile.rocketChatUrl': rocketChatUrl, - '_importFile.downloaded': true, - }, - $addToSet: { - attachments: attachment, + return this.updateMany( + { '_importFile.id': importFileId, '_importFile.downloaded': { $ne: true } }, + { + $set: { + '_hidden': false, + '_importFile.rocketChatUrl': rocketChatUrl, + '_importFile.downloaded': true, + }, + $unset: { '_importFile.lockedUntil': 1 }, + $addToSet: { attachments: attachment }, }, - }); + ); } countVisibleByRoomIdBetweenTimestampsInclusive(roomId: string, afterTimestamp: Date, beforeTimestamp: Date): Promise { @@ -1719,42 +1718,38 @@ export class MessagesRaw extends BaseRaw implements IMessagesModel { return this.findOne(query, { sort: { ts: 1 } }); } - findAllImportedMessagesWithFilesToDownload(): FindCursor { - const query = { - '_importFile.downloadUrl': { - $exists: true, - }, - '_importFile.rocketChatUrl': { - $exists: false, - }, - '_importFile.downloaded': { - $ne: true, - }, - '_importFile.external': { - $ne: true, - }, - }; + findOneAndClaimPendingFileImport(leaseMs: number, excludedMessageIds: IMessage['_id'][]): Promise { + const now = new Date(); - return this.find(query); + return this.col.findOneAndUpdate( + { + '_id': { $nin: excludedMessageIds }, + '_importFile.downloadUrl': { $exists: true }, + '_importFile.rocketChatUrl': { $exists: false }, + '_importFile.downloaded': { $ne: true }, + '_importFile.external': { $ne: true }, + '$or': [{ '_importFile.lockedUntil': { $exists: false } }, { '_importFile.lockedUntil': { $lt: now } }], + }, + { $set: { '_importFile.lockedUntil': new Date(now.getTime() + leaseMs) } }, + { returnDocument: 'after' }, + ) as Promise; } - countAllImportedMessagesWithFilesToDownload(): Promise { - const query = { - '_importFile.downloadUrl': { - $exists: true, - }, - '_importFile.rocketChatUrl': { - $exists: false, - }, - '_importFile.downloaded': { - $ne: true, - }, - '_importFile.external': { - $ne: true, - }, - }; + renewPendingFileImportLease(messageId: IMessage['_id'], leaseMs: number): Promise { + // The downloaded guard keeps a renewal racing the finalize from re-adding a stale lock. + return this.updateOne( + { '_id': messageId, '_importFile.downloaded': { $ne: true } }, + { $set: { '_importFile.lockedUntil': new Date(Date.now() + leaseMs) } }, + ); + } - return this.countDocuments(query); + countAllImportedMessagesWithFilesToDownload(): Promise { + return this.countDocuments({ + '_importFile.downloadUrl': { $exists: true }, + '_importFile.rocketChatUrl': { $exists: false }, + '_importFile.downloaded': { $ne: true }, + '_importFile.external': { $ne: true }, + }); } decreaseReplyCountById(_id: string, inc = -1): Promise {