groot/automatisch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(queue): auto clean up complete and fail jobs

Browse Source
This commit is contained in:
Ali BARIN
2022-11-27 19:25:10 +01:00
parent 4240849a2a
commit 2e391cc651
5 changed files with 36 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
packages/backend/src/graphql/mutations/update-flow-status.ts
View File

@@ -1,5 +1,6 @@
import Context from '../../types/express/context';
import flowQueue from '../../queues/flow';
import { REMOVE_AFTER_30_DAYS_OR_150_JOBS, REMOVE_AFTER_7_DAYS_OR_50_JOBS } from '../../helpers/remove-job-configuration';
type Params = {
input: {
@@ -51,6 +52,8 @@ const updateFlowStatus = async (
{
repeat: repeatOptions,
jobId: flow.id,
removeOnComplete: REMOVE_AFTER_7_DAYS_OR_50_JOBS,
removeOnFail: REMOVE_AFTER_30_DAYS_OR_150_JOBS
}
);
} else {

10
packages/backend/src/helpers/remove-job-configuration.ts Normal file
View File

@@ -0,0 +1,10 @@
export const REMOVE_AFTER_30_DAYS_OR_150_JOBS = {
age: 30 * 24 * 3600,
count: 150,
};
export const REMOVE_AFTER_7_DAYS_OR_50_JOBS = {
age: 7 * 24 * 3600,
count: 50,
};

10
packages/backend/src/workers/action.ts
View File

@@ -4,6 +4,7 @@ import logger from '../helpers/logger';
import Step from '../models/step';
import actionQueue from '../queues/action';
import { processAction } from '../services/action';
import { REMOVE_AFTER_30_DAYS_OR_150_JOBS, REMOVE_AFTER_7_DAYS_OR_50_JOBS } from '../helpers/remove-job-configuration';
type JobData = {
flowId: string;
@@ -31,7 +32,12 @@ export const worker = new Worker(
stepId: nextStep.id,
};
await actionQueue.add(jobName, jobPayload);
const jobOptions = {
removeOnComplete: REMOVE_AFTER_7_DAYS_OR_50_JOBS,
removeOnFail: REMOVE_AFTER_30_DAYS_OR_150_JOBS,
}
await actionQueue.add(jobName, jobPayload, jobOptions);
},
{ connection: redisConfig }
);
@@ -42,7 +48,7 @@ worker.on('completed', (job) => {
worker.on('failed', (job, err) => {
logger.info(
`JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has failed22 to start with ${err.message}`
`JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has failed to start with ${err.message}`
);
});

10
packages/backend/src/workers/flow.ts
View File

@@ -4,6 +4,7 @@ import logger from '../helpers/logger';
import triggerQueue from '../queues/trigger';
import { processFlow } from '../services/flow';
import Flow from '../models/flow';
import { REMOVE_AFTER_30_DAYS_OR_150_JOBS, REMOVE_AFTER_7_DAYS_OR_50_JOBS } from '../helpers/remove-job-configuration';
export const worker = new Worker(
'flow',
@@ -17,6 +18,11 @@ export const worker = new Worker(
const reversedData = data.reverse();
const jobOptions = {
removeOnComplete: REMOVE_AFTER_7_DAYS_OR_50_JOBS,
removeOnFail: REMOVE_AFTER_30_DAYS_OR_150_JOBS,
}
for (const triggerItem of reversedData) {
const jobName = `${triggerStep.id}-${triggerItem.meta.internalId}`;
@@ -26,7 +32,7 @@ export const worker = new Worker(
triggerItem,
};
await triggerQueue.add(jobName, jobPayload);
await triggerQueue.add(jobName, jobPayload, jobOptions);
}
if (error) {
@@ -38,7 +44,7 @@ export const worker = new Worker(
error,
};
await triggerQueue.add(jobName, jobPayload);
await triggerQueue.add(jobName, jobPayload, jobOptions);
}
},
{ connection: redisConfig }

8
packages/backend/src/workers/trigger.ts
View File

@@ -5,6 +5,7 @@ import { IJSONObject, ITriggerItem } from '@automatisch/types';
import actionQueue from '../queues/action';
import Step from '../models/step';
import { processTrigger } from '../services/trigger';
import { REMOVE_AFTER_30_DAYS_OR_150_JOBS, REMOVE_AFTER_7_DAYS_OR_50_JOBS } from '../helpers/remove-job-configuration';
type JobData = {
flowId: string;
@@ -32,7 +33,12 @@ export const worker = new Worker(
stepId: nextStep.id,
};
await actionQueue.add(jobName, jobPayload);
const jobOptions = {
removeOnComplete: REMOVE_AFTER_7_DAYS_OR_50_JOBS,
removeOnFail: REMOVE_AFTER_30_DAYS_OR_150_JOBS,
}
await actionQueue.add(jobName, jobPayload, jobOptions);
},
{ connection: redisConfig }
);