feat: OEmbed async processing

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

@@ -1,3 +1,4 @@
+import { Message } from '@rocket.chat/core-services';
 import type { IMessage, IThreadMainMessage } from '@rocket.chat/core-typings';
 import { MessageTypes } from '@rocket.chat/message-types';
 import { Messages, Users, Rooms, Subscriptions } from '@rocket.chat/models';
@@ -48,7 +49,6 @@ import { executeUpdateMessage } from '../../../lib/server/methods/updateMessage'
 import { applyAirGappedRestrictionsValidation } from '../../../license/server/airGappedRestrictionsWrapper';
 import { pinMessage, unpinMessage } from '../../../message-pin/server/pinMessage';
 import { starMessage } from '../../../message-star/server/starMessage';
-import { OEmbed } from '../../../oembed/server/server';
 import { executeSetReaction } from '../../../reactions/server/setReaction';
 import { settings } from '../../../settings/server';
 import { followMessage } from '../../../threads/server/methods/followMessage';
@@ -914,7 +914,7 @@ API.v1.addRoute(
 				throw new Meteor.Error('error-not-allowed', 'Not allowed');
 			}
 
-			const { urlPreview } = await OEmbed.parseUrl(url);
+			const { urlPreview } = await Message.parseOEmbedUrl(url);
 			urlPreview.ignoreParse = true;
 
 			return API.v1.success({ urlPreview });

apps/meteor/app/lib/server/functions/sendMessage.ts

@@ -11,7 +11,7 @@ import { hasPermissionAsync } from '../../../authorization/server/functions/hasP
 import { FileUpload } from '../../../file-upload/server';
 import { settings } from '../../../settings/server';
 import { afterSaveMessage } from '../lib/afterSaveMessage';
-import { notifyOnRoomChangedById, notifyOnMessageChange } from '../lib/notifyListener';
+import { notifyOnRoomChangedById } from '../lib/notifyListener';
 import { validateCustomMessageFields } from '../lib/validateCustomMessageFields';
 
 // TODO: most of the types here are wrong, but I don't want to change them now
@@ -286,11 +286,8 @@ export const sendMessage = async function (user: any, message: any, room: any, u
 		void Apps.getBridges()?.getListenerBridge().messageEvent(messageEvent, message);
 	}
 
-	// TODO: is there an opportunity to send returned data to notifyOnMessageChange?
 	await afterSaveMessage(message, room, user);
 
-	void notifyOnMessageChange({ id: message._id });
-
 	void notifyOnRoomChangedById(message.rid);
 
 	return message;

apps/meteor/app/lib/server/functions/updateMessage.ts

@@ -7,7 +7,7 @@ import { Meteor } from 'meteor/meteor';
 import { parseUrlsInMessage } from './parseUrlsInMessage';
 import { settings } from '../../../settings/server';
 import { afterSaveMessage } from '../lib/afterSaveMessage';
-import { notifyOnRoomChangedById, notifyOnMessageChange } from '../lib/notifyListener';
+import { notifyOnRoomChangedById } from '../lib/notifyListener';
 import { validateCustomMessageFields } from '../lib/validateCustomMessageFields';
 
 export const updateMessage = async function (
@@ -97,14 +97,7 @@ export const updateMessage = async function (
 			return;
 		}
 
-		// although this is an "afterSave" kind callback, we know they can extend message's properties
-		// so we wait for it to run before broadcasting
-		const data = await afterSaveMessage(msg, room, user);
-
-		void notifyOnMessageChange({
-			id: msg._id,
-			data,
-		});
+		await afterSaveMessage(msg, room, user);
 
 		if (room?.lastMessage?._id === msg._id) {
 			void notifyOnRoomChangedById(message.rid);

apps/meteor/app/lib/server/lib/afterSaveMessage.ts

@@ -1,3 +1,4 @@
+import { Message } from '@rocket.chat/core-services';
 import type { IMessage, IUser, IRoom } from '@rocket.chat/core-typings';
 import type { Updater } from '@rocket.chat/models';
 import { Rooms } from '@rocket.chat/models';
@@ -6,14 +7,16 @@ import { callbacks } from '../../../../server/lib/callbacks';
 
 export async function afterSaveMessage(message: IMessage, room: IRoom, user: IUser, roomUpdater?: Updater<IRoom>): Promise<IMessage> {
 	const updater = roomUpdater ?? Rooms.getUpdater();
-	const data = await callbacks.run('afterSaveMessage', message, { room, user, roomUpdater: updater });
+	const data: IMessage = (await callbacks.run('afterSaveMessage', message, { room, user, roomUpdater: updater })) as unknown as IMessage;
 
 	if (!roomUpdater && updater.hasChanges()) {
 		await Rooms.updateFromUpdater({ _id: room._id }, updater);
 	}
 
+	void Message.afterSave({ message: data });
+
 	// TODO: Fix type - callback configuration needs to be updated
-	return data as unknown as IMessage;
+	return data;
 }
 
 export function afterSaveMessageAsync(message: IMessage, room: IRoom, user: IUser, roomUpdater: Updater<IRoom> = Rooms.getUpdater()): void {
@@ -22,4 +25,6 @@ export function afterSaveMessageAsync(message: IMessage, room: IRoom, user: IUse
 	if (roomUpdater.hasChanges()) {
 		void Rooms.updateFromUpdater({ _id: room._id }, roomUpdater);
 	}
+
+	void Message.afterSave({ message });
 }

apps/meteor/server/importPackages.ts

@@ -42,7 +42,6 @@ import '../app/message-pin/server';
 import '../app/message-star/server';
 import '../app/nextcloud/server';
 import '../app/oauth2-server-config/server';
-import '../app/oembed/server';
 import '../app/push-notifications/server';
 import '../app/retention-policy/server';
 import '../app/slackbridge/server';

apps/meteor/server/lib/callbacks.ts

@@ -9,8 +9,6 @@ import type {
 	ILivechatInquiryRecord,
 	ILivechatVisitor,
 	VideoConference,
-	OEmbedMeta,
-	OEmbedUrlContent,
 	IOmnichannelRoom,
 	ILivechatTag,
 	ILivechatTagRecord,
@@ -164,16 +162,6 @@ type ChainedCallbackSignatures = {
 		BusinessHourBehaviorClass: { new (): IBusinessHourBehavior };
 	};
 	'renderMessage': <T extends IMessage & { html: string }>(message: T) => T;
-	'oembed:beforeGetUrlContent': (data: { urlObj: URL }) => {
-		urlObj: URL;
-		headerOverrides?: { [k: string]: string };
-	};
-	'oembed:afterParseContent': (data: { url: string; meta: OEmbedMeta; headers: { [k: string]: string }; content: OEmbedUrlContent }) => {
-		url: string;
-		meta: OEmbedMeta;
-		headers: { [k: string]: string };
-		content: OEmbedUrlContent;
-	};
 	'livechat.beforeListTags': () => ILivechatTag[];
 	'livechat.offlineMessage': (data: { name: string; email: string; message: string; department?: string; host?: string }) => void;
 	'livechat.leadCapture': (room: IOmnichannelRoom) => IOmnichannelRoom;

apps/meteor/app/oembed/server/server.ts → apps/meteor/server/services/messages/hooks/AfterSaveOEmbed.ts

@@ -1,25 +1,26 @@
 import type {
 	OEmbedUrlContentResult,
+	MessageUrl,
 	OEmbedUrlWithMetadata,
+	OEmbedMeta,
 	IMessage,
 	MessageAttachment,
-	OEmbedMeta,
-	MessageUrl,
+	OEmbedUrlContent,
 } from '@rocket.chat/core-typings';
 import { isOEmbedUrlWithMetadata } from '@rocket.chat/core-typings';
 import { Logger } from '@rocket.chat/logger';
-import { Messages, OEmbedCache } from '@rocket.chat/models';
+import { OEmbedCache, Messages } from '@rocket.chat/models';
 import { serverFetch as fetch } from '@rocket.chat/server-fetch';
-import { camelCase } from 'change-case';
 import he from 'he';
 import iconv from 'iconv-lite';
 import ipRangeCheck from 'ip-range-check';
 import jschardet from 'jschardet';
+import { camelCase } from 'lodash';
 
-import { isURL } from '../../../lib/utils/isURL';
-import { callbacks } from '../../../server/lib/callbacks';
-import { settings } from '../../settings/server';
-import { Info } from '../../utils/rocketchat.info';
+import { settings } from '../../../../app/settings/server';
+import { Info } from '../../../../app/utils/rocketchat.info';
+import { isURL } from '../../../../lib/utils/isURL';
+import { afterParseUrlContent, beforeGetUrlContent } from '../lib/oembed/providers';
 
 const MAX_EXTERNAL_URL_PREVIEWS = 5;
 const log = new Logger('OEmbed');
@@ -65,7 +66,7 @@ const toUtf8 = function (contentType: string, body: Buffer): string {
 	return iconv.decode(body, getCharset(contentType, body));
 };
 
-const getUrlContent = async (urlObj: URL, redirectCount = 5): Promise<OEmbedUrlContentResult> => {
+const getUrlContent = async (urlObj: URL, redirectCount = 5): Promise<OEmbedUrlContent> => {
 	const portsProtocol = new Map<string, string>(
 		Object.entries({
 			80: 'http:',
@@ -75,8 +76,40 @@ const getUrlContent = async (urlObj: URL, redirectCount = 5): Promise<OEmbedUrlC
 	);
 
 	const ignoredHosts = settings.get<string>('API_EmbedIgnoredHosts').replace(/\s/g, '').split(',') || [];
-	if (urlObj.hostname && (ignoredHosts.includes(urlObj.hostname) || ipRangeCheck(urlObj.hostname, ignoredHosts))) {
-		throw new Error('invalid host');
+
+	const isIgnoredHost = (hostname: string | null): boolean => {
+		if (!hostname || !ignoredHosts.length) {
+			return false;
+		}
+
+		const exactHosts = ignoredHosts.filter((h) => !h.includes('*'));
+		if (exactHosts.includes(hostname) || ipRangeCheck(hostname, exactHosts)) {
+			return true;
+		}
+
+		return ignoredHosts
+			.filter((h) => h.includes('*'))
+			.some((pattern) => {
+				const validationRegex = /^(?:\*\.)?(?:\*|[a-z0-9-]+)(?:\.(?:\*|[a-z0-9-]+))*$/i;
+				if (!validationRegex.test(pattern) || pattern === '*') {
+					return false;
+				}
+
+				const escaped = pattern.replace(/[-/\\^$+?.()|[\]{}]/g, '\\$&');
+				const source = `^${escaped.replace(/\*/g, '[^.]*')}$`;
+
+				try {
+					const regex = new RegExp(source, 'i');
+					return regex.test(hostname);
+				} catch {
+					// fail safe on invalid patterns
+					return false;
+				}
+			});
+	};
+
+	if (isIgnoredHost(urlObj.hostname)) {
+		throw new Error('host is ignored');
 	}
 
 	const safePorts = settings.get<string>('API_EmbedSafePorts').replace(/\s/g, '').split(',') || [];
@@ -91,14 +124,13 @@ const getUrlContent = async (urlObj: URL, redirectCount = 5): Promise<OEmbedUrlC
 		throw new Error('invalid/unsafe port');
 	}
 
-	const data = await callbacks.run('oembed:beforeGetUrlContent', {
-		urlObj,
-	});
+	const data = beforeGetUrlContent({ urlObj });
 
 	const url = data.urlObj.toString();
 	const sizeLimit = 250000;
 
 	log.debug({ msg: 'Fetching URL for OEmbed', url, redirectCount });
+	const start = Date.now();
 	const response = await fetch(
 		url,
 		{
@@ -109,10 +141,12 @@ const getUrlContent = async (urlObj: URL, redirectCount = 5): Promise<OEmbedUrlC
 				'Accept-Language': settings.get('Language') || 'en',
 				...data.headerOverrides,
 			},
+			timeout: settings.get<number>('API_EmbedTimeout') * 1000,
 			size: sizeLimit, // max size of the response body, this was not working as expected so I'm also manually verifying that on the iterator
 		},
 		settings.get('Allow_Invalid_SelfSigned_Certs'),
 	);
+	const end = Date.now();
 
 	let totalSize = 0;
 	const chunks = [];
@@ -126,13 +160,14 @@ const getUrlContent = async (urlObj: URL, redirectCount = 5): Promise<OEmbedUrlC
 		}
 	}
 
-	log.debug({ msg: 'Obtained response from server', length: totalSize });
+	log.debug({ msg: 'Obtained response from server', length: totalSize, timeSpent: `${end - start}ms` });
 	const buffer = Buffer.concat(chunks);
 
 	return {
 		headers: Object.fromEntries(response.headers),
 		body: toUtf8(response.headers.get('content-type') || 'text/plain', buffer),
 		statusCode: response.status,
+		urlObj,
 	};
 };
 
@@ -183,7 +218,7 @@ const getUrlMeta = async function (
 	}
 
 	log.debug({ msg: 'Fetching URL content', url: urlObj.toString() });
-	let content: OEmbedUrlContentResult | undefined;
+	let content: OEmbedUrlContent | undefined;
 	try {
 		content = await getUrlContent(urlObj, 5);
 	} catch (err) {
@@ -195,7 +230,7 @@ const getUrlMeta = async function (
 	}
 
 	log.debug({ msg: 'Parsing metadata for URL', url });
-	const metas: { [k: string]: string } = {};
+	const metas: OEmbedMeta = {} as any;
 
 	if (content?.body) {
 		const escapeMeta = (name: string, value: string): string => {
@@ -233,7 +268,7 @@ const getUrlMeta = async function (
 	if (content && content.statusCode !== 200) {
 		return;
 	}
-	return callbacks.run('oembed:afterParseContent', {
+	return afterParseUrlContent({
 		url,
 		meta: metas,
 		headers,
@@ -289,6 +324,10 @@ const insertMaxWidthInOembedHtml = (oembedHtml?: string): string | undefined =>
 const rocketUrlParser = async function (message: IMessage): Promise<IMessage> {
 	log.debug({ msg: 'Parsing message URLs' });
 
+	if (!settings.get('API_Embed')) {
+		return message;
+	}
+
 	if (!Array.isArray(message.urls)) {
 		return message;
 	}
@@ -318,6 +357,7 @@ const rocketUrlParser = async function (message: IMessage): Promise<IMessage> {
 		changed = changed || foundMeta;
 	}
 
+	// Why we have this here if there's no code for adding anything to the attachments array?
 	if (attachments.length) {
 		await Messages.setMessageAttachments(message._id, attachments);
 	}
@@ -326,26 +366,16 @@ const rocketUrlParser = async function (message: IMessage): Promise<IMessage> {
 		await Messages.setUrlsById(message._id, message.urls);
 	}
 
-	return message;
+	return {
+		...message,
+		...(attachments.length && { attachments }),
+	};
 };
 
-const OEmbed: {
-	getUrlMeta: (url: string, withFragment?: boolean) => Promise<OEmbedUrlWithMetadata | undefined | OEmbedUrlContentResult>;
-	getUrlMetaWithCache: (url: string, withFragment?: boolean) => Promise<OEmbedUrlWithMetadata | OEmbedUrlContentResult | undefined>;
+export const OEmbed: {
 	rocketUrlParser: (message: IMessage) => Promise<IMessage>;
 	parseUrl: (url: string) => Promise<{ urlPreview: MessageUrl; foundMeta: boolean }>;
 } = {
 	rocketUrlParser,
-	getUrlMetaWithCache,
-	getUrlMeta,
 	parseUrl,
 };
-
-settings.watch('API_Embed', (value) => {
-	if (value) {
-		return callbacks.add('afterSaveMessage', (message) => OEmbed.rocketUrlParser(message), callbacks.priority.LOW, 'API_Embed');
-	}
-	return callbacks.remove('afterSaveMessage', 'API_Embed');
-});
-
-export { OEmbed };