493 lines
15 KiB
TypeScript
493 lines
15 KiB
TypeScript
import { default as threadUtils } from "./utils"
|
|
threadUtils.threadSetup()
|
|
import { isRecurring, disableCron, isErrorInOutput } from "../automations/utils"
|
|
import { default as actions } from "../automations/actions"
|
|
import { default as automationUtils } from "../automations/automationUtils"
|
|
import { default as AutomationEmitter } from "../events/AutomationEmitter"
|
|
import { generateAutomationMetadataID, isProdAppID } from "../db/utils"
|
|
import { definitions as triggerDefs } from "../automations/triggerInfo"
|
|
import { AutomationErrors, MAX_AUTOMATION_RECURRING_ERRORS } from "../constants"
|
|
import { storeLog } from "../automations/logging"
|
|
import { Automation, AutomationStep, AutomationStatus } from "@budibase/types"
|
|
import {
|
|
LoopStep,
|
|
LoopStepType,
|
|
LoopInput,
|
|
AutomationEvent,
|
|
TriggerOutput,
|
|
AutomationContext,
|
|
AutomationMetadata,
|
|
} from "../definitions/automations"
|
|
import { WorkerCallback } from "./definitions"
|
|
const { doInAppContext, getAppDB } = require("@budibase/backend-core/context")
|
|
const { logAlertWithInfo, logWarn } = require("@budibase/backend-core/logging")
|
|
const { processObject } = require("@budibase/string-templates")
|
|
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: true, status: AutomationStatus.STOPPED }
|
|
const { cloneDeep } = require("lodash/fp")
|
|
const env = require("../environment")
|
|
|
|
function typecastForLooping(loopStep: LoopStep, input: LoopInput) {
|
|
if (!input || !input.binding) {
|
|
return null
|
|
}
|
|
try {
|
|
switch (loopStep.inputs.option) {
|
|
case LoopStepType.ARRAY:
|
|
if (typeof input.binding === "string") {
|
|
return JSON.parse(input.binding)
|
|
}
|
|
break
|
|
case LoopStepType.STRING:
|
|
if (Array.isArray(input.binding)) {
|
|
return input.binding.join(",")
|
|
}
|
|
break
|
|
}
|
|
} catch (err) {
|
|
throw new Error("Unable to cast to correct type")
|
|
}
|
|
return input.binding
|
|
}
|
|
|
|
function getLoopIterations(loopStep: LoopStep, input: LoopInput) {
|
|
const binding = typecastForLooping(loopStep, input)
|
|
if (!loopStep || !binding) {
|
|
return 1
|
|
}
|
|
return Array.isArray(binding)
|
|
? binding.length
|
|
: automationUtils.stringSplit(binding).length
|
|
}
|
|
|
|
/**
|
|
* The automation orchestrator is a class responsible for executing automations.
|
|
* It handles the context of the automation and makes sure each step gets the correct
|
|
* inputs and handles any outputs.
|
|
*/
|
|
class Orchestrator {
|
|
_chainCount: number
|
|
_appId: string
|
|
_automation: Automation
|
|
_emitter: any
|
|
_context: AutomationContext
|
|
_repeat?: { jobId: string; jobKey: string }
|
|
executionOutput: AutomationContext
|
|
|
|
constructor(automation: Automation, triggerOutput: TriggerOutput, opts: any) {
|
|
const metadata = triggerOutput.metadata
|
|
this._chainCount = metadata ? metadata.automationChainCount : 0
|
|
this._appId = triggerOutput.appId as string
|
|
if (opts?.repeat) {
|
|
this._repeat = {
|
|
jobId: opts.repeat.jobId,
|
|
jobKey: opts.repeat.key,
|
|
}
|
|
}
|
|
const triggerStepId = automation.definition.trigger.stepId
|
|
triggerOutput = this.cleanupTriggerOutputs(triggerStepId, triggerOutput)
|
|
// remove from context
|
|
delete triggerOutput.appId
|
|
delete triggerOutput.metadata
|
|
// step zero is never used as the template string is zero indexed for customer facing
|
|
this._context = { steps: [{}], trigger: triggerOutput }
|
|
this._automation = automation
|
|
// create an emitter which has the chain count for this automation run in it, so it can block
|
|
// excessive chaining if required
|
|
this._emitter = new AutomationEmitter(this._chainCount + 1)
|
|
this.executionOutput = { trigger: {}, steps: [] }
|
|
// setup the execution output
|
|
const triggerId = automation.definition.trigger.id
|
|
this.updateExecutionOutput(triggerId, triggerStepId, null, triggerOutput)
|
|
}
|
|
|
|
cleanupTriggerOutputs(stepId: string, triggerOutput: TriggerOutput) {
|
|
if (stepId === CRON_STEP_ID) {
|
|
triggerOutput.timestamp = Date.now()
|
|
}
|
|
return triggerOutput
|
|
}
|
|
|
|
async getStepFunctionality(stepId: string) {
|
|
let step = await actions.getAction(stepId)
|
|
if (step == null) {
|
|
throw `Cannot find automation step by name ${stepId}`
|
|
}
|
|
return step
|
|
}
|
|
|
|
async getMetadata(): Promise<AutomationMetadata> {
|
|
const metadataId = generateAutomationMetadataID(this._automation._id)
|
|
const db = getAppDB()
|
|
let metadata: AutomationMetadata
|
|
try {
|
|
metadata = await db.get(metadataId)
|
|
} catch (err) {
|
|
metadata = {
|
|
_id: metadataId,
|
|
errorCount: 0,
|
|
}
|
|
}
|
|
return metadata
|
|
}
|
|
|
|
async stopCron(reason: string) {
|
|
if (!this._repeat) {
|
|
return
|
|
}
|
|
logWarn(
|
|
`CRON disabled reason=${reason} - ${this._appId}/${this._automation._id}`
|
|
)
|
|
const automation = this._automation
|
|
const trigger = automation.definition.trigger
|
|
await disableCron(this._repeat?.jobId, this._repeat?.jobKey)
|
|
this.updateExecutionOutput(
|
|
trigger.id,
|
|
trigger.stepId,
|
|
{},
|
|
{
|
|
status: AutomationStatus.STOPPED_ERROR,
|
|
success: false,
|
|
}
|
|
)
|
|
await storeLog(automation, this.executionOutput)
|
|
}
|
|
|
|
async checkIfShouldStop(metadata: AutomationMetadata): Promise<boolean> {
|
|
if (!metadata.errorCount || !this._repeat) {
|
|
return false
|
|
}
|
|
if (metadata.errorCount >= MAX_AUTOMATION_RECURRING_ERRORS) {
|
|
await this.stopCron("errors")
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
async updateMetadata(metadata: AutomationMetadata) {
|
|
const output = this.executionOutput,
|
|
automation = this._automation
|
|
if (!output || !isRecurring(automation)) {
|
|
return
|
|
}
|
|
const count = metadata.errorCount
|
|
const isError = isErrorInOutput(output)
|
|
// nothing to do in this scenario, escape
|
|
if (!count && !isError) {
|
|
return
|
|
}
|
|
if (isError) {
|
|
metadata.errorCount = count ? count + 1 : 1
|
|
} else {
|
|
metadata.errorCount = 0
|
|
}
|
|
const db = getAppDB()
|
|
try {
|
|
await db.put(metadata)
|
|
} catch (err) {
|
|
logAlertWithInfo(
|
|
"Failed to write automation metadata",
|
|
db.name,
|
|
automation._id,
|
|
err
|
|
)
|
|
}
|
|
}
|
|
|
|
updateExecutionOutput(id: string, stepId: string, inputs: any, outputs: any) {
|
|
const stepObj = { id, stepId, inputs, outputs }
|
|
// replacing trigger when disabling CRON
|
|
if (
|
|
stepId === CRON_STEP_ID &&
|
|
outputs.status === AutomationStatus.STOPPED_ERROR
|
|
) {
|
|
this.executionOutput.trigger = stepObj
|
|
this.executionOutput.steps = [stepObj]
|
|
return
|
|
}
|
|
// first entry is always the trigger (constructor)
|
|
if (
|
|
this.executionOutput.steps.length === 0 ||
|
|
this.executionOutput.trigger.id === id
|
|
) {
|
|
this.executionOutput.trigger = stepObj
|
|
}
|
|
this.executionOutput.steps.push(stepObj)
|
|
}
|
|
|
|
updateContextAndOutput(
|
|
loopStepNumber: number | undefined,
|
|
step: AutomationStep,
|
|
output: any,
|
|
result: { success: boolean; status: string }
|
|
) {
|
|
if (!loopStepNumber) {
|
|
throw new Error("No loop step number provided.")
|
|
}
|
|
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
|
|
let stopped = false
|
|
let loopStep: AutomationStep | undefined = undefined
|
|
|
|
let stepCount = 0
|
|
let loopStepNumber: any = undefined
|
|
let loopSteps: LoopStep[] | undefined = []
|
|
let metadata
|
|
|
|
// check if this is a recurring automation,
|
|
if (isProdAppID(this._appId) && isRecurring(automation)) {
|
|
metadata = await this.getMetadata()
|
|
const shouldStop = await this.checkIfShouldStop(metadata)
|
|
if (shouldStop) {
|
|
return
|
|
}
|
|
}
|
|
|
|
for (let step of automation.definition.steps) {
|
|
stepCount++
|
|
let input,
|
|
iterations = 1,
|
|
iterationCount = 0
|
|
if (step.stepId === LOOP_STEP_ID) {
|
|
loopStep = step
|
|
loopStepNumber = stepCount
|
|
continue
|
|
}
|
|
|
|
if (loopStep) {
|
|
input = await processObject(loopStep.inputs, this._context)
|
|
iterations = getLoopIterations(loopStep as LoopStep, input)
|
|
}
|
|
|
|
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 && input.binding) {
|
|
let newInput = await processObject(
|
|
loopStep.inputs,
|
|
cloneDeep(this._context)
|
|
)
|
|
|
|
let tempOutput = { items: loopSteps, iterations: iterationCount }
|
|
try {
|
|
newInput.binding = typecastForLooping(
|
|
loopStep as LoopStep,
|
|
newInput
|
|
)
|
|
} catch (err) {
|
|
this.updateContextAndOutput(loopStepNumber, step, tempOutput, {
|
|
status: AutomationErrors.INCORRECT_TYPE,
|
|
success: false,
|
|
})
|
|
loopSteps = undefined
|
|
loopStep = undefined
|
|
break
|
|
}
|
|
|
|
let item
|
|
if (
|
|
typeof loopStep.inputs.binding === "string" &&
|
|
loopStep.inputs.option === "String"
|
|
) {
|
|
item = automationUtils.stringSplit(newInput.binding)
|
|
} else {
|
|
item = loopStep.inputs.binding
|
|
}
|
|
|
|
this._context.steps[loopStepNumber] = {
|
|
currentItem: item[index],
|
|
}
|
|
|
|
// 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 === env.AUTOMATION_MAX_ITERATION ||
|
|
index === parseInt(loopStep.inputs.iterations)
|
|
) {
|
|
this.updateContextAndOutput(loopStepNumber, step, tempOutput, {
|
|
status: AutomationErrors.MAX_ITERATIONS,
|
|
success: true,
|
|
})
|
|
loopSteps = undefined
|
|
loopStep = undefined
|
|
break
|
|
}
|
|
|
|
let isFailure = false
|
|
const currentItem = this._context.steps[loopStepNumber]?.currentItem
|
|
if (currentItem && typeof currentItem === "object") {
|
|
isFailure = Object.keys(currentItem).some(value => {
|
|
return currentItem[value] === loopStep?.inputs.failure
|
|
})
|
|
} else {
|
|
isFailure = currentItem && currentItem === loopStep.inputs.failure
|
|
}
|
|
|
|
if (isFailure) {
|
|
this.updateContextAndOutput(loopStepNumber, step, tempOutput, {
|
|
status: AutomationErrors.FAILURE_CONDITION,
|
|
success: false,
|
|
})
|
|
loopSteps = undefined
|
|
loopStep = undefined
|
|
break
|
|
}
|
|
}
|
|
|
|
// execution stopped, record state for that
|
|
if (stopped) {
|
|
this.updateExecutionOutput(step.id, step.stepId, {}, STOPPED_STATUS)
|
|
continue
|
|
}
|
|
|
|
// 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
|
|
const outputs = await 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.result) {
|
|
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 = undefined
|
|
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 = undefined
|
|
}
|
|
}
|
|
|
|
// store the logs for the automation run
|
|
await storeLog(this._automation, this.executionOutput)
|
|
if (isProdAppID(this._appId) && isRecurring(automation) && metadata) {
|
|
await this.updateMetadata(metadata)
|
|
}
|
|
return this.executionOutput
|
|
}
|
|
}
|
|
|
|
export function execute(input: AutomationEvent, callback: WorkerCallback) {
|
|
const appId = input.data.event.appId
|
|
if (!appId) {
|
|
throw new Error("Unable to execute, event doesn't contain app ID.")
|
|
}
|
|
doInAppContext(appId, async () => {
|
|
const automationOrchestrator = new Orchestrator(
|
|
input.data.automation,
|
|
input.data.event,
|
|
input.opts
|
|
)
|
|
try {
|
|
const response = await automationOrchestrator.execute()
|
|
callback(null, response)
|
|
} catch (err) {
|
|
callback(err)
|
|
}
|
|
})
|
|
}
|
|
|
|
export const removeStalled = async (input: AutomationEvent) => {
|
|
const appId = input.data.event.appId
|
|
if (!appId) {
|
|
throw new Error("Unable to execute, event doesn't contain app ID.")
|
|
}
|
|
await doInAppContext(appId, async () => {
|
|
const automationOrchestrator = new Orchestrator(
|
|
input.data.automation,
|
|
input.data.event,
|
|
input.opts
|
|
)
|
|
await automationOrchestrator.stopCron("stalled")
|
|
})
|
|
}
|