refactor: EmailInbox out of DB Watcher (#32501)

Co-authored-by: Diego Sampaio <chinello@gmail.com>

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

@@ -1,6 +1,8 @@
 import type { IEmailInbox } from '@rocket.chat/core-typings';
 import { EmailInbox, Users } from '@rocket.chat/models';
-import type { Filter, InsertOneResult, Sort, UpdateResult, WithId } from 'mongodb';
+import type { DeleteResult, Filter, InsertOneResult, Sort } from 'mongodb';
+
+import { notifyOnEmailInboxChanged } from '../../../lib/server/lib/notifyListener';
 
 export const findEmailInboxes = async ({
 	query = {},
@@ -34,33 +36,31 @@ export const findEmailInboxes = async ({
 	};
 };
 
-export const findOneEmailInbox = async ({ _id }: { _id: string }): Promise<IEmailInbox | null> => {
-	return EmailInbox.findOneById(_id);
-};
 export const insertOneEmailInbox = async (
 	userId: string,
 	emailInboxParams: Pick<IEmailInbox, 'active' | 'name' | 'email' | 'description' | 'senderInfo' | 'department' | 'smtp' | 'imap'>,
-): Promise<InsertOneResult<WithId<IEmailInbox>>> => {
+): Promise<InsertOneResult<IEmailInbox>> => {
 	const obj = {
 		...emailInboxParams,
 		_createdAt: new Date(),
 		_updatedAt: new Date(),
 		_createdBy: await Users.findOneById(userId, { projection: { username: 1 } }),
 	};
-	return EmailInbox.insertOne(obj);
+
+	const response = await EmailInbox.create(obj);
+
+	if (response.insertedId) {
+		void notifyOnEmailInboxChanged({ _id: response.insertedId, ...obj }, 'inserted');
+	}
+
+	return response;
 };
 
 export const updateEmailInbox = async (
 	emailInboxParams: Pick<IEmailInbox, '_id' | 'active' | 'name' | 'email' | 'description' | 'senderInfo' | 'department' | 'smtp' | 'imap'>,
-): Promise<InsertOneResult<WithId<IEmailInbox>> | UpdateResult> => {
+): Promise<Pick<IEmailInbox, '_id'> | null> => {
 	const { _id, active, name, email, description, senderInfo, department, smtp, imap } = emailInboxParams;
 
-	const emailInbox = await findOneEmailInbox({ _id });
-
-	if (!emailInbox) {
-		throw new Error('error-invalid-email-inbox');
-	}
-
 	const updateEmailInbox = {
 		$set: {
 			active,
@@ -76,5 +76,29 @@ export const updateEmailInbox = async (
 		...(department === 'All' && { $unset: { department: 1 as const } }),
 	};
 
-	return EmailInbox.updateOne({ _id }, updateEmailInbox);
+	const updatedResponse = await EmailInbox.updateById(_id, updateEmailInbox);
+
+	if (!updatedResponse.value) {
+		throw new Error('error-invalid-email-inbox');
+	}
+
+	void notifyOnEmailInboxChanged(
+		{
+			...updatedResponse.value,
+			...(department === 'All' && { department: undefined }),
+		},
+		'updated',
+	);
+
+	return updatedResponse.value;
+};
+
+export const removeEmailInbox = async (emailInboxId: IEmailInbox['_id']): Promise<DeleteResult> => {
+	const removeResponse = await EmailInbox.removeById(emailInboxId);
+
+	if (removeResponse.deletedCount) {
+		void notifyOnEmailInboxChanged({ _id: emailInboxId }, 'removed');
+	}
+
+	return removeResponse;
 };

apps/meteor/app/api/server/v1/email-inbox.ts

@@ -4,7 +4,7 @@ import { check, Match } from 'meteor/check';
 import { sendTestEmailToInbox } from '../../../../server/features/EmailInbox/EmailInbox_Outgoing';
 import { API } from '../api';
 import { getPaginationItems } from '../helpers/getPaginationItems';
-import { insertOneEmailInbox, findEmailInboxes, findOneEmailInbox, updateEmailInbox } from '../lib/emailInbox';
+import { insertOneEmailInbox, findEmailInboxes, updateEmailInbox, removeEmailInbox } from '../lib/emailInbox';
 
 API.v1.addRoute(
 	'email-inbox.list',
@@ -55,12 +55,23 @@ API.v1.addRoute(
 			let _id: string;
 
 			if (!emailInboxParams?._id) {
-				const emailInbox = await insertOneEmailInbox(this.userId, emailInboxParams);
-				_id = emailInbox.insertedId.toString();
+				const { insertedId } = await insertOneEmailInbox(this.userId, emailInboxParams);
+
+				if (!insertedId) {
+					return API.v1.failure('Failed to create email inbox');
+				}
+
+				_id = insertedId;
 			} else {
-				_id = emailInboxParams._id;
-				await updateEmailInbox({ ...emailInboxParams, _id });
+				const emailInbox = await updateEmailInbox({ ...emailInboxParams, _id: emailInboxParams._id });
+
+				if (!emailInbox?._id) {
+					return API.v1.failure('Failed to update email inbox');
+				}
+
+				_id = emailInbox._id;
 			}
+
 			return API.v1.success({ _id });
 		},
 	},
@@ -79,7 +90,7 @@ API.v1.addRoute(
 			if (!_id) {
 				throw new Error('error-invalid-param');
 			}
-			const emailInbox = await findOneEmailInbox({ _id });
+			const emailInbox = await EmailInbox.findOneById(_id);
 
 			if (!emailInbox) {
 				return API.v1.notFound();
@@ -97,11 +108,12 @@ API.v1.addRoute(
 				throw new Error('error-invalid-param');
 			}
 
-			const emailInboxes = await EmailInbox.findOneById(_id);
-			if (!emailInboxes) {
+			const { deletedCount } = await removeEmailInbox(_id);
+
+			if (!deletedCount) {
 				return API.v1.notFound();
 			}
-			await EmailInbox.removeById(_id);
+
 			return API.v1.success({ _id });
 		},
 	},
@@ -120,7 +132,7 @@ API.v1.addRoute(
 
 			// TODO: Chapter day backend - check if user has permission to view this email inbox instead of null values
 			// TODO: Chapter day: Remove this endpoint and move search to GET /email-inbox
-			const emailInbox = await EmailInbox.findOne({ email });
+			const emailInbox = await EmailInbox.findByEmail(email);
 
 			return API.v1.success({ emailInbox });
 		},
@@ -140,7 +152,7 @@ API.v1.addRoute(
 			if (!_id) {
 				throw new Error('error-invalid-param');
 			}
-			const emailInbox = await findOneEmailInbox({ _id });
+			const emailInbox = await EmailInbox.findOneById(_id);
 
 			if (!emailInbox) {
 				return API.v1.notFound();

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

@@ -10,6 +10,7 @@ import type {
 	IPbxEvent,
 	LoginServiceConfiguration as LoginServiceConfigurationData,
 	ILivechatPriority,
+	IEmailInbox,
 	IIntegrationHistory,
 	AtLeast,
 } from '@rocket.chat/core-typings';
@@ -266,6 +267,17 @@ export async function notifyOnIntegrationChangedByChannels<T extends IIntegratio
 	}
 }
 
+export async function notifyOnEmailInboxChanged<T extends IEmailInbox>(
+	data: Pick<T, '_id'> | T, // TODO: improve typing
+	clientAction: ClientAction = 'updated',
+): Promise<void> {
+	if (!dbWatchersDisabled) {
+		return;
+	}
+
+	void api.broadcast('watch.emailInbox', { clientAction, id: data._id, data });
+}
+
 export async function notifyOnIntegrationHistoryChanged<T extends IIntegrationHistory>(
 	data: AtLeast<T, '_id'>,
 	clientAction: ClientAction = 'updated',

apps/meteor/server/database/watchCollections.ts

@@ -34,7 +34,6 @@ export function getWatchCollections(): string[] {
 		LivechatInquiry.getCollectionName(),
 		LivechatDepartmentAgents.getCollectionName(),
 		InstanceStatus.getCollectionName(),
-		EmailInbox.getCollectionName(),
 		Settings.getCollectionName(),
 		Subscriptions.getCollectionName(),
 	];
@@ -49,6 +48,7 @@ export function getWatchCollections(): string[] {
 		collections.push(Permissions.getCollectionName());
 		collections.push(LivechatPriority.getCollectionName());
 		collections.push(LoginServiceConfiguration.getCollectionName());
+		collections.push(EmailInbox.getCollectionName());
 		collections.push(IntegrationHistory.getCollectionName());
 	}
 

apps/meteor/server/email/IMAPInterceptor.ts

@@ -6,6 +6,7 @@ import IMAP from 'imap';
 import type { ParsedMail } from 'mailparser';
 import { simpleParser } from 'mailparser';
 
+import { notifyOnEmailInboxChanged } from '../../app/lib/server/lib/notifyListener';
 import { logger } from '../features/EmailInbox/logger';
 
 type IMAPOptions = {
@@ -221,9 +222,15 @@ export class IMAPInterceptor extends EventEmitter {
 
 	async selfDisable(): Promise<void> {
 		logger.info(`Disabling inbox ${this.inboxId}`);
+
 		// Again, if there's 2 inboxes with the same email, this will prevent looping over the already disabled one
 		// Active filter is just in case :)
-		await EmailInbox.findOneAndUpdate({ _id: this.inboxId, active: true }, { $set: { active: false } });
+		const { value } = await EmailInbox.setDisabledById(this.inboxId);
+
+		if (value) {
+			void notifyOnEmailInboxChanged(value, 'updated');
+		}
+
 		logger.info(`IMAP inbox ${this.inboxId} automatically disabled`);
 	}
 }

apps/meteor/server/features/EmailInbox/EmailInbox.ts

@@ -18,9 +18,7 @@ export type Inbox = {
 export const inboxes = new Map<string, Inbox>();
 
 export async function configureEmailInboxes(): Promise<void> {
-	const emailInboxesCursor = EmailInbox.find({
-		active: true,
-	});
+	const emailInboxesCursor = EmailInbox.findActive();
 
 	logger.info('Clearing old email inbox registrations');
 	for (const { imap } of inboxes.values()) {

apps/meteor/server/models/raw/EmailInbox.ts

@@ -1,6 +1,6 @@
 import type { IEmailInbox, RocketChatRecordDeleted } from '@rocket.chat/core-typings';
 import type { IEmailInboxModel } from '@rocket.chat/model-typings';
-import type { Collection, Db, IndexDescription } from 'mongodb';
+import type { Collection, Db, FindCursor, IndexDescription, InsertOneResult, ModifyResult, UpdateFilter } from 'mongodb';
 
 import { BaseRaw } from './BaseRaw';
 
@@ -12,4 +12,27 @@ export class EmailInboxRaw extends BaseRaw<IEmailInbox> implements IEmailInboxMo
 	protected modelIndexes(): IndexDescription[] {
 		return [{ key: { email: 1 }, unique: true }];
 	}
+
+	async setDisabledById(id: IEmailInbox['_id']): Promise<ModifyResult<IEmailInbox>> {
+		return this.findOneAndUpdate({ _id: id, active: true }, { $set: { active: false } }, { returnDocument: 'after' });
+	}
+
+	async create(emailInbox: IEmailInbox): Promise<InsertOneResult<IEmailInbox>> {
+		return this.insertOne(emailInbox);
+	}
+
+	async updateById(id: IEmailInbox['_id'], data: UpdateFilter<IEmailInbox>): Promise<ModifyResult<Pick<IEmailInbox, '_id'>>> {
+		// findOneAndUpdate doesn't accept generics, so we had to type cast
+		return this.findOneAndUpdate({ _id: id }, data, { returnDocument: 'after', projection: { _id: 1 } }) as unknown as Promise<
+			ModifyResult<Pick<IEmailInbox, '_id'>>
+		>;
+	}
+
+	findActive(): FindCursor<IEmailInbox> {
+		return this.find({ active: true });
+	}
+
+	async findByEmail(email: IEmailInbox['email']): Promise<IEmailInbox | null> {
+		return this.findOne({ email });
+	}
 }

packages/model-typings/src/models/IEmailInboxModel.ts

@@ -1,8 +1,12 @@
 import type { IEmailInbox } from '@rocket.chat/core-typings';
+import type { FindCursor, InsertOneResult, ModifyResult, UpdateFilter } from 'mongodb';
 
 import type { IBaseModel } from './IBaseModel';
 
-// eslint-disable-next-line @typescript-eslint/no-empty-interface
 export interface IEmailInboxModel extends IBaseModel<IEmailInbox> {
-	//
+	setDisabledById(id: IEmailInbox['_id']): Promise<ModifyResult<IEmailInbox>>;
+	create(emailInbox: Omit<IEmailInbox, '_id'>): Promise<InsertOneResult<IEmailInbox>>;
+	updateById(id: IEmailInbox['_id'], data: UpdateFilter<IEmailInbox>): Promise<ModifyResult<Pick<IEmailInbox, '_id'>>>;
+	findActive(): FindCursor<IEmailInbox>;
+	findByEmail(email: IEmailInbox['email']): Promise<IEmailInbox | null>;
 }