diff --git a/packages/twenty-docs/developers/extend/apps/logic/logic-functions.mdx b/packages/twenty-docs/developers/extend/apps/logic/logic-functions.mdx index 8d6ed74c98..5032276247 100644 --- a/packages/twenty-docs/developers/extend/apps/logic/logic-functions.mdx +++ b/packages/twenty-docs/developers/extend/apps/logic/logic-functions.mdx @@ -59,7 +59,7 @@ To invoke a route-triggered logic function from a (headless) front component, se - **cron**: Runs your function on a schedule using a CRON expression. - **databaseEvent**: Runs on workspace object lifecycle events. When the event operation is `updated`, specific fields to listen to can be specified in the `updatedFields` array. If left undefined or empty, any update will trigger the function. > e.g. `person.updated`, `*.created`, `company.*` -- **serverRoute**: Exposes a single registration-scoped HTTP route. A **resolver** function (declared with `serverRouteTriggerSettings`) runs in the owner workspace and returns the target workspace AND the target logic function to dispatch to; the platform then runs that **target** function and returns its response. See [Server route trigger](#server-route-trigger). +- **serverRoute**: Exposes a single registration-scoped HTTP route. A **resolver** function (declared with `serverRouteTriggerSettings`) runs in the owner workspace and returns the target workspace AND the target logic function to dispatch to; the platform acks with `202` and runs that **target** function on the worker queue. See [Server route trigger](#server-route-trigger). You can also manually execute a function using the CLI: @@ -290,7 +290,7 @@ For request signatures, most providers sign with HMAC-SHA256; the parts that dif The resolver example above already shows the GitHub HMAC-SHA256 flow — adapt the header name, digest encoding, and signed-payload string per the provider you're integrating. -The target runs **synchronously** and its returned value becomes the HTTP response, so callers see your status code and can retry on non-2xx. Keep both handlers fast — some providers (e.g. Slack) time out in a few seconds. Because the resolver is reachable as a public endpoint, protect it with rate limiting at your edge. +The route responds `202 { queued: true }` right after the resolver returns and the target runs on the worker queue — the caller never observes the target's latency, result, or failures (those are recorded in execution logs). This keeps sender redeliveries from amplifying processing slowdowns, which is what you want for webhook ingestion. For endpoints whose caller must read the response body (challenge handshakes, Slack commands), use an `httpRouteTriggerSettings` route instead. Keep the resolver fast — some providers (e.g. Slack) time out in a few seconds. Because the resolver is reachable as a public endpoint, protect it with rate limiting at your edge. #### Database event trigger payload diff --git a/packages/twenty-server/src/engine/core-modules/server-route-trigger/__tests__/server-route-trigger.service.spec.ts b/packages/twenty-server/src/engine/core-modules/server-route-trigger/__tests__/server-route-trigger.service.spec.ts index edfc4cf159..be64576652 100644 --- a/packages/twenty-server/src/engine/core-modules/server-route-trigger/__tests__/server-route-trigger.service.spec.ts +++ b/packages/twenty-server/src/engine/core-modules/server-route-trigger/__tests__/server-route-trigger.service.spec.ts @@ -7,6 +7,8 @@ import { LogicFunctionExecutionExceptionCode, type LogicFunctionExecutorService, } from 'src/engine/core-modules/logic-function/logic-function-executor/logic-function-executor.service'; +import { LogicFunctionTriggerJob } from 'src/engine/core-modules/logic-function/logic-function-trigger/jobs/logic-function-trigger.job'; +import { type MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; import { ServerRouteTriggerExceptionCode } from 'src/engine/core-modules/server-route-trigger/exceptions/server-route-trigger.exception'; import { ServerRouteTriggerService } from 'src/engine/core-modules/server-route-trigger/server-route-trigger.service'; import { type LogicFunctionEntity } from 'src/engine/metadata-modules/logic-function/logic-function.entity'; @@ -72,6 +74,7 @@ describe('ServerRouteTriggerService', () => { let logicFunctionExecutorService: jest.Mocked< Pick >; + let messageQueueService: jest.Mocked>; let resolverRow: unknown; let queryBuilder: QueryBuilderMock; @@ -101,54 +104,57 @@ describe('ServerRouteTriggerService', () => { createQueryBuilder: jest.fn(() => queryBuilder), findOne: jest .fn() - // resolver lookup inside runFunction + // resolver lookup .mockResolvedValueOnce({ id: 'resolver-id' }) - // target lookup inside runFunction + // target lookup before enqueueing .mockResolvedValueOnce({ id: 'target-id' }), }; logicFunctionExecutorService = { - execute: jest - .fn() - // resolver returns { workspaceId, targetLogicFunctionUniversalIdentifier, payload } - .mockResolvedValueOnce( - buildExecuteResult({ - workspaceId: 'target-ws', - targetLogicFunctionUniversalIdentifier: TARGET_UID, - payload: { from: 'resolver' }, - }), - ) - // target returns the final response body - .mockResolvedValueOnce(buildExecuteResult({ ok: true })), + execute: jest.fn().mockResolvedValueOnce( + buildExecuteResult({ + workspaceId: 'target-ws', + targetLogicFunctionUniversalIdentifier: TARGET_UID, + payload: { from: 'resolver' }, + }), + ), + }; + + messageQueueService = { + add: jest.fn().mockResolvedValue(undefined), }; service = new ServerRouteTriggerService( logicFunctionRepository as unknown as Repository, logicFunctionExecutorService as unknown as LogicFunctionExecutorService, + messageQueueService as unknown as MessageQueueService, ); }); - it('runs the resolver in the owner workspace then the resolver-named target in the resolved workspace', async () => { + it('runs the resolver synchronously, enqueues the target, and acks with 202', async () => { const result = await handle(); - expect(logicFunctionExecutorService.execute).toHaveBeenNthCalledWith( - 1, + expect(logicFunctionExecutorService.execute).toHaveBeenCalledTimes(1); + expect(logicFunctionExecutorService.execute).toHaveBeenCalledWith( expect.objectContaining({ logicFunctionId: 'resolver-id', workspaceId: 'owner-ws', }), ); - expect(logicFunctionExecutorService.execute).toHaveBeenNthCalledWith( - 2, - expect.objectContaining({ - logicFunctionId: 'target-id', - workspaceId: 'target-ws', - payload: { from: 'resolver' }, - }), + expect(messageQueueService.add).toHaveBeenCalledWith( + LogicFunctionTriggerJob.name, + [ + { + logicFunctionId: 'target-id', + workspaceId: 'target-ws', + payload: { from: 'resolver' }, + }, + ], + { retryLimit: 3 }, ); expect(result).toEqual( expect.objectContaining({ - statusCode: 200, - body: { ok: true }, + statusCode: 202, + body: { queued: true }, }), ); }); @@ -184,6 +190,21 @@ describe('ServerRouteTriggerService', () => { ); }); + it('scopes the target lookup to the resolver application registration', async () => { + await handle(); + + expect(logicFunctionRepository.findOne).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ + where: expect.objectContaining({ + universalIdentifier: TARGET_UID, + workspaceId: 'target-ws', + application: { applicationRegistrationId: 'reg-1' }, + }), + }), + ); + }); + it('throws LOGIC_FUNCTION_NOT_FOUND and executes nothing when the query returns no server-route resolver', async () => { resolverRow = null; @@ -191,6 +212,7 @@ describe('ServerRouteTriggerService', () => { code: ServerRouteTriggerExceptionCode.LOGIC_FUNCTION_NOT_FOUND, }); expect(logicFunctionExecutorService.execute).not.toHaveBeenCalled(); + expect(messageQueueService.add).not.toHaveBeenCalled(); }); it('rejects and executes nothing when the resolver requires authentication', async () => { @@ -240,35 +262,16 @@ describe('ServerRouteTriggerService', () => { }); }); - it('throws LOGIC_FUNCTION_NOT_FOUND when the target named by the resolver is missing in the resolved workspace', async () => { + it('throws LOGIC_FUNCTION_NOT_FOUND and enqueues nothing when the target is missing', async () => { logicFunctionRepository.findOne.mockReset(); logicFunctionRepository.findOne - // resolver lookup succeeds .mockResolvedValueOnce({ id: 'resolver-id' }) - // target lookup returns null .mockResolvedValueOnce(null); await expect(handle()).rejects.toMatchObject({ code: ServerRouteTriggerExceptionCode.LOGIC_FUNCTION_NOT_FOUND, }); - }); - - it('surfaces a target userError as a server-route exception', async () => { - logicFunctionExecutorService.execute.mockReset(); - logicFunctionExecutorService.execute - .mockResolvedValueOnce( - buildExecuteResult({ - workspaceId: 'target-ws', - targetLogicFunctionUniversalIdentifier: TARGET_UID, - }), - ) - .mockResolvedValueOnce( - buildExecuteResult(null, { errorMessage: 'boom' }), - ); - - await expect(handle()).rejects.toMatchObject({ - code: ServerRouteTriggerExceptionCode.SERVER_ROUTE_USER_UNCAUGHT_ERROR, - }); + expect(messageQueueService.add).not.toHaveBeenCalled(); }); it('maps a LogicFunctionExecutionException(LOGIC_FUNCTION_NOT_FOUND) to the server-route not-found code', async () => { @@ -285,15 +288,6 @@ describe('ServerRouteTriggerService', () => { }); }); - it('falls back to PLATFORM_ERROR for any other thrown executor error', async () => { - logicFunctionExecutorService.execute.mockReset(); - logicFunctionExecutorService.execute.mockRejectedValue(new Error('boom')); - - await expect(handle()).rejects.toMatchObject({ - code: ServerRouteTriggerExceptionCode.SERVER_ROUTE_PLATFORM_ERROR, - }); - }); - it('maps a LogicFunctionExecutionException(RATE_LIMIT_EXCEEDED) to the server-route rate-limit code', async () => { logicFunctionExecutorService.execute.mockReset(); logicFunctionExecutorService.execute.mockRejectedValue( @@ -308,33 +302,11 @@ describe('ServerRouteTriggerService', () => { }); }); - it('scopes the target lookup to the resolver application registration', async () => { - await handle(); - - expect(logicFunctionRepository.findOne).toHaveBeenNthCalledWith( - 2, - expect.objectContaining({ - where: expect.objectContaining({ - universalIdentifier: TARGET_UID, - workspaceId: 'target-ws', - application: { applicationRegistrationId: 'reg-1' }, - }), - }), - ); - }); - - it('does not leak the raw executor error message to the caller', async () => { + it('does not leak the raw resolver executor error message to the caller', async () => { logicFunctionExecutorService.execute.mockReset(); - logicFunctionExecutorService.execute - .mockResolvedValueOnce( - buildExecuteResult({ - workspaceId: 'target-ws', - targetLogicFunctionUniversalIdentifier: TARGET_UID, - }), - ) - .mockRejectedValueOnce( - new Error('internal: connection to lambda-internal:5000 refused'), - ); + logicFunctionExecutorService.execute.mockRejectedValue( + new Error('internal: connection to lambda-internal:5000 refused'), + ); await expect(handle()).rejects.toMatchObject({ code: ServerRouteTriggerExceptionCode.SERVER_ROUTE_PLATFORM_ERROR, diff --git a/packages/twenty-server/src/engine/core-modules/server-route-trigger/server-route-trigger.service.ts b/packages/twenty-server/src/engine/core-modules/server-route-trigger/server-route-trigger.service.ts index 6c374d770b..520249024f 100644 --- a/packages/twenty-server/src/engine/core-modules/server-route-trigger/server-route-trigger.service.ts +++ b/packages/twenty-server/src/engine/core-modules/server-route-trigger/server-route-trigger.service.ts @@ -11,11 +11,15 @@ import { LogicFunctionExecutionExceptionCode, LogicFunctionExecutorService, } from 'src/engine/core-modules/logic-function/logic-function-executor/logic-function-executor.service'; -import { buildLogicFunctionEvent } from 'src/engine/core-modules/logic-function/logic-function-trigger/triggers/route/utils/build-logic-function-event.util'; import { - type RouteTriggerResponse, - buildRouteTriggerResponse, -} from 'src/engine/core-modules/logic-function/logic-function-trigger/triggers/route/utils/route-trigger-response.util'; + LogicFunctionTriggerJob, + type LogicFunctionTriggerJobData, +} from 'src/engine/core-modules/logic-function/logic-function-trigger/jobs/logic-function-trigger.job'; +import { buildLogicFunctionEvent } from 'src/engine/core-modules/logic-function/logic-function-trigger/triggers/route/utils/build-logic-function-event.util'; +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 { type RouteTriggerResponse } from 'src/engine/core-modules/logic-function/logic-function-trigger/triggers/route/utils/route-trigger-response.util'; import { ServerRouteTriggerException, ServerRouteTriggerExceptionCode, @@ -28,6 +32,8 @@ type ResolverResult = { payload?: object; }; +const QUEUED_TARGET_RETRY_LIMIT = 3; + @Injectable() export class ServerRouteTriggerService { private readonly logger = new Logger(ServerRouteTriggerService.name); @@ -36,6 +42,8 @@ export class ServerRouteTriggerService { @InjectRepository(LogicFunctionEntity) private readonly logicFunctionRepository: Repository, private readonly logicFunctionExecutorService: LogicFunctionExecutorService, + @InjectMessageQueue(MessageQueue.logicFunctionQueue) + private readonly messageQueueService: MessageQueueService, ) {} async handle({ @@ -89,22 +97,13 @@ export class ServerRouteTriggerService { }); const resolved = this.parseResolverResult(resolverResult); - const targetResult = await this.runFunction({ + return await this.enqueueTargetFunction({ logicFunctionUniversalIdentifier: resolved.targetLogicFunctionUniversalIdentifier, workspaceId: resolved.workspaceId, payload: resolved.payload ?? event, applicationRegistrationId, }); - - if (isDefined(targetResult.error)) { - throw new ServerRouteTriggerException( - targetResult.error.errorMessage, - ServerRouteTriggerExceptionCode.SERVER_ROUTE_USER_UNCAUGHT_ERROR, - ); - } - - return buildRouteTriggerResponse(targetResult.data); } private async findResolver({ @@ -169,7 +168,7 @@ export class ServerRouteTriggerService { }; } - private async runFunction({ + private async enqueueTargetFunction({ logicFunctionUniversalIdentifier, workspaceId, payload, @@ -178,8 +177,38 @@ export class ServerRouteTriggerService { logicFunctionUniversalIdentifier: string; workspaceId: string; payload: object; + applicationRegistrationId: string; + }): Promise { + const logicFunction = await this.findLogicFunctionOrFail({ + logicFunctionUniversalIdentifier, + workspaceId, + applicationRegistrationId, + }); + + await this.messageQueueService.add( + LogicFunctionTriggerJob.name, + [ + { + logicFunctionId: logicFunction.id, + workspaceId, + payload, + }, + ], + { retryLimit: QUEUED_TARGET_RETRY_LIMIT }, + ); + + return { statusCode: 202, headers: {}, body: { queued: true } }; + } + + private async findLogicFunctionOrFail({ + logicFunctionUniversalIdentifier, + workspaceId, + applicationRegistrationId, + }: { + logicFunctionUniversalIdentifier: string; + workspaceId: string; applicationRegistrationId?: string; - }): Promise<{ data: object | null; error?: { errorMessage: string } }> { + }): Promise { const logicFunction = await this.logicFunctionRepository.findOne({ where: { universalIdentifier: logicFunctionUniversalIdentifier, @@ -200,6 +229,23 @@ export class ServerRouteTriggerService { ); } + return logicFunction; + } + + private async runFunction({ + logicFunctionUniversalIdentifier, + workspaceId, + payload, + }: { + logicFunctionUniversalIdentifier: string; + workspaceId: string; + payload: object; + }): Promise<{ data: object | null; error?: { errorMessage: string } }> { + const logicFunction = await this.findLogicFunctionOrFail({ + logicFunctionUniversalIdentifier, + workspaceId, + }); + try { return await this.logicFunctionExecutorService.execute({ logicFunctionId: logicFunction.id, diff --git a/packages/twenty-server/test/integration/server-route-trigger/suites/server-route-trigger-authorization.integration-spec.ts b/packages/twenty-server/test/integration/server-route-trigger/suites/server-route-trigger-authorization.integration-spec.ts index b4a95667e1..8ac260e9c7 100644 --- a/packages/twenty-server/test/integration/server-route-trigger/suites/server-route-trigger-authorization.integration-spec.ts +++ b/packages/twenty-server/test/integration/server-route-trigger/suites/server-route-trigger-authorization.integration-spec.ts @@ -168,13 +168,13 @@ describe('ServerRouteTrigger authorization (integration)', () => { }); }); - it('dispatches a server-route-exposed resolver and returns the target function response', async () => { + it('dispatches a server-route-exposed resolver, queues the target, and acks with 202', async () => { const response = await request(baseUrl) .post(`/webhooks/server/${EXPOSED_RESOLVER_UNIVERSAL_IDENTIFIER}`) .send({ any: 'payload' }); - expect(response.status).toBe(200); - expect(response.body).toEqual(TARGET_FUNCTION_RESPONSE); + expect(response.status).toBe(202); + expect(response.body).toEqual({ queued: true }); }, 60000); it('rejects a server-route-exposed resolver that requires authentication before executing it', async () => {