build a generalizable setup for async sqs handling

Author: devksingh4

src/api/build.js

@@ -1,39 +1,55 @@
 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/index.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
   })
-  .then(() => console.log("Build completed successfully!"))
+  .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"],
+    outdir: "../../dist/sqsConsumer/",
+  })
+  .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/package.json

@@ -25,6 +25,9 @@
     "@fastify/aws-lambda": "^5.0.0",
     "@fastify/caching": "^9.0.1",
     "@fastify/cors": "^10.0.1",
+    "@middy/core": "^6.0.0",
+    "@middy/event-normalizer": "^6.0.0",
+    "@middy/sqs-partial-batch-failure": "^6.0.0",
     "@touch4it/ical-timezones": "^1.9.0",
     "base64-arraybuffer": "^1.0.2",
     "discord.js": "^14.15.3",
@@ -39,13 +42,15 @@
     "moment-timezone": "^0.5.45",
     "node-cache": "^5.1.2",
     "passkit-generator": "^3.3.1",
+    "pino": "^9.6.0",
     "pluralize": "^8.0.0",
     "zod": "^3.23.8",
     "zod-to-json-schema": "^3.23.2",
     "zod-validation-error": "^3.3.1"
   },
   "devDependencies": {
     "@tsconfig/node22": "^22.0.0",
+    "@types/aws-lambda": "^8.10.147",
     "nodemon": "^3.1.9"
   }
 }

src/api/sqs/handlers.ts

@@ -0,0 +1,16 @@
+import { AvailableSQSFunctions } from "common/types/sqsMessage.js";
+import { SQSHandlerFunction } from "./index.js";
+
+export const emailMembershipPassHandler: SQSHandlerFunction<
+  AvailableSQSFunctions.EmailMembershipPass
+> = async (payload, metadata, logger) => {
+  logger.error("Not implemented yet!");
+  return;
+};
+
+export const pingHandler: SQSHandlerFunction<
+  AvailableSQSFunctions.Ping
+> = async (payload, metadata, logger) => {
+  logger.error("Not implemented yet!");
+  return;
+};

src/api/sqs/index.ts

@@ -0,0 +1,60 @@
+import middy from "@middy/core";
+import eventNormalizerMiddleware from "@middy/event-normalizer";
+import sqsPartialBatchFailure from "@middy/sqs-partial-batch-failure";
+import { Context, SQSEvent } from "aws-lambda";
+import {
+  parseSQSPayload,
+  SQSPayload,
+  sqsPayloadSchemas,
+  AvailableSQSFunctions,
+  SQSMessageMetadata,
+} from "../../common/types/sqsMessage.js";
+import { logger } from "./logger.js";
+import { z, ZodError } from "zod";
+import pino from "pino";
+import { emailMembershipPassHandler, pingHandler } from "./handlers.js";
+
+export type SQSFunctionPayloadTypes = {
+  [K in keyof typeof sqsPayloadSchemas]: SQSHandlerFunction<K>;
+};
+
+export type SQSHandlerFunction<T extends AvailableSQSFunctions> = (
+  payload: z.infer<(typeof sqsPayloadSchemas)[T]>["payload"],
+  metadata: SQSMessageMetadata,
+  logger: pino.Logger,
+) => Promise<void>;
+
+const handlers: SQSFunctionPayloadTypes = {
+  [AvailableSQSFunctions.EmailMembershipPass]: emailMembershipPassHandler,
+  [AvailableSQSFunctions.Ping]: pingHandler,
+};
+
+export const handler = middy()
+  .use(eventNormalizerMiddleware())
+  .use(sqsPartialBatchFailure())
+  .handler((event: SQSEvent, context: Context, { signal }) => {
+    const recordsPromises = event.Records.map(async (record, index) => {
+      try {
+        let parsedBody = parseSQSPayload(record.body);
+        if (parsedBody instanceof ZodError) {
+          logger.error(
+            { sqsMessageId: record.messageId },
+            parsedBody.toString(),
+          );
+        }
+        parsedBody = parsedBody as SQSPayload;
+        const childLogger = logger.child({
+          sqsMessageId: record.messageId,
+          metadata: parsedBody.metadata,
+        });
+        childLogger.info("Processing started.");
+        const func = handlers[parsedBody.function] as SQSHandlerFunction<
+          typeof parsedBody.function
+        >;
+        return func(parsedBody.payload, parsedBody.metadata, childLogger);
+      } catch (e: any) {
+        logger.error({ sqsMessageId: record.messageId }, e.toString());
+      }
+    });
+    return Promise.allSettled(recordsPromises);
+  });

src/api/sqs/logger.ts

@@ -0,0 +1,2 @@
+import { pino } from "pino";
+export const logger = pino();

src/common/types/sqsMessage.ts

@@ -0,0 +1,58 @@
+import { z, ZodError, ZodType } from "zod";
+
+export enum AvailableSQSFunctions {
+  Ping = "pong",
+  EmailMembershipPass = "emailMembershipPass",
+}
+
+const sqsMessageMetadataSchema = z.object({
+  taskId: z.string().min(1),
+  reqId: z.string().min(1),
+  initiator: z.string().min(1),
+});
+
+export type SQSMessageMetadata = z.infer<typeof sqsMessageMetadataSchema>;
+
+const baseSchema = z.object({
+  metadata: sqsMessageMetadataSchema,
+});
+
+const createSQSSchema = <T extends AvailableSQSFunctions, P extends ZodType<any>>(
+  func: T,
+  payloadSchema: P
+) =>
+  baseSchema.extend({
+    function: z.literal(func),
+    payload: payloadSchema,
+  });
+
+export const sqsPayloadSchemas = {
+  [AvailableSQSFunctions.Ping]: createSQSSchema(AvailableSQSFunctions.Ping, z.object({})),
+  [AvailableSQSFunctions.EmailMembershipPass]: createSQSSchema(
+    AvailableSQSFunctions.EmailMembershipPass,
+    z.object({ email: z.string().email() })
+  ),
+} as const;
+
+export const sqsPayloadSchema = z.discriminatedUnion(
+  "function",
+  [
+    sqsPayloadSchemas[AvailableSQSFunctions.Ping],
+    sqsPayloadSchemas[AvailableSQSFunctions.EmailMembershipPass],
+  ] as const
+);
+
+export type SQSPayload = z.infer<typeof sqsPayloadSchema>;
+
+export type SQSFunctionPayloadTypes = {
+  [K in keyof typeof sqsPayloadSchemas]: z.infer<(typeof sqsPayloadSchemas)[K]>;
+};
+
+export function parseSQSPayload(json: unknown): SQSPayload | ZodError {
+  const parsed = sqsPayloadSchema.safeParse(json);
+  if (parsed.success) {
+    return parsed.data;
+  } else {
+    return parsed.error;
+  }
+}