Setup async process handling via SQS (#45)

* build a generalizable setup for async sqs handling

* setup sqs queues

* fix cfn

* fix missing max recieve count

* fix build entrypoints

* create the appropriate roles

* fix cfn

Signed-off-by: Dev Singh <[email protected]>

* enable partial batch item failures

Signed-off-by: Dev Singh <[email protected]>

* fix consumer

Signed-off-by: Dev Singh <[email protected]>

* fix alarm metric

Signed-off-by: Dev Singh <[email protected]>

* setup basic driver

* move membership pass creation into SQS queue

* move sqs unit tests

* fix unit test

* fix mobilewallet route test

* cleanup

* fix iam

---------

Signed-off-by: Dev Singh <[email protected]>

Author: devksingh4

Makefile

@@ -54,6 +54,7 @@ build: src/ cloudformation/ docs/
 	yarn -D
 	VITE_BUILD_HASH=$(GIT_HASH) yarn build
 	cp -r src/api/resources/ dist/api/resources
+	rm -rf dist/lambda/sqs
 	sam build --template-file cloudformation/main.yml
 
 local:

cloudformation/iam.yml

@@ -12,10 +12,14 @@ Parameters:
     AllowedPattern: ^[a-zA-Z0-9]+[a-zA-Z0-9-]+[a-zA-Z0-9]+$
   SesEmailDomain:
     Type: String
+  SqsQueueArn:
+    Type: String
 Resources:
   ApiLambdaIAMRole:
     Type: AWS::IAM::Role
     Properties:
+      ManagedPolicyArns:
+        - arn:aws:iam::aws:policy/service-role/AWSLambdaSQSQueueExecutionRole
       AssumeRolePolicyDocument:
         Version: '2012-10-17'
         Statement:
@@ -41,6 +45,14 @@ Resources:
                 ses:Recipients:
                   - "*@illinois.edu"
         PolicyName: ses-membership
+      - PolicyDocument:
+          Version: '2012-10-17'
+          Statement:
+          - Action:
+            - sqs:SendMessage
+            Effect: Allow
+            Resource: !Ref SqsQueueArn
+        PolicyName: lambda-sqs
       - PolicyDocument:
           Version: '2012-10-17'
           Statement:

cloudformation/main.yml

@@ -22,6 +22,14 @@ Parameters:
     Default: false
     Type: String
     AllowedValues: [true, false]
+  SqsLambdaTimeout:
+    Description: How long the SQS lambda is permitted to run (in seconds)
+    Default: 300
+    Type: Number
+  SqsMessageTimeout:
+    Description: MessageVisibilityTimeout for the SQS Lambda queue (should be at least 6xSqsLambdaTimeout)
+    Default: 1800
+    Type: Number
 
 Conditions:
   IsProd: !Equals [!Ref RunEnvironment, 'prod']
@@ -74,6 +82,7 @@ Resources:
         RunEnvironment: !Ref RunEnvironment
         LambdaFunctionName: !Sub ${ApplicationPrefix}-lambda
         SesEmailDomain: !FindInMap [General, !Ref RunEnvironment, SesDomain]
+        SqsQueueArn: !GetAtt AppSQSQueues.Outputs.MainQueueArn
 
   AppLogGroups:
     Type: AWS::Serverless::Application
@@ -83,6 +92,14 @@ Resources:
         LambdaFunctionName: !Sub ${ApplicationPrefix}-lambda
         LogRetentionDays: !FindInMap [General, !Ref RunEnvironment, LogRetentionDays]
 
+  AppSQSQueues:
+    Type: AWS::Serverless::Application
+    Properties:
+      Location: ./sqs.yml
+      Parameters:
+        QueueName: !Sub ${ApplicationPrefix}-sqs
+        MessageTimeout: !Ref SqsMessageTimeout
+
   IcalDomainProxy:
     Type: AWS::Serverless::Application
     Properties:
@@ -149,6 +166,40 @@ Resources:
             Path: /{proxy+}
             Method: ANY
 
+  AppSqsLambdaFunction:
+    Type: AWS::Serverless::Function
+    DependsOn:
+      - AppLogGroups
+    Properties:
+      Architectures: [arm64]
+      CodeUri: ../dist/sqsConsumer
+      AutoPublishAlias: live
+      Runtime: nodejs22.x
+      Description: !Sub "${ApplicationFriendlyName} SQS Lambda"
+      FunctionName: !Sub ${ApplicationPrefix}-sqs-lambda
+      Handler: index.handler
+      MemorySize: 512
+      Role: !GetAtt AppSecurityRoles.Outputs.MainFunctionRoleArn
+      Timeout: !Ref SqsLambdaTimeout
+      LoggingConfig:
+        LogGroup: !Sub /aws/lambda/${ApplicationPrefix}-lambda
+      Environment:
+        Variables:
+          RunEnvironment: !Ref RunEnvironment
+      VpcConfig:
+        Ipv6AllowedForDualStack: !If [ShouldAttachVpc, True, !Ref AWS::NoValue]
+        SecurityGroupIds: !If [ShouldAttachVpc, !FindInMap [EnvironmentToCidr, !Ref RunEnvironment, SecurityGroupIds], !Ref AWS::NoValue]
+        SubnetIds: !If [ShouldAttachVpc, !FindInMap [EnvironmentToCidr, !Ref RunEnvironment, SubnetIds], !Ref AWS::NoValue]
+
+  SQSLambdaEventMapping:
+    Type: AWS::Lambda::EventSourceMapping
+    Properties:
+      BatchSize: 5
+      EventSourceArn: !GetAtt AppSQSQueues.Outputs.MainQueueArn
+      FunctionName: !Sub ${ApplicationPrefix}-sqs-lambda
+      FunctionResponseTypes:
+        - ReportBatchItemFailures
+
   IamGroupRolesTable:
     Type: 'AWS::DynamoDB::Table'
     DeletionPolicy: "Retain"
@@ -348,6 +399,23 @@ Resources:
         - Name: 'ApiName'
           Value: !Sub ${ApplicationPrefix}-gateway
 
+
+  AppDLQMessagesAlarm:
+    Type: 'AWS::CloudWatch::Alarm'
+    Condition: IsProd
+    Properties:
+      AlarmName: !Sub ${ApplicationPrefix}-sqs-dlq
+      AlarmDescription: 'Items are present in the application DLQ, meaning some messages failed to process.'
+      Namespace: 'AWS/SQS'
+      MetricName: 'ApproximateNumberOfMessagesVisible'
+      Statistic: 'Sum'
+      Period: '60'
+      EvaluationPeriods: '1'
+      ComparisonOperator: 'GreaterThanThreshold'
+      Threshold: '0'
+      AlarmActions:
+        - !Ref AlertSNSArn
+
   APILambdaPermission:
     Type: AWS::Lambda::Permission
     Properties:

cloudformation/sqs.yml

@@ -0,0 +1,40 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Description: Stack SQS Queues
+Transform: AWS::Serverless-2016-10-31
+Parameters:
+  QueueName:
+    Type: String
+    AllowedPattern: ^[a-zA-Z0-9]+[a-zA-Z0-9-]+[a-zA-Z0-9]+$
+  MessageTimeout:
+    Type: Number
+Resources:
+  AppDLQ:
+    Type: AWS::SQS::Queue
+    Properties:
+      QueueName: !Sub ${QueueName}-dlq
+      VisibilityTimeout: !Ref MessageTimeout
+  AppQueue:
+    Type: AWS::SQS::Queue
+    Properties:
+      QueueName: !Ref QueueName
+      VisibilityTimeout: !Ref MessageTimeout
+      RedrivePolicy:
+        deadLetterTargetArn:
+          Fn::GetAtt:
+            - "AppDLQ"
+            - "Arn"
+        maxReceiveCount: 3
+
+Outputs:
+  MainQueueArn:
+    Description: Main Queue Arn
+    Value:
+      Fn::GetAtt:
+      - AppQueue
+      - Arn
+  DLQArn:
+    Description: Dead-letter Queue Arn
+    Value:
+      Fn::GetAtt:
+      - AppDLQ
+      - Arn

package.json

@@ -11,13 +11,13 @@
   "scripts": {
     "build": "yarn workspaces run build && yarn lockfile-manage",
     "dev": "concurrently --names 'api,ui' 'yarn workspace infra-core-api run dev' 'yarn workspace infra-core-ui run dev'",
-    "lockfile-manage": "synp --with-workspace --source-file yarn.lock && cp package-lock.json dist/lambda/ && cp src/api/package.lambda.json dist/lambda/package.json && rm package-lock.json",
+    "lockfile-manage": "synp --with-workspace --source-file yarn.lock && cp package-lock.json dist/lambda/ && cp package-lock.json dist/sqsConsumer/ && cp src/api/package.lambda.json dist/lambda/package.json && cp src/api/package.lambda.json dist/sqsConsumer/package.json && rm package-lock.json",
     "prettier": "yarn workspaces run prettier && prettier --check tests/**/*.ts",
     "prettier:write": "yarn workspaces run prettier:write && prettier --write tests/**/*.ts",
     "lint": "yarn workspaces run lint",
     "prepare": "node .husky/install.mjs || true",
     "typecheck": "yarn workspaces run typecheck",
-    "test:unit": "vitest run tests/unit --config tests/unit/vitest.config.ts && yarn workspace infra-core-ui run test:unit",
+    "test:unit": "cross-env RunEnvironment='dev' vitest run tests/unit --config tests/unit/vitest.config.ts && yarn workspace infra-core-ui run test:unit",
     "test:unit-ui": "yarn test:unit --ui",
     "test:unit-watch": "vitest tests/unit",
     "test:live": "vitest tests/live",
@@ -39,7 +39,7 @@
     "@typescript-eslint/parser": "^8.0.1",
     "@vitejs/plugin-react": "^4.3.1",
     "@vitest/ui": "^2.0.5",
-    "aws-sdk-client-mock": "^4.0.1",
+    "aws-sdk-client-mock": "^4.1.0",
     "concurrently": "^9.1.2",
     "cross-env": "^7.0.3",
     "esbuild": "^0.23.0",
@@ -81,4 +81,4 @@
   "resolutions": {
     "pdfjs-dist": "^4.8.69"
   }
-}
+}

src/api/build.js

@@ -1,39 +1,56 @@
 import esbuild from "esbuild";
 import { resolve } from "path";
 
+
+const commonParams = {
+  bundle: true,
+  format: "esm",
+  minify: true,
+  outExtension: { ".js": ".mjs" },
+  loader: {
+    ".png": "file",
+    ".pkpass": "file",
+    ".json": "file",
+  }, // File loaders
+  target: "es2022", // Target ES2022
+  sourcemap: false,
+  platform: "node",
+  external: ["aws-sdk", "moment-timezone", "passkit-generator", "fastify"],
+  alias: {
+    'moment-timezone': resolve(process.cwd(), '../../node_modules/moment-timezone/builds/moment-timezone-with-data-10-year-range.js')
+  },
+  banner: {
+    js: `
+      import path from 'path';
+      import { fileURLToPath } from 'url';
+      import { createRequire as topLevelCreateRequire } from 'module';
+      const require = topLevelCreateRequire(import.meta.url);
+      const __filename = fileURLToPath(import.meta.url);
+      const __dirname = path.dirname(__filename);
+    `.trim(),
+  }, // Banner for compatibility with CommonJS
+}
 esbuild
   .build({
-    entryPoints: ["api/lambda.js"], // Entry file
-    bundle: true,
-    format: "esm",
-    minify: true,
+    ...commonParams,
+    entryPoints: ["api/lambda.js"],
     outdir: "../../dist/lambda/",
-    outExtension: { ".js": ".mjs" },
-    loader: {
-      ".png": "file",
-      ".pkpass": "file",
-      ".json": "file",
-    }, // File loaders
-    target: "es2022", // Target ES2022
-    sourcemap: false,
-    platform: "node",
-    external: ["aws-sdk", "moment-timezone", "passkit-generator", "fastify"],
-    alias: {
-      'moment-timezone': resolve(process.cwd(), '../../node_modules/moment-timezone/builds/moment-timezone-with-data-10-year-range.js')
-    },
-    banner: {
-      js: `
-        import path from 'path';
-        import { fileURLToPath } from 'url';
-        import { createRequire as topLevelCreateRequire } from 'module';
-        const require = topLevelCreateRequire(import.meta.url);
-        const __filename = fileURLToPath(import.meta.url);
-        const __dirname = path.dirname(__filename);
-      `.trim(),
-    }, // Banner for compatibility with CommonJS
+    external: [...commonParams.external, "sqs/*"],
+  })
+  .then(() => console.log("API server build completed successfully!"))
+  .catch((error) => {
+    console.error("API server build failed:", error);
+    process.exit(1);
+  });
+
+  esbuild
+  .build({
+    ...commonParams,
+    entryPoints: ["api/sqs/index.js", "api/sqs/driver.js"],
+    outdir: "../../dist/sqsConsumer/",
   })
-  .then(() => console.log("Build completed successfully!"))
+  .then(() => console.log("SQS consumer build completed successfully!"))
   .catch((error) => {
-    console.error("Build failed:", error);
+    console.error("SQS consumer build failed:", error);
     process.exit(1);
   });

src/api/functions/entraId.ts

@@ -21,22 +21,22 @@ import {
 } from "../../common/types/iam.js";
 import { FastifyInstance } from "fastify";
 import { UserProfileDataBase } from "common/types/msGraphApi.js";
+import { SecretsManagerClient } from "@aws-sdk/client-secrets-manager";
+import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
 
 function validateGroupId(groupId: string): boolean {
   const groupIdPattern = /^[a-zA-Z0-9-]+$/; // Adjust the pattern as needed
   return groupIdPattern.test(groupId);
 }
 
 export async function getEntraIdToken(
-  fastify: FastifyInstance,
+  clients: { smClient: SecretsManagerClient; dynamoClient: DynamoDBClient },
   clientId: string,
   scopes: string[] = ["https://graph.microsoft.com/.default"],
 ) {
   const secretApiConfig =
-    (await getSecretValue(
-      fastify.secretsManagerClient,
-      genericConfig.ConfigSecretName,
-    )) || {};
+    (await getSecretValue(clients.smClient, genericConfig.ConfigSecretName)) ||
+    {};
   if (
     !secretApiConfig.entra_id_private_key ||
     !secretApiConfig.entra_id_thumbprint
@@ -50,7 +50,7 @@ export async function getEntraIdToken(
     "base64",
   ).toString("utf8");
   const cachedToken = await getItemFromCache(
-    fastify.dynamoClient,
+    clients.dynamoClient,
     "entra_id_access_token",
   );
   if (cachedToken) {
@@ -80,7 +80,7 @@ export async function getEntraIdToken(
     date.setTime(date.getTime() - 30000);
     if (result?.accessToken) {
       await insertItemIntoCache(
-        fastify.dynamoClient,
+        clients.dynamoClient,
         "entra_id_access_token",
         { token: result?.accessToken },
         date,