|
|
|
@ -8,10 +8,14 @@ const { DocumentTypes } = require("../db/utils") |
|
|
|
const { doInTenant } = require("@budibase/backend-core/tenancy") |
|
|
|
const { definitions: triggerDefs } = require("../automations/triggerInfo") |
|
|
|
const { doInAppContext, getAppDB } = require("@budibase/backend-core/context") |
|
|
|
|
|
|
|
const { AutomationErrors } = require("../constants") |
|
|
|
const FILTER_STEP_ID = actions.ACTION_DEFINITIONS.FILTER.stepId |
|
|
|
const LOOP_STEP_ID = actions.ACTION_DEFINITIONS.LOOP.stepId |
|
|
|
|
|
|
|
const CRON_STEP_ID = triggerDefs.CRON.stepId |
|
|
|
const STOPPED_STATUS = { success: false, status: "STOPPED" } |
|
|
|
const { cloneDeep } = require("lodash/fp") |
|
|
|
const env = require("../environment") |
|
|
|
|
|
|
|
/** |
|
|
|
* The automation orchestrator is a class responsible for executing automations. |
|
|
|
@ -74,50 +78,208 @@ class Orchestrator { |
|
|
|
this.executionOutput.steps.push(stepObj) |
|
|
|
} |
|
|
|
|
|
|
|
updateContextAndOutput(loopStepNumber, step, output, result) { |
|
|
|
this.executionOutput.steps.splice(loopStepNumber, 0, { |
|
|
|
id: step.id, |
|
|
|
stepId: step.stepId, |
|
|
|
outputs: { |
|
|
|
...output, |
|
|
|
success: result.success, |
|
|
|
status: result.status, |
|
|
|
}, |
|
|
|
inputs: step.inputs, |
|
|
|
}) |
|
|
|
this._context.steps.splice(loopStepNumber, 0, { |
|
|
|
...output, |
|
|
|
success: result.success, |
|
|
|
status: result.status, |
|
|
|
}) |
|
|
|
} |
|
|
|
|
|
|
|
async execute() { |
|
|
|
let automation = this._automation |
|
|
|
const app = await this.getApp() |
|
|
|
let stopped = false |
|
|
|
let loopStep |
|
|
|
|
|
|
|
let stepCount = 0 |
|
|
|
let loopStepNumber |
|
|
|
let loopSteps = [] |
|
|
|
for (let step of automation.definition.steps) { |
|
|
|
// execution stopped, record state for that
|
|
|
|
if (stopped) { |
|
|
|
this.updateExecutionOutput(step.id, step.stepId, {}, STOPPED_STATUS) |
|
|
|
stepCount++ |
|
|
|
let input |
|
|
|
if (step.stepId === LOOP_STEP_ID) { |
|
|
|
loopStep = step |
|
|
|
loopStepNumber = stepCount |
|
|
|
continue |
|
|
|
} |
|
|
|
let stepFn = await this.getStepFunctionality(step.stepId) |
|
|
|
step.inputs = await processObject(step.inputs, this._context) |
|
|
|
step.inputs = automationUtils.cleanInputValues( |
|
|
|
step.inputs, |
|
|
|
step.schema.inputs |
|
|
|
) |
|
|
|
// appId is always passed
|
|
|
|
try { |
|
|
|
let tenantId = app.tenantId || DEFAULT_TENANT_ID |
|
|
|
const outputs = await doInTenant(tenantId, () => { |
|
|
|
return stepFn({ |
|
|
|
inputs: step.inputs, |
|
|
|
appId: this._appId, |
|
|
|
emitter: this._emitter, |
|
|
|
context: this._context, |
|
|
|
}) |
|
|
|
}) |
|
|
|
this._context.steps.push(outputs) |
|
|
|
// if filter causes us to stop execution don't break the loop, set a var
|
|
|
|
// so that we can finish iterating through the steps and record that it stopped
|
|
|
|
if (step.stepId === FILTER_STEP_ID && !outputs.success) { |
|
|
|
stopped = true |
|
|
|
this.updateExecutionOutput(step.id, step.stepId, step.inputs, { |
|
|
|
...outputs, |
|
|
|
...STOPPED_STATUS, |
|
|
|
}) |
|
|
|
|
|
|
|
if (loopStep) { |
|
|
|
input = await processObject(loopStep.inputs, this._context) |
|
|
|
} |
|
|
|
let iterations = loopStep ? input.binding.length : 1 |
|
|
|
let iterationCount = 0 |
|
|
|
for (let index = 0; index < iterations; index++) { |
|
|
|
let originalStepInput = cloneDeep(step.inputs) |
|
|
|
|
|
|
|
// Handle if the user has set a max iteration count or if it reaches the max limit set by us
|
|
|
|
if (loopStep) { |
|
|
|
// lets first of all handle the input
|
|
|
|
// if the input is array then use it, if it is a string then split it on every new line
|
|
|
|
let newInput = await processObject( |
|
|
|
loopStep.inputs, |
|
|
|
cloneDeep(this._context) |
|
|
|
) |
|
|
|
newInput = automationUtils.cleanInputValues( |
|
|
|
newInput, |
|
|
|
loopStep.schema.inputs |
|
|
|
) |
|
|
|
this._context.steps[loopStepNumber] = { |
|
|
|
currentItem: newInput.binding[index], |
|
|
|
} |
|
|
|
|
|
|
|
let tempOutput = { items: loopSteps, iterations: iterationCount } |
|
|
|
if ( |
|
|
|
(loopStep.inputs.option === "Array" && |
|
|
|
!Array.isArray(newInput.binding)) || |
|
|
|
(loopStep.inputs.option === "String" && |
|
|
|
typeof newInput.binding !== "string") |
|
|
|
) { |
|
|
|
this.updateContextAndOutput(loopStepNumber, step, tempOutput, { |
|
|
|
status: AutomationErrors.INCORRECT_TYPE, |
|
|
|
success: false, |
|
|
|
}) |
|
|
|
loopSteps = null |
|
|
|
loopStep = null |
|
|
|
break |
|
|
|
} |
|
|
|
|
|
|
|
// The "Loop" binding in the front end is "fake", so replace it here so the context can understand it
|
|
|
|
// Pretty hacky because we need to account for the row object
|
|
|
|
for (let [key, value] of Object.entries(originalStepInput)) { |
|
|
|
if (typeof value === "object") { |
|
|
|
for (let [innerKey, innerValue] of Object.entries( |
|
|
|
originalStepInput[key] |
|
|
|
)) { |
|
|
|
if (typeof innerValue === "string") { |
|
|
|
originalStepInput[key][innerKey] = |
|
|
|
automationUtils.substituteLoopStep( |
|
|
|
innerValue, |
|
|
|
`steps.${loopStepNumber}` |
|
|
|
) |
|
|
|
} |
|
|
|
} |
|
|
|
} else { |
|
|
|
if (typeof value === "string") { |
|
|
|
originalStepInput[key] = automationUtils.substituteLoopStep( |
|
|
|
value, |
|
|
|
`steps.${loopStepNumber}` |
|
|
|
) |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
if ( |
|
|
|
index === parseInt(env.AUTOMATION_MAX_ITERATIONS) || |
|
|
|
index === loopStep.inputs.iterations |
|
|
|
) { |
|
|
|
this.updateContextAndOutput(loopStepNumber, step, tempOutput, { |
|
|
|
status: AutomationErrors.MAX_ITERATIONS, |
|
|
|
success: true, |
|
|
|
}) |
|
|
|
loopSteps = null |
|
|
|
loopStep = null |
|
|
|
break |
|
|
|
} |
|
|
|
|
|
|
|
if ( |
|
|
|
this._context.steps[loopStepNumber]?.currentItem === |
|
|
|
loopStep.inputs.failure |
|
|
|
) { |
|
|
|
this.updateContextAndOutput(loopStepNumber, step, tempOutput, { |
|
|
|
status: AutomationErrors.FAILURE_CONDITION, |
|
|
|
success: false, |
|
|
|
}) |
|
|
|
loopSteps = null |
|
|
|
loopStep = null |
|
|
|
break |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
// execution stopped, record state for that
|
|
|
|
if (stopped) { |
|
|
|
this.updateExecutionOutput(step.id, step.stepId, {}, STOPPED_STATUS) |
|
|
|
continue |
|
|
|
} |
|
|
|
this.updateExecutionOutput(step.id, step.stepId, step.inputs, outputs) |
|
|
|
} catch (err) { |
|
|
|
console.error(`Automation error - ${step.stepId} - ${err}`) |
|
|
|
return err |
|
|
|
|
|
|
|
// If it's a loop step, we need to manually add the bindings to the context
|
|
|
|
let stepFn = await this.getStepFunctionality(step.stepId) |
|
|
|
let inputs = await processObject(originalStepInput, this._context) |
|
|
|
inputs = automationUtils.cleanInputValues(inputs, step.schema.inputs) |
|
|
|
try { |
|
|
|
// appId is always passed
|
|
|
|
let tenantId = app.tenantId || DEFAULT_TENANT_ID |
|
|
|
const outputs = await doInTenant(tenantId, () => { |
|
|
|
return stepFn({ |
|
|
|
inputs: inputs, |
|
|
|
appId: this._appId, |
|
|
|
emitter: this._emitter, |
|
|
|
context: this._context, |
|
|
|
}) |
|
|
|
}) |
|
|
|
this._context.steps[stepCount] = outputs |
|
|
|
// if filter causes us to stop execution don't break the loop, set a var
|
|
|
|
// so that we can finish iterating through the steps and record that it stopped
|
|
|
|
if (step.stepId === FILTER_STEP_ID && !outputs.success) { |
|
|
|
stopped = true |
|
|
|
this.updateExecutionOutput(step.id, step.stepId, step.inputs, { |
|
|
|
...outputs, |
|
|
|
...STOPPED_STATUS, |
|
|
|
}) |
|
|
|
continue |
|
|
|
} |
|
|
|
if (loopStep && loopSteps) { |
|
|
|
loopSteps.push(outputs) |
|
|
|
} else { |
|
|
|
this.updateExecutionOutput( |
|
|
|
step.id, |
|
|
|
step.stepId, |
|
|
|
step.inputs, |
|
|
|
outputs |
|
|
|
) |
|
|
|
} |
|
|
|
} catch (err) { |
|
|
|
console.error(`Automation error - ${step.stepId} - ${err}`) |
|
|
|
return err |
|
|
|
} |
|
|
|
if (loopStep) { |
|
|
|
iterationCount++ |
|
|
|
if (index === iterations - 1) { |
|
|
|
loopStep = null |
|
|
|
this._context.steps.splice(loopStepNumber, 1) |
|
|
|
break |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
if (loopSteps && loopSteps.length) { |
|
|
|
let tempOutput = { |
|
|
|
success: true, |
|
|
|
items: loopSteps, |
|
|
|
iterations: iterationCount, |
|
|
|
} |
|
|
|
this.executionOutput.steps.splice(loopStepNumber + 1, 0, { |
|
|
|
id: step.id, |
|
|
|
stepId: step.stepId, |
|
|
|
outputs: tempOutput, |
|
|
|
inputs: step.inputs, |
|
|
|
}) |
|
|
|
|
|
|
|
this._context.steps.splice(loopStepNumber, 0, tempOutput) |
|
|
|
loopSteps = null |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
return this.executionOutput |
|
|
|
} |
|
|
|
} |
|
|
|
|
Peter Clement
4 years ago
GitHub