@SqsListener now works with any SQS queue, not only with messages d…

.changeset/starter-aws-sqs-listener-raw-messages.md

@@ -0,0 +1,10 @@
+---
+"@nodeboot/starter-aws": minor
+---
+
+`@SqsListener` now works with any SQS queue, not only with messages delivered by SNS.
+
+-   Messages sent straight to the queue are passed to the handler as they are: `message` is the parsed body, `messageId` the SQS message id and `timestamp` its `SentTimestamp`. Before, every message that was not an SNS notification failed and stayed on the queue. SNS notifications are still unwrapped as before. A body that is not JSON is passed as a string instead of failing. The envelope also carries the received SQS message as `raw`.
+-   FIFO queues (`.fifo`) and local emulators (LocalStack, ElasticMQ) are accepted: a queue URL is any `http(s)` URL with the path `/<account id>/<queue name>`. A new `integrations.aws.sqs.endpoint` setting points the SQS client at an emulator.
+-   `@SqsListener(queueUrl, options)` takes consumer options: `batchSize`, `waitTimeSeconds`, `visibilityTimeout`, `pollingWaitTimeMs`, `handleMessageTimeout`, `terminateVisibilityTimeout`, `attributeNames`, `messageSystemAttributeNames`, `messageAttributeNames`.
+-   Listeners stop polling on application shutdown. `stopSqsListeners()`, `SqsListenerAdapter` and `SqsListenerOptions` are exported from the package root.

starters/aws/README.md

@@ -305,30 +305,72 @@ com:
 
 #### The `MessageEnvelop` Payload
 
-Every `@SqsListener` handler receives a `MessageEnvelop<M>` — a parsed wrapper around the raw SQS message body,
-rather than the raw `sqs-consumer` `Message` object:
+Every `@SqsListener` handler receives a `MessageEnvelop<M>` rather than the raw `sqs-consumer` `Message`:
 
 ```typescript
 type MessageEnvelop<M = JsonObject> = {
-    messageId: string; // AWS-assigned message ID
-    timestamp: string; // SNS/SQS delivery timestamp
-    signature: string; // SNS signature, when the message originated from an SNS->SQS subscription
-    message: M; // The actual, JSON-parsed message payload
+    messageId: string; // the SNS notification's id, or the SQS message id
+    timestamp: string; // the SNS notification's Timestamp, or the SQS SentTimestamp (ISO 8601)
+    signature: string; // the SNS signature; empty for a message sent straight to SQS
+    message: M; // the JSON-parsed payload (the string itself when it is not JSON)
+    raw: Message; // the SQS message as received
 };
 ```
 
+Both kinds of queue messages are supported:
+
+-   **Sent straight to the queue** (`SendMessage`): `message` is the parsed body.
+-   **Delivered by an SNS topic subscription** (raw message delivery off): the notification is unwrapped and
+    `message` is its parsed `Message`.
+
+A body that is not JSON is passed as a string; parsing never fails the message.
+
 ```typescript
 @SqsListener("${com.example.aws.sqs.queue-url}")
 async onMessage(envelop: MessageEnvelop<{orderId: string}>): Promise<void> {
     console.log(`Processing order ${envelop.message.orderId} (messageId=${envelop.messageId})`);
 }
 ```
 
+#### Listener options
+
+The second argument configures the consumer (see sqs-consumer's
+[ConsumerOptions](https://bbc.github.io/sqs-consumer/interfaces/ConsumerOptions.html)):
+
+```typescript
+@SqsListener("${com.example.aws.sqs.queue-url}", {
+    batchSize: 10, // messages per receive, handled concurrently (default 1)
+    waitTimeSeconds: 20, // long polling (default 20)
+    visibilityTimeout: 60, // seconds (default: the queue's)
+    handleMessageTimeout: 30_000, // ms; a slower handler fails its message
+})
+async onMessage(envelop: MessageEnvelop): Promise<void> {}
+```
+
+#### Queue URLs, FIFO queues and local emulators
+
+The queue URL must have the path `/<12-digit account id>/<queue name>` on any `http(s)` host: AWS URLs
+(`https://sqs.<region>.amazonaws.com/...`), FIFO queues (`.../orders.fifo`) and local emulators. For LocalStack or
+ElasticMQ, point the client at the emulator:
+
+```yaml
+integrations:
+    aws:
+        sqs:
+            region: "us-east-1"
+            endpoint: "http://localhost:4566"
+```
+
+#### Shutdown
+
+Every listener stops polling on application shutdown (SIGINT/SIGTERM), before other shutdown hooks run; a message
+being handled finishes. `stopSqsListeners()` stops them on demand.
+
 #### SQS Message Handling
 
 -   The queue is polled continuously for messages using [long polling](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-long-polling.html).
 -   Throwing an error (or returning a rejected promise) from the handler function will cause the message to be left on the queue. An [SQS redrive policy](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/SQSDeadLetterQueue.html) can be used to move messages that cannot be processed to a dead letter queue.
--   By default messages are processed one at a time – a new message won't be received until the first one has been processed. To process messages in parallel, use the batchSize option [detailed here](https://bbc.github.io/sqs-consumer/interfaces/ConsumerOptions.html#batchSize).
+-   By default messages are processed one at a time – a new message won't be received until the first one has been processed. To process messages in parallel, pass `{batchSize}` in the listener options (above).
     -   It's also important to await any processing that you are doing to ensure that messages are processed one at a time.
 -   By default, messages that are sent to the handleMessage and handleMessageBatch functions will be considered as processed if they return without an error.
     -   To acknowledge individual messages, please return the message that you want to acknowledge if you are using handleMessage or the messages for handleMessageBatch.

starters/aws/src/adapter/SqsListenerAdapter.ts

@@ -6,31 +6,41 @@ import {
     extractPlaceholderKey,
     getActiveProfiles,
     Lifecycle,
+    ShutdownHookContext,
 } from "@nodeboot/context";
 import {AWS_SQS_FEATURE} from "../types";
-import {SQSClient} from "@aws-sdk/client-sqs";
+import {Message, SQSClient} from "@aws-sdk/client-sqs";
 import {Consumer} from "sqs-consumer";
-import {MessageEnvelop} from "./types";
+import {MessageEnvelop, SqsListenerOptions} from "./types";
 
-type SqsListenerOptions = {
+type SqsListenerAdapterOptions = {
     target: any;
     listenerFunction: Function;
     queueUrlOrConfigPlaceholder: string;
+    consumerOptions?: SqsListenerOptions;
 };
 
-const SQS_URL_PATTERN =
-    /^https:\/\/sqs\.(?<region>[a-z0-9-]+)\.amazonaws\.com\/(?<accountId>\d{12})\/(?<queueName>[a-zA-Z0-9_-]+)$/;
+/**
+ * A queue URL's path: `/<12-digit account id>/<queue name>[.fifo]`, on any http(s) host. Covers AWS
+ * (`https://sqs.<region>.amazonaws.com/...`, `.amazonaws.com.cn`, `.api.aws`) and local emulators such as LocalStack
+ * or ElasticMQ (`http://localhost:4566/000000000000/my-queue`; set `integrations.aws.sqs.endpoint` for them).
+ */
+const SQS_QUEUE_PATH = /^\/(?<accountId>\d{12})\/(?<queueName>[a-zA-Z0-9_-]{1,80}(\.fifo)?)$/;
+
+/** Stops polling on application shutdown, before other resources close. Registered once. */
+let shutdownHookRegistered = false;
 
 @Lifecycle("persistence.started")
 export class SqsListenerAdapter implements ApplicationFeatureAdapter {
     public static consumers: Consumer[] = [];
     public consumer?: Consumer;
-    private readonly options: SqsListenerOptions;
+    private readonly options: SqsListenerAdapterOptions;
 
-    constructor(options: SqsListenerOptions) {
+    constructor(options: SqsListenerAdapterOptions) {
         this.options = options;
     }
 
+    /** Stops every consumer: no new receives; messages being handled finish (and are deleted when they succeed). */
     public static stopAll(): void {
         for (const consumer of SqsListenerAdapter.consumers) {
             try {
@@ -42,8 +52,36 @@ export class SqsListenerAdapter implements ApplicationFeatureAdapter {
         SqsListenerAdapter.consumers = [];
     }
 
-    private isValidSqsUrl(url: string): boolean {
-        return SQS_URL_PATTERN.test(url);
+    static isValidSqsUrl(url: string): boolean {
+        try {
+            const {protocol, pathname} = new URL(url);
+            return (protocol === "https:" || protocol === "http:") && SQS_QUEUE_PATH.test(pathname);
+        } catch {
+            return false;
+        }
+    }
+
+    /** The envelope of one SQS message: an SNS notification unwrapped, or the body itself. */
+    static toEnvelop(msg: Message): MessageEnvelop<any> {
+        const body = msg.Body ?? "";
+        const parsed = parseJson(body);
+        if (isSnsNotification(parsed)) {
+            return {
+                messageId: parsed.MessageId ?? msg.MessageId ?? "",
+                signature: parsed.Signature ?? "",
+                timestamp: parsed.Timestamp ?? "",
+                message: parseJson(parsed.Message),
+                raw: msg,
+            };
+        }
+        const sent = msg.Attributes?.SentTimestamp;
+        return {
+            messageId: msg.MessageId ?? "",
+            signature: "",
+            timestamp: sent ? new Date(Number(sent)).toISOString() : "",
+            message: parsed,
+            raw: msg,
+        };
     }
 
     private getQueueUrl(config: any, queueUrlOrConfigPlaceholder: string) {
@@ -56,74 +94,104 @@ export class SqsListenerAdapter implements ApplicationFeatureAdapter {
     }
 
     bind({logger, iocContainer, config}: ApplicationFeatureContext): void {
-        const {target, queueUrlOrConfigPlaceholder, listenerFunction} = this.options;
+        const {target, queueUrlOrConfigPlaceholder, listenerFunction, consumerOptions = {}} = this.options;
+        const name = `${target.constructor.name}:::${listenerFunction.name}()`;
 
         // Check if SQS feature is enabled
-        if (ApplicationContext.get().applicationFeatures[AWS_SQS_FEATURE]) {
-            // Check if the current active profiles allow this SQS listener
-            // This is useful for conditional listener registration based on environment profiles
-            // e.g., only register certain listeners in production or development environments.
-            if (allowedProfiles(target)) {
-                // Retrieve the class instance (bean) from the DI container
-                const componentBean = iocContainer.get(target.constructor);
-
-                const queueUrl = this.getQueueUrl(config, queueUrlOrConfigPlaceholder);
-
-                // Validate SQS Queue url
-                if (this.isValidSqsUrl(queueUrl)) {
-                    logger.info(
-                        `Registering SQS Listener "${queueUrl}" --> "${target.constructor.name}:::${listenerFunction.name}()"`,
-                    );
-
-                    // "@ts-expect-error" SQSClient have a strange way to pass arguments to constructor that can make ts fail
-                    const sqsClient = iocContainer.get(SQSClient);
-
-                    const app = Consumer.create({
-                        queueUrl: queueUrl,
-                        pollingWaitTimeMs: 1000,
-                        handleMessage: async msg => {
-                            const body = JSON.parse(msg.Body!);
-
-                            const message: MessageEnvelop = {
-                                messageId: body.MessageId,
-                                signature: body.Signature,
-                                timestamp: body.Timestamp,
-                                message: JSON.parse(body.Message),
-                            };
-                            await listenerFunction.bind(componentBean)(message);
-                        },
-                        sqs: sqsClient,
-                    });
-
-                    app.on("error", err => {
-                        logger.error(`SQS Generic Error: ${err.message}`);
-                    });
-
-                    app.on("processing_error", err => {
-                        logger.error(`SQS Processing Error: ${err.message}`);
-                    });
-
-                    app.on("timeout_error", err => {
-                        logger.error(`SQS Timeout Error: ${err.message}`);
-                    });
-
-                    this.consumer = app;
-                    SqsListenerAdapter.consumers.push(app);
-                    app.start();
-                } else {
-                    logger.warn(`Invalid SQS queue URL for @SqsListener at function  "${target.constructor.name}:::${listenerFunction.name}()".
-                 Please provide a valid URL in the format "https://sqs.aws-region.amazonaws.com/account-id/queue-name"`);
-                }
-            } else {
-                logger.warn(`SqsListener ${target.constructor.name}:::${
-                    listenerFunction.name
-                }() with queueUrl ${queueUrlOrConfigPlaceholder} 
-                registered but the active profiles ${getActiveProfiles()} do not match the required profiles defined in the component/service class.`);
-            }
-        } else {
-            logger.warn(`⏰ SqsListener ${target.constructor.name}:::${listenerFunction.name}() with queueUrl ${queueUrlOrConfigPlaceholder} 
-            registered but AWS SQS is disabled. To enable AWS SQS, please decorate your Node-Boot application 
+        if (!ApplicationContext.get().applicationFeatures[AWS_SQS_FEATURE]) {
+            logger.warn(`⏰ SqsListener ${name} with queueUrl ${queueUrlOrConfigPlaceholder}
+            registered but AWS SQS is disabled. To enable AWS SQS, please decorate your Node-Boot application
             class with @EnableAws() and provide SQS region the configuration at "integrations.aws.sqs.region" config path.`);
+            return;
+        }
+        // Check if the current active profiles allow this SQS listener
+        // e.g., only register certain listeners in production or development environments.
+        if (!allowedProfiles(target)) {
+            logger.warn(`SqsListener ${name} with queueUrl ${queueUrlOrConfigPlaceholder}
+                registered but the active profiles ${getActiveProfiles()} do not match the required profiles defined in the component/service class.`);
+            return;
+        }
+
+        const queueUrl = this.getQueueUrl(config, queueUrlOrConfigPlaceholder);
+        if (!SqsListenerAdapter.isValidSqsUrl(queueUrl)) {
+            logger.warn(`Invalid SQS queue URL "${queueUrl}" for @SqsListener at function "${name}".
+                 Please provide a valid URL in the format "https://sqs.aws-region.amazonaws.com/account-id/queue-name"`);
+            return;
         }
+
+        logger.info(`Registering SQS Listener "${queueUrl}" --> "${name}"`);
+        // Retrieve the class instance (bean) from the DI container
+        const componentBean = iocContainer.get(target.constructor);
+        const sqsClient = iocContainer.get(SQSClient);
+
+        const app = Consumer.create({
+            pollingWaitTimeMs: 1000,
+            ...consumerOptions,
+            // The envelope's timestamp of a message that is not an SNS notification.
+            messageSystemAttributeNames: [
+                ...new Set([...(consumerOptions.messageSystemAttributeNames ?? []), "SentTimestamp" as const]),
+            ],
+            queueUrl,
+            sqs: sqsClient,
+            handleMessage: async msg => {
+                await listenerFunction.call(componentBean, SqsListenerAdapter.toEnvelop(msg));
+                // Acknowledged (deleted) only once the handler resolved.
+                return msg;
+            },
+        });
+
+        app.on("error", (err, msg) => {
+            logger.error(`SQS Generic Error (${name}${messageIdOf(msg)}): ${err.message}`);
+        });
+        app.on("processing_error", (err, msg) => {
+            logger.error(`SQS Processing Error (${name}${messageIdOf(msg)}): ${err.message}; left on the queue`);
+        });
+        app.on("timeout_error", (err, msg) => {
+            logger.error(`SQS Timeout Error (${name}${messageIdOf(msg)}): ${err.message}; left on the queue`);
+        });
+
+        this.consumer = app;
+        SqsListenerAdapter.consumers.push(app);
+        registerShutdownHook();
+        app.start();
     }
 }
+
+/** Stops every @SqsListener consumer. Also done automatically on application shutdown. */
+export function stopSqsListeners(): void {
+    SqsListenerAdapter.stopAll();
+}
+
+function registerShutdownHook(): void {
+    if (shutdownHookRegistered) return;
+    shutdownHookRegistered = true;
+    // No constructor: the shutdown context cannot resolve it from the IoC container and calls this object directly.
+    const target = Object.assign(Object.create(null), {stop: stopSqsListeners});
+    ShutdownHookContext.get().addShutdownHook({target, methodName: "stop", priority: 1000, timeout: 10_000});
+}
+
+function parseJson(text: string): any {
+    try {
+        return JSON.parse(text);
+    } catch {
+        return text;
+    }
+}
+
+/** An SNS → SQS delivery (raw message delivery off). `Type` is absent from some hand-built envelopes. */
+function isSnsNotification(
+    v: any,
+): v is {Type?: string; MessageId?: string; Message: string; Signature?: string; Timestamp?: string} {
+    return (
+        v !== null &&
+        typeof v === "object" &&
+        typeof v.Message === "string" &&
+        (v.Type === "Notification" || typeof v.MessageId === "string")
+    );
+}
+
+function messageIdOf(msg: Message | Message[] | undefined): string {
+    if (!msg) return "";
+    const ids = (Array.isArray(msg) ? msg : [msg]).map(m => m.MessageId).filter(Boolean);
+    return ids.length ? `, message ${ids.join(", ")}` : "";
+}

starters/aws/src/adapter/index.ts

@@ -1,2 +1,2 @@
-export {SqsListenerAdapter} from "./SqsListenerAdapter";
+export {SqsListenerAdapter, stopSqsListeners} from "./SqsListenerAdapter";
 export * from "./types";

starters/aws/src/adapter/types.ts

@@ -1,8 +1,40 @@
 import {JsonObject} from "@nodeboot/context";
+import type {Message, MessageSystemAttributeName, QueueAttributeName} from "@aws-sdk/client-sqs";
 
+/**
+ * What an `@SqsListener` handler receives.
+ *
+ * - A message delivered by an SNS topic subscription (`{Type: "Notification", MessageId, Message, ...}`) is unwrapped:
+ *   `messageId`, `timestamp` and `signature` are the notification's, `message` its `Message`.
+ * - Any other body is the message itself: `messageId` is the SQS message id, `timestamp` its `SentTimestamp`
+ *   attribute (when requested), `signature` empty.
+ *
+ * `message` is the JSON-parsed payload, or the string itself when it is not JSON. `raw` is the SQS message.
+ */
 export type MessageEnvelop<M = JsonObject> = {
     messageId: string;
     timestamp: string;
     signature: string;
     message: M;
+    raw: Message;
+};
+
+/** Consumer options of one `@SqsListener` (see sqs-consumer's ConsumerOptions). */
+export type SqsListenerOptions = {
+    /** Messages received per poll and processed concurrently (1-10). Default 1. */
+    batchSize?: number;
+    /** Long-polling wait per receive (0-20 s). Default 20. */
+    waitTimeSeconds?: number;
+    /** Visibility timeout of received messages, in seconds. Default: the queue's. */
+    visibilityTimeout?: number;
+    /** Pause between polls, in ms. Default 0 when messages keep arriving. */
+    pollingWaitTimeMs?: number;
+    /** A handler running longer than this fails its message (left on the queue), in ms. Default: none. */
+    handleMessageTimeout?: number;
+    /** On a failure, make the message visible again after this many seconds (true: immediately). */
+    terminateVisibilityTimeout?: boolean | number;
+    attributeNames?: QueueAttributeName[];
+    /** SentTimestamp is always requested. */
+    messageSystemAttributeNames?: MessageSystemAttributeName[];
+    messageAttributeNames?: string[];
 };