refactor: move index creation to onServerVersionChange

.vscode/settings.json

@@ -1,4 +1,5 @@
 {
+	"eslint.workingDirectories": ["."],
 	"typescript.tsdk": "./node_modules/typescript/lib",
 	"cSpell.words": [
 		"autotranslate",

apps/meteor/app/authorization/server/functions/upsertPermissions.ts

@@ -43,68 +43,89 @@ export const upsertPermissions = async (): Promise<void> => {
 
 	const createSettingPermission = async function (
 		setting: ISetting,
-		previousSettingPermissions: {
-			[key: string]: IPermission;
-		},
+		previousSettingPermissions: Record<string, IPermission>,
 	): Promise<void> {
+		const { _id: permissionId, doc } = buildSettingPermissionDoc(setting, previousSettingPermissions);
+		try {
+			await Permissions.updateOne({ _id: permissionId }, { $set: doc }, { upsert: true });
+		} catch (e) {
+			if (!(e as Error).message.includes('E11000')) {
+				await Permissions.updateOne({ _id: permissionId }, { $set: doc }, { upsert: true });
+			}
+		}
+		delete previousSettingPermissions[permissionId];
+	};
+
+	const buildSettingPermissionDoc = function (
+		setting: ISetting,
+		previousSettingPermissions: Record<string, IPermission>,
+	): { _id: string; doc: Omit<IPermission, '_id'> } {
 		const permissionId = getSettingPermissionId(setting._id);
 		const permission: Omit<IPermission, '_id' | '_updatedAt'> = {
 			level: CONSTANTS.SETTINGS_LEVEL as 'settings' | undefined,
-			// copy those setting-properties which are needed to properly publish the setting-based permissions
 			settingId: setting._id,
 			group: setting.group,
 			section: setting.section ?? undefined,
 			sorter: setting.sorter,
-			roles: [],
+			roles: previousSettingPermissions[permissionId]?.roles ?? [],
 		};
-		// copy previously assigned roles if available
-		if (previousSettingPermissions[permissionId]?.roles) {
-			permission.roles = previousSettingPermissions[permissionId].roles;
-		}
 		if (setting.group) {
 			permission.groupPermissionId = getSettingPermissionId(setting.group);
 		}
 		if (setting.section) {
 			permission.sectionPermissionId = getSettingPermissionId(setting.section);
 		}
+		return { _id: permissionId, doc: { ...permission, _updatedAt: new Date() } };
+	};
 
-		const existent = await Permissions.findOne(
-			{
-				_id: permissionId,
-				...permission,
-			},
-			{ projection: { _id: 1 } },
-		);
+	const BULK_WRITE_BATCH_SIZE = 500;
 
-		if (!existent) {
-			try {
-				await Permissions.updateOne({ _id: permissionId }, { $set: permission }, { upsert: true });
-			} catch (e) {
-				if (!(e as Error).message.includes('E11000')) {
-					// E11000 refers to a MongoDB error that can occur when using unique indexes for upserts
-					// https://docs.mongodb.com/manual/reference/method/db.collection.update/#use-unique-indexes
-					await Permissions.updateOne({ _id: permissionId }, { $set: permission }, { upsert: true });
-				}
-			}
-		}
-
-		delete previousSettingPermissions[permissionId];
+	type SettingPermissionUpdateOp = {
+		updateOne: {
+			filter: { _id: string };
+			update: { $set: Omit<IPermission, '_id'> };
+			upsert: true;
+		};
 	};
 
 	const createPermissionsForExistingSettings = async function (): Promise<void> {
 		const previousSettingPermissions = await getPreviousPermissions();
+		const settingsList = await Settings.findNotHidden().toArray();
 
-		const settings = await Settings.findNotHidden().toArray();
-		for await (const setting of settings) {
-			await createSettingPermission(setting, previousSettingPermissions);
+		const updateOps: SettingPermissionUpdateOp[] = [];
+		for (const setting of settingsList) {
+			const { _id: permissionId, doc } = buildSettingPermissionDoc(setting, previousSettingPermissions);
+			updateOps.push({
+				updateOne: {
+					filter: { _id: permissionId },
+					update: { $set: doc },
+					upsert: true,
+				},
+			});
+			delete previousSettingPermissions[permissionId];
 		}
 
-		// remove permissions for non-existent settings
-		for await (const obsoletePermission of Object.keys(previousSettingPermissions)) {
-			if (previousSettingPermissions.hasOwnProperty(obsoletePermission)) {
-				await Permissions.deleteOne({ _id: obsoletePermission });
+		// Batches run sequentially so E11000 retry applies per batch
+		/* eslint-disable no-await-in-loop */
+		for (let i = 0; i < updateOps.length; i += BULK_WRITE_BATCH_SIZE) {
+			const batch = updateOps.slice(i, i + BULK_WRITE_BATCH_SIZE);
+			try {
+				await Permissions.col.bulkWrite(batch, { ordered: false });
+			} catch (e) {
+				if ((e as Error).message.includes('E11000')) {
+					// E11000 duplicate key: retry without upsert for this batch (doc already exists)
+					await Promise.all(batch.map((op) => Permissions.updateOne(op.updateOne.filter, op.updateOne.update)));
+				} else {
+					throw e;
+				}
 			}
 		}
+		/* eslint-enable no-await-in-loop */
+
+		const obsoleteIds = Object.keys(previousSettingPermissions);
+		if (obsoleteIds.length > 0) {
+			await Permissions.deleteMany({ _id: { $in: obsoleteIds } });
+		}
 	};
 
 	// for each setting which already exists, create a permission to allow changing just this one setting

apps/meteor/app/settings/server/startup.ts

@@ -1,13 +1,11 @@
-import type { ISetting } from '@rocket.chat/core-typings';
 import type { Settings } from '@rocket.chat/models';
 
 import type { ICachedSettings } from './CachedSettings';
 
-// eslint-disable-next-line @typescript-eslint/naming-convention
 export async function initializeSettings({ model, settings }: { model: typeof Settings; settings: ICachedSettings }): Promise<void> {
-	await model.find().forEach((record: ISetting) => {
+	const records = await model.find().toArray();
+	for (const record of records) {
 		settings.set(record);
-	});
-
+	}
 	settings.initialized();
 }

apps/meteor/ee/server/apps/orchestrator.js

@@ -191,24 +191,30 @@ export class AppServerOrchestrator {
 			return;
 		}
 
+		const loadStart = Date.now();
 		await this.getManager().load();
 
 		// Before enabling each app we verify if there is still room for it
 		const apps = await this.getManager().get();
 
-		// This needs to happen sequentially to keep track of app limits
-		for await (const app of apps) {
-			try {
-				await canEnableApp(app.getStorageItem());
-
-				await this.getManager().loadOne(app.getID(), true);
-			} catch (error) {
-				this._rocketchatLogger.warn({
-					msg: 'App could not be enabled',
-					appName: app.getInfo().name,
-					err: error,
-				});
-			}
+		const CONCURRENCY_LIMIT = 4;
+		for (let i = 0; i < apps.length; i += CONCURRENCY_LIMIT) {
+			const chunk = apps.slice(i, i + CONCURRENCY_LIMIT);
+			// eslint-disable-next-line no-await-in-loop
+			await Promise.all(
+				chunk.map(async (app) => {
+					try {
+						await canEnableApp(app.getStorageItem());
+						await this.getManager().loadOne(app.getID(), true);
+					} catch (error) {
+						this._rocketchatLogger.warn({
+							msg: 'App could not be enabled',
+							appName: app.getInfo().name,
+							err: error,
+						});
+					}
+				}),
+			);
 		}
 
 		await this.getBridges().getSchedulerBridge().startScheduler();
@@ -218,6 +224,7 @@ export class AppServerOrchestrator {
 		this._rocketchatLogger.info({
 			msg: 'Loaded the Apps Framework and apps',
 			appCount,
+			durationMs: Date.now() - loadStart,
 		});
 	}
 

apps/meteor/server/database/trash.ts

@@ -1,6 +1,8 @@
-import { TrashRaw } from '@rocket.chat/models';
+import { registerModel, TrashRaw } from '@rocket.chat/models';
 
 import { db } from './utils';
 
 const Trash = new TrashRaw(db);
 export const trashCollection = Trash.col;
+
+registerModel('ITrashModel', Trash);

apps/meteor/server/lib/migrations.ts

@@ -311,7 +311,13 @@ export async function migrateDatabase(targetVersion: 'latest' | number, subcomma
 	return true;
 }
 
-export async function onServerVersionChange(cb: () => Promise<void>): Promise<void> {
+let hashVersion: string | undefined;
+
+const getHashVersion = async () => {
+	if (hashVersion) {
+		return hashVersion;
+	}
+
 	const result = await Migrations.findOneAndUpdate(
 		{
 			_id: 'upgrade',
@@ -326,9 +332,12 @@ export async function onServerVersionChange(cb: () => Promise<void>): Promise<vo
 		},
 	);
 
-	if (result?.hash === Info.commit.hash) {
-		return;
-	}
+	hashVersion = result?.hash;
+	return hashVersion;
+};
+
+export async function shouldRunServerVersionChange(): Promise<boolean> {
+	const hash = await getHashVersion();
 
-	await cb();
+	return hash !== Info.commit.hash;
 }

apps/meteor/server/startup/migrations/xrun.ts

@@ -1,9 +1,9 @@
-import { Settings } from '@rocket.chat/models';
+import { Settings, indexes } from '@rocket.chat/models';
 import type { UpdateResult } from 'mongodb';
 
 import { upsertPermissions } from '../../../app/authorization/server/functions/upsertPermissions';
 import { settings } from '../../../app/settings/server';
-import { migrateDatabase, onServerVersionChange } from '../../lib/migrations';
+import { migrateDatabase, shouldRunServerVersionChange } from '../../lib/migrations';
 import { ensureCloudWorkspaceRegistered } from '../cloudRegistration';
 
 const { MIGRATION_VERSION = 'latest' } = process.env;
@@ -57,10 +57,15 @@ const moveRetentionSetting = async () => {
 
 export const performMigrationProcedure = async (): Promise<void> => {
 	await migrateDatabase(version === 'latest' ? version : parseInt(version), subcommands);
-	// perform operations when the server is starting with a different version
-	await onServerVersionChange(async () => {
-		await upsertPermissions();
-		await ensureCloudWorkspaceRegistered();
-		await moveRetentionSetting();
-	});
+
+	if (!(await shouldRunServerVersionChange())) {
+		indexes.cancel();
+		return;
+	}
+
+	indexes.ensureIndexes();
+
+	await upsertPermissions();
+	await ensureCloudWorkspaceRegistered();
+	await moveRetentionSetting();
 };

packages/models/package.json

@@ -18,6 +18,7 @@
 		"unit": "jest"
 	},
 	"dependencies": {
+		"@rocket.chat/emitter": "^0.32.0",
 		"@rocket.chat/model-typings": "workspace:~",
 		"@rocket.chat/random": "workspace:^",
 		"@rocket.chat/rest-typings": "workspace:^",

packages/models/src/models/BaseRaw.ts

@@ -1,4 +1,5 @@
 import type { RocketChatRecordDeleted } from '@rocket.chat/core-typings';
+import { Emitter } from '@rocket.chat/emitter';
 import type { IBaseModel, DefaultFields, ResultFields, FindPaginated, InsertionModel } from '@rocket.chat/model-typings';
 import { traceInstanceMethods } from '@rocket.chat/tracing';
 import { ObjectId } from 'mongodb';
@@ -46,6 +47,30 @@ type ModelOptions = {
 	collection?: CollectionOptions;
 };
 
+export type IndexRegisterFn = () => Promise<void>;
+const ee = new Emitter<{
+	added: IndexRegisterFn;
+}>();
+// The idea is to accumulate the indexes that should be created in a set, and then create them all at once.
+// in case of a lazy model, we need to create the indexes when the model is instantiated.
+
+const indexesThatShouldBeCreated = new Set<IndexRegisterFn>();
+const onAdded = (fn: IndexRegisterFn) => indexesThatShouldBeCreated.add(fn);
+const onAddedExecute = (fn: IndexRegisterFn) => fn();
+ee.on('added', onAdded);
+export const indexes = {
+	ensureIndexes: () => {
+		indexesThatShouldBeCreated.forEach((fn) => fn());
+		indexesThatShouldBeCreated.clear();
+		ee.off('added', onAdded);
+		ee.on('added', onAddedExecute);
+	},
+	cancel: () => {
+		ee.off('added', onAdded);
+		indexesThatShouldBeCreated.clear();
+	},
+} as const;
+
 export abstract class BaseRaw<
 	T extends { _id: string },
 	C extends DefaultFields<T> = undefined,
@@ -79,10 +104,10 @@ export abstract class BaseRaw<
 
 		this.col = this.db.collection(this.collectionName, options?.collection || {});
 
-		void this.createIndexes();
-
 		this.preventSetUpdatedAt = options?.preventSetUpdatedAt ?? false;
 
+		void ee.emit('added', () => this.createIndexes());
+
 		return traceInstanceMethods(this);
 	}
 
@@ -363,7 +388,7 @@ export abstract class BaseRaw<
 			throw e;
 		}
 
-		return doc as WithId<T>;
+		return doc;
 	}
 
 	async deleteMany(filter: Filter<T>, options?: DeleteOptions & { onTrash?: (record: ResultFields<T, C>) => void }): Promise<DeleteResult> {

yarn.lock

@@ -9491,6 +9491,7 @@ __metadata:
   version: 0.0.0-use.local
   resolution: "@rocket.chat/models@workspace:packages/models"
   dependencies:
+    "@rocket.chat/emitter": "npm:^0.32.0"
     "@rocket.chat/jest-presets": "workspace:~"
     "@rocket.chat/model-typings": "workspace:~"
     "@rocket.chat/random": "workspace:^"