feat(notification): Handle long running transaction and add status support (#8900)

RESOLVES FRMW-2110
RESOLVES FRMW-2095

**What**
- Fix notification module to not retain transactions open while sending notification
- Add support for notification status [success, pending, failure]

packages/modules/notification/src/migrations/.snapshot-medusa-notification.json

@@ -212,6 +212,21 @@
           "nullable": true,
           "mappedType": "text"
         },
+        "status": {
+          "name": "status",
+          "type": "text",
+          "unsigned": false,
+          "autoincrement": false,
+          "primary": false,
+          "nullable": false,
+          "default": "'pending'",
+          "enumItems": [
+            "pending",
+            "success",
+            "failure"
+          ],
+          "mappedType": "enum"
+        },
         "provider_id": {
           "name": "provider_id",
           "type": "text",

packages/modules/notification/src/migrations/Migration20240830094712.ts

@@ -0,0 +1,13 @@
+import { Migration } from '@mikro-orm/migrations';
+
+export class Migration20240830094712 extends Migration {
+
+  async up(): Promise<void> {
+    this.addSql('alter table if exists "notification" add column if not exists "status" text check ("status" in (\'pending\', \'success\', \'failure\')) not null default \'pending\';');
+  }
+
+  async down(): Promise<void> {
+    this.addSql('alter table if exists "notification" drop column if exists "status";');
+  }
+
+}

packages/modules/notification/src/models/notification.ts

@@ -1,4 +1,4 @@
-import { model } from "@medusajs/utils"
+import { model, NotificationStatus } from "@medusajs/utils"
 import { NotificationProvider } from "./notification-provider"
 
 // We probably want to have a TTL for each entry, so we don't bloat the DB (and also for GDPR reasons if TTL < 30 days).
@@ -24,6 +24,9 @@ export const Notification = model.define("notification", {
   idempotency_key: model.text().unique().nullable(),
   // The ID of the notification in the external system, if applicable
   external_id: model.text().nullable(),
+  // The status of the notification
+  status: model.enum(NotificationStatus).default(NotificationStatus.PENDING),
+
   provider: model
     .belongsTo(() => NotificationProvider, { mappedBy: "notifications" })
     .nullable(),

packages/modules/notification/src/services/notification-module-service.ts

@@ -9,11 +9,12 @@ import {
 } from "@medusajs/types"
 import {
   EmitEvents,
+  generateEntityId,
   InjectManager,
-  InjectTransactionManager,
   MedusaContext,
   MedusaError,
   MedusaService,
+  NotificationStatus,
   promiseAll,
 } from "@medusajs/utils"
 import { Notification } from "@models"
@@ -94,7 +95,7 @@ export default class NotificationModuleService
     return Array.isArray(data) ? serialized : serialized[0]
   }
 
-  @InjectTransactionManager("baseRepository_")
+  @InjectManager("baseRepository_")
   protected async createNotifications_(
     data: NotificationTypes.CreateNotificationDTO[],
     @MedusaContext() sharedContext: Context = {}
@@ -108,57 +109,128 @@ export default class NotificationModuleService
     const idempotencyKeys = data
       .map((entry) => entry.idempotency_key)
       .filter(Boolean)
-    const alreadySentNotifications = await this.notificationService_.list(
-      {
-        idempotency_key: idempotencyKeys,
-      },
-      { take: null },
-      sharedContext
-    )
 
-    const existsMap = new Map(
-      alreadySentNotifications.map((n) => [n.idempotency_key as string, true])
-    )
-
-    const notificationsToProcess = data.filter(
-      (entry) => !entry.idempotency_key || !existsMap.has(entry.idempotency_key)
-    )
-
-    const notificationsToCreate = await promiseAll(
-      notificationsToProcess.map(async (entry) => {
-        const provider =
-          await this.notificationProviderService_.getProviderForChannel(
-            entry.channel
-          )
-
-        if (!provider) {
-          throw new MedusaError(
-            MedusaError.Types.NOT_FOUND,
-            `Could not find a notification provider for channel: ${entry.channel}`
-          )
+    let { notificationsToProcess, createdNotifications } =
+      await this.baseRepository_.transaction(async (txManager) => {
+        const context = {
+          ...sharedContext,
+          transactionManager: txManager,
         }
 
-        if (!provider.is_enabled) {
-          throw new MedusaError(
-            MedusaError.Types.NOT_FOUND,
-            `Notification provider ${provider.id} is not enabled. To enable it, configure it as a provider in the notification module options.`
-          )
-        }
-
-        const res = await this.notificationProviderService_.send(
-          provider,
-          entry
+        const alreadySentNotifications = await this.notificationService_.list(
+          {
+            idempotency_key: idempotencyKeys,
+          },
+          { take: null },
+          context
         )
-        return { ...entry, provider_id: provider.id, external_id: res.id }
-      })
-    )
 
-    // Currently we store notifications after they are sent, which might result in a notification being sent that is not registered in the database.
-    // If necessary, we can switch to a two-step process where we first create the notification, send it, and update it after it being sent.
-    const createdNotifications = await this.notificationService_.create(
-      notificationsToCreate,
-      sharedContext
-    )
+        const existsMap = new Map(
+          alreadySentNotifications.map((n) => [n.idempotency_key as string, n])
+        )
+
+        const notificationsToProcess = data.filter(
+          (entry) =>
+            !entry.idempotency_key ||
+            !existsMap.has(entry.idempotency_key) ||
+            (existsMap.has(entry.idempotency_key) &&
+              existsMap.get(entry.idempotency_key)!.status ===
+                NotificationStatus.FAILURE)
+        )
+
+        const channels = notificationsToProcess.map((not) => not.channel)
+        const providers =
+          await this.notificationProviderService_.getProviderForChannels(
+            channels
+          )
+
+        // Create the notifications to be sent to prevent concurrent actions listing the same notifications
+        const normalizedNotificationsToProcess = notificationsToProcess.map(
+          (entry) => {
+            const provider = providers.find((provider) =>
+              provider?.channels.includes(entry.channel)
+            )
+
+            return {
+              provider,
+              data: {
+                id: generateEntityId(undefined, "noti"),
+                ...entry,
+                provider_id: provider?.id,
+              },
+            }
+          }
+        )
+
+        const toCreate = normalizedNotificationsToProcess
+          .filter(
+            (e) =>
+              !e.data.idempotency_key || !existsMap.has(e.data.idempotency_key)
+          )
+          .map((e) => e.data)
+
+        const createdNotifications = toCreate.length
+          ? await this.notificationService_.create(toCreate, context)
+          : []
+
+        return {
+          notificationsToProcess: normalizedNotificationsToProcess,
+          createdNotifications,
+        }
+      })
+
+    const notificationToUpdate: { id: string; external_id?: string }[] = []
+
+    try {
+      await promiseAll(
+        notificationsToProcess.map(async (entry) => {
+          const provider = entry.provider
+
+          if (!provider?.is_enabled) {
+            entry.data.status = NotificationStatus.FAILURE
+            notificationToUpdate.push(entry.data)
+
+            const errorMessage = !provider
+              ? `Could not find a notification provider for channel: ${entry.data.channel} for notification id ${entry.data.id}`
+              : `Notification provider ${provider.id} is not enabled. To enable it, configure it as a provider in the notification module options.`
+
+            throw new MedusaError(MedusaError.Types.NOT_FOUND, errorMessage)
+          }
+
+          const res = await this.notificationProviderService_
+            .send(provider, entry.data)
+            .catch((e) => {
+              entry.data.status = NotificationStatus.FAILURE
+              notificationToUpdate.push(entry.data)
+              throw new MedusaError(
+                MedusaError.Types.UNEXPECTED_STATE,
+                `Failed to send notification with id ${entry.data.id}:\n${e.message}`
+              )
+            })
+
+          entry.data.external_id = res.id
+          entry.data.status = NotificationStatus.SUCCESS
+
+          notificationToUpdate.push(entry.data)
+        }),
+        {
+          aggregateErrors: true,
+        }
+      )
+    } finally {
+      const updatedNotifications = await this.notificationService_.update(
+        notificationToUpdate,
+        sharedContext
+      )
+      const updatedNotificationsMap = new Map(
+        updatedNotifications.map((n) => [n.id, n])
+      )
+
+      // Maintain the order of the notifications
+      createdNotifications = createdNotifications.map((notification) => {
+        return updatedNotificationsMap.get(notification.id) || notification
+      })
+    }
 
     return createdNotifications
   }

packages/modules/notification/src/services/notification-provider.ts

@@ -12,6 +12,8 @@ type InjectedDependencies = {
   ]: NotificationTypes.INotificationProvider
 }
 
+type Provider = InferEntityType<typeof NotificationProvider>
+
 export default class NotificationProviderService extends ModulesSdkUtils.MedusaInternalService<
   InjectedDependencies,
   typeof NotificationProvider
@@ -46,13 +48,13 @@ export default class NotificationProviderService extends ModulesSdkUtils.MedusaI
     }
   }
 
-  async getProviderForChannel(
-    channel: string
-  ): Promise<InferEntityType<typeof NotificationProvider> | undefined> {
+  async getProviderForChannels<
+    TChannel = string | string[],
+    TOutput = TChannel extends string[] ? Provider[] : Provider | undefined
+  >(channels: TChannel): Promise<TOutput> {
     if (!this.providersCache) {
       const providers = await this.notificationProviderRepository_.find()
 
-      type name = (typeof NotificationProvider)["name"]
       this.providersCache = new Map(
         providers.flatMap((provider) =>
           provider.channels.map((c) => [c, provider])
@@ -60,7 +62,12 @@ export default class NotificationProviderService extends ModulesSdkUtils.MedusaI
       )
     }
 
-    return this.providersCache.get(channel)
+    const normalizedChannels = Array.isArray(channels) ? channels : [channels]
+    const results = normalizedChannels
+      .map((channel) => this.providersCache.get(channel))
+      .filter(Boolean)
+
+    return (Array.isArray(channels) ? results : results[0]) as TOutput
   }
 
   async send(