Merge branch 'server-workflows' of github.com:Budibase/budibase into server-workflows

packages/server/Dockerfile

@@ -4,6 +4,7 @@ WORKDIR /app
 
 ENV CLOUD=1
 ENV COUCH_DB_URL=https://couchdb.budi.live:5984
+env BUDIBASE_ENVIRONMENT=PRODUCTION
 
 # copy files and install dependencies
 COPY . ./

packages/server/src/api/controllers/record.js

@@ -2,6 +2,16 @@ const CouchDB = require("../../db")
 const validateJs = require("validate.js")
 const newid = require("../../db/newid")
 
+function emitEvent(eventType, ctx, record) {
+  ctx.eventEmitter &&
+    ctx.eventEmitter.emit(eventType, {
+      args: {
+        record,
+      },
+      instanceId: ctx.user.instanceId,
+    })
+}
+
 validateJs.extend(validateJs.validators.datetime, {
   parse: function(value) {
     return new Date(value).getTime()
@@ -100,9 +110,7 @@ exports.save = async function(ctx) {
         const doc = row.doc
         return {
           ...doc,
-          [model.name]: doc[model.name]
-            ? [...doc[model.name], record._id]
-            : [record._id],
+          [model.name]: doc[model.name] ? [...doc[model.name], record._id] : [record._id],
         }
       })
 
@@ -110,13 +118,7 @@ exports.save = async function(ctx) {
     }
   }
 
-  ctx.eventEmitter &&
-    ctx.eventEmitter.emit(`record:save`, {
-      args: {
-        record,
-      },
-      instanceId: ctx.user.instanceId,
-    })
+  emitEvent(`record:save`, ctx, record)
   ctx.body = record
   ctx.status = 200
   ctx.message = `${model.name} created successfully`
@@ -179,7 +181,7 @@ exports.destroy = async function(ctx) {
     return
   }
   ctx.body = await db.remove(ctx.params.recordId, ctx.params.revId)
-  ctx.eventEmitter && ctx.eventEmitter.emit(`record:delete`, record)
+  emitEvent(`record:delete`, ctx, record)
 }
 
 exports.validate = async function(ctx) {
@@ -199,10 +201,7 @@ async function validate({ instanceId, modelId, record, model }) {
   }
   const errors = {}
   for (let fieldName in model.schema) {
-    const res = validateJs.single(
-      record[fieldName],
-      model.schema[fieldName].constraints
-    )
+    const res = validateJs.single(record[fieldName], model.schema[fieldName].constraints)
     if (res) errors[fieldName] = res
   }
   return { valid: Object.keys(errors).length === 0, errors }

packages/server/src/api/controllers/workflow/index.js

@@ -91,4 +91,9 @@ exports.trigger = async function(ctx) {
     ...ctx.request.body,
     instanceId: ctx.user.instanceId,
   })
+  ctx.status = 200
+  ctx.body = {
+    message: `Workflow ${workflow._id} has been triggered.`,
+    workflow,
+  }
 }

packages/server/src/workflows/index.js

@@ -1,6 +1,5 @@
 const triggers = require("./triggers")
 const workerFarm = require("worker-farm")
-const CouchDB = require("../db/client")
 const singleThread = require("./thread")
 
 let workers = workerFarm(require.resolve("./thread"))
@@ -22,7 +21,7 @@ function runWorker(job) {
  */
 module.exports.init = function() {
   triggers.workflowQueue.process(async job => {
-    if (CouchDB.preferredAdapters != null && CouchDB.preferredAdapters[0] !== "leveldb") {
+    if (process.env.BUDIBASE_ENVIRONMENT === "PRODUCTION") {
       await runWorker(job)
     } else {
       await singleThread(job)

packages/server/src/workflows/thread.js

@@ -52,6 +52,7 @@ class Orchestrator {
   }
 }
 
+// callback is required for worker-farm to state that the worker thread has completed
 module.exports = async (job, cb = null) => {
   try {
     const workflowOrchestrator = new Orchestrator(job.data.workflow)

packages/server/src/workflows/triggers.js

@@ -5,6 +5,9 @@ const InMemoryQueue = require("./queue/inMemoryQueue")
 let workflowQueue = new InMemoryQueue()
 
 async function queueRelevantWorkflows(event, eventType) {
+  if (event.instanceId == null) {
+    throw `No instanceId specified for ${eventType} - check event emitters.`
+  }
   const db = new CouchDB(event.instanceId)
   const workflowsToTrigger = await db.query("database/by_workflow_trigger", {
     key: [eventType],