feat: notifications (#172)

The Notifications API allows plugins to register Notification Providers which have `sendNotification` and `resendNotification`.

Each plugin can listen to any events transmittet over the event bus and the result of the notification send will be persisted in the database to allow for clear communications timeline + ability to resend notifications.

packages/medusa/src/services/__mocks__/event-bus.js

@@ -1,6 +1,9 @@
 export const EventBusServiceMock = {
   emit: jest.fn(),
   subscribe: jest.fn(),
+  withTransaction: function() {
+    return this
+  },
 }
 
 const mock = jest.fn().mockImplementation(() => {

packages/medusa/src/services/__tests__/notification.js

@@ -0,0 +1,54 @@
+import NotificationService from "../notification"
+import { IdMap, MockManager, MockRepository } from "medusa-test-utils"
+
+describe("NotificationService", () => {
+  describe("send", () => {
+    const notificationRepository = MockRepository({ create: c => c })
+
+    const container = {
+      manager: MockManager,
+      notificationRepository,
+      noti_test: {
+        sendNotification: jest.fn(() =>
+          Promise.resolve({
+            to: "test@mail.com",
+            data: { id: "something" },
+          })
+        ),
+      },
+    }
+
+    const notificationService = new NotificationService(container)
+
+    beforeEach(() => {
+      jest.clearAllMocks()
+    })
+
+    it("successfully calls provider and saves noti", async () => {
+      await notificationService.send("event.test", { id: "test" }, "test")
+
+      expect(container.noti_test.sendNotification).toHaveBeenCalledTimes(1)
+      expect(container.noti_test.sendNotification).toHaveBeenCalledWith(
+        "event.test",
+        { id: "test" },
+        null
+      )
+
+      const constructed = {
+        resource_type: "event",
+        resource_id: "test",
+        customer_id: null,
+        to: "test@mail.com",
+        data: { id: "something" },
+        event_name: "event.test",
+        provider_id: "test",
+      }
+
+      expect(notificationRepository.create).toHaveBeenCalledTimes(1)
+      expect(notificationRepository.create).toHaveBeenCalledWith(constructed)
+
+      expect(notificationRepository.save).toHaveBeenCalledTimes(1)
+      expect(notificationRepository.save).toHaveBeenCalledWith(constructed)
+    })
+  })
+})

packages/medusa/src/services/__tests__/order.js

@@ -10,7 +10,23 @@ describe("OrderService", () => {
     getRefundedTotal: o => {
       return o.refunded_total || 0
     },
+    getShippingTotal: o => {
+      return o.shipping_total || 0
+    },
+    getGiftCardTotal: o => {
+      return o.gift_card_total || 0
+    },
+    getDiscountTotal: o => {
+      return o.discount_total || 0
+    },
+    getTaxTotal: o => {
+      return o.tax_total || 0
+    },
+    getSubtotal: o => {
+      return o.subtotal || 0
+    },
   }
+
   const eventBusService = {
     emit: jest.fn(),
     withTransaction: function() {

packages/medusa/src/services/__tests__/swap.js

@@ -1,6 +1,13 @@
 import { IdMap, MockRepository, MockManager } from "medusa-test-utils"
 import SwapService from "../swap"
 
+const eventBusService = {
+  emit: jest.fn(),
+  withTransaction: function() {
+    return this
+  },
+}
+
 const generateOrder = (orderId, items, additional = {}) => {
   return {
     id: IdMap.getId(orderId),
@@ -70,7 +77,9 @@ describe("SwapService", () => {
     })
 
     it("fails if item is returned", async () => {
-      const swapService = new SwapService({})
+      const swapService = new SwapService({
+        eventBusService,
+      })
       const res = () =>
         swapService.validateReturnItems_(
           {
@@ -168,6 +177,7 @@ describe("SwapService", () => {
 
       const swapService = new SwapService({
         manager: MockManager,
+        eventBusService,
         swapRepository: swapRepo,
         cartService,
         lineItemService,
@@ -236,6 +246,7 @@ describe("SwapService", () => {
         })
         const swapService = new SwapService({
           manager: MockManager,
+          eventBusService,
           swapRepository: swapRepo,
         })
         const res = swapService.createCart(IdMap.getId("swap-1"))
@@ -274,6 +285,7 @@ describe("SwapService", () => {
 
       const swapService = new SwapService({
         manager: MockManager,
+        eventBusService,
         swapRepository: swapRepo,
         returnService,
         lineItemService,
@@ -358,6 +370,7 @@ describe("SwapService", () => {
       })
       const swapService = new SwapService({
         manager: MockManager,
+        eventBusService,
         swapRepository: swapRepo,
         returnService,
       })
@@ -401,6 +414,7 @@ describe("SwapService", () => {
       })
       const swapService = new SwapService({
         manager: MockManager,
+        eventBusService,
         swapRepository: swapRepo,
         returnService,
       })
@@ -475,6 +489,7 @@ describe("SwapService", () => {
       })
       const swapService = new SwapService({
         manager: MockManager,
+        eventBusService,
         swapRepository: swapRepo,
         fulfillmentService,
         lineItemService,
@@ -595,6 +610,7 @@ describe("SwapService", () => {
 
       const swapService = new SwapService({
         manager: MockManager,
+        eventBusService,
         swapRepository: swapRepo,
         lineItemService,
         eventBusService,
@@ -691,6 +707,7 @@ describe("SwapService", () => {
 
       const swapService = new SwapService({
         manager: MockManager,
+        eventBusService,
         swapRepository: swapRepo,
         totalsService,
         paymentProviderService,
@@ -770,6 +787,7 @@ describe("SwapService", () => {
 
       const swapService = new SwapService({
         manager: MockManager,
+        eventBusService,
         swapRepository: swapRepo,
         paymentProviderService,
         eventBusService,

packages/medusa/src/services/claim.js

@@ -99,7 +99,7 @@ class ClaimService extends BaseService {
       const { claim_items, shipping_methods, metadata } = data
 
       if (metadata) {
-        claim.metadata = this.setMetadata_(claim, update.metadata)
+        claim.metadata = this.setMetadata_(claim, metadata)
         await claimRepo.save(claim)
       }
 

packages/medusa/src/services/customer.js

@@ -110,6 +110,7 @@ class CustomerService extends BaseService {
     const token = jwt.sign(payload, secret)
     // Notify subscribers
     this.eventBus_.emit(CustomerService.Events.PASSWORD_RESET, {
+      id: customerId,
       email: customer.email,
       first_name: customer.first_name,
       last_name: customer.last_name,
@@ -292,6 +293,7 @@ class CustomerService extends BaseService {
 
       const {
         email,
+        password,
         password_hash,
         billing_address,
         metadata,
@@ -314,6 +316,10 @@ class CustomerService extends BaseService {
         customer[key] = value
       }
 
+      if (password) {
+        customer.password_hash = await this.hashPassword_(password)
+      }
+
       const updated = await customerRepository.save(customer)
       await this.eventBus_
         .withTransaction(manager)

packages/medusa/src/services/event-bus.js

@@ -217,14 +217,18 @@ class EventBusService {
    */
   worker_ = job => {
     const { eventName, data } = job.data
-    const observers = this.observers_[eventName] || []
+    const eventObservers = this.observers_[eventName] || []
+    const wildcardObservers = this.observers_["*"] || []
+
+    const observers = eventObservers.concat(wildcardObservers)
+
     this.logger_.info(
-      `Processing ${eventName} which has ${observers.length} subscribers`
+      `Processing ${eventName} which has ${eventObservers.length} subscribers`
     )
 
     return Promise.all(
       observers.map(subscriber => {
-        return subscriber(data).catch(err => {
+        return subscriber(data, eventName).catch(err => {
           this.logger_.warn(
             `An error occured while processing ${eventName}: ${err}`
           )
@@ -242,7 +246,7 @@ class EventBusService {
 
     return Promise.all(
       observers.map(subscriber => {
-        return subscriber(data).catch(err => {
+        return subscriber(data, eventName).catch(err => {
           this.logger_.warn(
             `An error occured while processing ${eventName}: ${err}`
           )

packages/medusa/src/services/fulfillment-provider.js

@@ -89,6 +89,18 @@ class FulfillmentProviderService {
     const provider = this.retrieveProvider(option.provider_id)
     return provider.createReturn(returnOrder)
   }
+
+  /**
+   * Fetches documents from the fulfillment provider
+   * @param {string} providerId - the id of the provider
+   * @param {object} fulfillmentData - the data relating to the fulfillment
+   * @param {"invoice" | "label"} documentType - the typ of
+   *  document to fetch
+   */
+  async retrieveDocuments(providerId, fulfillmentData, documentType) {
+    const provider = this.retrieveProvider(providerId)
+    return provider.retrieveDocuments(fulfillmentData, documentType)
+  }
 }
 
 export default FulfillmentProviderService

packages/medusa/src/services/notification.js

@@ -0,0 +1,254 @@
+import { MedusaError } from "medusa-core-utils"
+import { BaseService } from "medusa-interfaces"
+import _ from "lodash"
+
+/**
+ * Provides layer to manipulate orchestrate notifications.
+ * @implements BaseService
+ */
+class NotificationService extends BaseService {
+  constructor(container) {
+    super()
+
+    const {
+      manager,
+      notificationProviderRepository,
+      notificationRepository,
+      logger,
+    } = container
+
+    this.container_ = container
+
+    /** @private @const {EntityManager} */
+    this.manager_ = manager
+    this.logger_ = logger
+
+    /** @private @const {NotificationRepository} */
+    this.notificationRepository_ = notificationRepository
+    this.notificationProviderRepository_ = notificationProviderRepository
+
+    this.subscribers_ = {}
+    this.attachmentGenerator_ = null
+  }
+
+  /**
+   * Registers an attachment generator to the service. The generator can be
+   * used to generate on demand invoices or other documents.
+   */
+  registerAttachmentGenerator(service) {
+    this.attachmentGenerator_ = service
+  }
+
+  /**
+   * Sets the service's manager to a given transaction manager.
+   * @parma {EntityManager} transactionManager - the manager to use
+   * return {NotificationService} a cloned notification service
+   */
+  withTransaction(transactionManager) {
+    if (!transactionManager) {
+      return this
+    }
+
+    const cloned = new LineItemService({
+      manager: transactionManager,
+      notificationRepository: this.notificationRepository_,
+    })
+
+    cloned.transactionManager_ = transactionManager
+
+    return cloned
+  }
+
+  /**
+   * Takes a list of notification provider ids and persists them in the database.
+   * @param {Array<string>} providers - a list of provider ids
+   */
+  async registerInstalledProviders(providers) {
+    const { manager, notificationProviderRepository } = this.container_
+    const model = manager.getCustomRepository(notificationProviderRepository)
+    model.update({}, { is_installed: false })
+    for (const p of providers) {
+      const n = model.create({ id: p, is_installed: true })
+      await model.save(n)
+    }
+  }
+
+  /**
+   * Retrieves a list of notifications.
+   * @param {object} selector - the params to select the notifications by.
+   * @param {object} config - the configuration to apply to the query
+   * @return {Array<Notification>} the notifications that satisfy the query.
+   */
+  async list(
+    selector,
+    config = { skip: 0, take: 50, order: { created_at: "DESC" } }
+  ) {
+    const notiRepo = this.manager_.getCustomRepository(
+      this.notificationRepository_
+    )
+    const query = this.buildQuery_(selector, config)
+    return notiRepo.find(query)
+  }
+
+  /**
+   * Retrieves a notification with a given id
+   * @param {string} id - the id of the notification
+   * @return {Notification} the notification
+   */
+  async retrieve(id, config = {}) {
+    const notiRepository = this.manager_.getCustomRepository(
+      this.notificationRepository_
+    )
+
+    const validatedId = this.validateId_(id)
+    const query = this.buildQuery_({ id: validatedId }, config)
+
+    const notification = await notiRepository.findOne(query)
+
+    if (!notification) {
+      throw new MedusaError(
+        MedusaError.Types.NOT_FOUND,
+        `Notification with id: ${id} was not found.`
+      )
+    }
+
+    return notification
+  }
+
+  /**
+   * Subscribes a given provider to an event.
+   * @param {string} eventName - the event to subscribe to
+   * @param {string} providerId - the provider that the event will be sent to
+   */
+  subscribe(eventName, providerId) {
+    if (typeof providerId !== "string") {
+      throw new MedusaError(
+        MedusaError.Types.NOT_ALLOWED,
+        "providerId must be a string"
+      )
+    }
+
+    if (this.subscribers_[eventName]) {
+      this.subscribers_[eventName].push(providerId)
+    } else {
+      this.subscribers_[eventName] = [providerId]
+    }
+  }
+
+  /**
+   * Finds a provider with a given id. Will throw a NOT_FOUND error if the
+   * resolution fails.
+   * @param {string} id - the id of the provider
+   * @return {NotificationProvider} the notification provider
+   */
+  retrieveProvider_(id) {
+    try {
+      return this.container_[`noti_${id}`]
+    } catch (err) {
+      throw new MedusaError(
+        MedusaError.Types.NOT_FOUND,
+        `Could not find a notification provider with id: ${id}.`
+      )
+    }
+  }
+
+  /**
+   * Handles an event by relaying the event data to the subscribing providers.
+   * The result of the notification send will be persisted in the database in
+   * order to allow for resends. Will log any errors that are encountered.
+   * @param {string} eventName - the event to handle
+   * @param {object} data - the data the event was sent with
+   */
+  handleEvent(eventName, data) {
+    const subs = this.subscribers_[eventName]
+    if (!subs) {
+      return
+    }
+
+    return Promise.all(
+      subs.map(async providerId => {
+        return this.send(eventName, data, providerId).catch(err => {
+          console.log(err)
+          this.logger_.warn(
+            `An error occured while ${providerId} was processing a notification for ${eventName}: ${err.message}`
+          )
+        })
+      })
+    )
+  }
+
+  /**
+   * Sends a notification, by calling the given provider's sendNotification
+   * method. Persists the Notification in the database.
+   * @param {string} event - the name of the event
+   * @param {object} eventData - the data the event was sent with
+   * @param {string} providerId - the provider that should hande the event.
+   * @return {Notification} the created notification
+   */
+  async send(event, eventData, providerId) {
+    const provider = this.retrieveProvider_(providerId)
+    const result = await provider.sendNotification(
+      event,
+      eventData,
+      this.attachmentGenerator_
+    )
+
+    if (!result) {
+      return
+    }
+
+    const { to, data } = result
+    const notiRepo = this.manager_.getCustomRepository(
+      this.notificationRepository_
+    )
+
+    const [resource_type] = event.split(".")
+    const resource_id = eventData.id
+    const customer_id = eventData.customer_id || null
+
+    const created = notiRepo.create({
+      resource_type,
+      resource_id,
+      customer_id,
+      to,
+      data,
+      event_name: event,
+      provider_id: providerId,
+    })
+
+    return notiRepo.save(created)
+  }
+
+  /**
+   * Resends a notification by retrieving a prior notification and calling the
+   * underlying provider's resendNotification method.
+   * @param {string} id - the id of the notification
+   * @param {object} config - any configuration that might override the previous
+   *  send
+   * @return {Notification} the newly created notification
+   */
+  async resend(id, config = {}) {
+    const notification = await this.retrieve(id)
+
+    const provider = this.retrieveProvider_(notification.provider_id)
+    const { to, data } = await provider.resendNotification(
+      notification,
+      config,
+      this.attachmentGenerator_
+    )
+
+    const notiRepo = this.manager_.getCustomRepository(
+      this.notificationRepository_
+    )
+    const created = notiRepo.create({
+      ...notification,
+      to,
+      data,
+      parent_id: id,
+    })
+
+    return notiRepo.save(created)
+  }
+}
+
+export default NotificationService

packages/medusa/src/services/order.js

@@ -959,7 +959,16 @@ class OrderService extends BaseService {
   async createFulfillment(orderId, itemsToFulfill, metadata = {}) {
     return this.atomicPhase_(async manager => {
       const order = await this.retrieve(orderId, {
+        select: [
+          "subtotal",
+          "shipping_total",
+          "discount_total",
+          "tax_total",
+          "gift_card_total",
+          "total",
+        ],
         relations: [
+          "discounts",
           "region",
           "fulfillments",
           "shipping_address",

packages/medusa/src/services/swap.js

@@ -8,6 +8,7 @@ import { MedusaError } from "medusa-core-utils"
  */
 class SwapService extends BaseService {
   static Events = {
+    CREATED: "swap.created",
     SHIPMENT_CREATED: "swap.shipment_created",
     PAYMENT_COMPLETED: "swap.payment_completed",
     PAYMENT_CAPTURED: "swap.payment_captured",
@@ -247,6 +248,12 @@ class SwapService extends BaseService {
         order
       )
 
+      await this.eventBus_
+        .withTransaction(manager)
+        .emit(SwapService.Events.CREATED, {
+          id: result.id,
+        })
+
       return result
     })
   }

packages/medusa/src/services/totals.js

@@ -94,7 +94,10 @@ class TotalsService extends BaseService {
 
   getLineItemRefund(object, lineItem) {
     const { discounts } = object
-    const tax_rate = object.tax_rate || object.region.tax_rate
+    const tax_rate =
+      typeof object.tax_rate !== "undefined"
+        ? object.tax_rate
+        : object.region.tax_rate
     const taxRate = (tax_rate || 0) / 100
 
     const discount = discounts.find(({ rule }) => rule.type !== "free_shipping")