creates webhook and cron job to update inventory

packages/medusa/src/loaders/plugins.js

@@ -16,7 +16,7 @@ import { sync as existsSync } from "fs-exists-cached"
 /**
  * Registers all services in the services directory
  */
-export default ({ rootDirectory, container, app }) => {
+export default async ({ rootDirectory, container, app }) => {
   const { configModule, configFilePath } = getConfigFile(
     rootDirectory,
     `medusa-config`
@@ -55,6 +55,29 @@ export default ({ rootDirectory, container, app }) => {
     registerCoreRouters(pluginDetails, container)
     registerSubscribers(pluginDetails, container)
   })
+
+  await Promise.all(
+    resolved.map(async pluginDetails => runLoaders(pluginDetails, container))
+  )
+}
+
+async function runLoaders(pluginDetails, container) {
+  const loaderFiles = glob.sync(
+    `${pluginDetails.resolve}/loaders/[!__]*.js`,
+    {}
+  )
+  await Promise.all(
+    loaderFiles.map(async loader => {
+      try {
+        const module = require(loader).default
+        if (typeof module === "function") {
+          await module(container)
+        }
+      } catch (err) {
+        return Promise.resolve()
+      }
+    })
+  )
 }
 
 function registerMedusaApi(pluginDetails, container) {

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

@@ -13,11 +13,20 @@ class EventBusService {
     /** @private {object} */
     this.observers_ = {}
 
+    /** @private {object} to handle cron jobs */
+    this.cronHandlers_ = {}
+
+    /** @private {BullQueue} used for cron jobs */
+    this.cronQueue_ = new Bull(`cron-jobs:queue`, config.redisURI)
+
     /** @private {BullQueue} */
     this.queue_ = new Bull(`${this.constructor.name}:queue`, config.redisURI)
 
     // Register our worker to handle emit calls
     this.queue_.process(this.worker_)
+
+    // Register cron worker
+    this.cronQueue_.process(this.cronWorker_)
   }
 
   /**
@@ -38,6 +47,21 @@ class EventBusService {
     }
   }
 
+  /**
+   *
+   */
+  registerCronHandler_(event, subscriber) {
+    if (typeof subscriber !== "function") {
+      throw new Error("Handler must be a function")
+    }
+
+    if (this.observers_[event]) {
+      this.cronHandlers_[event].push(subscriber)
+    } else {
+      this.cronHandlers_[event] = [subscriber]
+    }
+  }
+
   /**
    * Calls all subscribers when an event occurs.
    * @param {string} eventName - the name of the event to be process.
@@ -77,6 +101,37 @@ class EventBusService {
       })
     )
   }
+
+  cronWorker_ = job => {
+    const { eventName, data } = job.data
+    const observers = this.cronHandlers_[eventName] || []
+    this.logger_.info(`Processing cron job: ${eventName}`)
+
+    return Promise.all(
+      observers.map(subscriber => {
+        return subscriber(data).catch(err => {
+          this.logger_.warn(
+            `An error occured while processing ${eventName}: ${err}`
+          )
+          return err
+        })
+      })
+    )
+  }
+
+  /**
+   * Registers a cron job.
+   */
+  createCronJob(eventName, data, cron, handler) {
+    this.registerCronHandler(eventName, handler)
+    return this.cronQueue_.add(
+      {
+        eventName,
+        data,
+      },
+      { repeat: { cron } }
+    )
+  }
 }
 
 export default EventBusService