chore(orchestration): idempotent (#7771)

packages/modules/workflow-engine-redis/src/index.ts

@@ -3,3 +3,4 @@ import { moduleDefinition } from "./module-definition"
 export default moduleDefinition
 
 export * from "./loaders"
+export * from "./models"

packages/modules/workflow-engine-redis/src/services/workflows-module.ts

@@ -1,26 +1,23 @@
 import {
   Context,
   DAL,
-  FindConfig,
   InternalModuleDeclaration,
-  IWorkflowEngineService,
   ModuleJoinerConfig,
   ModulesSdkTypes,
   WorkflowsSdkTypes,
 } from "@medusajs/types"
 import {
-  InjectManager,
   InjectSharedContext,
-  isString,
   MedusaContext,
-  MedusaError,
+  ModulesSdkUtils,
 } from "@medusajs/utils"
 import type {
   ReturnWorkflow,
   UnwrapWorkflowInputDataType,
 } from "@medusajs/workflows-sdk"
+import { WorkflowExecution } from "@models"
 import { WorkflowOrchestratorService } from "@services"
-import { joinerConfig } from "../joiner-config"
+import { entityNameToLinkableKeysMap, joinerConfig } from "../joiner-config"
 
 type InjectedDependencies = {
   baseRepository: DAL.RepositoryService
@@ -29,9 +26,13 @@ type InjectedDependencies = {
   redisDisconnectHandler: () => Promise<void>
 }
 
-export class WorkflowsModuleService implements IWorkflowEngineService {
+export class WorkflowsModuleService<
+  TWorkflowExecution extends WorkflowExecution = WorkflowExecution
+> extends ModulesSdkUtils.MedusaService<{
+  WorkflowExecution: { dto: WorkflowExecution }
+}>({ WorkflowExecution }, entityNameToLinkableKeysMap) {
   protected baseRepository_: DAL.RepositoryService
-  protected workflowExecutionService_: ModulesSdkTypes.IMedusaInternalService<any>
+  protected workflowExecutionService_: ModulesSdkTypes.IMedusaInternalService<TWorkflowExecution>
   protected workflowOrchestratorService_: WorkflowOrchestratorService
   protected redisDisconnectHandler_: () => Promise<void>
 
@@ -44,6 +45,9 @@ export class WorkflowsModuleService implements IWorkflowEngineService {
     }: InjectedDependencies,
     protected readonly moduleDeclaration: InternalModuleDeclaration
   ) {
+    // @ts-ignore
+    super(...arguments)
+
     this.baseRepository_ = baseRepository
     this.workflowExecutionService_ = workflowExecutionService
     this.workflowOrchestratorService_ = workflowOrchestratorService
@@ -64,122 +68,6 @@ export class WorkflowsModuleService implements IWorkflowEngineService {
     },
   }
 
-  @InjectManager("baseRepository_")
-  async retrieveWorkflowExecution(
-    idOrObject:
-      | string
-      | {
-          workflow_id: string
-          transaction_id: string
-        },
-    config: FindConfig<WorkflowsSdkTypes.WorkflowExecutionDTO> = {},
-    @MedusaContext() sharedContext: Context = {}
-  ): Promise<WorkflowsSdkTypes.WorkflowExecutionDTO> {
-    const objValue = isString(idOrObject)
-      ? { id: idOrObject }
-      : {
-          workflow_id: idOrObject.workflow_id,
-          transaction_id: idOrObject.transaction_id,
-        }
-
-    const wfExecution = await this.workflowExecutionService_.list(
-      objValue,
-      config,
-      sharedContext
-    )
-
-    if (wfExecution.length === 0) {
-      throw new MedusaError(
-        MedusaError.Types.NOT_FOUND,
-        `WorkflowExecution with ${Object.keys(objValue).join(
-          ", "
-        )}: ${Object.values(objValue).join(", ")} was not found`
-      )
-    }
-
-    // eslint-disable-next-line max-len
-    return await this.baseRepository_.serialize<WorkflowsSdkTypes.WorkflowExecutionDTO>(
-      wfExecution[0],
-      {
-        populate: true,
-      }
-    )
-  }
-
-  @InjectManager("baseRepository_")
-  async listWorkflowExecutions(
-    filters: WorkflowsSdkTypes.FilterableWorkflowExecutionProps = {},
-    config: FindConfig<WorkflowsSdkTypes.WorkflowExecutionDTO> = {},
-    @MedusaContext() sharedContext: Context = {}
-  ): Promise<WorkflowsSdkTypes.WorkflowExecutionDTO[]> {
-    if (filters.transaction_id) {
-      if (Array.isArray(filters.transaction_id)) {
-        filters.transaction_id = {
-          $in: filters.transaction_id,
-        }
-      }
-    }
-
-    if (filters.workflow_id) {
-      if (Array.isArray(filters.workflow_id)) {
-        filters.workflow_id = {
-          $in: filters.workflow_id,
-        }
-      }
-    }
-
-    const wfExecutions = await this.workflowExecutionService_.list(
-      filters,
-      config,
-      sharedContext
-    )
-
-    return await this.baseRepository_.serialize<
-      WorkflowsSdkTypes.WorkflowExecutionDTO[]
-    >(wfExecutions, {
-      populate: true,
-    })
-  }
-
-  @InjectManager("baseRepository_")
-  async listAndCountWorkflowExecutions(
-    filters: WorkflowsSdkTypes.FilterableWorkflowExecutionProps = {},
-    config: FindConfig<WorkflowsSdkTypes.WorkflowExecutionDTO> = {},
-    @MedusaContext() sharedContext: Context = {}
-  ): Promise<[WorkflowsSdkTypes.WorkflowExecutionDTO[], number]> {
-    if (filters.transaction_id) {
-      if (Array.isArray(filters.transaction_id)) {
-        filters.transaction_id = {
-          $in: filters.transaction_id,
-        }
-      }
-    }
-
-    if (filters.workflow_id) {
-      if (Array.isArray(filters.workflow_id)) {
-        filters.workflow_id = {
-          $in: filters.workflow_id,
-        }
-      }
-    }
-
-    const [wfExecutions, count] =
-      await this.workflowExecutionService_.listAndCount(
-        filters,
-        config,
-        sharedContext
-      )
-
-    return [
-      await this.baseRepository_.serialize<
-        WorkflowsSdkTypes.WorkflowExecutionDTO[]
-      >(wfExecutions, {
-        populate: true,
-      }),
-      count,
-    ]
-  }
-
   @InjectSharedContext()
   async run<TWorkflow extends string | ReturnWorkflow<any, any, any>>(
     workflowIdOrWorkflow: TWorkflow,

packages/modules/workflow-engine-redis/src/utils/workflow-orchestrator-storage.ts

@@ -4,9 +4,10 @@ import {
   IDistributedTransactionStorage,
   SchedulerOptions,
   TransactionCheckpoint,
+  TransactionOptions,
   TransactionStep,
 } from "@medusajs/orchestration"
-import { ModulesSdkTypes } from "@medusajs/types"
+import { Logger, ModulesSdkTypes } from "@medusajs/types"
 import { MedusaError, TransactionState, promiseAll } from "@medusajs/utils"
 import { WorkflowOrchestratorService } from "@services"
 import { Queue, Worker } from "bullmq"
@@ -19,12 +20,12 @@ enum JobType {
   TRANSACTION_TIMEOUT = "transaction_timeout",
 }
 
-// eslint-disable-next-line max-len
 export class RedisDistributedTransactionStorage
   implements IDistributedTransactionStorage, IDistributedSchedulerStorage
 {
-  private static TTL_AFTER_COMPLETED = 60 * 15 // 15 minutes
+  private static TTL_AFTER_COMPLETED = 60 * 2 // 2 minutes
   private workflowExecutionService_: ModulesSdkTypes.IMedusaInternalService<any>
+  private logger_: Logger
   private workflowOrchestratorService_: WorkflowOrchestratorService
 
   private redisClient: Redis
@@ -36,13 +37,16 @@ export class RedisDistributedTransactionStorage
     redisConnection,
     redisWorkerConnection,
     redisQueueName,
+    logger,
   }: {
     workflowExecutionService: ModulesSdkTypes.IMedusaInternalService<any>
     redisConnection: Redis
     redisWorkerConnection: Redis
     redisQueueName: string
+    logger: Logger
   }) {
     this.workflowExecutionService_ = workflowExecutionService
+    this.logger_ = logger
 
     this.redisClient = redisConnection
 
@@ -131,7 +135,7 @@ export class RedisDistributedTransactionStorage
       })
     } catch (e) {
       if (e instanceof MedusaError && e.type === MedusaError.Types.NOT_FOUND) {
-        console.warn(
+        this.logger_?.warn(
           `Tried to execute a scheduled workflow with ID ${jobId} that does not exist, removing it from the scheduler.`
         )
 
@@ -143,10 +147,42 @@ export class RedisDistributedTransactionStorage
     }
   }
 
-  async get(key: string): Promise<TransactionCheckpoint | undefined> {
+  async get(
+    key: string,
+    options?: TransactionOptions
+  ): Promise<TransactionCheckpoint | undefined> {
     const data = await this.redisClient.get(key)
 
-    return data ? JSON.parse(data) : undefined
+    if (data) {
+      return JSON.parse(data)
+    }
+
+    const { idempotent } = options ?? {}
+    if (!idempotent) {
+      return
+    }
+
+    const [_, workflowId, transactionId] = key.split(":")
+    const trx = await this.workflowExecutionService_
+      .retrieve(
+        {
+          workflow_id: workflowId,
+          transaction_id: transactionId,
+        },
+        {
+          select: ["execution", "context"],
+        }
+      )
+      .catch(() => undefined)
+
+    if (trx) {
+      return {
+        flow: trx.execution,
+        context: trx.context.data,
+        errors: trx.context.errors,
+      }
+    }
+    return
   }
 
   async list(): Promise<TransactionCheckpoint[]> {
@@ -166,10 +202,9 @@ export class RedisDistributedTransactionStorage
   async save(
     key: string,
     data: TransactionCheckpoint,
-    ttl?: number
+    ttl?: number,
+    options?: TransactionOptions
   ): Promise<void> {
-    let retentionTime
-
     /**
      * Store the retention time only if the transaction is done, failed or reverted.
      * From that moment, this tuple can be later on archived or deleted after the retention time.
@@ -180,8 +215,9 @@ export class RedisDistributedTransactionStorage
       TransactionState.REVERTED,
     ].includes(data.flow.state)
 
+    const { retentionTime, idempotent } = options ?? {}
+
     if (hasFinished) {
-      retentionTime = data.flow.options?.retentionTime
       Object.assign(data, {
         retention_time: retentionTime,
       })
@@ -198,14 +234,13 @@ export class RedisDistributedTransactionStorage
       }
     }
 
-    if (hasFinished && !retentionTime) {
+    if (hasFinished && !retentionTime && !idempotent) {
       await this.deleteFromDb(parsedData)
     } else {
       await this.saveToDb(parsedData)
     }
 
     if (hasFinished) {
-      // await this.redisClient.del(key)
       await this.redisClient.set(
         key,
         stringifiedData,