feat(types, product, utils, medusa): Include shared connection for modules (#4626)

.changeset/lucky-poets-jog.md

@@ -0,0 +1,9 @@
+---
+"@medusajs/medusa": patch
+"@medusajs/modules-sdk": patch
+"@medusajs/product": patch
+"@medusajs/types": patch
+"@medusajs/utils": patch
+---
+
+[WIP] feat(types, product, utils, medusa): Include shared connection for modules

packages/medusa/package.json

@@ -74,6 +74,7 @@
     "ioredis-mock": "8.4.0",
     "iso8601-duration": "^1.3.0",
     "jsonwebtoken": "^9.0.0",
+    "knex": "2.4.2",
     "lodash": "^4.17.21",
     "medusa-core-utils": "^1.2.0",
     "medusa-telemetry": "^0.0.16",
@@ -86,6 +87,7 @@
     "passport-http-bearer": "^1.0.1",
     "passport-jwt": "^4.0.1",
     "passport-local": "^1.0.0",
+    "pg": "^8.11.2",
     "qs": "^6.11.2",
     "randomatic": "^3.1.1",
     "redis": "^3.0.2",

packages/medusa/src/loaders/database.ts

@@ -40,12 +40,17 @@ export default async ({
 }: Options): Promise<DataSource> => {
   const entities = container.resolve("db_entities")
 
+  const connectionString = configModule.projectConfig.database_url
+  const database = configModule.projectConfig.database_database
+  const extra: any = configModule.projectConfig.database_extra || {}
+  const schema = configModule.projectConfig.database_schema || "public"
+
   dataSource = new DataSource({
     type: "postgres",
-    url: configModule.projectConfig.database_url,
+    url: connectionString,
-    database: configModule.projectConfig.database_database,
+    database: database,
-    extra: configModule.projectConfig.database_extra || {},
+    extra,
-    schema: configModule.projectConfig.database_schema,
+    schema,
     entities,
     migrations: customOptions?.migrations,
     logging:

packages/medusa/src/loaders/index.ts

@@ -25,6 +25,8 @@ import subscribersLoader from "./subscribers"
 
 import { moduleLoader, registerModules } from "@medusajs/modules-sdk"
 import { createMedusaContainer } from "medusa-core-utils"
+import pgConnectionLoader from "./pg-connection"
+import { ContainerRegistrationKeys } from "@medusajs/utils"
 
 type Options = {
   directory: string
@@ -44,7 +46,10 @@ export default async ({
   const configModule = loadConfig(rootDirectory)
 
   const container = createMedusaContainer()
-  container.register("configModule", asValue(configModule))
+  container.register(
+    ContainerRegistrationKeys.CONFIG_MODULE,
+    asValue(configModule)
+  )
 
   // Add additional information to context of request
   expressApp.use((req: Request, res: Response, next: NextFunction) => {
@@ -59,7 +64,7 @@ export default async ({
   track("FEATURE_FLAGS_LOADED")
 
   container.register({
-    logger: asValue(Logger),
+    [ContainerRegistrationKeys.LOGGER]: asValue(Logger),
     featureFlagRouter: asValue(featureFlagRouter),
   })
 
@@ -87,6 +92,8 @@ export default async ({
   const stratAct = Logger.success(stratActivity, "Strategies initialized") || {}
   track("STRATEGIES_INIT_COMPLETED", { duration: stratAct.duration })
 
+  await pgConnectionLoader({ container, configModule })
+
   const modulesActivity = Logger.activity(`Initializing modules${EOL}`)
   track("MODULES_INIT_STARTED")
   await moduleLoader({
@@ -112,7 +119,9 @@ export default async ({
   const rAct = Logger.success(repoActivity, "Repositories initialized") || {}
   track("REPOSITORIES_INIT_COMPLETED", { duration: rAct.duration })
 
-  container.register({ manager: asValue(dataSource.manager) })
+  container.register({
+    [ContainerRegistrationKeys.MANAGER]: asValue(dataSource.manager),
+  })
 
   const servicesActivity = Logger.activity(`Initializing services${EOL}`)
   track("SERVICES_INIT_STARTED")

packages/medusa/src/loaders/pg-connection.ts

@@ -0,0 +1,53 @@
+import { asValue, AwilixContainer } from "awilix"
+import { ConfigModule } from "../types/global"
+import { ContainerRegistrationKeys } from "@medusajs/utils"
+import { ClientConfig } from "pg"
+import knex from "knex"
+
+type Options = {
+  configModule: ConfigModule
+  container: AwilixContainer
+}
+
+export default async ({ container, configModule }: Options): Promise<void> => {
+  const connectionString = configModule.projectConfig.database_url
+  const database = configModule.projectConfig.database_database
+  const extra: any = configModule.projectConfig.database_extra || {}
+  const schema = configModule.projectConfig.database_schema || "public"
+
+  // Share a knex connection to be consumed by the shared modules
+  if (!container.hasRegistration(ContainerRegistrationKeys.PG_CONNECTION)) {
+    const config: ClientConfig = {
+      connectionString,
+      database,
+      ssl: extra?.ssl ?? false,
+      idle_in_transaction_session_timeout:
+        extra.idle_in_transaction_session_timeout ?? undefined, // prevent null to be passed
+    }
+
+    // TODO: see later if it is possible to use the same connection with multiple pool config using this shared connection across the modules
+    // const pgConnection = await new Client(config).connect() // or any other method to create your connection here
+
+    const pgConnection = knex<any, any>({
+      client: "pg",
+      searchPath: schema,
+      connection: {
+        connectionString: connectionString,
+        database,
+        ssl: extra?.ssl ?? false,
+        idle_in_transaction_session_timeout:
+          extra.idle_in_transaction_session_timeout ?? undefined, // prevent null to be passed
+      },
+      pool: {
+        min: 0,
+        max: extra.max,
+        idleTimeoutMillis: extra.idleTimeoutMillis ?? undefined, // prevent null to be passed
+      },
+    })
+
+    container.register(
+      ContainerRegistrationKeys.PG_CONNECTION,
+      asValue(pgConnection)
+    )
+  }
+}

packages/modules-sdk/src/definitions.ts

@@ -13,7 +13,19 @@ export enum Modules {
 }
 
 export enum ModuleRegistrationName {
-  EVENT_BUS = "eventBusModuleService"
+  EVENT_BUS = "eventBusModuleService",
+  STOCK_LOCATION = "stockLocationService",
+  INVENTORY = "inventoryService",
+  CACHE = "cacheService",
+  PRODUCT = "productModuleService",
+}
+
+export const MODULE_PACKAGE_NAMES = {
+  [Modules.PRODUCT]: "@medusajs/product",
+  [Modules.EVENT_BUS]: "@medusajs/event-bus-local",
+  [Modules.STOCK_LOCATION]: "@medusajs/stock-location",
+  [Modules.INVENTORY]: "@medusajs/inventory",
+  [Modules.CACHE]: "@medusajs/cache-inmemory",
 }
 
 export const ModulesDefinition: { [key: string | Modules]: ModuleDefinition } =
@@ -21,7 +33,7 @@ export const ModulesDefinition: { [key: string | Modules]: ModuleDefinition } =
     [Modules.EVENT_BUS]: {
       key: Modules.EVENT_BUS,
       registrationName: ModuleRegistrationName.EVENT_BUS,
-      defaultPackage: "@medusajs/event-bus-local",
+      defaultPackage: MODULE_PACKAGE_NAMES[Modules.EVENT_BUS],
       label: "EventBusModuleService",
       canOverride: true,
       isRequired: true,
@@ -33,7 +45,7 @@ export const ModulesDefinition: { [key: string | Modules]: ModuleDefinition } =
     },
     [Modules.STOCK_LOCATION]: {
       key: Modules.STOCK_LOCATION,
-      registrationName: "stockLocationService",
+      registrationName: ModuleRegistrationName.STOCK_LOCATION,
       defaultPackage: false,
       label: "StockLocationService",
       isRequired: false,
@@ -47,7 +59,7 @@ export const ModulesDefinition: { [key: string | Modules]: ModuleDefinition } =
     },
     [Modules.INVENTORY]: {
       key: Modules.INVENTORY,
-      registrationName: "inventoryService",
+      registrationName: ModuleRegistrationName.INVENTORY,
       defaultPackage: false,
       label: "InventoryService",
       isRequired: false,
@@ -61,8 +73,8 @@ export const ModulesDefinition: { [key: string | Modules]: ModuleDefinition } =
     },
     [Modules.CACHE]: {
       key: Modules.CACHE,
-      registrationName: "cacheService",
+      registrationName: ModuleRegistrationName.CACHE,
-      defaultPackage: "@medusajs/cache-inmemory",
+      defaultPackage: MODULE_PACKAGE_NAMES[Modules.CACHE],
       label: "CacheService",
       isRequired: true,
       canOverride: true,
@@ -73,15 +85,16 @@ export const ModulesDefinition: { [key: string | Modules]: ModuleDefinition } =
     },
     [Modules.PRODUCT]: {
       key: Modules.PRODUCT,
-      registrationName: "productModuleService",
+      registrationName: ModuleRegistrationName.PRODUCT,
       defaultPackage: false,
       label: "ProductModuleService",
       isRequired: false,
       canOverride: true,
       isQueryable: true,
-      dependencies: [ModuleRegistrationName.EVENT_BUS],
+      dependencies: [ModuleRegistrationName.EVENT_BUS, "logger"],
       defaultModuleDeclaration: {
-        scope: MODULE_SCOPE.EXTERNAL,
+        scope: MODULE_SCOPE.INTERNAL,
+        resources: MODULE_RESOURCE_TYPE.SHARED,
       },
     },
   }
@@ -89,8 +102,4 @@ export const ModulesDefinition: { [key: string | Modules]: ModuleDefinition } =
 export const MODULE_DEFINITIONS: ModuleDefinition[] =
   Object.values(ModulesDefinition)
 
-export const MODULE_PACKAGE_NAMES = {
-  [Modules.PRODUCT]: "@medusajs/product",
-}
-
 export default MODULE_DEFINITIONS

packages/modules-sdk/src/loaders/utils/load-internal.ts

@@ -8,7 +8,10 @@ import {
   ModuleExports,
   ModuleResolution,
 } from "@medusajs/types"
-import { createMedusaContainer } from "@medusajs/utils"
+import {
+  ContainerRegistrationKeys,
+  createMedusaContainer,
+} from "@medusajs/utils"
 import { asFunction, asValue } from "awilix"
 
 export async function loadInternalModule(
@@ -73,7 +76,12 @@ export async function loadInternalModule(
 
   const dependencies = resolution?.dependencies ?? []
   if (resources === MODULE_RESOURCE_TYPE.SHARED) {
-    dependencies.push("manager", "configModule")
+    dependencies.push(
+      ContainerRegistrationKeys.MANAGER,
+      ContainerRegistrationKeys.CONFIG_MODULE,
+      ContainerRegistrationKeys.LOGGER,
+      ContainerRegistrationKeys.PG_CONNECTION
+    )
   }
 
   for (const dependency of dependencies) {

packages/product/integration-tests/__fixtures__/module.ts

@@ -1,3 +1,4 @@
+import { Context } from "@medusajs/types"
 import { DALUtils } from "@medusajs/utils"
 
 class CustomRepository extends DALUtils.MikroOrmBaseRepository {
@@ -5,12 +6,44 @@ class CustomRepository extends DALUtils.MikroOrmBaseRepository {
     // @ts-ignore
     super(...arguments)
   }
-}
 
-CustomRepository.prototype.find = jest.fn().mockImplementation(async () => [])
+  find = jest.fn().mockImplementation(async () => [])
-CustomRepository.prototype.findAndCount = jest
+  findAndCount = jest.fn().mockImplementation(async () => [])
-  .fn()
+  create = jest.fn()
-  .mockImplementation(async () => [])
+  update = jest.fn()
+  delete = jest.fn()
+  softDelete = jest.fn()
+  restore = jest.fn()
+
+  async transaction<TManager = unknown>(
+    task: (transactionManager: TManager) => Promise<any>,
+    options: {
+      isolationLevel?: string
+      enableNestedTransactions?: boolean
+      transaction?: TManager
+    } = {}
+  ): Promise<any> {
+    return super.transaction(task, options)
+  }
+
+  getActiveManager<TManager = unknown>({
+    transactionManager,
+    manager,
+  }: Context = {}): TManager {
+    return super.getActiveManager({ transactionManager, manager })
+  }
+
+  getFreshManager<TManager = unknown>(): TManager {
+    return super.getFreshManager()
+  }
+
+  async serialize<TOutput extends object | object[]>(
+    data: any,
+    options?: any
+  ): Promise<TOutput> {
+    return super.serialize(data, options)
+  }
+}
 
 export class ProductRepository extends CustomRepository {}
 export class ProductTagRepository extends CustomRepository {}

packages/product/integration-tests/__tests__/module.ts

@@ -1,14 +1,24 @@
 import { MedusaModule } from "@medusajs/modules-sdk"
 import { initialize } from "../../src"
 import * as CustomRepositories from "../__fixtures__/module"
-import { ProductRepository } from "../__fixtures__/module"
+import {
-import { createProductAndTags } from "../__fixtures__/product"
+  buildProductAndRelationsData,
+  createProductAndTags,
+} from "../__fixtures__/product"
 import { productsData } from "../__fixtures__/product/data"
 import { DB_URL, TestDatabase } from "../utils"
-import { buildProductAndRelationsData } from "../__fixtures__/product/data/create-product"
-
-import { IProductModuleService } from "@medusajs/types"
 import { kebabCase } from "@medusajs/utils"
+import { IProductModuleService } from "@medusajs/types"
+import { knex } from "knex"
+import { EventBusService } from "../__fixtures__/event-bus"
+
+const sharedPgConnection = knex<any, any>({
+  client: "pg",
+  searchPath: process.env.MEDUSA_PRODUCT_DB_SCHEMA,
+  connection: {
+    connectionString: DB_URL,
+  },
+})
 
 const beforeEach_ = async () => {
   await TestDatabase.setupDatabase()
@@ -20,6 +30,8 @@ const afterEach_ = async () => {
 }
 
 describe("Product module", function () {
+  const eventBus = new EventBusService()
+
   describe("Using built-in data access layer", function () {
     let module: IProductModuleService
 
@@ -27,12 +39,17 @@ describe("Product module", function () {
       const testManager = await beforeEach_()
       await createProductAndTags(testManager, productsData)
 
-      module = await initialize({
+      module = await initialize(
+        {
           database: {
             clientUrl: DB_URL,
             schema: process.env.MEDUSA_PRODUCT_DB_SCHEMA,
           },
-      })
+        },
+        {
+          eventBusModuleService: eventBus,
+        }
+      )
     })
 
     afterEach(afterEach_)
@@ -41,6 +58,12 @@ describe("Product module", function () {
       expect(module).toBeDefined()
     })
 
+    it("should have a connection that is not the shared connection", async () => {
+      expect(
+        (module as any).baseRepository_.manager_.getConnection().client
+      ).not.toEqual(sharedPgConnection)
+    })
+
     it("should return a list of product", async () => {
       const products = await module.list()
       expect(products).toHaveLength(2)
@@ -48,20 +71,25 @@ describe("Product module", function () {
   })
 
   describe("Using custom data access layer", function () {
-    let module: IProductModuleService
+    let module
 
     beforeEach(async () => {
       const testManager = await beforeEach_()
 
       await createProductAndTags(testManager, productsData)
 
-      module = await initialize({
+      module = await initialize(
+        {
           database: {
             clientUrl: DB_URL,
             schema: process.env.MEDUSA_PRODUCT_DB_SCHEMA,
           },
           repositories: CustomRepositories,
-      })
+        },
+        {
+          eventBusModuleService: eventBus,
+        }
+      )
     })
 
     afterEach(afterEach_)
@@ -70,16 +98,22 @@ describe("Product module", function () {
       expect(module).toBeDefined()
     })
 
+    it("should have a connection that is not the shared connection", async () => {
+      expect(
+        (module as any).baseRepository_.manager_.getConnection().client
+      ).not.toEqual(sharedPgConnection)
+    })
+
     it("should return a list of product", async () => {
       const products = await module.list()
 
-      expect(ProductRepository.prototype.find).toHaveBeenCalled()
+      expect(module.productService_.productRepository_.find).toHaveBeenCalled()
       expect(products).toHaveLength(0)
     })
   })
 
-  describe("Using custom data access layer and connection", function () {
+  describe("Using custom data access layer and manager", function () {
-    let module: IProductModuleService
+    let module
 
     beforeEach(async () => {
       const testManager = await beforeEach_()
@@ -87,10 +121,15 @@ describe("Product module", function () {
 
       MedusaModule.clearInstances()
 
-      module = await initialize({
+      module = await initialize(
+        {
           manager: testManager,
           repositories: CustomRepositories,
-      })
+        },
+        {
+          eventBusModuleService: eventBus,
+        }
+      )
     })
 
     afterEach(afterEach_)
@@ -99,14 +138,54 @@ describe("Product module", function () {
       expect(module).toBeDefined()
     })
 
+    it("should have a connection that is not the shared connection", async () => {
+      expect(
+        (module as any).baseRepository_.manager_.getConnection().client
+      ).not.toEqual(sharedPgConnection)
+    })
+
     it("should return a list of product", async () => {
       const products = await module.list()
 
-      expect(ProductRepository.prototype.find).toHaveBeenCalled()
+      expect(module.productService_.productRepository_.find).toHaveBeenCalled()
       expect(products).toHaveLength(0)
     })
   })
 
+  describe("Using an existing connection", function () {
+    let module: IProductModuleService
+
+    beforeEach(async () => {
+      const testManager = await beforeEach_()
+      await createProductAndTags(testManager, productsData)
+
+      MedusaModule.clearInstances()
+
+      module = await initialize(
+        {
+          database: {
+            connection: sharedPgConnection,
+          },
+        },
+        {
+          eventBusModuleService: eventBus,
+        }
+      )
+    })
+
+    afterEach(afterEach_)
+
+    it("should initialize and return a list of product", async () => {
+      expect(module).toBeDefined()
+    })
+
+    it("should have a connection that is the shared connection", async () => {
+      expect(
+        (module as any).baseRepository_.manager_.getConnection().client
+      ).toEqual(sharedPgConnection)
+    })
+  })
+
   describe("create", function () {
     let module: IProductModuleService
     let images = ["image-1"]
@@ -116,12 +195,17 @@ describe("Product module", function () {
 
       MedusaModule.clearInstances()
 
-      module = await initialize({
+      module = await initialize(
+        {
           database: {
             clientUrl: DB_URL,
             schema: process.env.MEDUSA_PRODUCT_DB_SCHEMA,
           },
-      })
+        },
+        {
+          eventBusModuleService: eventBus,
+        }
+      )
     })
 
     afterEach(afterEach_)
@@ -212,12 +296,17 @@ describe("Product module", function () {
 
       MedusaModule.clearInstances()
 
-      module = await initialize({
+      module = await initialize(
+        {
           database: {
             clientUrl: DB_URL,
             schema: process.env.MEDUSA_PRODUCT_DB_SCHEMA,
           },
-      })
+        },
+        {
+          eventBusModuleService: eventBus,
+        }
+      )
     })
 
     afterEach(afterEach_)
@@ -282,12 +371,17 @@ describe("Product module", function () {
 
       MedusaModule.clearInstances()
 
-      module = await initialize({
+      module = await initialize(
+        {
           database: {
             clientUrl: DB_URL,
             schema: process.env.MEDUSA_PRODUCT_DB_SCHEMA,
           },
-      })
+        },
+        {
+          eventBusModuleService: eventBus,
+        }
+      )
     })
 
     afterEach(afterEach_)

packages/product/package.json

@@ -57,6 +57,7 @@
     "@mikro-orm/postgresql": "5.7.12",
     "awilix": "^8.0.0",
     "dotenv": "^16.1.4",
-    "lodash": "^4.17.21"
+    "lodash": "^4.17.21",
+    "knex": "2.4.2"
   }
 }

packages/product/src/initialize/index.ts

@@ -13,7 +13,8 @@ export const initialize = async (
   options?:
     | ModulesSdkTypes.ModuleServiceInitializeOptions
     | ModulesSdkTypes.ModuleServiceInitializeCustomDataLayerOptions
-    | ExternalModuleDeclaration,
+    | ExternalModuleDeclaration
+    | InternalModuleDeclaration,
   injectedDependencies?: InitializeModuleInjectableDependencies
 ): Promise<IProductModuleService> => {
   const serviceKey = Modules.PRODUCT

packages/product/src/loaders/connection.ts

@@ -1,68 +1,27 @@
-import { asValue } from "awilix"
+import { InternalModuleDeclaration, LoaderOptions } from "@medusajs/modules-sdk"
-
+import { ModulesSdkTypes } from "@medusajs/types"
-import {
+import { ModulesSdkUtils } from "@medusajs/utils"
-  InternalModuleDeclaration,
+import * as ProductModels from "../models"
-  LoaderOptions,
-  MODULE_RESOURCE_TYPE,
-  MODULE_SCOPE,
-} from "@medusajs/modules-sdk"
-import { DALUtils, MedusaError, ModulesSdkUtils } from "@medusajs/utils"
-
 import { EntitySchema } from "@mikro-orm/core"
 
-import * as ProductModels from "@models"
-import { ConfigModule, ModulesSdkTypes } from "@medusajs/types"
-
 export default async (
   {
     options,
     container,
+    logger,
   }: LoaderOptions<
     | ModulesSdkTypes.ModuleServiceInitializeOptions
     | ModulesSdkTypes.ModuleServiceInitializeCustomDataLayerOptions
   >,
   moduleDeclaration?: InternalModuleDeclaration
 ): Promise<void> => {
-  if (
-    moduleDeclaration?.scope === MODULE_SCOPE.INTERNAL &&
-    moduleDeclaration.resources === MODULE_RESOURCE_TYPE.SHARED
-  ) {
-    const { projectConfig } = container.resolve("configModule") as ConfigModule
-    options = {
-      database: {
-        clientUrl: projectConfig.database_url!,
-        driverOptions: projectConfig.database_extra!,
-        schema: projectConfig.database_schema!,
-      },
-    }
-  }
-
-  const customManager = (
-    options as ModulesSdkTypes.ModuleServiceInitializeCustomDataLayerOptions
-  )?.manager
-
-  if (!customManager) {
-    const dbData = ModulesSdkUtils.loadDatabaseConfig("product", options)
-    await loadDefault({ database: dbData, container })
-  } else {
-    container.register({
-      manager: asValue(customManager),
-    })
-  }
-}
-
-async function loadDefault({ database, container }) {
-  if (!database) {
-    throw new MedusaError(
-      MedusaError.Types.INVALID_ARGUMENT,
-      `Database config is not present at module config "options.database"`
-    )
-  }
-
   const entities = Object.values(ProductModels) as unknown as EntitySchema[]
-  const orm = await DALUtils.mikroOrmCreateConnection(database, entities)
 
-  container.register({
+  await ModulesSdkUtils.mikroOrmConnectionLoader({
-    manager: asValue(orm.em.fork()),
+    entities,
+    container,
+    options,
+    moduleDeclaration,
+    logger,
   })
 }

packages/product/src/loaders/container.ts

@@ -51,9 +51,9 @@ export default async ({
   })
 
   if (customRepositories) {
-    await loadCustomRepositories({ customRepositories, container })
+    loadCustomRepositories({ customRepositories, container })
   } else {
-    await loadDefaultRepositories({ container })
+    loadDefaultRepositories({ container })
   }
 }
 

packages/product/src/scripts/migration-down.ts

@@ -16,15 +16,12 @@ export async function revertMigration({
   options,
   logger,
 }: Pick<
-  LoaderOptions<
+  LoaderOptions<ModulesSdkTypes.ModuleServiceInitializeOptions>,
-    | ModulesSdkTypes.ModuleServiceInitializeOptions
-    | ModulesSdkTypes.ModuleServiceInitializeCustomDataLayerOptions
-  >,
   "options" | "logger"
 > = {}) {
   logger ??= console as unknown as Logger
 
-  const dbData = ModulesSdkUtils.loadDatabaseConfig("product", options)
+  const dbData = ModulesSdkUtils.loadDatabaseConfig("product", options)!
   const entities = Object.values(ProductModels) as unknown as EntitySchema[]
 
   const orm = await DALUtils.mikroOrmCreateConnection(dbData, entities)

packages/product/src/scripts/migration-up.ts

@@ -14,15 +14,12 @@ export async function runMigrations({
   options,
   logger,
 }: Pick<
-  LoaderOptions<
+  LoaderOptions<ModulesSdkTypes.ModuleServiceInitializeOptions>,
-    | ModulesSdkTypes.ModuleServiceInitializeOptions
-    | ModulesSdkTypes.ModuleServiceInitializeCustomDataLayerOptions
-  >,
   "options" | "logger"
 > = {}) {
   logger ??= console as unknown as Logger
 
-  const dbData = ModulesSdkUtils.loadDatabaseConfig("product", options)
+  const dbData = ModulesSdkUtils.loadDatabaseConfig("product", options)!
   const entities = Object.values(ProductModels) as unknown as EntitySchema[]
 
   const orm = await DALUtils.mikroOrmCreateConnection(dbData, entities)

packages/product/src/scripts/seed.ts

@@ -13,10 +13,7 @@ export async function run({
   path,
 }: Partial<
   Pick<
-    LoaderOptions<
+    LoaderOptions<ModulesSdkTypes.ModuleServiceInitializeOptions>,
-      | ModulesSdkTypes.ModuleServiceInitializeOptions
-      | ModulesSdkTypes.ModuleServiceInitializeCustomDataLayerOptions
-    >,
     "options" | "logger"
   >
 > & {
@@ -34,7 +31,7 @@ export async function run({
 
   logger ??= console as unknown as Logger
 
-  const dbData = ModulesSdkUtils.loadDatabaseConfig("product", options)
+  const dbData = ModulesSdkUtils.loadDatabaseConfig("product", options)!
   const entities = Object.values(ProductModels) as unknown as EntitySchema[]
 
   const orm = await DALUtils.mikroOrmCreateConnection(dbData, entities)

packages/product/src/services/product-module-service.ts

@@ -22,10 +22,10 @@ import {
   CreateProductOnlyDTO,
   DAL,
   FindConfig,
+  IEventBusModuleService,
   InternalModuleDeclaration,
   JoinerServiceConfig,
   ProductTypes,
-  IEventBusModuleService,
 } from "@medusajs/types"
 
 import ProductImageService from "./product-image"
@@ -739,12 +739,6 @@ export default class ProductModuleService<
         productVariantsMap.set(productData.handle!, variants ?? [])
         productOptionsMap.set(productData.handle!, options ?? [])
 
-        if (!productData.thumbnail && productData.images?.length) {
-          productData.thumbnail = isString(productData.images[0])
-            ? (productData.images[0] as string)
-            : (productData.images[0] as { url: string }).url
-        }
-
         if (productData.is_giftcard) {
           productData.discountable = false
         }
@@ -1084,7 +1078,7 @@ export default class ProductModuleService<
     },
     sharedContext: Context = {}
   ): Promise<Record<Lowercase<keyof typeof LinkableKeys>, string[]> | void> {
-    let [products, cascadedEntitiesMap] = await this.softDelete_(
+    const [products, cascadedEntitiesMap] = await this.softDelete_(
       productIds,
       sharedContext
     )

packages/product/src/types/index.ts

@@ -1,8 +1,9 @@
 export * from "./services"
 
-import { IEventBusModuleService } from "@medusajs/types"
+import { IEventBusModuleService, Logger } from "@medusajs/types"
 
 export type InitializeModuleInjectableDependencies = {
+  logger?: Logger
   eventBusModuleService?: IEventBusModuleService
 }
 

packages/types/src/modules-sdk/index.ts

@@ -108,7 +108,11 @@ export type ModuleExports = {
 
 export interface ModuleServiceInitializeOptions {
   database: {
-    clientUrl: string
+    /**
+     * Forces to use a shared knex connection
+     */
+    connection?: any
+    clientUrl?: string
     schema?: string
     driverOptions?: Record<string, unknown>
     debug?: boolean
@@ -117,5 +121,7 @@ export interface ModuleServiceInitializeOptions {
 
 export type ModuleServiceInitializeCustomDataLayerOptions = {
   manager?: any
-  repositories?: { [key: string]: Constructor<RepositoryService> }
+  repositories?: {
+    [key: string]: Constructor<RepositoryService>
+  }
 }

packages/utils/src/common/container.ts

@@ -0,0 +1,6 @@
+export const ContainerRegistrationKeys = {
+  PG_CONNECTION: "__pg_connection__",
+  MANAGER: "manager",
+  CONFIG_MODULE: "configModule",
+  LOGGER: "logger",
+}

packages/utils/src/common/index.ts

@@ -22,4 +22,4 @@ export * from "./to-kebab-case"
 export * from "./to-pascal-case"
 export * from "./upper-case-first"
 export * from "./wrap-handler"
-
+export * from "./container"

packages/utils/src/dal/mikro-orm/mikro-orm-create-connection.ts

@@ -1,28 +1,39 @@
 import { ModuleServiceInitializeOptions } from "@medusajs/types"
 
 export async function mikroOrmCreateConnection(
-  database: ModuleServiceInitializeOptions["database"],
+  database: ModuleServiceInitializeOptions["database"] & { connection?: any },
   entities: any[]
 ) {
-  const { MikroORM } = await import("@mikro-orm/postgresql")
+  let schema = database.schema || "public"
 
-  const schema = database.schema || "public"
+  let driverOptions = database.driverOptions ?? {
-  const orm = await MikroORM.init({
+    connection: { ssl: true },
+  }
+
+  let clientUrl = database.clientUrl
+
+  if (database.connection) {
+    // Reuse already existing connection
+    // It is important that the knex package version is the same as the one used by MikroORM knex package
+    driverOptions = database.connection
+    clientUrl =
+      database.connection.context.client.config.connection.connectionString
+    schema = database.connection.context.client.config.searchPath
+  }
+
+  const { MikroORM } = await import("@mikro-orm/postgresql")
+  return await MikroORM.init({
     discovery: { disableDynamicFileAccess: true },
     entities,
     debug: database.debug ?? process.env.NODE_ENV?.startsWith("dev") ?? false,
     baseDir: process.cwd(),
-    clientUrl: database.clientUrl,
+    clientUrl,
     schema,
-    driverOptions: database.driverOptions ?? {
+    driverOptions,
-      connection: { ssl: true },
-    },
     tsNode: process.env.APP_ENV === "development",
     type: "postgresql",
     migrations: {
       path: __dirname + "/../migrations",
     },
   })
-
-  return orm
 }

packages/utils/src/modules-sdk/index.ts

@@ -2,3 +2,4 @@ export * from "./load-module-database-config"
 export * from "./decorators"
 export * from "./build-query"
 export * from "./retrieve-entity"
+export * from "./loaders/mikro-orm-connection-loader"

packages/utils/src/modules-sdk/load-module-database-config.ts

@@ -47,9 +47,8 @@ function getDefaultDriverOptions(clientUrl: string) {
  */
 export function loadDatabaseConfig(
   moduleName: string,
-  options?:
+  options?: ModulesSdkTypes.ModuleServiceInitializeOptions,
-    | ModulesSdkTypes.ModuleServiceInitializeOptions
+  silent: boolean = false
-    | ModulesSdkTypes.ModuleServiceInitializeCustomDataLayerOptions
 ): ModulesSdkTypes.ModuleServiceInitializeOptions["database"] {
   const clientUrl = getEnv("POSTGRES_URL", moduleName)
 
@@ -64,19 +63,20 @@ export function loadDatabaseConfig(
   }
 
   if (isModuleServiceInitializeOptions(options)) {
-    database.clientUrl = options.database.clientUrl ?? database.clientUrl
+    database.clientUrl = options.database!.clientUrl ?? database.clientUrl
-    database.schema = options.database.schema ?? database.schema
+    database.schema = options.database!.schema ?? database.schema
     database.driverOptions =
-      options.database.driverOptions ??
+      options.database!.driverOptions ??
       getDefaultDriverOptions(database.clientUrl)
-    database.debug = options.database.debug ?? database.debug
+    database.debug = options.database!.debug ?? database.debug
   }
 
-  if (!database.clientUrl) {
+  if (!database.clientUrl && !silent) {
     throw new MedusaError(
       MedusaError.Types.INVALID_ARGUMENT,
       "No database clientUrl provided. Please provide the clientUrl through the PRODUCT_POSTGRES_URL or POSTGRES_URL environment variable or the options object in the initialize function."
     )
   }
+
   return database
 }

packages/utils/src/modules-sdk/loaders/mikro-orm-connection-loader.ts

@@ -0,0 +1,112 @@
+import {
+  Logger,
+  MedusaContainer,
+  MODULE_RESOURCE_TYPE,
+  MODULE_SCOPE,
+  ModulesSdkTypes,
+} from "@medusajs/types"
+import { asValue } from "awilix"
+import { PostgreSqlDriver, SqlEntityManager } from "@mikro-orm/postgresql"
+import { ContainerRegistrationKeys, MedusaError } from "../../common"
+import { loadDatabaseConfig } from "../load-module-database-config"
+import { mikroOrmCreateConnection } from "../../dal"
+
+export async function mikroOrmConnectionLoader({
+  container,
+  options,
+  moduleDeclaration,
+  entities,
+}: {
+  entities: any[]
+  container: MedusaContainer
+  options?:
+    | ModulesSdkTypes.ModuleServiceInitializeOptions
+    | ModulesSdkTypes.ModuleServiceInitializeCustomDataLayerOptions
+  moduleDeclaration?: ModulesSdkTypes.InternalModuleDeclaration
+  logger?: Logger
+}) {
+  let manager = (
+    options as ModulesSdkTypes.ModuleServiceInitializeCustomDataLayerOptions
+  )?.manager
+
+  // Custom manager provided
+  if (manager) {
+    container.register({
+      manager: asValue(manager),
+    })
+    return
+  }
+
+  if (
+    moduleDeclaration?.scope === MODULE_SCOPE.INTERNAL &&
+    moduleDeclaration.resources === MODULE_RESOURCE_TYPE.SHARED
+  ) {
+    return await loadShared({ container, entities })
+  }
+
+  /**
+   * Reuse an existing connection if it is passed in the options
+   */
+  let dbConfig
+  const shouldSwallowError = !!(
+    options as ModulesSdkTypes.ModuleServiceInitializeOptions
+  )?.database.connection
+  dbConfig = {
+    ...loadDatabaseConfig(
+      "product",
+      (options ?? {}) as ModulesSdkTypes.ModuleServiceInitializeOptions,
+      shouldSwallowError
+    ),
+    connection: (options as ModulesSdkTypes.ModuleServiceInitializeOptions)
+      ?.database.connection,
+  }
+
+  manager ??= await loadDefault({
+    database: dbConfig,
+    entities,
+  })
+
+  container.register({
+    manager: asValue(manager),
+  })
+}
+
+async function loadDefault({
+  database,
+  entities,
+}): Promise<SqlEntityManager<PostgreSqlDriver>> {
+  if (!database) {
+    throw new MedusaError(
+      MedusaError.Types.INVALID_ARGUMENT,
+      `Database config is not present at module config "options.database"`
+    )
+  }
+
+  const orm = await mikroOrmCreateConnection(database, entities)
+
+  return orm.em.fork()
+}
+
+async function loadShared({ container, entities }) {
+  const sharedConnection = container.resolve(
+    ContainerRegistrationKeys.PG_CONNECTION,
+    {
+      allowUnregistered: true,
+    }
+  )
+  if (!sharedConnection) {
+    throw new Error(
+      "The module is setup to use a shared resources but no shared connection is present. A new connection will be created"
+    )
+  }
+
+  const manager = await loadDefault({
+    entities,
+    database: {
+      connection: sharedConnection,
+    },
+  })
+  container.register({
+    manager: asValue(manager),
+  })
+}

yarn.lock

@@ -6384,6 +6384,7 @@ __metadata:
     iso8601-duration: ^1.3.0
     jest: ^25.5.4
     jsonwebtoken: ^9.0.0
+    knex: 2.4.2
     lodash: ^4.17.21
     medusa-core-utils: ^1.2.0
     medusa-interfaces: ^1.3.7
@@ -6397,6 +6398,7 @@ __metadata:
     passport-http-bearer: ^1.0.1
     passport-jwt: ^4.0.1
     passport-local: ^1.0.0
+    pg: ^8.11.2
     qs: ^6.11.2
     randomatic: ^3.1.1
     redis: ^3.0.2
@@ -6514,6 +6516,7 @@ __metadata:
     dotenv: ^16.1.4
     faker: ^6.6.6
     jest: ^25.5.4
+    knex: 2.4.2
     lodash: ^4.17.21
     medusa-test-utils: ^1.1.40
     pg-god: ^1.0.12