fix: log errors in app routing

youngsecurity · Aug 12, 2024 · 7615f6e · 7615f6e
1 parent 77a8ad6
commit 7615f6e
Showing 1 changed file with 37 additions and 32 deletions.
diff --git a/packages/server/api/src/app/app-event-routing/app-event-routing.module.ts b/packages/server/api/src/app/app-event-routing/app-event-routing.module.ts
@@ -12,7 +12,7 @@ import {
     isNil,
 } from '@activepieces/shared'
 import { FastifyPluginAsyncTypebox, Type } from '@fastify/type-provider-typebox'
-import { flowQueue } from '../workers/queue'
+import { StatusCodes } from 'http-status-codes'
 import { DEFAULT_PRIORITY } from '../workers/queue/queue-manager'
 import { appEventRoutingService } from './app-event-routing.service'
 
@@ -56,43 +56,48 @@ export const appEventRoutingController: FastifyPluginAsyncTypebox = async (
                 },
             })
         }
-        const pieceName = pieceNames[pieceUrl]
-        assertNotNullOrUndefined(piece.events, 'Event is not defined')
+        const appName = pieceNames[pieceUrl]
+        assertNotNullOrUndefined(piece.events, 'Event is possible in this piece')
         const { reply, event, identifierValue } = piece.events.parseAndReply({ payload })
+        if (!isNil(reply)) {
+            logger.info({
+                reply,
+                piece: pieceUrl,
+            }, '[AppEventRoutingController#event] reply')
+            return requestReply.status(StatusCodes.OK).headers(reply?.headers ?? {}).send(reply?.body ?? {})
+        }
         logger.info({
             event,
             identifierValue,
         }, '[AppEventRoutingController#event] event')
-        if (event && identifierValue) {
-            const listeners = await appEventRoutingService.listListeners({
-                appName: pieceName,
-                event,
-                identifierValue,
-            })
-            rejectedPromiseHandler(Promise.all(listeners.map(async (listener) => {
-                const requestId = apId()
-                await flowQueue.add({
-                    id: requestId,
-                    type: JobType.WEBHOOK,
-                    data: {
-                        projectId: listener.projectId,
-                        schemaVersion: LATEST_JOB_DATA_SCHEMA_VERSION,
-                        requestId,
-                        synchronousHandlerId: null,
-                        payload,
-                        flowId: listener.flowId,
-                        simulate: false,
-                    },
-                    priority: DEFAULT_PRIORITY,
-                })
-            })))
+        if (isNil(event) || isNil(identifierValue)) {
+            return requestReply.status(StatusCodes.BAD_REQUEST).send({})
         }
-        return requestReply
-            .status(200)
-            .headers(reply?.headers ?? {})
-            .send(reply?.body ?? {})
-    },
-    )
+        const listeners = await appEventRoutingService.listListeners({
+            appName,
+            event,
+            identifierValue,
+        })
+        const eventsQueue = listeners.map(async (listener) => {
+            const requestId = apId()
+            return {
+                id: requestId,
+                type: JobType.WEBHOOK,
+                data: {
+                    projectId: listener.projectId,
+                    schemaVersion: LATEST_JOB_DATA_SCHEMA_VERSION,
+                    requestId,
+                    synchronousHandlerId: null,
+                    payload,
+                    flowId: listener.flowId,
+                    simulate: false,
+                },
+                priority: DEFAULT_PRIORITY,
+            }
+        })
+        rejectedPromiseHandler(Promise.all(eventsQueue))
+        return requestReply.status(StatusCodes.OK).send({})
+    })
 }
 
 const EventRoutingParam = {