Improve workflow crons (#17720)

Issue 1: no info to debug cron trigger. Stop catching exception + using
logs instead of throwing for now

Issue 2: sentry often send timeouts errors for workflow crons. Probably
not real ones, it sends it if the job takes more than 5 minutes to run.
To fix, on each workflow cron we do:
- loop over active workspaces
- perform a query check that workspace is relevant, using count for
performances
- send a job if relevant

packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/workflow-cron-trigger-cron.job.ts

@@ -1,3 +1,4 @@
+import { Logger } from '@nestjs/common';
 import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
 
 import { isDefined } from 'twenty-shared/utils';
@@ -25,6 +26,8 @@ export const WORKFLOW_CRON_TRIGGER_CRON_PATTERN = '* * * * *';
 
 @Processor(MessageQueue.cronQueue)
 export class WorkflowCronTriggerCronJob {
+  private readonly logger = new Logger(WorkflowCronTriggerCronJob.name);
+
   constructor(
     @InjectDataSource()
     private readonly coreDataSource: DataSource,
@@ -41,12 +44,16 @@ export class WorkflowCronTriggerCronJob {
     WORKFLOW_CRON_TRIGGER_CRON_PATTERN,
   )
   async handle() {
+    this.logger.log('WorkflowCronTriggerCronJob started');
+
     const activeWorkspaces = await this.workspaceRepository.find({
       where: {
         activationStatus: WorkspaceActivationStatus.ACTIVE,
       },
     });
 
+    this.logger.log(`Found ${activeWorkspaces.length} active workspaces`);
+
     const now = new Date();
 
     for (const activeWorkspace of activeWorkspaces) {
@@ -57,18 +64,39 @@ export class WorkflowCronTriggerCronJob {
           `SELECT * FROM ${schemaName}."workflowAutomatedTrigger" WHERE type = '${AutomatedTriggerType.CRON}'`,
         );
 
+        this.logger.log(
+          `Workspace ${activeWorkspace.id}: found ${workflowAutomatedCronTriggers.length} cron triggers`,
+        );
+
         for (const workflowAutomatedCronTrigger of workflowAutomatedCronTriggers) {
           const settings =
             workflowAutomatedCronTrigger.settings as CronTriggerSettings;
 
+          this.logger.log(
+            `Trigger ${workflowAutomatedCronTrigger.id} for workflow ${workflowAutomatedCronTrigger.workflowId}: pattern=${settings.pattern}`,
+          );
+
           if (!isDefined(settings.pattern)) {
+            this.logger.warn(
+              `Trigger ${workflowAutomatedCronTrigger.id}: skipping - pattern not defined`,
+            );
             continue;
           }
 
-          if (!shouldRunNow(settings.pattern, now)) {
+          const shouldRun = 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}`,
+          );
+
           await this.messageQueueService.add<WorkflowTriggerJobData>(
             WorkflowTriggerJob.name,
             {
@@ -80,6 +108,9 @@ export class WorkflowCronTriggerCronJob {
           );
         }
       } catch (error) {
+        this.logger.error(
+          `Error processing workspace ${activeWorkspace.id}: ${error}`,
+        );
         this.exceptionHandlerService.captureExceptions([error], {
           workspace: {
             id: activeWorkspace.id,
@@ -87,5 +118,7 @@ export class WorkflowCronTriggerCronJob {
         });
       }
     }
+
+    this.logger.log('WorkflowCronTriggerCronJob completed');
   }
 }

packages/twenty-server/src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job.ts

@@ -1,27 +1,19 @@
-import { Scope } from '@nestjs/common';
+import { Logger, Scope } from '@nestjs/common';
 
 import isEmpty from 'lodash.isempty';
 import { FieldActorSource } from 'twenty-shared/types';
 import { isDefined } from 'twenty-shared/utils';
 
-import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
-import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
-import { handleWorkflowTriggerException } from 'src/engine/core-modules/workflow/filters/workflow-trigger-graphql-api-exception.filter';
 import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
 import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
-import {
-  WorkflowVersionStatus,
-  type WorkflowVersionWorkspaceEntity,
-} from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
+import { WorkflowVersionStatus } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
 import { type WorkflowWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity';
+import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
 import { WorkflowRunnerWorkspaceService } from 'src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service';
-import {
-  WorkflowTriggerException,
-  WorkflowTriggerExceptionCode,
-} from 'src/modules/workflow/workflow-trigger/exceptions/workflow-trigger.exception';
+import { WorkflowTriggerExceptionCode } from 'src/modules/workflow/workflow-trigger/exceptions/workflow-trigger.exception';
 
 export type WorkflowTriggerJobData = {
   workspaceId: string;
@@ -33,11 +25,11 @@ const DEFAULT_WORKFLOW_NAME = 'Workflow';
 
 @Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST })
 export class WorkflowTriggerJob {
+  private readonly logger = new Logger(WorkflowTriggerJob.name);
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly workflowCommonWorkspaceService: WorkflowCommonWorkspaceService,
     private readonly workflowRunnerWorkspaceService: WorkflowRunnerWorkspaceService,
-    @InjectMessageQueue(MessageQueue.workflowQueue)
-    private readonly messageQueueService: MessageQueueService,
   ) {}
 
   @Process(WorkflowTriggerJob.name)
@@ -45,77 +37,64 @@ export class WorkflowTriggerJob {
     const authContext = buildSystemAuthContext(data.workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      try {
-        const workflowRepository =
-          await this.globalWorkspaceOrmManager.getRepository<WorkflowWorkspaceEntity>(
-            data.workspaceId,
-            'workflow',
-            { shouldBypassPermissionChecks: true },
-          );
+      const workflowRepository =
+        await this.globalWorkspaceOrmManager.getRepository<WorkflowWorkspaceEntity>(
+          data.workspaceId,
+          'workflow',
+          { shouldBypassPermissionChecks: true },
+        );
 
-        const workflow = await workflowRepository.findOneBy({
-          id: data.workflowId,
-        });
+      const workflow = await workflowRepository.findOneBy({
+        id: data.workflowId,
+      });
 
-        if (!workflow) {
-          throw new WorkflowTriggerException(
-            `Workflow ${data.workflowId} not found in workspace ${data.workspaceId}`,
-            WorkflowTriggerExceptionCode.NOT_FOUND,
-          );
-        }
+      if (!workflow) {
+        this.logger.error(
+          `Workflow ${data.workflowId} not found in workspace ${data.workspaceId}`,
+          WorkflowTriggerExceptionCode.NOT_FOUND,
+        );
 
-        if (!workflow.lastPublishedVersionId) {
-          throw new WorkflowTriggerException(
-            `Workflow ${data.workflowId} has no published version in workspace ${data.workspaceId}`,
-            WorkflowTriggerExceptionCode.INTERNAL_ERROR,
-          );
-        }
+        return;
+      }
 
-        const workflowVersionRepository =
-          await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
-            data.workspaceId,
-            'workflowVersion',
-            { shouldBypassPermissionChecks: true },
-          );
+      if (!workflow.lastPublishedVersionId) {
+        this.logger.error(
+          `Workflow ${data.workflowId} has no published version in workspace ${data.workspaceId}`,
+          WorkflowTriggerExceptionCode.INTERNAL_ERROR,
+        );
 
-        const workflowVersion = await workflowVersionRepository.findOneBy({
-          id: workflow.lastPublishedVersionId,
-        });
+        return;
+      }
 
-        if (!workflowVersion) {
-          throw new WorkflowTriggerException(
-            `Workflow version ${workflow.lastPublishedVersionId} not found in workspace ${data.workspaceId}`,
-            WorkflowTriggerExceptionCode.NOT_FOUND,
-          );
-        }
-        if (workflowVersion.status !== WorkflowVersionStatus.ACTIVE) {
-          throw new WorkflowTriggerException(
-            `Workflow version ${workflowVersion.id} is not active in workspace ${data.workspaceId}`,
-            WorkflowTriggerExceptionCode.INTERNAL_ERROR,
-          );
-        }
-
-        await this.workflowRunnerWorkspaceService.run({
+      const workflowVersion =
+        await this.workflowCommonWorkspaceService.getWorkflowVersionOrFail({
           workspaceId: data.workspaceId,
           workflowVersionId: workflow.lastPublishedVersionId,
-          payload: data.payload,
-          source: {
-            source: FieldActorSource.WORKFLOW,
-            name:
-              isDefined(workflow.name) && !isEmpty(workflow.name)
-                ? workflow.name
-                : DEFAULT_WORKFLOW_NAME,
-            context: {},
-            workspaceMemberId: null,
-          },
         });
-      } catch (e) {
-        await this.messageQueueService.removeCron({
-          jobName: WorkflowTriggerJob.name,
-          jobId: data.workflowId,
-        });
-        handleWorkflowTriggerException(e);
+
+      if (workflowVersion.status !== WorkflowVersionStatus.ACTIVE) {
+        this.logger.error(
+          `Workflow version ${workflowVersion?.id} is not active in workspace ${data.workspaceId}`,
+          WorkflowTriggerExceptionCode.INTERNAL_ERROR,
+        );
+
+        return;
       }
+
+      await this.workflowRunnerWorkspaceService.run({
+        workspaceId: data.workspaceId,
+        workflowVersionId: workflow.lastPublishedVersionId,
+        payload: data.payload,
+        source: {
+          source: FieldActorSource.WORKFLOW,
+          name:
+            isDefined(workflow.name) && !isEmpty(workflow.name)
+              ? workflow.name
+              : DEFAULT_WORKFLOW_NAME,
+          context: {},
+          workspaceMemberId: null,
+        },
+      });
     }, authContext);
   }
 }