Compare commits

..

1 Commits

Author SHA1 Message Date
Ali BARIN
e7df19ae17 refactor(queues): abstract queue management 2024-12-03 11:22:03 +00:00
32 changed files with 329 additions and 219 deletions

View File

@@ -13,7 +13,7 @@ if (appConfig.redisSentinelHost) {
{ {
host: appConfig.redisSentinelHost, host: appConfig.redisSentinelHost,
port: appConfig.redisSentinelPort, port: appConfig.redisSentinelPort,
} },
]; ];
redisConfig.sentinelUsername = appConfig.redisSentinelUsername; redisConfig.sentinelUsername = appConfig.redisSentinelUsername;

View File

@@ -87,14 +87,14 @@ describe('GET /api/v1/apps/:appKey/connections', () => {
it('should return not found response for invalid connection UUID', async () => { it('should return not found response for invalid connection UUID', async () => {
await createPermission({ await createPermission({
action: 'read', action: 'update',
subject: 'Connection', subject: 'Connection',
roleId: currentUserRole.id, roleId: currentUserRole.id,
conditions: ['isCreator'], conditions: ['isCreator'],
}); });
await request(app) await request(app)
.get('/api/v1/apps/invalid-connection-id/connections') .get('/api/v1/connections/invalid-connection-id/connections')
.set('Authorization', token) .set('Authorization', token)
.expect(404); .expect(404);
}); });

View File

@@ -193,7 +193,7 @@ describe('POST /api/v1/steps/:stepId/dynamic-data', () => {
const notExistingStepUUID = Crypto.randomUUID(); const notExistingStepUUID = Crypto.randomUUID();
await request(app) await request(app)
.post(`/api/v1/steps/${notExistingStepUUID}/dynamic-data`) .get(`/api/v1/steps/${notExistingStepUUID}/dynamic-data`)
.set('Authorization', token) .set('Authorization', token)
.expect(404); .expect(404);
}); });
@@ -216,7 +216,7 @@ describe('POST /api/v1/steps/:stepId/dynamic-data', () => {
const step = await createStep({ appKey: null }); const step = await createStep({ appKey: null });
await request(app) await request(app)
.post(`/api/v1/steps/${step.id}/dynamic-data`) .get(`/api/v1/steps/${step.id}/dynamic-data`)
.set('Authorization', token) .set('Authorization', token)
.expect(404); .expect(404);
}); });

View File

@@ -118,7 +118,7 @@ describe('POST /api/v1/steps/:stepId/dynamic-fields', () => {
const notExistingStepUUID = Crypto.randomUUID(); const notExistingStepUUID = Crypto.randomUUID();
await request(app) await request(app)
.post(`/api/v1/steps/${notExistingStepUUID}/dynamic-fields`) .get(`/api/v1/steps/${notExistingStepUUID}/dynamic-fields`)
.set('Authorization', token) .set('Authorization', token)
.expect(404); .expect(404);
}); });
@@ -138,11 +138,10 @@ describe('POST /api/v1/steps/:stepId/dynamic-fields', () => {
conditions: [], conditions: [],
}); });
const step = await createStep(); const step = await createStep({ appKey: null });
await step.$query().patch({ appKey: null });
await request(app) await request(app)
.post(`/api/v1/steps/${step.id}/dynamic-fields`) .get(`/api/v1/steps/${step.id}/dynamic-fields`)
.set('Authorization', token) .set('Authorization', token)
.expect(404); .expect(404);
}); });

View File

@@ -12,7 +12,7 @@ import appConfig from '../config/app.js';
const serverAdapter = new ExpressAdapter(); const serverAdapter = new ExpressAdapter();
const queues = [ const queues = [
new BullMQAdapter(flowQueue), new BullMQAdapter(flowQueue.queue),
new BullMQAdapter(triggerQueue), new BullMQAdapter(triggerQueue),
new BullMQAdapter(actionQueue), new BullMQAdapter(actionQueue),
new BullMQAdapter(emailQueue), new BullMQAdapter(emailQueue),

View File

@@ -386,10 +386,7 @@ class Flow extends Base {
} }
); );
} else { } else {
const repeatableJobs = await flowQueue.getRepeatableJobs(); await flowQueue.removeRepeatableJobById(this.id);
const job = repeatableJobs.find((job) => job.id === this.id);
await flowQueue.removeRepeatableByKey(job.key);
} }
} }

View File

@@ -305,14 +305,8 @@ class User extends Base {
active: true, active: true,
}); });
const repeatableJobs = await flowQueue.getRepeatableJobs();
for (const flow of flows) { for (const flow of flows) {
const job = repeatableJobs.find((job) => job.id === flow.id); await flowQueue.removeRepeatableJobById(flow.id);
if (job) {
await flowQueue.removeRepeatableByKey(job.key);
}
} }
const executionIds = ( const executionIds = (

View File

@@ -996,9 +996,21 @@ describe('User model', () => {
const user = await createUser(); const user = await createUser();
const presentDate = DateTime.fromObject(
{ year: 2024, month: 11, day: 17, hour: 11, minute: 30 },
{ zone: 'UTC+0' }
);
vi.setSystemTime(presentDate);
await user.startTrialPeriod(); await user.startTrialPeriod();
vi.setSystemTime(DateTime.now().plus({ month: 1 })); const futureDate = DateTime.fromObject(
{ year: 2025, month: 1, day: 1 },
{ zone: 'UTC+0' }
);
vi.setSystemTime(futureDate);
const refetchedUser = await user.$query(); const refetchedUser = await user.$query();
@@ -1106,9 +1118,7 @@ describe('User model', () => {
const user = await createUser(); const user = await createUser();
await expect(() => user.getPlanAndUsage()).rejects.toThrow( expect(() => user.getPlanAndUsage()).rejects.toThrow('NotFoundError');
'NotFoundError'
);
}); });
}); });
@@ -1179,7 +1189,7 @@ describe('User model', () => {
}); });
it('should throw not found error when user role does not exist', async () => { it('should throw not found error when user role does not exist', async () => {
await expect(() => expect(() =>
User.registerUser({ User.registerUser({
fullName: 'Sample user', fullName: 'Sample user',
email: 'user@automatisch.io', email: 'user@automatisch.io',
@@ -1247,8 +1257,6 @@ describe('User model', () => {
describe('createUsageData', () => { describe('createUsageData', () => {
it('should create usage data if Automatisch is a cloud installation', async () => { it('should create usage data if Automatisch is a cloud installation', async () => {
vi.useFakeTimers();
vi.spyOn(appConfig, 'isCloud', 'get').mockReturnValue(true); vi.spyOn(appConfig, 'isCloud', 'get').mockReturnValue(true);
const user = await createUser({ const user = await createUser({
@@ -1256,14 +1264,10 @@ describe('User model', () => {
email: 'user@automatisch.io', email: 'user@automatisch.io',
}); });
vi.setSystemTime(DateTime.now().plus({ month: 1 }));
const usageData = await user.createUsageData(); const usageData = await user.createUsageData();
const currentUsageData = await user.$relatedQuery('currentUsageData'); const currentUsageData = await user.$relatedQuery('currentUsageData');
expect(usageData).toStrictEqual(currentUsageData); expect(usageData).toStrictEqual(currentUsageData);
vi.useRealTimers();
}); });
it('should not create usage data if Automatisch is not a cloud installation', async () => { it('should not create usage data if Automatisch is not a cloud installation', async () => {

View File

@@ -11,6 +11,10 @@ const redisConnection = {
const actionQueue = new Queue('action', redisConnection); const actionQueue = new Queue('action', redisConnection);
process.on('SIGTERM', async () => {
await actionQueue.close();
});
actionQueue.on('error', (error) => { actionQueue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) { if (error.code === CONNECTION_REFUSED) {
logger.error( logger.error(

View File

@@ -0,0 +1,30 @@
/* eslint-disable no-unused-vars */
class BaseQueue {
constructor(name) {
if (new.target === BaseQueue) {
throw new Error('Cannot instantiate abstract class BaseQueue directly.');
}
this.name = name;
}
// Abstract methods to be implemented by subclasses
async add(jobName, data, options) {
throw new Error('Method "add" must be implemented.');
}
async remove(jobId) {
throw new Error('Method "remove" must be implemented.');
}
async getRepeatableJobs() {
throw new Error('Method "getRepeatableJobs" must be implemented.');
}
async removeRepeatableJobByKey(jobKey) {
throw new Error('Method "removeRepeatableJobByKey" must be implemented.');
}
}
export default BaseQueue;

View File

@@ -0,0 +1,154 @@
import { Queue, Worker } from 'bullmq';
import redisConfig from '../config/redis.js';
import logger from '../helpers/logger.js';
import BaseQueue from './base.js';
const CONNECTION_REFUSED = 'ECONNREFUSED';
class BullMQQueue extends BaseQueue {
static queueOptions = {
connection: redisConfig,
};
constructor(name) {
super(name);
this.queue = new Queue(name, this.constructor.queueOptions);
this.workers = [];
this.setupErrorHandlers();
this.setupGracefulShutdown();
}
async add(jobName, data, options = {}) {
try {
const job = await this.queue.add(jobName, data, options);
return job;
} catch (error) {
logger.error(`Failed to add job to queue "${this.name}":`, error);
throw error;
}
}
async remove(jobId) {
try {
const job = await this.getJob(jobId);
if (job) {
await job.remove();
return true;
}
return false;
} catch (error) {
logger.error(`Failed to remove job from queue "${this.name}":`, error);
throw error;
}
}
async getJob(jobId) {
return await this.queue.getJob(jobId);
}
async getRepeatableJobById(jobId) {
const repeatableJobs = await this.getRepeatableJobs();
const job = repeatableJobs.find((job) => job.id === jobId);
return job;
}
async getRepeatableJobs() {
try {
return await this.queue.getRepeatableJobs();
} catch (error) {
logger.error(
`Failed to get repeatable jobs from queue "${this.name}":`,
error
);
throw error;
}
}
async removeRepeatableJobByKey(jobKey) {
try {
await this.queue.removeRepeatableByKey(jobKey);
return true;
} catch (error) {
logger.error(
`Failed to remove repeatable job from queue "${this.name}":`,
error
);
throw error;
}
}
async removeRepeatableJobById(jobId) {
const job = await this.getRepeatableJobById(jobId);
return await this.removeRepeatableJobByKey(job.key);
}
startWorker(processor, workerOptions = {}) {
const worker = new Worker(this.name, processor, {
...this.queueOptions,
...workerOptions,
});
worker.on('error', (error) => {
logger.error(`Worker error in queue "${this.name}":`, error);
});
this.workers.push(worker);
return worker;
}
setupErrorHandlers() {
this.queue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) {
logger.error(
'Make sure you have installed Redis and it is running.',
error
);
process.exit();
}
logger.error(`Queue error in "${this.name}":`, error);
});
}
setupGracefulShutdown() {
const shutdown = async () => {
logger.log(`Shutting down queue "${this.name}"...`);
try {
// Close all workers gracefully
for (const worker of this.workers) {
await worker.close();
}
await this.queue.close();
logger.log(`Queue "${this.name}" shut down successfully.`);
process.exit();
} catch (error) {
logger.error(`Error during shutdown of queue "${this.name}":`, error);
process.exit(1);
}
};
process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);
}
}
export default BullMQQueue;

View File

@@ -11,6 +11,10 @@ const redisConnection = {
const deleteUserQueue = new Queue('delete-user', redisConnection); const deleteUserQueue = new Queue('delete-user', redisConnection);
process.on('SIGTERM', async () => {
await deleteUserQueue.close();
});
deleteUserQueue.on('error', (error) => { deleteUserQueue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) { if (error.code === CONNECTION_REFUSED) {
logger.error( logger.error(

View File

@@ -11,6 +11,10 @@ const redisConnection = {
const emailQueue = new Queue('email', redisConnection); const emailQueue = new Queue('email', redisConnection);
process.on('SIGTERM', async () => {
await emailQueue.close();
});
emailQueue.on('error', (error) => { emailQueue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) { if (error.code === CONNECTION_REFUSED) {
logger.error( logger.error(

View File

@@ -1,27 +1,5 @@
import process from 'process'; import BullMQQueue from './bullmq.js';
import { Queue } from 'bullmq';
import redisConfig from '../config/redis.js';
import logger from '../helpers/logger.js';
const CONNECTION_REFUSED = 'ECONNREFUSED'; const flowQueue = new BullMQQueue('flow');
const redisConnection = {
connection: redisConfig,
};
const flowQueue = new Queue('flow', redisConnection);
flowQueue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) {
logger.error(
'Make sure you have installed Redis and it is running.',
error
);
process.exit();
}
logger.error('Error happened in flow queue!', error);
});
export default flowQueue; export default flowQueue;

View File

@@ -1,21 +0,0 @@
import appConfig from '../config/app.js';
import actionQueue from './action.js';
import emailQueue from './email.js';
import flowQueue from './flow.js';
import triggerQueue from './trigger.js';
import deleteUserQueue from './delete-user.ee.js';
import removeCancelledSubscriptionsQueue from './remove-cancelled-subscriptions.ee.js';
const queues = [
actionQueue,
emailQueue,
flowQueue,
triggerQueue,
deleteUserQueue,
];
if (appConfig.isCloud) {
queues.push(removeCancelledSubscriptionsQueue);
}
export default queues;

View File

@@ -14,6 +14,10 @@ const removeCancelledSubscriptionsQueue = new Queue(
redisConnection redisConnection
); );
process.on('SIGTERM', async () => {
await removeCancelledSubscriptionsQueue.close();
});
removeCancelledSubscriptionsQueue.on('error', (error) => { removeCancelledSubscriptionsQueue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) { if (error.code === CONNECTION_REFUSED) {
logger.error( logger.error(

View File

@@ -11,6 +11,10 @@ const redisConnection = {
const triggerQueue = new Queue('trigger', redisConnection); const triggerQueue = new Queue('trigger', redisConnection);
process.on('SIGTERM', async () => {
await triggerQueue.close();
});
triggerQueue.on('error', (error) => { triggerQueue.on('error', (error) => {
if (error.code === CONNECTION_REFUSED) { if (error.code === CONNECTION_REFUSED) {
logger.error( logger.error(

View File

@@ -1,22 +1,20 @@
import * as Sentry from './helpers/sentry.ee.js'; import * as Sentry from './helpers/sentry.ee.js';
import process from 'node:process'; import appConfig from './config/app.js';
Sentry.init(); Sentry.init();
import './config/orm.js'; import './config/orm.js';
import './helpers/check-worker-readiness.js'; import './helpers/check-worker-readiness.js';
import queues from './queues/index.js'; import './workers/flow.js';
import workers from './workers/index.js'; import './workers/trigger.js';
import './workers/action.js';
import './workers/email.js';
import './workers/delete-user.ee.js';
process.on('SIGTERM', async () => { if (appConfig.isCloud) {
for (const queue of queues) { import('./workers/remove-cancelled-subscriptions.ee.js');
await queue.close(); import('./queues/remove-cancelled-subscriptions.ee.js');
} }
for (const worker of workers) {
await worker.close();
}
});
import telemetry from './helpers/telemetry/index.js'; import telemetry from './helpers/telemetry/index.js';

View File

@@ -1,4 +1,5 @@
import { Worker } from 'bullmq'; import { Worker } from 'bullmq';
import process from 'node:process';
import * as Sentry from '../helpers/sentry.ee.js'; import * as Sentry from '../helpers/sentry.ee.js';
import redisConfig from '../config/redis.js'; import redisConfig from '../config/redis.js';
@@ -14,7 +15,7 @@ import delayAsMilliseconds from '../helpers/delay-as-milliseconds.js';
const DEFAULT_DELAY_DURATION = 0; const DEFAULT_DELAY_DURATION = 0;
const actionWorker = new Worker( export const worker = new Worker(
'action', 'action',
async (job) => { async (job) => {
const { stepId, flowId, executionId, computedParameters, executionStep } = const { stepId, flowId, executionId, computedParameters, executionStep } =
@@ -54,11 +55,11 @@ const actionWorker = new Worker(
{ connection: redisConfig } { connection: redisConfig }
); );
actionWorker.on('completed', (job) => { worker.on('completed', (job) => {
logger.info(`JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has started!`); logger.info(`JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has started!`);
}); });
actionWorker.on('failed', (job, err) => { worker.on('failed', (job, err) => {
const errorMessage = ` const errorMessage = `
JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has failed to start with ${err.message} JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has failed to start with ${err.message}
\n ${err.stack} \n ${err.stack}
@@ -73,4 +74,6 @@ actionWorker.on('failed', (job, err) => {
}); });
}); });
export default actionWorker; process.on('SIGTERM', async () => {
await worker.close();
});

View File

@@ -1,4 +1,5 @@
import { Worker } from 'bullmq'; import { Worker } from 'bullmq';
import process from 'node:process';
import * as Sentry from '../helpers/sentry.ee.js'; import * as Sentry from '../helpers/sentry.ee.js';
import redisConfig from '../config/redis.js'; import redisConfig from '../config/redis.js';
@@ -7,7 +8,7 @@ import appConfig from '../config/app.js';
import User from '../models/user.js'; import User from '../models/user.js';
import ExecutionStep from '../models/execution-step.js'; import ExecutionStep from '../models/execution-step.js';
const deleteUserWorker = new Worker( export const worker = new Worker(
'delete-user', 'delete-user',
async (job) => { async (job) => {
const { id } = job.data; const { id } = job.data;
@@ -45,13 +46,13 @@ const deleteUserWorker = new Worker(
{ connection: redisConfig } { connection: redisConfig }
); );
deleteUserWorker.on('completed', (job) => { worker.on('completed', (job) => {
logger.info( logger.info(
`JOB ID: ${job.id} - The user with the ID of '${job.data.id}' has been deleted!` `JOB ID: ${job.id} - The user with the ID of '${job.data.id}' has been deleted!`
); );
}); });
deleteUserWorker.on('failed', (job, err) => { worker.on('failed', (job, err) => {
const errorMessage = ` const errorMessage = `
JOB ID: ${job.id} - The user with the ID of '${job.data.id}' has failed to be deleted! ${err.message} JOB ID: ${job.id} - The user with the ID of '${job.data.id}' has failed to be deleted! ${err.message}
\n ${err.stack} \n ${err.stack}
@@ -66,4 +67,6 @@ deleteUserWorker.on('failed', (job, err) => {
}); });
}); });
export default deleteUserWorker; process.on('SIGTERM', async () => {
await worker.close();
});

View File

@@ -1,4 +1,5 @@
import { Worker } from 'bullmq'; import { Worker } from 'bullmq';
import process from 'node:process';
import * as Sentry from '../helpers/sentry.ee.js'; import * as Sentry from '../helpers/sentry.ee.js';
import redisConfig from '../config/redis.js'; import redisConfig from '../config/redis.js';
@@ -15,7 +16,7 @@ const isAutomatischEmail = (email) => {
return email.endsWith('@automatisch.io'); return email.endsWith('@automatisch.io');
}; };
const emailWorker = new Worker( export const worker = new Worker(
'email', 'email',
async (job) => { async (job) => {
const { email, subject, template, params } = job.data; const { email, subject, template, params } = job.data;
@@ -38,13 +39,13 @@ const emailWorker = new Worker(
{ connection: redisConfig } { connection: redisConfig }
); );
emailWorker.on('completed', (job) => { worker.on('completed', (job) => {
logger.info( logger.info(
`JOB ID: ${job.id} - ${job.data.subject} email sent to ${job.data.email}!` `JOB ID: ${job.id} - ${job.data.subject} email sent to ${job.data.email}!`
); );
}); });
emailWorker.on('failed', (job, err) => { worker.on('failed', (job, err) => {
const errorMessage = ` const errorMessage = `
JOB ID: ${job.id} - ${job.data.subject} email to ${job.data.email} has failed to send with ${err.message} JOB ID: ${job.id} - ${job.data.subject} email to ${job.data.email} has failed to send with ${err.message}
\n ${err.stack} \n ${err.stack}
@@ -59,4 +60,6 @@ emailWorker.on('failed', (job, err) => {
}); });
}); });
export default emailWorker; process.on('SIGTERM', async () => {
await worker.close();
});

View File

@@ -1,4 +1,5 @@
import { Worker } from 'bullmq'; import { Worker } from 'bullmq';
import process from 'node:process';
import * as Sentry from '../helpers/sentry.ee.js'; import * as Sentry from '../helpers/sentry.ee.js';
import redisConfig from '../config/redis.js'; import redisConfig from '../config/redis.js';
@@ -12,7 +13,7 @@ import {
REMOVE_AFTER_7_DAYS_OR_50_JOBS, REMOVE_AFTER_7_DAYS_OR_50_JOBS,
} from '../helpers/remove-job-configuration.js'; } from '../helpers/remove-job-configuration.js';
const flowWorker = new Worker( export const worker = new Worker(
'flow', 'flow',
async (job) => { async (job) => {
const { flowId } = job.data; const { flowId } = job.data;
@@ -63,11 +64,11 @@ const flowWorker = new Worker(
{ connection: redisConfig } { connection: redisConfig }
); );
flowWorker.on('completed', (job) => { worker.on('completed', (job) => {
logger.info(`JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has started!`); logger.info(`JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has started!`);
}); });
flowWorker.on('failed', async (job, err) => { worker.on('failed', async (job, err) => {
const errorMessage = ` const errorMessage = `
JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has failed to start with ${err.message} JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has failed to start with ${err.message}
\n ${err.stack} \n ${err.stack}
@@ -78,7 +79,7 @@ flowWorker.on('failed', async (job, err) => {
const flow = await Flow.query().findById(job.data.flowId); const flow = await Flow.query().findById(job.data.flowId);
if (!flow) { if (!flow) {
await flowQueue.removeRepeatableByKey(job.repeatJobKey); await flowQueue.removeRepeatableJobByKey(job.repeatJobKey);
const flowNotFoundErrorMessage = ` const flowNotFoundErrorMessage = `
JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has been deleted from Redis because flow was not found! JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has been deleted from Redis because flow was not found!
@@ -94,4 +95,6 @@ flowWorker.on('failed', async (job, err) => {
}); });
}); });
export default flowWorker; process.on('SIGTERM', async () => {
await worker.close();
});

View File

@@ -1,21 +0,0 @@
import appConfig from '../config/app.js';
import actionWorker from './action.js';
import emailWorker from './email.js';
import flowWorker from './flow.js';
import triggerWorker from './trigger.js';
import deleteUserWorker from './delete-user.ee.js';
import removeCancelledSubscriptionsWorker from './remove-cancelled-subscriptions.ee.js';
const workers = [
actionWorker,
emailWorker,
flowWorker,
triggerWorker,
deleteUserWorker,
];
if (appConfig.isCloud) {
workers.push(removeCancelledSubscriptionsWorker);
}
export default workers;

View File

@@ -1,11 +1,12 @@
import { Worker } from 'bullmq'; import { Worker } from 'bullmq';
import process from 'node:process';
import { DateTime } from 'luxon'; import { DateTime } from 'luxon';
import * as Sentry from '../helpers/sentry.ee.js'; import * as Sentry from '../helpers/sentry.ee.js';
import redisConfig from '../config/redis.js'; import redisConfig from '../config/redis.js';
import logger from '../helpers/logger.js'; import logger from '../helpers/logger.js';
import Subscription from '../models/subscription.ee.js'; import Subscription from '../models/subscription.ee.js';
const removeCancelledSubscriptionsWorker = new Worker( export const worker = new Worker(
'remove-cancelled-subscriptions', 'remove-cancelled-subscriptions',
async () => { async () => {
await Subscription.query() await Subscription.query()
@@ -22,13 +23,13 @@ const removeCancelledSubscriptionsWorker = new Worker(
{ connection: redisConfig } { connection: redisConfig }
); );
removeCancelledSubscriptionsWorker.on('completed', (job) => { worker.on('completed', (job) => {
logger.info( logger.info(
`JOB ID: ${job.id} - The cancelled subscriptions have been removed!` `JOB ID: ${job.id} - The cancelled subscriptions have been removed!`
); );
}); });
removeCancelledSubscriptionsWorker.on('failed', (job, err) => { worker.on('failed', (job, err) => {
const errorMessage = ` const errorMessage = `
JOB ID: ${job.id} - ERROR: The cancelled subscriptions can not be removed! ${err.message} JOB ID: ${job.id} - ERROR: The cancelled subscriptions can not be removed! ${err.message}
\n ${err.stack} \n ${err.stack}
@@ -41,4 +42,6 @@ removeCancelledSubscriptionsWorker.on('failed', (job, err) => {
}); });
}); });
export default removeCancelledSubscriptionsWorker; process.on('SIGTERM', async () => {
await worker.close();
});

View File

@@ -1,4 +1,5 @@
import { Worker } from 'bullmq'; import { Worker } from 'bullmq';
import process from 'node:process';
import * as Sentry from '../helpers/sentry.ee.js'; import * as Sentry from '../helpers/sentry.ee.js';
import redisConfig from '../config/redis.js'; import redisConfig from '../config/redis.js';
@@ -11,7 +12,7 @@ import {
REMOVE_AFTER_7_DAYS_OR_50_JOBS, REMOVE_AFTER_7_DAYS_OR_50_JOBS,
} from '../helpers/remove-job-configuration.js'; } from '../helpers/remove-job-configuration.js';
const triggerWorker = new Worker( export const worker = new Worker(
'trigger', 'trigger',
async (job) => { async (job) => {
const { flowId, executionId, stepId, executionStep } = await processTrigger( const { flowId, executionId, stepId, executionStep } = await processTrigger(
@@ -40,11 +41,11 @@ const triggerWorker = new Worker(
{ connection: redisConfig } { connection: redisConfig }
); );
triggerWorker.on('completed', (job) => { worker.on('completed', (job) => {
logger.info(`JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has started!`); logger.info(`JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has started!`);
}); });
triggerWorker.on('failed', (job, err) => { worker.on('failed', (job, err) => {
const errorMessage = ` const errorMessage = `
JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has failed to start with ${err.message} JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has failed to start with ${err.message}
\n ${err.stack} \n ${err.stack}
@@ -59,4 +60,6 @@ triggerWorker.on('failed', (job, err) => {
}); });
}); });
export default triggerWorker; process.on('SIGTERM', async () => {
await worker.close();
});

View File

@@ -9,7 +9,6 @@ export class DeleteRoleModal {
this.modal = page.getByTestId('delete-role-modal'); this.modal = page.getByTestId('delete-role-modal');
this.cancelButton = this.modal.getByTestId('confirmation-cancel-button'); this.cancelButton = this.modal.getByTestId('confirmation-cancel-button');
this.deleteButton = this.modal.getByTestId('confirmation-confirm-button'); this.deleteButton = this.modal.getByTestId('confirmation-confirm-button');
this.deleteAlert = this.modal.getByTestId('confirmation-dialog-error-alert');
} }
async close () { async close () {

View File

@@ -218,7 +218,12 @@ test.describe('Role management page', () => {
const row = await adminRolesPage.getRoleRowByName('Delete Role'); const row = await adminRolesPage.getRoleRowByName('Delete Role');
const modal = await adminRolesPage.clickDeleteRole(row); const modal = await adminRolesPage.clickDeleteRole(row);
await modal.deleteButton.click(); await modal.deleteButton.click();
await expect(modal.deleteAlert).toHaveCount(1); await adminRolesPage.snackbar.waitFor({
state: 'attached',
});
const snackbar = await adminRolesPage.getSnackbarData('snackbar-delete-role-error');
await expect(snackbar.variant).toBe('error');
await adminRolesPage.closeSnackbar();
await modal.close(); await modal.close();
} }
); );
@@ -313,6 +318,7 @@ test.describe('Role management page', () => {
const row = await adminUsersPage.findUserPageWithEmail( const row = await adminUsersPage.findUserPageWithEmail(
'user-delete-role-test@automatisch.io' 'user-delete-role-test@automatisch.io'
); );
// await test.waitForTimeout(10000);
const modal = await adminUsersPage.clickDeleteUser(row); const modal = await adminUsersPage.clickDeleteUser(row);
await modal.deleteButton.click(); await modal.deleteButton.click();
await adminUsersPage.snackbar.waitFor({ await adminUsersPage.snackbar.waitFor({
@@ -329,7 +335,15 @@ test.describe('Role management page', () => {
const row = await adminRolesPage.getRoleRowByName('Cannot Delete Role'); const row = await adminRolesPage.getRoleRowByName('Cannot Delete Role');
const modal = await adminRolesPage.clickDeleteRole(row); const modal = await adminRolesPage.clickDeleteRole(row);
await modal.deleteButton.click(); await modal.deleteButton.click();
await expect(modal.deleteAlert).toHaveCount(1); await adminRolesPage.snackbar.waitFor({
state: 'attached',
});
/*
* TODO: await snackbar - make assertions based on product
* decisions
const snackbar = await adminRolesPage.getSnackbarData();
await expect(snackbar.variant).toBe('...');
*/
await adminRolesPage.closeSnackbar(); await adminRolesPage.closeSnackbar();
}); });
}); });

View File

@@ -6,7 +6,6 @@ import DialogActions from '@mui/material/DialogActions';
import DialogContent from '@mui/material/DialogContent'; import DialogContent from '@mui/material/DialogContent';
import DialogContentText from '@mui/material/DialogContentText'; import DialogContentText from '@mui/material/DialogContentText';
import DialogTitle from '@mui/material/DialogTitle'; import DialogTitle from '@mui/material/DialogTitle';
import Alert from '@mui/material/Alert';
function ConfirmationDialog(props) { function ConfirmationDialog(props) {
const { const {
@@ -17,7 +16,6 @@ function ConfirmationDialog(props) {
cancelButtonChildren, cancelButtonChildren,
confirmButtonChildren, confirmButtonChildren,
open = true, open = true,
errorMessage,
} = props; } = props;
const dataTest = props['data-test']; const dataTest = props['data-test'];
return ( return (
@@ -46,11 +44,6 @@ function ConfirmationDialog(props) {
</Button> </Button>
)} )}
</DialogActions> </DialogActions>
{errorMessage && (
<Alert data-test="confirmation-dialog-error-alert" severity="error">
{errorMessage}
</Alert>
)}
</Dialog> </Dialog>
); );
} }
@@ -64,7 +57,6 @@ ConfirmationDialog.propTypes = {
confirmButtonChildren: PropTypes.node.isRequired, confirmButtonChildren: PropTypes.node.isRequired,
open: PropTypes.bool, open: PropTypes.bool,
'data-test': PropTypes.string, 'data-test': PropTypes.string,
errorMessage: PropTypes.string,
}; };
export default ConfirmationDialog; export default ConfirmationDialog;

View File

@@ -4,7 +4,6 @@ import IconButton from '@mui/material/IconButton';
import useEnqueueSnackbar from 'hooks/useEnqueueSnackbar'; import useEnqueueSnackbar from 'hooks/useEnqueueSnackbar';
import * as React from 'react'; import * as React from 'react';
import { getGeneralErrorMessage, getFieldErrorMessage } from 'helpers/errors';
import Can from 'components/Can'; import Can from 'components/Can';
import ConfirmationDialog from 'components/ConfirmationDialog'; import ConfirmationDialog from 'components/ConfirmationDialog';
import useFormatMessage from 'hooks/useFormatMessage'; import useFormatMessage from 'hooks/useFormatMessage';
@@ -16,21 +15,7 @@ function DeleteRoleButton(props) {
const formatMessage = useFormatMessage(); const formatMessage = useFormatMessage();
const enqueueSnackbar = useEnqueueSnackbar(); const enqueueSnackbar = useEnqueueSnackbar();
const { const { mutateAsync: deleteRole } = useAdminDeleteRole(roleId);
mutateAsync: deleteRole,
error: deleteRoleError,
reset: resetDeleteRole,
} = useAdminDeleteRole(roleId);
const roleErrorMessage = getFieldErrorMessage({
fieldName: 'role',
error: deleteRoleError,
});
const generalErrorMessage = getGeneralErrorMessage({
error: deleteRoleError,
fallbackMessage: formatMessage('deleteRoleButton.generalError'),
});
const handleConfirm = React.useCallback(async () => { const handleConfirm = React.useCallback(async () => {
try { try {
@@ -43,13 +28,23 @@ function DeleteRoleButton(props) {
'data-test': 'snackbar-delete-role-success', 'data-test': 'snackbar-delete-role-success',
}, },
}); });
} catch {} } catch (error) {
}, [deleteRole, enqueueSnackbar, formatMessage]); const errors = Object.values(
error.response.data.errors || [['Failed while deleting!']],
);
const handleClose = () => { for (const [error] of errors) {
setShowConfirmation(false); enqueueSnackbar(error, {
resetDeleteRole(); variant: 'error',
}; SnackbarProps: {
'data-test': 'snackbar-delete-role-error',
},
});
}
throw new Error('Failed while deleting!');
}
}, [deleteRole, enqueueSnackbar, formatMessage]);
return ( return (
<> <>
@@ -70,12 +65,11 @@ function DeleteRoleButton(props) {
open={showConfirmation} open={showConfirmation}
title={formatMessage('deleteRoleButton.title')} title={formatMessage('deleteRoleButton.title')}
description={formatMessage('deleteRoleButton.description')} description={formatMessage('deleteRoleButton.description')}
onClose={handleClose} onClose={() => setShowConfirmation(false)}
onConfirm={handleConfirm} onConfirm={handleConfirm}
cancelButtonChildren={formatMessage('deleteRoleButton.cancel')} cancelButtonChildren={formatMessage('deleteRoleButton.cancel')}
confirmButtonChildren={formatMessage('deleteRoleButton.confirm')} confirmButtonChildren={formatMessage('deleteRoleButton.confirm')}
data-test="delete-role-modal" data-test="delete-role-modal"
errorMessage={roleErrorMessage || generalErrorMessage}
/> />
</> </>
); );

View File

@@ -3,7 +3,6 @@ import DeleteIcon from '@mui/icons-material/Delete';
import IconButton from '@mui/material/IconButton'; import IconButton from '@mui/material/IconButton';
import { useQueryClient } from '@tanstack/react-query'; import { useQueryClient } from '@tanstack/react-query';
import { getGeneralErrorMessage } from 'helpers/errors';
import useEnqueueSnackbar from 'hooks/useEnqueueSnackbar'; import useEnqueueSnackbar from 'hooks/useEnqueueSnackbar';
import * as React from 'react'; import * as React from 'react';
import ConfirmationDialog from 'components/ConfirmationDialog'; import ConfirmationDialog from 'components/ConfirmationDialog';
@@ -13,21 +12,12 @@ import useAdminUserDelete from 'hooks/useAdminUserDelete';
function DeleteUserButton(props) { function DeleteUserButton(props) {
const { userId } = props; const { userId } = props;
const [showConfirmation, setShowConfirmation] = React.useState(false); const [showConfirmation, setShowConfirmation] = React.useState(false);
const { const { mutateAsync: deleteUser } = useAdminUserDelete(userId);
mutateAsync: deleteUser,
error: deleteUserError,
reset: resetDeleteUser,
} = useAdminUserDelete(userId);
const formatMessage = useFormatMessage(); const formatMessage = useFormatMessage();
const enqueueSnackbar = useEnqueueSnackbar(); const enqueueSnackbar = useEnqueueSnackbar();
const queryClient = useQueryClient(); const queryClient = useQueryClient();
const generalErrorMessage = getGeneralErrorMessage({
error: deleteUserError,
fallbackMessage: formatMessage('deleteUserButton.deleteError'),
});
const handleConfirm = React.useCallback(async () => { const handleConfirm = React.useCallback(async () => {
try { try {
await deleteUser(); await deleteUser();
@@ -39,14 +29,16 @@ function DeleteUserButton(props) {
'data-test': 'snackbar-delete-user-success', 'data-test': 'snackbar-delete-user-success',
}, },
}); });
} catch {} } catch (error) {
enqueueSnackbar(
error?.message || formatMessage('deleteUserButton.deleteError'),
{
variant: 'error',
},
);
}
}, [deleteUser]); }, [deleteUser]);
const handleClose = () => {
setShowConfirmation(false);
resetDeleteUser();
};
return ( return (
<> <>
<IconButton <IconButton
@@ -61,12 +53,11 @@ function DeleteUserButton(props) {
open={showConfirmation} open={showConfirmation}
title={formatMessage('deleteUserButton.title')} title={formatMessage('deleteUserButton.title')}
description={formatMessage('deleteUserButton.description')} description={formatMessage('deleteUserButton.description')}
onClose={handleClose} onClose={() => setShowConfirmation(false)}
onConfirm={handleConfirm} onConfirm={handleConfirm}
cancelButtonChildren={formatMessage('deleteUserButton.cancel')} cancelButtonChildren={formatMessage('deleteUserButton.cancel')}
confirmButtonChildren={formatMessage('deleteUserButton.confirm')} confirmButtonChildren={formatMessage('deleteUserButton.confirm')}
data-test="delete-user-modal" data-test="delete-user-modal"
errorMessage={generalErrorMessage}
/> />
</> </>
); );

View File

@@ -1,29 +0,0 @@
// Helpers to extract errors received from the API
export const getGeneralErrorMessage = ({ error, fallbackMessage }) => {
if (!error) {
return;
}
const errors = error?.response?.data?.errors;
const generalError = errors?.general;
if (generalError && Array.isArray(generalError)) {
return generalError.join(' ');
}
if (!errors) {
return error?.message || fallbackMessage;
}
};
export const getFieldErrorMessage = ({ fieldName, error }) => {
const errors = error?.response?.data?.errors;
const fieldErrors = errors?.[fieldName];
if (fieldErrors && Array.isArray(fieldErrors)) {
return fieldErrors.join(', ');
}
return '';
};

View File

@@ -245,7 +245,6 @@
"deleteRoleButton.cancel": "Cancel", "deleteRoleButton.cancel": "Cancel",
"deleteRoleButton.confirm": "Delete", "deleteRoleButton.confirm": "Delete",
"deleteRoleButton.successfullyDeleted": "The role has been deleted.", "deleteRoleButton.successfullyDeleted": "The role has been deleted.",
"deleteRoleButton.generalError": "Failed while deleting!",
"editRolePage.title": "Edit role", "editRolePage.title": "Edit role",
"createRolePage.title": "Create role", "createRolePage.title": "Create role",
"roleForm.name": "Name", "roleForm.name": "Name",