test(messaging): messaging and calendar sync integration suites (#22567)

13 integration suites driving the real sync pipeline end to end — OAuth connect via the actual `/auth/google-apis/get-access-token` / `microsoft-apis` callbacks (transient token + mocked provider token exchange), real queue workers, provider APIs mocked at the HTTP layer with msw. **Messaging (8):** Gmail list fetch + import, Gmail folder discovery, Microsoft folder discovery, history-based incremental sync, stale-sync recovery, sync failure lifecycle (429 throttle → exhaustion → relaunch; declined refresh token → insufficient permissions), token refresh, connected-account cleanup cascade. **Calendar (5):** Google events import (full + sync-token incremental), Microsoft events import (delta fetch + import), stale-sync recovery, failure lifecycle, cleanup cascade.  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/22567?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
2026-07-07 18:38:05 +05:30
parent 71ad0fb5fc
commit d99e6db93d
49 changed files with 2385 additions and 10 deletions
@@ -14,13 +14,17 @@ IS_WORKSPACE_CREATION_LIMITED_TO_SERVER_ADMINS=false

 ENTERPRISE_VALIDITY_TOKEN=eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJkZXYtc3Vic2NyaXB0aW9uLWlkIiwic3RhdHVzIjoidmFsaWQiLCJpYXQiOjE3NzMzMDg4MzMsImV4cCI6NDg5NzUxMTIzM30.qhfrW_SV2Y86fWtWXsALlAVhxmMxylUUIefN0fki10Q2NTGGqFVXZrNn2WacJY37yq3m5y4WgwZw34ua6E0ff_YUXsrlY5OHJWHT9DMqKCRn-JujHJnnYp3VHLncy5CvxH5r9mfPFp-5AWe1pYeR1T63sTiejH3sfDrNE357SB7KVti8LCcnsJxEtXB2tRnvyvdun7A-GKoKYEIam-16ZRKKFs6GaWo8ObHdfm8yBt6uK4DZSGPWb644QyWh9FtDxbzJ0ti54DuHSlErLgIp1NNEsMA0MK7zFY7StRaOdt72rxE1ZHwN7e6HhweTU4ORVUPfYkjDFLB2fF7Pa7Kvdg

-AUTH_GOOGLE_ENABLED=false
-MESSAGING_PROVIDER_GMAIL_ENABLED=false
+AUTH_GOOGLE_ENABLED=true
+AUTH_GOOGLE_CLIENT_ID=mock-google-client-id
+AUTH_GOOGLE_CLIENT_SECRET=mock-google-client-secret
+MESSAGING_PROVIDER_GMAIL_ENABLED=true
 IS_IMAP_SMTP_CALDAV_ENABLED=true
 IS_IMAP_SMTP_CALDAV_CONNECTION_TEST_ENABLED=false
-CALENDAR_PROVIDER_GOOGLE_ENABLED=false
-MESSAGING_PROVIDER_MICROSOFT_ENABLED=false
-CALENDAR_PROVIDER_MICROSOFT_ENABLED=false
+CALENDAR_PROVIDER_GOOGLE_ENABLED=true
+AUTH_MICROSOFT_CLIENT_ID=mock-microsoft-client-id
+AUTH_MICROSOFT_CLIENT_SECRET=mock-microsoft-client-secret
+MESSAGING_PROVIDER_MICROSOFT_ENABLED=true
+CALENDAR_PROVIDER_MICROSOFT_ENABLED=true

 AUTH_GOOGLE_CALLBACK_URL=http://localhost:3000/auth/google/redirect
 AUTH_GOOGLE_APIS_CALLBACK_URL=http://localhost:3000/auth/google-apis/get-access-token
@@ -38,11 +38,12 @@ const jestConfig: JestConfigWithTsJest = {
  setupFilesAfterEnv: ['<rootDir>/test/integration/utils/setup-wait-for-all-jobs-between-tests.ts'],
  testTimeout: 20000,
  maxWorkers: 1,
-  // jsdom 29 pulls ESM-only transitive deps (parse5, entities, tough-cookie,
-  // @exodus/bytes via html-encoding-sniffer, @csstools/@asamuzakjp css engine);
-  // let swc transform them (and .mjs below) so jest can require jsdom.
+  // jsdom 29 and msw ship ESM-only transitive deps (parse5, entities,
+  // tough-cookie, @exodus/bytes via html-encoding-sniffer, @csstools/@asamuzakjp
+  // css engine, @mswjs/interceptors and friends); let swc transform them
+  // (and .mjs below) so jest can require them.
  transformIgnorePatterns: [
-    '/node_modules/(?!(jsdom|html-encoding-sniffer|whatwg-encoding|@exodus|parse5|entities|tough-cookie|@csstools|@asamuzakjp)/)',
+    '/node_modules/(?!(jsdom|html-encoding-sniffer|whatwg-encoding|@exodus|parse5|entities|tough-cookie|@csstools|@asamuzakjp|msw|@mswjs|until-async|@bundled-es-modules|@open-draft|strict-event-emitter|headers-polyfill|outvariant|is-node-process|path-to-regexp|statuses|cookie|digest-fetch|md5|email-reply-parser)/)',
  ],
  transform: {
    '^.+\\.(t|j|mj)s$': [
@@ -207,6 +207,7 @@
    "@types/lodash.uniqby": "^4.7.9",
    "@types/ms": "^0.7.31",
    "@types/node": "^24.0.0",
+    "@types/node-fetch": "^2.6.12",
    "@types/nodemailer": "^7.0.3",
    "@types/passport-google-oauth20": "^2.0.11",
    "@types/passport-jwt": "^3.0.8",
@@ -220,6 +221,8 @@
    "@types/uuid": "^9.0.2",
    "jest": "29.7.0",
    "jest-environment-node": "^29.4.1",
+    "msw": "^2.12.7",
+    "node-fetch": "^2.7.0",
    "prettier": "^3.1.1",
    "rimraf": "^5.0.5",
    "supertest": "^6.1.3",
@@ -0,0 +1,102 @@
+import { randomUUID } from 'node:crypto';
+
+import { ConnectedAccountProvider } from 'twenty-shared/types';
+
+import { googleCalendarEvent } from 'test/integration/google/mocks/google-calendar-event.util';
+import { setupGoogleMock } from 'test/integration/google/mocks/setup-google-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import {
+  findRecordIdsByFilter,
+  findRecordNodesByFilter,
+} from 'test/integration/utils/find-records-by-filter.util';
+import { deleteConnectedAccount } from 'test/integration/utils/query-messaging.util';
+import { runCalendarChannelEventsImport } from 'test/integration/utils/run-calendar-channel-events-import.util';
+import { runCalendarChannelListFetch } from 'test/integration/utils/run-calendar-channel-list-fetch.util';
+
+const HANDLE = 'calendar-cleanup@apple.dev';
+
+describe('Calendar connected account cleanup (integration)', () => {
+  const eventId = `google-calendar-event-${randomUUID()}`;
+
+  const gmail = setupGoogleMock({ handle: HANDLE });
+
+  let channel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    channel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: HANDLE,
+    });
+
+    gmail.serveCalendarEvents([
+      googleCalendarEvent({
+        id: eventId,
+        attendees: [
+          { email: `organizer-${eventId}@example.com`, organizer: true },
+          { email: `attendee-${eventId}@example.com` },
+        ],
+      }),
+    ]);
+
+    await runCalendarChannelListFetch(channel.calendarChannelId);
+    await runCalendarChannelEventsImport(channel.calendarChannelId);
+  }, 60000);
+
+  afterAll(async () => {
+    await channel?.cleanup().catch(() => undefined);
+  });
+
+  it('deletes all associated calendar data when the connected account is removed', async () => {
+    const associations = await findRecordNodesByFilter<{
+      id: string;
+      calendarEventId: string;
+    }>(
+      'calendarChannelEventAssociation',
+      'calendarChannelEventAssociations',
+      `id
+        calendarEventId`,
+      { calendarChannelId: { eq: channel.calendarChannelId } },
+    );
+
+    expect(associations).toHaveLength(1);
+
+    const eventIds = associations.map(
+      (association) => association.calendarEventId,
+    );
+
+    expect(
+      await findRecordIdsByFilter('calendarEvent', 'calendarEvents', {
+        id: { in: eventIds },
+      }),
+    ).not.toHaveLength(0);
+    expect(
+      await findRecordIdsByFilter(
+        'calendarEventParticipant',
+        'calendarEventParticipants',
+        { calendarEventId: { in: eventIds } },
+      ),
+    ).not.toHaveLength(0);
+
+    await deleteConnectedAccount(channel.connectedAccountId);
+
+    expect(
+      await findRecordIdsByFilter(
+        'calendarChannelEventAssociation',
+        'calendarChannelEventAssociations',
+        { calendarChannelId: { eq: channel.calendarChannelId } },
+      ),
+    ).toHaveLength(0);
+    expect(
+      await findRecordIdsByFilter('calendarEvent', 'calendarEvents', {
+        id: { in: eventIds },
+      }),
+    ).toHaveLength(0);
+    expect(
+      await findRecordIdsByFilter(
+        'calendarEventParticipant',
+        'calendarEventParticipants',
+        { calendarEventId: { in: eventIds } },
+      ),
+    ).toHaveLength(0);
+  }, 60000);
+});
@@ -0,0 +1,76 @@
+import { randomUUID } from 'node:crypto';
+
+import {
+  CalendarChannelSyncStage,
+  ConnectedAccountProvider,
+} from 'twenty-shared/types';
+
+import { googleCalendarEvent } from 'test/integration/google/mocks/google-calendar-event.util';
+import { setupGoogleMock } from 'test/integration/google/mocks/setup-google-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import { findImportedCalendarEventTitles } from 'test/integration/utils/find-imported-records.util';
+import { queryCalendarChannel } from 'test/integration/utils/query-messaging.util';
+import { runCalendarChannelEventsImport } from 'test/integration/utils/run-calendar-channel-events-import.util';
+import { runCalendarChannelListFetch } from 'test/integration/utils/run-calendar-channel-list-fetch.util';
+
+const HANDLE = 'google-calendar-events-import@apple.dev';
+
+describe('Google calendar events import (integration)', () => {
+  const gmail = setupGoogleMock({ handle: HANDLE });
+
+  let channel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    channel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: HANDLE,
+    });
+  }, 60000);
+
+  afterAll(async () => {
+    await channel?.cleanup().catch(() => undefined);
+  });
+
+  it('imports calendar events through the real list-fetch and import pipeline', async () => {
+    const eventTitle = `Calendar event ${randomUUID()}`;
+
+    gmail.serveCalendarEvents([googleCalendarEvent({ summary: eventTitle })]);
+
+    await runCalendarChannelListFetch(channel.calendarChannelId);
+
+    const channelState = await queryCalendarChannel(channel);
+
+    expect(channelState.syncStage).toBe(
+      CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING,
+    );
+
+    await runCalendarChannelEventsImport(channel.calendarChannelId);
+
+    expect(await findImportedCalendarEventTitles([eventTitle])).toEqual([
+      eventTitle,
+    ]);
+  }, 60000);
+
+  it('imports a newly created event through the sync-token incremental fetch', async () => {
+    const newEventTitle = `Calendar event ${randomUUID()}`;
+
+    gmail.serveCalendarEvents(
+      [googleCalendarEvent({ summary: newEventTitle })],
+      { nextSyncToken: 'mock-calendar-sync-token-2' },
+    );
+
+    await runCalendarChannelListFetch(channel.calendarChannelId);
+
+    const channelState = await queryCalendarChannel(channel);
+
+    expect(channelState.syncStage).toBe(
+      CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING,
+    );
+
+    await runCalendarChannelEventsImport(channel.calendarChannelId);
+
+    expect(await findImportedCalendarEventTitles([newEventTitle])).toEqual([
+      newEventTitle,
+    ]);
+  }, 60000);
+});
@@ -0,0 +1,90 @@
+import {
+  CalendarChannelSyncStage,
+  ConnectedAccountProvider,
+} from 'twenty-shared/types';
+
+import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity';
+import { CalendarOngoingStaleCronJob } from 'src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-ongoing-stale.cron.job';
+
+import { setupGoogleMock } from 'test/integration/google/mocks/setup-google-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import { getCoreRepository } from 'test/integration/utils/get-core-repository.util';
+import { queryCalendarChannel } from 'test/integration/utils/query-messaging.util';
+import { runSyncCron } from 'test/integration/utils/run-sync-cron.util';
+
+const STALE_HANDLE = 'calendar-stale-sync@apple.dev';
+const RECENT_HANDLE = 'calendar-recent-sync@apple.dev';
+
+const STALE_STARTED_AT = new Date(Date.now() - 61 * 60 * 1000);
+const RECENT_STARTED_AT = new Date(Date.now() - 60 * 1000);
+
+describe('Calendar stale-sync recovery (integration)', () => {
+  const gmail = setupGoogleMock({ handle: STALE_HANDLE });
+
+  let staleChannel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+  let recentChannel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    staleChannel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: STALE_HANDLE,
+    });
+
+    gmail.actAsAccount(RECENT_HANDLE);
+
+    recentChannel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: RECENT_HANDLE,
+    });
+
+    for (const connectedChannel of [staleChannel, recentChannel]) {
+      const channelState = await queryCalendarChannel(connectedChannel);
+
+      expect(channelState.syncStage).toBe(
+        CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING,
+      );
+    }
+  }, 120000);
+
+  afterAll(async () => {
+    await staleChannel?.cleanup().catch(() => undefined);
+    await recentChannel?.cleanup().catch(() => undefined);
+  });
+
+  it('resets a stale ongoing channel to pending and leaves a recent one running', async () => {
+    const calendarChannelRepository = getCoreRepository<CalendarChannelEntity>(
+      CalendarChannelEntity,
+    );
+
+    await calendarChannelRepository.update(
+      { id: staleChannel.calendarChannelId },
+      {
+        syncStage: CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_ONGOING,
+        syncStageStartedAt: STALE_STARTED_AT,
+      },
+    );
+
+    await calendarChannelRepository.update(
+      { id: recentChannel.calendarChannelId },
+      {
+        syncStage: CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_ONGOING,
+        syncStageStartedAt: RECENT_STARTED_AT,
+      },
+    );
+
+    await runSyncCron(CalendarOngoingStaleCronJob);
+
+    const staleChannelAfter = await queryCalendarChannel(staleChannel);
+
+    expect(staleChannelAfter.syncStage).toBe(
+      CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING,
+    );
+    expect(staleChannelAfter.syncStageStartedAt).toBeNull();
+
+    const recentChannelAfter = await queryCalendarChannel(recentChannel);
+
+    expect(recentChannelAfter.syncStage).toBe(
+      CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_ONGOING,
+    );
+  }, 60000);
+});
@@ -0,0 +1,153 @@
+import {
+  CalendarChannelSyncStage,
+  CalendarChannelSyncStatus,
+  ConnectedAccountProvider,
+} from 'twenty-shared/types';
+
+import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity';
+import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
+import { CALENDAR_THROTTLE_MAX_ATTEMPTS } from 'src/modules/calendar/calendar-event-import-manager/constants/calendar-throttle-max-attempts';
+import { CalendarEventListFetchCronJob } from 'src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-event-list-fetch.cron.job';
+import { CalendarRelaunchFailedCalendarChannelsCronJob } from 'src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-relaunch-failed-calendar-channels.cron.job';
+
+import { setupGoogleMock } from 'test/integration/google/mocks/setup-google-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import { getCoreRepository } from 'test/integration/utils/get-core-repository.util';
+import {
+  queryCalendarChannel,
+  queryConnectedAccount,
+  updateCalendarChannel,
+} from 'test/integration/utils/query-messaging.util';
+import { runSyncCron } from 'test/integration/utils/run-sync-cron.util';
+
+const THROTTLED_HANDLE = 'calendar-throttled@apple.dev';
+const REVOKED_HANDLE = 'calendar-revoked@apple.dev';
+
+const ONE_HOUR_AGO = new Date(Date.now() - 60 * 60 * 1000);
+const EXPIRED_CREDENTIALS_AT = new Date(Date.now() - 56 * 60 * 1000);
+
+describe('Calendar sync failure lifecycle (integration)', () => {
+  const gmail = setupGoogleMock({ handle: THROTTLED_HANDLE });
+
+  let throttledChannel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+  let revokedChannel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    throttledChannel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: THROTTLED_HANDLE,
+    });
+
+    gmail.actAsAccount(REVOKED_HANDLE);
+
+    revokedChannel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: REVOKED_HANDLE,
+    });
+
+    await updateCalendarChannel(revokedChannel.calendarChannelId, {
+      isSyncEnabled: false,
+    });
+  }, 120000);
+
+  afterAll(async () => {
+    await throttledChannel?.cleanup().catch(() => undefined);
+    await revokedChannel?.cleanup().catch(() => undefined);
+  });
+
+  it('counts a throttle failure on a 429 and keeps the channel alive', async () => {
+    gmail.rateLimitCalendarEventList();
+
+    await runSyncCron(CalendarEventListFetchCronJob);
+
+    const channelState = await queryCalendarChannel(throttledChannel);
+
+    expect(channelState.throttleFailureCount).toBe(1);
+    expect(channelState.syncStatus).not.toBe(
+      CalendarChannelSyncStatus.FAILED_UNKNOWN,
+    );
+  }, 60000);
+
+  it('fails the channel as unknown once the throttle attempts are exhausted', async () => {
+    gmail.rateLimitCalendarEventList();
+
+    const calendarChannelRepository = getCoreRepository<CalendarChannelEntity>(
+      CalendarChannelEntity,
+    );
+
+    let channelState = await queryCalendarChannel(throttledChannel);
+
+    for (
+      let attempt = channelState.throttleFailureCount;
+      attempt <= CALENDAR_THROTTLE_MAX_ATTEMPTS &&
+      channelState.syncStatus !== CalendarChannelSyncStatus.FAILED_UNKNOWN;
+      attempt++
+    ) {
+      await calendarChannelRepository.update(
+        { id: throttledChannel.calendarChannelId },
+        { syncStageStartedAt: ONE_HOUR_AGO },
+      );
+
+      await runSyncCron(CalendarEventListFetchCronJob);
+
+      channelState = await queryCalendarChannel(throttledChannel);
+    }
+
+    expect(channelState.syncStatus).toBe(
+      CalendarChannelSyncStatus.FAILED_UNKNOWN,
+    );
+    expect(channelState.syncStage).toBe(CalendarChannelSyncStage.FAILED);
+  }, 120000);
+
+  it('fails the channel as insufficient-permissions when the refresh token is declined', async () => {
+    await updateCalendarChannel(revokedChannel.calendarChannelId, {
+      isSyncEnabled: true,
+    });
+
+    await getCoreRepository<ConnectedAccountEntity>(
+      ConnectedAccountEntity,
+    ).update(
+      { id: revokedChannel.connectedAccountId },
+      { lastCredentialsRefreshedAt: EXPIRED_CREDENTIALS_AT },
+    );
+
+    gmail.declineTokenRefresh();
+
+    await runSyncCron(CalendarEventListFetchCronJob);
+
+    const channelState = await queryCalendarChannel(revokedChannel);
+
+    expect(channelState.syncStatus).toBe(
+      CalendarChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS,
+    );
+    expect(channelState.syncStage).toBe(CalendarChannelSyncStage.FAILED);
+
+    const account = await queryConnectedAccount(
+      revokedChannel.connectedAccountId,
+    );
+
+    expect(account.authFailedAt).not.toBeNull();
+  }, 60000);
+
+  it('relaunches the unknown-failure channel and leaves the permissions-failure channel untouched', async () => {
+    await runSyncCron(CalendarRelaunchFailedCalendarChannelsCronJob);
+
+    const relaunchedChannel = await queryCalendarChannel(throttledChannel);
+
+    expect(relaunchedChannel.syncStatus).toBe(
+      CalendarChannelSyncStatus.ACTIVE,
+    );
+    expect(relaunchedChannel.syncStage).toBe(
+      CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING,
+    );
+    expect(relaunchedChannel.throttleFailureCount).toBe(0);
+    expect(relaunchedChannel.syncStageStartedAt).toBeNull();
+
+    const untouchedChannel = await queryCalendarChannel(revokedChannel);
+
+    expect(untouchedChannel.syncStatus).toBe(
+      CalendarChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS,
+    );
+    expect(untouchedChannel.syncStage).toBe(CalendarChannelSyncStage.FAILED);
+  }, 60000);
+});
@@ -0,0 +1,120 @@
+import { ConnectedAccountProvider } from 'twenty-shared/types';
+import { isDefined } from 'twenty-shared/utils';
+
+import { gmailMessage } from 'test/integration/google/mocks/gmail-message.util';
+import { setupGoogleMock } from 'test/integration/google/mocks/setup-google-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import {
+  findRecordIdsByFilter,
+  findRecordNodesByFilter,
+} from 'test/integration/utils/find-records-by-filter.util';
+import { deleteConnectedAccount } from 'test/integration/utils/query-messaging.util';
+import { runMessageChannelSync } from 'test/integration/utils/run-message-channel-sync.util';
+
+const HANDLE = 'messaging-cleanup@apple.dev';
+
+describe('Messaging connected account cleanup (integration)', () => {
+  const inbox = [gmailMessage(), gmailMessage()];
+
+  setupGoogleMock({ handle: HANDLE, inbox });
+
+  let channel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    channel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: HANDLE,
+    });
+
+    await runMessageChannelSync(channel.channelId);
+  }, 60000);
+
+  afterAll(async () => {
+    await channel?.cleanup().catch(() => undefined);
+  });
+
+  it('deletes all associated messaging data when the connected account is removed', async () => {
+    const associations = await findRecordNodesByFilter<{
+      id: string;
+      messageId: string;
+    }>(
+      'messageChannelMessageAssociation',
+      'messageChannelMessageAssociations',
+      `id
+        messageId`,
+      { messageChannelId: { eq: channel.channelId } },
+    );
+
+    expect(associations).toHaveLength(inbox.length);
+
+    const associationIds = associations.map((association) => association.id);
+    const messageIds = associations.map((association) => association.messageId);
+
+    const messages = await findRecordNodesByFilter<{
+      id: string;
+      messageThreadId: string | null;
+    }>(
+      'message',
+      'messages',
+      `id
+        messageThreadId`,
+      { id: { in: messageIds } },
+    );
+
+    expect(messages).toHaveLength(inbox.length);
+
+    const threadIds = [
+      ...new Set(messages.map((message) => message.messageThreadId)),
+    ].filter(isDefined);
+
+    expect(
+      await findRecordIdsByFilter(
+        'messageChannelMessageAssociationMessageFolder',
+        'messageChannelMessageAssociationMessageFolders',
+        { messageChannelMessageAssociationId: { in: associationIds } },
+      ),
+    ).not.toHaveLength(0);
+    expect(
+      await findRecordIdsByFilter('messageParticipant', 'messageParticipants', {
+        messageId: { in: messageIds },
+      }),
+    ).not.toHaveLength(0);
+    expect(
+      await findRecordIdsByFilter('messageThread', 'messageThreads', {
+        id: { in: threadIds },
+      }),
+    ).not.toHaveLength(0);
+
+    await deleteConnectedAccount(channel.connectedAccountId);
+
+    expect(
+      await findRecordIdsByFilter('message', 'messages', {
+        id: { in: messageIds },
+      }),
+    ).toHaveLength(0);
+    expect(
+      await findRecordIdsByFilter(
+        'messageChannelMessageAssociation',
+        'messageChannelMessageAssociations',
+        { messageChannelId: { eq: channel.channelId } },
+      ),
+    ).toHaveLength(0);
+    expect(
+      await findRecordIdsByFilter(
+        'messageChannelMessageAssociationMessageFolder',
+        'messageChannelMessageAssociationMessageFolders',
+        { messageChannelMessageAssociationId: { in: associationIds } },
+      ),
+    ).toHaveLength(0);
+    expect(
+      await findRecordIdsByFilter('messageParticipant', 'messageParticipants', {
+        messageId: { in: messageIds },
+      }),
+    ).toHaveLength(0);
+    expect(
+      await findRecordIdsByFilter('messageThread', 'messageThreads', {
+        id: { in: threadIds },
+      }),
+    ).toHaveLength(0);
+  }, 60000);
+});
@@ -0,0 +1,75 @@
+import {
+  ConnectedAccountProvider,
+  MessageFolderImportPolicy,
+} from 'twenty-shared/types';
+
+import { setupGoogleMock } from 'test/integration/google/mocks/setup-google-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import {
+  queryMessageFolders,
+  updateMessageChannel,
+} from 'test/integration/utils/query-messaging.util';
+import { runMessageChannelSync } from 'test/integration/utils/run-message-channel-sync.util';
+import { startChannelSyncAndAwait } from 'test/integration/utils/start-channel-sync-and-await.util';
+
+const HANDLE = 'gmail-folder-discovery@apple.dev';
+
+describe('Gmail folder discovery (integration)', () => {
+  const gmail = setupGoogleMock({
+    handle: HANDLE,
+    labels: [
+      { id: 'INBOX', name: 'INBOX' },
+      { id: 'SENT', name: 'SENT' },
+      { id: 'Label_Work', name: 'Work' },
+    ],
+  });
+
+  let channel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    channel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: HANDLE,
+      skipChannelConfiguration: false,
+    });
+  }, 60000);
+
+  afterAll(async () => {
+    await channel?.cleanup().catch(() => undefined);
+  });
+
+  it('syncs every folder discovered at connect time under the default all-folders policy', async () => {
+    await startChannelSyncAndAwait(channel.connectedAccountId);
+
+    const folders = await queryMessageFolders(channel.channelId);
+
+    expect(
+      Object.fromEntries(folders.map((folder) => [folder.name, folder.isSynced])),
+    ).toEqual({
+      INBOX: true,
+      SENT: true,
+      Work: true,
+    });
+  }, 60000);
+
+  it('leaves a folder discovered after switching to selected-folders unsynced', async () => {
+    await updateMessageChannel(channel.channelId, {
+      messageFolderImportPolicy: MessageFolderImportPolicy.SELECTED_FOLDERS,
+    });
+
+    gmail.labels.add({ id: 'Label_Archive', name: 'Archive' });
+
+    await runMessageChannelSync(channel.channelId);
+
+    const folders = await queryMessageFolders(channel.channelId);
+
+    expect(
+      Object.fromEntries(folders.map((folder) => [folder.name, folder.isSynced])),
+    ).toEqual({
+      INBOX: true,
+      SENT: true,
+      Work: true,
+      Archive: false,
+    });
+  }, 60000);
+});
@@ -0,0 +1,60 @@
+import { ConnectedAccountProvider } from 'twenty-shared/types';
+
+import { getGmailMessageSubject } from 'test/integration/google/mocks/gmail-message-subject.util';
+import { gmailMessage } from 'test/integration/google/mocks/gmail-message.util';
+import { setupGoogleMock } from 'test/integration/google/mocks/setup-google-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import { findImportedMessageSubjects } from 'test/integration/utils/find-imported-records.util';
+import { runMessageChannelSync } from 'test/integration/utils/run-message-channel-sync.util';
+
+const HANDLE = 'messaging-incremental-sync@apple.dev';
+
+describe('Messaging incremental sync (integration)', () => {
+  const inbox = [gmailMessage()];
+
+  const gmail = setupGoogleMock({ handle: HANDLE, inbox });
+
+  let channel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    channel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: HANDLE,
+    });
+  }, 60000);
+
+  afterAll(async () => {
+    await channel?.cleanup().catch(() => undefined);
+  });
+
+  it('imports the initial inbox through a full sync', async () => {
+    await runMessageChannelSync(channel.channelId);
+
+    const initialSubject = getGmailMessageSubject(inbox[0]);
+
+    expect(await findImportedMessageSubjects([initialSubject])).toEqual([
+      initialSubject,
+    ]);
+  }, 60000);
+
+  it('imports a newly arrived message through the history-based incremental sync', async () => {
+    const initialMessages = [...inbox];
+    const newMessage = gmailMessage();
+
+    inbox.push(newMessage);
+
+    // The list endpoint keeps serving only the initial inbox: the new message
+    // is reachable through the history endpoint alone, so a regression that
+    // re-runs a full list fetch instead of the history-based sync cannot pass.
+    gmail.serveMessageList(initialMessages);
+    gmail.serveHistory([newMessage]);
+
+    await runMessageChannelSync(channel.channelId);
+
+    const newSubject = getGmailMessageSubject(newMessage);
+
+    expect(await findImportedMessageSubjects([newSubject])).toEqual([
+      newSubject,
+    ]);
+  }, 60000);
+});
@@ -0,0 +1,47 @@
+import { ConnectedAccountProvider } from 'twenty-shared/types';
+
+import { getGmailMessageSubject } from 'test/integration/google/mocks/gmail-message-subject.util';
+import { gmailMessage } from 'test/integration/google/mocks/gmail-message.util';
+import { setupGoogleMock } from 'test/integration/google/mocks/setup-google-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import { findImportedMessageSubjects } from 'test/integration/utils/find-imported-records.util';
+import { queryMessageFolders } from 'test/integration/utils/query-messaging.util';
+import { runMessageChannelSync } from 'test/integration/utils/run-message-channel-sync.util';
+
+const HANDLE = 'gmail-message-list-fetch@apple.dev';
+
+describe('Gmail message list fetch (integration)', () => {
+  const inbox = [gmailMessage(), gmailMessage()];
+
+  setupGoogleMock({ handle: HANDLE, inbox });
+
+  let channel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    channel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: HANDLE,
+    });
+  }, 60000);
+
+  afterAll(async () => {
+    await channel?.cleanup().catch(() => undefined);
+  });
+
+  it('runs the full sync pipeline on the real worker: folders synced, messages imported', async () => {
+    await runMessageChannelSync(channel.channelId);
+
+    const expectedSubjects = inbox.map(getGmailMessageSubject);
+
+    expect(await findImportedMessageSubjects(expectedSubjects)).toEqual(
+      [...expectedSubjects].sort(),
+    );
+
+    const folders = await queryMessageFolders(channel.channelId);
+
+    expect(folders.map((folder) => folder.name).sort()).toEqual([
+      'INBOX',
+      'SENT',
+    ]);
+  }, 60000);
+});
@@ -0,0 +1,89 @@
+import {
+  ConnectedAccountProvider,
+  MessageChannelSyncStage,
+} from 'twenty-shared/types';
+
+import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
+import { MessagingOngoingStaleCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-ongoing-stale.cron.job';
+
+import { setupGoogleMock } from 'test/integration/google/mocks/setup-google-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import { getCoreRepository } from 'test/integration/utils/get-core-repository.util';
+import { queryMessageChannel } from 'test/integration/utils/query-messaging.util';
+import { runSyncCron } from 'test/integration/utils/run-sync-cron.util';
+
+const STALE_HANDLE = 'messaging-stale-sync@apple.dev';
+const RECENT_HANDLE = 'messaging-recent-sync@apple.dev';
+
+const STALE_STARTED_AT = new Date(Date.now() - 31 * 60 * 1000);
+const RECENT_STARTED_AT = new Date(Date.now() - 60 * 1000);
+
+describe('Messaging stale-sync recovery (integration)', () => {
+  const gmail = setupGoogleMock({ handle: STALE_HANDLE });
+
+  let staleChannel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+  let recentChannel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    staleChannel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: STALE_HANDLE,
+    });
+
+    gmail.actAsAccount(RECENT_HANDLE);
+
+    recentChannel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: RECENT_HANDLE,
+    });
+
+    for (const connectedChannel of [staleChannel, recentChannel]) {
+      const channelState = await queryMessageChannel(connectedChannel);
+
+      expect(channelState.syncStage).toBe(
+        MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+      );
+    }
+  }, 120000);
+
+  afterAll(async () => {
+    await staleChannel?.cleanup().catch(() => undefined);
+    await recentChannel?.cleanup().catch(() => undefined);
+  });
+
+  it('resets a stale ongoing channel to pending and leaves a recent one running', async () => {
+    const messageChannelRepository =
+      getCoreRepository<MessageChannelEntity>(MessageChannelEntity);
+
+    await messageChannelRepository.update(
+      { id: staleChannel.channelId },
+      {
+        syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING,
+        syncStageStartedAt: STALE_STARTED_AT,
+      },
+    );
+
+    await messageChannelRepository.update(
+      { id: recentChannel.channelId },
+      {
+        syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING,
+        syncStageStartedAt: RECENT_STARTED_AT,
+      },
+    );
+
+    await runSyncCron(MessagingOngoingStaleCronJob);
+
+    const staleChannelAfter = await queryMessageChannel(staleChannel);
+
+    expect(staleChannelAfter.syncStage).toBe(
+      MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+    );
+    expect(staleChannelAfter.syncStageStartedAt).toBeNull();
+
+    const recentChannelAfter = await queryMessageChannel(recentChannel);
+
+    expect(recentChannelAfter.syncStage).toBe(
+      MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING,
+    );
+  }, 60000);
+});
@@ -0,0 +1,156 @@
+import {
+  ConnectedAccountProvider,
+  MessageChannelSyncStage,
+  MessageChannelSyncStatus,
+} from 'twenty-shared/types';
+
+import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
+import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
+import { MESSAGING_THROTTLE_MAX_ATTEMPTS } from 'src/modules/messaging/message-import-manager/constants/messaging-throttle-max-attempts';
+import { MessagingMessageListFetchCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job';
+import { MessagingRelaunchFailedMessageChannelsCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-relaunch-failed-message-channels.cron.job';
+
+import { setupGoogleMock } from 'test/integration/google/mocks/setup-google-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import { getCoreRepository } from 'test/integration/utils/get-core-repository.util';
+import {
+  queryConnectedAccount,
+  queryMessageChannel,
+  updateMessageChannel,
+} from 'test/integration/utils/query-messaging.util';
+import { runSyncCron } from 'test/integration/utils/run-sync-cron.util';
+
+const THROTTLED_HANDLE = 'messaging-throttled@apple.dev';
+const REVOKED_HANDLE = 'messaging-revoked@apple.dev';
+
+const FUTURE_RETRY_AFTER_ISO = '2099-12-31T10:30:00.000Z';
+
+const ONE_HOUR_AGO = new Date(Date.now() - 60 * 60 * 1000);
+const EXPIRED_CREDENTIALS_AT = new Date(Date.now() - 56 * 60 * 1000);
+
+describe('Messaging sync failure lifecycle (integration)', () => {
+  const gmail = setupGoogleMock({ handle: THROTTLED_HANDLE });
+
+  let throttledChannel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+  let revokedChannel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    throttledChannel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: THROTTLED_HANDLE,
+    });
+
+    gmail.actAsAccount(REVOKED_HANDLE);
+
+    revokedChannel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: REVOKED_HANDLE,
+    });
+
+    await updateMessageChannel(revokedChannel.channelId, {
+      isSyncEnabled: false,
+    });
+  }, 120000);
+
+  afterAll(async () => {
+    await throttledChannel?.cleanup().catch(() => undefined);
+    await revokedChannel?.cleanup().catch(() => undefined);
+  });
+
+  it('records the throttle backoff on a 429 and keeps the channel alive', async () => {
+    gmail.rateLimitMessageList(FUTURE_RETRY_AFTER_ISO);
+
+    await runSyncCron(MessagingMessageListFetchCronJob);
+
+    const channelState = await queryMessageChannel(throttledChannel);
+
+    expect(channelState.throttleFailureCount).toBe(1);
+    expect(channelState.throttleRetryAfter).toBe(FUTURE_RETRY_AFTER_ISO);
+    expect(channelState.syncStatus).not.toBe(
+      MessageChannelSyncStatus.FAILED_UNKNOWN,
+    );
+  }, 60000);
+
+  it('fails the channel as unknown once the throttle attempts are exhausted', async () => {
+    gmail.rateLimitMessageList(FUTURE_RETRY_AFTER_ISO);
+
+    const messageChannelRepository =
+      getCoreRepository<MessageChannelEntity>(MessageChannelEntity);
+
+    let channelState = await queryMessageChannel(throttledChannel);
+
+    for (
+      let attempt = channelState.throttleFailureCount;
+      attempt <= MESSAGING_THROTTLE_MAX_ATTEMPTS &&
+      channelState.syncStatus !== MessageChannelSyncStatus.FAILED_UNKNOWN;
+      attempt++
+    ) {
+      await messageChannelRepository.update(
+        { id: throttledChannel.channelId },
+        { syncStageStartedAt: ONE_HOUR_AGO, throttleRetryAfter: null },
+      );
+
+      await runSyncCron(MessagingMessageListFetchCronJob);
+
+      channelState = await queryMessageChannel(throttledChannel);
+    }
+
+    expect(channelState.syncStatus).toBe(
+      MessageChannelSyncStatus.FAILED_UNKNOWN,
+    );
+    expect(channelState.syncStage).toBe(MessageChannelSyncStage.FAILED);
+  }, 120000);
+
+  it('fails the channel as insufficient-permissions when the refresh token is declined', async () => {
+    gmail.actAsAccount(REVOKED_HANDLE);
+
+    await updateMessageChannel(revokedChannel.channelId, {
+      isSyncEnabled: true,
+    });
+
+    await getCoreRepository<ConnectedAccountEntity>(
+      ConnectedAccountEntity,
+    ).update(
+      { id: revokedChannel.connectedAccountId },
+      { lastCredentialsRefreshedAt: EXPIRED_CREDENTIALS_AT },
+    );
+
+    gmail.declineTokenRefresh();
+
+    await runSyncCron(MessagingMessageListFetchCronJob);
+
+    const channelState = await queryMessageChannel(revokedChannel);
+
+    expect(channelState.syncStatus).toBe(
+      MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS,
+    );
+    expect(channelState.syncStage).toBe(MessageChannelSyncStage.FAILED);
+
+    const account = await queryConnectedAccount(
+      revokedChannel.connectedAccountId,
+    );
+
+    expect(account.authFailedAt).not.toBeNull();
+  }, 60000);
+
+  it('relaunches the unknown-failure channel and leaves the permissions-failure channel untouched', async () => {
+    await runSyncCron(MessagingRelaunchFailedMessageChannelsCronJob);
+
+    const relaunchedChannel = await queryMessageChannel(throttledChannel);
+
+    expect(relaunchedChannel.syncStatus).toBe(MessageChannelSyncStatus.ACTIVE);
+    expect(relaunchedChannel.syncStage).toBe(
+      MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+    );
+    expect(relaunchedChannel.throttleFailureCount).toBe(0);
+    expect(relaunchedChannel.throttleRetryAfter).toBeNull();
+    expect(relaunchedChannel.syncStageStartedAt).toBeNull();
+
+    const untouchedChannel = await queryMessageChannel(revokedChannel);
+
+    expect(untouchedChannel.syncStatus).toBe(
+      MessageChannelSyncStatus.FAILED_INSUFFICIENT_PERMISSIONS,
+    );
+    expect(untouchedChannel.syncStage).toBe(MessageChannelSyncStage.FAILED);
+  }, 60000);
+});
@@ -0,0 +1,49 @@
+import { ConnectedAccountProvider } from 'twenty-shared/types';
+
+import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
+
+import { gmailMessage } from 'test/integration/google/mocks/gmail-message.util';
+import { setupGoogleMock } from 'test/integration/google/mocks/setup-google-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import { getCoreRepository } from 'test/integration/utils/get-core-repository.util';
+import { queryConnectedAccount } from 'test/integration/utils/query-messaging.util';
+import { runMessageChannelSync } from 'test/integration/utils/run-message-channel-sync.util';
+
+const HANDLE = 'messaging-token-refresh@apple.dev';
+
+const EXPIRED_CREDENTIALS_AT = new Date(Date.now() - 56 * 60 * 1000);
+
+describe('Messaging token refresh (integration)', () => {
+  setupGoogleMock({ handle: HANDLE, inbox: [gmailMessage()] });
+
+  let channel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    channel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.GOOGLE,
+      handle: HANDLE,
+    });
+  }, 60000);
+
+  afterAll(async () => {
+    await channel?.cleanup().catch(() => undefined);
+  });
+
+  it('refreshes expired credentials and completes the sync', async () => {
+    await getCoreRepository<ConnectedAccountEntity>(
+      ConnectedAccountEntity,
+    ).update(
+      { id: channel.connectedAccountId },
+      { lastCredentialsRefreshedAt: EXPIRED_CREDENTIALS_AT },
+    );
+
+    await runMessageChannelSync(channel.channelId);
+
+    const account = await queryConnectedAccount(channel.connectedAccountId);
+
+    expect(account.authFailedAt).toBeNull();
+    expect(
+      new Date(account.lastCredentialsRefreshedAt ?? 0).getTime(),
+    ).toBeGreaterThan(EXPIRED_CREDENTIALS_AT.getTime());
+  }, 60000);
+});
@@ -0,0 +1,20 @@
+import { type gmail_v1 } from 'googleapis';
+import { http, HttpResponse } from 'msw';
+
+import { type MswHandler } from 'test/integration/utils/http-mock.util';
+
+export const gmailHistoryHandler = (
+  addedMessages: gmail_v1.Schema$Message[],
+): MswHandler =>
+  http.get('*/gmail/v1/users/me/history', () =>
+    HttpResponse.json<gmail_v1.Schema$ListHistoryResponse>({
+      history: [
+        {
+          messagesAdded: addedMessages.map((message) => ({
+            message: { id: message.id, threadId: message.threadId },
+          })),
+        },
+      ],
+      historyId: '987654322',
+    }),
+  );
@@ -0,0 +1,77 @@
+import { type gmail_v1 } from 'googleapis';
+import { http, HttpResponse } from 'msw';
+
+import { gmailMessageListHandler } from 'test/integration/google/mocks/gmail-message-list-handler.util';
+import { type MswHandler } from 'test/integration/utils/http-mock.util';
+import { type MockEntityStore } from 'test/integration/utils/mock-entity-store.util';
+
+const buildBatchMultipartResponse = (
+  messages: gmail_v1.Schema$Message[],
+): { body: string; contentType: string } => {
+  const boundary = 'batch_boundary';
+  const subResponses = messages
+    .map((message) =>
+      [
+        `--${boundary}`,
+        'Content-Type: application/http',
+        '',
+        'HTTP/1.1 200 OK',
+        'Content-Type: application/json; charset=UTF-8',
+        '',
+        JSON.stringify(message),
+      ].join('\r\n'),
+    )
+    .join('\r\n');
+
+  return {
+    body: `${subResponses}\r\n--${boundary}--`,
+    contentType: `multipart/mixed; boundary=${boundary}`,
+  };
+};
+
+export const gmailMailboxHandlers = (
+  inbox: gmail_v1.Schema$Message[],
+  labelStore: MockEntityStore<gmail_v1.Schema$Label>,
+): MswHandler[] => [
+  http.get('*/gmail/v1/users/me/labels', () =>
+    HttpResponse.json<gmail_v1.Schema$ListLabelsResponse>({
+      labels: labelStore.list(),
+    }),
+  ),
+  gmailMessageListHandler(inbox),
+  http.get('*/gmail/v1/users/me/history', () =>
+    HttpResponse.json<gmail_v1.Schema$ListHistoryResponse>({
+      history: [],
+      historyId: inbox[0]?.historyId ?? '987654321',
+    }),
+  ),
+  http.get('*/gmail/v1/users/me/messages/:messageId', ({ params }) => {
+    const message = inbox.find(
+      (candidate) => candidate.id === params.messageId,
+    );
+
+    if (!message) {
+      return HttpResponse.json(
+        { error: { code: 404, message: 'Not Found' } },
+        { status: 404 },
+      );
+    }
+
+    return HttpResponse.json<gmail_v1.Schema$Message>(message);
+  }),
+  http.post('*/batch', async ({ request }) => {
+    const requestedIds = [
+      ...(await request.text()).matchAll(/messages\/([\w-]+)/g),
+    ].map((match) => match[1]);
+    const requestedMessages = inbox.filter((message) =>
+      requestedIds.includes(message.id ?? ''),
+    );
+
+    const { body, contentType } =
+      buildBatchMultipartResponse(requestedMessages);
+
+    return new HttpResponse(body, {
+      headers: { 'Content-Type': contentType },
+    });
+  }),
+];
@@ -0,0 +1,17 @@
+import { type gmail_v1 } from 'googleapis';
+import { http, HttpResponse } from 'msw';
+
+import { type MswHandler } from 'test/integration/utils/http-mock.util';
+
+export const gmailMessageListHandler = (
+  messages: gmail_v1.Schema$Message[],
+): MswHandler =>
+  http.get('*/gmail/v1/users/me/messages', () =>
+    HttpResponse.json<gmail_v1.Schema$ListMessagesResponse>({
+      messages: messages.map((message) => ({
+        id: message.id,
+        threadId: message.threadId,
+      })),
+      resultSizeEstimate: messages.length,
+    }),
+  );
@@ -0,0 +1,7 @@
+import { type gmail_v1 } from 'googleapis';
+
+export const getGmailMessageSubject = (
+  message: gmail_v1.Schema$Message,
+): string =>
+  message.payload?.headers?.find((header) => header.name === 'Subject')
+    ?.value ?? '';
@@ -0,0 +1,32 @@
+import { randomUUID } from 'node:crypto';
+
+import { type gmail_v1 } from 'googleapis';
+
+export const gmailMessage = (
+  overrides: Partial<gmail_v1.Schema$Message> = {},
+): gmail_v1.Schema$Message => {
+  const id = overrides.id ?? `gmail-msg-${randomUUID()}`;
+
+  return {
+    id,
+    threadId: id,
+    historyId: '987654321',
+    internalDate: '1700000000000',
+    labelIds: ['INBOX'],
+    payload: {
+      mimeType: 'text/plain',
+      headers: [
+        { name: 'From', value: `sender-${id}@example.com` },
+        { name: 'To', value: `recipient-${id}@example.com` },
+        { name: 'Subject', value: `Subject ${id}` },
+        { name: 'Message-ID', value: `<${id}@example.com>` },
+        { name: 'Date', value: 'Wed, 15 Nov 2023 00:00:00 +0000' },
+      ],
+      body: {
+        data: Buffer.from(`body ${id}`).toString('base64'),
+        size: 10,
+      },
+    },
+    ...overrides,
+  };
+};
@@ -0,0 +1,22 @@
+import { randomUUID } from 'node:crypto';
+
+import { type calendar_v3 } from 'googleapis';
+
+export const googleCalendarEvent = (
+  overrides: Partial<calendar_v3.Schema$Event> = {},
+): calendar_v3.Schema$Event => {
+  const id = overrides.id ?? `google-calendar-event-${randomUUID()}`;
+
+  return {
+    id,
+    iCalUID: `${id}@google.com`,
+    summary: `Calendar event ${id}`,
+    status: 'confirmed',
+    start: { dateTime: '2023-11-15T10:00:00Z' },
+    end: { dateTime: '2023-11-15T11:00:00Z' },
+    created: '2023-11-01T00:00:00.000Z',
+    updated: '2023-11-01T00:00:00.000Z',
+    attendees: [],
+    ...overrides,
+  };
+};
@@ -0,0 +1,22 @@
+import { type calendar_v3 } from 'googleapis';
+import { http, HttpResponse } from 'msw';
+
+import { GOOGLE_CALENDAR_EVENTS_URL } from 'test/integration/google/mocks/google-calendar-events-url.constant';
+import { type MswHandler } from 'test/integration/utils/http-mock.util';
+
+export const googleCalendarEventsHandlers = (
+  events: calendar_v3.Schema$Event[],
+  nextSyncToken: string,
+): MswHandler[] => [
+  http.get(GOOGLE_CALENDAR_EVENTS_URL, () =>
+    HttpResponse.json<calendar_v3.Schema$Events>({
+      items: events,
+      nextSyncToken,
+    }),
+  ),
+  ...events.map((event) =>
+    http.get(`${GOOGLE_CALENDAR_EVENTS_URL}/${event.id}`, () =>
+      HttpResponse.json<calendar_v3.Schema$Event>(event),
+    ),
+  ),
+];
@@ -0,0 +1,2 @@
+export const GOOGLE_CALENDAR_EVENTS_URL =
+  'https://www.googleapis.com/calendar/v3/calendars/primary/events';
@@ -0,0 +1,26 @@
+import { http, HttpResponse } from 'msw';
+
+import { GOOGLE_OAUTH_SCOPES } from 'test/integration/google/mocks/google-oauth-scopes.constant';
+import { type MswHandler } from 'test/integration/utils/http-mock.util';
+
+export const googleIdentityHandlers = (handle: string): MswHandler[] => [
+  http.get('https://www.googleapis.com/oauth2/v3/userinfo', () =>
+    HttpResponse.json({
+      sub: `google-user-id-${handle}`,
+      email: handle,
+      email_verified: true,
+      name: 'Jane Austen',
+      given_name: 'Jane',
+      family_name: 'Austen',
+    }),
+  ),
+  http.get('https://www.googleapis.com/oauth2/v3/tokeninfo', () =>
+    HttpResponse.json({ scope: GOOGLE_OAUTH_SCOPES, email: handle }),
+  ),
+  http.get('https://gmail.googleapis.com/gmail/v1/users/me/profile', () =>
+    HttpResponse.json({ emailAddress: handle, messagesTotal: 0 }),
+  ),
+  http.get('*/gmail/v1/users/me/settings/sendAs', () =>
+    HttpResponse.json({ sendAs: [{ sendAsEmail: handle, isPrimary: true }] }),
+  ),
+];
@@ -0,0 +1,9 @@
+export const GOOGLE_OAUTH_SCOPES = [
+  'email',
+  'profile',
+  'https://www.googleapis.com/auth/gmail.readonly',
+  'https://www.googleapis.com/auth/calendar.events',
+  'https://www.googleapis.com/auth/profile.emails.read',
+  'https://www.googleapis.com/auth/gmail.send',
+  'https://www.googleapis.com/auth/gmail.compose',
+].join(' ');
@@ -0,0 +1,22 @@
+import { http, HttpResponse } from 'msw';
+
+import { GOOGLE_OAUTH_SCOPES } from 'test/integration/google/mocks/google-oauth-scopes.constant';
+import { type MswHandler } from 'test/integration/utils/http-mock.util';
+
+export const GOOGLE_TOKEN_URLS = [
+  'https://oauth2.googleapis.com/token',
+  'https://www.googleapis.com/oauth2/v4/token',
+];
+
+export const googleTokenHandlers = (): MswHandler[] =>
+  GOOGLE_TOKEN_URLS.map((url) =>
+    http.post(url, () =>
+      HttpResponse.json({
+        access_token: 'mock-access-token',
+        refresh_token: 'mock-refresh-token',
+        expires_in: 3600,
+        scope: GOOGLE_OAUTH_SCOPES,
+        token_type: 'Bearer',
+      }),
+    ),
+  );
@@ -0,0 +1,124 @@
+import { type calendar_v3, type gmail_v1 } from 'googleapis';
+import { http, HttpResponse } from 'msw';
+
+import { gmailHistoryHandler } from 'test/integration/google/mocks/gmail-history-handler.util';
+import { gmailMailboxHandlers } from 'test/integration/google/mocks/gmail-mailbox-handlers.util';
+import { gmailMessageListHandler } from 'test/integration/google/mocks/gmail-message-list-handler.util';
+import { googleCalendarEventsHandlers } from 'test/integration/google/mocks/google-calendar-events-handlers.util';
+import { GOOGLE_CALENDAR_EVENTS_URL } from 'test/integration/google/mocks/google-calendar-events-url.constant';
+import { googleIdentityHandlers } from 'test/integration/google/mocks/google-identity-handlers.util';
+import {
+  GOOGLE_TOKEN_URLS,
+  googleTokenHandlers,
+} from 'test/integration/google/mocks/google-token-handlers.util';
+import { setupHttpMock } from 'test/integration/utils/http-mock.util';
+import {
+  createMockEntityStore,
+  type MockEntityStore,
+} from 'test/integration/utils/mock-entity-store.util';
+
+const DEFAULT_LABELS: gmail_v1.Schema$Label[] = [
+  { id: 'INBOX', name: 'INBOX', type: 'system' },
+  { id: 'SENT', name: 'SENT', type: 'system' },
+];
+
+export type GoogleMock = {
+  labels: MockEntityStore<gmail_v1.Schema$Label>;
+  actAsAccount: (handle: string) => void;
+  serveMessageList: (messages: gmail_v1.Schema$Message[]) => void;
+  serveHistory: (addedMessages: gmail_v1.Schema$Message[]) => void;
+  serveCalendarEvents: (
+    events: calendar_v3.Schema$Event[],
+    options?: { nextSyncToken?: string },
+  ) => void;
+  rateLimitMessageList: (retryAfterIso: string) => void;
+  rateLimitCalendarEventList: () => void;
+  declineTokenRefresh: () => void;
+};
+
+export const setupGoogleMock = ({
+  handle,
+  inbox = [],
+  labels = DEFAULT_LABELS,
+}: {
+  handle: string;
+  inbox?: gmail_v1.Schema$Message[];
+  labels?: gmail_v1.Schema$Label[];
+}): GoogleMock => {
+  const labelStore = createMockEntityStore(labels, (label) => label.id ?? '');
+
+  const httpMock = setupHttpMock(
+    ...googleTokenHandlers(),
+    ...googleIdentityHandlers(handle),
+    ...googleCalendarEventsHandlers([], 'mock-calendar-sync-token'),
+    ...gmailMailboxHandlers(inbox, labelStore),
+  );
+
+  return {
+    labels: labelStore,
+    actAsAccount: (accountHandle) =>
+      httpMock.use(...googleIdentityHandlers(accountHandle)),
+    serveMessageList: (messages) =>
+      httpMock.use(gmailMessageListHandler(messages)),
+    serveHistory: (addedMessages) =>
+      httpMock.use(gmailHistoryHandler(addedMessages)),
+    serveCalendarEvents: (
+      events,
+      { nextSyncToken = 'mock-calendar-sync-token' } = {},
+    ) => httpMock.use(...googleCalendarEventsHandlers(events, nextSyncToken)),
+    rateLimitMessageList: (retryAfterIso) =>
+      httpMock.use(
+        http.get('*/gmail/v1/users/me/messages', () =>
+          HttpResponse.json(
+            {
+              error: {
+                code: 429,
+                message: 'Rate Limit Exceeded',
+                errors: [
+                  {
+                    reason: 'rateLimitExceeded',
+                    message: `Rate Limit Exceeded. Retry after ${retryAfterIso}`,
+                  },
+                ],
+              },
+            },
+            { status: 429 },
+          ),
+        ),
+      ),
+    rateLimitCalendarEventList: () =>
+      httpMock.use(
+        http.get(GOOGLE_CALENDAR_EVENTS_URL, () =>
+          HttpResponse.json(
+            {
+              error: {
+                code: 429,
+                message: 'Rate Limit Exceeded',
+                errors: [
+                  {
+                    reason: 'rateLimitExceeded',
+                    message: 'Rate Limit Exceeded',
+                  },
+                ],
+              },
+            },
+            { status: 429 },
+          ),
+        ),
+      ),
+    declineTokenRefresh: () =>
+      httpMock.use(
+        ...GOOGLE_TOKEN_URLS.map((url) =>
+          http.post(url, () =>
+            HttpResponse.json(
+              {
+                error: 'invalid_grant',
+                error_description: 'Token has been revoked',
+              },
+              { status: 400 },
+            ),
+          ),
+        ),
+      ),
+  };
+};
@@ -0,0 +1,78 @@
+import { randomUUID } from 'node:crypto';
+
+import {
+  CalendarChannelSyncStage,
+  ConnectedAccountProvider,
+} from 'twenty-shared/types';
+
+import { microsoftCalendarEvent } from 'test/integration/microsoft/mocks/microsoft-calendar-event.util';
+import { setupMicrosoftMock } from 'test/integration/microsoft/mocks/setup-microsoft-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import { findImportedCalendarEventTitles } from 'test/integration/utils/find-imported-records.util';
+import { queryCalendarChannel } from 'test/integration/utils/query-messaging.util';
+import { runCalendarChannelEventsImport } from 'test/integration/utils/run-calendar-channel-events-import.util';
+import { runCalendarChannelListFetch } from 'test/integration/utils/run-calendar-channel-list-fetch.util';
+
+const HANDLE = 'microsoft-calendar-events-import@apple.dev';
+
+describe('Microsoft calendar events import (integration)', () => {
+  const microsoft = setupMicrosoftMock({ handle: HANDLE });
+
+  let channel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    channel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.MICROSOFT,
+      handle: HANDLE,
+    });
+  }, 60000);
+
+  afterAll(async () => {
+    await channel?.cleanup().catch(() => undefined);
+  });
+
+  it('imports calendar events through the real delta-fetch and import pipeline', async () => {
+    const eventTitle = `Calendar event ${randomUUID()}`;
+
+    microsoft.serveCalendarEvents([
+      microsoftCalendarEvent({ subject: eventTitle }),
+    ]);
+
+    await runCalendarChannelListFetch(channel.calendarChannelId);
+
+    const channelState = await queryCalendarChannel(channel);
+
+    expect(channelState.syncStage).toBe(
+      CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING,
+    );
+
+    await runCalendarChannelEventsImport(channel.calendarChannelId);
+
+    expect(await findImportedCalendarEventTitles([eventTitle])).toEqual([
+      eventTitle,
+    ]);
+  }, 60000);
+
+  it('imports a newly created event through the delta-token continuation', async () => {
+    const newEventTitle = `Calendar event ${randomUUID()}`;
+
+    microsoft.serveCalendarEvents(
+      [microsoftCalendarEvent({ subject: newEventTitle })],
+      { deltaToken: 'mock-calendar-delta-token-2' },
+    );
+
+    await runCalendarChannelListFetch(channel.calendarChannelId);
+
+    const channelState = await queryCalendarChannel(channel);
+
+    expect(channelState.syncStage).toBe(
+      CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING,
+    );
+
+    await runCalendarChannelEventsImport(channel.calendarChannelId);
+
+    expect(await findImportedCalendarEventTitles([newEventTitle])).toEqual([
+      newEventTitle,
+    ]);
+  }, 60000);
+});
@@ -0,0 +1,40 @@
+import { ConnectedAccountProvider } from 'twenty-shared/types';
+
+import { setupMicrosoftMock } from 'test/integration/microsoft/mocks/setup-microsoft-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import { queryMessageFolders } from 'test/integration/utils/query-messaging.util';
+import { runMessageChannelSync } from 'test/integration/utils/run-message-channel-sync.util';
+
+const HANDLE = 'microsoft-folder-discovery@apple.dev';
+
+describe('Microsoft folder discovery (integration)', () => {
+  setupMicrosoftMock({ handle: HANDLE });
+
+  let channel: Awaited<ReturnType<typeof connectMessagingAccount>>;
+
+  beforeAll(async () => {
+    channel = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.MICROSOFT,
+      handle: HANDLE,
+    });
+  }, 60000);
+
+  afterAll(async () => {
+    await channel?.cleanup().catch(() => undefined);
+  });
+
+  it('discovers Microsoft mail folders through the Graph delta sync', async () => {
+    await runMessageChannelSync(channel.channelId);
+
+    const folders = await queryMessageFolders(channel.channelId);
+
+    expect(
+      Object.fromEntries(
+        folders.map((folder) => [folder.name, folder.isSynced]),
+      ),
+    ).toEqual({
+      Inbox: true,
+      'Sent Items': true,
+    });
+  }, 60000);
+});
@@ -0,0 +1,25 @@
+import { http, HttpResponse } from 'msw';
+
+import { type MswHandler } from 'test/integration/utils/http-mock.util';
+
+export const microsoftAuthHandlers = (handle: string): MswHandler[] => [
+  http.post('https://login.microsoftonline.com/common/oauth2/v2.0/token', () =>
+    HttpResponse.json({
+      token_type: 'Bearer',
+      access_token: 'mock-access-token',
+      refresh_token: 'mock-refresh-token',
+      expires_in: 3600,
+      scope: 'openid profile email offline_access',
+    }),
+  ),
+  http.get('https://graph.microsoft.com/v1.0/me', () =>
+    HttpResponse.json({
+      id: 'microsoft-user-id',
+      displayName: 'Jane Austen',
+      givenName: 'Jane',
+      surname: 'Austen',
+      mail: handle,
+      userPrincipalName: handle,
+    }),
+  ),
+];
@@ -0,0 +1,21 @@
+import { randomUUID } from 'node:crypto';
+
+import { type Event } from '@microsoft/microsoft-graph-types';
+
+export const microsoftCalendarEvent = (overrides: Partial<Event> = {}): Event => {
+  const id = overrides.id ?? `microsoft-calendar-event-${randomUUID()}`;
+
+  return {
+    id,
+    iCalUId: `${id}@microsoft.com`,
+    subject: `Calendar event ${id}`,
+    isCancelled: false,
+    isAllDay: false,
+    start: { dateTime: '2023-11-15T10:00:00.000Z', timeZone: 'UTC' },
+    end: { dateTime: '2023-11-15T11:00:00.000Z', timeZone: 'UTC' },
+    createdDateTime: '2023-11-01T00:00:00.000Z',
+    lastModifiedDateTime: '2023-11-01T00:00:00.000Z',
+    attendees: [],
+    ...overrides,
+  };
+};
@@ -0,0 +1,21 @@
+import { type Event } from '@microsoft/microsoft-graph-types';
+import { http, HttpResponse } from 'msw';
+
+import { type MswHandler } from 'test/integration/utils/http-mock.util';
+
+export const microsoftCalendarEventsHandlers = (
+  events: Event[],
+  deltaToken: string,
+): MswHandler[] => [
+  http.get('*/me/calendar/events/delta', () =>
+    HttpResponse.json({
+      value: events.map((event) => ({ id: event.id })),
+      '@odata.deltaLink': `https://graph.microsoft.com/beta/me/calendar/events/delta?$deltatoken=${deltaToken}`,
+    }),
+  ),
+  ...events.map((event) =>
+    http.get(`*/me/calendar/events/${event.id}`, () =>
+      HttpResponse.json(event),
+    ),
+  ),
+];
@@ -0,0 +1,23 @@
+import { type MailFolder } from '@microsoft/microsoft-graph-types';
+import { http, HttpResponse } from 'msw';
+
+import { type MswHandler } from 'test/integration/utils/http-mock.util';
+import { type MockEntityStore } from 'test/integration/utils/mock-entity-store.util';
+
+export const microsoftMailboxHandlers = (
+  folderStore: MockEntityStore<MailFolder>,
+): MswHandler[] => [
+  http.get('*/me/mailFolders', () =>
+    HttpResponse.json<{ value: MailFolder[] }>({ value: folderStore.list() }),
+  ),
+  http.get('*/messages/delta', () =>
+    HttpResponse.json({
+      value: [],
+      '@odata.deltaLink':
+        'https://graph.microsoft.com/beta/me/mailfolders/inbox/messages/delta?$deltatoken=mock-delta-token',
+    }),
+  ),
+  http.post('*/$batch', () =>
+    HttpResponse.json<{ responses: never[] }>({ responses: [] }),
+  ),
+];
@@ -0,0 +1,49 @@
+import { type Event, type MailFolder } from '@microsoft/microsoft-graph-types';
+
+import { setupHttpMock } from 'test/integration/utils/http-mock.util';
+import { microsoftAuthHandlers } from 'test/integration/microsoft/mocks/microsoft-auth-handlers.util';
+import { microsoftCalendarEventsHandlers } from 'test/integration/microsoft/mocks/microsoft-calendar-events-handlers.util';
+import { microsoftMailboxHandlers } from 'test/integration/microsoft/mocks/microsoft-mailbox-handlers.util';
+import {
+  createMockEntityStore,
+  type MockEntityStore,
+} from 'test/integration/utils/mock-entity-store.util';
+
+const DEFAULT_FOLDERS: MailFolder[] = [
+  { id: 'inbox', displayName: 'Inbox' },
+  { id: 'sentitems', displayName: 'Sent Items' },
+];
+
+export type MicrosoftMock = {
+  folders: MockEntityStore<MailFolder>;
+  serveCalendarEvents: (
+    events: Event[],
+    options?: { deltaToken?: string },
+  ) => void;
+};
+
+export const setupMicrosoftMock = ({
+  handle,
+  folders = DEFAULT_FOLDERS,
+}: {
+  handle: string;
+  folders?: MailFolder[];
+}): MicrosoftMock => {
+  const folderStore = createMockEntityStore(
+    folders,
+    (folder) => folder.id ?? '',
+  );
+
+  const httpMock = setupHttpMock(
+    ...microsoftAuthHandlers(handle),
+    ...microsoftMailboxHandlers(folderStore),
+  );
+
+  return {
+    folders: folderStore,
+    serveCalendarEvents: (
+      events,
+      { deltaToken = 'mock-calendar-delta-token' } = {},
+    ) => httpMock.use(...microsoftCalendarEventsHandlers(events, deltaToken)),
+  };
+};
@@ -0,0 +1,109 @@
+import gql from 'graphql-tag';
+import request from 'supertest';
+
+import {
+  CalendarChannelVisibility,
+  ConnectedAccountProvider,
+  MessageChannelVisibility,
+} from 'twenty-shared/types';
+
+import {
+  deleteConnectedAccount,
+  getDataOrThrow,
+  queryCalendarChannels,
+  queryMessageChannels,
+} from 'test/integration/utils/query-messaging.util';
+import { makeMetadataAPIRequest } from 'test/integration/metadata/suites/utils/make-metadata-api-request.util';
+import { waitForAllJobsToFinish } from 'test/integration/utils/wait-for-all-jobs-to-finish.util';
+
+type ConnectMessagingAccountInput = {
+  provider: ConnectedAccountProvider;
+  handle: string;
+  skipChannelConfiguration?: boolean;
+};
+
+type ConnectMessagingAccountResult = {
+  channelId: string;
+  calendarChannelId: string;
+  connectedAccountId: string;
+  handle: string;
+  cleanup: () => Promise<void>;
+};
+
+const OAUTH_CALLBACK_PATH: Partial<Record<ConnectedAccountProvider, string>> = {
+  [ConnectedAccountProvider.GOOGLE]: '/auth/google-apis/get-access-token',
+  [ConnectedAccountProvider.MICROSOFT]: '/auth/microsoft-apis/get-access-token',
+};
+
+const generateTransientToken = async (): Promise<string> => {
+  const response = await makeMetadataAPIRequest({
+    query: gql`
+      mutation GenerateTransientToken {
+        generateTransientToken {
+          transientToken {
+            token
+          }
+        }
+      }
+    `,
+  });
+
+  const data = getDataOrThrow(response) as {
+    generateTransientToken: { transientToken: { token: string } };
+  };
+
+  return data.generateTransientToken.transientToken.token;
+};
+
+export const connectMessagingAccount = async ({
+  provider,
+  handle,
+  skipChannelConfiguration = true,
+}: ConnectMessagingAccountInput): Promise<ConnectMessagingAccountResult> => {
+  const callbackPath = OAUTH_CALLBACK_PATH[provider];
+
+  if (!callbackPath) {
+    throw new Error(`Unsupported OAuth provider: ${provider}`);
+  }
+
+  const state = JSON.stringify({
+    transientToken: await generateTransientToken(),
+    messageVisibility: MessageChannelVisibility.SHARE_EVERYTHING,
+    calendarVisibility: CalendarChannelVisibility.SHARE_EVERYTHING,
+    skipMessageChannelConfiguration: skipChannelConfiguration,
+  });
+
+  const callbackResponse = await request(`http://localhost:${APP_PORT}`)
+    .get(callbackPath)
+    .query({ code: 'mock-authorization-code', state });
+
+  await waitForAllJobsToFinish();
+
+  const connectedChannel = (await queryMessageChannels()).find(
+    (channel) => channel.handle === handle,
+  );
+
+  if (!connectedChannel) {
+    throw new Error(
+      `OAuth connect for ${provider} created no message channel for ${handle} (callback redirected to ${callbackResponse.headers.location})`,
+    );
+  }
+
+  const [calendarChannel] = await queryCalendarChannels(
+    connectedChannel.connectedAccountId,
+  );
+
+  if (!calendarChannel) {
+    throw new Error(
+      `OAuth connect for ${provider} created no calendar channel for ${handle}`,
+    );
+  }
+
+  return {
+    channelId: connectedChannel.id,
+    calendarChannelId: calendarChannel.id,
+    connectedAccountId: connectedChannel.connectedAccountId,
+    handle,
+    cleanup: () => deleteConnectedAccount(connectedChannel.connectedAccountId),
+  };
+};
@@ -0,0 +1,25 @@
+import { type MessageQueueJobData } from 'src/engine/core-modules/message-queue/interfaces/message-queue-job.interface';
+
+import { type MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { type MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import { getQueueToken } from 'src/engine/core-modules/message-queue/utils/get-queue-token.util';
+
+import { waitForAllJobsToFinish } from 'test/integration/utils/wait-for-all-jobs-to-finish.util';
+
+// Enqueues on the real BullMQ queue and waits for every queue to drain, so
+// follow-up jobs the worker chains (imports, contact creation, ...) are also
+// done when this resolves. Failure-path suites rely on the job being allowed
+// to fail: they assert the resulting channel state afterwards.
+export const enqueueJobAndDrain = async <TData extends MessageQueueJobData>(
+  queue: MessageQueue,
+  jobName: string,
+  data: TData,
+): Promise<void> => {
+  const messageQueueService = global.app.get<MessageQueueService>(
+    getQueueToken(queue),
+    { strict: false },
+  );
+
+  await messageQueueService.add(jobName, data);
+  await waitForAllJobsToFinish();
+};
@@ -0,0 +1,27 @@
+import { findRecordNodesByFilter } from 'test/integration/utils/find-records-by-filter.util';
+
+export const findImportedMessageSubjects = async (
+  subjects: string[],
+): Promise<string[]> => {
+  const messages = await findRecordNodesByFilter<{ subject: string }>(
+    'message',
+    'messages',
+    'subject',
+    { subject: { in: subjects } },
+  );
+
+  return messages.map((message) => message.subject).sort();
+};
+
+export const findImportedCalendarEventTitles = async (
+  titles: string[],
+): Promise<string[]> => {
+  const events = await findRecordNodesByFilter<{ title: string }>(
+    'calendarEvent',
+    'calendarEvents',
+    'title',
+    { title: { in: titles } },
+  );
+
+  return events.map((event) => event.title).sort();
+};
@@ -0,0 +1,45 @@
+import { findManyOperationFactory } from 'test/integration/graphql/utils/find-many-operation-factory.util';
+import { makeGraphqlAPIRequest } from 'test/integration/graphql/utils/make-graphql-api-request.util';
+
+export const findRecordNodesByFilter = async <TNode>(
+  objectMetadataSingularName: string,
+  objectMetadataPluralName: string,
+  gqlFields: string,
+  filter: object,
+): Promise<TNode[]> => {
+  const response = await makeGraphqlAPIRequest(
+    findManyOperationFactory({
+      objectMetadataSingularName,
+      objectMetadataPluralName,
+      gqlFields,
+      filter,
+    }),
+  );
+
+  if (response.body.errors?.length) {
+    throw new Error(
+      `findRecordNodesByFilter(${objectMetadataPluralName}) failed: ${response.body.errors
+        .map((error: { message: string }) => error.message)
+        .join('; ')}`,
+    );
+  }
+
+  return response.body.data[objectMetadataPluralName].edges.map(
+    (edge: { node: TNode }) => edge.node,
+  );
+};
+
+export const findRecordIdsByFilter = async (
+  objectMetadataSingularName: string,
+  objectMetadataPluralName: string,
+  filter: object,
+): Promise<string[]> => {
+  const nodes = await findRecordNodesByFilter<{ id: string }>(
+    objectMetadataSingularName,
+    objectMetadataPluralName,
+    'id',
+    filter,
+  );
+
+  return nodes.map((node) => node.id);
+};
@@ -0,0 +1,11 @@
+import { getRepositoryToken } from '@nestjs/typeorm';
+import { type EntityClassOrSchema } from '@nestjs/typeorm/dist/interfaces/entity-class-or-schema.type';
+
+import { type ObjectLiteral, type Repository } from 'typeorm';
+
+export const getCoreRepository = <Entity extends ObjectLiteral>(
+  target: EntityClassOrSchema,
+): Repository<Entity> =>
+  global.app.get<Repository<Entity>>(getRepositoryToken(target), {
+    strict: false,
+  });
@@ -0,0 +1,41 @@
+import { http, passthrough, type RequestHandler } from 'msw';
+import { setupServer } from 'msw/node';
+
+const localhostPassthroughHandlers = [
+  http.all('http://127.0.0.1*', () => passthrough()),
+  http.all('http://localhost*', () => passthrough()),
+];
+
+const server = setupServer(...localhostPassthroughHandlers);
+
+export type MswHandler = RequestHandler;
+
+export type HttpMock = {
+  use: (...handlers: MswHandler[]) => void;
+};
+
+export const setupHttpMock = (...baseHandlers: MswHandler[]): HttpMock => {
+  const applyBaseHandlers = () => {
+    if (baseHandlers.length > 0) {
+      server.use(...baseHandlers);
+    }
+  };
+
+  beforeAll(() => {
+    server.listen({ onUnhandledRequest: 'error' });
+    applyBaseHandlers();
+  });
+
+  afterEach(() => {
+    server.resetHandlers();
+    applyBaseHandlers();
+  });
+
+  afterAll(() => {
+    server.close();
+  });
+
+  return {
+    use: (...handlers: MswHandler[]) => server.use(...handlers),
+  };
+};
@@ -0,0 +1,27 @@
+export type MockEntityStore<TEntity> = {
+  add: (entity: TEntity) => void;
+  remove: (entityId: string) => void;
+  reset: () => void;
+  list: () => TEntity[];
+};
+
+export const createMockEntityStore = <TEntity>(
+  initialEntities: TEntity[],
+  getEntityId: (entity: TEntity) => string,
+): MockEntityStore<TEntity> => {
+  const seedEntities = [...initialEntities];
+  let entities = [...seedEntities];
+
+  return {
+    add: (entity) => {
+      entities = [...entities, entity];
+    },
+    remove: (entityId) => {
+      entities = entities.filter((entity) => getEntityId(entity) !== entityId);
+    },
+    reset: () => {
+      entities = [...seedEntities];
+    },
+    list: () => [...entities],
+  };
+};
@@ -0,0 +1,301 @@
+import gql from 'graphql-tag';
+
+import { type MessageFolderImportPolicy } from 'twenty-shared/types';
+
+import { type CalendarChannelDTO } from 'src/engine/metadata-modules/calendar-channel/dtos/calendar-channel.dto';
+import { type ConnectedAccountDTO } from 'src/engine/metadata-modules/connected-account/dtos/connected-account.dto';
+import { type MessageChannelDTO } from 'src/engine/metadata-modules/message-channel/dtos/message-channel.dto';
+import { type MessageFolderDTO } from 'src/engine/metadata-modules/message-folder/dtos/message-folder.dto';
+
+import { makeMetadataAPIRequest } from 'test/integration/metadata/suites/utils/make-metadata-api-request.util';
+import { waitForAllJobsToFinish } from 'test/integration/utils/wait-for-all-jobs-to-finish.util';
+
+type MetadataAPIResponse = {
+  body: { data: Record<string, unknown>; errors?: { message: string }[] };
+};
+
+export const getDataOrThrow = (response: MetadataAPIResponse) => {
+  if (response.body.errors?.length) {
+    throw new Error(
+      `Metadata API request failed: ${response.body.errors
+        .map((error) => error.message)
+        .join('; ')}`,
+    );
+  }
+
+  return response.body.data;
+};
+
+export type MessageFolderDto = Pick<
+  MessageFolderDTO,
+  'id' | 'name' | 'isSynced'
+>;
+
+export type MessageChannelDto = Pick<
+  MessageChannelDTO,
+  | 'id'
+  | 'handle'
+  | 'connectedAccountId'
+  | 'syncStatus'
+  | 'syncStage'
+  | 'syncStageStartedAt'
+  | 'throttleFailureCount'
+  | 'throttleRetryAfter'
+>;
+
+export type CalendarChannelDto = Pick<
+  CalendarChannelDTO,
+  | 'id'
+  | 'handle'
+  | 'connectedAccountId'
+  | 'syncStatus'
+  | 'syncStage'
+  | 'syncStageStartedAt'
+  | 'throttleFailureCount'
+>;
+
+export type ConnectedAccountDto = Pick<
+  ConnectedAccountDTO,
+  'id' | 'handle' | 'provider' | 'lastCredentialsRefreshedAt' | 'authFailedAt'
+>;
+
+type MessageChannelUpdate = {
+  messageFolderImportPolicy?: MessageFolderImportPolicy;
+  isSyncEnabled?: boolean;
+  isContactAutoCreationEnabled?: boolean;
+};
+
+type CalendarChannelUpdate = {
+  isSyncEnabled?: boolean;
+  isContactAutoCreationEnabled?: boolean;
+};
+
+const MESSAGE_CHANNEL_FIELDS = gql`
+  fragment TestMessageChannelFields on MessageChannel {
+    id
+    handle
+    connectedAccountId
+    syncStatus
+    syncStage
+    syncStageStartedAt
+    throttleFailureCount
+    throttleRetryAfter
+  }
+`;
+
+export const queryMessageChannels = async (): Promise<MessageChannelDto[]> => {
+  const response = await makeMetadataAPIRequest({
+    query: gql`
+      query MessageChannelsForTest {
+        myMessageChannels {
+          ...TestMessageChannelFields
+        }
+      }
+      ${MESSAGE_CHANNEL_FIELDS}
+    `,
+  });
+
+  return getDataOrThrow(response).myMessageChannels as MessageChannelDto[];
+};
+
+export const queryMessageChannel = async ({
+  connectedAccountId,
+  channelId,
+}: {
+  connectedAccountId: string;
+  channelId: string;
+}): Promise<MessageChannelDto> => {
+  const response = await makeMetadataAPIRequest({
+    query: gql`
+      query MessageChannelForTest($connectedAccountId: UUID) {
+        myMessageChannels(connectedAccountId: $connectedAccountId) {
+          ...TestMessageChannelFields
+        }
+      }
+      ${MESSAGE_CHANNEL_FIELDS}
+    `,
+    variables: { connectedAccountId },
+  });
+
+  const channel = (
+    getDataOrThrow(response).myMessageChannels as MessageChannelDto[]
+  ).find((candidate) => candidate.id === channelId);
+
+  if (!channel) {
+    throw new Error(`Message channel ${channelId} not found`);
+  }
+
+  return channel;
+};
+
+export const queryMessageFolders = async (
+  messageChannelId: string,
+): Promise<MessageFolderDto[]> => {
+  const response = await makeMetadataAPIRequest({
+    query: gql`
+      query MessageFoldersForTest($messageChannelId: UUID) {
+        myMessageFolders(messageChannelId: $messageChannelId) {
+          id
+          name
+          isSynced
+        }
+      }
+    `,
+    variables: { messageChannelId },
+  });
+
+  return getDataOrThrow(response).myMessageFolders as MessageFolderDto[];
+};
+
+export const queryCalendarChannels = async (
+  connectedAccountId: string,
+): Promise<CalendarChannelDto[]> => {
+  const response = await makeMetadataAPIRequest({
+    query: gql`
+      query CalendarChannelsForTest($connectedAccountId: UUID) {
+        myCalendarChannels(connectedAccountId: $connectedAccountId) {
+          id
+          handle
+          connectedAccountId
+          syncStatus
+          syncStage
+          syncStageStartedAt
+          throttleFailureCount
+        }
+      }
+    `,
+    variables: { connectedAccountId },
+  });
+
+  return getDataOrThrow(response).myCalendarChannels as CalendarChannelDto[];
+};
+
+export const queryCalendarChannel = async ({
+  connectedAccountId,
+  calendarChannelId,
+}: {
+  connectedAccountId: string;
+  calendarChannelId: string;
+}): Promise<CalendarChannelDto> => {
+  const channel = (await queryCalendarChannels(connectedAccountId)).find(
+    (candidate) => candidate.id === calendarChannelId,
+  );
+
+  if (!channel) {
+    throw new Error(`Calendar channel ${calendarChannelId} not found`);
+  }
+
+  return channel;
+};
+
+export const queryConnectedAccount = async (
+  connectedAccountId: string,
+): Promise<ConnectedAccountDto> => {
+  const response = await makeMetadataAPIRequest({
+    query: gql`
+      query ConnectedAccountsForTest {
+        myConnectedAccounts {
+          id
+          handle
+          provider
+          lastCredentialsRefreshedAt
+          authFailedAt
+        }
+      }
+    `,
+  });
+
+  const account = (
+    getDataOrThrow(response).myConnectedAccounts as ConnectedAccountDto[]
+  ).find((candidate) => candidate.id === connectedAccountId);
+
+  if (!account) {
+    throw new Error(`Connected account ${connectedAccountId} not found`);
+  }
+
+  return account;
+};
+
+export const updateMessageChannel = async (
+  messageChannelId: string,
+  update: MessageChannelUpdate,
+): Promise<void> => {
+  const response = await makeMetadataAPIRequest({
+    query: gql`
+      mutation UpdateMessageChannelForTest($input: UpdateMessageChannelInput!) {
+        updateMessageChannel(input: $input) {
+          id
+        }
+      }
+    `,
+    variables: {
+      input: {
+        id: messageChannelId,
+        update,
+      },
+    },
+  });
+
+  getDataOrThrow(response);
+};
+
+export const updateCalendarChannel = async (
+  calendarChannelId: string,
+  update: CalendarChannelUpdate,
+): Promise<void> => {
+  const response = await makeMetadataAPIRequest({
+    query: gql`
+      mutation UpdateCalendarChannelForTest(
+        $input: UpdateCalendarChannelInput!
+      ) {
+        updateCalendarChannel(input: $input) {
+          id
+        }
+      }
+    `,
+    variables: {
+      input: {
+        id: calendarChannelId,
+        update,
+      },
+    },
+  });
+
+  getDataOrThrow(response);
+};
+
+export const startChannelSync = async (
+  connectedAccountId: string,
+): Promise<void> => {
+  const response = await makeMetadataAPIRequest({
+    query: gql`
+      mutation StartChannelSyncForTest($connectedAccountId: UUID!) {
+        startChannelSync(connectedAccountId: $connectedAccountId) {
+          success
+        }
+      }
+    `,
+    variables: { connectedAccountId },
+  });
+
+  getDataOrThrow(response);
+};
+
+export const deleteConnectedAccount = async (
+  connectedAccountId: string,
+): Promise<void> => {
+  const response = await makeMetadataAPIRequest({
+    query: gql`
+      mutation DeleteConnectedAccountForTest($id: UUID!) {
+        deleteConnectedAccount(id: $id) {
+          id
+        }
+      }
+    `,
+    variables: { id: connectedAccountId },
+  });
+
+  getDataOrThrow(response);
+
+  await waitForAllJobsToFinish();
+};
@@ -0,0 +1,24 @@
+import { CalendarChannelSyncStage } from 'twenty-shared/types';
+
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity';
+import { CalendarEventsImportJob } from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-events-import.job';
+
+import { enqueueJobAndDrain } from 'test/integration/utils/enqueue-job-and-drain.util';
+import { scheduleChannelStage } from 'test/integration/utils/schedule-channel-stage.util';
+
+export const runCalendarChannelEventsImport = async (
+  calendarChannelId: string,
+): Promise<void> => {
+  const workspaceId = await scheduleChannelStage(
+    CalendarChannelEntity,
+    calendarChannelId,
+    CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED,
+  );
+
+  await enqueueJobAndDrain(
+    MessageQueue.calendarQueue,
+    CalendarEventsImportJob.name,
+    { workspaceId, calendarChannelId },
+  );
+};
@@ -0,0 +1,24 @@
+import { CalendarChannelSyncStage } from 'twenty-shared/types';
+
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity';
+import { CalendarEventListFetchJob } from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job';
+
+import { enqueueJobAndDrain } from 'test/integration/utils/enqueue-job-and-drain.util';
+import { scheduleChannelStage } from 'test/integration/utils/schedule-channel-stage.util';
+
+export const runCalendarChannelListFetch = async (
+  calendarChannelId: string,
+): Promise<void> => {
+  const workspaceId = await scheduleChannelStage(
+    CalendarChannelEntity,
+    calendarChannelId,
+    CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED,
+  );
+
+  await enqueueJobAndDrain(
+    MessageQueue.calendarQueue,
+    CalendarEventListFetchJob.name,
+    { workspaceId, calendarChannelId },
+  );
+};
@@ -0,0 +1,24 @@
+import { MessageChannelSyncStage } from 'twenty-shared/types';
+
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
+import { MessagingMessageListFetchJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job';
+
+import { enqueueJobAndDrain } from 'test/integration/utils/enqueue-job-and-drain.util';
+import { scheduleChannelStage } from 'test/integration/utils/schedule-channel-stage.util';
+
+export const runMessageChannelSync = async (
+  messageChannelId: string,
+): Promise<void> => {
+  const workspaceId = await scheduleChannelStage(
+    MessageChannelEntity,
+    messageChannelId,
+    MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED,
+  );
+
+  await enqueueJobAndDrain(
+    MessageQueue.messagingQueue,
+    MessagingMessageListFetchJob.name,
+    { workspaceId, messageChannelId },
+  );
+};
@@ -0,0 +1,9 @@
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+
+import { enqueueJobAndDrain } from 'test/integration/utils/enqueue-job-and-drain.util';
+
+export const runSyncCron = async (cronJob: {
+  name: string;
+}): Promise<void> => {
+  await enqueueJobAndDrain(MessageQueue.cronQueue, cronJob.name, {});
+};
@@ -0,0 +1,29 @@
+import { type EntityClassOrSchema } from '@nestjs/typeorm/dist/interfaces/entity-class-or-schema.type';
+
+import {
+  type CalendarChannelSyncStage,
+  type MessageChannelSyncStage,
+} from 'twenty-shared/types';
+
+import { getCoreRepository } from 'test/integration/utils/get-core-repository.util';
+
+// The sync crons scan every active workspace's channels, so triggering them from
+// a test fans out across co-located suites. The per-channel triggers instead
+// drive a SINGLE channel through its leaf job: this sets the stage the leaf job
+// guards on and returns the workspaceId the job payload needs.
+export const scheduleChannelStage = async (
+  channelEntity: EntityClassOrSchema,
+  channelId: string,
+  scheduledStage: MessageChannelSyncStage | CalendarChannelSyncStage,
+): Promise<string> => {
+  const repository = getCoreRepository<{
+    id: string;
+    workspaceId: string;
+    syncStage: MessageChannelSyncStage | CalendarChannelSyncStage;
+  }>(channelEntity);
+  const channel = await repository.findOneByOrFail({ id: channelId });
+
+  await repository.update({ id: channelId }, { syncStage: scheduledStage });
+
+  return channel.workspaceId;
+};
@@ -1,3 +1,4 @@
+import nodeFetch from 'node-fetch';
 import { type JestConfigWithTsJest } from 'ts-jest';
 import 'tsconfig-paths/register';

@@ -6,6 +7,10 @@ import { rawDataSource } from 'src/database/typeorm/raw/raw.datasource';
 import { createApp } from './create-app';

 export default async (_: unknown, projectConfig: JestConfigWithTsJest) => {
+  // node-fetch rides node:http, which msw patches; native undici fetch
+  // escapes interception.
+  globalThis.fetch = nodeFetch as unknown as typeof globalThis.fetch;
+
  const app = await createApp({});

  if (!projectConfig.globals) {
@@ -0,0 +1,9 @@
+import { startChannelSync } from 'test/integration/utils/query-messaging.util';
+import { waitForAllJobsToFinish } from 'test/integration/utils/wait-for-all-jobs-to-finish.util';
+
+export const startChannelSyncAndAwait = async (
+  connectedAccountId: string,
+): Promise<void> => {
+  await startChannelSync(connectedAccountId);
+  await waitForAllJobsToFinish();
+};
@@ -22277,7 +22277,7 @@ __metadata:
  languageName: node
  linkType: hard

-"@types/node-fetch@npm:^2.5.10":
+"@types/node-fetch@npm:^2.5.10, @types/node-fetch@npm:^2.6.12":
  version: 2.6.13
  resolution: "@types/node-fetch@npm:2.6.13"
  dependencies:
@@ -53185,6 +53185,7 @@ __metadata:
    "@types/lodash.uniqby": "npm:^4.7.9"
    "@types/ms": "npm:^0.7.31"
    "@types/node": "npm:^24.0.0"
+    "@types/node-fetch": "npm:^2.6.12"
    "@types/nodemailer": "npm:^7.0.3"
    "@types/passport-google-oauth20": "npm:^2.0.11"
    "@types/passport-jwt": "npm:^3.0.8"
@@ -53260,7 +53261,9 @@ __metadata:
    microdiff: "npm:1.4.0"
    mrmime: "npm:^2.0.1"
    ms: "npm:2.1.3"
+    msw: "npm:^2.12.7"
    nest-commander: "npm:^3.19.1"
+    node-fetch: "npm:^2.7.0"
    node-ical: "npm:^0.21.0"
    nodemailer: "npm:^9.0.1"
    openapi-types: "npm:12.1.3"