refactor: SqsFifoProcessor for functionalities used in fifo processor classes

Author: arnabrahman

packages/batch/src/SqsFifoPartialProcessor.ts

@@ -1,5 +1,6 @@
 import type { SQSRecord } from 'aws-lambda';
 import { BatchProcessorSync } from './BatchProcessorSync.js';
+import { SqsFifoProcessor } from './SqsFifoProcessor.js';
 import { EventType } from './constants.js';
 import {
   type BatchProcessingError,
@@ -46,17 +47,13 @@ import type {
  */
 class SqsFifoPartialProcessor extends BatchProcessorSync {
   /**
-   * The ID of the current message group being processed.
+   *  Processor for handling SQS FIFO message
    */
-  #currentGroupId?: string;
-  /**
-   * A set of group IDs that have already encountered failures.
-   */
-  #failedGroupIds: Set<string>;
+  readonly #processor: SqsFifoProcessor;
 
   public constructor() {
     super(EventType.SQS);
-    this.#failedGroupIds = new Set<string>();
+    this.#processor = new SqsFifoProcessor();
   }
 
   /**
@@ -70,9 +67,7 @@ class SqsFifoPartialProcessor extends BatchProcessorSync {
     record: EventSourceDataClassTypes,
     exception: Error
   ): FailureResponse {
-    if (this.options?.skipGroupOnError && this.#currentGroupId) {
-      this.#addToFailedGroup(this.#currentGroupId);
-    }
+    this.#processor.processFailureForCurrentGroup(this.options);
 
     return super.failureHandler(record, exception);
   }
@@ -101,24 +96,17 @@ class SqsFifoPartialProcessor extends BatchProcessorSync {
     const processedRecords: (SuccessResponse | FailureResponse)[] = [];
     let currentIndex = 0;
     for (const record of this.records) {
-      this.#setCurrentGroup((record as SQSRecord).attributes?.MessageGroupId);
+      this.#processor.setCurrentGroup(
+        (record as SQSRecord).attributes?.MessageGroupId
+      );
 
-      // If we have any failed messages, we should then short circuit the process and
-      // fail remaining messages unless `skipGroupOnError` is true
-      const shouldShortCircuit =
-        !this.options?.skipGroupOnError && this.failureMessages.length !== 0;
-      if (shouldShortCircuit) {
+      if (
+        this.#processor.shouldShortCircuit(this.failureMessages, this.options)
+      ) {
         return this.shortCircuitProcessing(currentIndex, processedRecords);
       }
 
-      // If `skipGroupOnError` is true and the current group has previously failed,
-      // then we should skip processing the current group.
-      const shouldSkipCurrentGroup =
-        this.options?.skipGroupOnError &&
-        this.#currentGroupId &&
-        this.#failedGroupIds.has(this.#currentGroupId);
-
-      const result = shouldSkipCurrentGroup
+      const result = this.#processor.shouldSkipCurrentGroup(this.options)
         ? this.#processFailRecord(
             record,
             new SqsFifoMessageGroupShortCircuitError()
@@ -161,15 +149,6 @@ class SqsFifoPartialProcessor extends BatchProcessorSync {
     return processedRecords;
   }
 
-  /**
-   * Adds the specified group ID to the set of failed group IDs.
-   *
-   * @param group - The group ID to be added to the set of failed group IDs.
-   */
-  #addToFailedGroup(group: string): void {
-    this.#failedGroupIds.add(group);
-  }
-
   /**
    * Processes a fail record.
    *
@@ -184,15 +163,6 @@ class SqsFifoPartialProcessor extends BatchProcessorSync {
 
     return this.failureHandler(data, exception);
   }
-
-  /**
-   * Sets the current group ID for the message being processed.
-   *
-   * @param group - The group ID of the current message being processed.
-   */
-  #setCurrentGroup(group?: string): void {
-    this.#currentGroupId = group;
-  }
 }
 
 export { SqsFifoPartialProcessor };

packages/batch/src/SqsFifoPartialProcessorAsync.ts

@@ -1,6 +1,6 @@
 import type { SQSRecord } from 'aws-lambda';
 import { BatchProcessor } from './BatchProcessor.js';
-import { SqsFifoProcessingUtils } from './SqsFifoProcessingUtils.js';
+import { SqsFifoProcessor } from './SqsFifoProcessor.js';
 import { EventType } from './constants.js';
 import {
   type BatchProcessingError,
@@ -47,13 +47,13 @@ import type {
  */
 class SqsFifoPartialProcessorAsync extends BatchProcessor {
   /**
-   * Utility class for processing SQS FIFO queues
+   *  Processor for handling SQS FIFO message
    */
-  readonly #utils: SqsFifoProcessingUtils;
+  readonly #processor: SqsFifoProcessor;
 
   public constructor() {
     super(EventType.SQS);
-    this.#utils = new SqsFifoProcessingUtils();
+    this.#processor = new SqsFifoProcessor();
   }
 
   /**
@@ -66,7 +66,7 @@ class SqsFifoPartialProcessorAsync extends BatchProcessor {
     record: EventSourceDataClassTypes,
     exception: Error
   ): FailureResponse {
-    this.#utils.processFailureForCurrentGroup(this.options);
+    this.#processor.processFailureForCurrentGroup(this.options);
 
     return super.failureHandler(record, exception);
   }
@@ -95,15 +95,17 @@ class SqsFifoPartialProcessorAsync extends BatchProcessor {
     const processedRecords: (SuccessResponse | FailureResponse)[] = [];
     let currentIndex = 0;
     for (const record of this.records) {
-      this.#utils.setCurrentGroup(
+      this.#processor.setCurrentGroup(
         (record as SQSRecord).attributes?.MessageGroupId
       );
 
-      if (this.#utils.shouldShortCircuit(this.failureMessages, this.options)) {
+      if (
+        this.#processor.shouldShortCircuit(this.failureMessages, this.options)
+      ) {
         return this.shortCircuitProcessing(currentIndex, processedRecords);
       }
 
-      const result = this.#utils.shouldSkipCurrentGroup(this.options)
+      const result = this.#processor.shouldSkipCurrentGroup(this.options)
         ? this.#processFailRecord(
             record,
             new SqsFifoMessageGroupShortCircuitError()

packages/batch/src/SqsFifoProcessingUtils.ts renamed to packages/batch/src/SqsFifoProcessor.ts

@@ -4,9 +4,11 @@ import type {
 } from './types.js';
 
 /**
- * Utility class to handle processing of SQS FIFO messages.
+ * Class representing a processor for SQS FIFO messages.
+ * This class provides utilities for handling message groups, including tracking failed groups,
+ * determining whether to short-circuit processing, and skipping groups based on processing options.
  */
-class SqsFifoProcessingUtils {
+class SqsFifoProcessor {
   /**
    * The ID of the current message group being processed.
    */
@@ -84,4 +86,4 @@ class SqsFifoProcessingUtils {
   }
 }
 
-export { SqsFifoProcessingUtils };
+export { SqsFifoProcessor };