Merge pull request #11648 from Budibase/revert-automation-changes
Revert "Merge pull request #11489 from Budibase/fix/automation-disable"
This commit is contained in:
commit
47fb79b708
11 changed files with 56 additions and 157 deletions
|
@ -8,7 +8,6 @@ import {
|
||||||
DatabasePutOpts,
|
DatabasePutOpts,
|
||||||
DatabaseCreateIndexOpts,
|
DatabaseCreateIndexOpts,
|
||||||
DatabaseDeleteIndexOpts,
|
DatabaseDeleteIndexOpts,
|
||||||
DocExistsResponse,
|
|
||||||
Document,
|
Document,
|
||||||
isDocument,
|
isDocument,
|
||||||
} from "@budibase/types"
|
} from "@budibase/types"
|
||||||
|
@ -121,19 +120,6 @@ export class DatabaseImpl implements Database {
|
||||||
return this.updateOutput(() => db.get(id))
|
return this.updateOutput(() => db.get(id))
|
||||||
}
|
}
|
||||||
|
|
||||||
async docExists(docId: string): Promise<DocExistsResponse> {
|
|
||||||
const db = await this.checkSetup()
|
|
||||||
let _rev, exists
|
|
||||||
try {
|
|
||||||
const { etag } = await db.head(docId)
|
|
||||||
_rev = etag
|
|
||||||
exists = true
|
|
||||||
} catch (err) {
|
|
||||||
exists = false
|
|
||||||
}
|
|
||||||
return { _rev, exists }
|
|
||||||
}
|
|
||||||
|
|
||||||
async remove(idOrDoc: string | Document, rev?: string) {
|
async remove(idOrDoc: string | Document, rev?: string) {
|
||||||
const db = await this.checkSetup()
|
const db = await this.checkSetup()
|
||||||
let _id: string
|
let _id: string
|
||||||
|
|
|
@ -1 +1 @@
|
||||||
Subproject commit b7815e099bbd5e1410185c464dbd54f7287e732f
|
Subproject commit 02626390cde905a248cb60729968667c9e49fae9
|
|
@ -39,8 +39,9 @@ import {
|
||||||
} from "../../db/defaultData/datasource_bb_default"
|
} from "../../db/defaultData/datasource_bb_default"
|
||||||
import { removeAppFromUserRoles } from "../../utilities/workerRequests"
|
import { removeAppFromUserRoles } from "../../utilities/workerRequests"
|
||||||
import { stringToReadStream } from "../../utilities"
|
import { stringToReadStream } from "../../utilities"
|
||||||
import { doesUserHaveLock } from "../../utilities/redis"
|
import { doesUserHaveLock, getLocksById } from "../../utilities/redis"
|
||||||
import { cleanupAutomations } from "../../automations/utils"
|
import { cleanupAutomations } from "../../automations/utils"
|
||||||
|
import { checkAppMetadata } from "../../automations/logging"
|
||||||
import { getUniqueRows } from "../../utilities/usageQuota/rows"
|
import { getUniqueRows } from "../../utilities/usageQuota/rows"
|
||||||
import { groups, licensing, quotas } from "@budibase/pro"
|
import { groups, licensing, quotas } from "@budibase/pro"
|
||||||
import {
|
import {
|
||||||
|
@ -50,6 +51,7 @@ import {
|
||||||
PlanType,
|
PlanType,
|
||||||
Screen,
|
Screen,
|
||||||
UserCtx,
|
UserCtx,
|
||||||
|
ContextUser,
|
||||||
} from "@budibase/types"
|
} from "@budibase/types"
|
||||||
import { BASE_LAYOUT_PROP_IDS } from "../../constants/layouts"
|
import { BASE_LAYOUT_PROP_IDS } from "../../constants/layouts"
|
||||||
import sdk from "../../sdk"
|
import sdk from "../../sdk"
|
||||||
|
|
|
@ -20,7 +20,7 @@ import {
|
||||||
Automation,
|
Automation,
|
||||||
AutomationActionStepId,
|
AutomationActionStepId,
|
||||||
AutomationResults,
|
AutomationResults,
|
||||||
Ctx,
|
BBContext,
|
||||||
} from "@budibase/types"
|
} from "@budibase/types"
|
||||||
import { getActionDefinitions as actionDefs } from "../../automations/actions"
|
import { getActionDefinitions as actionDefs } from "../../automations/actions"
|
||||||
import sdk from "../../sdk"
|
import sdk from "../../sdk"
|
||||||
|
@ -73,7 +73,7 @@ function cleanAutomationInputs(automation: Automation) {
|
||||||
return automation
|
return automation
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function create(ctx: Ctx) {
|
export async function create(ctx: BBContext) {
|
||||||
const db = context.getAppDB()
|
const db = context.getAppDB()
|
||||||
let automation = ctx.request.body
|
let automation = ctx.request.body
|
||||||
automation.appId = ctx.appId
|
automation.appId = ctx.appId
|
||||||
|
@ -142,7 +142,7 @@ export async function handleStepEvents(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function update(ctx: Ctx) {
|
export async function update(ctx: BBContext) {
|
||||||
const db = context.getAppDB()
|
const db = context.getAppDB()
|
||||||
let automation = ctx.request.body
|
let automation = ctx.request.body
|
||||||
automation.appId = ctx.appId
|
automation.appId = ctx.appId
|
||||||
|
@ -193,7 +193,7 @@ export async function update(ctx: Ctx) {
|
||||||
builderSocket?.emitAutomationUpdate(ctx, automation)
|
builderSocket?.emitAutomationUpdate(ctx, automation)
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function fetch(ctx: Ctx) {
|
export async function fetch(ctx: BBContext) {
|
||||||
const db = context.getAppDB()
|
const db = context.getAppDB()
|
||||||
const response = await db.allDocs(
|
const response = await db.allDocs(
|
||||||
getAutomationParams(null, {
|
getAutomationParams(null, {
|
||||||
|
@ -203,11 +203,12 @@ export async function fetch(ctx: Ctx) {
|
||||||
ctx.body = response.rows.map(row => row.doc)
|
ctx.body = response.rows.map(row => row.doc)
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function find(ctx: Ctx) {
|
export async function find(ctx: BBContext) {
|
||||||
ctx.body = await sdk.automations.get(ctx.params.id)
|
const db = context.getAppDB()
|
||||||
|
ctx.body = await db.get(ctx.params.id)
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function destroy(ctx: Ctx) {
|
export async function destroy(ctx: BBContext) {
|
||||||
const db = context.getAppDB()
|
const db = context.getAppDB()
|
||||||
const automationId = ctx.params.id
|
const automationId = ctx.params.id
|
||||||
const oldAutomation = await db.get<Automation>(automationId)
|
const oldAutomation = await db.get<Automation>(automationId)
|
||||||
|
@ -221,11 +222,11 @@ export async function destroy(ctx: Ctx) {
|
||||||
builderSocket?.emitAutomationDeletion(ctx, automationId)
|
builderSocket?.emitAutomationDeletion(ctx, automationId)
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function logSearch(ctx: Ctx) {
|
export async function logSearch(ctx: BBContext) {
|
||||||
ctx.body = await automations.logs.logSearch(ctx.request.body)
|
ctx.body = await automations.logs.logSearch(ctx.request.body)
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function clearLogError(ctx: Ctx) {
|
export async function clearLogError(ctx: BBContext) {
|
||||||
const { automationId, appId } = ctx.request.body
|
const { automationId, appId } = ctx.request.body
|
||||||
await context.doInAppContext(appId, async () => {
|
await context.doInAppContext(appId, async () => {
|
||||||
const db = context.getProdAppDB()
|
const db = context.getProdAppDB()
|
||||||
|
@ -244,15 +245,15 @@ export async function clearLogError(ctx: Ctx) {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function getActionList(ctx: Ctx) {
|
export async function getActionList(ctx: BBContext) {
|
||||||
ctx.body = await getActionDefinitions()
|
ctx.body = await getActionDefinitions()
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function getTriggerList(ctx: Ctx) {
|
export async function getTriggerList(ctx: BBContext) {
|
||||||
ctx.body = getTriggerDefinitions()
|
ctx.body = getTriggerDefinitions()
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function getDefinitionList(ctx: Ctx) {
|
export async function getDefinitionList(ctx: BBContext) {
|
||||||
ctx.body = {
|
ctx.body = {
|
||||||
trigger: getTriggerDefinitions(),
|
trigger: getTriggerDefinitions(),
|
||||||
action: await getActionDefinitions(),
|
action: await getActionDefinitions(),
|
||||||
|
@ -265,7 +266,7 @@ export async function getDefinitionList(ctx: Ctx) {
|
||||||
* *
|
* *
|
||||||
*********************/
|
*********************/
|
||||||
|
|
||||||
export async function trigger(ctx: Ctx) {
|
export async function trigger(ctx: BBContext) {
|
||||||
const db = context.getAppDB()
|
const db = context.getAppDB()
|
||||||
let automation = await db.get<Automation>(ctx.params.id)
|
let automation = await db.get<Automation>(ctx.params.id)
|
||||||
|
|
||||||
|
@ -310,7 +311,7 @@ function prepareTestInput(input: any) {
|
||||||
return input
|
return input
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function test(ctx: Ctx) {
|
export async function test(ctx: BBContext) {
|
||||||
const db = context.getAppDB()
|
const db = context.getAppDB()
|
||||||
let automation = await db.get<Automation>(ctx.params.id)
|
let automation = await db.get<Automation>(ctx.params.id)
|
||||||
await setTestFlag(automation._id!)
|
await setTestFlag(automation._id!)
|
||||||
|
|
|
@ -6,11 +6,11 @@ import { isDevAppID } from "../db/utils"
|
||||||
// need this to call directly, so we can get a response
|
// need this to call directly, so we can get a response
|
||||||
import { automationQueue } from "./bullboard"
|
import { automationQueue } from "./bullboard"
|
||||||
import { checkTestFlag } from "../utilities/redis"
|
import { checkTestFlag } from "../utilities/redis"
|
||||||
|
import * as utils from "./utils"
|
||||||
import env from "../environment"
|
import env from "../environment"
|
||||||
import { context, db as dbCore } from "@budibase/backend-core"
|
import { context, db as dbCore } from "@budibase/backend-core"
|
||||||
import { Automation, Row, AutomationData, AutomationJob } from "@budibase/types"
|
import { Automation, Row, AutomationData, AutomationJob } from "@budibase/types"
|
||||||
import { executeSynchronously } from "../threads/automation"
|
import { executeSynchronously } from "../threads/automation"
|
||||||
import sdk from "../sdk"
|
|
||||||
|
|
||||||
export const TRIGGER_DEFINITIONS = definitions
|
export const TRIGGER_DEFINITIONS = definitions
|
||||||
const JOB_OPTS = {
|
const JOB_OPTS = {
|
||||||
|
@ -142,7 +142,7 @@ export async function rebootTrigger() {
|
||||||
let automations = await getAllAutomations()
|
let automations = await getAllAutomations()
|
||||||
let rebootEvents = []
|
let rebootEvents = []
|
||||||
for (let automation of automations) {
|
for (let automation of automations) {
|
||||||
if (sdk.automations.isReboot(automation)) {
|
if (utils.isRebootTrigger(automation)) {
|
||||||
const job = {
|
const job = {
|
||||||
automation,
|
automation,
|
||||||
event: {
|
event: {
|
||||||
|
|
|
@ -16,14 +16,13 @@ import {
|
||||||
} from "@budibase/types"
|
} from "@budibase/types"
|
||||||
import sdk from "../sdk"
|
import sdk from "../sdk"
|
||||||
|
|
||||||
|
const REBOOT_CRON = "@reboot"
|
||||||
const WH_STEP_ID = definitions.WEBHOOK.stepId
|
const WH_STEP_ID = definitions.WEBHOOK.stepId
|
||||||
|
const CRON_STEP_ID = definitions.CRON.stepId
|
||||||
const Runner = new Thread(ThreadType.AUTOMATION)
|
const Runner = new Thread(ThreadType.AUTOMATION)
|
||||||
|
|
||||||
function loggingArgs(
|
function loggingArgs(job: AutomationJob) {
|
||||||
job: AutomationJob,
|
return [
|
||||||
timing?: { start: number; complete?: boolean }
|
|
||||||
) {
|
|
||||||
const logs: any[] = [
|
|
||||||
{
|
{
|
||||||
_logKey: "automation",
|
_logKey: "automation",
|
||||||
trigger: job.data.automation.definition.trigger.event,
|
trigger: job.data.automation.definition.trigger.event,
|
||||||
|
@ -33,53 +32,24 @@ function loggingArgs(
|
||||||
jobId: job.id,
|
jobId: job.id,
|
||||||
},
|
},
|
||||||
]
|
]
|
||||||
if (timing?.start) {
|
|
||||||
logs.push({
|
|
||||||
_logKey: "startTime",
|
|
||||||
start: timing.start,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
if (timing?.start && timing?.complete) {
|
|
||||||
const end = new Date().getTime()
|
|
||||||
const duration = end - timing.start
|
|
||||||
logs.push({
|
|
||||||
_logKey: "endTime",
|
|
||||||
end,
|
|
||||||
})
|
|
||||||
logs.push({
|
|
||||||
_logKey: "duration",
|
|
||||||
duration,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
return logs
|
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function processEvent(job: AutomationJob) {
|
export async function processEvent(job: AutomationJob) {
|
||||||
const appId = job.data.event.appId!
|
const appId = job.data.event.appId!
|
||||||
const automationId = job.data.automation._id!
|
const automationId = job.data.automation._id!
|
||||||
const start = new Date().getTime()
|
|
||||||
const task = async () => {
|
const task = async () => {
|
||||||
try {
|
try {
|
||||||
// need to actually await these so that an error can be captured properly
|
// need to actually await these so that an error can be captured properly
|
||||||
console.log("automation running", ...loggingArgs(job, { start }))
|
console.log("automation running", ...loggingArgs(job))
|
||||||
|
|
||||||
const runFn = () => Runner.run(job)
|
const runFn = () => Runner.run(job)
|
||||||
const result = await quotas.addAutomation(runFn, {
|
const result = await quotas.addAutomation(runFn, {
|
||||||
automationId,
|
automationId,
|
||||||
})
|
})
|
||||||
const end = new Date().getTime()
|
console.log("automation completed", ...loggingArgs(job))
|
||||||
const duration = end - start
|
|
||||||
console.log(
|
|
||||||
"automation completed",
|
|
||||||
...loggingArgs(job, { start, complete: true })
|
|
||||||
)
|
|
||||||
return result
|
return result
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
console.error(
|
console.error(`automation was unable to run`, err, ...loggingArgs(job))
|
||||||
`automation was unable to run`,
|
|
||||||
err,
|
|
||||||
...loggingArgs(job, { start, complete: true })
|
|
||||||
)
|
|
||||||
return { err }
|
return { err }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -158,6 +128,19 @@ export async function clearMetadata() {
|
||||||
await db.bulkDocs(automationMetadata)
|
await db.bulkDocs(automationMetadata)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export function isCronTrigger(auto: Automation) {
|
||||||
|
return (
|
||||||
|
auto &&
|
||||||
|
auto.definition.trigger &&
|
||||||
|
auto.definition.trigger.stepId === CRON_STEP_ID
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
export function isRebootTrigger(auto: Automation) {
|
||||||
|
const trigger = auto ? auto.definition.trigger : null
|
||||||
|
return isCronTrigger(auto) && trigger?.inputs.cron === REBOOT_CRON
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* This function handles checking of any cron jobs that need to be enabled/updated.
|
* This function handles checking of any cron jobs that need to be enabled/updated.
|
||||||
* @param {string} appId The ID of the app in which we are checking for webhooks
|
* @param {string} appId The ID of the app in which we are checking for webhooks
|
||||||
|
@ -165,13 +148,13 @@ export async function clearMetadata() {
|
||||||
*/
|
*/
|
||||||
export async function enableCronTrigger(appId: any, automation: Automation) {
|
export async function enableCronTrigger(appId: any, automation: Automation) {
|
||||||
const trigger = automation ? automation.definition.trigger : null
|
const trigger = automation ? automation.definition.trigger : null
|
||||||
const validCron = sdk.automations.isCron(automation) && trigger?.inputs.cron
|
|
||||||
const needsCreated =
|
|
||||||
!sdk.automations.isReboot(automation) &&
|
|
||||||
!sdk.automations.disabled(automation)
|
|
||||||
|
|
||||||
// need to create cron job
|
// need to create cron job
|
||||||
if (validCron && needsCreated) {
|
if (
|
||||||
|
isCronTrigger(automation) &&
|
||||||
|
!isRebootTrigger(automation) &&
|
||||||
|
trigger?.inputs.cron
|
||||||
|
) {
|
||||||
// make a job id rather than letting Bull decide, makes it easier to handle on way out
|
// make a job id rather than letting Bull decide, makes it easier to handle on way out
|
||||||
const jobId = `${appId}_cron_${newid()}`
|
const jobId = `${appId}_cron_${newid()}`
|
||||||
const job: any = await automationQueue.add(
|
const job: any = await automationQueue.add(
|
||||||
|
|
|
@ -1,38 +0,0 @@
|
||||||
import { context } from "@budibase/backend-core"
|
|
||||||
import { Automation, AutomationState, DocumentType } from "@budibase/types"
|
|
||||||
import { definitions } from "../../../automations/triggerInfo"
|
|
||||||
|
|
||||||
const REBOOT_CRON = "@reboot"
|
|
||||||
|
|
||||||
export async function exists(automationId: string) {
|
|
||||||
if (!automationId?.startsWith(DocumentType.AUTOMATION)) {
|
|
||||||
throw new Error("Invalid automation ID.")
|
|
||||||
}
|
|
||||||
const db = context.getAppDB()
|
|
||||||
return db.docExists(automationId)
|
|
||||||
}
|
|
||||||
|
|
||||||
export async function get(automationId: string) {
|
|
||||||
const db = context.getAppDB()
|
|
||||||
return (await db.get(automationId)) as Automation
|
|
||||||
}
|
|
||||||
|
|
||||||
export function disabled(automation: Automation) {
|
|
||||||
return automation.state === AutomationState.DISABLED || !hasSteps(automation)
|
|
||||||
}
|
|
||||||
|
|
||||||
export function isCron(automation: Automation) {
|
|
||||||
return (
|
|
||||||
automation?.definition.trigger &&
|
|
||||||
automation?.definition.trigger.stepId === definitions.CRON.stepId
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
export function isReboot(automation: Automation) {
|
|
||||||
const trigger = automation?.definition.trigger
|
|
||||||
return isCron(automation) && trigger?.inputs.cron === REBOOT_CRON
|
|
||||||
}
|
|
||||||
|
|
||||||
export function hasSteps(automation: Automation) {
|
|
||||||
return automation?.definition?.steps?.length > 0
|
|
||||||
}
|
|
|
@ -1,9 +1,7 @@
|
||||||
import * as webhook from "./webhook"
|
import * as webhook from "./webhook"
|
||||||
import * as utils from "./utils"
|
import * as utils from "./utils"
|
||||||
import * as automations from "./automations"
|
|
||||||
|
|
||||||
export default {
|
export default {
|
||||||
webhook,
|
webhook,
|
||||||
utils,
|
utils,
|
||||||
...automations,
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -2,9 +2,9 @@ import { default as threadUtils } from "./utils"
|
||||||
import { Job } from "bull"
|
import { Job } from "bull"
|
||||||
threadUtils.threadSetup()
|
threadUtils.threadSetup()
|
||||||
import {
|
import {
|
||||||
|
isRecurring,
|
||||||
disableCronById,
|
disableCronById,
|
||||||
isErrorInOutput,
|
isErrorInOutput,
|
||||||
isRecurring,
|
|
||||||
} from "../automations/utils"
|
} from "../automations/utils"
|
||||||
import * as actions from "../automations/actions"
|
import * as actions from "../automations/actions"
|
||||||
import * as automationUtils from "../automations/automationUtils"
|
import * as automationUtils from "../automations/automationUtils"
|
||||||
|
@ -15,17 +15,17 @@ import { AutomationErrors, MAX_AUTOMATION_RECURRING_ERRORS } from "../constants"
|
||||||
import { storeLog } from "../automations/logging"
|
import { storeLog } from "../automations/logging"
|
||||||
import {
|
import {
|
||||||
Automation,
|
Automation,
|
||||||
AutomationData,
|
|
||||||
AutomationJob,
|
|
||||||
AutomationMetadata,
|
|
||||||
AutomationStatus,
|
|
||||||
AutomationStep,
|
AutomationStep,
|
||||||
|
AutomationStatus,
|
||||||
|
AutomationMetadata,
|
||||||
|
AutomationJob,
|
||||||
|
AutomationData,
|
||||||
} from "@budibase/types"
|
} from "@budibase/types"
|
||||||
import {
|
import {
|
||||||
AutomationContext,
|
|
||||||
LoopInput,
|
|
||||||
LoopStep,
|
LoopStep,
|
||||||
|
LoopInput,
|
||||||
TriggerOutput,
|
TriggerOutput,
|
||||||
|
AutomationContext,
|
||||||
} from "../definitions/automations"
|
} from "../definitions/automations"
|
||||||
import { WorkerCallback } from "./definitions"
|
import { WorkerCallback } from "./definitions"
|
||||||
import { context, logging } from "@budibase/backend-core"
|
import { context, logging } from "@budibase/backend-core"
|
||||||
|
@ -34,8 +34,6 @@ import { cloneDeep } from "lodash/fp"
|
||||||
import { performance } from "perf_hooks"
|
import { performance } from "perf_hooks"
|
||||||
import * as sdkUtils from "../sdk/utils"
|
import * as sdkUtils from "../sdk/utils"
|
||||||
import env from "../environment"
|
import env from "../environment"
|
||||||
import sdk from "../sdk"
|
|
||||||
|
|
||||||
const FILTER_STEP_ID = actions.BUILTIN_ACTION_DEFINITIONS.FILTER.stepId
|
const FILTER_STEP_ID = actions.BUILTIN_ACTION_DEFINITIONS.FILTER.stepId
|
||||||
const LOOP_STEP_ID = actions.BUILTIN_ACTION_DEFINITIONS.LOOP.stepId
|
const LOOP_STEP_ID = actions.BUILTIN_ACTION_DEFINITIONS.LOOP.stepId
|
||||||
const CRON_STEP_ID = triggerDefs.CRON.stepId
|
const CRON_STEP_ID = triggerDefs.CRON.stepId
|
||||||
|
@ -516,8 +514,7 @@ class Orchestrator {
|
||||||
|
|
||||||
export function execute(job: Job<AutomationData>, callback: WorkerCallback) {
|
export function execute(job: Job<AutomationData>, callback: WorkerCallback) {
|
||||||
const appId = job.data.event.appId
|
const appId = job.data.event.appId
|
||||||
const automation = job.data.automation
|
const automationId = job.data.automation._id
|
||||||
const automationId = automation._id
|
|
||||||
if (!appId) {
|
if (!appId) {
|
||||||
throw new Error("Unable to execute, event doesn't contain app ID.")
|
throw new Error("Unable to execute, event doesn't contain app ID.")
|
||||||
}
|
}
|
||||||
|
@ -528,30 +525,10 @@ export function execute(job: Job<AutomationData>, callback: WorkerCallback) {
|
||||||
appId,
|
appId,
|
||||||
automationId,
|
automationId,
|
||||||
task: async () => {
|
task: async () => {
|
||||||
let automation = job.data.automation,
|
|
||||||
isCron = sdk.automations.isCron(job.data.automation),
|
|
||||||
notFound = false
|
|
||||||
try {
|
|
||||||
automation = await sdk.automations.get(automationId)
|
|
||||||
} catch (err: any) {
|
|
||||||
// automation no longer exists
|
|
||||||
notFound = err
|
|
||||||
}
|
|
||||||
const disabled = sdk.automations.disabled(automation)
|
|
||||||
const stopAutomation = disabled || notFound
|
|
||||||
const envVars = await sdkUtils.getEnvironmentVariables()
|
const envVars = await sdkUtils.getEnvironmentVariables()
|
||||||
// put into automation thread for whole context
|
// put into automation thread for whole context
|
||||||
await context.doInEnvironmentContext(envVars, async () => {
|
await context.doInEnvironmentContext(envVars, async () => {
|
||||||
const automationOrchestrator = new Orchestrator(job)
|
const automationOrchestrator = new Orchestrator(job)
|
||||||
// hard stop on automations
|
|
||||||
if (isCron && stopAutomation) {
|
|
||||||
await automationOrchestrator.stopCron(
|
|
||||||
disabled ? "disabled" : "not_found"
|
|
||||||
)
|
|
||||||
}
|
|
||||||
if (stopAutomation) {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
try {
|
try {
|
||||||
const response = await automationOrchestrator.execute()
|
const response = await automationOrchestrator.execute()
|
||||||
callback(null, response)
|
callback(null, response)
|
||||||
|
@ -580,10 +557,11 @@ export function executeSynchronously(job: Job) {
|
||||||
// put into automation thread for whole context
|
// put into automation thread for whole context
|
||||||
return context.doInEnvironmentContext(envVars, async () => {
|
return context.doInEnvironmentContext(envVars, async () => {
|
||||||
const automationOrchestrator = new Orchestrator(job)
|
const automationOrchestrator = new Orchestrator(job)
|
||||||
return await Promise.race([
|
const response = await Promise.race([
|
||||||
automationOrchestrator.execute(),
|
automationOrchestrator.execute(),
|
||||||
timeoutPromise,
|
timeoutPromise,
|
||||||
])
|
])
|
||||||
|
return response
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
|
@ -100,10 +100,6 @@ export const AutomationStepIdArray = [
|
||||||
...Object.values(AutomationTriggerStepId),
|
...Object.values(AutomationTriggerStepId),
|
||||||
]
|
]
|
||||||
|
|
||||||
export enum AutomationState {
|
|
||||||
DISABLED = "disabled",
|
|
||||||
}
|
|
||||||
|
|
||||||
export interface Automation extends Document {
|
export interface Automation extends Document {
|
||||||
definition: {
|
definition: {
|
||||||
steps: AutomationStep[]
|
steps: AutomationStep[]
|
||||||
|
@ -116,7 +112,6 @@ export interface Automation extends Document {
|
||||||
name: string
|
name: string
|
||||||
internal?: boolean
|
internal?: boolean
|
||||||
type?: string
|
type?: string
|
||||||
state?: AutomationState
|
|
||||||
}
|
}
|
||||||
|
|
||||||
interface BaseIOStructure {
|
interface BaseIOStructure {
|
||||||
|
|
|
@ -40,11 +40,6 @@ export type DatabasePutOpts = {
|
||||||
force?: boolean
|
force?: boolean
|
||||||
}
|
}
|
||||||
|
|
||||||
export type DocExistsResponse = {
|
|
||||||
_rev?: string
|
|
||||||
exists: boolean
|
|
||||||
}
|
|
||||||
|
|
||||||
export type DatabaseCreateIndexOpts = {
|
export type DatabaseCreateIndexOpts = {
|
||||||
index: {
|
index: {
|
||||||
fields: string[]
|
fields: string[]
|
||||||
|
@ -95,7 +90,6 @@ export interface Database {
|
||||||
exists(): Promise<boolean>
|
exists(): Promise<boolean>
|
||||||
checkSetup(): Promise<Nano.DocumentScope<any>>
|
checkSetup(): Promise<Nano.DocumentScope<any>>
|
||||||
get<T>(id?: string): Promise<T>
|
get<T>(id?: string): Promise<T>
|
||||||
docExists(id: string): Promise<DocExistsResponse>
|
|
||||||
remove(
|
remove(
|
||||||
id: string | Document,
|
id: string | Document,
|
||||||
rev?: string
|
rev?: string
|
||||||
|
|
Loading…
Reference in a new issue