diff --git a/packages/sqs/README.md b/packages/sqs/README.md index dc537e1..0663061 100644 --- a/packages/sqs/README.md +++ b/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' }) diff --git a/packages/sqs/index.ts b/packages/sqs/index.ts index f86cfad..95e1896 100644 --- a/packages/sqs/index.ts +++ b/packages/sqs/index.ts @@ -86,6 +86,7 @@ export { type SqsQueueSourceConfig, } from './lib/triggers/SqsQueueInvalidationTrigger.js' export { + buildSnsTriggerConsumerOptions, deriveSnsTopicChannelName, SnsTopicInvalidationTrigger, type SnsTopicInvalidationSource, diff --git a/packages/sqs/lib/triggers/AbstractSqsTrigger.ts b/packages/sqs/lib/triggers/AbstractSqsTrigger.ts index e625de3..a8e5ae5 100644 --- a/packages/sqs/lib/triggers/AbstractSqsTrigger.ts +++ b/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) { diff --git a/packages/sqs/lib/triggers/SnsTopicGroupInvalidationTrigger.ts b/packages/sqs/lib/triggers/SnsTopicGroupInvalidationTrigger.ts index 5edf751..e44fc53 100644 --- a/packages/sqs/lib/triggers/SnsTopicGroupInvalidationTrigger.ts +++ b/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,7 +60,7 @@ 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, @@ -72,30 +68,10 @@ export class SnsTopicGroupInvalidationTrigger extends AbstractSqsTrigger { 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, - undefined - >, - context, + buildSnsTriggerConsumerOptions(consumerOptions, built), + built.context, ) } } diff --git a/packages/sqs/lib/triggers/SnsTopicInvalidationTrigger.ts b/packages/sqs/lib/triggers/SnsTopicInvalidationTrigger.ts index 6c4eb8b..eda0586 100644 --- a/packages/sqs/lib/triggers/SnsTopicInvalidationTrigger.ts +++ b/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,7 +82,7 @@ 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, @@ -89,34 +90,76 @@ export class SnsTopicInvalidationTrigger extends AbstractSqsTrigger { 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, - 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( + consumerOptions: SnsTopicSourceConfig, + built: Pick, 'handlers' | 'messageTypeResolver'>, +): SNSSQSConsumerOptions, undefined> { + const { subscriptionConfig, ...rest } = consumerOptions + + return { + ...rest, + handlers: built.handlers, + messageTypeResolver: built.messageTypeResolver, + subscriptionConfig: stripSubscriptionFilterPolicy(subscriptionConfig), + } as SNSSQSConsumerOptions, 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 diff --git a/packages/sqs/test/triggerInternals.spec.ts b/packages/sqs/test/triggerInternals.spec.ts new file mode 100644 index 0000000..0dca7ec --- /dev/null +++ b/packages/sqs/test/triggerInternals.spec.ts @@ -0,0 +1,182 @@ +import { describe, expect, it } from 'vitest' +import { z } from 'zod' +import { + AbstractSqsTrigger, + type InternalConsumerHandle, +} from '../lib/triggers/AbstractSqsTrigger.js' +import { buildFlatBindings } from '../lib/triggers/bindingHelpers.js' +import { buildSnsTriggerConsumerOptions } from '../lib/triggers/SnsTopicInvalidationTrigger.js' + +const EVENT_SCHEMA = z.object({ id: z.string() }) + +function buildBindings() { + return buildFlatBindings( + [{ messageSchema: EVENT_SCHEMA, resolver: () => null }], + undefined, + { invalidate: async () => {}, invalidateMany: async () => {} } as never, + 'test-channel', + undefined, + ) +} + +describe('buildSnsTriggerConsumerOptions', () => { + it('injects the binding-derived handlers and message-type resolver', () => { + const built = buildBindings() + const options = buildSnsTriggerConsumerOptions( + { creationConfig: { topic: { Name: 't' }, queue: { QueueName: 'q' } } }, + built, + ) + + expect(options.handlers).toBe(built.handlers) + expect(options.messageTypeResolver).toBe(built.messageTypeResolver) + }) + + it('drops the handler-derived subscription filter policy (Attributes)', () => { + const built = buildBindings() + const options = buildSnsTriggerConsumerOptions( + { + creationConfig: { topic: { Name: 't' }, queue: { QueueName: 'q' } }, + subscriptionConfig: { + updateAttributesIfExists: true, + // A reject-all policy resolveConsumerOptions() derived from empty handlers. + Attributes: { + FilterPolicy: JSON.stringify({ type: ['__never__'] }), + FilterPolicyScope: 'MessageBody', + }, + }, + }, + built, + ) + + expect(options.subscriptionConfig?.Attributes).toBeUndefined() + // ...but the caller's other subscription settings are preserved. + expect(options.subscriptionConfig?.updateAttributesIfExists).toBe(true) + }) + + it('preserves a subscription-level RedrivePolicy while dropping the filter policy', () => { + const redrivePolicy = JSON.stringify({ deadLetterTargetArn: 'arn:aws:sqs:::sub-dlq' }) + const options = buildSnsTriggerConsumerOptions( + { + creationConfig: { topic: { Name: 't' }, queue: { QueueName: 'q' } }, + subscriptionConfig: { + updateAttributesIfExists: true, + Attributes: { + FilterPolicy: JSON.stringify({ type: ['__never__'] }), + RedrivePolicy: redrivePolicy, + RawMessageDelivery: 'true', + }, + }, + }, + buildBindings(), + ) + + // The subscription DLQ (and other non-filter attributes) survive... + expect(options.subscriptionConfig?.Attributes).toEqual({ + RedrivePolicy: redrivePolicy, + RawMessageDelivery: 'true', + }) + // ...while the handler-derived filter policy is gone. + expect(options.subscriptionConfig?.Attributes?.FilterPolicy).toBeUndefined() + }) + + it('defaults updateAttributesIfExists to false when no subscriptionConfig is given', () => { + const options = buildSnsTriggerConsumerOptions( + { creationConfig: { topic: { Name: 't' }, queue: { QueueName: 'q' } } }, + buildBindings(), + ) + + expect(options.subscriptionConfig?.updateAttributesIfExists).toBe(false) + }) + + it('forwards a spread-in subscriptionDeadLetterQueue untouched', () => { + const options = buildSnsTriggerConsumerOptions( + { + creationConfig: { topic: { Name: 't' }, queue: { QueueName: 'q' } }, + deadLetterQueue: { + redrivePolicy: { maxReceiveCount: 3 }, + creationConfig: { queue: { QueueName: 'q-dlq' } }, + }, + // Not part of the vendored typed surface, but rides along on a + // resolveConsumerOptions() spread / newer toolkit versions. + subscriptionDeadLetterQueue: { reuseConsumerDeadLetterQueue: true }, + } as Parameters[0], + buildBindings(), + ) + + // The subscription-level DLQ directive flows through to the consumer. + expect((options as { subscriptionDeadLetterQueue?: unknown }).subscriptionDeadLetterQueue).toEqual( + { reuseConsumerDeadLetterQueue: true }, + ) + // The queue-level DLQ is forwarded untouched as well. + expect(options.deadLetterQueue).toEqual({ + redrivePolicy: { maxReceiveCount: 3 }, + creationConfig: { queue: { QueueName: 'q-dlq' } }, + }) + }) + + it('forwards unrelated consumer options untouched', () => { + const options = buildSnsTriggerConsumerOptions( + { + creationConfig: { topic: { Name: 't' }, queue: { QueueName: 'q' } }, + concurrentConsumersAmount: 4, + }, + buildBindings(), + ) + + expect(options.creationConfig).toEqual({ topic: { Name: 't' }, queue: { QueueName: 'q' } }) + expect(options.concurrentConsumersAmount).toBe(4) + }) +}) + +describe('AbstractSqsTrigger lifecycle', () => { + class CountingHandle implements InternalConsumerHandle { + initCalls = 0 + startCalls = 0 + closeCalls = 0 + async init() { + this.initCalls++ + } + async start() { + this.startCalls++ + } + async close() { + this.closeCalls++ + } + } + + class TestTrigger extends AbstractSqsTrigger { + constructor(private readonly handles: readonly InternalConsumerHandle[]) { + super() + } + protected createConsumers(): readonly InternalConsumerHandle[] { + return this.handles + } + } + + it('starts each consumer exactly once and never calls init() separately', async () => { + // The underlying consumer's start() runs init() internally; the trigger must + // not also call init(), or every queue re-subscribes and conflicts with the + // attributes (filter/redrive policy) set on the first subscribe. + const handles = [new CountingHandle(), new CountingHandle()] + const trigger = new TestTrigger(handles) + + await trigger.start() + + for (const handle of handles) { + expect(handle.startCalls).toBe(1) + expect(handle.initCalls).toBe(0) + } + }) + + it('closes every consumer on stop()', async () => { + const handles = [new CountingHandle(), new CountingHandle()] + const trigger = new TestTrigger(handles) + + await trigger.start() + await trigger.stop() + + for (const handle of handles) { + expect(handle.closeCalls).toBe(1) + } + }) +})