feat(connections): run a logic function on connection provider connect (#23167)

## What

Adds an optional `onConnectLogicFunctionUniversalIdentifier` field to
the connection provider manifest. When set, the referenced logic
function is dispatched right after an OAuth connection is successfully
established for that provider.

This gives apps a first-class "on connect" hook — e.g. the Slack app can
resolve the workspace's `team_id` via `auth.test` and claim the `team_id
-> workspaceId` mapping in the SERVER key-value store immediately on
connect, instead of racing against later events.

Follow-up to the app key-value store PR (#23089).

## How

- **twenty-shared**: add `onConnectLogicFunctionUniversalIdentifier` to
`ConnectionProviderManifest`.
- **twenty-sdk**: expose the field in `defineConnectionProvider` and
validate it is a UUID `universalIdentifier`.
- **twenty-server**:
- add a nullable `onConnectLogicFunctionUniversalIdentifier` column to
`ConnectionProviderEntity` (+ fast instance command / migration).
  - map the field through the

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

packages/twenty-server/src/engine/core-modules/application/application-manifest/converters/__tests__/from-connection-provider-manifest-to-universal-flat-connection-provider.util.spec.ts

@@ -50,11 +50,32 @@ describe('fromConnectionProviderManifestToUniversalFlatConnectionProvider', () =
         tokenRequestContentType: 'json',
         usePkce: true,
       },
+      onConnectLogicFunctionUniversalIdentifier: null,
       createdAt: NOW,
       updatedAt: NOW,
     });
   });
 
+  it('resolves the onConnectLogicFunction universalIdentifier into the flat field when provided', () => {
+    const onConnectLogicFunctionUniversalIdentifier =
+      'c1c1c1c1-c1c1-4c1c-c1c1-c1c1c1c1c1c1';
+
+    const result =
+      fromConnectionProviderManifestToUniversalFlatConnectionProvider({
+        connectionProviderManifest: buildManifest({
+          onConnectLogicFunction: {
+            universalIdentifier: onConnectLogicFunctionUniversalIdentifier,
+          },
+        }),
+        applicationUniversalIdentifier: APP_UID,
+        now: NOW,
+      });
+
+    expect(result.onConnectLogicFunctionUniversalIdentifier).toBe(
+      onConnectLogicFunctionUniversalIdentifier,
+    );
+  });
+
   it('passes through optional oauth config when provided', () => {
     const result =
       fromConnectionProviderManifestToUniversalFlatConnectionProvider({

packages/twenty-server/src/engine/core-modules/application/application-manifest/converters/from-connection-provider-manifest-to-universal-flat-connection-provider.util.ts

@@ -43,6 +43,9 @@ export const fromConnectionProviderManifestToUniversalFlatConnectionProvider =
       displayName: connectionProviderManifest.displayName,
       type: connectionProviderManifest.type,
       oauthConfig,
+      onConnectLogicFunctionUniversalIdentifier:
+        connectionProviderManifest.onConnectLogicFunction
+          ?.universalIdentifier ?? null,
       createdAt: now,
       updatedAt: now,
     };

packages/twenty-server/src/engine/core-modules/application/connection-provider/__tests__/connection-provider-oauth-flow.service.spec.ts

@@ -19,12 +19,16 @@ import { type ConnectionProviderEntity } from 'src/engine/core-modules/applicati
 import { ConnectionProviderOAuthFlowService } from 'src/engine/core-modules/application/connection-provider/connection-provider-oauth-flow.service';
 import { ConnectionProviderService } from 'src/engine/core-modules/application/connection-provider/connection-provider.service';
 import { JwtTokenTypeEnum } from 'src/engine/core-modules/auth/types/jwt-token-type.enum';
+import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
 import { JwtWrapperService } from 'src/engine/core-modules/jwt/services/jwt-wrapper.service';
 import { SecureHttpClientService } from 'src/engine/core-modules/secure-http-client/secure-http-client.service';
 import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { getQueueToken } from 'src/engine/core-modules/message-queue/utils/get-queue-token.util';
 import { SECRET_ENCRYPTION_ENVELOPE_V2_PREFIX } from 'src/engine/core-modules/secret-encryption/constants/secret-encryption.constant';
 import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
 import { ConnectedAccountTokenEncryptionService } from 'src/engine/metadata-modules/connected-account/services/connected-account-token-encryption.service';
+import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
 
 const FAKE_CIPHER_PREFIX = `${SECRET_ENCRYPTION_ENVELOPE_V2_PREFIX}keyid:`;
 
@@ -47,6 +51,9 @@ describe('ConnectionProviderOAuthFlowService', () => {
     findOne: jest.Mock;
     findOneByOrFail: jest.Mock;
   };
+  let workspaceCacheService: { getOrRecompute: jest.Mock };
+  let messageQueueService: { add: jest.Mock };
+  let exceptionHandlerService: { captureExceptions: jest.Mock };
 
   const baseProvider: ConnectionProviderEntity = {
     id: 'provider-1',
@@ -95,6 +102,13 @@ describe('ConnectionProviderOAuthFlowService', () => {
         provider: ConnectedAccountProvider.APP,
       })),
     };
+    workspaceCacheService = {
+      getOrRecompute: jest.fn(async () => ({
+        flatLogicFunctionMaps: { byUniversalIdentifier: {} },
+      })),
+    };
+    messageQueueService = { add: jest.fn() };
+    exceptionHandlerService = { captureExceptions: jest.fn() };
 
     const module: TestingModule = await Test.createTestingModule({
       providers: [
@@ -113,6 +127,18 @@ describe('ConnectionProviderOAuthFlowService', () => {
           provide: getRepositoryToken(ConnectedAccountEntity),
           useValue: connectedAccountRepository,
         },
+        {
+          provide: WorkspaceCacheService,
+          useValue: workspaceCacheService,
+        },
+        {
+          provide: getQueueToken(MessageQueue.logicFunctionQueue),
+          useValue: messageQueueService,
+        },
+        {
+          provide: ExceptionHandlerService,
+          useValue: exceptionHandlerService,
+        },
         {
           // Real prefix/round-trip behavior is asserted in
           // connected-account-token-encryption.service.spec.ts; here we use a
@@ -437,5 +463,107 @@ describe('ConnectionProviderOAuthFlowService', () => {
         }),
       ).rejects.toThrow(/state/);
     });
+
+    describe('on-connect hook', () => {
+      const ON_CONNECT_UID = 'c1c1c1c1-c1c1-4c1c-c1c1-c1c1c1c1c1c1';
+
+      it('does not dispatch a hook when the provider declares none', async () => {
+        await service.completeAuthorizationFlow({
+          code: 'auth_code',
+          state: 'signed-state',
+        });
+
+        expect(workspaceCacheService.getOrRecompute).not.toHaveBeenCalled();
+        expect(messageQueueService.add).not.toHaveBeenCalled();
+      });
+
+      it('enqueues the declared on-connect logic function in the connecting workspace', async () => {
+        connectionProviderService.findOneByIdOrThrow.mockResolvedValue({
+          ...baseProvider,
+          onConnectLogicFunctionUniversalIdentifier: ON_CONNECT_UID,
+        });
+        workspaceCacheService.getOrRecompute.mockResolvedValue({
+          flatLogicFunctionMaps: {
+            byUniversalIdentifier: {
+              [ON_CONNECT_UID]: { id: 'logic-function-1' },
+            },
+          },
+        });
+
+        const result = await service.completeAuthorizationFlow({
+          code: 'auth_code',
+          state: 'signed-state',
+        });
+
+        expect(workspaceCacheService.getOrRecompute).toHaveBeenCalledWith(
+          'workspace-1',
+          ['flatLogicFunctionMaps'],
+        );
+        expect(messageQueueService.add).toHaveBeenCalledWith(
+          'LogicFunctionTriggerJob',
+          [
+            {
+              logicFunctionId: 'logic-function-1',
+              workspaceId: 'workspace-1',
+              payload: {
+                connectionProviderId: 'provider-1',
+                connectionProviderName: 'linear',
+                connectedAccountId: result.connectedAccountId,
+              },
+            },
+          ],
+          { retryLimit: 3 },
+        );
+      });
+
+      it('reports to Sentry without failing the connection when the hook function is missing', async () => {
+        connectionProviderService.findOneByIdOrThrow.mockResolvedValue({
+          ...baseProvider,
+          onConnectLogicFunctionUniversalIdentifier: ON_CONNECT_UID,
+        });
+        workspaceCacheService.getOrRecompute.mockResolvedValue({
+          flatLogicFunctionMaps: { byUniversalIdentifier: {} },
+        });
+
+        const result = await service.completeAuthorizationFlow({
+          code: 'auth_code',
+          state: 'signed-state',
+        });
+
+        expect(result.connectedAccountId).toBe('new-account-id');
+        expect(messageQueueService.add).not.toHaveBeenCalled();
+        expect(exceptionHandlerService.captureExceptions).toHaveBeenCalledTimes(
+          1,
+        );
+      });
+
+      it('treats a soft-deleted hook function as missing instead of enqueuing it', async () => {
+        connectionProviderService.findOneByIdOrThrow.mockResolvedValue({
+          ...baseProvider,
+          onConnectLogicFunctionUniversalIdentifier: ON_CONNECT_UID,
+        });
+        workspaceCacheService.getOrRecompute.mockResolvedValue({
+          flatLogicFunctionMaps: {
+            byUniversalIdentifier: {
+              [ON_CONNECT_UID]: {
+                id: 'logic-function-1',
+                deletedAt: new Date().toISOString(),
+              },
+            },
+          },
+        });
+
+        const result = await service.completeAuthorizationFlow({
+          code: 'auth_code',
+          state: 'signed-state',
+        });
+
+        expect(result.connectedAccountId).toBe('new-account-id');
+        expect(messageQueueService.add).not.toHaveBeenCalled();
+        expect(exceptionHandlerService.captureExceptions).toHaveBeenCalledTimes(
+          1,
+        );
+      });
+    });
   });
 });

packages/twenty-server/src/engine/core-modules/application/connection-provider/connection-provider-exception-code.enum.ts

@@ -9,4 +9,5 @@ export enum ConnectionProviderExceptionCode {
   INVALID_CONNECTION_PROVIDER_INPUT = 'INVALID_CONNECTION_PROVIDER_INPUT',
   CONNECTION_PROVIDER_NOT_FOUND = 'CONNECTION_PROVIDER_NOT_FOUND',
   CONNECTION_PROVIDER_NAME_ALREADY_EXISTS = 'CONNECTION_PROVIDER_NAME_ALREADY_EXISTS',
+  ON_CONNECT_LOGIC_FUNCTION_NOT_FOUND = 'ON_CONNECT_LOGIC_FUNCTION_NOT_FOUND',
 }

packages/twenty-server/src/engine/core-modules/application/connection-provider/connection-provider-oauth-flow.service.ts

@@ -21,11 +21,20 @@ import { exchangeCodeForToken } from 'src/engine/core-modules/application/connec
 import { generatePkceVerifier } from 'src/engine/core-modules/application/connection-provider/utils/generate-pkce-verifier.util';
 import { type AppOAuthStateJwtPayload } from 'src/engine/core-modules/auth/types/app-oauth-state-jwt-payload.type';
 import { JwtTokenTypeEnum } from 'src/engine/core-modules/auth/types/jwt-token-type.enum';
+import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
 import { JwtWrapperService } from 'src/engine/core-modules/jwt/services/jwt-wrapper.service';
+import {
+  LogicFunctionTriggerJob,
+  type LogicFunctionTriggerJobData,
+} from 'src/engine/core-modules/logic-function/logic-function-trigger/jobs/logic-function-trigger.job';
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
 import { SecureHttpClientService } from 'src/engine/core-modules/secure-http-client/secure-http-client.service';
 import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
 import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
 import { ConnectedAccountTokenEncryptionService } from 'src/engine/metadata-modules/connected-account/services/connected-account-token-encryption.service';
+import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
 
 const STATE_JWT_EXPIRES_IN = '10m';
 
@@ -63,6 +72,10 @@ export class ConnectionProviderOAuthFlowService {
     private readonly connectedAccountTokenEncryptionService: ConnectedAccountTokenEncryptionService,
     @InjectRepository(ConnectedAccountEntity)
     private readonly connectedAccountRepository: Repository<ConnectedAccountEntity>,
+    @InjectMessageQueue(MessageQueue.logicFunctionQueue)
+    private readonly messageQueueService: MessageQueueService,
+    private readonly workspaceCacheService: WorkspaceCacheService,
+    private readonly exceptionHandlerService: ExceptionHandlerService,
   ) {}
 
   async startAuthorizationFlow(
@@ -186,6 +199,12 @@ export class ConnectionProviderOAuthFlowService {
         statePayload.reconnectingConnectedAccountId,
     });
 
+    await this.dispatchOnConnectHook({
+      provider,
+      workspaceId: statePayload.workspaceId,
+      connectedAccountId: connectedAccount.id,
+    });
+
     return {
       connectedAccountId: connectedAccount.id,
       workspaceId: statePayload.workspaceId,
@@ -194,6 +213,67 @@ export class ConnectionProviderOAuthFlowService {
     };
   }
 
+  private async dispatchOnConnectHook({
+    provider,
+    workspaceId,
+    connectedAccountId,
+  }: {
+    provider: OAuthConnectionProvider;
+    workspaceId: string;
+    connectedAccountId: string;
+  }): Promise<void> {
+    const { onConnectLogicFunctionUniversalIdentifier } = provider;
+
+    if (!isDefined(onConnectLogicFunctionUniversalIdentifier)) {
+      return;
+    }
+
+    // The on-connect hook is best-effort: the ConnectedAccount is already
+    // persisted, so a misconfigured or failing hook must not break the OAuth
+    // callback. We still report failures to Sentry so they don't go unnoticed.
+    try {
+      const { flatLogicFunctionMaps } =
+        await this.workspaceCacheService.getOrRecompute(workspaceId, [
+          'flatLogicFunctionMaps',
+        ]);
+
+      const flatLogicFunction =
+        flatLogicFunctionMaps.byUniversalIdentifier[
+          onConnectLogicFunctionUniversalIdentifier
+        ];
+
+      if (
+        !isDefined(flatLogicFunction) ||
+        isDefined(flatLogicFunction.deletedAt)
+      ) {
+        throw new ConnectionProviderException(
+          `Connection provider ${provider.id} references on-connect logic function ${onConnectLogicFunctionUniversalIdentifier}, which was not found in workspace ${workspaceId}.`,
+          ConnectionProviderExceptionCode.ON_CONNECT_LOGIC_FUNCTION_NOT_FOUND,
+        );
+      }
+
+      await this.messageQueueService.add<LogicFunctionTriggerJobData[]>(
+        LogicFunctionTriggerJob.name,
+        [
+          {
+            logicFunctionId: flatLogicFunction.id,
+            workspaceId,
+            payload: {
+              connectionProviderId: provider.id,
+              connectionProviderName: provider.name,
+              connectedAccountId,
+            },
+          },
+        ],
+        { retryLimit: 3 },
+      );
+    } catch (error) {
+      this.exceptionHandlerService.captureExceptions([error], {
+        workspace: { id: workspaceId },
+      });
+    }
+  }
+
   private async signState(payload: AppOAuthStateJwtPayload): Promise<string> {
     return this.jwtWrapperService.signAsyncOrThrow(payload, {
       expiresIn: STATE_JWT_EXPIRES_IN,

packages/twenty-server/src/engine/core-modules/application/connection-provider/connection-provider.entity.ts

@@ -12,6 +12,7 @@ import {
   UpdateDateColumn,
 } from 'typeorm';
 
+import { WasIntroducedInUpgrade } from 'src/engine/core-modules/upgrade/decorators/was-introduced-in-upgrade.decorator';
 import { SyncableEntity } from 'src/engine/workspace-manager/types/syncable-entity.interface';
 
 @Entity({ name: 'connectionProvider', schema: 'core' })
@@ -39,6 +40,13 @@ export class ConnectionProviderEntity
   @Column({ nullable: true, type: 'jsonb' })
   oauthConfig: StoredOAuthConnectionProviderConfig | null;
 
+  @Column({ nullable: true, type: 'uuid' })
+  @WasIntroducedInUpgrade({
+    upgradeCommandName:
+      '2.24.0_AddOnConnectLogicFunctionToConnectionProviderFastInstanceCommand_1784712843602',
+  })
+  onConnectLogicFunctionUniversalIdentifier: string | null;
+
   @CreateDateColumn({ type: 'timestamptz' })
   createdAt: Date;
 

packages/twenty-server/src/engine/core-modules/application/connection-provider/connection-provider.exception.ts

@@ -29,6 +29,8 @@ const getConnectionProviderExceptionUserFriendlyMessage = (
       return msg`Connection provider not found.`;
     case ConnectionProviderExceptionCode.CONNECTION_PROVIDER_NAME_ALREADY_EXISTS:
       return msg`A connection provider with this name already exists for this application.`;
+    case ConnectionProviderExceptionCode.ON_CONNECT_LOGIC_FUNCTION_NOT_FOUND:
+      return msg`The logic function to run on connect was not found.`;
     default:
       assertUnreachable(code);
   }

packages/twenty-server/src/engine/core-modules/application/connection-provider/connection-provider.module.ts

@@ -14,6 +14,7 @@ import { TwentyConfigModule } from 'src/engine/core-modules/twenty-config/twenty
 import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
 import { ConnectedAccountTokenEncryptionModule } from 'src/engine/metadata-modules/connected-account/services/connected-account-token-encryption.module';
 import { FlatConnectionProviderModule } from 'src/engine/metadata-modules/flat-connection-provider/flat-connection-provider.module';
+import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache.module';
 
 @Module({
   imports: [
@@ -29,6 +30,7 @@ import { FlatConnectionProviderModule } from 'src/engine/metadata-modules/flat-c
     TwentyConfigModule,
     FlatConnectionProviderModule,
     ConnectedAccountTokenEncryptionModule,
+    WorkspaceCacheModule,
   ],
   providers: [
     ConnectionProviderService,

packages/twenty-server/src/engine/metadata-modules/flat-entity/constant/__tests__/__snapshots__/all-universal-flat-entity-properties-to-compare-and-stringify.constant.spec.ts.snap

@@ -60,6 +60,7 @@ exports[`ALL_UNIVERSAL_FLAT_ENTITY_PROPERTIES_TO_COMPARE_AND_STRINGIFY should ma
       "displayName",
       "type",
       "oauthConfig",
+      "onConnectLogicFunctionUniversalIdentifier",
     ],
     "propertiesToStringify": [
       "oauthConfig",

packages/twenty-server/src/engine/metadata-modules/flat-entity/constant/all-entity-properties-configuration-by-metadata-name.constant.ts

@@ -1802,6 +1802,11 @@ export const ALL_ENTITY_PROPERTIES_CONFIGURATION_BY_METADATA_NAME = {
       toStringify: true,
       universalProperty: undefined,
     },
+    onConnectLogicFunctionUniversalIdentifier: {
+      toCompare: true,
+      toStringify: false,
+      universalProperty: undefined,
+    },
     createdAt: {
       toCompare: false,
       toStringify: false,