Merge pull request #4472 from mahendraHegde/idempotency-impl

feat: add idempotency interceptor
novuhq · Oct 15, 2023 · e4b3d8f · e4b3d8f
2 parents c7f68d3 + a45eb28
commit e4b3d8f
Show file tree

Hide file tree

Showing 12 changed files with 509 additions and 3 deletions.
diff --git a/.cspell.json b/.cspell.json
@@ -510,7 +510,11 @@
     "mediumdark",
     "Docgen",
     "clicksend",
-    "Clicksend"
+    "Clicksend",
+    "idempotency", 
+    "IDEMPOTENCY", 
+    "Idempotency",
+    "occured"
   ],
   "flagWords": [],
   "patterns": [

diff --git a/apps/api/e2e/idempotency.e2e.ts b/apps/api/e2e/idempotency.e2e.ts
@@ -0,0 +1,217 @@
+import { UserSession } from '@novu/testing';
+import { expect } from 'chai';
+describe('Idempotency Test', async () => {
+  let session: UserSession;
+  const path = '/v1/testing/idempotency';
+  const HEADER_KEYS = {
+    IDEMPOTENCY: 'idempotency-key',
+    RETRY_AFTER: 'retry-after',
+    IDEMPOTENCY_CONFLICT: 'x-idempotency-conflict',
+  };
+
+  describe('when enabled', () => {
+    before(async () => {
+      session = new UserSession();
+      await session.initialize();
+      process.env.API_IDEMPOTENCY_ENABLED = 'true';
+    });
+
+    it('should return cached same response for duplicate requests', async () => {
+      const key = Math.random().toString();
+      const { body, headers } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({ data: 201 })
+        .expect(201);
+      const { body: bodyDupe, headers: headerDupe } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({ data: 201 })
+        .expect(201);
+      expect(typeof body.data.number === 'number').to.be.true;
+      expect(body.data.number).to.equal(bodyDupe.data.number);
+      expect(headers[HEADER_KEYS.IDEMPOTENCY]).to.eq(key);
+      expect(headerDupe[HEADER_KEYS.IDEMPOTENCY]).to.eq(key);
+    });
+    it('should return conflict when concurrent requests are made', async () => {
+      const key = Math.random().toString();
+      const [{ headers, body, status }, { headers: headerDupe, body: bodyDupe, status: statusDupe }] =
+        await Promise.all([
+          session.testAgent.post(path).set(HEADER_KEYS.IDEMPOTENCY, key).send({ data: 250 }),
+          session.testAgent.post(path).set(HEADER_KEYS.IDEMPOTENCY, key).send({ data: 250 }),
+        ]);
+      const oneSuccess = status === 201 || statusDupe === 201;
+      const oneConflict = status === 429 || statusDupe === 429;
+      const conflictBody = status === 201 ? bodyDupe : body;
+      const retryHeader = headers[HEADER_KEYS.RETRY_AFTER] || headerDupe[HEADER_KEYS.RETRY_AFTER];
+      const conflictHeader = headers[HEADER_KEYS.IDEMPOTENCY_CONFLICT] || headerDupe[HEADER_KEYS.IDEMPOTENCY_CONFLICT];
+      expect(oneSuccess).to.be.true;
+      expect(oneConflict).to.be.true;
+      expect(headers[HEADER_KEYS.IDEMPOTENCY]).to.eq(key);
+      expect(headerDupe[HEADER_KEYS.IDEMPOTENCY]).to.eq(key);
+      expect(retryHeader).to.eq(`1`);
+      expect(conflictHeader).to.eq('IDEMPOTENCY_REQUEST_PROCESSING');
+      expect(JSON.stringify(conflictBody)).to.eq(
+        JSON.stringify({
+          error: 'IDEMPOTENCY_REQUEST_PROCESSING',
+          message: `request ${key} is currently being processed. Please retry after 1 second`,
+        })
+      );
+    });
+    it('should return conflict when different body is sent for same key', async () => {
+      const key = Math.random().toString();
+      const { headers, body, status } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({ data: 250 });
+      const {
+        headers: headerDupe,
+        body: bodyDupe,
+        status: statusDupe,
+      } = await session.testAgent.post(path).set(HEADER_KEYS.IDEMPOTENCY, key).send({ data: 251 });
+
+      const oneSuccess = status === 201 || statusDupe === 201;
+      const oneConflict = status === 409 || statusDupe === 409;
+      const conflictBody = status === 201 ? bodyDupe : body;
+      const conflictHeader = headers[HEADER_KEYS.IDEMPOTENCY_CONFLICT] || headerDupe[HEADER_KEYS.IDEMPOTENCY_CONFLICT];
+      expect(oneSuccess).to.be.true;
+      expect(oneConflict).to.be.true;
+      expect(headers[HEADER_KEYS.IDEMPOTENCY]).to.eq(key);
+      expect(headerDupe[HEADER_KEYS.IDEMPOTENCY]).to.eq(key);
+      expect(conflictHeader).to.eq('IDEMPOTENCY_BODY_CONFLICT');
+      expect(JSON.stringify(conflictBody)).to.eq(
+        JSON.stringify({
+          error: 'IDEMPOTENCY_BODY_CONFLICT',
+          message: `request ${key} is being reused for difefrent body`,
+        })
+      );
+    });
+    it('should return non cached response for unique requests', async () => {
+      const key = Math.random().toString();
+      const key1 = Math.random().toString();
+      const { body, headers } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({ data: 201 })
+        .expect(201);
+
+      const { body: bodyDupe, headers: headerDupe } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key1)
+        .send({ data: 201 })
+        .expect(201);
+      expect(typeof body.data.number === 'number').to.be.true;
+      expect(typeof bodyDupe.data.number === 'number').to.be.true;
+      expect(body.data.number).not.to.equal(bodyDupe.data.number);
+      expect(headers[HEADER_KEYS.IDEMPOTENCY]).to.eq(key);
+      expect(headerDupe[HEADER_KEYS.IDEMPOTENCY]).to.eq(key1);
+    });
+    it('should return non cached response for GET requests', async () => {
+      const key = Math.random().toString();
+      const { body, headers } = await session.testAgent
+        .get(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({})
+        .expect(200);
+
+      const { body: bodyDupe } = await session.testAgent
+        .get(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({})
+        .expect(200);
+      expect(typeof body.data.number === 'number').to.be.true;
+      expect(typeof bodyDupe.data.number === 'number').to.be.true;
+      expect(body.data.number).not.to.equal(bodyDupe.data.number);
+      expect(headers[HEADER_KEYS.IDEMPOTENCY]).to.eq(undefined);
+    });
+    it('should return cached error response for duplicate requests', async () => {
+      const key = Math.random().toString();
+      const { body, headers } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({ data: 422 })
+        .expect(422);
+
+      const { body: bodyDupe, headers: headerDupe } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({ data: 422 })
+        .expect(422);
+      expect(JSON.stringify(body)).to.equal(JSON.stringify(bodyDupe));
+
+      expect(headers[HEADER_KEYS.IDEMPOTENCY]).to.eq(key);
+      expect(headerDupe[HEADER_KEYS.IDEMPOTENCY]).to.eq(key);
+    });
+  });
+
+  describe('when disabled', () => {
+    before(async () => {
+      session = new UserSession();
+      await session.initialize();
+      process.env.API_IDEMPOTENCY_ENABLED = 'false';
+    });
+
+    it('should not return cached same response for duplicate requests', async () => {
+      const key = Math.random().toString();
+      const { body } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({ data: 201 })
+        .expect(201);
+
+      const { body: bodyDupe } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({ data: 201 })
+        .expect(201);
+      expect(typeof body.data.number === 'number').to.be.true;
+      expect(body.data.number).not.to.equal(bodyDupe.data.number);
+    });
+    it('should return non cached response for unique requests', async () => {
+      const key = Math.random().toString();
+      const key1 = Math.random().toString();
+      const { body } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({ data: 201 })
+        .expect(201);
+
+      const { body: bodyDupe } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key1)
+        .send({ data: 201 })
+        .expect(201);
+      expect(typeof body.data.number === 'number').to.be.true;
+      expect(typeof bodyDupe.data.number === 'number').to.be.true;
+      expect(body.data.number).not.to.equal(bodyDupe.data.number);
+    });
+    it('should return non cached response for GET requests', async () => {
+      const key = Math.random().toString();
+      const { body } = await session.testAgent.get(path).set(HEADER_KEYS.IDEMPOTENCY, key).send({}).expect(200);
+
+      const { body: bodyDupe } = await session.testAgent
+        .get(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({})
+        .expect(200);
+      expect(typeof body.data.number === 'number').to.be.true;
+      expect(typeof bodyDupe.data.number === 'number').to.be.true;
+      expect(body.data.number).not.to.equal(bodyDupe.data.number);
+    });
+    it('should not return cached error response for duplicate requests', async () => {
+      const key = Math.random().toString();
+      const { body } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({ data: 500 })
+        .expect(500);
+
+      const { body: bodyDupe } = await session.testAgent
+        .post(path)
+        .set(HEADER_KEYS.IDEMPOTENCY, key)
+        .send({ data: 500 })
+        .expect(500);
+      expect(JSON.stringify(body)).not.to.equal(JSON.stringify(bodyDupe));
+    });
+  });
+});
diff --git a/apps/api/src/.env.development b/apps/api/src/.env.development
@@ -64,3 +64,5 @@ NOVU_SMS_INTEGRATION_SENDER=
 INTERCOM_IDENTITY_VERIFICATION_SECRET_KEY=
 
 LAUNCH_DARKLY_SDK_KEY=
+
+API_IDEMPOTENCY_ENABLED=true
diff --git a/apps/api/src/.env.production b/apps/api/src/.env.production
@@ -53,3 +53,5 @@ NOVU_SMS_INTEGRATION_SENDER=
 INTERCOM_IDENTITY_VERIFICATION_SECRET_KEY=
 
 LAUNCH_DARKLY_SDK_KEY=
+
+API_IDEMPOTENCY_ENABLED=true
diff --git a/apps/api/src/.env.test b/apps/api/src/.env.test
@@ -90,3 +90,5 @@ MAX_NOVU_INTEGRATION_SMS_REQUESTS=20
 NOVU_SMS_INTEGRATION_ACCOUNT_SID=test
 NOVU_SMS_INTEGRATION_TOKEN=test
 NOVU_SMS_INTEGRATION_SENDER=1234567890
+
+API_IDEMPOTENCY_ENABLED=true
diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts
@@ -31,6 +31,7 @@ import { TopicsModule } from './app/topics/topics.module';
 import { InboundParseModule } from './app/inbound-parse/inbound-parse.module';
 import { BlueprintModule } from './app/blueprint/blueprint.module';
 import { TenantModule } from './app/tenant/tenant.module';
+import { IdempotencyInterceptor } from './app/shared/framework/idempotency.interceptor';
 
 const enterpriseImports = (): Array<Type | DynamicModule | Promise<DynamicModule> | ForwardReference> => {
   const modules: Array<Type | DynamicModule | Promise<DynamicModule> | ForwardReference> = [];
@@ -78,7 +79,12 @@ const enterpriseModules = enterpriseImports();
 
 const modules = baseModules.concat(enterpriseModules);
 
-const providers: Provider[] = [];
+const providers: Provider[] = [
+  {
+    provide: APP_INTERCEPTOR,
+    useClass: IdempotencyInterceptor,
+  },
+];
 
 if (process.env.SENTRY_DSN) {
   modules.push(RavenModule);
Original file line number	Diff line number	Diff line change
Expand Up		@@ -64,3 +64,5 @@ NOVU_SMS_INTEGRATION_SENDER=
		INTERCOM_IDENTITY_VERIFICATION_SECRET_KEY=

		LAUNCH_DARKLY_SDK_KEY=

		API_IDEMPOTENCY_ENABLED=true
Original file line number	Diff line number	Diff line change
Expand Up		@@ -53,3 +53,5 @@ NOVU_SMS_INTEGRATION_SENDER=
		INTERCOM_IDENTITY_VERIFICATION_SECRET_KEY=

		LAUNCH_DARKLY_SDK_KEY=

		API_IDEMPOTENCY_ENABLED=true