chore(workflow-engine): Migrate to DML

.changeset/clean-paws-build.md

@@ -0,0 +1,7 @@
+---
+"@medusajs/workflow-engine-inmemory": patch
+"@medusajs/workflow-engine-redis": patch
+"@medusajs/utils": patch
+---
+
+chore(workflow-engine): Migrate to DML

packages/core/utils/src/modules-sdk/__tests__/joiner-config-builder.spec.ts

@@ -668,8 +668,8 @@ describe("joiner-config-builder", () => {
         serviceName: "myService",
         field: "car",
         entity: "Car",
-        linkable: "car_number_plate",
-        primaryKey: "number_plate",
+        linkable: "car_id",
+        primaryKey: "id",
       })
       expect(linkConfig.user.toJSON()).toEqual({
         serviceName: "myService",

packages/core/utils/src/modules-sdk/joiner-config-builder.ts

@@ -17,7 +17,7 @@ import {
   toCamelCase,
   upperCaseFirst,
 } from "../common"
-import { DmlEntity } from "../dml"
+import { DmlEntity, IdProperty } from "../dml"
 import { toGraphQLSchema } from "../dml/helpers/create-graphql"
 import { PrimaryKeyModifier } from "../dml/properties/primary-key"
 import { BaseRelationship } from "../dml/relations/base"
@@ -396,7 +396,10 @@ export function buildLinkConfigFromModelObjects<
       }
 
       const parsedProperty = (value as PropertyType<any>).parse(property)
-      if (PrimaryKeyModifier.isPrimaryKeyModifier(value)) {
+      if (
+        PrimaryKeyModifier.isPrimaryKeyModifier(value) ||
+        IdProperty.isIdProperty(value)
+      ) {
         const linkableKeyName =
           parsedProperty.dataType.options?.linkable ??
           `${camelToSnakeCase(model.name).toLowerCase()}_${property}`

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

@@ -1,4 +1,7 @@
-import { WorkflowManager } from "@medusajs/framework/orchestration"
+import {
+  DistributedTransactionType,
+  WorkflowManager,
+} from "@medusajs/framework/orchestration"
 import {
   Context,
   IWorkflowEngineService,
@@ -60,6 +63,20 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
               serviceName: "workflows",
               field: "workflowExecution",
             },
+            transaction_id: {
+              linkable: "workflow_execution_transaction_id",
+              entity: "WorkflowExecution",
+              primaryKey: "transaction_id",
+              serviceName: "workflows",
+              field: "workflowExecution",
+            },
+            workflow_id: {
+              linkable: "workflow_execution_workflow_id",
+              entity: "WorkflowExecution",
+              primaryKey: "workflow_id",
+              serviceName: "workflows",
+              field: "workflowExecution",
+            },
           },
         })
       })
@@ -87,12 +104,12 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
         })
 
         // Validate context event group id
-        expect(workflowEventGroupIdStep1Mock.mock.calls[0][1]).toEqual(
-          expect.objectContaining({ eventGroupId })
-        )
-        expect(workflowEventGroupIdStep2Mock.mock.calls[0][1]).toEqual(
-          expect.objectContaining({ eventGroupId })
-        )
+        expect(
+          (workflowEventGroupIdStep1Mock.mock.calls[0] as any[])[1]
+        ).toEqual(expect.objectContaining({ eventGroupId }))
+        expect(
+          (workflowEventGroupIdStep2Mock.mock.calls[0] as any[])[1]
+        ).toEqual(expect.objectContaining({ eventGroupId }))
       })
 
       it("should execute an async workflow keeping track of the event group id that has been auto generated", async () => {
@@ -114,14 +131,19 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
           stepResponse: { hey: "oh" },
         })
 
-        const generatedEventGroupId = (workflowEventGroupIdStep1Mock.mock
-          .calls[0][1] as unknown as Context)!.eventGroupId
+        const generatedEventGroupId = ((
+          workflowEventGroupIdStep1Mock.mock.calls[0] as any[]
+        )[1] as unknown as Context)!.eventGroupId
 
         // Validate context event group id
-        expect(workflowEventGroupIdStep1Mock.mock.calls[0][1]).toEqual(
+        expect(
+          (workflowEventGroupIdStep1Mock.mock.calls[0] as any[])[1]
+        ).toEqual(
           expect.objectContaining({ eventGroupId: generatedEventGroupId })
         )
-        expect(workflowEventGroupIdStep2Mock.mock.calls[0][1]).toEqual(
+        expect(
+          (workflowEventGroupIdStep2Mock.mock.calls[0] as any[])[1]
+        ).toEqual(
           expect.objectContaining({ eventGroupId: generatedEventGroupId })
         )
       })
@@ -139,10 +161,9 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
             throwOnError: true,
           })
 
-          let executionsList = await query({
-            workflow_executions: {
-              fields: ["workflow_id", "transaction_id", "state"],
-            },
+          let { data: executionsList } = await query.graph({
+            entity: "workflow_executions",
+            fields: ["workflow_id", "transaction_id", "state"],
           })
 
           expect(executionsList).toHaveLength(1)
@@ -157,11 +178,10 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
             stepResponse: { uhuuuu: "yeaah!" },
           })
 
-          executionsList = await query({
-            workflow_executions: {
-              fields: ["id"],
-            },
-          })
+          ;({ data: executionsList } = await query.graph({
+            entity: "workflow_executions",
+            fields: ["id"],
+          }))
 
           expect(executionsList).toHaveLength(0)
           expect(result).toEqual({
@@ -180,10 +200,9 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
             transactionId: "transaction_1",
           })
 
-          let executionsList = await query({
-            workflow_executions: {
-              fields: ["id"],
-            },
+          let { data: executionsList } = await query.graph({
+            entity: "workflow_executions",
+            fields: ["id"],
           })
 
           expect(executionsList).toHaveLength(1)
@@ -208,40 +227,38 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
           expect(workflow2Step3Invoke.mock.calls[0][0]).toEqual({
             uhuuuu: "yeaah!",
           })
-
-          executionsList = await query({
-            workflow_executions: {
-              fields: ["id"],
-            },
-          })
+          ;({ data: executionsList } = await query.graph({
+            entity: "workflow_executions",
+            fields: ["id"],
+          }))
 
           expect(executionsList).toHaveLength(1)
         })
 
         it("should revert the entire transaction when a step timeout expires", async () => {
-          const { transaction } = await workflowOrcModule.run(
+          const { transaction } = (await workflowOrcModule.run(
             "workflow_step_timeout",
             {
               input: {},
               throwOnError: false,
             }
-          )
+          )) as Awaited<{ transaction: DistributedTransactionType }>
 
-          expect(transaction.flow.state).toEqual("reverted")
+          expect(transaction.getFlow().state).toEqual("reverted")
         })
 
         it("should revert the entire transaction when the transaction timeout expires", async () => {
-          const { transaction } = await workflowOrcModule.run(
+          const { transaction } = (await workflowOrcModule.run(
             "workflow_transaction_timeout",
             {
               input: {},
               throwOnError: false,
             }
-          )
+          )) as Awaited<{ transaction: DistributedTransactionType }>
 
           await setTimeoutPromise(200)
 
-          expect(transaction.flow.state).toEqual("reverted")
+          expect(transaction.getFlow().state).toEqual("reverted")
         })
 
         it.skip("should subscribe to a async workflow and receive the response when it finishes", (done) => {
@@ -393,7 +410,7 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
         })
 
         it("should fetch an idempotent workflow after its completion", async () => {
-          const { transaction: firstRun } = await workflowOrcModule.run(
+          const { transaction: firstRun } = (await workflowOrcModule.run(
             "workflow_idempotent",
             {
               input: {
@@ -402,15 +419,14 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
               throwOnError: true,
               transactionId: "transaction_1",
             }
-          )
+          )) as Awaited<{ transaction: DistributedTransactionType }>
 
-          let executionsList = await query({
-            workflow_executions: {
-              fields: ["id"],
-            },
+          let { data: executionsList } = await query.graph({
+            entity: "workflow_executions",
+            fields: ["id"],
           })
 
-          const { transaction: secondRun } = await workflowOrcModule.run(
+          const { transaction: secondRun } = (await workflowOrcModule.run(
             "workflow_idempotent",
             {
               input: {
@@ -419,15 +435,16 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
               throwOnError: true,
               transactionId: "transaction_1",
             }
-          )
+          )) as Awaited<{ transaction: DistributedTransactionType }>
 
-          const executionsListAfter = await query({
-            workflow_executions: {
-              fields: ["id"],
-            },
+          const { data: executionsListAfter } = await query.graph({
+            entity: "workflow_executions",
+            fields: ["id"],
           })
 
-          expect(secondRun.flow.startedAt).toEqual(firstRun.flow.startedAt)
+          expect(secondRun.getFlow().startedAt).toEqual(
+            firstRun.getFlow().startedAt
+          )
           expect(executionsList).toHaveLength(1)
           expect(executionsListAfter).toHaveLength(1)
         })