asyavee/medusa-store
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

chore(workflow-engine): Migrate to DML (#10477)

Browse Source 
RESOLVES FRMW-2832
RESOLVES FRMW-2833

**What**
Migrate workflow engines to DML. Alos includes and update to the linkable generation which now takes into account id and primary keys to generate the linkable instead of only primary keys
This commit is contained in:
Adrien de Peretti
2024-12-06 14:23:07 +01:00
committed by GitHub
parent b0448a7c35
commit 0a077d48e1
16 changed files with 608 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/modules/workflow-engine-inmemory/integration-tests/__tests__/index.spec.ts
+66 -49
View File
@@ -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)
})
packages/modules/workflow-engine-inmemory/src/migrations/.snapshot-medusa-workflows.json
+171
View File
@@ -0,0 +1,171 @@
{
"namespaces": [
"public"
],
"name": "public",
"tables": [
{
"columns": {
"workflow_id": {
"name": "workflow_id",
"type": "text",
"unsigned": false,
"autoincrement": false,
"primary": false,
"nullable": false,
"mappedType": "text"
},
"transaction_id": {
"name": "transaction_id",
"type": "text",
"unsigned": false,
"autoincrement": false,
"primary": false,
"nullable": false,
"mappedType": "text"
},
"id": {
"name": "id",
"type": "text",
"unsigned": false,
"autoincrement": false,
"primary": false,
"nullable": false,
"mappedType": "text"
},
"execution": {
"name": "execution",
"type": "jsonb",
"unsigned": false,
"autoincrement": false,
"primary": false,
"nullable": true,
"mappedType": "json"
},
"context": {
"name": "context",
"type": "jsonb",
"unsigned": false,
"autoincrement": false,
"primary": false,
"nullable": true,
"mappedType": "json"
},
"state": {
"name": "state",
"type": "text",
"unsigned": false,
"autoincrement": false,
"primary": false,
"nullable": false,
"enumItems": [
"not_started",
"invoking",
"waiting_to_compensate",
"compensating",
"done",
"reverted",
"failed"
],
"mappedType": "enum"
},
"created_at": {
"name": "created_at",
"type": "timestamptz",
"unsigned": false,
"autoincrement": false,
"primary": false,
"nullable": false,
"length": 6,
"default": "now()",
"mappedType": "datetime"
},
"updated_at": {
"name": "updated_at",
"type": "timestamptz",
"unsigned": false,
"autoincrement": false,
"primary": false,
"nullable": false,
"length": 6,
"default": "now()",
"mappedType": "datetime"
},
"deleted_at": {
"name": "deleted_at",
"type": "timestamptz",
"unsigned": false,
"autoincrement": false,
"primary": false,
"nullable": true,
"length": 6,
"mappedType": "datetime"
}
},
"name": "workflow_execution",
"schema": "public",
"indexes": [
{
"keyName": "IDX_workflow_execution_deleted_at",
"columnNames": [],
"composite": false,
"primary": false,
"unique": false,
"expression": "CREATE INDEX IF NOT EXISTS \"IDX_workflow_execution_deleted_at\" ON \"workflow_execution\" (deleted_at) WHERE deleted_at IS NULL"
},
{
"keyName": "IDX_workflow_execution_id",
"columnNames": [],
"composite": false,
"primary": false,
"unique": false,
"expression": "CREATE INDEX IF NOT EXISTS \"IDX_workflow_execution_id\" ON \"workflow_execution\" (id) WHERE deleted_at IS NULL"
},
{
"keyName": "IDX_workflow_execution_workflow_id_transaction_id_unique",
"columnNames": [],
"composite": false,
"primary": false,
"unique": false,
"expression": "CREATE UNIQUE INDEX IF NOT EXISTS \"IDX_workflow_execution_workflow_id_transaction_id_unique\" ON \"workflow_execution\" (workflow_id, transaction_id) WHERE deleted_at IS NULL"
},
{
"keyName": "IDX_workflow_execution_workflow_id",
"columnNames": [],
"composite": false,
"primary": false,
"unique": false,
"expression": "CREATE INDEX IF NOT EXISTS \"IDX_workflow_execution_workflow_id\" ON \"workflow_execution\" (workflow_id) WHERE deleted_at IS NULL"
},
{
"keyName": "IDX_workflow_execution_transaction_id",
"columnNames": [],
"composite": false,
"primary": false,
"unique": false,
"expression": "CREATE INDEX IF NOT EXISTS \"IDX_workflow_execution_transaction_id\" ON \"workflow_execution\" (transaction_id) WHERE deleted_at IS NULL"
},
{
"keyName": "IDX_workflow_execution_state",
"columnNames": [],
"composite": false,
"primary": false,
"unique": false,
"expression": "CREATE INDEX IF NOT EXISTS \"IDX_workflow_execution_state\" ON \"workflow_execution\" (state) WHERE deleted_at IS NULL"
},
{
"keyName": "workflow_execution_pkey",
"columnNames": [
"workflow_id",
"transaction_id"
],
"composite": true,
"primary": true,
"unique": true
}
],
"checks": [],
"foreignKeys": {}
}
]
}
packages/modules/workflow-engine-inmemory/src/migrations/Migration20241206101446.ts
+27
View File
@@ -0,0 +1,27 @@
import { Migration } from "@mikro-orm/migrations"
export class Migration20241206101446 extends Migration {
async up(): Promise<void> {
this.addSql(
`DROP INDEX IF EXISTS "IDX_workflow_execution_id";
DROP INDEX IF EXISTS "IDX_workflow_execution_workflow_id";
DROP INDEX IF EXISTS "IDX_workflow_execution_transaction_id";
DROP INDEX IF EXISTS "IDX_workflow_execution_state";`
)
this.addSql(
'CREATE INDEX IF NOT EXISTS "IDX_workflow_execution_deleted_at" ON "workflow_execution" (deleted_at) WHERE deleted_at IS NULL;'
)
this.addSql(
'CREATE INDEX IF NOT EXISTS "IDX_workflow_execution_id" ON "workflow_execution" (id) WHERE deleted_at IS NULL;'
)
this.addSql(
'CREATE INDEX IF NOT EXISTS "IDX_workflow_execution_workflow_id" ON "workflow_execution" (workflow_id) WHERE deleted_at IS NULL;'
)
this.addSql(
'CREATE INDEX IF NOT EXISTS "IDX_workflow_execution_transaction_id" ON "workflow_execution" (transaction_id) WHERE deleted_at IS NULL;'
)
this.addSql(
'CREATE INDEX IF NOT EXISTS "IDX_workflow_execution_state" ON "workflow_execution" (state) WHERE deleted_at IS NULL;'
)
}
}
packages/modules/workflow-engine-inmemory/src/models/index.ts
+1 -1
View File
@@ -1 +1 @@
export { default as WorkflowExecution } from "./workflow-execution"
export { WorkflowExecution } from "./workflow-execution"
packages/modules/workflow-engine-inmemory/src/models/workflow-execution.ts
+27 -73
View File
@@ -1,76 +1,30 @@
import { TransactionState } from "@medusajs/framework/orchestration"
import { DALUtils, generateEntityId } from "@medusajs/framework/utils"
import {
BeforeCreate,
Entity,
Enum,
Filter,
Index,
OnInit,
OptionalProps,
PrimaryKey,
Property,
Unique,
} from "@mikro-orm/core"
import { model } from "@medusajs/framework/utils"
type OptionalFields = "deleted_at"
@Entity()
@Unique({
name: "IDX_workflow_execution_workflow_id_transaction_id_unique",
properties: ["workflow_id", "transaction_id"],
})
@Filter(DALUtils.mikroOrmSoftDeletableFilterOptions)
export default class WorkflowExecution {
[OptionalProps]?: OptionalFields
@Property({ columnType: "text", nullable: false })
@Index({ name: "IDX_workflow_execution_id" })
id!: string
@Index({ name: "IDX_workflow_execution_workflow_id" })
@PrimaryKey({ columnType: "text" })
workflow_id: string
@Index({ name: "IDX_workflow_execution_transaction_id" })
@PrimaryKey({ columnType: "text" })
transaction_id: string
@Property({ columnType: "jsonb", nullable: true })
execution: Record<string, unknown> | null = null
@Property({ columnType: "jsonb", nullable: true })
context: Record<string, unknown> | null = null
@Index({ name: "IDX_workflow_execution_state" })
@Enum(() => TransactionState)
state: TransactionState
@Property({
onCreate: () => new Date(),
columnType: "timestamptz",
defaultRaw: "now()",
export const WorkflowExecution = model
.define("workflow_execution", {
id: model.id({ prefix: "wf_exec" }),
workflow_id: model.text().primaryKey(),
transaction_id: model.text().primaryKey(),
execution: model.json().nullable(),
context: model.json().nullable(),
state: model.enum(TransactionState),
})
created_at: Date
@Property({
onCreate: () => new Date(),
onUpdate: () => new Date(),
columnType: "timestamptz",
defaultRaw: "now()",
})
updated_at: Date
@Property({ columnType: "timestamptz", nullable: true })
deleted_at: Date | null = null
@BeforeCreate()
onCreate() {
this.id = generateEntityId(this.id, "wf_exec")
}
@OnInit()
onInit() {
this.id = generateEntityId(this.id, "wf_exec")
}
}
.indexes([
{
on: ["id"],
where: "deleted_at IS NULL",
},
{
on: ["workflow_id"],
where: "deleted_at IS NULL",
},
{
on: ["transaction_id"],
where: "deleted_at IS NULL",
},
{
on: ["state"],
where: "deleted_at IS NULL",
},
])
packages/modules/workflow-engine-inmemory/src/services/workflows-module.ts
+5 -2
View File
@@ -1,6 +1,7 @@
import {
Context,
DAL,
InferEntityType,
InternalModuleDeclaration,
MedusaContainer,
ModulesSdkTypes,
@@ -25,9 +26,11 @@ type InjectedDependencies = {
}
export class WorkflowsModuleService<
TWorkflowExecution extends WorkflowExecution = WorkflowExecution
TWorkflowExecution extends InferEntityType<
typeof WorkflowExecution
> = InferEntityType<typeof WorkflowExecution>
> extends ModulesSdkUtils.MedusaService<{
WorkflowExecution: { dto: WorkflowExecution }
WorkflowExecution: { dto: InferEntityType<typeof WorkflowExecution> }
}>({ WorkflowExecution }) {
protected baseRepository_: DAL.RepositoryService
protected workflowExecutionService_: ModulesSdkTypes.IMedusaInternalService<TWorkflowExecution>