chore: configurable database migration in concurrency (#14004)

> [!NOTE]
> Adds a configurable concurrency for link migrations (CLI/commands), forces concurrency=1 when pgstream is detected, and ignores duplicate link-table inserts.
> 
> - **CLI**
>   - Add `--concurrency` option to `db:migrate` and `db:sync-links`.
> - **Medusa commands**
>   - `migrate` and `sync-links` accept `concurrency`; set `DB_MIGRATION_CONCURRENCY` and force `1` when `pgstream` schema exists via new `isPgstreamEnabled`.
> - **Link Modules (migrations)**
>   - Execute plan actions with `executeWithConcurrency` using `DB_MIGRATION_CONCURRENCY`.
>   - Make link-table tracking inserts idempotent with `ON CONFLICT DO NOTHING` (including bulk/upsert and per-create).
> 
> <sup>Written by [Cursor Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit 07432293c8fe8de30b07920fa47823b9081edacc. This will update automatically on new commits. Configure [here](https://cursor.com/dashboard?tab=bugbot).</sup>

packages/modules/link-modules/src/migration/index.ts

@@ -6,18 +6,18 @@ import {
   PlannerActionLinkDescriptor,
 } from "@medusajs/framework/types"
 
-import {
-  arrayDifference,
-  DALUtils,
-  ModulesSdkUtils,
-  normalizeMigrationSQL,
-  promiseAll,
-} from "@medusajs/framework/utils"
 import { EntitySchema, MikroORM } from "@medusajs/framework/mikro-orm/core"
 import {
   DatabaseSchema,
   PostgreSqlDriver,
 } from "@medusajs/framework/mikro-orm/postgresql"
+import {
+  arrayDifference,
+  DALUtils,
+  executeWithConcurrency,
+  ModulesSdkUtils,
+  normalizeMigrationSQL,
+} from "@medusajs/framework/utils"
 import { generateEntity } from "../utils"
 
 /**
@@ -195,7 +195,7 @@ export class MigrationsExecutionPlanner implements ILinkMigrationsPlanner {
       
       INSERT INTO "${
         this.tableName
-      }" (table_name, link_descriptor) VALUES (?, ?);
+      }" (table_name, link_descriptor) VALUES (?, ?) ON CONFLICT DO NOTHING;
 
       ${sql}
     `,
@@ -513,22 +513,30 @@ export class MigrationsExecutionPlanner implements ILinkMigrationsPlanner {
   async executePlan(actionPlan: LinkMigrationsPlannerAction[]): Promise<void> {
     const orm = await this.createORM()
 
-    await promiseAll(
-      actionPlan.map(async (action) => {
-        switch (action.action) {
-          case "delete":
-            return await this.dropLinkTable(orm, action.tableName)
-          case "create":
-            return await this.createLinkTable(orm, action)
-          case "update":
-            const sql = `SET LOCAL search_path TO "${this.#schema}"; \n\n${
-              action.sql
-            }`
-            return await orm.em.getDriver().getConnection().execute(sql)
-          default:
-            return
-        }
-      })
-    ).finally(() => orm.close(true))
+    try {
+      const concurrency = parseInt(process.env.DB_MIGRATION_CONCURRENCY ?? "1")
+      await executeWithConcurrency(
+        actionPlan.map((action) => {
+          return async () => {
+            switch (action.action) {
+              case "delete":
+                return await this.dropLinkTable(orm, action.tableName)
+              case "create":
+                return await this.createLinkTable(orm, action)
+              case "update":
+                const sql = `SET LOCAL search_path TO "${this.#schema}"; \n\n${
+                  action.sql
+                }`
+                return await orm.em.getDriver().getConnection().execute(sql)
+              default:
+                return
+            }
+          }
+        }),
+        concurrency
+      )
+    } finally {
+      await orm.close(true)
+    }
   }
 }