diff --git a/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts b/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts index 567abfe960..c914368a6d 100644 --- a/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts +++ b/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts @@ -286,6 +286,78 @@ export class CacheStorageService { return newValue; } + async hashGetValues(key: string): Promise { + if (!this.isRedisCache()) { + throw new Error('hashGetValues is only supported with Redis cache'); + } + + const redisClient = (this.cache as RedisCache).store.client; + + return redisClient.hVals(this.getKey(key)); + } + + async hashSet({ + key, + field, + value, + }: { + key: string; + field: string; + value: string; + }): Promise { + if (!this.isRedisCache()) { + throw new Error('hashSet is only supported with Redis cache'); + } + + const redisClient = (this.cache as RedisCache).store.client; + + return redisClient.hSet(this.getKey(key), field, value); + } + + async hashSetIfExists({ + key, + field, + value, + }: { + key: string; + field: string; + value: string; + }): Promise { + if (!this.isRedisCache()) { + throw new Error('hashSetIfExists is only supported with Redis cache'); + } + + const redisClient = (this.cache as RedisCache).store.client; + + const script = ` +if redis.call('EXISTS', KEYS[1]) == 1 then + return redis.call('HSET', KEYS[1], ARGV[1], ARGV[2]) +else + return 0 +end`; + + return redisClient.eval(script, { + keys: [this.getKey(key)], + arguments: [field, value], + }) as Promise; + } + + async hashDelete({ + key, + field, + }: { + key: string; + field: string; + }): Promise { + if (!this.isRedisCache()) { + throw new Error('hashDelete is only supported with Redis cache'); + } + + const redisClient = (this.cache as RedisCache).store.client; + + return redisClient.hDel(this.getKey(key), field); + } + async expire(key: string, ttlMs: Milliseconds): Promise { if (this.isRedisCache()) { return (this.cache as RedisCache).store.client.expire( diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/__tests__/automated-trigger.workspace-service.spec.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/__tests__/automated-trigger.workspace-service.spec.ts new file mode 100644 index 0000000000..a336c72bd4 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/__tests__/automated-trigger.workspace-service.spec.ts @@ -0,0 +1,144 @@ +import { Test, type TestingModule } from '@nestjs/testing'; + +import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; +import { AutomatedTriggerType } from 'src/modules/workflow/common/standard-objects/workflow-automated-trigger.workspace-entity'; +import { AutomatedTriggerWorkspaceService } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.workspace-service'; + +const mockRepository = { + insert: jest.fn().mockResolvedValue(undefined), + delete: jest.fn().mockResolvedValue(undefined), +}; + +const mockGlobalWorkspaceOrmManager = { + getRepository: jest.fn().mockResolvedValue(mockRepository), + executeInWorkspaceContext: jest + .fn() + .mockImplementation((fn: () => any) => fn()), +}; + +describe('AutomatedTriggerWorkspaceService', () => { + let service: AutomatedTriggerWorkspaceService; + + beforeEach(async () => { + jest.clearAllMocks(); + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + AutomatedTriggerWorkspaceService, + { + provide: GlobalWorkspaceOrmManager, + useValue: mockGlobalWorkspaceOrmManager, + }, + ], + }).compile(); + + service = module.get( + AutomatedTriggerWorkspaceService, + ); + }); + + describe('addAutomatedTrigger', () => { + const workspaceId = 'workspace-1'; + const workflowId = 'workflow-1'; + const cronSettings = { pattern: '* * * * *' }; + const databaseEventSettings = { + eventName: 'company.created', + fields: [], + }; + + it('should insert a CRON trigger', async () => { + await service.addAutomatedTrigger({ + workflowId, + type: AutomatedTriggerType.CRON, + settings: cronSettings, + workspaceId, + }); + + expect(mockRepository.insert).toHaveBeenCalledWith({ + type: AutomatedTriggerType.CRON, + settings: cronSettings, + workflowId, + }); + }); + + it('should insert a DATABASE_EVENT trigger', async () => { + await service.addAutomatedTrigger({ + workflowId, + type: AutomatedTriggerType.DATABASE_EVENT, + settings: databaseEventSettings, + workspaceId, + }); + + expect(mockRepository.insert).toHaveBeenCalledWith({ + type: AutomatedTriggerType.DATABASE_EVENT, + settings: databaseEventSettings, + workflowId, + }); + }); + + it('should use entityManager when provided', async () => { + const mockEntityManager = {} as any; + + await service.addAutomatedTrigger({ + workflowId, + type: AutomatedTriggerType.CRON, + settings: cronSettings, + workspaceId, + entityManager: mockEntityManager, + }); + + expect(mockRepository.insert).toHaveBeenCalledWith( + { type: AutomatedTriggerType.CRON, settings: cronSettings, workflowId }, + mockEntityManager, + ); + expect( + mockGlobalWorkspaceOrmManager.executeInWorkspaceContext, + ).not.toHaveBeenCalled(); + }); + + it('should use executeInWorkspaceContext when no entityManager provided', async () => { + await service.addAutomatedTrigger({ + workflowId, + type: AutomatedTriggerType.CRON, + settings: cronSettings, + workspaceId, + }); + + expect( + mockGlobalWorkspaceOrmManager.executeInWorkspaceContext, + ).toHaveBeenCalled(); + }); + }); + + describe('deleteAutomatedTrigger', () => { + const workspaceId = 'workspace-1'; + const workflowId = 'workflow-1'; + + it('should delete trigger by workflowId', async () => { + await service.deleteAutomatedTrigger({ + workflowId, + workspaceId, + }); + + expect(mockRepository.delete).toHaveBeenCalledWith({ workflowId }); + }); + + it('should use entityManager when provided', async () => { + const mockEntityManager = {} as any; + + await service.deleteAutomatedTrigger({ + workflowId, + workspaceId, + entityManager: mockEntityManager, + }); + + expect(mockRepository.delete).toHaveBeenCalledWith( + { workflowId }, + mockEntityManager, + ); + expect( + mockGlobalWorkspaceOrmManager.executeInWorkspaceContext, + ).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts index 12d3fe3fea..71cc30c885 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts @@ -1,6 +1,7 @@ import { Module } from '@nestjs/common'; import { TypeOrmModule } from '@nestjs/typeorm'; +import { CacheStorageModule } from 'src/engine/core-modules/cache-storage/cache-storage.module'; import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module'; import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module'; @@ -12,6 +13,7 @@ import { WorkflowDatabaseEventTriggerListener } from 'src/modules/workflow/workf @Module({ imports: [ TypeOrmModule.forFeature([WorkspaceEntity]), + CacheStorageModule, WorkflowCommonModule, WorkspaceDataSourceModule, ], diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/constants/workflow-cron-trigger-cache-key.constant.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/constants/workflow-cron-trigger-cache-key.constant.ts new file mode 100644 index 0000000000..5a50a656b7 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/constants/workflow-cron-trigger-cache-key.constant.ts @@ -0,0 +1 @@ +export const WORKFLOW_CRON_TRIGGER_CACHE_KEY = 'workflow-cron-triggers'; diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/constants/workflow-cron-trigger-cache-ttl.constant.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/constants/workflow-cron-trigger-cache-ttl.constant.ts new file mode 100644 index 0000000000..8fa26ef67d --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/constants/workflow-cron-trigger-cache-ttl.constant.ts @@ -0,0 +1 @@ +export const WORKFLOW_CRON_TRIGGER_CACHE_TTL_MS = 60 * 60 * 1000; diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/__tests__/workflow-cron-trigger-cron.job.spec.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/__tests__/workflow-cron-trigger-cron.job.spec.ts new file mode 100644 index 0000000000..52abb33330 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/__tests__/workflow-cron-trigger-cron.job.spec.ts @@ -0,0 +1,303 @@ +import { Test, type TestingModule } from '@nestjs/testing'; +import { getDataSourceToken, getRepositoryToken } from '@nestjs/typeorm'; + +import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; +import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; +import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; +import { WORKFLOW_CRON_TRIGGER_CACHE_KEY } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/constants/workflow-cron-trigger-cache-key.constant'; +import { WORKFLOW_CRON_TRIGGER_CACHE_TTL_MS } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/constants/workflow-cron-trigger-cache-ttl.constant'; +import { WorkflowCronTriggerCronJob } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/workflow-cron-trigger-cron.job'; +import { WorkflowTriggerJob } from 'src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job'; + +const WORKSPACE_1 = '20202020-0000-0000-0000-000000000001'; +const WORKSPACE_2 = '20202020-0000-0000-0000-000000000002'; +const WORKSPACE_3 = '20202020-0000-0000-0000-000000000003'; + +const mockCoreDataSource = { + query: jest.fn(), +}; + +const mockWorkspaceRepository = { + find: jest.fn(), +}; + +const mockMessageQueueService = { + add: jest.fn(), +}; + +const mockExceptionHandlerService = { + captureExceptions: jest.fn(), +}; + +const mockCacheStorageService = { + hashGetValues: jest.fn(), + hashSet: jest.fn(), + expire: jest.fn(), +}; + +describe('WorkflowCronTriggerCronJob', () => { + let job: WorkflowCronTriggerCronJob; + + beforeEach(async () => { + jest.clearAllMocks(); + jest.useFakeTimers(); + jest.setSystemTime(new Date('2026-04-02T15:00:30.000Z')); + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + WorkflowCronTriggerCronJob, + { + provide: getDataSourceToken(), + useValue: mockCoreDataSource, + }, + { + provide: getRepositoryToken(WorkspaceEntity), + useValue: mockWorkspaceRepository, + }, + { + provide: 'MESSAGE_QUEUE_workflow-queue', + useValue: mockMessageQueueService, + }, + { + provide: ExceptionHandlerService, + useValue: mockExceptionHandlerService, + }, + { + provide: CacheStorageNamespace.ModuleWorkflow, + useValue: mockCacheStorageService, + }, + ], + }).compile(); + + job = module.get(WorkflowCronTriggerCronJob); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + describe('handle - cache hit', () => { + it('should process cached triggers without querying the database', async () => { + mockCacheStorageService.hashGetValues.mockResolvedValue([ + JSON.stringify({ + workspaceId: WORKSPACE_1, + workflowId: 'workflow-1', + pattern: '* * * * *', + }), + ]); + + await job.handle(); + + expect(mockCacheStorageService.hashGetValues).toHaveBeenCalledWith( + WORKFLOW_CRON_TRIGGER_CACHE_KEY, + ); + expect(mockWorkspaceRepository.find).not.toHaveBeenCalled(); + expect(mockCoreDataSource.query).not.toHaveBeenCalled(); + }); + + it('should enqueue jobs for matching cron triggers', async () => { + mockCacheStorageService.hashGetValues.mockResolvedValue([ + JSON.stringify({ + workspaceId: WORKSPACE_1, + workflowId: 'workflow-1', + pattern: '* * * * *', + }), + ]); + + await job.handle(); + + expect(mockMessageQueueService.add).toHaveBeenCalledWith( + WorkflowTriggerJob.name, + { + workspaceId: WORKSPACE_1, + workflowId: 'workflow-1', + payload: {}, + }, + { retryLimit: 3 }, + ); + }); + + it('should not enqueue jobs when cron pattern does not match', async () => { + mockCacheStorageService.hashGetValues.mockResolvedValue([ + JSON.stringify({ + workspaceId: WORKSPACE_1, + workflowId: 'workflow-1', + pattern: '0 0 1 1 *', + }), + ]); + + await job.handle(); + + expect(mockMessageQueueService.add).not.toHaveBeenCalled(); + }); + + it('should skip triggers with undefined pattern', async () => { + mockCacheStorageService.hashGetValues.mockResolvedValue([ + JSON.stringify({ + workspaceId: WORKSPACE_1, + workflowId: 'workflow-1', + }), + ]); + + await job.handle(); + + expect(mockMessageQueueService.add).not.toHaveBeenCalled(); + }); + + it('should not rebuild cache on cache hit', async () => { + mockCacheStorageService.hashGetValues.mockResolvedValue([ + JSON.stringify({ + workspaceId: WORKSPACE_1, + workflowId: 'workflow-1', + pattern: '* * * * *', + }), + ]); + + await job.handle(); + + expect(mockCacheStorageService.hashSet).not.toHaveBeenCalled(); + }); + }); + + describe('handle - cache miss', () => { + it('should perform full scan when cache is empty', async () => { + mockCacheStorageService.hashGetValues.mockResolvedValue([]); + mockWorkspaceRepository.find.mockResolvedValue([ + { id: WORKSPACE_1 }, + { id: WORKSPACE_2 }, + { id: WORKSPACE_3 }, + ]); + mockCoreDataSource.query.mockResolvedValue([]); + + await job.handle(); + + expect(mockWorkspaceRepository.find).toHaveBeenCalled(); + expect(mockCoreDataSource.query).toHaveBeenCalledTimes(3); + }); + + it('should write each trigger to cache immediately and set TTL', async () => { + mockCacheStorageService.hashGetValues.mockResolvedValue([]); + mockWorkspaceRepository.find.mockResolvedValue([ + { id: WORKSPACE_1 }, + { id: WORKSPACE_2 }, + { id: WORKSPACE_3 }, + ]); + + mockCoreDataSource.query + .mockResolvedValueOnce([ + { + id: 'trigger-1', + workflowId: 'workflow-1', + settings: { pattern: '* * * * *' }, + }, + ]) + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([ + { + id: 'trigger-2', + workflowId: 'workflow-2', + settings: { pattern: '* * * * *' }, + }, + ]); + + await job.handle(); + + expect(mockCacheStorageService.hashSet).toHaveBeenCalledTimes(2); + expect(mockCacheStorageService.hashSet).toHaveBeenCalledWith({ + key: WORKFLOW_CRON_TRIGGER_CACHE_KEY, + field: 'workflow-1', + value: JSON.stringify({ + workspaceId: WORKSPACE_1, + workflowId: 'workflow-1', + pattern: '* * * * *', + }), + }); + expect(mockCacheStorageService.hashSet).toHaveBeenCalledWith({ + key: WORKFLOW_CRON_TRIGGER_CACHE_KEY, + field: 'workflow-2', + value: JSON.stringify({ + workspaceId: WORKSPACE_3, + workflowId: 'workflow-2', + pattern: '* * * * *', + }), + }); + expect(mockCacheStorageService.expire).toHaveBeenCalledWith( + WORKFLOW_CRON_TRIGGER_CACHE_KEY, + WORKFLOW_CRON_TRIGGER_CACHE_TTL_MS, + ); + }); + + it('should not write to cache when no workspaces have cron triggers', async () => { + mockCacheStorageService.hashGetValues.mockResolvedValue([]); + mockWorkspaceRepository.find.mockResolvedValue([{ id: WORKSPACE_1 }]); + mockCoreDataSource.query.mockResolvedValue([]); + + await job.handle(); + + expect(mockCacheStorageService.hashSet).not.toHaveBeenCalled(); + expect(mockCacheStorageService.expire).not.toHaveBeenCalled(); + }); + }); + + describe('error handling', () => { + it('should catch errors for cached triggers and continue processing', async () => { + mockCacheStorageService.hashGetValues.mockResolvedValue([ + 'invalid-json', + JSON.stringify({ + workspaceId: WORKSPACE_2, + workflowId: 'workflow-1', + pattern: '* * * * *', + }), + ]); + + await job.handle(); + + expect( + mockExceptionHandlerService.captureExceptions, + ).toHaveBeenCalledWith([expect.any(Error)]); + expect(mockMessageQueueService.add).toHaveBeenCalledWith( + WorkflowTriggerJob.name, + { + workspaceId: WORKSPACE_2, + workflowId: 'workflow-1', + payload: {}, + }, + { retryLimit: 3 }, + ); + }); + + it('should catch errors per workspace during full scan and continue', async () => { + mockCacheStorageService.hashGetValues.mockResolvedValue([]); + mockWorkspaceRepository.find.mockResolvedValue([ + { id: WORKSPACE_1 }, + { id: WORKSPACE_2 }, + ]); + mockCoreDataSource.query + .mockRejectedValueOnce(new Error('Schema not found')) + .mockResolvedValueOnce([ + { + id: 'trigger-1', + workflowId: 'workflow-1', + settings: { pattern: '* * * * *' }, + }, + ]); + + await job.handle(); + + expect( + mockExceptionHandlerService.captureExceptions, + ).toHaveBeenCalledWith([expect.any(Error)], { + workspace: { id: WORKSPACE_1 }, + }); + expect(mockMessageQueueService.add).toHaveBeenCalledWith( + WorkflowTriggerJob.name, + { + workspaceId: WORKSPACE_2, + workflowId: 'workflow-1', + payload: {}, + }, + { retryLimit: 3 }, + ); + }); + }); +}); diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/workflow-cron-trigger-cron.job.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/workflow-cron-trigger-cron.job.ts index 23e073b023..85b084cee2 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/workflow-cron-trigger-cron.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/workflow-cron-trigger-cron.job.ts @@ -5,6 +5,9 @@ import { isDefined } from 'twenty-shared/utils'; import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; import { DataSource, Repository } from 'typeorm'; +import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator'; +import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service'; +import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; @@ -16,6 +19,9 @@ import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.ent import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util'; import { AutomatedTriggerType } from 'src/modules/workflow/common/standard-objects/workflow-automated-trigger.workspace-entity'; import { type CronTriggerSettings } from 'src/modules/workflow/workflow-trigger/automated-trigger/constants/automated-trigger-settings'; +import { WORKFLOW_CRON_TRIGGER_CACHE_KEY } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/constants/workflow-cron-trigger-cache-key.constant'; +import { WORKFLOW_CRON_TRIGGER_CACHE_TTL_MS } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/constants/workflow-cron-trigger-cache-ttl.constant'; +import { type CachedCronTrigger } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/types/cached-cron-trigger.type'; import { WorkflowTriggerJob, type WorkflowTriggerJobData, @@ -36,6 +42,8 @@ export class WorkflowCronTriggerCronJob { @InjectMessageQueue(MessageQueue.workflowQueue) private readonly messageQueueService: MessageQueueService, private readonly exceptionHandlerService: ExceptionHandlerService, + @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow) + private readonly cacheStorageService: CacheStorageService, ) {} @Process(WorkflowCronTriggerCronJob.name) @@ -46,79 +54,160 @@ export class WorkflowCronTriggerCronJob { async handle() { this.logger.log('WorkflowCronTriggerCronJob started'); + const now = new Date(); + + const cachedValues = await this.cacheStorageService.hashGetValues( + WORKFLOW_CRON_TRIGGER_CACHE_KEY, + ); + + if (cachedValues.length > 0) { + this.logger.log(`Cache hit: ${cachedValues.length} cached cron triggers`); + + await this.getAndRunTriggersFromCache(cachedValues, now); + } else { + this.logger.log('Cache miss: performing full scan of all workspaces'); + + await this.getAndRunTriggersFromDatabase(now); + } + + this.logger.log('WorkflowCronTriggerCronJob completed'); + } + + private async getAndRunTriggersFromCache(cachedValues: string[], now: Date) { + for (const serialized of cachedValues) { + try { + const trigger = JSON.parse(serialized) as CachedCronTrigger; + + if (!isDefined(trigger.pattern)) { + continue; + } + + if (!shouldRunNow(trigger.pattern, now)) { + continue; + } + + this.logger.log( + `Enqueuing WorkflowTriggerJob for workflow ${trigger.workflowId} in workspace ${trigger.workspaceId}`, + ); + + await this.messageQueueService.add( + WorkflowTriggerJob.name, + { + workspaceId: trigger.workspaceId, + workflowId: trigger.workflowId, + payload: {}, + }, + { retryLimit: 3 }, + ); + } catch (error) { + this.logger.error(`Error processing cached trigger: ${error}`); + this.exceptionHandlerService.captureExceptions([error]); + } + } + } + + private async getAndRunTriggersFromDatabase(now: Date) { const activeWorkspaces = await this.workspaceRepository.find({ where: { activationStatus: WorkspaceActivationStatus.ACTIVE, }, + select: ['id'], }); this.logger.log(`Found ${activeWorkspaces.length} active workspaces`); - const now = new Date(); + let triggerCount = 0; - for (const activeWorkspace of activeWorkspaces) { - try { - const schemaName = getWorkspaceSchemaName(activeWorkspace.id); + for (const workspace of activeWorkspaces) { + const triggersToCache = await this.getAndRunWorkspaceTriggersFromDatabase( + workspace.id, + now, + ); - const workflowAutomatedCronTriggers = await this.coreDataSource.query( - `SELECT * FROM ${schemaName}."workflowAutomatedTrigger" WHERE type = '${AutomatedTriggerType.CRON}'`, - ); + for (const trigger of triggersToCache) { + await this.cacheStorageService.hashSet({ + key: WORKFLOW_CRON_TRIGGER_CACHE_KEY, + field: trigger.workflowId, + value: JSON.stringify(trigger), + }); + triggerCount++; + } + } - this.logger.log( - `Workspace ${activeWorkspace.id}: found ${workflowAutomatedCronTriggers.length} cron triggers`, - ); + if (triggerCount > 0) { + await this.cacheStorageService.expire( + WORKFLOW_CRON_TRIGGER_CACHE_KEY, + WORKFLOW_CRON_TRIGGER_CACHE_TTL_MS, + ); + } - for (const workflowAutomatedCronTrigger of workflowAutomatedCronTriggers) { - const settings = - workflowAutomatedCronTrigger.settings as CronTriggerSettings; + this.logger.log(`Cache rebuilt with ${triggerCount} cron triggers`); + } - this.logger.log( - `Trigger ${workflowAutomatedCronTrigger.id} for workflow ${workflowAutomatedCronTrigger.workflowId}: pattern=${settings.pattern}`, + private async getAndRunWorkspaceTriggersFromDatabase( + workspaceId: string, + now: Date, + ): Promise { + try { + const schemaName = getWorkspaceSchemaName(workspaceId); + + const workflowAutomatedCronTriggers = await this.coreDataSource.query( + `SELECT * FROM ${schemaName}."workflowAutomatedTrigger" WHERE type = '${AutomatedTriggerType.CRON}'`, + ); + + if (workflowAutomatedCronTriggers.length === 0) { + return []; + } + + this.logger.log( + `Workspace ${workspaceId}: found ${workflowAutomatedCronTriggers.length} cron triggers`, + ); + + const triggersToCache: CachedCronTrigger[] = []; + + for (const trigger of workflowAutomatedCronTriggers) { + const settings = trigger.settings as CronTriggerSettings; + + if (!isDefined(settings.pattern)) { + this.logger.warn( + `Trigger ${trigger.id}: skipping - pattern not defined`, ); + continue; + } - if (!isDefined(settings.pattern)) { - this.logger.warn( - `Trigger ${workflowAutomatedCronTrigger.id}: skipping - pattern not defined`, - ); - continue; - } + const cachedTrigger: CachedCronTrigger = { + workspaceId, + workflowId: trigger.workflowId, + pattern: settings.pattern, + }; - const shouldRun = shouldRunNow(settings.pattern, now); + triggersToCache.push(cachedTrigger); + if (shouldRunNow(settings.pattern, now)) { this.logger.log( - `Trigger ${workflowAutomatedCronTrigger.id}: shouldRunNow(${settings.pattern}, ${now.toISOString()}) = ${shouldRun}`, - ); - - if (!shouldRun) { - continue; - } - - this.logger.log( - `Trigger ${workflowAutomatedCronTrigger.id}: enqueuing WorkflowTriggerJob for workflow ${workflowAutomatedCronTrigger.workflowId}`, + `Trigger ${trigger.id}: enqueuing WorkflowTriggerJob for workflow ${trigger.workflowId}`, ); await this.messageQueueService.add( WorkflowTriggerJob.name, { - workspaceId: activeWorkspace.id, - workflowId: workflowAutomatedCronTrigger.workflowId, + workspaceId, + workflowId: trigger.workflowId, payload: {}, }, { retryLimit: 3 }, ); } - } catch (error) { - this.logger.error( - `Error processing workspace ${activeWorkspace.id}: ${error}`, - ); - this.exceptionHandlerService.captureExceptions([error], { - workspace: { - id: activeWorkspace.id, - }, - }); } - } - this.logger.log('WorkflowCronTriggerCronJob completed'); + return triggersToCache; + } catch (error) { + this.logger.error(`Error processing workspace ${workspaceId}: ${error}`); + this.exceptionHandlerService.captureExceptions([error], { + workspace: { id: workspaceId }, + }); + + return []; + } } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/types/cached-cron-trigger.type.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/types/cached-cron-trigger.type.ts new file mode 100644 index 0000000000..b73f2bae1e --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/types/cached-cron-trigger.type.ts @@ -0,0 +1,5 @@ +export type CachedCronTrigger = { + workspaceId: string; + workflowId: string; + pattern: string; +}; diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/workflow-trigger.module.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/workflow-trigger.module.ts index b008bf3440..ae9f0100fa 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/workflow-trigger.module.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/workflow-trigger.module.ts @@ -2,6 +2,7 @@ import { Module } from '@nestjs/common'; import { NestjsQueryTypeOrmModule } from '@ptc-org/nestjs-query-typeorm'; +import { CacheStorageModule } from 'src/engine/core-modules/cache-storage/cache-storage.module'; import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module'; import { CommandMenuItemModule } from 'src/engine/metadata-modules/command-menu-item/command-menu-item.module'; import { LogicFunctionModule } from 'src/engine/metadata-modules/logic-function/logic-function.module'; @@ -19,6 +20,7 @@ import { WorkflowTriggerWorkspaceService } from 'src/modules/workflow/workflow-t CodeStepBuildModule, WorkflowRunnerModule, AutomatedTriggerModule, + CacheStorageModule, CommandMenuItemModule, FeatureFlagModule, NestjsQueryTypeOrmModule.forFeature([ObjectMetadataEntity]), diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/workspace-services/workflow-trigger.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/workspace-services/workflow-trigger.workspace-service.ts index 926d4c937f..1ea0201089 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/workspace-services/workflow-trigger.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/workspace-services/workflow-trigger.workspace-service.ts @@ -3,6 +3,9 @@ import { msg } from '@lingui/core/macro'; import { isNonEmptyString } from '@sniptt/guards'; import { type ActorMetadata } from 'twenty-shared/types'; +import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator'; +import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service'; +import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; import { CommandMenuItemService } from 'src/engine/metadata-modules/command-menu-item/command-menu-item.service'; import { CommandMenuItemAvailabilityType } from 'src/engine/metadata-modules/command-menu-item/enums/command-menu-item-availability-type.enum'; import { EngineComponentKey } from 'src/engine/metadata-modules/command-menu-item/enums/engine-component-key.enum'; @@ -25,6 +28,8 @@ import { WORKFLOW_VERSION_STATUS_UPDATED } from 'src/modules/workflow/workflow-s import { type WorkflowVersionStatusUpdate } from 'src/modules/workflow/workflow-status/jobs/workflow-statuses-update.job'; import { AutomatedTriggerWorkspaceService } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.workspace-service'; import { type DatabaseEventTriggerSettings } from 'src/modules/workflow/workflow-trigger/automated-trigger/constants/automated-trigger-settings'; +import { WORKFLOW_CRON_TRIGGER_CACHE_KEY } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/constants/workflow-cron-trigger-cache-key.constant'; +import { type CachedCronTrigger } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/types/cached-cron-trigger.type'; import { WorkflowTriggerException, WorkflowTriggerExceptionCode, @@ -49,6 +54,8 @@ export class WorkflowTriggerWorkspaceService { private readonly automatedTriggerWorkspaceService: AutomatedTriggerWorkspaceService, private readonly workspaceEventEmitter: WorkspaceEventEmitter, private readonly commandMenuItemService: CommandMenuItemService, + @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow) + private readonly cacheStorageService: CacheStorageService, ) {} async runWorkflowVersion({ @@ -496,6 +503,18 @@ export class WorkflowTriggerWorkspaceService { entityManager: transactionContext?.entityManager, }); + const cachedTrigger: CachedCronTrigger = { + workspaceId, + workflowId: workflowVersion.workflowId, + pattern, + }; + + await this.cacheStorageService.hashSetIfExists({ + key: WORKFLOW_CRON_TRIGGER_CACHE_KEY, + field: workflowVersion.workflowId, + value: JSON.stringify(cachedTrigger), + }); + return; } default: @@ -514,6 +533,13 @@ export class WorkflowTriggerWorkspaceService { switch (workflowVersion.trigger.type) { case WorkflowTriggerType.DATABASE_EVENT: + await this.automatedTriggerWorkspaceService.deleteAutomatedTrigger({ + workflowId: workflowVersion.workflowId, + workspaceId, + entityManager: transactionContext?.entityManager, + }); + + return; case WorkflowTriggerType.CRON: await this.automatedTriggerWorkspaceService.deleteAutomatedTrigger({ workflowId: workflowVersion.workflowId, @@ -521,6 +547,11 @@ export class WorkflowTriggerWorkspaceService { entityManager: transactionContext?.entityManager, }); + await this.cacheStorageService.hashDelete({ + key: WORKFLOW_CRON_TRIGGER_CACHE_KEY, + field: workflowVersion.workflowId, + }); + return; case WorkflowTriggerType.MANUAL: case WorkflowTriggerType.WEBHOOK: