chore(workflows-sdk): add idempotencyKey to shared context (#6860)
This commit is contained in:
1 parent
412111e1ea
commit
45c49e89f2
5 files changed
+61
-29
No files matched your search
@@ -0,0 +1,6 @@
|
|||||||
|
---
|
||||||
|
"@medusajs/workflows-sdk": patch
|
||||||
|
"@medusajs/types": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
Add workflow idempotencyKey to shared context
|
||||||
@@ -32,7 +32,7 @@ export type WorkflowHandler = Map<
|
|||||||
{ invoke: WorkflowStepHandler; compensate?: WorkflowStepHandler }
|
{ invoke: WorkflowStepHandler; compensate?: WorkflowStepHandler }
|
||||||
>
|
>
|
||||||
|
|
||||||
export type WorkflowStepHandler = (args: {
|
export type WorkflowStepHandlerArguments = {
|
||||||
container: MedusaContainer
|
container: MedusaContainer
|
||||||
payload: unknown
|
payload: unknown
|
||||||
invoke: { [actions: string]: unknown }
|
invoke: { [actions: string]: unknown }
|
||||||
@@ -40,7 +40,11 @@ export type WorkflowStepHandler = (args: {
|
|||||||
metadata: TransactionMetadata
|
metadata: TransactionMetadata
|
||||||
transaction: DistributedTransaction
|
transaction: DistributedTransaction
|
||||||
context?: Context
|
context?: Context
|
||||||
}) => unknown
|
}
|
||||||
|
|
||||||
|
export type WorkflowStepHandler = (
|
||||||
|
args: WorkflowStepHandlerArguments
|
||||||
|
) => unknown
|
||||||
|
|
||||||
export class WorkflowManager {
|
export class WorkflowManager {
|
||||||
protected static workflows: Map<string, WorkflowDefinition> = new Map()
|
protected static workflows: Map<string, WorkflowDefinition> = new Map()
|
||||||
|
|||||||
@@ -76,4 +76,9 @@ export type Context<TManager = unknown> = {
|
|||||||
* A string indicating the ID of the current request.
|
* A string indicating the ID of the current request.
|
||||||
*/
|
*/
|
||||||
requestId?: string
|
requestId?: string
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A string indicating the idempotencyKey of the current workflow execution.
|
||||||
|
*/
|
||||||
|
idempotencyKey?: string
|
||||||
}
|
}
|
||||||
@@ -1,6 +1,7 @@
|
|||||||
import {
|
import {
|
||||||
TransactionStepsDefinition,
|
TransactionStepsDefinition,
|
||||||
WorkflowManager,
|
WorkflowManager,
|
||||||
|
WorkflowStepHandlerArguments,
|
||||||
} from "@medusajs/orchestration"
|
} from "@medusajs/orchestration"
|
||||||
import { OrchestrationUtils, deepCopy, isString } from "@medusajs/utils"
|
import { OrchestrationUtils, deepCopy, isString } from "@medusajs/utils"
|
||||||
import { ulid } from "ulid"
|
import { ulid } from "ulid"
|
||||||
@@ -120,16 +121,20 @@ function applyStep<
|
|||||||
}
|
}
|
||||||
|
|
||||||
const handler = {
|
const handler = {
|
||||||
invoke: async (transactionContext) => {
|
invoke: async (transactionContext: WorkflowStepHandlerArguments) => {
|
||||||
|
const metadata = transactionContext.metadata
|
||||||
|
const idempotencyKey = metadata.idempotency_key
|
||||||
|
|
||||||
|
transactionContext.context!.idempotencyKey = idempotencyKey
|
||||||
const executionContext: StepExecutionContext = {
|
const executionContext: StepExecutionContext = {
|
||||||
workflowId: transactionContext.model_id,
|
workflowId: metadata.model_id,
|
||||||
stepName: transactionContext.action,
|
stepName: metadata.action,
|
||||||
action: "invoke",
|
action: "invoke",
|
||||||
idempotencyKey: transactionContext.idempotency_key,
|
idempotencyKey,
|
||||||
attempt: transactionContext.attempt,
|
attempt: metadata.attempt,
|
||||||
container: transactionContext.container,
|
container: transactionContext.container,
|
||||||
metadata: transactionContext.metadata,
|
metadata,
|
||||||
context: transactionContext.context,
|
context: transactionContext.context!,
|
||||||
}
|
}
|
||||||
|
|
||||||
const argInput = input
|
const argInput = input
|
||||||
@@ -151,19 +156,25 @@ function applyStep<
|
|||||||
}
|
}
|
||||||
},
|
},
|
||||||
compensate: compensateFn
|
compensate: compensateFn
|
||||||
? async (transactionContext) => {
|
? async (transactionContext: WorkflowStepHandlerArguments) => {
|
||||||
|
const metadata = transactionContext.metadata
|
||||||
|
const idempotencyKey = metadata.idempotency_key
|
||||||
|
|
||||||
|
transactionContext.context!.idempotencyKey = idempotencyKey
|
||||||
|
|
||||||
const executionContext: StepExecutionContext = {
|
const executionContext: StepExecutionContext = {
|
||||||
workflowId: transactionContext.model_id,
|
workflowId: metadata.model_id,
|
||||||
stepName: transactionContext.action,
|
stepName: metadata.action,
|
||||||
action: "compensate",
|
action: "compensate",
|
||||||
idempotencyKey: transactionContext.idempotency_key,
|
idempotencyKey,
|
||||||
attempt: transactionContext.attempt,
|
attempt: metadata.attempt,
|
||||||
container: transactionContext.container,
|
container: transactionContext.container,
|
||||||
metadata: transactionContext.metadata,
|
metadata,
|
||||||
context: transactionContext.context,
|
context: transactionContext.context!,
|
||||||
}
|
}
|
||||||
|
|
||||||
const stepOutput = transactionContext.invoke[stepName]?.output
|
const stepOutput = (transactionContext.invoke[stepName] as any)
|
||||||
|
?.output
|
||||||
const invokeResult =
|
const invokeResult =
|
||||||
stepOutput?.__type ===
|
stepOutput?.__type ===
|
||||||
OrchestrationUtils.SymbolWorkflowStepResponse
|
OrchestrationUtils.SymbolWorkflowStepResponse
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
import { OrchestrationUtils } from "@medusajs/utils"
|
import { WorkflowStepHandlerArguments } from "@medusajs/orchestration"
|
||||||
|
import { OrchestrationUtils, deepCopy } from "@medusajs/utils"
|
||||||
import { resolveValue } from "./helpers"
|
import { resolveValue } from "./helpers"
|
||||||
import {
|
import {
|
||||||
CreateWorkflowComposerContext,
|
CreateWorkflowComposerContext,
|
||||||
@@ -108,22 +109,27 @@ export function hook<TOutput>(
|
|||||||
|
|
||||||
return hookBinder(name, function (context) {
|
return hookBinder(name, function (context) {
|
||||||
return {
|
return {
|
||||||
__value: async function (transactionContext) {
|
__value: async function (
|
||||||
|
transactionContext: WorkflowStepHandlerArguments
|
||||||
|
) {
|
||||||
|
const metadata = transactionContext.metadata
|
||||||
|
const idempotencyKey = metadata.idempotency_key
|
||||||
|
|
||||||
|
transactionContext.context!.idempotencyKey = idempotencyKey
|
||||||
|
|
||||||
const executionContext: StepExecutionContext = {
|
const executionContext: StepExecutionContext = {
|
||||||
workflowId: transactionContext.model_id,
|
workflowId: metadata.model_id,
|
||||||
stepName: transactionContext.action,
|
stepName: metadata.action,
|
||||||
action: transactionContext.action_type,
|
action: metadata.action_type,
|
||||||
idempotencyKey: transactionContext.idempotency_key,
|
idempotencyKey,
|
||||||
attempt: transactionContext.attempt,
|
attempt: metadata.attempt,
|
||||||
container: transactionContext.container,
|
container: transactionContext.container,
|
||||||
metadata: transactionContext.metadata,
|
metadata,
|
||||||
context: transactionContext.context,
|
context: transactionContext.context!,
|
||||||
}
|
}
|
||||||
|
|
||||||
const allValues = await resolveValue(value, transactionContext)
|
const allValues = await resolveValue(value, transactionContext)
|
||||||
const stepValue = allValues
|
const stepValue = allValues ? deepCopy(allValues) : allValues
|
||||||
? JSON.parse(JSON.stringify(allValues))
|
|
||||||
: allValues
|
|
||||||
|
|
||||||
let finalResult
|
let finalResult
|
||||||
const functions = context.hooksCallback_[name]
|
const functions = context.hooksCallback_[name]
|
||||||
|
|||||||
Reference in new issue
Block a user