diff --git a/packages/serverless-offline-dynamodb-streams/package.json b/packages/serverless-offline-dynamodb-streams/package.json index fefc0d6..f659721 100644 --- a/packages/serverless-offline-dynamodb-streams/package.json +++ b/packages/serverless-offline-dynamodb-streams/package.json @@ -35,6 +35,7 @@ "dependencies": { "@aws-sdk/client-dynamodb": "^3.0.0", "@aws-sdk/client-dynamodb-streams": "^3.0.0", + "@aws-sdk/client-sqs": "^3.0.0", "@smithy/node-http-handler": "^4.0.0", "dynamodb-streams-readable": "^3.0.0", "lodash": "^4.17.21", diff --git a/packages/serverless-offline-dynamodb-streams/src/destinations.js b/packages/serverless-offline-dynamodb-streams/src/destinations.js new file mode 100644 index 0000000..98ac992 --- /dev/null +++ b/packages/serverless-offline-dynamodb-streams/src/destinations.js @@ -0,0 +1,164 @@ +const {GetQueueUrlCommand, SendMessageCommand} = require('@aws-sdk/client-sqs'); +const {get, has, isNil, isPlainObject, isString} = require('lodash/fp'); + +// --------------------------------------------------------------------------- +// Lambda async-invoke destinations (onFailure / onSuccess) — pure helpers + a +// single injected-client dispatcher. Extracted from EventBridge._onFailure so the +// same destinations contract can be reused by serverless-offline-sqs and +// serverless-offline-dynamodb-streams (spec B2). SQS-only targets, by design. +// --------------------------------------------------------------------------- + +// CloudFormation pseudo-parameters we can resolve offline from the plugin's region/accountId. +// `resources` (the resolved CFN Resources map) is optional and only used to resolve a Ref / Fn::GetAtt +// that points at an AWS::SQS::Queue declared in the stack (spec EARS6). +const pseudoParams = (region, accountId, resources) => ({ + 'AWS::Region': region, + 'AWS::AccountId': accountId, + region, + accountId, + resources +}); + +// resolveQueueRefArn(resources, refName, region, accountId) -> string | undefined +// Mirrors serverless-offline-sqs' resolveSqsRefArn: a Ref / Fn::GetAtt targeting an AWS::SQS::Queue +// declared in resources.Resources resolves to a stable ARN whose last segment is the queue name +// (preferring the declared Properties.QueueName, falling back to the logical id serverless uses for an +// auto-named queue locally). Returns undefined for anything that is not an SQS queue. Pure, non-throwing. +const resolveQueueRefArn = (resources, refName, region, accountId) => { + if (get([refName, 'Type'], resources) !== 'AWS::SQS::Queue') return undefined; + const queueName = get([refName, 'Properties', 'QueueName'], resources) || refName; + return `arn:aws:sqs:${region}:${accountId}:${queueName}`; +}; + +// resolveCfnValue(value, ctx) -> string | undefined +// Mirrors serverless-offline-sqs' resolver: flattens the intrinsic / pseudo-parameter forms a +// destination ARN can take into a plain string. Resolves a Ref / Fn::GetAtt to an AWS::SQS::Queue +// declared in `ctx.resources` to that queue's ARN (spec EARS6). Returns undefined for anything +// unresolvable offline (Fn::ImportValue, a Ref/Fn::GetAtt to a non-queue or absent resource, a +// non-string) so the caller can warn + skip. +const resolveCfnValue = (value, ctx) => { + if (isString(value)) + return value.replace(/[#$]\{(AWS::[A-Za-z]+)\}/g, (match, key) => + isNil(ctx[key]) ? match : ctx[key] + ); + + if (isPlainObject(value)) { + if (has('Ref', value)) { + // A Ref to a pseudo-parameter (AWS::Region/AWS::AccountId) resolves to that value; otherwise try + // resolving it against the stack Resources as an AWS::SQS::Queue (spec EARS6). + if (!isNil(ctx[value.Ref])) return ctx[value.Ref]; + return resolveQueueRefArn(ctx.resources, value.Ref, ctx.region, ctx.accountId); + } + if (has('Fn::GetAtt', value)) { + // CloudFormation `{Fn::GetAtt: [, 'Arn']}` -> the queue's ARN (spec EARS6). + const [resourceName, attribute] = value['Fn::GetAtt']; + if (attribute !== 'Arn') return undefined; + return resolveQueueRefArn(ctx.resources, resourceName, ctx.region, ctx.accountId); + } + if (has('Fn::Sub', value)) return resolveCfnValue(value['Fn::Sub'], ctx); + if (has('Fn::Join', value)) { + const [separator, parts] = value['Fn::Join']; + return (parts || []).map(part => resolveCfnValue(part, ctx)).join(separator); + } + // Fn::ImportValue (and any other intrinsic) is not resolvable offline. + } + + return undefined; +}; + +// resolveDestinationArn(target, ctx) -> string | undefined +// A destination target is a literal string ARN, a {arn: }, or a bare intrinsic. +// Resolves any of these to a string ARN through resolveCfnValue, or undefined when unresolvable. +const resolveDestinationArn = (target, ctx = {}) => { + if (isNil(target)) return undefined; + const arn = isPlainObject(target) && has('arn', target) ? target.arn : target; + const resolved = resolveCfnValue(arn, ctx); + return isString(resolved) && resolved !== '' ? resolved : undefined; +}; + +// queueNameFromArn(arn) -> string | undefined +// The queue name is the final ':'-delimited segment (robust to pseudo-params that inject extra ':'). +const queueNameFromArn = arn => { + if (!isString(arn) || arn === '') return undefined; + const segments = arn.split(':'); + return segments[segments.length - 1] || undefined; +}; + +// The eventbridge onFailure contract, preserved byte-for-byte (spec EARS8). +const buildFailurePayload = (requestPayload, err) => ({ + requestPayload, + responsePayload: {errorMessage: err && err.message, errorType: err && err.name} +}); + +const buildSuccessPayload = (requestPayload, result) => ({ + requestPayload, + responsePayload: result +}); + +// dispatchDestination({client, target, ctx, payload, log}) -> Promise +// The single impure edge (still unit-tested with an injected fake client). Best-effort: +// - nil target -> no-op (no client call) +// - unresolvable ARN -> warn + skip +// - resolvable target -> GetQueueUrl + SendMessage +// ANY error is log.warning-ed and swallowed — destinations dispatch never throws or blocks the +// originating event source (spec EARS7, edge cases 2 & 3). +const dispatchDestination = async ({client, target, ctx = {}, payload, log}) => { + if (isNil(target)) return; + + const arn = resolveDestinationArn(target, ctx); + if (isNil(arn)) { + log.warning( + `destinations: cannot resolve target ARN offline, skipping: ${JSON.stringify(target)}` + ); + return; + } + + try { + const queueName = queueNameFromArn(arn); + const {QueueUrl} = await client.send(new GetQueueUrlCommand({QueueName: queueName})); + await client.send(new SendMessageCommand({QueueUrl, MessageBody: JSON.stringify(payload)})); + } catch (err) { + log.warning(err && err.stack ? err.stack : String(err)); + } +}; + +// runDestinations(args) -> Promise +// The orchestrator the event sources call. Gated behind simulateDestinations (default true). On an +// error it dispatches destinations.onFailure; otherwise it dispatches destinations.onSuccess. The SQS +// client is built lazily via makeClient() so NO client is constructed when there is nothing to send +// (spec EARS2, edge case 1). +const runDestinations = async ({ + simulateDestinations, + destinations, + ctx = {}, + makeClient, + log, + requestPayload, + error, + result +}) => { + if (simulateDestinations === false) return; + + const target = error + ? destinations && destinations.onFailure + : destinations && destinations.onSuccess; + if (isNil(target)) return; + + const payload = error + ? buildFailurePayload(requestPayload, error) + : buildSuccessPayload(requestPayload, result); + + await dispatchDestination({client: makeClient(), target, ctx, payload, log}); +}; + +module.exports = { + pseudoParams, + resolveQueueRefArn, + resolveCfnValue, + resolveDestinationArn, + queueNameFromArn, + buildFailurePayload, + buildSuccessPayload, + dispatchDestination, + runDestinations +}; diff --git a/packages/serverless-offline-dynamodb-streams/src/dynamodb-streams.js b/packages/serverless-offline-dynamodb-streams/src/dynamodb-streams.js index ab41707..310d5c9 100644 --- a/packages/serverless-offline-dynamodb-streams/src/dynamodb-streams.js +++ b/packages/serverless-offline-dynamodb-streams/src/dynamodb-streams.js @@ -10,11 +10,13 @@ const { GetRecordsCommand, GetShardIteratorCommand } = require('@aws-sdk/client-dynamodb-streams'); +const {SQSClient} = require('@aws-sdk/client-sqs'); const DynamodbStreamsReadable = require('dynamodb-streams-readable'); const {assign, isEmpty, last, get} = require('lodash/fp'); const {normalizeLog} = require('./log'); const {buildClientConfig} = require('./client-config'); +const {pseudoParams, runDestinations} = require('./destinations'); const {buildCallbackClient} = require('./callback-adapter'); const DynamodbStreamsEventDefinition = require('./dynamodb-streams-event-definition'); const DynamodbStreamsEvent = require('./dynamodb-streams-event'); @@ -95,6 +97,9 @@ class DynamodbStreams { this.readables = []; + // B2 destinations: lazily built only when a function declares a destination (see _sqsClient). + this.sqsClient = null; + // #178 (ddb-streams-checkpoint): resolve and load the restart checkpoint once. The // configured `checkpointFile` (custom option) is anchored at process cwd; a missing // file is a clean cold start (loadState returns {}). The in-memory map is the single @@ -105,7 +110,9 @@ class DynamodbStreams { create(events) { return Promise.all( - events.map(({functionKey, dynamodbStreams}) => this._create(functionKey, dynamodbStreams)) + events.map(({functionKey, destinations, dynamodbStreams}) => + this._create(functionKey, destinations, dynamodbStreams) + ) ); } @@ -117,14 +124,14 @@ class DynamodbStreams { this.readables.forEach(readable => readable.pause()); } - _create(functionKey, rawDynamodbStreamsEventDefinition) { + _create(functionKey, destinations, rawDynamodbStreamsEventDefinition) { const dynamodbStreamsEvent = new DynamodbStreamsEventDefinition( rawDynamodbStreamsEventDefinition, this.options.region, this.options.accountId ); - return this._dynamodbStreamsEvent(functionKey, dynamodbStreamsEvent); + return this._dynamodbStreamsEvent(functionKey, destinations, dynamodbStreamsEvent); } // #248 (aws-sdk v3): the bounded waiter, extracted as a seam so unit tests can stub the @@ -153,7 +160,7 @@ class DynamodbStreams { } } - async _dynamodbStreamsEvent(functionKey, dynamodbStreamsEvent) { + async _dynamodbStreamsEvent(functionKey, destinations, dynamodbStreamsEvent) { const { enabled, tableName, @@ -229,20 +236,27 @@ class DynamodbStreams { return cb(); } + const event = new DynamodbStreamsEvent(matching, this.options.region, arn); + const task = async remainingAttempts => { try { const lambdaFunction = this.lambda.get(functionKey); - const event = new DynamodbStreamsEvent(matching, this.options.region, arn); lambdaFunction.setEvent(event); - await lambdaFunction.runHandler(); + const result = await lambdaFunction.runHandler(); + // B2 destinations: a clean handler run dispatches `destinations.onSuccess` once + // (records as requestPayload). Best-effort — never blocks the stream. + await this._dispatchDestination(destinations, event, {result}); } catch (err) { this.log.warning(err.stack); if (remainingAttempts > 0) { await delay(500); return task(remainingAttempts - 1); } + // B2 destinations: retries exhausted — dispatch `destinations.onFailure` with the + // stream records + error. A handler that succeeds on a retry never reaches here. + await this._dispatchDestination(destinations, event, {error: err}); } }; @@ -268,6 +282,34 @@ class DynamodbStreams { }); } + // B2 destinations: lazily build (and memoize) the SQS client used for destination dispatch, so no + // client is constructed unless a function actually declares a destination. + _sqsClient() { + if (!this.sqsClient) this.sqsClient = new SQSClient(buildClientConfig(this.options)); + return this.sqsClient; + } + + // B2 destinations: best-effort onFailure/onSuccess dispatch to the resolved SQS target. Gated + // behind simulateDestinations, never throws (a dispatch failure must not break the stream). + async _dispatchDestination(destinations, requestPayload, {error, result} = {}) { + await runDestinations({ + simulateDestinations: this.options.simulateDestinations, + destinations, + // Surface the CFN Resources map so a Ref/Fn::GetAtt destination ARN resolves against the + // declared AWS::SQS::Queue (spec EARS6). + ctx: pseudoParams( + this.options.region, + this.options.accountId, + get(['resources', 'Resources'], this.options) || {} + ), + makeClient: () => this._sqsClient(), + log: this.log, + requestPayload, + error, + result + }); + } + // #178 (ddb-streams-checkpoint): persist the per-shard high-water mark. Thin I/O edge: // updates the in-memory map (pure setCheckpoint) then best-effort writes it. A write // failure is logged, never thrown — losing a checkpoint must not break the stream. diff --git a/packages/serverless-offline-dynamodb-streams/src/index.js b/packages/serverless-offline-dynamodb-streams/src/index.js index 69256a6..0337430 100644 --- a/packages/serverless-offline-dynamodb-streams/src/index.js +++ b/packages/serverless-offline-dynamodb-streams/src/index.js @@ -129,7 +129,11 @@ class ServerlessOfflineDynamodbStreams { omitUndefined(provider), omitUndefined(pick(['location', 'localEnvironment'], offlineOptions)), // serverless-webpack support omitUndefined(customOptions), - omitUndefined(this.cliOptions) + omitUndefined(this.cliOptions), + // B2 destinations: surface the raw CFN Resources so a Ref/Fn::GetAtt destination ARN can resolve + // to an AWS::SQS::Queue declared in the stack (spec EARS6). Always overwritten so user config + // can never shadow it. + {resources: get(['service', 'resources'], this.serverless) || {Resources: {}}} ); this.log.debug('dynamodb-streams options:', this.options); @@ -161,6 +165,7 @@ class ServerlessOfflineDynamodbStreams { dynamodbStreamsEvents.push({ functionKey, handler: functionDefinition.handler, + destinations: get('destinations', functionDefinition), dynamodbStreams: this._resolveFn(stream) }); } diff --git a/packages/serverless-offline-dynamodb-streams/test/index.js b/packages/serverless-offline-dynamodb-streams/test/index.js index cb39505..b708d35 100644 --- a/packages/serverless-offline-dynamodb-streams/test/index.js +++ b/packages/serverless-offline-dynamodb-streams/test/index.js @@ -17,6 +17,14 @@ const { TABLE_DESCRIBE_MAX_ATTEMPTS } = DynamodbStreams; const {resolveTableName} = require('../src/resolve-arn'); +const { + resolveDestinationArn, + queueNameFromArn, + buildFailurePayload, + buildSuccessPayload, + dispatchDestination, + runDestinations +} = require('../src/destinations'); const {recordMatchesFilterPatterns, filterRecords} = require('../src/filter-patterns'); const { DEFAULT_STATE_FILE, @@ -688,6 +696,7 @@ test('write: a batch fully dropped by filterPatterns still advances the checkpoi await streams._dynamodbStreamsEvent( 'fnKey', + undefined, new DynamodbStreamsEventDefinition( { arn: HARNESS_STREAM_ARN, @@ -729,6 +738,7 @@ test('write: a partially-matching batch runs the handler and advances to the FUL await streams._dynamodbStreamsEvent( 'fnKey', + undefined, new DynamodbStreamsEventDefinition( { arn: HARNESS_STREAM_ARN, @@ -837,7 +847,7 @@ test('_dynamodbStreamsEvent rejects with a clear error when the table is missing streams._describeTable = () => Promise.reject(new Error('Table ghostTable not found')); const error = await t.throwsAsync(() => - streams._dynamodbStreamsEvent('fnKey', missingTableEvent()) + streams._dynamodbStreamsEvent('fnKey', undefined, missingTableEvent()) ); t.regex(error.message, /ghostTable/); t.is(streams.readables.length, 0, 'no readable was wired for the missing resource'); @@ -849,7 +859,9 @@ test('_dynamodbStreamsEvent warns and skips a missing table when continueOnMissi const {streams, warnings} = buildMissingTableStreams({continueOnMissingResource: true}); streams._describeTable = () => Promise.reject(new Error('Table ghostTable not found')); - await t.notThrowsAsync(() => streams._dynamodbStreamsEvent('fnKey', missingTableEvent())); + await t.notThrowsAsync(() => + streams._dynamodbStreamsEvent('fnKey', undefined, missingTableEvent()) + ); t.is(streams.readables.length, 0, 'the missing event source is skipped'); t.is(warnings.length, 1, 'a single warning was logged'); t.regex(warnings[0], /ghostTable/, 'the warning names the table'); @@ -862,7 +874,9 @@ test('_dynamodbStreamsEvent warns and skips a table without streams when continu // Table exists, but LatestStreamArn is absent -> assertStreamEnabled throws. streams._describeTable = () => Promise.resolve({Table: {LatestStreamArn: undefined}}); - await t.notThrowsAsync(() => streams._dynamodbStreamsEvent('fnKey', missingTableEvent())); + await t.notThrowsAsync(() => + streams._dynamodbStreamsEvent('fnKey', undefined, missingTableEvent()) + ); t.is(streams.readables.length, 0); t.is(warnings.length, 1); t.regex(warnings[0], /ghostTable/); @@ -874,7 +888,7 @@ test('_dynamodbStreamsEvent rejects for a table without streams when no opt-in ( streams._describeTable = () => Promise.resolve({Table: {LatestStreamArn: undefined}}); const error = await t.throwsAsync(() => - streams._dynamodbStreamsEvent('fnKey', missingTableEvent()) + streams._dynamodbStreamsEvent('fnKey', undefined, missingTableEvent()) ); t.regex(error.message, /streams/i); }); @@ -887,3 +901,228 @@ test('src/dynamodb-streams.js no longer re-waits unconditionally in _describeTab 'the unbounded `catch { return this._describeTable(tableName) }` must be removed' ); }); + +// --------------------------------------------------------------------------- +// Lambda async destinations (onFailure / onSuccess) — spec B2 +// --------------------------------------------------------------------------- + +const CTX = {'AWS::Region': 'eu-west-1', 'AWS::AccountId': '000000000000'}; + +const fakeSqsClient = (impl = {}) => { + const calls = []; + return { + calls, + send: command => { + const name = command.constructor.name; + calls.push({name, input: command.input}); + if (name === 'GetQueueUrlCommand') { + if (impl.getQueueUrlError) throw impl.getQueueUrlError; + return Promise.resolve({QueueUrl: impl.queueUrl || 'http://localhost:9324/000/q'}); + } + if (name === 'SendMessageCommand') return Promise.resolve({MessageId: 'm-1'}); + return Promise.resolve({}); + } + }; +}; + +test('destinations: resolveDestinationArn / queueNameFromArn (EARS1, EARS6)', t => { + t.is(resolveDestinationArn({arn: {Ref: 'AWS::Region'}}, CTX), 'eu-west-1'); + t.is(resolveDestinationArn({arn: {'Fn::ImportValue': 'X'}}, CTX), undefined); + t.is(queueNameFromArn('arn:aws:sqs:eu-west-1:0:my-dlq'), 'my-dlq'); +}); + +test('destinations: build*Payload shapes (EARS5, EARS8)', t => { + t.deepEqual(buildFailurePayload({a: 1}, Object.assign(new Error('boom'), {name: 'E'})), { + requestPayload: {a: 1}, + responsePayload: {errorMessage: 'boom', errorType: 'E'} + }); + t.deepEqual(buildSuccessPayload({a: 1}, {ok: 1}), { + requestPayload: {a: 1}, + responsePayload: {ok: 1} + }); +}); + +test('destinations: dispatchDestination swallows a client error and warns (EARS7)', async t => { + const client = fakeSqsClient({getQueueUrlError: new Error('gone')}); + const warnings = []; + await t.notThrowsAsync( + dispatchDestination({ + client, + target: 'arn:aws:sqs:eu-west-1:0:dlq', + ctx: CTX, + payload: {}, + log: normalizeLog({warning: m => warnings.push(m)}) + }) + ); + t.is(warnings.length, 1); +}); + +test('destinations: runDestinations is a no-op (no client) when simulateDestinations is false (EARS2)', async t => { + let built = false; + await runDestinations({ + simulateDestinations: false, + destinations: {onFailure: 'arn:aws:sqs:eu-west-1:0:dlq'}, + ctx: CTX, + makeClient: () => { + built = true; + return fakeSqsClient(); + }, + log: normalizeLog(), + requestPayload: {}, + error: new Error('boom') + }); + t.false(built); +}); + +// A harness that drives the REAL writable with a configurable handler + injected SQS client, so we +// can exercise the retry-vs-exhaustion destinations wiring end-to-end (no AWS, no docker). +const buildDestinationsHarness = ({runHandler, destinations, sqsClient}) => { + const file = path.join(fs.mkdtempSync(path.join(os.tmpdir(), 'ddb-dest-')), DEFAULT_STATE_FILE); + const handlerCalls = []; + const lambda = { + get: () => ({ + setEvent: event => handlerCalls.push(event), + runHandler + }) + }; + + const streams = new DynamodbStreams( + lambda, + { + region: 'eu-west-1', + accountId: '000000000000', + checkpointFile: file, + simulateDestinations: true + }, + {} + ); + + streams._describeTable = () => Promise.resolve({Table: {LatestStreamArn: HARNESS_STREAM_ARN}}); + streams.streamsClient.send = () => + Promise.resolve({StreamDescription: {Shards: [{ShardId: SHARD_ID}]}}); + // Inject the fake SQS client so destinations dispatch hits no real AWS. + streams.sqsClient = sqsClient; + + return {streams, handlerCalls, destinations}; +}; + +const runOneBatch = async ({runHandler, destinations, sqsClient}) => { + const {streams} = buildDestinationsHarness({runHandler, destinations, sqsClient}); + await streams._dynamodbStreamsEvent( + 'fnKey', + destinations, + new DynamodbStreamsEventDefinition( + {arn: HARNESS_STREAM_ARN, startingPosition: 'TRIM_HORIZON', maximumRetryAttempts: 3}, + 'eu-west-1', + '000000000000' + ) + ); + const writable = writableOf(streams.readables[0]); + await writeChunk(writable, [ + {...insertRecord, dynamodb: {...insertRecord.dynamodb, SequenceNumber: '100'}} + ]); +}; + +// EARS4: retries exhausted -> exactly one onFailure dispatch with the stream records + error. +test('DDB: a handler that exhausts its retries dispatches onFailure once (EARS4)', async t => { + const client = fakeSqsClient(); + await runOneBatch({ + runHandler: () => Promise.reject(Object.assign(new Error('boom'), {name: 'E'})), + destinations: {onFailure: 'arn:aws:sqs:eu-west-1:0:my-dlq'}, + sqsClient: client + }); + const sends = client.calls.filter(({name}) => name === 'SendMessageCommand'); + t.is(sends.length, 1, 'exactly one onFailure dispatch'); + const body = JSON.parse(sends[0].input.MessageBody); + t.is(body.responsePayload.errorMessage, 'boom'); + t.is(body.responsePayload.errorType, 'E'); + t.truthy(body.requestPayload.Records, 'requestPayload carries the stream records'); +}); + +// Edge 5: a handler that succeeds on a retry does NOT fire onFailure; onSuccess fires once. +test('DDB: a handler that succeeds on a retry fires onSuccess once, never onFailure (edge 5, EARS5)', async t => { + const client = fakeSqsClient(); + let attempts = 0; + await runOneBatch({ + runHandler: () => { + attempts += 1; + return attempts < 2 + ? Promise.reject(new Error('transient')) + : Promise.resolve({statusCode: 200}); + }, + destinations: { + onFailure: 'arn:aws:sqs:eu-west-1:0:dlq', + onSuccess: 'arn:aws:sqs:eu-west-1:0:ok' + }, + sqsClient: client + }); + const getQueue = client.calls.find(({name}) => name === 'GetQueueUrlCommand'); + const sends = client.calls.filter(({name}) => name === 'SendMessageCommand'); + t.is(sends.length, 1, 'exactly one dispatch'); + t.is(getQueue.input.QueueName, 'ok', 'dispatched to onSuccess, not onFailure'); + t.deepEqual(JSON.parse(sends[0].input.MessageBody).responsePayload, {statusCode: 200}); +}); + +// EARS5: a clean first-attempt run fires onSuccess once. +test('DDB: a clean handler run dispatches onSuccess once (EARS5)', async t => { + const client = fakeSqsClient(); + await runOneBatch({ + runHandler: () => Promise.resolve({ok: true}), + destinations: {onSuccess: 'arn:aws:sqs:eu-west-1:0:ok'}, + sqsClient: client + }); + const sends = client.calls.filter(({name}) => name === 'SendMessageCommand'); + t.is(sends.length, 1); + t.deepEqual(JSON.parse(sends[0].input.MessageBody).responsePayload, {ok: true}); +}); + +// Edge 1: no destinations declared -> no SQS calls at all. +test('DDB: no destinations -> no SQS dispatch (edge 1)', async t => { + const client = fakeSqsClient(); + await runOneBatch({ + runHandler: () => Promise.resolve({ok: true}), + destinations: undefined, + sqsClient: client + }); + t.deepEqual(client.calls, []); +}); + +// EARS6: a Fn::GetAtt / Ref destination ARN pointing at a Queue declared in this.options.resources +// resolves to that queue. Build a minimal streams object so the test targets only the resolution path. +const buildStreamsForDispatch = (client, resources) => { + const streams = Object.create(DynamodbStreams.prototype); + streams.options = { + region: 'eu-west-1', + accountId: '000000000000', + simulateDestinations: true, + resources: {Resources: resources} + }; + streams.sqsClient = client; + streams.log = normalizeLog(); + return streams; +}; + +test('DDB._dispatchDestination resolves a Fn::GetAtt destination via options.resources (EARS6)', async t => { + const client = fakeSqsClient(); + const streams = buildStreamsForDispatch(client, { + MyDlq: {Type: 'AWS::SQS::Queue', Properties: {QueueName: 'my-dlq'}} + }); + await streams._dispatchDestination( + {onFailure: {'Fn::GetAtt': ['MyDlq', 'Arn']}}, + {Records: [{eventID: '1'}]}, + {error: Object.assign(new Error('boom'), {name: 'E'})} + ); + t.is(client.calls[0].input.QueueName, 'my-dlq'); +}); + +test('DDB._dispatchDestination resolves a {Ref: } destination via options.resources (EARS6)', async t => { + const client = fakeSqsClient(); + const streams = buildStreamsForDispatch(client, {OkQueue: {Type: 'AWS::SQS::Queue'}}); + await streams._dispatchDestination( + {onSuccess: {Ref: 'OkQueue'}}, + {Records: [{eventID: '1'}]}, + {result: {ok: true}} + ); + // No explicit QueueName -> falls back to the logical id. + t.is(client.calls[0].input.QueueName, 'OkQueue'); +}); diff --git a/packages/serverless-offline-eventbridge/src/destinations.js b/packages/serverless-offline-eventbridge/src/destinations.js new file mode 100644 index 0000000..98ac992 --- /dev/null +++ b/packages/serverless-offline-eventbridge/src/destinations.js @@ -0,0 +1,164 @@ +const {GetQueueUrlCommand, SendMessageCommand} = require('@aws-sdk/client-sqs'); +const {get, has, isNil, isPlainObject, isString} = require('lodash/fp'); + +// --------------------------------------------------------------------------- +// Lambda async-invoke destinations (onFailure / onSuccess) — pure helpers + a +// single injected-client dispatcher. Extracted from EventBridge._onFailure so the +// same destinations contract can be reused by serverless-offline-sqs and +// serverless-offline-dynamodb-streams (spec B2). SQS-only targets, by design. +// --------------------------------------------------------------------------- + +// CloudFormation pseudo-parameters we can resolve offline from the plugin's region/accountId. +// `resources` (the resolved CFN Resources map) is optional and only used to resolve a Ref / Fn::GetAtt +// that points at an AWS::SQS::Queue declared in the stack (spec EARS6). +const pseudoParams = (region, accountId, resources) => ({ + 'AWS::Region': region, + 'AWS::AccountId': accountId, + region, + accountId, + resources +}); + +// resolveQueueRefArn(resources, refName, region, accountId) -> string | undefined +// Mirrors serverless-offline-sqs' resolveSqsRefArn: a Ref / Fn::GetAtt targeting an AWS::SQS::Queue +// declared in resources.Resources resolves to a stable ARN whose last segment is the queue name +// (preferring the declared Properties.QueueName, falling back to the logical id serverless uses for an +// auto-named queue locally). Returns undefined for anything that is not an SQS queue. Pure, non-throwing. +const resolveQueueRefArn = (resources, refName, region, accountId) => { + if (get([refName, 'Type'], resources) !== 'AWS::SQS::Queue') return undefined; + const queueName = get([refName, 'Properties', 'QueueName'], resources) || refName; + return `arn:aws:sqs:${region}:${accountId}:${queueName}`; +}; + +// resolveCfnValue(value, ctx) -> string | undefined +// Mirrors serverless-offline-sqs' resolver: flattens the intrinsic / pseudo-parameter forms a +// destination ARN can take into a plain string. Resolves a Ref / Fn::GetAtt to an AWS::SQS::Queue +// declared in `ctx.resources` to that queue's ARN (spec EARS6). Returns undefined for anything +// unresolvable offline (Fn::ImportValue, a Ref/Fn::GetAtt to a non-queue or absent resource, a +// non-string) so the caller can warn + skip. +const resolveCfnValue = (value, ctx) => { + if (isString(value)) + return value.replace(/[#$]\{(AWS::[A-Za-z]+)\}/g, (match, key) => + isNil(ctx[key]) ? match : ctx[key] + ); + + if (isPlainObject(value)) { + if (has('Ref', value)) { + // A Ref to a pseudo-parameter (AWS::Region/AWS::AccountId) resolves to that value; otherwise try + // resolving it against the stack Resources as an AWS::SQS::Queue (spec EARS6). + if (!isNil(ctx[value.Ref])) return ctx[value.Ref]; + return resolveQueueRefArn(ctx.resources, value.Ref, ctx.region, ctx.accountId); + } + if (has('Fn::GetAtt', value)) { + // CloudFormation `{Fn::GetAtt: [, 'Arn']}` -> the queue's ARN (spec EARS6). + const [resourceName, attribute] = value['Fn::GetAtt']; + if (attribute !== 'Arn') return undefined; + return resolveQueueRefArn(ctx.resources, resourceName, ctx.region, ctx.accountId); + } + if (has('Fn::Sub', value)) return resolveCfnValue(value['Fn::Sub'], ctx); + if (has('Fn::Join', value)) { + const [separator, parts] = value['Fn::Join']; + return (parts || []).map(part => resolveCfnValue(part, ctx)).join(separator); + } + // Fn::ImportValue (and any other intrinsic) is not resolvable offline. + } + + return undefined; +}; + +// resolveDestinationArn(target, ctx) -> string | undefined +// A destination target is a literal string ARN, a {arn: }, or a bare intrinsic. +// Resolves any of these to a string ARN through resolveCfnValue, or undefined when unresolvable. +const resolveDestinationArn = (target, ctx = {}) => { + if (isNil(target)) return undefined; + const arn = isPlainObject(target) && has('arn', target) ? target.arn : target; + const resolved = resolveCfnValue(arn, ctx); + return isString(resolved) && resolved !== '' ? resolved : undefined; +}; + +// queueNameFromArn(arn) -> string | undefined +// The queue name is the final ':'-delimited segment (robust to pseudo-params that inject extra ':'). +const queueNameFromArn = arn => { + if (!isString(arn) || arn === '') return undefined; + const segments = arn.split(':'); + return segments[segments.length - 1] || undefined; +}; + +// The eventbridge onFailure contract, preserved byte-for-byte (spec EARS8). +const buildFailurePayload = (requestPayload, err) => ({ + requestPayload, + responsePayload: {errorMessage: err && err.message, errorType: err && err.name} +}); + +const buildSuccessPayload = (requestPayload, result) => ({ + requestPayload, + responsePayload: result +}); + +// dispatchDestination({client, target, ctx, payload, log}) -> Promise +// The single impure edge (still unit-tested with an injected fake client). Best-effort: +// - nil target -> no-op (no client call) +// - unresolvable ARN -> warn + skip +// - resolvable target -> GetQueueUrl + SendMessage +// ANY error is log.warning-ed and swallowed — destinations dispatch never throws or blocks the +// originating event source (spec EARS7, edge cases 2 & 3). +const dispatchDestination = async ({client, target, ctx = {}, payload, log}) => { + if (isNil(target)) return; + + const arn = resolveDestinationArn(target, ctx); + if (isNil(arn)) { + log.warning( + `destinations: cannot resolve target ARN offline, skipping: ${JSON.stringify(target)}` + ); + return; + } + + try { + const queueName = queueNameFromArn(arn); + const {QueueUrl} = await client.send(new GetQueueUrlCommand({QueueName: queueName})); + await client.send(new SendMessageCommand({QueueUrl, MessageBody: JSON.stringify(payload)})); + } catch (err) { + log.warning(err && err.stack ? err.stack : String(err)); + } +}; + +// runDestinations(args) -> Promise +// The orchestrator the event sources call. Gated behind simulateDestinations (default true). On an +// error it dispatches destinations.onFailure; otherwise it dispatches destinations.onSuccess. The SQS +// client is built lazily via makeClient() so NO client is constructed when there is nothing to send +// (spec EARS2, edge case 1). +const runDestinations = async ({ + simulateDestinations, + destinations, + ctx = {}, + makeClient, + log, + requestPayload, + error, + result +}) => { + if (simulateDestinations === false) return; + + const target = error + ? destinations && destinations.onFailure + : destinations && destinations.onSuccess; + if (isNil(target)) return; + + const payload = error + ? buildFailurePayload(requestPayload, error) + : buildSuccessPayload(requestPayload, result); + + await dispatchDestination({client: makeClient(), target, ctx, payload, log}); +}; + +module.exports = { + pseudoParams, + resolveQueueRefArn, + resolveCfnValue, + resolveDestinationArn, + queueNameFromArn, + buildFailurePayload, + buildSuccessPayload, + dispatchDestination, + runDestinations +}; diff --git a/packages/serverless-offline-eventbridge/src/eventbridge.js b/packages/serverless-offline-eventbridge/src/eventbridge.js index b21c79b..cd892b9 100644 --- a/packages/serverless-offline-eventbridge/src/eventbridge.js +++ b/packages/serverless-offline-eventbridge/src/eventbridge.js @@ -1,6 +1,6 @@ const http = require('http'); const {randomUUID} = require('crypto'); -const {SQSClient, GetQueueUrlCommand, SendMessageCommand} = require('@aws-sdk/client-sqs'); +const {SQSClient} = require('@aws-sdk/client-sqs'); const { castArray, @@ -16,6 +16,7 @@ const { const {normalizeLog} = require('./log'); const {buildClientConfig} = require('./client-config'); +const {pseudoParams, runDestinations} = require('./destinations'); const EventBridgeEventDefinition = require('./eventbridge-event-definition'); const EventBridgeEvent = require('./eventbridge-event'); const {buildEventBridgeEvent} = require('./eventbridge-event'); @@ -371,7 +372,8 @@ class EventBridge { try { const lambdaFunction = this.lambda.get(functionKey); lambdaFunction.setEvent(new EventBridgeEvent(event, this.options.region)); - await lambdaFunction.runHandler(); + const result = await lambdaFunction.runHandler(); + await this._onSuccess(destinations, event, result); } catch (err) { this.log.warning(err.stack); if (remainingAttempts > 0) { @@ -385,32 +387,49 @@ class EventBridge { return task(maximumRetryAttempts - 1); } + // Lazily build (and memoize) the SQS client used for destination dispatch, so no client is ever + // constructed when no function declares a destination / simulateDestinations is off. + _sqsClient() { + if (!this.sqsClient) this.sqsClient = new SQSClient(buildClientConfig(this.options)); + return this.sqsClient; + } + // onFailure (async-invoke DLQ): best-effort push of a failure record to the function's // `destinations.onFailure` SQS ARN. Gated behind `simulateDestinations` (default true), never - // throws. + // throws. Delegates to the shared destinations helper (spec B2). async _onFailure(destinations, event, err) { - if (this.options.simulateDestinations === false) return; - - const onFailure = getOr(undefined, ['onFailure'], destinations); - const arn = isPlainObject(onFailure) ? getOr(undefined, 'arn', onFailure) : onFailure; - if (isNil(arn)) return; - - try { - if (!this.sqsClient) this.sqsClient = new SQSClient(buildClientConfig(this.options)); - const queueName = String(arn).split(':').pop(); - const {QueueUrl} = await this.sqsClient.send(new GetQueueUrlCommand({QueueName: queueName})); - await this.sqsClient.send( - new SendMessageCommand({ - QueueUrl, - MessageBody: JSON.stringify({ - requestPayload: event, - responsePayload: {errorMessage: err.message, errorType: err.name} - }) - }) - ); - } catch (dlqErr) { - this.log.warning(dlqErr.stack); - } + await runDestinations({ + simulateDestinations: this.options.simulateDestinations, + destinations, + ctx: pseudoParams( + this.options.region, + this.options.accountId, + getOr({}, ['resources', 'Resources'], this.options) + ), + makeClient: () => this._sqsClient(), + log: this.log, + requestPayload: event, + error: err + }); + } + + // onSuccess (async-invoke destination): best-effort push of `{requestPayload, responsePayload}` to + // the function's `destinations.onSuccess` SQS ARN after a clean handler run. Same gate, same + // swallow-and-warn semantics as onFailure. + async _onSuccess(destinations, event, result) { + await runDestinations({ + simulateDestinations: this.options.simulateDestinations, + destinations, + ctx: pseudoParams( + this.options.region, + this.options.accountId, + getOr({}, ['resources', 'Resources'], this.options) + ), + makeClient: () => this._sqsClient(), + log: this.log, + requestPayload: event, + result + }); } // Opt-in LocalStack poll loop placeholder: when `subscribe` is set, teams running a real `events` diff --git a/packages/serverless-offline-eventbridge/test/index.js b/packages/serverless-offline-eventbridge/test/index.js index 519d0f1..39adb17 100644 --- a/packages/serverless-offline-eventbridge/test/index.js +++ b/packages/serverless-offline-eventbridge/test/index.js @@ -13,6 +13,15 @@ const { } = require('../src/eventbridge'); const EventBridgeEvent = require('../src/eventbridge-event'); const EventBridgeEventDefinition = require('../src/eventbridge-event-definition'); +const { + pseudoParams, + resolveDestinationArn, + queueNameFromArn, + buildFailurePayload, + buildSuccessPayload, + dispatchDestination, + runDestinations +} = require('../src/destinations'); const { buildClientConfig, buildCredentials, @@ -482,3 +491,287 @@ test('src/eventbridge.js builds the event with this.options.region, not a bare t t.true(source.includes('this.options.region')); t.false(/new EventBridgeEvent\(event, this\.region\b/.test(source)); }); + +// --------------------------------------------------------------------------- +// Group H — Lambda async destinations (onFailure / onSuccess) — spec B2 +// --------------------------------------------------------------------------- + +// A fake SQS client recording every command it is `.send()` a — no SDK transport, no docker. +const fakeSqsClient = (impl = {}) => { + const calls = []; + return { + calls, + send: command => { + const name = command.constructor.name; + calls.push({name, input: command.input}); + if (name === 'GetQueueUrlCommand') { + if (impl.getQueueUrlError) throw impl.getQueueUrlError; + return Promise.resolve({QueueUrl: impl.queueUrl || 'http://localhost:9324/000/q'}); + } + if (name === 'SendMessageCommand') { + if (impl.sendMessageError) throw impl.sendMessageError; + return Promise.resolve({MessageId: 'm-1'}); + } + return Promise.resolve({}); + } + }; +}; + +const collectLog = () => { + const warnings = []; + return {log: {...normalizeLog(), warning: msg => warnings.push(msg)}, warnings}; +}; + +const CTX = {'AWS::Region': 'eu-west-1', 'AWS::AccountId': '000000000000'}; + +// EARS8: the failure payload shape is preserved byte-for-byte. +test('buildFailurePayload returns the eventbridge {requestPayload, responsePayload} shape (EARS8)', t => { + const err = Object.assign(new Error('boom'), {name: 'BoomError'}); + t.deepEqual(buildFailurePayload({a: 1}, err), { + requestPayload: {a: 1}, + responsePayload: {errorMessage: 'boom', errorType: 'BoomError'} + }); +}); + +// EARS5: onSuccess payload carries the handler result. +test('buildSuccessPayload returns {requestPayload, responsePayload: result} (EARS5)', t => { + t.deepEqual(buildSuccessPayload({a: 1}, {ok: true}), { + requestPayload: {a: 1}, + responsePayload: {ok: true} + }); +}); + +// EARS6 / edge 2: resolve literal, {arn}, intrinsic, pseudo-param; undefined when unresolvable. +test('resolveDestinationArn resolves every supported target shape (EARS6)', t => { + t.is(resolveDestinationArn('arn:aws:sqs:eu-west-1:0:dlq', CTX), 'arn:aws:sqs:eu-west-1:0:dlq'); + t.is( + resolveDestinationArn({arn: 'arn:aws:sqs:eu-west-1:0:dlq'}, CTX), + 'arn:aws:sqs:eu-west-1:0:dlq' + ); + t.is(resolveDestinationArn({Ref: 'AWS::AccountId'}, CTX), '000000000000'); + t.is( + // eslint-disable-next-line no-template-curly-in-string -- deliberate CloudFormation Fn::Sub token + resolveDestinationArn({arn: {'Fn::Sub': 'arn:aws:sqs:${AWS::Region}:0:dlq'}}, CTX), + 'arn:aws:sqs:eu-west-1:0:dlq' + ); + t.is( + resolveDestinationArn({'Fn::Join': [':', ['arn', 'aws', 'sqs', 'eu-west-1', '0', 'dlq']]}, CTX), + 'arn:aws:sqs:eu-west-1:0:dlq' + ); + t.is( + resolveDestinationArn('arn:aws:sqs:#{AWS::Region}:#{AWS::AccountId}:dlq', CTX), + 'arn:aws:sqs:eu-west-1:000000000000:dlq' + ); +}); + +test('resolveDestinationArn returns undefined for an unresolvable intrinsic (edge 2)', t => { + t.is(resolveDestinationArn({arn: {'Fn::ImportValue': 'Exported'}}, CTX), undefined); + t.is(resolveDestinationArn({Ref: 'SomeStackResource'}, CTX), undefined); + t.is(resolveDestinationArn(undefined, CTX), undefined); +}); + +// EARS6: a Ref / Fn::GetAtt destination pointing at an AWS::SQS::Queue declared in the stack resolves +// to that queue's real ARN — through the SAME CFN resolution the event-source ARNs use. +const CTX_WITH_QUEUE = pseudoParams('eu-west-1', '000000000000', { + MyDlq: {Type: 'AWS::SQS::Queue', Properties: {QueueName: 'my-dlq'}}, + AutoNamedDlq: {Type: 'AWS::SQS::Queue'}, + NotAQueue: {Type: 'AWS::S3::Bucket'} +}); + +test('resolveDestinationArn resolves a {Ref: } destination to the queue ARN (EARS6)', t => { + t.is( + resolveDestinationArn({Ref: 'MyDlq'}, CTX_WITH_QUEUE), + 'arn:aws:sqs:eu-west-1:000000000000:my-dlq' + ); + t.is( + resolveDestinationArn({arn: {Ref: 'MyDlq'}}, CTX_WITH_QUEUE), + 'arn:aws:sqs:eu-west-1:000000000000:my-dlq' + ); + // A queue with no explicit QueueName falls back to its logical id (serverless' local auto-name). + t.is( + resolveDestinationArn({Ref: 'AutoNamedDlq'}, CTX_WITH_QUEUE), + 'arn:aws:sqs:eu-west-1:000000000000:AutoNamedDlq' + ); +}); + +test('resolveDestinationArn resolves a {Fn::GetAtt: [, Arn]} destination to the queue ARN (EARS6)', t => { + t.is( + resolveDestinationArn({'Fn::GetAtt': ['MyDlq', 'Arn']}, CTX_WITH_QUEUE), + 'arn:aws:sqs:eu-west-1:000000000000:my-dlq' + ); + t.is( + resolveDestinationArn({arn: {'Fn::GetAtt': ['AutoNamedDlq', 'Arn']}}, CTX_WITH_QUEUE), + 'arn:aws:sqs:eu-west-1:000000000000:AutoNamedDlq' + ); +}); + +test('resolveDestinationArn returns undefined for a Ref/Fn::GetAtt to a non-queue or absent resource (EARS6, edge 2)', t => { + t.is(resolveDestinationArn({Ref: 'NotAQueue'}, CTX_WITH_QUEUE), undefined); + t.is(resolveDestinationArn({Ref: 'Missing'}, CTX_WITH_QUEUE), undefined); + t.is(resolveDestinationArn({'Fn::GetAtt': ['NotAQueue', 'Arn']}, CTX_WITH_QUEUE), undefined); + // Only the Arn attribute is resolvable offline; QueueName/QueueUrl etc. are not. + t.is(resolveDestinationArn({'Fn::GetAtt': ['MyDlq', 'QueueName']}, CTX_WITH_QUEUE), undefined); +}); + +// EARS6 end-to-end: a {Fn::GetAtt} onFailure destination dispatches to the resolved queue. +test('runDestinations dispatches onFailure to a Fn::GetAtt-resolved queue (EARS6)', async t => { + const client = fakeSqsClient(); + const {log} = collectLog(); + await runDestinations({ + simulateDestinations: true, + destinations: {onFailure: {'Fn::GetAtt': ['MyDlq', 'Arn']}}, + ctx: CTX_WITH_QUEUE, + makeClient: () => client, + log, + requestPayload: {detail: 'x'}, + error: Object.assign(new Error('boom'), {name: 'BoomError'}) + }); + t.is(client.calls[0].input.QueueName, 'my-dlq'); +}); + +// EARS1: queue name = final ':'-segment. +test('queueNameFromArn returns the final segment, undefined for garbage (EARS1)', t => { + t.is(queueNameFromArn('arn:aws:sqs:eu-west-1:0:my-dlq'), 'my-dlq'); + t.is(queueNameFromArn(''), undefined); + t.is(queueNameFromArn(undefined), undefined); +}); + +// Edge 1: nil target is a no-op — no client call at all. +test('dispatchDestination is a no-op for a nil target (edge 1)', async t => { + const client = fakeSqsClient(); + const {log} = collectLog(); + await dispatchDestination({client, target: undefined, ctx: CTX, payload: {}, log}); + t.deepEqual(client.calls, []); +}); + +// Edge 2: unresolvable ARN warns and skips, never calls GetQueueUrl. +test('dispatchDestination warns and skips an unresolvable ARN (edge 2)', async t => { + const client = fakeSqsClient(); + const {log, warnings} = collectLog(); + await dispatchDestination({ + client, + target: {arn: {'Fn::ImportValue': 'X'}}, + ctx: CTX, + payload: {}, + log + }); + t.deepEqual(client.calls, []); + t.is(warnings.length, 1); +}); + +// EARS1: a resolvable target -> GetQueueUrl(queueName) then SendMessage(payload). +test('dispatchDestination sends the payload to the resolved queue (EARS1)', async t => { + const client = fakeSqsClient({queueUrl: 'http://localhost:9324/000/my-dlq'}); + const {log} = collectLog(); + await dispatchDestination({ + client, + target: 'arn:aws:sqs:eu-west-1:0:my-dlq', + ctx: CTX, + payload: {requestPayload: {a: 1}, responsePayload: {errorMessage: 'x', errorType: 'E'}}, + log + }); + t.deepEqual(client.calls[0], {name: 'GetQueueUrlCommand', input: {QueueName: 'my-dlq'}}); + t.is(client.calls[1].name, 'SendMessageCommand'); + t.is(client.calls[1].input.QueueUrl, 'http://localhost:9324/000/my-dlq'); + t.deepEqual(JSON.parse(client.calls[1].input.MessageBody), { + requestPayload: {a: 1}, + responsePayload: {errorMessage: 'x', errorType: 'E'} + }); +}); + +// EARS7 / edge 3: any client error is warning-logged and swallowed. +test('dispatchDestination swallows a GetQueueUrl failure and warns (EARS7, edge 3)', async t => { + const client = fakeSqsClient({getQueueUrlError: new Error('queue gone')}); + const {log, warnings} = collectLog(); + await t.notThrowsAsync( + dispatchDestination({client, target: 'arn:aws:sqs:eu-west-1:0:dlq', ctx: CTX, payload: {}, log}) + ); + t.is(warnings.length, 1); +}); + +test('dispatchDestination swallows a SendMessage failure and warns (EARS7)', async t => { + const client = fakeSqsClient({sendMessageError: new Error('send failed')}); + const {log, warnings} = collectLog(); + await t.notThrowsAsync( + dispatchDestination({client, target: 'arn:aws:sqs:eu-west-1:0:dlq', ctx: CTX, payload: {}, log}) + ); + t.is(warnings.length, 1); +}); + +// EARS2: simulateDestinations === false is a no-op AND never builds a client (lazy). +test('runDestinations is a no-op and never builds a client when simulateDestinations is false (EARS2)', async t => { + let built = false; + const {log} = collectLog(); + await runDestinations({ + simulateDestinations: false, + destinations: {onFailure: 'arn:aws:sqs:eu-west-1:0:dlq'}, + ctx: CTX, + makeClient: () => { + built = true; + return fakeSqsClient(); + }, + log, + requestPayload: {a: 1}, + error: new Error('boom') + }); + t.false(built); +}); + +// Edge 1: no matching destination -> never builds a client. +test('runDestinations never builds a client when there is no matching destination (edge 1)', async t => { + let built = false; + const {log} = collectLog(); + await runDestinations({ + simulateDestinations: true, + destinations: {onSuccess: 'arn:aws:sqs:eu-west-1:0:ok'}, + ctx: CTX, + makeClient: () => { + built = true; + return fakeSqsClient(); + }, + log, + requestPayload: {a: 1}, + error: new Error('boom') // error path -> looks for onFailure, which is absent + }); + t.false(built); +}); + +// EARS1/8: end-to-end onFailure through runDestinations dispatches the failure payload. +test('runDestinations dispatches onFailure with the failure payload (EARS1, EARS8)', async t => { + const client = fakeSqsClient(); + const {log} = collectLog(); + await runDestinations({ + simulateDestinations: true, + destinations: {onFailure: 'arn:aws:sqs:eu-west-1:0:my-dlq'}, + ctx: CTX, + makeClient: () => client, + log, + requestPayload: {detail: 'x'}, + error: Object.assign(new Error('boom'), {name: 'BoomError'}) + }); + t.is(client.calls[0].input.QueueName, 'my-dlq'); + t.deepEqual(JSON.parse(client.calls[1].input.MessageBody), { + requestPayload: {detail: 'x'}, + responsePayload: {errorMessage: 'boom', errorType: 'BoomError'} + }); +}); + +// EARS5: end-to-end onSuccess through runDestinations dispatches the success payload. +test('runDestinations dispatches onSuccess with the result payload (EARS5)', async t => { + const client = fakeSqsClient(); + const {log} = collectLog(); + await runDestinations({ + simulateDestinations: true, + destinations: {onSuccess: 'arn:aws:sqs:eu-west-1:0:ok-queue'}, + ctx: CTX, + makeClient: () => client, + log, + requestPayload: {detail: 'x'}, + result: {statusCode: 200} + }); + t.is(client.calls[0].input.QueueName, 'ok-queue'); + t.deepEqual(JSON.parse(client.calls[1].input.MessageBody), { + requestPayload: {detail: 'x'}, + responsePayload: {statusCode: 200} + }); +}); diff --git a/packages/serverless-offline-sqs/src/destinations.js b/packages/serverless-offline-sqs/src/destinations.js new file mode 100644 index 0000000..98ac992 --- /dev/null +++ b/packages/serverless-offline-sqs/src/destinations.js @@ -0,0 +1,164 @@ +const {GetQueueUrlCommand, SendMessageCommand} = require('@aws-sdk/client-sqs'); +const {get, has, isNil, isPlainObject, isString} = require('lodash/fp'); + +// --------------------------------------------------------------------------- +// Lambda async-invoke destinations (onFailure / onSuccess) — pure helpers + a +// single injected-client dispatcher. Extracted from EventBridge._onFailure so the +// same destinations contract can be reused by serverless-offline-sqs and +// serverless-offline-dynamodb-streams (spec B2). SQS-only targets, by design. +// --------------------------------------------------------------------------- + +// CloudFormation pseudo-parameters we can resolve offline from the plugin's region/accountId. +// `resources` (the resolved CFN Resources map) is optional and only used to resolve a Ref / Fn::GetAtt +// that points at an AWS::SQS::Queue declared in the stack (spec EARS6). +const pseudoParams = (region, accountId, resources) => ({ + 'AWS::Region': region, + 'AWS::AccountId': accountId, + region, + accountId, + resources +}); + +// resolveQueueRefArn(resources, refName, region, accountId) -> string | undefined +// Mirrors serverless-offline-sqs' resolveSqsRefArn: a Ref / Fn::GetAtt targeting an AWS::SQS::Queue +// declared in resources.Resources resolves to a stable ARN whose last segment is the queue name +// (preferring the declared Properties.QueueName, falling back to the logical id serverless uses for an +// auto-named queue locally). Returns undefined for anything that is not an SQS queue. Pure, non-throwing. +const resolveQueueRefArn = (resources, refName, region, accountId) => { + if (get([refName, 'Type'], resources) !== 'AWS::SQS::Queue') return undefined; + const queueName = get([refName, 'Properties', 'QueueName'], resources) || refName; + return `arn:aws:sqs:${region}:${accountId}:${queueName}`; +}; + +// resolveCfnValue(value, ctx) -> string | undefined +// Mirrors serverless-offline-sqs' resolver: flattens the intrinsic / pseudo-parameter forms a +// destination ARN can take into a plain string. Resolves a Ref / Fn::GetAtt to an AWS::SQS::Queue +// declared in `ctx.resources` to that queue's ARN (spec EARS6). Returns undefined for anything +// unresolvable offline (Fn::ImportValue, a Ref/Fn::GetAtt to a non-queue or absent resource, a +// non-string) so the caller can warn + skip. +const resolveCfnValue = (value, ctx) => { + if (isString(value)) + return value.replace(/[#$]\{(AWS::[A-Za-z]+)\}/g, (match, key) => + isNil(ctx[key]) ? match : ctx[key] + ); + + if (isPlainObject(value)) { + if (has('Ref', value)) { + // A Ref to a pseudo-parameter (AWS::Region/AWS::AccountId) resolves to that value; otherwise try + // resolving it against the stack Resources as an AWS::SQS::Queue (spec EARS6). + if (!isNil(ctx[value.Ref])) return ctx[value.Ref]; + return resolveQueueRefArn(ctx.resources, value.Ref, ctx.region, ctx.accountId); + } + if (has('Fn::GetAtt', value)) { + // CloudFormation `{Fn::GetAtt: [, 'Arn']}` -> the queue's ARN (spec EARS6). + const [resourceName, attribute] = value['Fn::GetAtt']; + if (attribute !== 'Arn') return undefined; + return resolveQueueRefArn(ctx.resources, resourceName, ctx.region, ctx.accountId); + } + if (has('Fn::Sub', value)) return resolveCfnValue(value['Fn::Sub'], ctx); + if (has('Fn::Join', value)) { + const [separator, parts] = value['Fn::Join']; + return (parts || []).map(part => resolveCfnValue(part, ctx)).join(separator); + } + // Fn::ImportValue (and any other intrinsic) is not resolvable offline. + } + + return undefined; +}; + +// resolveDestinationArn(target, ctx) -> string | undefined +// A destination target is a literal string ARN, a {arn: }, or a bare intrinsic. +// Resolves any of these to a string ARN through resolveCfnValue, or undefined when unresolvable. +const resolveDestinationArn = (target, ctx = {}) => { + if (isNil(target)) return undefined; + const arn = isPlainObject(target) && has('arn', target) ? target.arn : target; + const resolved = resolveCfnValue(arn, ctx); + return isString(resolved) && resolved !== '' ? resolved : undefined; +}; + +// queueNameFromArn(arn) -> string | undefined +// The queue name is the final ':'-delimited segment (robust to pseudo-params that inject extra ':'). +const queueNameFromArn = arn => { + if (!isString(arn) || arn === '') return undefined; + const segments = arn.split(':'); + return segments[segments.length - 1] || undefined; +}; + +// The eventbridge onFailure contract, preserved byte-for-byte (spec EARS8). +const buildFailurePayload = (requestPayload, err) => ({ + requestPayload, + responsePayload: {errorMessage: err && err.message, errorType: err && err.name} +}); + +const buildSuccessPayload = (requestPayload, result) => ({ + requestPayload, + responsePayload: result +}); + +// dispatchDestination({client, target, ctx, payload, log}) -> Promise +// The single impure edge (still unit-tested with an injected fake client). Best-effort: +// - nil target -> no-op (no client call) +// - unresolvable ARN -> warn + skip +// - resolvable target -> GetQueueUrl + SendMessage +// ANY error is log.warning-ed and swallowed — destinations dispatch never throws or blocks the +// originating event source (spec EARS7, edge cases 2 & 3). +const dispatchDestination = async ({client, target, ctx = {}, payload, log}) => { + if (isNil(target)) return; + + const arn = resolveDestinationArn(target, ctx); + if (isNil(arn)) { + log.warning( + `destinations: cannot resolve target ARN offline, skipping: ${JSON.stringify(target)}` + ); + return; + } + + try { + const queueName = queueNameFromArn(arn); + const {QueueUrl} = await client.send(new GetQueueUrlCommand({QueueName: queueName})); + await client.send(new SendMessageCommand({QueueUrl, MessageBody: JSON.stringify(payload)})); + } catch (err) { + log.warning(err && err.stack ? err.stack : String(err)); + } +}; + +// runDestinations(args) -> Promise +// The orchestrator the event sources call. Gated behind simulateDestinations (default true). On an +// error it dispatches destinations.onFailure; otherwise it dispatches destinations.onSuccess. The SQS +// client is built lazily via makeClient() so NO client is constructed when there is nothing to send +// (spec EARS2, edge case 1). +const runDestinations = async ({ + simulateDestinations, + destinations, + ctx = {}, + makeClient, + log, + requestPayload, + error, + result +}) => { + if (simulateDestinations === false) return; + + const target = error + ? destinations && destinations.onFailure + : destinations && destinations.onSuccess; + if (isNil(target)) return; + + const payload = error + ? buildFailurePayload(requestPayload, error) + : buildSuccessPayload(requestPayload, result); + + await dispatchDestination({client: makeClient(), target, ctx, payload, log}); +}; + +module.exports = { + pseudoParams, + resolveQueueRefArn, + resolveCfnValue, + resolveDestinationArn, + queueNameFromArn, + buildFailurePayload, + buildSuccessPayload, + dispatchDestination, + runDestinations +}; diff --git a/packages/serverless-offline-sqs/src/index.js b/packages/serverless-offline-sqs/src/index.js index 6663a4f..abc4e6a 100644 --- a/packages/serverless-offline-sqs/src/index.js +++ b/packages/serverless-offline-sqs/src/index.js @@ -215,6 +215,7 @@ class ServerlessOfflineSQS { sqsEvents.push({ functionKey, handler: functionDefinition.handler, + destinations: get('destinations', functionDefinition), sqs }); } diff --git a/packages/serverless-offline-sqs/src/sqs.js b/packages/serverless-offline-sqs/src/sqs.js index 5084cc1..e778633 100644 --- a/packages/serverless-offline-sqs/src/sqs.js +++ b/packages/serverless-offline-sqs/src/sqs.js @@ -43,6 +43,7 @@ const { const {default: PQueue} = require('p-queue'); const {normalizeLog} = require('./log'); const {buildClientConfig, ensureArray} = require('./client-config'); +const {pseudoParams, runDestinations} = require('./destinations'); const SQSEventDefinition = require('./sqs-event-definition'); const SQSEvent = require('./sqs-event'); @@ -345,8 +346,8 @@ class SQS { // comma-separated queueName, event-level or via the --queueName override) before creating; each // `def` is a fully-resolved single-queue definition. const definitions = flatMap( - ({functionKey, sqs}) => - expandSqsEventDefinitions(this.options, sqs).map(def => ({functionKey, def})), + ({functionKey, destinations, sqs}) => + expandSqsEventDefinitions(this.options, sqs).map(def => ({functionKey, destinations, def})), events ); @@ -380,7 +381,11 @@ class SQS { ); } - return Promise.all(definitions.map(({functionKey, def}) => this._create(functionKey, def))); + return Promise.all( + definitions.map(({functionKey, destinations, def}) => + this._create(functionKey, destinations, def) + ) + ); } start() { @@ -391,13 +396,13 @@ class SQS { this.queue.pause(); } - _create(functionKey, def) { + _create(functionKey, destinations, def) { // #262 (renanlido): `def` is already a fully-resolved single-queue definition produced by // expandSqsEventDefinitions (override-wins + arn-strip handled there), so SQSEventDefinition // only ever sees one queue at a time. const sqsEvent = new SQSEventDefinition(def, this.options.region, this.options.accountId); - return this._sqsEvent(functionKey, sqsEvent); + return this._sqsEvent(functionKey, destinations, sqsEvent); } _rewriteQueueUrl(queueUrl) { @@ -432,7 +437,7 @@ class SQS { } } - async _sqsEvent(functionKey, sqsEvent) { + async _sqsEvent(functionKey, destinations, sqsEvent) { const {enabled, arn, queueName, batchSize = 10, functionResponseType} = sqsEvent; if (!enabled) return; @@ -475,6 +480,7 @@ class SQS { // try/catch, so a transient SQS-client / network failure rejected the p-queue task and became // an unhandled rejection. Pull it inside: a receive failure is now logged via log.warning and // the loop re-schedules, exactly like a thrown handler — the offline session keeps polling. + let dispatchEvent; try { const messages = await getMessages(batchSize); @@ -482,6 +488,7 @@ class SQS { const lambdaFunction = this.lambda.get(functionKey); const event = new SQSEvent(messages, this.options.region, arn); + dispatchEvent = event; lambdaFunction.setEvent(event); // #221 (successkrisz): capture the handler result so ReportBatchItemFailures can keep the @@ -502,9 +509,16 @@ class SQS { ) ); } + + // B2 destinations: a clean handler run dispatches `destinations.onSuccess`. Best-effort — + // a dispatch failure never blocks the poll loop. + await this._dispatchDestination(destinations, event, {result}); } } catch (err) { this.log.warning(err.stack); + // B2 destinations: a poll iteration has no per-batch retry budget, so one throw IS exhaustion + // for that batch — dispatch `destinations.onFailure` with the SQS event payload + error. + await this._dispatchDestination(destinations, dispatchEvent, {error: err}); } // #226: re-enqueue through enqueueLoop so the floating task promise can never go unhandled. @@ -513,6 +527,25 @@ class SQS { enqueueLoop(this.queue, job, this.log); } + // B2 destinations: best-effort onFailure/onSuccess dispatch reusing this.client (the SQS client is + // already constructed for the plugin). No-op when there is no requestPayload (the receive itself + // failed before any batch was built) or no matching destination. Never throws. + async _dispatchDestination(destinations, requestPayload, {error, result} = {}) { + if (isNil(requestPayload)) return; + await runDestinations({ + simulateDestinations: this.options.simulateDestinations, + destinations, + // this.resources is the already-_resolveFn-flattened CFN Resources map; pass it so a + // Ref/Fn::GetAtt destination ARN resolves against the declared AWS::SQS::Queue (spec EARS6). + ctx: pseudoParams(this.options.region, this.options.accountId, this.resources || {}), + makeClient: () => this.client, + log: this.log, + requestPayload, + error, + result + }); + } + _getResourceProperties(queueName) { return pipe( values, diff --git a/packages/serverless-offline-sqs/test/index.js b/packages/serverless-offline-sqs/test/index.js index 476f793..61b3bf2 100644 --- a/packages/serverless-offline-sqs/test/index.js +++ b/packages/serverless-offline-sqs/test/index.js @@ -22,6 +22,14 @@ const { } = require('../src/sqs'); const SQSEvent = require('../src/sqs-event'); const SQSEventDefinition = require('../src/sqs-event-definition'); +const { + resolveDestinationArn, + queueNameFromArn, + buildFailurePayload, + buildSuccessPayload, + dispatchDestination, + runDestinations +} = require('../src/destinations'); const {defaultOptions, isPluginEnabled} = require('../src'); const {extractQueueNameFromARN, resolveCfnValue} = require('../src/sqs-event-definition'); const { @@ -787,7 +795,7 @@ const captureReceiveMessage = async (options, rawDefinition) => { }; const sqsEvent = new SQSEventDefinition(rawDefinition || {queueName: 'q'}, 'eu-west-1', '0'); - await sqs._sqsEvent('fn', sqsEvent); + await sqs._sqsEvent('fn', undefined, sqsEvent); sqs.queue.start(); // let the queued job drain await new Promise(resolve => { @@ -1237,7 +1245,7 @@ const runPollLoopWithReceive = async (receive, {stopAfter = 3} = {}) => { }; const sqsEvent = new SQSEventDefinition({queueName: 'q'}, 'eu-west-1', '0'); - await sqs._sqsEvent('fn', sqsEvent); + await sqs._sqsEvent('fn', undefined, sqsEvent); sqs.queue.start(); await drain(); sqs.queue.pause(); @@ -1396,3 +1404,173 @@ test('#189 _getQueueUrl returns the URL on the first candidate without trying th t.is(QueueUrl, 'http://local/QueueName.fifo'); t.deepEqual(seen, ['QueueName.fifo']); }); + +// --------------------------------------------------------------------------- +// Lambda async destinations (onFailure / onSuccess) — spec B2 +// --------------------------------------------------------------------------- + +const fakeSqsClient = (impl = {}) => { + const calls = []; + return { + calls, + send: command => { + const name = command.constructor.name; + calls.push({name, input: command.input}); + if (name === 'GetQueueUrlCommand') { + if (impl.getQueueUrlError) throw impl.getQueueUrlError; + return Promise.resolve({QueueUrl: impl.queueUrl || 'http://localhost:9324/000/q'}); + } + if (name === 'SendMessageCommand') { + if (impl.sendMessageError) throw impl.sendMessageError; + return Promise.resolve({MessageId: 'm-1'}); + } + return Promise.resolve({}); + } + }; +}; + +const CTX = {'AWS::Region': 'eu-west-1', 'AWS::AccountId': '000000000000'}; + +test('destinations: resolveDestinationArn handles string/{arn}/Ref/pseudo-param (EARS6)', t => { + t.is(resolveDestinationArn('arn:aws:sqs:eu-west-1:0:dlq', CTX), 'arn:aws:sqs:eu-west-1:0:dlq'); + t.is(resolveDestinationArn({arn: {Ref: 'AWS::Region'}}, CTX), 'eu-west-1'); + t.is(resolveDestinationArn({arn: {'Fn::ImportValue': 'X'}}, CTX), undefined); +}); + +test('destinations: queueNameFromArn returns the final segment (EARS1)', t => { + t.is(queueNameFromArn('arn:aws:sqs:eu-west-1:0:my-dlq'), 'my-dlq'); +}); + +test('destinations: buildFailurePayload preserves the eventbridge shape (EARS8)', t => { + t.deepEqual(buildFailurePayload({a: 1}, Object.assign(new Error('boom'), {name: 'E'})), { + requestPayload: {a: 1}, + responsePayload: {errorMessage: 'boom', errorType: 'E'} + }); +}); + +test('destinations: buildSuccessPayload carries the result (EARS5)', t => { + t.deepEqual(buildSuccessPayload({a: 1}, {ok: 1}), { + requestPayload: {a: 1}, + responsePayload: {ok: 1} + }); +}); + +test('destinations: dispatchDestination sends to the resolved queue (EARS1)', async t => { + const client = fakeSqsClient(); + await dispatchDestination({ + client, + target: 'arn:aws:sqs:eu-west-1:0:my-dlq', + ctx: CTX, + payload: {x: 1}, + log: normalizeLog() + }); + t.is(client.calls[0].input.QueueName, 'my-dlq'); + t.is(client.calls[1].name, 'SendMessageCommand'); +}); + +test('destinations: dispatchDestination swallows a client error and warns (EARS7)', async t => { + const client = fakeSqsClient({getQueueUrlError: new Error('gone')}); + const warnings = []; + await t.notThrowsAsync( + dispatchDestination({ + client, + target: 'arn:aws:sqs:eu-west-1:0:dlq', + ctx: CTX, + payload: {}, + log: normalizeLog({warning: m => warnings.push(m)}) + }) + ); + t.is(warnings.length, 1); +}); + +test('destinations: runDestinations is a no-op (no client) when simulateDestinations is false (EARS2)', async t => { + let built = false; + await runDestinations({ + simulateDestinations: false, + destinations: {onFailure: 'arn:aws:sqs:eu-west-1:0:dlq'}, + ctx: CTX, + makeClient: () => { + built = true; + return fakeSqsClient(); + }, + log: normalizeLog(), + requestPayload: {}, + error: new Error('boom') + }); + t.false(built); +}); + +// EARS3: a thrown handler in the poll job dispatches onFailure with the SQS event + error. +const buildSqsForDispatch = client => { + const sqs = Object.create(SQS.prototype); + sqs.client = client; + sqs.options = {region: 'eu-west-1', accountId: '000000000000', simulateDestinations: true}; + sqs.log = normalizeLog(); + return sqs; +}; + +test('SQS._dispatchDestination sends onFailure on a thrown handler (EARS3)', async t => { + const client = fakeSqsClient(); + const sqs = buildSqsForDispatch(client); + await sqs._dispatchDestination( + {onFailure: 'arn:aws:sqs:eu-west-1:0:my-dlq'}, + {Records: [{messageId: '1'}]}, + {error: Object.assign(new Error('boom'), {name: 'E'})} + ); + t.is(client.calls[0].input.QueueName, 'my-dlq'); + t.deepEqual(JSON.parse(client.calls[1].input.MessageBody), { + requestPayload: {Records: [{messageId: '1'}]}, + responsePayload: {errorMessage: 'boom', errorType: 'E'} + }); +}); + +test('SQS._dispatchDestination sends onSuccess on a clean result (EARS5)', async t => { + const client = fakeSqsClient(); + const sqs = buildSqsForDispatch(client); + await sqs._dispatchDestination( + {onSuccess: 'arn:aws:sqs:eu-west-1:0:ok'}, + {Records: [{messageId: '1'}]}, + {result: {ok: true}} + ); + t.is(client.calls[0].input.QueueName, 'ok'); + t.deepEqual(JSON.parse(client.calls[1].input.MessageBody), { + requestPayload: {Records: [{messageId: '1'}]}, + responsePayload: {ok: true} + }); +}); + +test('SQS._dispatchDestination is a no-op without a requestPayload (receive failed before a batch)', async t => { + const client = fakeSqsClient(); + const sqs = buildSqsForDispatch(client); + await sqs._dispatchDestination({onFailure: 'arn:aws:sqs:eu-west-1:0:dlq'}, undefined, { + error: new Error('receive failed') + }); + t.deepEqual(client.calls, []); +}); + +// EARS6: a Fn::GetAtt / Ref destination ARN pointing at a Queue declared in this.resources resolves +// to that queue (the Resources map SQS holds is the already-_resolveFn-flattened one). +test('SQS._dispatchDestination resolves a Fn::GetAtt destination via this.resources (EARS6)', async t => { + const client = fakeSqsClient(); + const sqs = buildSqsForDispatch(client); + sqs.resources = {MyDlq: {Type: 'AWS::SQS::Queue', Properties: {QueueName: 'my-dlq'}}}; + await sqs._dispatchDestination( + {onFailure: {'Fn::GetAtt': ['MyDlq', 'Arn']}}, + {Records: [{messageId: '1'}]}, + {error: Object.assign(new Error('boom'), {name: 'E'})} + ); + t.is(client.calls[0].input.QueueName, 'my-dlq'); +}); + +test('SQS._dispatchDestination resolves a {Ref: } destination via this.resources (EARS6)', async t => { + const client = fakeSqsClient(); + const sqs = buildSqsForDispatch(client); + sqs.resources = {OkQueue: {Type: 'AWS::SQS::Queue'}}; + await sqs._dispatchDestination( + {onSuccess: {Ref: 'OkQueue'}}, + {Records: [{messageId: '1'}]}, + {result: {ok: true}} + ); + // No explicit QueueName -> falls back to the logical id. + t.is(client.calls[0].input.QueueName, 'OkQueue'); +});