Expire webhook subscriptions whose refresh token is dead (#23907)

Follow-up to #23707. Token errors thrown while building the OAuth client
never reach the driver-exception mapping, so they landed in
`handleUnknownException`: channel marked FAILED, captured to Sentry,
rethrown (captured again by the queue explorer). The renewal cron
re-selects FAILED channels every tick, so a dead refresh token looped
forever.

Routes REFRESH_TOKEN_NOT_FOUND and INVALID_REFRESH_TOKEN to the existing
expiry path, matching the message and calendar import handlers. Sampled
160 events across
[TWENTY-SERVER-J1F](https://twenty-v7.sentry.io/issues/7603992092/)
(~5.9k/day) and
[TWENTY-SERVER-JBZ](https://twenty-v7.sentry.io/issues/7617075015/)
(~1.5k/day): 100% originate here.

<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/23907?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>
<!-- End of auto-generated description by cubic. -->

---------

Co-authored-by: neo773 <huzef@twenty.com>

packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-exception-handler.service.ts

@@ -3,6 +3,10 @@ import { Injectable } from '@nestjs/common';
 import { type WebhookSubscriptionChannelType } from 'twenty-shared/types';
 
 import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
+import {
+  ConnectedAccountRefreshAccessTokenException,
+  ConnectedAccountRefreshAccessTokenExceptionCode,
+} from 'src/engine/metadata-modules/connected-account/exceptions/connected-account-refresh-tokens.exception';
 import {
   WebhookSubscriptionDriverException,
   WebhookSubscriptionDriverExceptionCode,
@@ -67,6 +71,32 @@ export class WebhookSubscriptionExceptionHandlerService {
       }
     }
 
+    if (exception instanceof ConnectedAccountRefreshAccessTokenException) {
+      switch (exception.code) {
+        case ConnectedAccountRefreshAccessTokenExceptionCode.REFRESH_TOKEN_NOT_FOUND:
+        case ConnectedAccountRefreshAccessTokenExceptionCode.INVALID_REFRESH_TOKEN:
+          return await this.handleInsufficientPermissionsException(
+            channelType,
+            channel,
+          );
+        case ConnectedAccountRefreshAccessTokenExceptionCode.TEMPORARY_NETWORK_ERROR:
+          return await this.handleTemporaryException(
+            exception,
+            channelType,
+            channel,
+          );
+        case ConnectedAccountRefreshAccessTokenExceptionCode.ACCESS_TOKEN_NOT_FOUND:
+        case ConnectedAccountRefreshAccessTokenExceptionCode.PROVIDER_NOT_SUPPORTED:
+        default:
+          return await this.handleUnknownException(
+            exception,
+            channelType,
+            channel,
+            workspaceId,
+          );
+      }
+    }
+
     return await this.handleUnknownException(
       exception,
       channelType,
@@ -112,7 +142,9 @@ export class WebhookSubscriptionExceptionHandlerService {
   }
 
   private async handleTemporaryException(
-    exception: WebhookSubscriptionDriverException,
+    exception:
+      | WebhookSubscriptionDriverException
+      | ConnectedAccountRefreshAccessTokenException,
     channelType: WebhookSubscriptionChannelType,
     channel: WebhookSubscribableChannelReference,
   ): Promise<WebhookSubscriptionRecoveryAction> {

packages/twenty-server/test/integration/microsoft/webhook/renewal-auth-failure.integration-spec.ts

@@ -0,0 +1,136 @@
+import {
+  ConnectedAccountProvider,
+  WebhookSubscriptionStatus,
+} 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 { type EncryptedString } from 'src/engine/core-modules/secret-encryption/branded-strings/encrypted-string.type';
+import { WebhookSubscriptionRenewalCronJob } from 'src/modules/connected-account/webhook-subscription-manager/crons/jobs/webhook-subscription-renewal.cron.job';
+
+import { setupMicrosoftMock } from 'test/integration/microsoft/mocks/setup-microsoft-mock.util';
+import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util';
+import { getCoreRepository } from 'test/integration/utils/get-core-repository.util';
+import { runSyncCron } from 'test/integration/utils/run-sync-cron.util';
+
+const HANDLE = 'microsoft-webhook-renewal-auth@apple.dev';
+const CLIENT_STATE = 'renewal-auth-client-state';
+const SUBSCRIPTION_ID = 'subscription-due-for-renewal';
+
+describe('Microsoft webhook subscription renewal (integration)', () => {
+  const microsoft = setupMicrosoftMock({ handle: HANDLE });
+
+  let account: Awaited<ReturnType<typeof connectMessagingAccount>>;
+  let storedRefreshToken: EncryptedString | null;
+
+  const calendarChannelRepository = () =>
+    getCoreRepository<CalendarChannelEntity>(CalendarChannelEntity);
+
+  const connectedAccountRepository = () =>
+    getCoreRepository<ConnectedAccountEntity>(ConnectedAccountEntity);
+
+  const readChannel = async () =>
+    await calendarChannelRepository().findOneOrFail({
+      where: { id: account.calendarChannelId },
+    });
+
+  const makeSubscriptionDueForRenewal = async () => {
+    await calendarChannelRepository().update(account.calendarChannelId, {
+      webhookSubscriptionExternalId: SUBSCRIPTION_ID,
+      webhookSubscriptionClientState: CLIENT_STATE,
+      webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE,
+      webhookSubscriptionExpiresAt: new Date(Date.now() + 60 * 1000),
+    });
+  };
+
+  // resolveTokens short-circuits on a still-valid access token, so the refresh
+  // token is only consulted once lastCredentialsRefreshedAt is stale.
+  const revokeRefreshToken = async () => {
+    await connectedAccountRepository().update(account.connectedAccountId, {
+      refreshToken: null,
+      lastCredentialsRefreshedAt: null,
+    });
+  };
+
+  const restoreRefreshToken = async () => {
+    await connectedAccountRepository().update(account.connectedAccountId, {
+      refreshToken: storedRefreshToken,
+      lastCredentialsRefreshedAt: null,
+    });
+  };
+
+  beforeAll(async () => {
+    account = await connectMessagingAccount({
+      provider: ConnectedAccountProvider.MICROSOFT,
+      handle: HANDLE,
+    });
+
+    const connectedAccount = await connectedAccountRepository().findOneOrFail({
+      where: { id: account.connectedAccountId },
+    });
+
+    storedRefreshToken = connectedAccount.refreshToken;
+  }, 120000);
+
+  beforeEach(async () => {
+    microsoft.subscriptions.reset();
+    await makeSubscriptionDueForRenewal();
+  });
+
+  afterAll(async () => {
+    await account?.cleanup().catch(() => undefined);
+  });
+
+  it('renews a subscription that is close to expiring', async () => {
+    await runSyncCron(WebhookSubscriptionRenewalCronJob);
+
+    expect(microsoft.subscriptions.renewed).toContain(SUBSCRIPTION_ID);
+
+    const channel = await readChannel();
+
+    expect(channel.webhookSubscriptionStatus).toBe(
+      WebhookSubscriptionStatus.ACTIVE,
+    );
+  }, 60000);
+
+  describe('when the connected account has no usable refresh token', () => {
+    beforeEach(async () => {
+      await revokeRefreshToken();
+    });
+
+    it('expires the channel rather than marking it failed', async () => {
+      await runSyncCron(WebhookSubscriptionRenewalCronJob);
+
+      const channel = await readChannel();
+
+      expect(channel.webhookSubscriptionStatus).toBe(
+        WebhookSubscriptionStatus.EXPIRED,
+      );
+      expect(microsoft.subscriptions.renewed).not.toContain(SUBSCRIPTION_ID);
+    }, 60000);
+
+    // Restoring the token before the second run is what makes this meaningful:
+    // a re-selected channel would reach the provider and get a subscription.
+    it('stops selecting the channel on subsequent cron runs', async () => {
+      await runSyncCron(WebhookSubscriptionRenewalCronJob);
+
+      expect((await readChannel()).webhookSubscriptionStatus).toBe(
+        WebhookSubscriptionStatus.EXPIRED,
+      );
+
+      await restoreRefreshToken();
+      microsoft.subscriptions.reset();
+
+      await runSyncCron(WebhookSubscriptionRenewalCronJob);
+
+      const channel = await readChannel();
+
+      expect(channel.webhookSubscriptionStatus).toBe(
+        WebhookSubscriptionStatus.EXPIRED,
+      );
+      expect(channel.webhookSubscriptionExternalId).toBe(SUBSCRIPTION_ID);
+      expect(microsoft.subscriptions.created).toHaveLength(0);
+      expect(microsoft.subscriptions.renewed).not.toContain(SUBSCRIPTION_ID);
+    }, 60000);
+  });
+});