Release v1.40.1 (#961)

Fix:
* Too many sigterm listeners on process

Hotfixes:
* Slow GetFlows query due to inconsistent query plan
* Check that Trigger step, app and event are valid

Author: kevinkim-ogp

package-lock.json

@@ -106,5 +106,5 @@
     "tsconfig-paths": "^4.2.0",
     "type-fest": "4.10.3"
   },
-  "version": "1.40.0"
+  "version": "1.40.1"
 }

packages/backend/package.json

@@ -72,30 +72,41 @@ export default async (request: IRequest, response: Response) => {
 
   const testRun = !flow.active
   const triggerStep = await flow.getTriggerStep()
-  const triggerCommand = await triggerStep.getTriggerCommand()
-  const app = await triggerStep.getApp()
-  const isWebhookApp = app.key === 'webhook' || app.key === 'formsg'
 
-  // Allow all webhook test runs to work
-  if (testRun && !isWebhookApp) {
+  const triggerApp = await triggerStep?.getApp()
+
+  if (!triggerApp) {
+    logger.info(`Trigger app not set up for flow ${flowId}`)
+    return response.sendStatus(404)
+  }
+
+  const isWebhookApp =
+    triggerApp.key === 'webhook' || triggerApp.key === 'formsg'
+
+  if (!isWebhookApp) {
+    logger.info(`Invalid trigger app for flow${flowId}`)
     return response.sendStatus(404)
   }
 
+  const triggerCommand = await triggerStep?.getTriggerCommand()
   // If trigger event is not selected, this should also return 404
   if (triggerCommand?.type !== 'webhook') {
+    logger.info(
+      `Trigger command not found or is not webhook type for flow${flowId}`,
+    )
     return response.sendStatus(404)
   }
 
-  if (app.auth?.verifyWebhook) {
+  if (triggerApp.auth?.verifyWebhook) {
     const $ = await globalVariable({
       flow,
       connection: await triggerStep.$relatedQuery('connection'),
-      app,
+      app: triggerApp,
       step: triggerStep,
       request,
     })
 
-    const verified = await app.auth.verifyWebhook($)
+    const verified = await triggerApp.auth.verifyWebhook($)
 
     if (!verified) {
       return response.sendStatus(401)
@@ -131,7 +142,7 @@ export default async (request: IRequest, response: Response) => {
     flowId,
     executionId,
     stepId: triggerStep.id,
-    appKey: app.key,
+    appKey: triggerApp.key,
     testRun,
   })
 

packages/backend/src/controllers/webhooks/handler.ts

@@ -1,4 +1,5 @@
 import paginate from '@/helpers/pagination'
+import Flow from '@/models/flow'
 
 import type { QueryResolvers } from '../__generated__/types.generated'
 
@@ -7,31 +8,55 @@ const getFlows: QueryResolvers['getFlows'] = async (
   params,
   context,
 ) => {
-  const flowsQuery = context.currentUser
-    .$relatedQuery('flows')
-    .joinRelated({
-      steps: true,
+  const filteredFlowIds = (
+    await context.currentUser
+      .$relatedQuery('flows')
+      .distinct('id')
+      .where((builder) => {
+        if (params.name) {
+          builder.where('name', 'ilike', `%${params.name}%`)
+        }
+      })
+  ).map((f) => f.id)
+
+  if (!filteredFlowIds.length) {
+    return {
+      pageInfo: {
+        currentPage: 1,
+        totalCount: 0,
+      },
+      edges: [],
+    }
+  }
+
+  const flowsQuery = Flow.query()
+    .with('filtered_steps', (builder) => {
+      builder
+        .distinct('flow_id')
+        .from('steps')
+        .where((stepBuilder) => {
+          if (params.connectionId) {
+            stepBuilder.where('connection_id', params.connectionId)
+          }
+
+          if (params.appKey) {
+            stepBuilder.where('app_key', params.appKey)
+          }
+
+          stepBuilder.withSoftDeleted()
+        })
+        .whereNull('deleted_at')
+        .whereIn('flow_id', filteredFlowIds)
+        .withSoftDeleted()
     })
+    .innerJoin('filtered_steps', 'id', 'filtered_steps.flow_id')
     .withGraphFetched({
       steps: {
         connection: true,
       },
       pendingTransfer: true,
     })
-    .where((builder) => {
-      if (params.connectionId) {
-        builder.where('steps.connection_id', params.connectionId)
-      }
-
-      if (params.name) {
-        builder.where('flows.name', 'ilike', `%${params.name}%`)
-      }
-
-      if (params.appKey) {
-        builder.where('steps.app_key', params.appKey)
-      }
-    })
-    .groupBy('flows.id')
+    .groupBy('id')
     .orderBy('active', 'desc')
     .orderBy('updated_at', 'desc')
 

packages/backend/src/graphql/queries/get-flows.ts

@@ -7,6 +7,7 @@ import {
 } from '@taskforcesh/bullmq-pro'
 
 import apps from '@/apps'
+import logger from '@/helpers/logger'
 import { makeActionQueue } from '@/queues/helpers/make-action-queue'
 
 //
@@ -133,3 +134,10 @@ export async function getActionJob(
   const { queueName, bullMqJobId } = parseActionJobId(actionJobId)
   return await actionQueuesByName[queueName].getJob(bullMqJobId)
 }
+
+process.on('SIGTERM', async () => {
+  logger.info('SIGTERM: gracefully closing all action queues')
+  const allQueues = [mainActionQueue, ...Object.values(actionQueuesByName)]
+  await Promise.all(allQueues.map((q) => q?.close()))
+  logger.info('SIGTERM: all action queues closed')
+})

packages/backend/src/queues/action.ts

@@ -31,9 +31,5 @@ export function makeActionQueue(
     }
   })
 
-  process.on('SIGTERM', async () => {
-    await queue.close()
-  })
-
   return queue
 }

packages/backend/src/queues/helpers/make-action-queue.ts

@@ -1,4 +1,5 @@
 import apps from '@/apps'
+import logger from '@/helpers/logger'
 import {
   appActionQueues,
   MAIN_ACTION_QUEUE_NAME,
@@ -38,3 +39,10 @@ for (const [appKey, app] of Object.entries(apps)) {
     queueConfig: app.queue,
   })
 }
+
+process.on('SIGTERM', async () => {
+  logger.info('SIGTERM: gracefully closing all action workers')
+  const allWorkers = [mainActionWorker, ...Object.values(appActionWorkers)]
+  await Promise.all(allWorkers.map((w) => w?.close()))
+  logger.info('SIGTERM: all action workers closed')
+})

packages/backend/src/workers/action.ts

@@ -290,9 +290,5 @@ export function makeActionWorker(
     })
   })
 
-  process.on('SIGTERM', async () => {
-    await worker.close()
-  })
-
   return worker
 }

packages/backend/src/workers/helpers/make-action-worker.ts

@@ -1,6 +1,6 @@
 {
   "name": "frontend",
-  "version": "1.40.0",
+  "version": "1.40.1",
   "scripts": {
     "dev": "wait-on tcp:3000 && vite --host --force",
     "build": "tsc && vite build --mode=${VITE_MODE:-prod}",

packages/frontend/package.json

@@ -2,5 +2,5 @@
   "name": "@plumber/types",
   "description": "Shared types for plumber",
   "types": "./index.d.ts",
-  "version": "1.40.0"
+  "version": "1.40.1"
 }