feat(sqs): triggers own subscription config so spread needs no workarounds

packages/sqs/README.md

@@ -610,7 +610,12 @@ A trigger source **is** the underlying `message-queue-toolkit` consumer options
 - **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.
+A trigger is not a plain consumer — it owns its handlers and its subscription *filtering* — so just those two things are claimed by the trigger and everything else flows through untouched. You can spread a resolved options object straight in without un-setting anything:
+
+- **`handlers`** — always rebuilt from `bindings`.
+- **subscription filter policy** (`subscriptionConfig.Attributes.FilterPolicy` / `FilterPolicyScope`) — `resolveConsumerOptions` derives a `FilterPolicy` from the consumer's handlers. The trigger has its own handlers (from `bindings`), so the resolver's policy — built from an empty handler list — would reject every message. The trigger drops just those keys and defaults the subscription to accept-all.
+
+Everything else flows through as-is, including `creationConfig`/`locatorConfig`, `deadLetterQueue`, `concurrentConsumersAmount`, `consumerOverrides`, and the **subscription-level dead-letter queue** — both `subscriptionConfig.Attributes.RedrivePolicy` and a spread-in `subscriptionDeadLetterQueue` are preserved.
 
 ```ts
 const resolver = getSnsMqtOptionsResolver({ appEnv: 'production' })

packages/sqs/index.ts

@@ -86,6 +86,7 @@ export {
   type SqsQueueSourceConfig,
 } from './lib/triggers/SqsQueueInvalidationTrigger.js'
 export {
+  buildSnsTriggerConsumerOptions,
   deriveSnsTopicChannelName,
   SnsTopicInvalidationTrigger,
   type SnsTopicInvalidationSource,

packages/sqs/lib/triggers/AbstractSqsTrigger.ts

@@ -80,7 +80,13 @@ export abstract class AbstractSqsTrigger implements InvalidationTrigger {
         const consumers = this.createConsumers()
         if (consumers.length === 0) return
         try {
-          await Promise.all(consumers.map((c) => c.init()))
+          // Only call start(): the underlying message-queue-toolkit consumer's
+          // start() already runs init() internally. Calling init() here as well
+          // would init() every consumer twice, and the second pass re-subscribes
+          // the queue to its topic — which conflicts with subscription attributes
+          // (filter policy, redrive policy) applied on the first pass. start()
+          // also provisions all resources, so nothing is lost by dropping the
+          // separate init() barrier.
           await Promise.all(consumers.map((c) => c.start()))
           this.internalConsumers = [...consumers]
         } catch (err) {

packages/sqs/lib/triggers/SnsTopicGroupInvalidationTrigger.ts

@@ -1,18 +1,14 @@
-import type {
-  SNSSQSConsumerDependencies,
-  SNSSQSConsumerOptions,
-} from '@message-queue-toolkit/sns'
+import type { SNSSQSConsumerDependencies } from '@message-queue-toolkit/sns'
 import {
   AbstractSqsTrigger,
   type InternalConsumerHandle,
   SnsTopicTriggerConsumer,
 } from './AbstractSqsTrigger.js'
+import { buildGroupBindings, type GroupBinding } from './bindingHelpers.js'
 import {
-  type BindingHandlerContext,
-  buildGroupBindings,
-  type GroupBinding,
-} from './bindingHelpers.js'
-import type { SnsTopicSourceConfig } from './SnsTopicInvalidationTrigger.js'
+  buildSnsTriggerConsumerOptions,
+  type SnsTopicSourceConfig,
+} from './SnsTopicInvalidationTrigger.js'
 import type { GroupInvalidationTarget, TriggerErrorHandler } from './types.js'
 
 /**
@@ -64,38 +60,18 @@ 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(
+    const built = buildGroupBindings(
       bindings,
       messageTypeField,
       this.target,
       channel,
       this.errorHandler,
     )
 
-    // 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 },
-    }
-
     return new SnsTopicTriggerConsumer(
       this.dependencies,
-      options as SNSSQSConsumerOptions<
-        object,
-        BindingHandlerContext<GroupInvalidationTarget>,
-        undefined
-      >,
-      context,
+      buildSnsTriggerConsumerOptions(consumerOptions, built),
+      built.context,
     )
   }
 }

packages/sqs/lib/triggers/SnsTopicInvalidationTrigger.ts

@@ -10,6 +10,7 @@ import {
 import {
   type BindingHandlerContext,
   buildFlatBindings,
+  type BuiltBindings,
   type FlatBinding,
 } from './bindingHelpers.js'
 import type { InvalidationTarget, TriggerErrorHandler } from './types.js'
@@ -81,42 +82,84 @@ 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(
+    const built = buildFlatBindings(
       bindings,
       messageTypeField,
       this.target,
       channel,
       this.errorHandler,
     )
 
-    // 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 },
-    }
-
     return new SnsTopicTriggerConsumer(
       this.dependencies,
-      options as SNSSQSConsumerOptions<
-        object,
-        BindingHandlerContext<InvalidationTarget>,
-        undefined
-      >,
-      context,
+      buildSnsTriggerConsumerOptions(consumerOptions, built),
+      built.context,
     )
   }
 }
 
+/**
+ * Build the underlying SNS→SQS consumer options for a trigger source.
+ *
+ * A trigger is not a full consumer: it owns its handlers and its subscription
+ * *filtering*. So while the source still supports spreading in a pre-resolved
+ * options object — e.g. `@lokalise/aws-config`'s `resolveConsumerOptions(...)` —
+ * for `creationConfig`/`locatorConfig`, `deadLetterQueue`,
+ * `concurrentConsumersAmount`, `consumerOverrides`, etc., the two things the
+ * trigger genuinely owns are reset here rather than left for the caller to
+ * un-set at every call site:
+ *
+ * - **`handlers`** — always rebuilt from `bindings` (the caller's destructure
+ *   already drops any spread-in handlers; we re-inject the binding-derived ones).
+ * - **subscription filter policy** (`subscriptionConfig.Attributes.FilterPolicy`
+ *   / `FilterPolicyScope`) — `resolveConsumerOptions` derives a `FilterPolicy`
+ *   from the *consumer's* handlers. The trigger supplies its own handlers from
+ *   `bindings`, so the policy the resolver built (from the empty handler list it
+ *   was given) would reject every message. We drop just those keys and default
+ *   the subscription to accept-all.
+ *
+ * Everything else on `subscriptionConfig` is preserved — notably any
+ * `Attributes.RedrivePolicy` (the SNS subscription-level dead-letter queue) and
+ * a spread-in `subscriptionDeadLetterQueue` both flow through untouched.
+ */
+export function buildSnsTriggerConsumerOptions<TTarget>(
+  consumerOptions: SnsTopicSourceConfig,
+  built: Pick<BuiltBindings<TTarget>, 'handlers' | 'messageTypeResolver'>,
+): SNSSQSConsumerOptions<object, BindingHandlerContext<TTarget>, undefined> {
+  const { subscriptionConfig, ...rest } = consumerOptions
+
+  return {
+    ...rest,
+    handlers: built.handlers,
+    messageTypeResolver: built.messageTypeResolver,
+    subscriptionConfig: stripSubscriptionFilterPolicy(subscriptionConfig),
+  } as SNSSQSConsumerOptions<object, BindingHandlerContext<TTarget>, undefined>
+}
+
+/**
+ * Strip only the handler-derived filter-policy keys from a subscription config,
+ * leaving every other attribute (e.g. `RedrivePolicy`, `RawMessageDelivery`)
+ * intact. Defaults `updateAttributesIfExists` to `false` when unset.
+ */
+function stripSubscriptionFilterPolicy(
+  subscriptionConfig: SnsTopicSourceConfig['subscriptionConfig'],
+): SnsTopicSourceConfig['subscriptionConfig'] {
+  const { Attributes, updateAttributesIfExists, ...rest } = subscriptionConfig ?? {}
+  const {
+    FilterPolicy: _filterPolicy,
+    FilterPolicyScope: _filterPolicyScope,
+    ...remainingAttributes
+  } = Attributes ?? {}
+
+  return {
+    updateAttributesIfExists: updateAttributesIfExists ?? false,
+    ...rest,
+    // Omit Attributes entirely when nothing but the filter policy was present,
+    // so the subscription stays accept-all instead of carrying an empty object.
+    ...(Object.keys(remainingAttributes).length > 0 ? { Attributes: remainingAttributes } : {}),
+  }
+}
+
 /** Human-readable channel name derived from an SNS-topic source config. */
 export function deriveSnsTopicChannelName(source: SnsTopicInvalidationSource): string {
   if (source.creationConfig?.topic?.Name) return source.creationConfig.topic.Name