fix: create queue for Tiles Find Single Row action (#865)

## Problem
* The findSingleRow action in Tiles encounters ‘Throughput Limit
Exceeded’ errors when there are a high number of executions attempting
to find rows.

## Solution
* Create a queue to rate limit the number of findSingleRow actions

**BEFORE**:
<img width="1512" alt="Screenshot 2025-02-17 at 3 27 40 PM"
src="https://github.com/user-attachments/assets/f53ff49c-1cde-44c8-a7a1-faad17ddaf8f"
/>

**AFTER**:
<img width="1496" alt="Screenshot 2025-02-17 at 3 29 49 PM"
src="https://github.com/user-attachments/assets/bfef0b2e-2b27-4799-af16-8d2026435994"
/>

## Tests
- [x] Test that existing executions with Tile actions execute as
execpted
- [x] Verify from admin dashboard that multiple concurrent findSingleRow
actions on the same Tile are rate limited, and that only

## Environment Variables
Note that rate limit defaults to 1 findSingleRow action per second.
If a different rate limit is desired, create a new environment variable
in the parameter store named
`plumber-<ENVIRONMENT>-tiles-interval-between-find-single-row-ms`

Author: kevinkim-ogp

ecs/env.json

@@ -221,6 +221,10 @@
     {
       "name": "ADMIN_JWT_SECRET_KEY",
       "valueFrom": "plumber-<ENVIRONMENT>-admin-jwt-secret-key"
+    },
+    {
+      "name": "TILES_INTERVAL_BETWEEN_FIND_SINGLE_ROW_MS",
+      "valueFrom": "plumber-<ENVIRONMENT>-tiles-interval-between-find-single-row-ms"
     }
   ]
 }

packages/backend/src/apps/tiles/__tests__/queue.test.ts

@@ -0,0 +1,75 @@
+// Avoid cyclic imports when importing m365ExcelApp
+import '@/apps'
+
+import { afterEach, describe, expect, it, vi } from 'vitest'
+
+import tilesApp from '..'
+
+const mocks = vi.hoisted(() => ({
+  stepQueryResult: vi.fn(),
+}))
+
+vi.mock('@/models/step', () => ({
+  default: {
+    query: vi.fn(() => ({
+      findById: vi.fn(() => ({
+        throwIfNotFound: mocks.stepQueryResult,
+      })),
+    })),
+  },
+}))
+
+describe('Queue config', () => {
+  afterEach(() => {
+    vi.restoreAllMocks()
+  })
+
+  it('configures a delayable queue', () => {
+    expect(tilesApp.queue.isQueueDelayable).toEqual(true)
+  })
+
+  it('sets group ID to the file ID', async () => {
+    mocks.stepQueryResult.mockResolvedValueOnce({
+      parameters: {
+        tableId: 'mock-table-id',
+      },
+      key: 'findSingleRow',
+      appKey: 'tiles',
+    })
+    const groupConfig = await tilesApp.queue.getGroupConfigForJob({
+      flowId: 'test-flow-id',
+      stepId: 'test-step-id',
+      executionId: 'test-step-id',
+    })
+    expect(groupConfig).toEqual({
+      id: 'mock-table-id-findSingleRow',
+    })
+  })
+
+  it('sets group ID to null', async () => {
+    mocks.stepQueryResult.mockResolvedValueOnce({
+      parameters: {
+        tableId: 'mock-table-id',
+      },
+      key: 'createRow',
+      appKey: 'tiles',
+    })
+    const groupConfig = await tilesApp.queue.getGroupConfigForJob({
+      flowId: 'test-flow-id',
+      stepId: 'test-step-id',
+      executionId: 'test-step-id',
+    })
+    expect(groupConfig).toEqual(null)
+  })
+
+  it('sets group concurrency to 1', () => {
+    expect(tilesApp.queue.groupLimits).toEqual({
+      type: 'concurrency',
+      concurrency: 1,
+    })
+  })
+
+  it('avoids bursting via a leaky bucket approach', () => {
+    expect(tilesApp.queue.queueRateLimit.max).toEqual(1)
+  })
+})

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

@@ -3,6 +3,7 @@ import { IApp } from '@plumber/types'
 import getTransferDetails from './common/get-transfer-details'
 import actions from './actions'
 import dynamicData from './dynamic-data'
+import queue from './queue'
 
 const app: IApp = {
   name: 'Tiles',
@@ -17,6 +18,7 @@ const app: IApp = {
   actions,
   dynamicData,
   getTransferDetails,
+  queue,
 }
 
 export default app

packages/backend/src/apps/tiles/queue/index.ts

@@ -0,0 +1,48 @@
+import type { IAppQueue } from '@plumber/types'
+
+import { TILES_INTERVAL_BETWEEN_FIND_SINGLE_ROW_MS } from '@/config/app-env-vars/tiles'
+import Step from '@/models/step'
+
+// Define actions that should be queued
+const QUEUED_ACTIONS = new Set(['findSingleRow'])
+
+//
+// This config sets up a per-Tile findSingleRow action queue, i.e., only findSingleRow
+// actions are grouped and rate limited.
+// All other Tile actions are not grouped and do not need to be rate limited.
+//
+// This is necessary because the findSingleRow action is the most expensive
+// operation in the Tile app, due to the need to scan the entire DynamoDB table.
+//
+
+const getGroupConfigForJob: IAppQueue['getGroupConfigForJob'] = async (
+  jobData,
+) => {
+  const step = await Step.query().findById(jobData.stepId).throwIfNotFound()
+  const tableId = step.parameters['tableId'] as string
+
+  if (QUEUED_ACTIONS.has(step.key)) {
+    return {
+      id: `${tableId}-${step.key}`,
+    }
+  }
+
+  // All other Tile actions are not grouped and do not need to be rate limited
+  // as the write operations are fast and less expensive.
+  return null
+}
+
+const queueSettings = {
+  getGroupConfigForJob,
+  groupLimits: {
+    type: 'concurrency',
+    concurrency: 1,
+  },
+  isQueueDelayable: true,
+  queueRateLimit: {
+    max: 1,
+    duration: TILES_INTERVAL_BETWEEN_FIND_SINGLE_ROW_MS,
+  },
+} satisfies IAppQueue
+
+export default queueSettings

packages/backend/src/config/app-env-vars/index.ts

@@ -24,3 +24,4 @@
 import './m365'
 import './formsg'
 import './postman-sms'
+import './tiles'

packages/backend/src/config/app-env-vars/tiles.ts

@@ -0,0 +1,4 @@
+// Default to 1 action per second.
+export const TILES_INTERVAL_BETWEEN_FIND_SINGLE_ROW_MS = Number(
+  process.env.TILES_INTERVAL_BETWEEN_FIND_SINGLE_ROW_MS ?? '1000',
+)