refactor: use SqsFifoProcessingUtils inside SqsFifoPartialProcessorAsync

Author: arnabrahman

packages/batch/src/SqsFifoPartialProcessorAsync.ts

@@ -1,5 +1,6 @@
 import type { SQSRecord } from 'aws-lambda';
 import { BatchProcessor } from './BatchProcessor.js';
+import { SqsFifoProcessingUtils } from './SqsFifoProcessingUtils.js';
 import { EventType } from './constants.js';
 import {
   type BatchProcessingError,
@@ -46,17 +47,13 @@ import type {
  */
 class SqsFifoPartialProcessorAsync extends BatchProcessor {
   /**
-   * The ID of the current message group being processed.
+   * Utility class for processing SQS FIFO queues
    */
-  #currentGroupId?: string;
-  /**
-   * A set of group IDs that have already encountered failures.
-   */
-  readonly #failedGroupIds: Set<string>;
+  readonly #utils: SqsFifoProcessingUtils;
 
   public constructor() {
     super(EventType.SQS);
-    this.#failedGroupIds = new Set<string>();
+    this.#utils = new SqsFifoProcessingUtils();
   }
 
   /**
@@ -70,9 +67,7 @@ class SqsFifoPartialProcessorAsync extends BatchProcessor {
     record: EventSourceDataClassTypes,
     exception: Error
   ): FailureResponse {
-    if (this.options?.skipGroupOnError && this.#currentGroupId) {
-      this.#addToFailedGroup(this.#currentGroupId);
-    }
+    this.#utils.processFailureForCurrentGroup(this.options);
 
     return super.failureHandler(record, exception);
   }
@@ -101,13 +96,15 @@ class SqsFifoPartialProcessorAsync extends BatchProcessor {
     const processedRecords: (SuccessResponse | FailureResponse)[] = [];
     let currentIndex = 0;
     for (const record of this.records) {
-      this.#setCurrentGroup((record as SQSRecord).attributes?.MessageGroupId);
+      this.#utils.setCurrentGroup(
+        (record as SQSRecord).attributes?.MessageGroupId
+      );
 
-      if (this.#shouldShortCircuit()) {
+      if (this.#utils.shouldShortCircuit(this.failureMessages, this.options)) {
         return this.shortCircuitProcessing(currentIndex, processedRecords);
       }
 
-      const result = this.#shouldSkipCurrentGroup()
+      const result = this.#utils.shouldSkipCurrentGroup(this.options)
         ? this.#processFailRecord(
             record,
             new SqsFifoMessageGroupShortCircuitError()
@@ -150,15 +147,6 @@ class SqsFifoPartialProcessorAsync extends BatchProcessor {
     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.
    *
@@ -173,39 +161,6 @@ class SqsFifoPartialProcessorAsync extends BatchProcessor {
 
     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;
-  }
-
-  /**
-   * Determines whether the current group should be short-circuited.
-   *
-   * If we have any failed messages, we should then short circuit the process and
-   * fail remaining messages unless `skipGroupOnError` is true
-   */
-  #shouldShortCircuit(): boolean {
-    return !this.options?.skipGroupOnError && this.failureMessages.length !== 0;
-  }
-
-  /**
-   * Determines whether the current group should be skipped.
-   *
-   * If `skipGroupOnError` is true and the current group has previously failed,
-   * then we should skip processing the current group.
-   */
-  #shouldSkipCurrentGroup(): boolean {
-    return (
-      (this.options?.skipGroupOnError ?? false) &&
-      this.#currentGroupId &&
-      this.#failedGroupIds.has(this.#currentGroupId)
-    );
-  }
 }
 
 export { SqsFifoPartialProcessorAsync };

packages/batch/src/SqsFifoProcessingUtils.ts

@@ -0,0 +1,86 @@
+import type {
+  BatchProcessingOptions,
+  EventSourceDataClassTypes,
+} from './types.js';
+
+/**
+ * Utility class to handle processing of SQS FIFO messages.
+ */
+class SqsFifoProcessingUtils {
+  /**
+   * The ID of the current message group being processed.
+   */
+  #currentGroupId?: string;
+
+  /**
+   * A set of group IDs that have already encountered failures.
+   */
+  readonly #failedGroupIds: Set<string>;
+
+  public constructor() {
+    this.#failedGroupIds = new Set<string>();
+  }
+
+  /**
+   * 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.
+   */
+  public addToFailedGroup(group: string): void {
+    this.#failedGroupIds.add(group);
+  }
+
+  /**
+   * Sets the current group ID for the message being processed.
+   *
+   * @param group - The group ID of the current message being processed.
+   */
+  public setCurrentGroup(group?: string): void {
+    this.#currentGroupId = group;
+  }
+
+  /**
+   * Determines whether the current group should be short-circuited.
+   *
+   * If we have any failed messages, we should then short circuit the process and
+   * fail remaining messages unless `skipGroupOnError` is true
+   *
+   * @param failureMessages - The list of failure messages.
+   * @param options - The options for the batch processing.
+   */
+  public shouldShortCircuit(
+    failureMessages: EventSourceDataClassTypes[],
+    options?: BatchProcessingOptions
+  ): boolean {
+    return !options?.skipGroupOnError && failureMessages.length !== 0;
+  }
+
+  /**
+   * Determines whether the current group should be skipped.
+   *
+   * If `skipGroupOnError` is true and the current group has previously failed,
+   * then we should skip processing the current group.
+   *
+   * @param options - The options for the batch processing.
+   */
+  public shouldSkipCurrentGroup(options?: BatchProcessingOptions): boolean {
+    return (
+      (options?.skipGroupOnError ?? false) &&
+      this.#currentGroupId &&
+      this.#failedGroupIds.has(this.#currentGroupId)
+    );
+  }
+
+  /**
+   * Handles failure for current group
+   *
+   * @param options - The options for the batch processing.
+   */
+  public processFailureForCurrentGroup(options?: BatchProcessingOptions) {
+    if (options?.skipGroupOnError && this.#currentGroupId) {
+      this.addToFailedGroup(this.#currentGroupId);
+    }
+  }
+}
+
+export { SqsFifoProcessingUtils };