[comp] Production Deploy

apps/api/src/trigger/integration-platform/run-integration-checks-schedule.spec.ts

@@ -3,9 +3,11 @@ import { db } from '@db';
 import { getManifest } from '@trycompai/integration-platform';
 import {
   filterDueTasks,
+  groupTasksByOrg,
   integrationChecksSchedule,
   resolveProviderChecks,
 } from './run-integration-checks-schedule';
+import { runOrgIntegrationChecks } from './run-org-integration-checks';
 
 // Mock @db at the module boundary so importing the orchestrator does not try
 // to connect to Postgres. We never call the scheduled `run` function itself
@@ -42,8 +44,11 @@ jest.mock('@trigger.dev/sdk', () => ({
   },
 }));
 
-jest.mock('./run-task-integration-checks', () => ({
-  runTaskIntegrationChecks: { batchTrigger: jest.fn() },
+// The orchestrator now dispatches ONE per-org runner instead of N per-task
+// runs. Mock the runner module so importing the orchestrator doesn't load the
+// real runner (which calls queue()/task() at module load).
+jest.mock('./run-org-integration-checks', () => ({
+  runOrgIntegrationChecks: { batchTrigger: jest.fn() },
 }));
 
 jest.mock('./run-device-sync', () => ({
@@ -181,6 +186,54 @@ describe('resolveProviderChecks (static vs dynamic)', () => {
   });
 });
 
+describe('groupTasksByOrg (bundling key)', () => {
+  const base = {
+    taskTitle: 'T',
+    connectionId: 'conn',
+    providerSlug: 'github',
+    checkIds: ['c1'],
+  };
+
+  it('groups tasks by organization and attaches the org name', () => {
+    const groups = groupTasksByOrg({
+      tasksToRun: [
+        { ...base, taskId: 't1', organizationId: 'org1' },
+        { ...base, taskId: 't2', organizationId: 'org1' },
+        { ...base, taskId: 't3', organizationId: 'org2' },
+      ],
+      orgNameById: new Map([
+        ['org1', 'Org One'],
+        ['org2', 'Org Two'],
+      ]),
+    });
+
+    expect(groups).toHaveLength(2);
+    const org1 = groups.find((g) => g.organizationId === 'org1');
+    expect(org1?.organizationName).toBe('Org One');
+    expect(org1?.tasks.map((t) => t.taskId)).toEqual(['t1', 't2']);
+    // organizationId is hoisted to the group; tasks carry the per-task shape.
+    expect(org1?.tasks[0]).not.toHaveProperty('organizationId');
+    expect(groups.find((g) => g.organizationId === 'org2')?.tasks).toHaveLength(
+      1,
+    );
+  });
+
+  it('falls back to "your organization" when the name is unknown', () => {
+    const groups = groupTasksByOrg({
+      tasksToRun: [{ ...base, taskId: 't1', organizationId: 'orgX' }],
+      orgNameById: new Map(),
+    });
+
+    expect(groups[0]?.organizationName).toBe('your organization');
+  });
+
+  it('returns [] for no tasks', () => {
+    expect(
+      groupTasksByOrg({ tasksToRun: [], orgNameById: new Map() }),
+    ).toEqual([]);
+  });
+});
+
 describe('orchestrator excludes MANUAL tasks from scheduled runs', () => {
   // Cast the db/getManifest mocks to their jest.Mock shape for setup.
   const taskFindMany = (db as unknown as { task: { findMany: jest.Mock } }).task
@@ -236,3 +289,94 @@ describe('orchestrator excludes MANUAL tasks from scheduled runs', () => {
     );
   });
 });
+
+describe('orchestrator dispatches ONE runner per org', () => {
+  const taskFindMany = (db as unknown as { task: { findMany: jest.Mock } }).task
+    .findMany;
+  const connectionFindMany = (
+    db as unknown as { integrationConnection: { findMany: jest.Mock } }
+  ).integrationConnection.findMany;
+  const dynamicFindMany = (
+    db as unknown as { dynamicIntegration: { findMany: jest.Mock } }
+  ).dynamicIntegration.findMany;
+  const orgFindMany = (
+    db as unknown as { organization: { findMany: jest.Mock } }
+  ).organization.findMany;
+  const getManifestMock = getManifest as jest.Mock;
+  const batchTrigger = (
+    runOrgIntegrationChecks as unknown as { batchTrigger: jest.Mock }
+  ).batchTrigger;
+
+  const runOrchestrator = (
+    integrationChecksSchedule as unknown as {
+      run: (p: { timestamp: Date; lastTimestamp?: Date }) => Promise<unknown>;
+    }
+  ).run;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    // Two orgs, each with one active connection mapped to the same template.
+    connectionFindMany.mockResolvedValue([
+      {
+        id: 'conn1',
+        provider: { slug: 'github' },
+        organizationId: 'org1',
+        organization: { id: 'org1', name: 'Org One' },
+        metadata: null,
+      },
+      {
+        id: 'conn2',
+        provider: { slug: 'github' },
+        organizationId: 'org2',
+        organization: { id: 'org2', name: 'Org Two' },
+        metadata: null,
+      },
+    ]);
+    dynamicFindMany.mockResolvedValue([]);
+    getManifestMock.mockReturnValue({
+      checks: [{ id: 'c1', taskMapping: 'tpl_a' }],
+    });
+    // One due (daily) task per connection.
+    taskFindMany
+      .mockResolvedValueOnce([
+        {
+          id: 'task_org1',
+          title: 'Org1 Task',
+          taskTemplateId: 'tpl_a',
+          integrationScheduleFrequency: 'daily',
+          integrationLastRunAt: null,
+        },
+      ])
+      .mockResolvedValueOnce([
+        {
+          id: 'task_org2',
+          title: 'Org2 Task',
+          taskTemplateId: 'tpl_a',
+          integrationScheduleFrequency: 'daily',
+          integrationLastRunAt: null,
+        },
+      ]);
+    orgFindMany.mockResolvedValue([]);
+  });
+
+  it('triggers the per-org runner once with one payload per org (not per task)', async () => {
+    await runOrchestrator({ timestamp: new Date(), lastTimestamp: new Date() });
+
+    expect(batchTrigger).toHaveBeenCalledTimes(1);
+    const payloads = batchTrigger.mock.calls[0][0] as Array<{
+      payload: {
+        organizationId: string;
+        organizationName: string;
+        tasks: Array<{ taskId: string }>;
+      };
+    }>;
+    expect(payloads).toHaveLength(2);
+
+    const org1 = payloads.find((p) => p.payload.organizationId === 'org1');
+    expect(org1?.payload.organizationName).toBe('Org One');
+    expect(org1?.payload.tasks.map((t) => t.taskId)).toEqual(['task_org1']);
+
+    const org2 = payloads.find((p) => p.payload.organizationId === 'org2');
+    expect(org2?.payload.tasks.map((t) => t.taskId)).toEqual(['task_org2']);
+  });
+});

apps/api/src/trigger/integration-platform/run-integration-checks-schedule.ts

@@ -1,7 +1,10 @@
 import { getManifest } from '@trycompai/integration-platform';
 import { db, TaskAutomationStatus, TaskFrequency } from '@db';
 import { logger, schedules } from '@trigger.dev/sdk';
-import { runTaskIntegrationChecks } from './run-task-integration-checks';
+import {
+  runOrgIntegrationChecks,
+  type OrgTaskCheck,
+} from './run-org-integration-checks';
 import { runDeviceSync } from './run-device-sync';
 import { isCheckDisabledForTask } from '../../integration-platform/utils/disabled-task-checks';
 import { isDueToday } from '../shared/is-due-today';
@@ -66,6 +69,59 @@ export function resolveProviderChecks({
   return dynamicChecks ?? [];
 }
 
+/** A flat scheduled-task entry, before grouping by org. */
+export interface ScheduledTask {
+  taskId: string;
+  taskTitle: string;
+  connectionId: string;
+  providerSlug: string;
+  organizationId: string;
+  checkIds: string[];
+}
+
+/** One org's worth of scheduled tasks, ready to hand to the per-org runner. */
+export interface OrgTaskGroup {
+  organizationId: string;
+  organizationName: string;
+  tasks: OrgTaskCheck[];
+}
+
+/**
+ * Group the flat task list into one entry per organization, attaching each
+ * org's display name. Pure + exported for unit testing. This is what makes the
+ * failure email bundle-able: one runner (and thus one email) per org instead of
+ * one independent run (and email) per task.
+ */
+export function groupTasksByOrg({
+  tasksToRun,
+  orgNameById,
+}: {
+  tasksToRun: ScheduledTask[];
+  orgNameById: Map<string, string>;
+}): OrgTaskGroup[] {
+  const byOrg = new Map<string, OrgTaskGroup>();
+  for (const t of tasksToRun) {
+    let group = byOrg.get(t.organizationId);
+    if (!group) {
+      group = {
+        organizationId: t.organizationId,
+        organizationName:
+          orgNameById.get(t.organizationId) ?? 'your organization',
+        tasks: [],
+      };
+      byOrg.set(t.organizationId, group);
+    }
+    group.tasks.push({
+      taskId: t.taskId,
+      taskTitle: t.taskTitle,
+      connectionId: t.connectionId,
+      providerSlug: t.providerSlug,
+      checkIds: t.checkIds,
+    });
+  }
+  return Array.from(byOrg.values());
+}
+
 /**
  * Daily scheduled task (orchestrator) that finds all tasks with integration checks
  * and triggers individual check runs for each.
@@ -95,6 +151,17 @@ export const integrationChecksSchedule = schedules.task({
 
     logger.info(`Found ${activeConnections.length} active connections`);
 
+    // Org display names, used by the per-org runner's bundled failure email.
+    const orgNameById = new Map<string, string>();
+    for (const connection of activeConnections) {
+      if (connection.organization?.name) {
+        orgNameById.set(
+          connection.organizationId,
+          connection.organization.name,
+        );
+      }
+    }
+
     // Dynamic (DB-backed) integrations are NOT in the Trigger.dev manifest
     // registry, so getManifest() returns undefined for them below. Load their
     // enabled check → task mappings straight from the DB so this orchestrator
@@ -118,14 +185,7 @@ export const integrationChecksSchedule = schedules.task({
     );
 
     // For each connection, find tasks that have checks mapped to them
-    const tasksToRun: Array<{
-      taskId: string;
-      taskTitle: string;
-      connectionId: string;
-      providerSlug: string;
-      organizationId: string;
-      checkIds: string[];
-    }> = [];
+    const tasksToRun: ScheduledTask[] = [];
 
     for (const connection of activeConnections) {
       // Static providers resolve from the code manifest; dynamic ones from the
@@ -202,35 +262,48 @@ export const integrationChecksSchedule = schedules.task({
       }
     }
 
-    // Trigger integration checks in batches
-    let totalTriggered = 0;
+    // Dispatch ONE runner per org (fire-and-forget). Each runner waits on its
+    // own tasks and sends a SINGLE bundled failure email — instead of every
+    // per-task run emailing independently (the old one-email-per-failure spam).
+    // Keeping the top level fire-and-forget means the orchestrator stays fast
+    // and a slow/failing org can't hold the whole daily run open.
+    let tasksTriggered = 0;
+    let orgsTriggered = 0;
+
+    const orgGroups = groupTasksByOrg({ tasksToRun, orgNameById });
 
-    if (tasksToRun.length === 0) {
+    if (orgGroups.length === 0) {
       logger.info('No tasks with mapped integration checks found');
     } else {
       logger.info(
-        `Found ${tasksToRun.length} tasks with integration checks to run`,
+        `Found ${tasksToRun.length} task(s) across ${orgGroups.length} org(s) to run`,
       );
 
-      const BATCH_SIZE = 500;
-      const triggerPayloads = tasksToRun.map((t) => ({ payload: t }));
+      const ORG_BATCH_SIZE = 100;
+      const triggerPayloads = orgGroups.map((g) => ({ payload: g }));
 
       try {
-        for (let i = 0; i < triggerPayloads.length; i += BATCH_SIZE) {
-          const batch = triggerPayloads.slice(i, i + BATCH_SIZE);
-          await runTaskIntegrationChecks.batchTrigger(batch);
-          totalTriggered += batch.length;
+        for (let i = 0; i < triggerPayloads.length; i += ORG_BATCH_SIZE) {
+          const batch = triggerPayloads.slice(i, i + ORG_BATCH_SIZE);
+          await runOrgIntegrationChecks.batchTrigger(batch);
+          orgsTriggered += batch.length;
+          tasksTriggered += batch.reduce(
+            (n, p) => n + p.payload.tasks.length,
+            0,
+          );
 
           logger.info(
-            `Triggered batch ${Math.floor(i / BATCH_SIZE) + 1}: ${batch.length} tasks`,
+            `Triggered org batch ${Math.floor(i / ORG_BATCH_SIZE) + 1}: ${batch.length} org(s)`,
           );
         }
 
-        logger.info(`Triggered ${totalTriggered} task integration check runs`);
+        logger.info(
+          `Triggered ${orgsTriggered} org runner(s) covering ${tasksTriggered} task(s)`,
+        );
       } catch (error) {
-        logger.error('Failed to trigger task integration checks', {
+        logger.error('Failed to trigger org integration check runs', {
           error: error instanceof Error ? error.message : String(error),
-          triggeredBeforeError: totalTriggered,
+          orgsTriggeredBeforeError: orgsTriggered,
         });
       }
     }
@@ -281,10 +354,13 @@ export const integrationChecksSchedule = schedules.task({
     logger.info(`Triggered ${deviceSyncsTriggered} device syncs`);
 
     return {
-      // Report failure when not every queued task batch was dispatched OR a
-      // device-sync dispatch threw, so partial/failed runs aren't masked.
-      success: totalTriggered === tasksToRun.length && deviceSyncFailures === 0,
-      tasksTriggered: totalTriggered,
+      // Report failure when not every queued task was dispatched (via its org
+      // runner) OR a device-sync dispatch threw, so partial/failed runs aren't
+      // masked.
+      success:
+        tasksTriggered === tasksToRun.length && deviceSyncFailures === 0,
+      tasksTriggered,
+      orgsTriggered,
       deviceSyncsTriggered,
     };
   },