Optimize migrate-email-fields-command

packages/twenty-server/src/database/commands/upgrade-version/0-30/0-30-migrate-email-fields-to-emails.command.ts

@@ -55,9 +55,12 @@ export class MigrateEmailFieldsToEmailsCommand extends ActiveWorkspacesCommandRu
     );
 
     for (const workspaceId of workspaceIds) {
+      let dataSourceMetadata;
+      let workspaceQueryRunner;
+
       this.logger.log(`Running command for workspace ${workspaceId}`);
       try {
-        const dataSourceMetadata =
+        dataSourceMetadata =
           await this.dataSourceService.getLastDataSourceMetadataFromWorkspaceId(
             workspaceId,
           );
@@ -77,10 +80,19 @@ export class MigrateEmailFieldsToEmailsCommand extends ActiveWorkspacesCommandRu
           );
         }
 
-        const workspaceQueryRunner = workspaceDataSource.createQueryRunner();
+        workspaceQueryRunner = workspaceDataSource.createQueryRunner();
 
         await workspaceQueryRunner.connect();
+      } catch (error) {
+        this.logger.log(
+          chalk.red(
+            `Could not connect to workspace data source for workspace ${workspaceId}`,
+          ),
+        );
+        continue;
+      }
 
+      try {
         const customFieldsWithEmailType =
           await this.fieldMetadataRepository.find({
             where: {
@@ -267,17 +279,19 @@ export class MigrateEmailFieldsToEmailsCommand extends ActiveWorkspacesCommandRu
                 workspaceId,
               );
             }
-          } finally {
-            await workspaceQueryRunner.release();
           }
         }
       } catch (error) {
+        await workspaceQueryRunner.release();
+
         this.logger.log(
           chalk.red(
             `Running command on workspace ${workspaceId} failed with error: ${error}`,
           ),
         );
         continue;
+      } finally {
+        await workspaceQueryRunner.release();
       }
 
       this.logger.log(chalk.green(`Command completed!`));

packages/twenty-server/src/engine/twenty-orm/factories/workspace-datasource.factory.ts

@@ -30,28 +30,11 @@ export class WorkspaceDatasourceFactory {
     workspaceId: string,
     workspaceMetadataVersion: number | null,
   ): Promise<WorkspaceDataSource> {
-    const latestWorkspaceMetadataVersion =
-      await this.workspaceCacheStorageService.getMetadataVersion(workspaceId);
-
-    if (latestWorkspaceMetadataVersion === undefined) {
-      await this.workspaceMetadataCacheService.recomputeMetadataCache(
-        workspaceId,
-      );
-      throw new TwentyORMException(
-        `Metadata version not found for workspace ${workspaceId}`,
-        TwentyORMExceptionCode.METADATA_VERSION_NOT_FOUND,
-      );
-    }
-
     const desiredWorkspaceMetadataVersion =
-      workspaceMetadataVersion ?? latestWorkspaceMetadataVersion;
-
-    if (latestWorkspaceMetadataVersion !== desiredWorkspaceMetadataVersion) {
-      throw new TwentyORMException(
-        `Workspace metadata version mismatch detected for workspace ${workspaceId}. Current version: ${latestWorkspaceMetadataVersion}. Desired version: ${desiredWorkspaceMetadataVersion}`,
-        TwentyORMExceptionCode.METADATA_VERSION_MISMATCH,
+      await this.computeDesiredWorkspaceMetadataVersion(
+        workspaceId,
+        workspaceMetadataVersion,
       );
-    }
 
     const workspaceDataSource = await this.cacheManager.execute(
       `${workspaceId}-${desiredWorkspaceMetadataVersion}`,
@@ -166,4 +149,46 @@ export class WorkspaceDatasourceFactory {
 
     return workspaceDataSource;
   }
+
+  private async computeDesiredWorkspaceMetadataVersion(
+    workspaceId: string,
+    workspaceMetadataVersion: number | null,
+  ): Promise<number> {
+    const latestWorkspaceMetadataVersion =
+      await this.workspaceCacheStorageService.getMetadataVersion(workspaceId);
+
+    if (latestWorkspaceMetadataVersion === undefined) {
+      await this.workspaceMetadataCacheService.recomputeMetadataCache(
+        workspaceId,
+      );
+      throw new TwentyORMException(
+        `Metadata version not found for workspace ${workspaceId}`,
+        TwentyORMExceptionCode.METADATA_VERSION_NOT_FOUND,
+      );
+    }
+
+    const desiredWorkspaceMetadataVersion =
+      workspaceMetadataVersion ?? latestWorkspaceMetadataVersion;
+
+    if (latestWorkspaceMetadataVersion !== desiredWorkspaceMetadataVersion) {
+      throw new TwentyORMException(
+        `Workspace metadata version mismatch detected for workspace ${workspaceId}. Current version: ${latestWorkspaceMetadataVersion}. Desired version: ${desiredWorkspaceMetadataVersion}`,
+        TwentyORMExceptionCode.METADATA_VERSION_MISMATCH,
+      );
+    }
+
+    return desiredWorkspaceMetadataVersion;
+  }
+
+  public async destroy(
+    workspaceId: string,
+    metadataVersion: number | null,
+  ): Promise<void> {
+    const desiredWorkspaceMetadataVersion =
+      this.computeDesiredWorkspaceMetadataVersion(workspaceId, metadataVersion);
+
+    await this.cacheManager.clearKey(
+      `${workspaceId}-${desiredWorkspaceMetadataVersion}`,
+    );
+  }
 }

packages/twenty-server/src/engine/twenty-orm/storage/cache-manager.storage.ts

@@ -34,6 +34,16 @@ export class CacheManager<T> {
     return value;
   }
 
+  async clearKey(
+    cacheKey: CacheKey,
+    onDelete?: (value: T) => Promise<void> | void,
+  ): Promise<void> {
+    if (this.cache.has(cacheKey)) {
+      await onDelete?.(this.cache.get(cacheKey)!);
+      this.cache.delete(cacheKey);
+    }
+  }
+
   async clear(onDelete?: (value: T) => Promise<void> | void): Promise<void> {
     for (const value of this.cache.values()) {
       await onDelete?.(value);

packages/twenty-server/src/engine/twenty-orm/twenty-orm-global.manager.ts

@@ -53,4 +53,8 @@ export class TwentyORMGlobalManager {
   async loadDataSourceForWorkspace(workspaceId: string) {
     await this.workspaceDataSourceFactory.create(workspaceId, null);
   }
+
+  async destroyDataSourceForWorkspace(workspaceId: string) {
+    await this.workspaceDataSourceFactory.destroy(workspaceId, null);
+  }
 }