chore: remove redundant update flow status mutation
This commit is contained in:
@@ -7,7 +7,6 @@ import generateAuthUrl from './mutations/generate-auth-url.js';
|
|||||||
import createConnection from './mutations/create-connection.js';
|
import createConnection from './mutations/create-connection.js';
|
||||||
import resetConnection from './mutations/reset-connection.js';
|
import resetConnection from './mutations/reset-connection.js';
|
||||||
import updateConnection from './mutations/update-connection.js';
|
import updateConnection from './mutations/update-connection.js';
|
||||||
import updateFlowStatus from './mutations/update-flow-status.js';
|
|
||||||
|
|
||||||
const mutationResolvers = {
|
const mutationResolvers = {
|
||||||
createConnection,
|
createConnection,
|
||||||
@@ -16,7 +15,6 @@ const mutationResolvers = {
|
|||||||
resetConnection,
|
resetConnection,
|
||||||
updateConnection,
|
updateConnection,
|
||||||
updateCurrentUser,
|
updateCurrentUser,
|
||||||
updateFlowStatus,
|
|
||||||
updateUser,
|
updateUser,
|
||||||
verifyConnection,
|
verifyConnection,
|
||||||
};
|
};
|
||||||
|
@@ -1,91 +0,0 @@
|
|||||||
import Flow from '../../models/flow.js';
|
|
||||||
import flowQueue from '../../queues/flow.js';
|
|
||||||
import {
|
|
||||||
REMOVE_AFTER_30_DAYS_OR_150_JOBS,
|
|
||||||
REMOVE_AFTER_7_DAYS_OR_50_JOBS,
|
|
||||||
} from '../../helpers/remove-job-configuration.js';
|
|
||||||
import globalVariable from '../../helpers/global-variable.js';
|
|
||||||
|
|
||||||
const JOB_NAME = 'flow';
|
|
||||||
const EVERY_15_MINUTES_CRON = '*/15 * * * *';
|
|
||||||
|
|
||||||
const updateFlowStatus = async (_parent, params, context) => {
|
|
||||||
const conditions = context.currentUser.can('publish', 'Flow');
|
|
||||||
const isCreator = conditions.isCreator;
|
|
||||||
const allFlows = Flow.query();
|
|
||||||
const userFlows = context.currentUser.$relatedQuery('flows');
|
|
||||||
const baseQuery = isCreator ? userFlows : allFlows;
|
|
||||||
|
|
||||||
let flow = await baseQuery
|
|
||||||
.clone()
|
|
||||||
.findOne({
|
|
||||||
id: params.input.id,
|
|
||||||
})
|
|
||||||
.throwIfNotFound();
|
|
||||||
|
|
||||||
const newActiveValue = params.input.active;
|
|
||||||
|
|
||||||
if (flow.active === newActiveValue) {
|
|
||||||
return flow;
|
|
||||||
}
|
|
||||||
|
|
||||||
const triggerStep = await flow.getTriggerStep();
|
|
||||||
|
|
||||||
if (triggerStep.status === 'incomplete') {
|
|
||||||
throw flow.IncompleteStepsError;
|
|
||||||
}
|
|
||||||
|
|
||||||
const trigger = await triggerStep.getTriggerCommand();
|
|
||||||
const interval = trigger.getInterval?.(triggerStep.parameters);
|
|
||||||
const repeatOptions = {
|
|
||||||
pattern: interval || EVERY_15_MINUTES_CRON,
|
|
||||||
};
|
|
||||||
|
|
||||||
if (trigger.type === 'webhook') {
|
|
||||||
const $ = await globalVariable({
|
|
||||||
flow,
|
|
||||||
connection: await triggerStep.$relatedQuery('connection'),
|
|
||||||
app: await triggerStep.getApp(),
|
|
||||||
step: triggerStep,
|
|
||||||
testRun: false,
|
|
||||||
});
|
|
||||||
|
|
||||||
if (newActiveValue && trigger.registerHook) {
|
|
||||||
await trigger.registerHook($);
|
|
||||||
} else if (!newActiveValue && trigger.unregisterHook) {
|
|
||||||
await trigger.unregisterHook($);
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
if (newActiveValue) {
|
|
||||||
flow = await flow.$query().patchAndFetch({
|
|
||||||
publishedAt: new Date().toISOString(),
|
|
||||||
});
|
|
||||||
|
|
||||||
const jobName = `${JOB_NAME}-${flow.id}`;
|
|
||||||
|
|
||||||
await flowQueue.add(
|
|
||||||
jobName,
|
|
||||||
{ flowId: flow.id },
|
|
||||||
{
|
|
||||||
repeat: repeatOptions,
|
|
||||||
jobId: flow.id,
|
|
||||||
removeOnComplete: REMOVE_AFTER_7_DAYS_OR_50_JOBS,
|
|
||||||
removeOnFail: REMOVE_AFTER_30_DAYS_OR_150_JOBS,
|
|
||||||
}
|
|
||||||
);
|
|
||||||
} else {
|
|
||||||
const repeatableJobs = await flowQueue.getRepeatableJobs();
|
|
||||||
const job = repeatableJobs.find((job) => job.id === flow.id);
|
|
||||||
|
|
||||||
await flowQueue.removeRepeatableByKey(job.key);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
flow = await flow.$query().withGraphFetched('steps').patchAndFetch({
|
|
||||||
active: newActiveValue,
|
|
||||||
});
|
|
||||||
|
|
||||||
return flow;
|
|
||||||
};
|
|
||||||
|
|
||||||
export default updateFlowStatus;
|
|
@@ -8,7 +8,6 @@ type Mutation {
|
|||||||
resetConnection(input: ResetConnectionInput): Connection
|
resetConnection(input: ResetConnectionInput): Connection
|
||||||
updateConnection(input: UpdateConnectionInput): Connection
|
updateConnection(input: UpdateConnectionInput): Connection
|
||||||
updateCurrentUser(input: UpdateCurrentUserInput): User
|
updateCurrentUser(input: UpdateCurrentUserInput): User
|
||||||
updateFlowStatus(input: UpdateFlowStatusInput): Flow
|
|
||||||
updateUser(input: UpdateUserInput): User
|
updateUser(input: UpdateUserInput): User
|
||||||
verifyConnection(input: VerifyConnectionInput): Connection
|
verifyConnection(input: VerifyConnectionInput): Connection
|
||||||
}
|
}
|
||||||
@@ -234,11 +233,6 @@ input VerifyConnectionInput {
|
|||||||
id: String!
|
id: String!
|
||||||
}
|
}
|
||||||
|
|
||||||
input UpdateFlowStatusInput {
|
|
||||||
id: String!
|
|
||||||
active: Boolean!
|
|
||||||
}
|
|
||||||
|
|
||||||
input ExecuteFlowInput {
|
input ExecuteFlowInput {
|
||||||
stepId: String!
|
stepId: String!
|
||||||
}
|
}
|
||||||
|
@@ -1,9 +0,0 @@
|
|||||||
import { gql } from '@apollo/client';
|
|
||||||
export const UPDATE_FLOW_STATUS = gql`
|
|
||||||
mutation UpdateFlowStatus($input: UpdateFlowStatusInput) {
|
|
||||||
updateFlowStatus(input: $input) {
|
|
||||||
id
|
|
||||||
active
|
|
||||||
}
|
|
||||||
}
|
|
||||||
`;
|
|
Reference in New Issue
Block a user