feat: Initial comit

.gitignore

@@ -0,0 +1,10 @@
+.vscode/
+node_modules/
+dist/
+coverage/
+.DS_Store
+test.js
+.nyc_output
+public
+test/reports/*
+!test/reports/.keep

.mocharc.json

@@ -0,0 +1,6 @@
+{
+  "extensions": ["ts"],
+  "spec": "lib/**/*.spec.ts",
+  "require": ["ts-node/register"],
+  "node-option": ["loader=ts-node/esm"]
+}

.npmignore

@@ -0,0 +1,9 @@
+node_modules
+coverage
+*.log
+.vscode
+examples
+test
+public
+.nyc_output
+scripts

.nvmrc

@@ -0,0 +1 @@
+v18.20.3

.prettierignore

@@ -0,0 +1,5 @@
+node_modules
+coverage
+dist
+public
+test/reports

.prettierrc.json

@@ -0,0 +1,5 @@
+{
+  "singleQuote": true,
+  "arrowParens": "always",
+  "trailingComma": "none"
+}

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2024 Francisco Couceiro
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

examples/inbound-webhooks/cron.ts

@@ -0,0 +1,9 @@
+import cron from 'node-cron';
+
+import debouncer from './debouncedSqs.js';
+
+// Runs every hour at the start of the hour
+cron.schedule('0 * * * *', async () => {
+  // Push out all deduped messages received so far (see `./debounceWorker.js`)
+  debouncer.dispatchStoredMessages();
+});

examples/inbound-webhooks/debounceWorker.ts

@@ -0,0 +1,6 @@
+import debouncer from './debouncedSqs';
+
+// Keep this consumer active while you want to move messages
+// from your queue to the unique index. `./cron.js` takes care
+// of sending these messages to the debounced queue later on
+debouncer.createInputConsumer().start();

examples/inbound-webhooks/debouncedSqs.ts

@@ -0,0 +1,48 @@
+import { S3Client } from '@aws-sdk/client-s3';
+import { SQSClient } from '@aws-sdk/client-sqs';
+
+import { Debouncer } from '../../lib/deouncer.js';
+import { UniqueIndex } from '../../lib/unique-index/uniqueIndex.js';
+import { ConnectorS3 } from '../../lib/unique-index/connectors/s3.js';
+
+const s3Client = new S3Client({
+  region: 'your-region'
+});
+const S3_BUCKET_NAME = 'your-s3-bucket-name';
+
+const sqsClient = new SQSClient({
+  region: 'your-region'
+});
+export const SQS_QUEUE_URL = 'your-sqs-queue-url';
+export const SQS_DEBOUNCED_QUEUE_URL = 'your-sqs-debounced-queue-url';
+
+const debouncer = new Debouncer({
+  mode: 'withPayload',
+  index: new UniqueIndex({
+    name: 'inbound-webhooks',
+    connector: new ConnectorS3(s3Client, S3_BUCKET_NAME)
+  }),
+  inputQueueUrl: SQS_QUEUE_URL,
+  outputQueueUrl: SQS_DEBOUNCED_QUEUE_URL,
+  messageMapper: {
+    mapInputMessage: async ({ tenantId, webhookId, data }) => {
+      return {
+        groupId: tenantId,
+        entryId: webhookId,
+        payload: data
+      };
+    },
+    mapOutputMessages: async (groupId, entriesIds, payloadsByEntryId) => {
+      return entriesIds.map((entryId) => {
+        return {
+          tenantId: Number(groupId),
+          webhookId: Number(entryId),
+          data: payloadsByEntryId[entryId]
+        };
+      });
+    }
+  },
+  sqs: sqsClient
+});
+
+export default debouncer;

examples/inbound-webhooks/restApi.ts

@@ -0,0 +1,31 @@
+import express from 'express';
+import bodyParser from 'body-parser';
+import debouncer, { SQS_QUEUE_URL } from './debouncedSqs';
+
+const app = express();
+app.use(bodyParser.json());
+
+app.post('/:tenantId/webhook/:webhookId', async (req, res) => {
+  const { tenantId, webhookId } = req.params;
+  const data = req.body;
+
+  const message = {
+    tenantId,
+    webhookId,
+    data
+  };
+
+  // This message will be consumed by `debounceWorker.js` and written to the unique
+  // distributed index. Later on all duduped messages are delivered to SQS_DEBOUNCED_QUEUE_URL
+  // when `cron.js` triggers.
+  //
+  // Note: you can use your existing way of sending messages to SQS (SQS_QUEUE_URL)
+  await debouncer.enqueue([message], SQS_QUEUE_URL);
+
+  res.sendStatus(200);
+});
+
+const PORT = process.env.PORT || 3000;
+app.listen(PORT, () => {
+  console.log(`Server is running on port ${PORT}`);
+});

examples/inbound-webhooks/worker.ts

@@ -0,0 +1,17 @@
+import { Consumer } from 'sqs-consumer';
+import { SQS_DEBOUNCED_QUEUE_URL } from './debouncedSqs';
+
+const app = Consumer.create({
+  queueUrl: SQS_DEBOUNCED_QUEUE_URL,
+  handleMessage: async (message) => {
+    const { tenantId, webhookId, data } = JSON.parse(message.Body);
+
+    // Do something with this debounced data
+  }
+});
+
+app.on('error', (err) => {
+  console.error(err.message);
+});
+
+app.start();

lib/debouncer.integration.spec.ts

@@ -0,0 +1,123 @@
+import * as chai from 'chai';
+import chaiSorted from 'chai-sorted';
+
+import { Debouncer } from './deouncer.js';
+import { givenIOConsumers } from './test-utils/consumers.js';
+import { S3Mocks } from './test-utils/s3Mocks.js';
+import { SQSMocks } from './test-utils/sqsMocks.js';
+
+chai.use(chaiSorted);
+const { expect } = chai;
+
+describe('DebouncedSQS Integration Tests with SQS + S3', () => {
+  let sqsMocks: SQSMocks;
+  let s3Mocks: S3Mocks;
+
+  before(async () => {
+    sqsMocks = new SQSMocks('test-queue', 'test-debounced-queue');
+    await sqsMocks.init();
+
+    s3Mocks = new S3Mocks('test-bucket-debounced', 'test-index-debounced');
+    await s3Mocks.init();
+  });
+
+  after(async () => {
+    await sqsMocks.clear();
+    await s3Mocks.clear();
+  });
+
+  afterEach(async () => {
+    await s3Mocks.clearIndexFiles();
+  });
+
+  it('should debounce end-to-end - webhook example', async () => {
+    // Given
+    const debouncer = new Debouncer({
+      mode: 'withPayload',
+      index: s3Mocks.uniqueIndex,
+      inputQueueUrl: sqsMocks.sqsQueueUrl,
+      outputQueueUrl: sqsMocks.sqsDebouncedQueueUrl,
+      messageMapper: {
+        mapInputMessage: async ({ tenantId, webhookId, data }) => {
+          return {
+            groupId: tenantId,
+            entryId: webhookId,
+            payload: data
+          };
+        },
+        mapOutputMessages: async (groupId, entriesIds, payloadsByEntryId) => {
+          return entriesIds.map((entryId) => {
+            return {
+              tenantId: Number(groupId),
+              webhookId: Number(entryId),
+              data: payloadsByEntryId[entryId]
+            };
+          });
+        }
+      },
+      sqs: sqsMocks.sqsClient
+    });
+    const consumers = givenIOConsumers(debouncer);
+
+    // Given - simulate sending messages including duplicates
+    const message1 = {
+      tenantId: 100,
+      webhookId: 4001,
+      data: { external_id: 'example' }
+    };
+    const message1Duplicate = {
+      tenantId: 100,
+      webhookId: 4001,
+      data: { external_id: 'example' }
+    };
+    const message2 = {
+      tenantId: 100,
+      webhookId: 7654,
+      data: { external_id: 'example' }
+    };
+    const message3 = {
+      tenantId: 987,
+      webhookId: 2000,
+      data: { external_id: 'example' }
+    };
+    await debouncer.enqueue(
+      [message1, message1Duplicate, message2, message3],
+      debouncer.inputQueueUrl
+    );
+
+    // Wait for input messages to be processed
+    consumers.startInputConsumer();
+    const processedInputMessages = await Promise.all([
+      consumers.inputMessages.next().then(parseMessage),
+      consumers.inputMessages.next().then(parseMessage),
+      consumers.inputMessages.next().then(parseMessage),
+      consumers.inputMessages.next().then(parseMessage)
+    ]);
+    consumers.stopInputConsumer();
+
+    // When - Dispatching indexed messages
+    await debouncer.dispatchStoredMessages();
+    consumers.startOutputConsumer();
+    const processedOutputMessages = await Promise.all([
+      consumers.outputMessages.next().then(parseMessage),
+      consumers.outputMessages.next().then(parseMessage),
+      consumers.outputMessages.next().then(parseMessage)
+    ]);
+    consumers.stopOutputConsumer();
+
+    // Then
+    expect(processedInputMessages).to.have.deep.members([
+      message1,
+      message1Duplicate,
+      message2,
+      message3
+    ]);
+    expect(processedOutputMessages).to.have.deep.members([
+      message1,
+      message2,
+      message3
+    ]);
+  });
+});
+
+const parseMessage = ({ value }) => JSON.parse(value.Body);