Adds publish/subscribe pattern to allow plugins and projects to hook into events (#55)

Also adds CLI to ease development.

packages/medusa/src/commands/develop.js

@@ -0,0 +1,45 @@
+import { spawn, execSync } from "child_process"
+import mongoose from "mongoose"
+import chokidar from "chokidar"
+import express from "express"
+import path from "path"
+
+import loaders from "../loaders"
+import Logger from "../loaders/logger"
+
+export default async function({ port, directory }) {
+  const args = process.argv
+  args.shift()
+  args.shift()
+  args.shift()
+
+  execSync("babel src -d dist", {
+    cwd: directory,
+    stdio: ["ignore", process.stdout, process.stderr],
+  })
+
+  let child = spawn("medusa", [`start`, ...args], {
+    cwd: directory,
+    env: process.env,
+    stdio: ["pipe", process.stdout, process.stderr],
+  })
+
+  chokidar.watch(`${directory}/src`).on("change", file => {
+    const f = file.split("src")[1]
+    Logger.info(`${f} changed: restarting...`)
+    child.kill("SIGINT")
+
+    execSync(`babel src -d dist`, {
+      cwd: directory,
+      stdio: ["pipe", process.stdout, process.stderr],
+    })
+
+    Logger.info("Rebuilt")
+
+    child = spawn("medusa", [`start`, ...args], {
+      cwd: directory,
+      env: process.env,
+      stdio: ["pipe", process.stdout, process.stderr],
+    })
+  })
+}

packages/medusa/src/commands/start.js

@@ -0,0 +1,28 @@
+import "core-js/stable"
+import "regenerator-runtime/runtime"
+
+import mongoose from "mongoose"
+import chokidar from "chokidar"
+import express from "express"
+import cwdResolve from "resolve-cwd"
+
+import loaders from "../loaders"
+import Logger from "../loaders/logger"
+
+export default async function({ port, directory }) {
+  async function start() {
+    const app = express()
+
+    const { dbConnection } = await loaders({ directory, expressApp: app })
+    const server = app.listen(port, err => {
+      if (err) {
+        return
+      }
+      Logger.info(`Server is ready on port: ${port}!`)
+    })
+
+    return { dbConnection, server }
+  }
+
+  let { dbConnection, server } = await start()
+}

packages/medusa/src/commands/user.js

@@ -0,0 +1,23 @@
+import "core-js/stable"
+import "regenerator-runtime/runtime"
+
+import mongoose from "mongoose"
+import chokidar from "chokidar"
+import express from "express"
+import cwdResolve from "resolve-cwd"
+
+import loaders from "../loaders"
+import Logger from "../loaders/logger"
+
+export default async function({ directory, email, password }) {
+  const app = express()
+  const { container, dbConnection } = await loaders({
+    directory,
+    expressApp: app,
+  })
+
+  const userService = container.resolve("userService")
+  const user = await userService.create({ email }, password)
+
+  process.exit()
+}

packages/medusa/src/loaders/index.js

@@ -8,7 +8,7 @@ import passportLoader from "./passport"
 import pluginsLoader from "./plugins"
 import Logger from "./logger"
 
-export default async ({ expressApp }) => {
+export default async ({ directory: rootDirectory, expressApp }) => {
   const container = createContainer()
   container.registerAdd = function(name, registration) {
     let storeKey = name + "_STORE"
@@ -36,7 +36,7 @@ export default async ({ expressApp }) => {
   await servicesLoader({ container })
   Logger.info("Services initialized")
 
-  await mongooseLoader({ container })
+  const dbConnection = await mongooseLoader({ container })
   Logger.info("MongoDB Intialized")
 
   await expressLoader({ app: expressApp })
@@ -51,11 +51,13 @@ export default async ({ expressApp }) => {
     next()
   })
 
-  await pluginsLoader({ container, app: expressApp })
+  await pluginsLoader({ container, rootDirectory, app: expressApp })
   Logger.info("Plugins Intialized")
 
   await apiLoader({ container, app: expressApp })
   Logger.info("API initialized")
+
+  return { container, dbConnection, app: expressApp }
 }
 
 function asArray(resolvers) {

packages/medusa/src/loaders/mongoose.js

@@ -3,17 +3,19 @@ import config from "../config"
 
 export default async ({ container }) => {
   const logger = container.resolve("logger")
-  await mongoose
-    .connect(config.databaseURL, {
-      useNewUrlParser: true,
-      useCreateIndex: true,
-      useUnifiedTopology: true,
-    })
-    .catch(err => {
-      logger.error(err)
-    })
 
   mongoose.connection.on("error", err => {
     logger.error(err)
   })
+
+  return mongoose
+    .connect(config.databaseURL, {
+      useNewUrlParser: true,
+      useCreateIndex: true,
+      useUnifiedTopology: true,
+      useFindAndModify: false,
+    })
+    .catch(err => {
+      logger.error(err)
+    })
 }

packages/medusa/src/loaders/plugins.js

@@ -15,9 +15,9 @@ import { sync as existsSync } from "fs-exists-cached"
 /**
  * Registers all services in the services directory
  */
-export default ({ container, app }) => {
+export default ({ rootDirectory, container, app }) => {
   const { configModule, configFilePath } = getConfigFile(
-    process.cwd(),
+    rootDirectory,
     `medusa-config`
   )
 
@@ -39,7 +39,7 @@ export default ({ container, app }) => {
   })
 
   resolved.push({
-    resolve: process.cwd(),
+    resolve: `${rootDirectory}/dist`,
     name: `project-plugin`,
     id: createPluginId(`project-plugin`),
     options: {},
@@ -51,6 +51,7 @@ export default ({ container, app }) => {
     registerServices(pluginDetails, container)
     registerMedusaApi(pluginDetails, container)
     registerApi(pluginDetails, app)
+    registerSubscribers(pluginDetails, container)
   })
 }
 
@@ -156,6 +157,31 @@ function registerServices(pluginDetails, container) {
   })
 }
 
+/**
+ * Registers a plugin's models at the right location in our container. Models
+ * must inherit from BaseModel. Models are registered directly in the container.
+ * Names are camelCase formatted and namespaced by the folder i.e:
+ * models/example-person -> examplePersonModel
+ * @param {object} pluginDetails - the plugin details including plugin options,
+ *    version, id, resolved path, etc. See resolvePlugin
+ * @param {object} container - the container where the services will be
+ *    registered
+ * @return {void}
+ */
+function registerSubscribers(pluginDetails, container) {
+  const files = glob.sync(`${pluginDetails.resolve}/subscribers/*.js`, {})
+  files.forEach(fn => {
+    const loaded = require(fn).default
+
+    const name = formatRegistrationName(fn)
+    container.build(
+      asFunction(
+        cradle => new loaded(cradle, pluginDetails.options)
+      ).singleton()
+    )
+  })
+}
+
 /**
  * Registers a plugin's models at the right location in our container. Models
  * must inherit from BaseModel. Models are registered directly in the container.

packages/medusa/src/models/__mocks__/cart.js

@@ -312,9 +312,9 @@ export const carts = {
 }
 
 export const CartModelMock = {
-  create: jest.fn().mockReturnValue(Promise.resolve()),
+  create: jest.fn().mockImplementation(data => Promise.resolve(data)),
   updateOne: jest.fn().mockImplementation((query, update) => {
-    return Promise.resolve()
+    return Promise.resolve(update)
   }),
   deleteOne: jest.fn().mockReturnValue(Promise.resolve()),
   findOne: jest.fn().mockImplementation(query => {

packages/medusa/src/models/cart.js

@@ -15,15 +15,15 @@ class CartModel extends BaseModel {
 
   static schema = {
     email: { type: String },
-    billing_address: { type: AddressSchema, default: {} },
-    shipping_address: { type: AddressSchema, default: {} },
+    billing_address: { type: AddressSchema },
+    shipping_address: { type: AddressSchema },
     items: { type: [LineItemSchema], default: [] },
     region_id: { type: String, required: true },
     discounts: { type: [DiscountModel.schema], default: [] },
     customer_id: { type: String, default: "" },
     payment_sessions: { type: [PaymentMethodSchema], default: [] },
     shipping_options: { type: [ShippingMethodSchema], default: [] },
-    payment_method: { type: PaymentMethodSchema, default: {} },
+    payment_method: { type: PaymentMethodSchema },
     shipping_methods: { type: [ShippingMethodSchema], default: [] },
     metadata: { type: mongoose.Schema.Types.Mixed, default: {} },
   }

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

@@ -0,0 +1,10 @@
+export const EventBusServiceMock = {
+  emit: jest.fn(),
+  subscribe: jest.fn(),
+}
+
+const mock = jest.fn().mockImplementation(() => {
+  return EventBusServiceMock
+})
+
+export default mock

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

@@ -7,6 +7,7 @@ import {
 } from "../__mocks__/payment-provider"
 import { ProductVariantServiceMock } from "../__mocks__/product-variant"
 import { RegionServiceMock } from "../__mocks__/region"
+import { EventBusServiceMock } from "../__mocks__/event-bus"
 import { ShippingOptionServiceMock } from "../__mocks__/shipping-option"
 import { ShippingProfileServiceMock } from "../__mocks__/shipping-profile"
 import { CartModelMock, carts } from "../../models/__mocks__/cart"
@@ -40,6 +41,7 @@ describe("CartService", () => {
   describe("setMetadata", () => {
     const cartService = new CartService({
       cartModel: CartModelMock,
+      eventBusService: EventBusServiceMock,
     })
 
     beforeEach(() => {
@@ -50,6 +52,12 @@ describe("CartService", () => {
       const id = mongoose.Types.ObjectId()
       await cartService.setMetadata(`${id}`, "metadata", "testMetadata")
 
+      expect(EventBusServiceMock.emit).toHaveBeenCalledTimes(1)
+      expect(EventBusServiceMock.emit).toHaveBeenCalledWith(
+        "cart.updated",
+        expect.any(Object)
+      )
+
       expect(CartModelMock.updateOne).toBeCalledTimes(1)
       expect(CartModelMock.updateOne).toBeCalledWith(
         { _id: `${id}` },
@@ -74,6 +82,7 @@ describe("CartService", () => {
     const cartService = new CartService({
       cartModel: CartModelMock,
       regionService: RegionServiceMock,
+      eventBusService: EventBusServiceMock,
     })
 
     beforeEach(() => {
@@ -85,6 +94,12 @@ describe("CartService", () => {
         region_id: IdMap.getId("testRegion"),
       })
 
+      expect(EventBusServiceMock.emit).toHaveBeenCalledTimes(1)
+      expect(EventBusServiceMock.emit).toHaveBeenCalledWith(
+        "cart.created",
+        expect.any(Object)
+      )
+
       expect(CartModelMock.create).toHaveBeenCalledTimes(1)
       expect(CartModelMock.create).toHaveBeenCalledWith({
         region_id: IdMap.getId("testRegion"),
@@ -97,6 +112,7 @@ describe("CartService", () => {
       cartModel: CartModelMock,
       productVariantService: ProductVariantServiceMock,
       lineItemService: LineItemServiceMock,
+      eventBusService: EventBusServiceMock,
     })
 
     beforeEach(() => {
@@ -123,6 +139,12 @@ describe("CartService", () => {
 
       await cartService.addLineItem(IdMap.getId("emptyCart"), lineItem)
 
+      expect(EventBusServiceMock.emit).toHaveBeenCalledTimes(1)
+      expect(EventBusServiceMock.emit).toHaveBeenCalledWith(
+        "cart.updated",
+        expect.any(Object)
+      )
+
       expect(CartModelMock.updateOne).toHaveBeenCalledTimes(1)
       expect(CartModelMock.updateOne).toHaveBeenCalledWith(
         {
@@ -280,6 +302,7 @@ describe("CartService", () => {
       cartModel: CartModelMock,
       productVariantService: ProductVariantServiceMock,
       lineItemService: LineItemServiceMock,
+      eventBusService: EventBusServiceMock,
     })
 
     beforeEach(() => {
@@ -310,6 +333,12 @@ describe("CartService", () => {
         lineItem
       )
 
+      expect(EventBusServiceMock.emit).toHaveBeenCalledTimes(1)
+      expect(EventBusServiceMock.emit).toHaveBeenCalledWith(
+        "cart.updated",
+        expect.any(Object)
+      )
+
       expect(CartModelMock.updateOne).toHaveBeenCalledTimes(1)
       expect(CartModelMock.updateOne).toHaveBeenCalledWith(
         {
@@ -357,6 +386,7 @@ describe("CartService", () => {
   describe("updateEmail", () => {
     const cartService = new CartService({
       cartModel: CartModelMock,
+      eventBusService: EventBusServiceMock,
     })
 
     beforeEach(() => {
@@ -369,6 +399,12 @@ describe("CartService", () => {
         "test@testdom.com"
       )
 
+      expect(EventBusServiceMock.emit).toHaveBeenCalledTimes(1)
+      expect(EventBusServiceMock.emit).toHaveBeenCalledWith(
+        "cart.updated",
+        expect.any(Object)
+      )
+
       expect(CartModelMock.updateOne).toHaveBeenCalledTimes(1)
       expect(CartModelMock.updateOne).toHaveBeenCalledWith(
         {
@@ -394,6 +430,7 @@ describe("CartService", () => {
   describe("updateBillingAddress", () => {
     const cartService = new CartService({
       cartModel: CartModelMock,
+      eventBusService: EventBusServiceMock,
     })
 
     beforeEach(() => {
@@ -413,6 +450,12 @@ describe("CartService", () => {
 
       await cartService.updateBillingAddress(IdMap.getId("emptyCart"), address)
 
+      expect(EventBusServiceMock.emit).toHaveBeenCalledTimes(1)
+      expect(EventBusServiceMock.emit).toHaveBeenCalledWith(
+        "cart.updated",
+        expect.any(Object)
+      )
+
       expect(CartModelMock.updateOne).toHaveBeenCalledTimes(1)
       expect(CartModelMock.updateOne).toHaveBeenCalledWith(
         {
@@ -450,6 +493,7 @@ describe("CartService", () => {
   describe("updateShippingAddress", () => {
     const cartService = new CartService({
       cartModel: CartModelMock,
+      eventBusService: EventBusServiceMock,
     })
 
     beforeEach(() => {
@@ -469,6 +513,12 @@ describe("CartService", () => {
 
       await cartService.updateShippingAddress(IdMap.getId("emptyCart"), address)
 
+      expect(EventBusServiceMock.emit).toHaveBeenCalledTimes(1)
+      expect(EventBusServiceMock.emit).toHaveBeenCalledWith(
+        "cart.updated",
+        expect.any(Object)
+      )
+
       expect(CartModelMock.updateOne).toHaveBeenCalledTimes(1)
       expect(CartModelMock.updateOne).toHaveBeenCalledWith(
         {
@@ -508,6 +558,7 @@ describe("CartService", () => {
       cartModel: CartModelMock,
       regionService: RegionServiceMock,
       productVariantService: ProductVariantServiceMock,
+      eventBusService: EventBusServiceMock,
     })
 
     beforeEach(() => {
@@ -520,6 +571,12 @@ describe("CartService", () => {
         IdMap.getId("region-us")
       )
 
+      expect(EventBusServiceMock.emit).toHaveBeenCalledTimes(1)
+      expect(EventBusServiceMock.emit).toHaveBeenCalledWith(
+        "cart.updated",
+        expect.any(Object)
+      )
+
       expect(CartModelMock.updateOne).toHaveBeenCalledTimes(1)
       expect(CartModelMock.updateOne).toHaveBeenCalledWith(
         {
@@ -645,6 +702,7 @@ describe("CartService", () => {
       cartModel: CartModelMock,
       regionService: RegionServiceMock,
       paymentProviderService: PaymentProviderServiceMock,
+      eventBusService: EventBusServiceMock,
     })
 
     beforeEach(() => {
@@ -664,6 +722,12 @@ describe("CartService", () => {
         paymentMethod
       )
 
+      expect(EventBusServiceMock.emit).toHaveBeenCalledTimes(1)
+      expect(EventBusServiceMock.emit).toHaveBeenCalledWith(
+        "cart.updated",
+        expect.any(Object)
+      )
+
       expect(RegionServiceMock.retrieve).toHaveBeenCalledTimes(1)
       expect(RegionServiceMock.retrieve).toHaveBeenCalledWith(
         IdMap.getId("testRegion")
@@ -787,6 +851,7 @@ describe("CartService", () => {
       cartModel: CartModelMock,
       regionService: RegionServiceMock,
       paymentProviderService: PaymentProviderServiceMock,
+      eventBusService: EventBusServiceMock,
     })
 
     beforeEach(() => {
@@ -796,6 +861,12 @@ describe("CartService", () => {
     it("initializes payment sessions for each of the providers", async () => {
       await cartService.setPaymentSessions(IdMap.getId("cartWithLine"))
 
+      expect(EventBusServiceMock.emit).toHaveBeenCalledTimes(1)
+      expect(EventBusServiceMock.emit).toHaveBeenCalledWith(
+        "cart.updated",
+        expect.any(Object)
+      )
+
       expect(PaymentProviderServiceMock.createSession).toHaveBeenCalledTimes(2)
       expect(PaymentProviderServiceMock.createSession).toHaveBeenCalledWith(
         "default_provider",
@@ -938,6 +1009,7 @@ describe("CartService", () => {
   describe("retrievePaymentSession", () => {
     const cartService = new CartService({
       cartModel: CartModelMock,
+      eventBusService: EventBusServiceMock,
     })
 
     let res
@@ -999,6 +1071,7 @@ describe("CartService", () => {
       cartModel: CartModelMock,
       shippingProfileService: ShippingProfileServiceMock,
       shippingOptionService: ShippingOptionServiceMock,
+      eventBusService: EventBusServiceMock,
     })
 
     describe("successfully adds the shipping method", () => {
@@ -1043,6 +1116,12 @@ describe("CartService", () => {
       })
 
       it("updates cart", () => {
+        expect(EventBusServiceMock.emit).toHaveBeenCalledTimes(1)
+        expect(EventBusServiceMock.emit).toHaveBeenCalledWith(
+          "cart.updated",
+          expect.any(Object)
+        )
+
         expect(CartModelMock.updateOne).toHaveBeenCalledTimes(1)
         expect(CartModelMock.updateOne).toHaveBeenCalledWith(
           {
@@ -1171,6 +1250,7 @@ describe("CartService", () => {
     const cartService = new CartService({
       cartModel: CartModelMock,
       discountService: DiscountServiceMock,
+      eventBusService: EventBusServiceMock,
     })
     beforeEach(async () => {
       jest.clearAllMocks()
@@ -1183,6 +1263,12 @@ describe("CartService", () => {
         _id: IdMap.getId("fr-cart"),
       })
 
+      expect(EventBusServiceMock.emit).toHaveBeenCalledTimes(1)
+      expect(EventBusServiceMock.emit).toHaveBeenCalledWith(
+        "cart.updated",
+        expect.any(Object)
+      )
+
       expect(DiscountServiceMock.retrieveByCode).toHaveBeenCalledTimes(1)
       expect(DiscountServiceMock.retrieveByCode).toHaveBeenCalledWith("10%OFF")
 

packages/medusa/src/services/cart.js

@@ -7,6 +7,11 @@ import { BaseService } from "medusa-interfaces"
  * @implements BaseService
  */
 class CartService extends BaseService {
+  static Events = {
+    CREATED: "cart.created",
+    UPDATED: "cart.updated",
+  }
+
   constructor({
     cartModel,
     eventBusService,
@@ -199,6 +204,10 @@ class CartService extends BaseService {
       .create({
         region_id: region._id,
       })
+      .then(result => {
+        this.eventBus_.emit(CartService.Events.CREATED, result)
+        return result
+      })
       .catch(err => {
         throw new MedusaError(MedusaError.Types.DB_ERROR, err.message)
       })
@@ -246,17 +255,23 @@ class CartService extends BaseService {
         )
       }
 
-      return this.cartModel_.updateOne(
-        {
-          _id: cartId,
-          "items._id": currentItem._id,
-        },
-        {
-          $set: {
-            "items.$.quantity": newQuantity,
+      return this.cartModel_
+        .updateOne(
+          {
+            _id: cartId,
+            "items._id": currentItem._id,
           },
-        }
-      )
+          {
+            $set: {
+              "items.$.quantity": newQuantity,
+            },
+          }
+        )
+        .then(result => {
+          // Notify subscribers
+          this.eventBus_.emit(CartService.Events.UPDATED, result)
+          return result
+        })
     }
 
     // Confirm inventory
@@ -273,14 +288,20 @@ class CartService extends BaseService {
     }
 
     // The line we are adding doesn't already exist so it is safe to push
-    return this.cartModel_.updateOne(
-      {
-        _id: cartId,
-      },
-      {
-        $push: { items: validatedLineItem },
-      }
-    )
+    return this.cartModel_
+      .updateOne(
+        {
+          _id: cartId,
+        },
+        {
+          $push: { items: validatedLineItem },
+        }
+      )
+      .then(result => {
+        // Notify subscribers
+        this.eventBus_.emit(CartService.Events.UPDATED, result)
+        return result
+      })
   }
 
   /**
@@ -325,17 +346,23 @@ class CartService extends BaseService {
     }
 
     // Update the line item
-    return this.cartModel_.updateOne(
-      {
-        _id: cartId,
-        "items._id": lineItemId,
-      },
-      {
-        $set: {
-          "items.$": validatedLineItem,
+    return this.cartModel_
+      .updateOne(
+        {
+          _id: cartId,
+          "items._id": lineItemId,
         },
-      }
-    )
+        {
+          $set: {
+            "items.$": validatedLineItem,
+          },
+        }
+      )
+      .then(result => {
+        // Notify subscribers
+        this.eventBus_.emit(CartService.Events.UPDATED, result)
+        return result
+      })
   }
 
   /**
@@ -357,14 +384,20 @@ class CartService extends BaseService {
       )
     }
 
-    return this.cartModel_.updateOne(
-      {
-        _id: cart._id,
-      },
-      {
-        $set: { email: value },
-      }
-    )
+    return this.cartModel_
+      .updateOne(
+        {
+          _id: cart._id,
+        },
+        {
+          $set: { email: value },
+        }
+      )
+      .then(result => {
+        // Notify subscribers
+        this.eventBus_.emit(CartService.Events.UPDATED, result)
+        return result
+      })
   }
 
   /**
@@ -383,14 +416,20 @@ class CartService extends BaseService {
       )
     }
 
-    return this.cartModel_.updateOne(
-      {
-        _id: cart._id,
-      },
-      {
-        $set: { billing_address: value },
-      }
-    )
+    return this.cartModel_
+      .updateOne(
+        {
+          _id: cart._id,
+        },
+        {
+          $set: { billing_address: value },
+        }
+      )
+      .then(result => {
+        // Notify subscribers
+        this.eventBus_.emit(CartService.Events.UPDATED, result)
+        return result
+      })
   }
 
   /**
@@ -409,14 +448,20 @@ class CartService extends BaseService {
       )
     }
 
-    return this.cartModel_.updateOne(
-      {
-        _id: cartId,
-      },
-      {
-        $set: { shipping_address: value },
-      }
-    )
+    return this.cartModel_
+      .updateOne(
+        {
+          _id: cartId,
+        },
+        {
+          $set: { shipping_address: value },
+        }
+      )
+      .then(result => {
+        // Notify subscribers
+        this.eventBus_.emit(CartService.Events.UPDATED, result)
+        return result
+      })
   }
   /**
    * Updates the cart's discounts.
@@ -455,49 +500,73 @@ class CartService extends BaseService {
       shippingDisc.length === 0 &&
       discount.discount_rule.type === "free_shipping"
     ) {
-      return this.cartModel_.updateOne(
-        {
-          _id: cart._id,
-        },
-        {
-          $push: { discounts: discount },
-        }
-      )
+      return this.cartModel_
+        .updateOne(
+          {
+            _id: cart._id,
+          },
+          {
+            $push: { discounts: discount },
+          }
+        )
+        .then(result => {
+          // Notify subscribers
+          this.eventBus_.emit(CartService.Events.UPDATED, result)
+          return result
+        })
     } else if (
       shippingDisc.length > 0 &&
       discount.discount_rule.type === "free_shipping"
     ) {
-      return this.cartModel_.updateOne(
-        {
-          _id: cart._id,
-        },
-        {
-          $pull: { discounts: { _id: shippingDisc[0]._id } },
-          $push: { discounts: discount },
-        }
-      )
+      return this.cartModel_
+        .updateOne(
+          {
+            _id: cart._id,
+          },
+          {
+            $pull: { discounts: { _id: shippingDisc[0]._id } },
+            $push: { discounts: discount },
+          }
+        )
+        .then(result => {
+          // Notify subscribers
+          this.eventBus_.emit(CartService.Events.UPDATED, result)
+          return result
+        })
     }
 
     // replace the current discount if there, else add the new one
     if (otherDisc.length === 0) {
-      return this.cartModel_.updateOne(
-        {
-          _id: cart._id,
-        },
-        {
-          $push: { discounts: discount },
-        }
-      )
+      return this.cartModel_
+        .updateOne(
+          {
+            _id: cart._id,
+          },
+          {
+            $push: { discounts: discount },
+          }
+        )
+        .then(result => {
+          // Notify subscribers
+          this.eventBus_.emit(CartService.Events.UPDATED, result)
+          return result
+        })
     } else {
-      return this.cartModel_.updateOne(
-        {
-          _id: cart._id,
-        },
-        {
-          $pull: { discounts: { _id: otherDisc[0]._id } },
-          $push: { discounts: discount },
-        }
-      )
+      return this.cartModel_
+        .updateOne(
+          {
+            _id: cart._id,
+          },
+          {
+            $pull: { discounts: { _id: otherDisc[0]._id } },
+            $push: { discounts: discount },
+          }
+        )
+        .then(result => {
+          // Notify subscribers
+          this.eventBus_.emit(CartService.Events.UPDATED, result)
+          return result
+        })
     }
   }
 
@@ -571,14 +640,20 @@ class CartService extends BaseService {
     }
 
     // At this point we can register the payment method.
-    return this.cartModel_.updateOne(
-      {
-        _id: cart._id,
-      },
-      {
-        $set: { payment_method: paymentMethod },
-      }
-    )
+    return this.cartModel_
+      .updateOne(
+        {
+          _id: cart._id,
+        },
+        {
+          $set: { payment_method: paymentMethod },
+        }
+      )
+      .then(result => {
+        // Notify subscribers
+        this.eventBus_.emit(CartService.Events.UPDATED, result)
+        return result
+      })
   }
 
   /**
@@ -627,14 +702,20 @@ class CartService extends BaseService {
 
     // Update the payment sessions with the concatenated array of updated and
     // newly created payment sessions
-    return this.cartModel_.updateOne(
-      {
-        _id: cart._id,
-      },
-      {
-        $set: { payment_sessions: sessions.concat(newSessions) },
-      }
-    )
+    return this.cartModel_
+      .updateOne(
+        {
+          _id: cart._id,
+        },
+        {
+          $set: { payment_sessions: sessions.concat(newSessions) },
+        }
+      )
+      .then(result => {
+        // Notify subscribers
+        this.eventBus_.emit(CartService.Events.UPDATED, result)
+        return result
+      })
   }
 
   /**
@@ -693,14 +774,20 @@ class CartService extends BaseService {
       newMethods.push(option)
     }
 
-    return this.cartModel_.updateOne(
-      {
-        _id: cart._id,
-      },
-      {
-        $set: { shipping_methods: newMethods },
-      }
-    )
+    return this.cartModel_
+      .updateOne(
+        {
+          _id: cart._id,
+        },
+        {
+          $set: { shipping_methods: newMethods },
+        }
+      )
+      .then(result => {
+        // Notify subscribers
+        this.eventBus_.emit(CartService.Events.UPDATED, result)
+        return result
+      })
   }
 
   /**
@@ -763,7 +850,13 @@ class CartService extends BaseService {
       update.payment_method = undefined
     }
 
-    return this.cartModel_.updateOne({ _id: cart._id }, { $set: update })
+    return this.cartModel_
+      .updateOne({ _id: cart._id }, { $set: update })
+      .then(result => {
+        // Notify subscribers
+        this.eventBus_.emit(CartService.Events.UPDATED, result)
+        return result
+      })
   }
 
   /**
@@ -788,6 +881,11 @@ class CartService extends BaseService {
     const keyPath = `metadata.${key}`
     return this.cartModel_
       .updateOne({ _id: validatedId }, { $set: { [keyPath]: value } })
+      .then(result => {
+        // Notify subscribers
+        this.eventBus_.emit(CartService.Events.UPDATED, result)
+        return result
+      })
       .catch(err => {
         throw new MedusaError(MedusaError.Types.DB_ERROR, err.message)
       })

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

@@ -59,7 +59,8 @@ class EventBusService {
    *
    * @returns {Promise} resolves to the results of the subscriber calls.
    */
-  async worker_({ eventName, data }) {
+  worker_ = job => {
+    const { eventName, data } = job.data
     const observers = this.observers_[eventName] || []
     this.logger_.info(
       `Processing ${eventName} which has ${observers.length} subscribers`