give rules control over throwing conditional check failures (#399)

Author: codyjmoody

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "aws-lambda-stream",
-  "version": "1.0.29",
+  "version": "1.0.30",
   "description": "Create stream processors with AWS Lambda functions.",
   "keywords": [
     "aws",

package.json

@@ -27,13 +27,15 @@ class Connector {
     removeUndefinedValues = true,
     timeout = Number(process.env.DYNAMODB_TIMEOUT) || Number(process.env.TIMEOUT) || 1000,
     retryConfig = defaultRetryConfig,
+    throwConditionFailure = false,
     additionalClientOpts = {},
     ...opt
   }) {
     this.debug = (msg) => debug('%j', msg);
     this.tableName = tableName || /* istanbul ignore next */ 'undefined';
     this.client = Connector.getClient(pipelineId, debug, convertEmptyValues, removeUndefinedValues, timeout, additionalClientOpts);
     this.retryConfig = retryConfig;
+    this.throwConditionFailure = throwConditionFailure;
     this.opt = opt;
   }
 
@@ -73,7 +75,7 @@ class Connector {
     return this._sendCommand(new UpdateCommand(params), ctx)
       .catch((err) => {
         /* istanbul ignore else */
-        if (err.name === 'ConditionalCheckFailedException') {
+        if (err.name === 'ConditionalCheckFailedException' && !this.throwConditionFailure) {
           return {};
         }
         /* istanbul ignore next */

src/connectors/dynamodb.js

@@ -59,11 +59,12 @@ export const updateDynamoDB = ({
   parallel = Number(process.env.UPDATE_PARALLEL) || Number(process.env.PARALLEL) || 4,
   timeout = Number(process.env.DYNAMODB_TIMEOUT) || Number(process.env.TIMEOUT) || 1000,
   removeUndefinedValues = true,
+  throwConditionFailure = false,
   step = 'save',
   ...opt
 } = {}) => {
   const connector = new Connector({
-    pipelineId, debug, tableName, timeout, removeUndefinedValues, ...opt,
+    pipelineId, debug, tableName, timeout, removeUndefinedValues, throwConditionFailure, ...opt,
   });
 
   const invoke = (uow) => {

src/sinks/dynamodb.js

@@ -3,6 +3,9 @@ import { expect } from 'chai';
 import sinon from 'sinon';
 
 import { KmsConnector, MOCK_GEN_DK_RESPONSE } from 'aws-kms-ee';
+import { DynamoDBDocumentClient, UpdateCommand } from '@aws-sdk/lib-dynamodb';
+import { ConditionalCheckFailedException } from '@aws-sdk/client-dynamodb';
+import { mockClient } from 'aws-sdk-client-mock';
 
 import {
   initialize, initializeFrom,
@@ -17,16 +20,21 @@ import {
 import {
   updateExpression, timestampCondition,
 } from '../../../src/sinks/dynamodb';
-import { DynamoDBConnector } from '../../../src/connectors';
+import { DynamoDBConnector, EventBridgeConnector } from '../../../src/connectors';
 
 import { update } from '../../../src/flavors/update';
 
 describe('flavors/update.js', () => {
+  let mockDdb;
+
   beforeEach(() => {
     sinon.stub(DynamoDBConnector.prototype, 'update').resolves({});
   });
 
-  afterEach(sinon.restore);
+  afterEach(() => {
+    sinon.restore();
+    mockDdb?.restore();
+  });
 
   it('should execute', (done) => {
     sinon.stub(DynamoDBConnector.prototype, 'query').resolves([]);
@@ -157,6 +165,54 @@ describe('flavors/update.js', () => {
       })
       .done(done);
   });
+
+  it('should optionally throw conditional check', (done) => {
+    sinon.restore();
+    sinon.stub(EventBridgeConnector.prototype, 'putEvents').resolves({});
+    mockDdb = mockClient(DynamoDBDocumentClient);
+    mockDdb.on(UpdateCommand).rejects(new ConditionalCheckFailedException({}));
+
+    const events = toDynamodbRecords([
+      {
+        timestamp: 1572832690,
+        keys: {
+          pk: '1',
+          sk: 'thing',
+        },
+        newImage: {
+          pk: '1',
+          sk: 'thing',
+          discriminator: 'thing',
+          ttl: 1549053422,
+          timestamp: 1548967022000,
+        },
+      },
+    ]);
+
+    const rule = {
+      id: 'updateThrow',
+      flavor: update,
+      eventType: /thing-*/,
+      toUpdateRequest: () => ({}),
+      throwConditionFailure: true,
+    };
+
+    initialize({
+      ...initializeFrom([
+        rule,
+      ]),
+    }, { ...defaultOptions, AES: false })
+      .assemble(fromDynamodb(events), true)
+      .collect()
+      // .tap((collected) => console.log(JSON.stringify(collected, null, 2)))
+      .tap((collected) => {
+        expect(collected.length).to.equal(1);
+        expect(collected[0].event.tags.pipeline).to.equal('updateThrow');
+        expect(collected[0].event.type).to.equal('fault');
+        expect(collected[0].event.err.name).to.equal('ConditionalCheckFailedException');
+      })
+      .done(done);
+  });
 });
 
 const toUpdateRequest = (uow) => ({