fix: replace N+1 queries with single batch query in CronTriggerCronJob

The cron trigger job was executing a separate database query for each
active workspace to find logic functions with cron triggers. As the
number of active workspaces grew, these sequential queries accumulated
and eventually caused a query read timeout.

Replace the per-workspace loop query with a single batched query using
TypeORM's In() operator to fetch all logic functions with cron triggers
across all active workspaces at once. Also add try-catch around the
message queue enqueue step so one failure doesn't crash the entire job.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

packages/twenty-server/src/engine/core-modules/logic-function/logic-function-trigger/triggers/cron/cron-trigger.cron.job.ts

@@ -1,8 +1,9 @@
+import { Logger } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 
 import { isDefined } from 'twenty-shared/utils';
 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
-import { IsNull, Not, Repository } from 'typeorm';
+import { In, IsNull, Not, Repository } from 'typeorm';
 
 import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
 import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
@@ -22,6 +23,8 @@ export const CRON_TRIGGER_CRON_PATTERN = '* * * * *';
 
 @Processor(MessageQueue.cronQueue)
 export class CronTriggerCronJob {
+  private readonly logger = new Logger(CronTriggerCronJob.name);
+
   constructor(
     @InjectMessageQueue(MessageQueue.logicFunctionQueue)
     private readonly messageQueueService: MessageQueueService,
@@ -41,29 +44,37 @@ export class CronTriggerCronJob {
       select: ['id'],
     });
 
+    if (activeWorkspaces.length === 0) {
+      return;
+    }
+
+    const activeWorkspaceIds = activeWorkspaces.map(
+      (workspace) => workspace.id,
+    );
+
     const now = new Date();
 
-    for (const activeWorkspace of activeWorkspaces) {
-      const logicFunctionsWithCronTrigger =
-        await this.logicFunctionRepository.find({
-          where: {
-            workspaceId: activeWorkspace.id,
-            cronTriggerSettings: Not(IsNull()),
-          },
-          select: ['id', 'cronTriggerSettings', 'workspaceId'],
-        });
+    const logicFunctionsWithCronTrigger =
+      await this.logicFunctionRepository.find({
+        where: {
+          workspaceId: In(activeWorkspaceIds),
+          cronTriggerSettings: Not(IsNull()),
+        },
+        select: ['id', 'cronTriggerSettings', 'workspaceId'],
+      });
 
-      for (const logicFunction of logicFunctionsWithCronTrigger) {
-        const cronSettings = logicFunction.cronTriggerSettings;
+    for (const logicFunction of logicFunctionsWithCronTrigger) {
+      const cronSettings = logicFunction.cronTriggerSettings;
 
-        if (!isDefined(cronSettings?.pattern)) {
-          continue;
-        }
+      if (!isDefined(cronSettings?.pattern)) {
+        continue;
+      }
 
-        if (!shouldRunNow(cronSettings.pattern, now)) {
-          continue;
-        }
+      if (!shouldRunNow(cronSettings.pattern, now)) {
+        continue;
+      }
 
+      try {
         await this.messageQueueService.add<LogicFunctionTriggerJobData[]>(
           LogicFunctionTriggerJob.name,
           [
@@ -75,6 +86,11 @@ export class CronTriggerCronJob {
           ],
           { retryLimit: 3 },
         );
+      } catch (error) {
+        this.logger.error(
+          `Failed to enqueue cron trigger for logic function ${logicFunction.id} in workspace ${logicFunction.workspaceId}`,
+          error instanceof Error ? error.stack : error,
+        );
       }
     }
   }