From 8069ed5e99dc53a912df9bb860114d2258044108 Mon Sep 17 00:00:00 2001
From: Sebastian Rindom <skrindom@gmail.com>
Date: Wed, 9 Nov 2022 13:24:26 +0100
Subject: [PATCH] fix(medusa): add support for retrying failed event bus jobs
 (#2566)

Closes CORE-770
---
 .changeset/clever-eels-sit.md             |  5 +++++
 packages/medusa/src/services/event-bus.ts | 19 +++++++++++++++++--
 2 files changed, 22 insertions(+), 2 deletions(-)
 create mode 100644 .changeset/clever-eels-sit.md

diff --git a/.changeset/clever-eels-sit.md b/.changeset/clever-eels-sit.md
new file mode 100644
index 0000000000..0a6ff890bb
--- /dev/null
+++ b/.changeset/clever-eels-sit.md
@@ -0,0 +1,5 @@
+---
+"@medusajs/medusa": patch
+---
+
+fix(medusa): add support for retrying failed event bus jobs
diff --git a/packages/medusa/src/services/event-bus.ts b/packages/medusa/src/services/event-bus.ts
index eaf94ae91d..1f4f5fc234 100644
--- a/packages/medusa/src/services/event-bus.ts
+++ b/packages/medusa/src/services/event-bus.ts
@@ -16,6 +16,15 @@ type InjectedDependencies = {
 
 type Subscriber<T = unknown> = (data: T, eventName: string) => Promise<void>
 
+type EmitOptions = {
+  delay?: number
+  attempts?: number
+  backoff?: {
+    type: "fixed" | "exponential"
+    delay: number
+  }
+}
+
 /**
  * Can keep track of multiple subscribers to different events and run the
  * subscribers when events happen. Events will run asynchronously.
@@ -179,7 +188,7 @@ export default class EventBusService {
   async emit<T>(
     eventName: string,
     data: T,
-    options: { delay?: number } = {}
+    options: EmitOptions = {}
   ): Promise<StagedJob | void> {
     if (this.transactionManager_) {
       const stagedJobRepository = this.transactionManager_.getCustomRepository(
@@ -192,9 +201,15 @@ export default class EventBusService {
       })
       return await stagedJobRepository.save(stagedJobInstance)
     } else {
-      const opts: { removeOnComplete: boolean; delay?: number } = {
+      const opts: { removeOnComplete: boolean } & EmitOptions = {
         removeOnComplete: true,
       }
+      if (typeof options.attempts === "number") {
+        opts.attempts = options.attempts
+        if (typeof options.backoff !== "undefined") {
+          opts.backoff = options.backoff
+        }
+      }
       if (typeof options.delay === "number") {
         opts.delay = options.delay
       }