feat(sqs): dead-letter queue support + full consumer-option config for invalidation triggers

packages/sqs/README.md

@@ -29,6 +29,8 @@ The implementation is built on top of [`@message-queue-toolkit/sns`](https://git
   - [Mixing source kinds with `composeTriggers`](#mixing-source-kinds-with-composetriggers)
   - [Resolver semantics](#resolver-semantics)
   - [Error handling and retries](#error-handling-and-retries)
+  - [Dead-letter queues](#dead-letter-queues)
+  - [Explicit vs spread configuration](#explicit-vs-spread-configuration)
 - [Testing with fauxqs](#testing-with-fauxqs)
 - [API reference](#api-reference)
 
@@ -560,7 +562,64 @@ If the resolver or apply step throws, the trigger:
 1. Invokes the optional `errorHandler(err, channel)` for observability.
 2. Re-throws so `message-queue-toolkit` can apply its standard SQS retry / dead-letter behaviour.
 
-For schema-violation errors, the message is failed by `message-queue-toolkit` before the resolver runs and goes back to the queue (and ultimately to a DLQ if you configured one). Configure DLQ on the trigger's queue creation config to bound retries.
+For schema-violation errors, the message is failed by `message-queue-toolkit` before the resolver runs and goes back to the queue (and ultimately to a DLQ if you configured one).
+
+### Dead-letter queues
+
+To bound retries, add a `deadLetterQueue` to any trigger source. With a `creationConfig` the trigger **auto-creates the DLQ, attaches the redrive policy to its own queue, and (for SNS sources) wires the subscription** — no manual AWS setup required:
+
+```ts
+const trigger = new SnsTopicInvalidationTrigger({
+  target: userLoader,
+  dependencies: consumerDeps,
+  sources: [
+    {
+      creationConfig: {
+        topic: { Name: 'domain-events.users' },
+        queue: { QueueName: `cache-trigger-${process.env.HOSTNAME}` },
+      },
+      deadLetterQueue: {
+        // Move a message to the DLQ after this many failed receives.
+        redrivePolicy: { maxReceiveCount: 3 },
+        // Auto-create the DLQ. Use `locatorConfig` instead to point at an existing one.
+        creationConfig: { queue: { QueueName: `cache-trigger-${process.env.HOSTNAME}-dlq` } },
+      },
+      bindings: [/* ... */],
+    },
+  ],
+})
+```
+
+The `deadLetterQueue` field is the same shape `message-queue-toolkit` exposes on its consumers. It is available on all four trigger source types.
+
+### Explicit vs spread configuration
+
+A trigger source **is** the underlying `message-queue-toolkit` consumer options (minus the `handlers` list, which the trigger builds from `bindings`). That means both styles are fully type-checked — autocomplete and typo detection on every option:
+
+- **Explicit** — spell options out inline (`creationConfig`/`locatorConfig`, `deadLetterQueue`, `subscriptionConfig`, `concurrentConsumersAmount`, `consumerOverrides`, ...). An unknown key or a wrong-typed value is a compile error.
+- **Spread** — resolve options elsewhere — e.g. with `@lokalise/aws-config`'s `getSnsMqtOptionsResolver()` — and spread the result straight in.
+
+The trigger always overrides `handlers` with the ones it builds from `bindings`; everything else flows through untouched.
+
+```ts
+const resolver = getSnsMqtOptionsResolver({ appEnv: 'production' })
+const options = resolver.resolveConsumerOptions(topicName, queueName, {
+  /* awsConfig, logger, deadLetterQueue, ... */
+})
+
+const trigger = new SnsTopicInvalidationTrigger({
+  target,
+  dependencies,
+  sources: [
+    {
+      ...options,
+      bindings: [
+        { messageSchema: PROJECT_LANGUAGE_EVENT_SCHEMA, resolver: async (message) => { /* ... */ } },
+      ],
+    },
+  ],
+})
+```
 
 ### Lifecycle
 
@@ -624,7 +683,7 @@ Then point your AWS SDK clients at `http://localhost:4566`.
 | `SqsQueueInvalidationTrigger` | Flat-cache trigger consuming directly from upstream SQS queues. |
 | `SnsTopicGroupInvalidationTrigger` / `SqsQueueGroupInvalidationTrigger` | `GroupLoader` counterparts. |
 | `composeTriggers(...triggers)` | Wraps multiple `InvalidationTrigger`s into one combined `start()` / `stop()`. |
-| `SnsTopicInvalidationSource` / `SqsQueueInvalidationSource` | A single upstream source plus its bindings and optional `messageTypeField`. Group counterparts have parallel names. |
+| `SnsTopicInvalidationSource` / `SqsQueueInvalidationSource` | A single upstream source: every `message-queue-toolkit` consumer option (`creationConfig`/`locatorConfig`, `deadLetterQueue`, `subscriptionConfig`, ...) minus `handlers`, plus `bindings` and optional `messageTypeField`. Fully typed for both explicit and spread configuration. Group counterparts have parallel names. |
 | `FlatBinding<TMessage>` / `GroupBinding<TMessage>` | One `(messageSchema, resolver, messageType?)` triple. |
 | `InvalidationTarget` / `GroupInvalidationTarget` | Structural interfaces that `Loader` / `GroupLoader` satisfy. |
 | `InvalidationAction` / `GroupInvalidationAction` | Action ADTs returned by resolvers. |

packages/sqs/lib/triggers/SnsTopicGroupInvalidationTrigger.ts

@@ -1,10 +1,7 @@
 import type {
   SNSSQSConsumerDependencies,
   SNSSQSConsumerOptions,
-  SNSSQSCreationConfig,
-  SNSSQSQueueLocatorType,
 } from '@message-queue-toolkit/sns'
-import type { SqsSubscriptionOptions } from '../SqsNotificationConsumer.js'
 import {
   AbstractSqsTrigger,
   type InternalConsumerHandle,
@@ -15,14 +12,19 @@ import {
   buildGroupBindings,
   type GroupBinding,
 } from './bindingHelpers.js'
+import type { SnsTopicSourceConfig } from './SnsTopicInvalidationTrigger.js'
 import type { GroupInvalidationTarget, TriggerErrorHandler } from './types.js'
 
-export type SnsTopicGroupSourceConfig =
-  | { creationConfig: SNSSQSCreationConfig; locatorConfig?: never }
-  | { creationConfig?: never; locatorConfig: SNSSQSQueueLocatorType }
+/**
+ * The per-source config surface is identical to the flat-cache one — every
+ * `message-queue-toolkit` SNS→SQS consumer option minus `handlers` — so it is
+ * aliased to {@link SnsTopicSourceConfig} rather than re-declared. The handler
+ * execution context (the only place the two would differ) lives on the omitted
+ * `handlers` field, so the resulting types are structurally the same.
+ */
+export type SnsTopicGroupSourceConfig = SnsTopicSourceConfig
 
 export type SnsTopicGroupInvalidationSource = SnsTopicGroupSourceConfig & {
-  subscriptionConfig?: SqsSubscriptionOptions
   messageTypeField?: string
   // biome-ignore lint/suspicious/noExplicitAny: bindings heterogeneous over TMessage
   bindings: readonly GroupBinding<any>[]
@@ -61,22 +63,30 @@ export class SnsTopicGroupInvalidationTrigger extends AbstractSqsTrigger {
 
   private buildConsumer(source: SnsTopicGroupInvalidationSource): InternalConsumerHandle {
     const channel = this.channelOverride ?? deriveSnsTopicGroupChannelName(source)
+    const { bindings, messageTypeField, ...consumerOptions } = source
     const { handlers, context, messageTypeResolver } = buildGroupBindings(
-      source.bindings,
-      source.messageTypeField,
+      bindings,
+      messageTypeField,
       this.target,
       channel,
       this.errorHandler,
     )
 
-    const base = {
+    // Forward every consumer option the caller supplied. This deliberately
+    // spreads the whole source (minus the trigger-only `bindings`/
+    // `messageTypeField`) so a pre-resolved options object — e.g. the output of
+    // `@lokalise/aws-config`'s `resolveConsumerOptions(...)` — can be spread
+    // straight into the source and have all of its fields (`creationConfig`/
+    // `locatorConfig`, `deadLetterQueue`, `concurrentConsumersAmount`,
+    // `consumerOverrides`, ...) flow through untouched. The trigger owns
+    // `handlers`, so the bindings-derived handlers always override any in the
+    // spread.
+    const options = {
+      ...consumerOptions,
       handlers,
       messageTypeResolver,
       subscriptionConfig: source.subscriptionConfig ?? { updateAttributesIfExists: false },
     }
-    const options = source.creationConfig
-      ? { ...base, creationConfig: source.creationConfig }
-      : { ...base, locatorConfig: source.locatorConfig }
 
     return new SnsTopicTriggerConsumer(
       this.dependencies,

packages/sqs/lib/triggers/SnsTopicInvalidationTrigger.ts

@@ -1,10 +1,7 @@
 import type {
   SNSSQSConsumerDependencies,
   SNSSQSConsumerOptions,
-  SNSSQSCreationConfig,
-  SNSSQSQueueLocatorType,
 } from '@message-queue-toolkit/sns'
-import type { SqsSubscriptionOptions } from '../SqsNotificationConsumer.js'
 import {
   AbstractSqsTrigger,
   type InternalConsumerHandle,
@@ -17,12 +14,25 @@ import {
 } from './bindingHelpers.js'
 import type { InvalidationTarget, TriggerErrorHandler } from './types.js'
 
-export type SnsTopicSourceConfig =
-  | { creationConfig: SNSSQSCreationConfig; locatorConfig?: never }
-  | { creationConfig?: never; locatorConfig: SNSSQSQueueLocatorType }
+/**
+ * Every option the underlying `message-queue-toolkit` SNS→SQS consumer accepts,
+ * minus the `handlers` list (which the trigger builds from `bindings`).
+ *
+ * This is the per-source config surface, so callers get the same flexibility in
+ * two interchangeable styles, both fully type-checked:
+ *
+ * - **Explicit** — spell out `creationConfig`/`locatorConfig`, `subscriptionConfig`,
+ *   `deadLetterQueue`, `concurrentConsumersAmount`, etc. with autocomplete and
+ *   typo detection.
+ * - **Spread** — drop in a pre-resolved options object (e.g.
+ *   `@lokalise/aws-config`'s `resolveConsumerOptions(...)`) with `...options`.
+ */
+export type SnsTopicSourceConfig = Omit<
+  SNSSQSConsumerOptions<object, BindingHandlerContext<InvalidationTarget>, undefined>,
+  'handlers'
+>
 
 export type SnsTopicInvalidationSource = SnsTopicSourceConfig & {
-  subscriptionConfig?: SqsSubscriptionOptions
   messageTypeField?: string
   // biome-ignore lint/suspicious/noExplicitAny: bindings heterogeneous over TMessage
   bindings: readonly FlatBinding<any>[]
@@ -70,22 +80,30 @@ export class SnsTopicInvalidationTrigger extends AbstractSqsTrigger {
 
   private buildConsumer(source: SnsTopicInvalidationSource): InternalConsumerHandle {
     const channel = this.channelOverride ?? deriveSnsTopicChannelName(source)
+    const { bindings, messageTypeField, ...consumerOptions } = source
     const { handlers, context, messageTypeResolver } = buildFlatBindings(
-      source.bindings,
-      source.messageTypeField,
+      bindings,
+      messageTypeField,
       this.target,
       channel,
       this.errorHandler,
     )
 
-    const base = {
+    // Forward every consumer option the caller supplied. This deliberately
+    // spreads the whole source (minus the trigger-only `bindings`/
+    // `messageTypeField`) so a pre-resolved options object — e.g. the output of
+    // `@lokalise/aws-config`'s `resolveConsumerOptions(...)` — can be spread
+    // straight into the source and have all of its fields (`creationConfig`/
+    // `locatorConfig`, `deadLetterQueue`, `concurrentConsumersAmount`,
+    // `consumerOverrides`, ...) flow through untouched. The trigger owns
+    // `handlers`, so the bindings-derived handlers always override any in the
+    // spread.
+    const options = {
+      ...consumerOptions,
       handlers,
       messageTypeResolver,
       subscriptionConfig: source.subscriptionConfig ?? { updateAttributesIfExists: false },
     }
-    const options = source.creationConfig
-      ? { ...base, creationConfig: source.creationConfig }
-      : { ...base, locatorConfig: source.locatorConfig }
 
     return new SnsTopicTriggerConsumer(
       this.dependencies,

packages/sqs/lib/triggers/SqsQueueGroupInvalidationTrigger.ts

@@ -1,8 +1,6 @@
 import type {
   SQSConsumerDependencies,
   SQSConsumerOptions,
-  SQSCreationConfig,
-  SQSQueueLocatorType,
 } from '@message-queue-toolkit/sqs'
 import {
   AbstractSqsTrigger,
@@ -14,11 +12,17 @@ import {
   buildGroupBindings,
   type GroupBinding,
 } from './bindingHelpers.js'
+import type { SqsQueueSourceConfig } from './SqsQueueInvalidationTrigger.js'
 import type { GroupInvalidationTarget, TriggerErrorHandler } from './types.js'
 
-export type SqsQueueGroupSourceConfig =
-  | { creationConfig: SQSCreationConfig; locatorConfig?: never }
-  | { creationConfig?: never; locatorConfig: SQSQueueLocatorType }
+/**
+ * The per-source config surface is identical to the flat-cache one — every
+ * `message-queue-toolkit` SQS consumer option minus `handlers` — so it is
+ * aliased to {@link SqsQueueSourceConfig} rather than re-declared. The handler
+ * execution context (the only place the two would differ) lives on the omitted
+ * `handlers` field, so the resulting types are structurally the same.
+ */
+export type SqsQueueGroupSourceConfig = SqsQueueSourceConfig
 
 export type SqsQueueGroupInvalidationSource = SqsQueueGroupSourceConfig & {
   messageTypeField?: string
@@ -59,18 +63,25 @@ export class SqsQueueGroupInvalidationTrigger extends AbstractSqsTrigger {
 
   private buildConsumer(source: SqsQueueGroupInvalidationSource): InternalConsumerHandle {
     const channel = this.channelOverride ?? deriveSqsQueueGroupChannelName(source)
+    const { bindings, messageTypeField, ...consumerOptions } = source
     const { handlers, context, messageTypeResolver } = buildGroupBindings(
-      source.bindings,
-      source.messageTypeField,
+      bindings,
+      messageTypeField,
       this.target,
       channel,
       this.errorHandler,
     )
 
-    const base = { handlers, messageTypeResolver }
-    const options = source.creationConfig
-      ? { ...base, creationConfig: source.creationConfig }
-      : { ...base, locatorConfig: source.locatorConfig }
+    // Forward every consumer option the caller supplied. This deliberately
+    // spreads the whole source (minus the trigger-only `bindings`/
+    // `messageTypeField`) so a pre-resolved options object — e.g. the output of
+    // `@lokalise/aws-config`'s `resolveConsumerOptions(...)` — can be spread
+    // straight into the source and have all of its fields (`creationConfig`/
+    // `locatorConfig`, `deadLetterQueue`, `concurrentConsumersAmount`,
+    // `consumerOverrides`, ...) flow through untouched. The trigger owns
+    // `handlers`, so the bindings-derived handlers always override any in the
+    // spread.
+    const options = { ...consumerOptions, handlers, messageTypeResolver }
 
     return new SqsQueueTriggerConsumer(
       this.dependencies,

packages/sqs/lib/triggers/SqsQueueInvalidationTrigger.ts

@@ -1,8 +1,6 @@
 import type {
   SQSConsumerDependencies,
   SQSConsumerOptions,
-  SQSCreationConfig,
-  SQSQueueLocatorType,
 } from '@message-queue-toolkit/sqs'
 import {
   AbstractSqsTrigger,
@@ -16,9 +14,22 @@ import {
 } from './bindingHelpers.js'
 import type { InvalidationTarget, TriggerErrorHandler } from './types.js'
 
-export type SqsQueueSourceConfig =
-  | { creationConfig: SQSCreationConfig; locatorConfig?: never }
-  | { creationConfig?: never; locatorConfig: SQSQueueLocatorType }
+/**
+ * Every option the underlying `message-queue-toolkit` SQS consumer accepts,
+ * minus the `handlers` list (which the trigger builds from `bindings`).
+ *
+ * This is the per-source config surface, so callers get the same flexibility in
+ * two interchangeable styles, both fully type-checked:
+ *
+ * - **Explicit** — spell out `creationConfig`/`locatorConfig`, `deadLetterQueue`,
+ *   `concurrentConsumersAmount`, etc. with autocomplete and typo detection.
+ * - **Spread** — drop in a pre-resolved options object (e.g.
+ *   `@lokalise/aws-config`'s `resolveConsumerOptions(...)`) with `...options`.
+ */
+export type SqsQueueSourceConfig = Omit<
+  SQSConsumerOptions<object, BindingHandlerContext<InvalidationTarget>, undefined>,
+  'handlers'
+>
 
 export type SqsQueueInvalidationSource = SqsQueueSourceConfig & {
   /**
@@ -83,18 +94,25 @@ export class SqsQueueInvalidationTrigger extends AbstractSqsTrigger {
 
   private buildConsumer(source: SqsQueueInvalidationSource): InternalConsumerHandle {
     const channel = this.channelOverride ?? deriveSqsQueueChannelName(source)
+    const { bindings, messageTypeField, ...consumerOptions } = source
     const { handlers, context, messageTypeResolver } = buildFlatBindings(
-      source.bindings,
-      source.messageTypeField,
+      bindings,
+      messageTypeField,
       this.target,
       channel,
       this.errorHandler,
     )
 
-    const base = { handlers, messageTypeResolver }
-    const options = source.creationConfig
-      ? { ...base, creationConfig: source.creationConfig }
-      : { ...base, locatorConfig: source.locatorConfig }
+    // Forward every consumer option the caller supplied. This deliberately
+    // spreads the whole source (minus the trigger-only `bindings`/
+    // `messageTypeField`) so a pre-resolved options object — e.g. the output of
+    // `@lokalise/aws-config`'s `resolveConsumerOptions(...)` — can be spread
+    // straight into the source and have all of its fields (`creationConfig`/
+    // `locatorConfig`, `deadLetterQueue`, `concurrentConsumersAmount`,
+    // `consumerOverrides`, ...) flow through untouched. The trigger owns
+    // `handlers`, so the bindings-derived handlers always override any in the
+    // spread.
+    const options = { ...consumerOptions, handlers, messageTypeResolver }
 
     return new SqsQueueTriggerConsumer(
       this.dependencies,