From 5db08ba6d3f2c6461d310d2ba095740c6b028ded Mon Sep 17 00:00:00 2001 From: manusant Date: Sun, 25 Oct 2026 12:06:30 +0000 Subject: [PATCH] `@SqsListener` now works with any SQS queue, not only with messages delivered by SNS. --- .../starter-aws-sqs-listener-raw-messages.md | 10 + starters/aws/README.md | 56 ++++- .../aws/src/adapter/SqsListenerAdapter.ts | 216 ++++++++++++------ starters/aws/src/adapter/index.ts | 2 +- starters/aws/src/adapter/types.ts | 32 +++ .../aws/src/config/SqsClientConfiguration.ts | 10 +- starters/aws/src/decorator/SqsListener.ts | 8 +- starters/aws/src/index.ts | 2 +- .../test/aws-sqs-listener-enabled.it.test.ts | 14 ++ .../aws/test/sqs-listener-envelop.test.ts | 73 ++++++ 10 files changed, 338 insertions(+), 85 deletions(-) create mode 100644 .changeset/starter-aws-sqs-listener-raw-messages.md create mode 100644 starters/aws/test/sqs-listener-envelop.test.ts diff --git a/.changeset/starter-aws-sqs-listener-raw-messages.md b/.changeset/starter-aws-sqs-listener-raw-messages.md new file mode 100644 index 00000000..ebbbbdcc --- /dev/null +++ b/.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 `//`. 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. diff --git a/starters/aws/README.md b/starters/aws/README.md index 49339720..8572df1a 100644 --- a/starters/aws/README.md +++ b/starters/aws/README.md @@ -305,18 +305,26 @@ com: #### The `MessageEnvelop` Payload -Every `@SqsListener` handler receives a `MessageEnvelop` — a parsed wrapper around the raw SQS message body, -rather than the raw `sqs-consumer` `Message` object: +Every `@SqsListener` handler receives a `MessageEnvelop` rather than the raw `sqs-consumer` `Message`: ```typescript type MessageEnvelop = { - 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 { @@ -324,11 +332,45 @@ async onMessage(envelop: MessageEnvelop<{orderId: string}>): Promise { } ``` +#### 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 {} +``` + +#### Queue URLs, FIFO queues and local emulators + +The queue URL must have the path `/<12-digit account id>/` on any `http(s)` host: AWS URLs +(`https://sqs..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. diff --git a/starters/aws/src/adapter/SqsListenerAdapter.ts b/starters/aws/src/adapter/SqsListenerAdapter.ts index fea6fa2e..88de411e 100644 --- a/starters/aws/src/adapter/SqsListenerAdapter.ts +++ b/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\.(?[a-z0-9-]+)\.amazonaws\.com\/(?\d{12})\/(?[a-zA-Z0-9_-]+)$/; +/** + * A queue URL's path: `/<12-digit account id>/[.fifo]`, on any http(s) host. Covers AWS + * (`https://sqs..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 = /^\/(?\d{12})\/(?[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 { + 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(", ")}` : ""; +} diff --git a/starters/aws/src/adapter/index.ts b/starters/aws/src/adapter/index.ts index 8d2c175a..11878a4f 100644 --- a/starters/aws/src/adapter/index.ts +++ b/starters/aws/src/adapter/index.ts @@ -1,2 +1,2 @@ -export {SqsListenerAdapter} from "./SqsListenerAdapter"; +export {SqsListenerAdapter, stopSqsListeners} from "./SqsListenerAdapter"; export * from "./types"; diff --git a/starters/aws/src/adapter/types.ts b/starters/aws/src/adapter/types.ts index 0b878803..f803c06e 100644 --- a/starters/aws/src/adapter/types.ts +++ b/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 = { 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[]; }; diff --git a/starters/aws/src/config/SqsClientConfiguration.ts b/starters/aws/src/config/SqsClientConfiguration.ts index 8a5241d9..c55e3430 100644 --- a/starters/aws/src/config/SqsClientConfiguration.ts +++ b/starters/aws/src/config/SqsClientConfiguration.ts @@ -25,7 +25,15 @@ export class SqsClientConfiguration { ); } - const client = new SQSClient({region, credentials, logger, useQueueUrlAsEndpoint: false}); + // A local emulator (LocalStack, ElasticMQ); unset for AWS. + const endpoint = config.getOptionalString("integrations.aws.sqs.endpoint"); + const client = new SQSClient({ + region, + credentials, + logger, + useQueueUrlAsEndpoint: false, + ...(endpoint ? {endpoint} : {}), + }); iocContainer.set(SQSClient, client); // Enable AWS SQS feature in the application context diff --git a/starters/aws/src/decorator/SqsListener.ts b/starters/aws/src/decorator/SqsListener.ts index 80eb5510..ad4ed97e 100644 --- a/starters/aws/src/decorator/SqsListener.ts +++ b/starters/aws/src/decorator/SqsListener.ts @@ -1,12 +1,18 @@ import {ApplicationContext} from "@nodeboot/context"; import {SqsListenerAdapter} from "../adapter"; +import type {SqsListenerOptions} from "../adapter"; -export function SqsListener(queueUrlOrConfigPlaceholder: string): MethodDecorator { +/** + * Consumes an SQS queue with the decorated method. `queueUrlOrConfigPlaceholder` is a queue URL or a `${config.path}` + * placeholder resolving to one. The method receives a {@link MessageEnvelop}; throwing leaves the message on the queue. + */ +export function SqsListener(queueUrlOrConfigPlaceholder: string, options: SqsListenerOptions = {}): MethodDecorator { return function (target: any, _: string | symbol, descriptor: PropertyDescriptor) { const sqsListenerAdapter = new SqsListenerAdapter({ target, queueUrlOrConfigPlaceholder, listenerFunction: descriptor.value, + consumerOptions: options, }); ApplicationContext.get().applicationFeatureAdapters.push(sqsListenerAdapter); }; diff --git a/starters/aws/src/index.ts b/starters/aws/src/index.ts index 3d73e2fd..911c6992 100644 --- a/starters/aws/src/index.ts +++ b/starters/aws/src/index.ts @@ -1,2 +1,2 @@ export * from "./decorator"; -export {MessageEnvelop} from "./adapter"; +export {MessageEnvelop, SqsListenerOptions, SqsListenerAdapter, stopSqsListeners} from "./adapter"; diff --git a/starters/aws/test/aws-sqs-listener-enabled.it.test.ts b/starters/aws/test/aws-sqs-listener-enabled.it.test.ts index 86577204..f68978a3 100644 --- a/starters/aws/test/aws-sqs-listener-enabled.it.test.ts +++ b/starters/aws/test/aws-sqs-listener-enabled.it.test.ts @@ -71,4 +71,18 @@ describe("@nodeboot/starter-aws - SQS listener message consumption", () => { assert.equal(received?.messageId, "msg-789"); assert.deepEqual(received?.message, payload); }); + + test("hands a raw SQS message (not sent through SNS) to the handler as it is", async () => { + const service = ApplicationContext.getIocContainer()!.get(SampleSqsListenerService); + const before = service.receivedMessages.length; + mockServer.pushSqsMessage(JSON.stringify({orderId: "order-raw"})); + + const start = Date.now(); + while (service.receivedMessages.length === before && Date.now() - start < 3000) { + await new Promise(r => setTimeout(r, 100)); + } + const received = service.receivedMessages[before]; + assert.deepEqual(received?.message, {orderId: "order-raw"}); + assert.match(received?.messageId ?? "", /^msg-/); + }); }); diff --git a/starters/aws/test/sqs-listener-envelop.test.ts b/starters/aws/test/sqs-listener-envelop.test.ts new file mode 100644 index 00000000..98707d13 --- /dev/null +++ b/starters/aws/test/sqs-listener-envelop.test.ts @@ -0,0 +1,73 @@ +/** + * Unit tests of how @SqsListener reads queue URLs and SQS messages. + */ +import {describe, test} from "node:test"; +import assert from "node:assert/strict"; +import {SqsListenerAdapter} from "../src/adapter"; + +describe("@nodeboot/starter-aws - SqsListenerAdapter", () => { + test("accepts AWS, FIFO and local emulator queue URLs", () => { + for (const url of [ + "https://sqs.eu-west-1.amazonaws.com/123456789012/orders", + "https://sqs.eu-west-1.amazonaws.com/123456789012/orders.fifo", + "https://sqs.cn-north-1.amazonaws.com.cn/123456789012/orders", + "http://localhost:4566/000000000000/orders", + "http://sqs.us-east-1.localhost.localstack.cloud:4566/000000000000/orders.fifo", + ]) + assert.equal(SqsListenerAdapter.isValidSqsUrl(url), true, url); + }); + + test("refuses URLs that are not a queue", () => { + for (const url of [ + "http://invalid-queue-url/123", + "https://sqs.eu-west-1.amazonaws.com/orders", + "https://sqs.eu-west-1.amazonaws.com/123456789012/orders/extra", + "ftp://host/123456789012/orders", + "${not.resolved}", + "", + ]) + assert.equal(SqsListenerAdapter.isValidSqsUrl(url), false, url); + }); + + test("unwraps an SNS notification", () => { + const envelop = SqsListenerAdapter.toEnvelop({ + MessageId: "sqs-1", + Body: JSON.stringify({ + Type: "Notification", + MessageId: "sns-1", + Timestamp: "2026-10-25T10:00:00.000Z", + Signature: "sig", + Message: JSON.stringify({orderId: "o-1"}), + }), + }); + assert.deepEqual( + {id: envelop.messageId, ts: envelop.timestamp, sig: envelop.signature, message: envelop.message}, + {id: "sns-1", ts: "2026-10-25T10:00:00.000Z", sig: "sig", message: {orderId: "o-1"}}, + ); + assert.equal(envelop.raw.MessageId, "sqs-1"); + }); + + test("hands a raw SQS message over as it is", () => { + const envelop = SqsListenerAdapter.toEnvelop({ + MessageId: "sqs-2", + Body: JSON.stringify({orderId: "o-2"}), + Attributes: {SentTimestamp: "1761386400000"}, + }); + assert.deepEqual( + {id: envelop.messageId, ts: envelop.timestamp, sig: envelop.signature, message: envelop.message}, + {id: "sqs-2", ts: new Date(1761386400000).toISOString(), sig: "", message: {orderId: "o-2"}}, + ); + }); + + test("passes a body that is not JSON as a string, never throwing", () => { + assert.equal(SqsListenerAdapter.toEnvelop({MessageId: "sqs-3", Body: "plain text"}).message, "plain text"); + assert.equal( + SqsListenerAdapter.toEnvelop({ + MessageId: "sqs-4", + Body: JSON.stringify({Type: "Notification", Message: "hi"}), + }).message, + "hi", + ); + assert.equal(SqsListenerAdapter.toEnvelop({MessageId: "sqs-5"}).message, ""); + }); +});