chore(modules-sdk): parallel migrations (#13898)

.changeset/young-eggs-rush.md

@@ -0,0 +1,9 @@
+---
+"@medusajs/modules-sdk": patch
+"@medusajs/framework": patch
+"@medusajs/utils": patch
+"@medusajs/test-utils": patch
+"@medusajs/medusa": patch
+---
+
+chore(modules-sdk): parallel migrations

packages/core/framework/src/migrations/migrator.ts

@@ -1,12 +1,12 @@
 import { MedusaContainer } from "@medusajs/types"
-import { Knex } from "../deps/mikro-orm-knex"
 import { glob } from "glob"
 import { join } from "path"
+import { Knex } from "../deps/mikro-orm-knex"
 import { logger } from "../logger"
 import { ContainerRegistrationKeys } from "../utils"
 
-export abstract class Migrator {
-  protected abstract migration_table_name: string
+export class Migrator {
+  protected migration_table_name: string
 
   protected container: MedusaContainer
   protected pgConnection: Knex<any>
@@ -18,6 +18,7 @@ export abstract class Migrator {
     this.pgConnection = this.container.resolve(
       ContainerRegistrationKeys.PG_CONNECTION
     )
+    this.migration_table_name = "mikro_orm_migrations"
   }
 
   /**
@@ -158,7 +159,21 @@ export abstract class Migrator {
     return allScripts
   }
 
-  protected abstract createMigrationTable(): Promise<void>
-  abstract run(...args: any[]): Promise<any>
-  abstract getPendingMigrations(migrationPaths: string[]): Promise<string[]>
+  protected async createMigrationTable(): Promise<void> {
+    await this.pgConnection.raw(`
+      CREATE TABLE IF NOT EXISTS ${this.migration_table_name} (
+        id serial PRIMARY KEY,
+        name varchar(255),
+        executed_at timestamptz DEFAULT CURRENT_TIMESTAMP
+      )
+    `)
+  }
+
+  run(...args: any[]): Promise<any> {
+    throw new Error("Method not implemented")
+  }
+
+  getPendingMigrations(migrationPaths: string[]): Promise<string[]> {
+    throw new Error("Method not implemented")
+  }
 }

packages/core/modules-sdk/src/medusa-app.ts

@@ -1,3 +1,4 @@
+import { asValue } from "@medusajs/deps/awilix"
 import { RemoteFetchDataCallback } from "@medusajs/orchestration"
 import {
   ConfigModule,
@@ -21,6 +22,7 @@ import {
   createMedusaContainer,
   discoverFeatureFlagsFromDir,
   dynamicImport,
+  executeWithConcurrency,
   FeatureFlag,
   GraphQLUtils,
   isObject,
@@ -33,7 +35,6 @@ import {
   promiseAll,
   registerFeatureFlag,
 } from "@medusajs/utils"
-import { asValue } from "@medusajs/deps/awilix"
 import { Link } from "./link"
 import {
   MedusaModule,
@@ -501,12 +502,14 @@ async function MedusaApp_({
     modulesNames: string[]
     action?: "run" | "revert" | "generate"
   }) => {
-    const moduleResolutions = modulesNames.map((moduleName) => {
-      return {
-        moduleName,
-        resolution: MedusaModule.getModuleResolutions(moduleName),
+    const moduleResolutions = Array.from(new Set(modulesNames)).map(
+      (moduleName) => {
+        return {
+          moduleName,
+          resolution: MedusaModule.getModuleResolutions(moduleName),
+        }
       }
-    })
+    )
 
     const missingModules = moduleResolutions
       .filter(({ resolution }) => !resolution)
@@ -524,7 +527,7 @@ async function MedusaApp_({
       throw error
     }
 
-    for (const { resolution: moduleResolution } of moduleResolutions) {
+    const run = async ({ resolution: moduleResolution }) => {
       if (
         !moduleResolution.options?.database &&
         moduleResolution.moduleDeclaration?.scope === MODULE_SCOPE.INTERNAL
@@ -554,6 +557,11 @@ async function MedusaApp_({
         await MedusaModule.migrateGenerate(migrationOptions)
       }
     }
+
+    await executeWithConcurrency(
+      moduleResolutions.map((a) => () => run(a)),
+      8 // parallel migrations
+    )
   }
 
   const runMigrations: RunMigrationFn = async (): Promise<void> => {

packages/core/utils/src/common/execute-with-concurrency.ts

@@ -0,0 +1,31 @@
+/**
+ * Execute functions with a concurrency limit
+ * @param functions Array of functions to execute in parallel
+ * @param concurrency Maximum number of concurrent executions
+ */
+export async function executeWithConcurrency<T>(
+  functions: (() => Promise<T>)[],
+  concurrency: number
+): Promise<PromiseSettledResult<Awaited<T>>[]> {
+  const results: PromiseSettledResult<Awaited<T>>[] = new Array(
+    functions.length
+  )
+  let currentIndex = 0
+
+  const executeNext = async (): Promise<void> => {
+    while (currentIndex < functions.length) {
+      const index = currentIndex++
+      const result = await Promise.allSettled([functions[index]()])
+      results[index] = result[0]
+    }
+  }
+
+  const workers: Promise<void>[] = []
+  for (let i = 0; i < concurrency; i++) {
+    workers.push(executeNext())
+  }
+
+  await Promise.all(workers)
+
+  return results
+}

packages/core/utils/src/common/index.ts

@@ -19,6 +19,7 @@ export * from "./define-file-config"
 export * from "./dynamic-import"
 export * from "./env-editor"
 export * from "./errors"
+export * from "./execute-with-concurrency"
 export * from "./file-system"
 export * from "./filter-object-by-keys"
 export * from "./filter-operator-map"

packages/medusa-test-utils/src/medusa-test-runner.ts

@@ -1,4 +1,6 @@
+import { asValue } from "@medusajs/framework/awilix"
 import { logger } from "@medusajs/framework/logger"
+import { Migrator } from "@medusajs/framework/migrations"
 import { MedusaAppOutput } from "@medusajs/framework/modules-sdk"
 import { MedusaContainer } from "@medusajs/framework/types"
 import {
@@ -7,7 +9,6 @@ import {
   getResolvedPlugins,
   mergePluginModules,
 } from "@medusajs/framework/utils"
-import { asValue } from "@medusajs/framework/awilix"
 import { dbTestUtilFactory, getDatabaseURL } from "./database"
 import {
   applyEnvVarsToProcess,
@@ -178,6 +179,9 @@ class MedusaTestRunner {
 
     await this.initializeDatabase()
 
+    const migrator = new Migrator({ container })
+    await migrator.ensureMigrationsTable()
+
     logger.info(
       `Migrating database with core migrations and links ${this.dbName}`
     )

packages/medusa/src/commands/db/generate.ts

@@ -1,4 +1,4 @@
-import { MedusaAppLoader } from "@medusajs/framework"
+import { MedusaAppLoader, Migrator } from "@medusajs/framework"
 import { LinkLoader } from "@medusajs/framework/links"
 import {
   ContainerRegistrationKeys,
@@ -41,6 +41,9 @@ const main = async function ({ directory, modules }) {
      */
     logger.info("Generating migrations...")
 
+    const migrator = new Migrator({ container })
+    await migrator.ensureMigrationsTable()
+
     await medusaAppLoader.runModulesMigrations({
       moduleNames: modules,
       action: "generate",

packages/medusa/src/commands/db/migrate.ts

@@ -1,4 +1,4 @@
-import { MEDUSA_CLI_PATH, MedusaAppLoader } from "@medusajs/framework"
+import { MEDUSA_CLI_PATH, MedusaAppLoader, Migrator } from "@medusajs/framework"
 import { LinkLoader } from "@medusajs/framework/links"
 import {
   ContainerRegistrationKeys,
@@ -60,6 +60,10 @@ export async function migrate({
    * Run migrations
    */
   logger.info("Running migrations...")
+
+  const migrator = new Migrator({ container })
+  await migrator.ensureMigrationsTable()
+
   await medusaAppLoader.runModulesMigrations({
     action: "run",
   })

packages/medusa/src/commands/db/rollback.ts

@@ -1,4 +1,4 @@
-import { MedusaAppLoader } from "@medusajs/framework"
+import { MedusaAppLoader, Migrator } from "@medusajs/framework"
 import { LinkLoader } from "@medusajs/framework/links"
 import {
   ContainerRegistrationKeys,
@@ -40,6 +40,10 @@ const main = async function ({ directory, modules }) {
      * Reverting migrations
      */
     logger.info("Reverting migrations...")
+
+    const migrator = new Migrator({ container })
+    await migrator.ensureMigrationsTable()
+
     await medusaAppLoader.runModulesMigrations({
       moduleNames: modules,
       action: "revert",

packages/modules/index/integration-tests/__tests__/config-sync.spec.ts

@@ -3,13 +3,14 @@ import {
   container,
   logger,
   MedusaAppLoader,
+  Migrator,
 } from "@medusajs/framework"
+import { asValue } from "@medusajs/framework/awilix"
 import { MedusaAppOutput, MedusaModule } from "@medusajs/framework/modules-sdk"
 import { ContainerRegistrationKeys, Modules } from "@medusajs/framework/utils"
 import { initDb, TestDatabaseUtils } from "@medusajs/test-utils"
 import { IndexTypes, ModulesSdkTypes } from "@medusajs/types"
 import { Configuration } from "@utils"
-import { asValue } from "@medusajs/framework/awilix"
 import path from "path"
 import { setTimeout } from "timers/promises"
 import { EventBusServiceMock } from "../__fixtures__"
@@ -49,6 +50,10 @@ const beforeAll_ = async () => {
     medusaAppLoader = new MedusaAppLoader()
 
     // Migrations
+
+    const migrator = new Migrator({ container })
+    await migrator.ensureMigrationsTable()
+
     await medusaAppLoader.runModulesMigrations()
     const linkPlanner = await medusaAppLoader.getLinksExecutionPlanner()
     const plan = await linkPlanner.createPlan()

packages/modules/index/integration-tests/__tests__/data-synchronizer.spec.ts

@@ -3,7 +3,10 @@ import {
   container,
   logger,
   MedusaAppLoader,
+  Migrator,
 } from "@medusajs/framework"
+import { asValue } from "@medusajs/framework/awilix"
+import { EntityManager } from "@medusajs/framework/mikro-orm/postgresql"
 import { MedusaAppOutput, MedusaModule } from "@medusajs/framework/modules-sdk"
 import { IndexTypes, InferEntityType } from "@medusajs/framework/types"
 import {
@@ -12,14 +15,12 @@ import {
   toMikroORMEntity,
 } from "@medusajs/framework/utils"
 import { initDb, TestDatabaseUtils } from "@medusajs/test-utils"
-import { EntityManager } from "@medusajs/framework/mikro-orm/postgresql"
 import { IndexData, IndexRelation } from "@models"
 import { DataSynchronizer } from "@services"
-import { asValue } from "@medusajs/framework/awilix"
 import * as path from "path"
 import { setTimeout } from "timers/promises"
 import { EventBusServiceMock } from "../__fixtures__"
-import config, { dbName } from "../__fixtures__/medusa-config"
+import { dbName } from "../__fixtures__/medusa-config"
 
 const eventBusMock = new EventBusServiceMock()
 const queryMock = {
@@ -86,6 +87,9 @@ const beforeAll_ = async () => {
     medusaAppLoader = new MedusaAppLoader()
 
     // Migrations
+    const migrator = new Migrator({ container })
+    await migrator.ensureMigrationsTable()
+
     await medusaAppLoader.runModulesMigrations()
     const linkPlanner = await medusaAppLoader.getLinksExecutionPlanner()
     const plan = await linkPlanner.createPlan()

packages/modules/index/integration-tests/__tests__/index-engine-module-sync.spec.ts

@@ -3,6 +3,7 @@ import {
   container,
   logger,
   MedusaAppLoader,
+  Migrator,
 } from "@medusajs/framework"
 import { asValue } from "@medusajs/framework/awilix"
 import { EntityManager } from "@medusajs/framework/mikro-orm/postgresql"
@@ -14,7 +15,7 @@ import {
   toMikroORMEntity,
 } from "@medusajs/framework/utils"
 import { initDb, TestDatabaseUtils } from "@medusajs/test-utils"
-import { IndexData, IndexRelation, IndexMetadata, IndexSync } from "@models"
+import { IndexData, IndexMetadata, IndexRelation, IndexSync } from "@models"
 import { IndexMetadataStatus } from "@utils"
 import * as path from "path"
 import { setTimeout } from "timers/promises"
@@ -63,6 +64,9 @@ const beforeAll_ = async ({
     medusaAppLoader = new MedusaAppLoader(container as any)
 
     // Migrations
+    const migrator = new Migrator({ container })
+    await migrator.ensureMigrationsTable()
+
     await medusaAppLoader.runModulesMigrations()
     const linkPlanner = await medusaAppLoader.getLinksExecutionPlanner()
     const plan = await linkPlanner.createPlan()

packages/modules/index/integration-tests/__tests__/index-engine-module.spec.ts

@@ -3,6 +3,7 @@ import {
   container,
   logger,
   MedusaAppLoader,
+  Migrator,
 } from "@medusajs/framework"
 import { asValue } from "@medusajs/framework/awilix"
 import { EntityManager } from "@medusajs/framework/mikro-orm/postgresql"
@@ -124,6 +125,9 @@ const beforeAll_ = async () => {
     medusaAppLoader = new MedusaAppLoader(container as any)
 
     // Migrations
+    const migrator = new Migrator({ container })
+    await migrator.ensureMigrationsTable()
+
     await medusaAppLoader.runModulesMigrations()
     const linkPlanner = await medusaAppLoader.getLinksExecutionPlanner()
     const plan = await linkPlanner.createPlan()

packages/modules/index/integration-tests/__tests__/query-builder.spec.ts

@@ -3,14 +3,15 @@ import {
   container,
   logger,
   MedusaAppLoader,
+  Migrator,
 } from "@medusajs/framework"
+import { asValue } from "@medusajs/framework/awilix"
+import { EntityManager } from "@medusajs/framework/mikro-orm/postgresql"
 import { MedusaAppOutput, MedusaModule } from "@medusajs/framework/modules-sdk"
 import { IndexTypes } from "@medusajs/framework/types"
 import { ContainerRegistrationKeys, Modules } from "@medusajs/framework/utils"
 import { initDb, TestDatabaseUtils } from "@medusajs/test-utils"
-import { EntityManager } from "@medusajs/framework/mikro-orm/postgresql"
 import { IndexData, IndexRelation } from "@models"
-import { asValue } from "@medusajs/framework/awilix"
 import path from "path"
 import { EventBusServiceMock } from "../__fixtures__"
 import { dbName } from "../__fixtures__/medusa-config"
@@ -47,6 +48,9 @@ const beforeAll_ = async () => {
     medusaAppLoader = new MedusaAppLoader(container as any)
 
     // Migrations
+    const migrator = new Migrator({ container })
+    await migrator.ensureMigrationsTable()
+
     await medusaAppLoader.runModulesMigrations()
     const linkPlanner = await medusaAppLoader.getLinksExecutionPlanner()
     const plan = await linkPlanner.createPlan()