feat: queues for postman email, tele and slack

Author: kevinkim-ogp

packages/backend/src/apps/postman/index.ts

@@ -1,5 +1,7 @@
 import { IApp } from '@plumber/types'
 
+import { getGenericAppQueue } from '@/queues/helpers/get-generic-app-queue'
+
 import actions from './actions'
 
 const app: IApp = {
@@ -18,6 +20,7 @@ const app: IApp = {
     url: 'https://demo.arcade.software/VppMAbGKfFXFEsKxnKiw?embed&show_copy_link=true',
     title: 'Setting up Email by Postman',
   },
+  queue: getGenericAppQueue('POSTMAN_EMAIL'),
 }
 
 export default app

packages/backend/src/apps/slack/index.ts

@@ -1,5 +1,7 @@
 import { IApp } from '@plumber/types'
 
+import { getGenericAppQueue } from '@/queues/helpers/get-generic-app-queue'
+
 import addAuthHeader from './common/add-auth-header'
 import actions from './actions'
 import auth from './auth'
@@ -17,6 +19,7 @@ const app: IApp = {
   auth,
   actions,
   dynamicData,
+  queue: getGenericAppQueue('SLACK'),
 }
 
 export default app

packages/backend/src/apps/telegram-bot/index.ts

@@ -1,5 +1,7 @@
 import { IApp } from '@plumber/types'
 
+import { getGenericAppQueue } from '@/queues/helpers/get-generic-app-queue'
+
 import addAuthHeader from './common/add-auth-header'
 import rateLimitHandler from './common/interceptor/rate-limit'
 import actions from './actions'
@@ -19,6 +21,7 @@ const app: IApp = {
   dynamicData,
   auth,
   actions,
+  queue: getGenericAppQueue('TELEGRAM'),
 }
 
 export default app

packages/backend/src/config/queues.ts

@@ -0,0 +1,5 @@
+export const QUEUE_CONCURRENCY = {
+  POSTMAN_EMAIL: 3,
+  SLACK: 3,
+  TELEGRAM: 3,
+}

packages/backend/src/queues/__tests__/helpers.get-generic-app-queue.test.ts

@@ -0,0 +1,53 @@
+import { describe, expect, it } from 'vitest'
+
+import { QUEUE_CONCURRENCY } from '@/config/queues'
+
+import { getGenericAppQueue } from '../helpers/get-generic-app-queue'
+
+describe('tests for apps with generic queue', () => {
+  // apps that have a generic queue
+  const apps = Object.keys(QUEUE_CONCURRENCY)
+
+  it.each(apps)(
+    'should return a generic queue with the correct group config for %s',
+    async (app) => {
+      const appKey = app as keyof typeof QUEUE_CONCURRENCY
+      const queue = getGenericAppQueue(appKey)
+      const groupConfig = await queue.getGroupConfigForJob({
+        flowId: 'some-flow-id',
+        executionId: 'some-execution-id',
+        stepId: 'some-step-id',
+      })
+      expect(groupConfig).toEqual({ id: app })
+    },
+  )
+
+  it.each(apps)(
+    'should return a generic queue with the correct concurrency for %s',
+    (app) => {
+      const appKey = app as keyof typeof QUEUE_CONCURRENCY
+      const queue = getGenericAppQueue(appKey)
+      expect(queue.groupLimits.type).toBe('concurrency')
+      if (queue.groupLimits.type === 'concurrency') {
+        expect(queue.groupLimits.concurrency).toBe(QUEUE_CONCURRENCY[appKey])
+      }
+    },
+  )
+
+  it('should return a generic queue with concurrency 2 if the app is not in the QUEUE_CONCURRENCY object', async () => {
+    const queue = getGenericAppQueue(
+      'NOT_A_REAL_APP' as keyof typeof QUEUE_CONCURRENCY,
+    )
+    const groupConfig = await queue.getGroupConfigForJob({
+      flowId: 'some-flow-id',
+      executionId: 'some-execution-id',
+      stepId: 'some-step-id',
+    })
+    expect(groupConfig).toEqual({ id: 'NOT_A_REAL_APP' })
+
+    expect(queue.groupLimits.type).toBe('concurrency')
+    if (queue.groupLimits.type === 'concurrency') {
+      expect(queue.groupLimits.concurrency).toBe(2)
+    }
+  })
+})

packages/backend/src/queues/helpers/get-generic-app-queue.ts

@@ -0,0 +1,28 @@
+import type { IAppQueue } from '@plumber/types'
+
+import { QUEUE_CONCURRENCY } from '@/config/queues'
+
+/**
+ * NOTE: this creates a generic queue for apps that do not require
+ * any special queue configuration.
+ * It adds a generic group id to the jobs to ensure that the
+ * concurrency limit is applied.
+ */
+export function getGenericAppQueue(
+  app: keyof typeof QUEUE_CONCURRENCY,
+): IAppQueue {
+  const concurrency = QUEUE_CONCURRENCY[app] || 2
+
+  const getGroupConfigForJob = async () => {
+    return { id: app }
+  }
+
+  return {
+    getGroupConfigForJob,
+    groupLimits: {
+      type: 'concurrency',
+      concurrency: concurrency,
+    },
+    isQueueDelayable: false,
+  } satisfies IAppQueue
+}