feat(node): add exponential retry mechanism with idempotency headers

packages/node/package.json

@@ -43,19 +43,23 @@
   },
   "dependencies": {
     "@novu/shared": "^0.20.0-alpha.0",
+    "axios-retry": "^3.8.0",
     "handlebars": "^4.7.7",
     "lodash.get": "^4.4.2",
-    "lodash.merge": "^4.6.2"
+    "lodash.merge": "^4.6.2",
+    "uuid": "^9.0.1"
   },
   "devDependencies": {
     "@sendgrid/mail": "^7.4.6",
     "@types/jest": "29.5.2",
     "@types/lodash.get": "^4.4.6",
     "@types/lodash.merge": "^4.6.6",
     "@types/node": "^14.6.0",
+    "@types/uuid": "^8.3.4",
     "axios": "^1.4.0",
     "codecov": "^3.5.0",
     "jest": "^27.0.6",
+    "nock": "^13.1.3",
     "npm-run-all": "^4.1.5",
     "open-cli": "^6.0.1",
     "rimraf": "^3.0.2",
@@ -77,7 +81,7 @@
     "preset": "ts-jest",
     "testEnvironment": "node",
     "moduleNameMapper": {
-      "axios": "axios/dist/node/axios.cjs"
+      "^axios$": "axios/dist/node/axios.cjs"
     }
   },
   "prettier": {

packages/node/src/index.ts

@@ -30,3 +30,4 @@ export * from './lib/feeds/feeds.interface';
 export * from './lib/topics/topic.interface';
 export * from './lib/integrations/integrations.interface';
 export * from './lib/messages/messages.interface';
+export { defaultRetryCondition } from './lib/retries';

packages/node/src/lib/novu.interface.ts

@@ -1,7 +1,16 @@
-import { AxiosInstance } from 'axios';
+import { AxiosError, AxiosInstance } from 'axios';
+
+export interface IRetryConfig {
+  initialDelay?: number;
+  waitMin?: number;
+  waitMax?: number;
+  retryMax?: number;
+  retryCondition?: (err: AxiosError) => boolean;
+}
 
 export interface INovuConfiguration {
   backendUrl?: string;
+  retryConfig?: IRetryConfig;
 }
 
 export class WithHttp {

packages/node/src/lib/novu.ts

@@ -13,6 +13,7 @@ import { Topics } from './topics/topics';
 import { Integrations } from './integrations/integrations';
 import { Messages } from './messages/messages';
 import { Tenants } from './tenants/tenants';
+import { makeRetriable } from './retries';
 
 export class Novu extends EventEmitter {
   private readonly apiKey?: string;
@@ -33,14 +34,19 @@ export class Novu extends EventEmitter {
   constructor(apiKey: string, config?: INovuConfiguration) {
     super();
     this.apiKey = apiKey;
-
-    this.http = axios.create({
+    const axiosInstance = axios.create({
       baseURL: this.buildBackendUrl(config),
       headers: {
         Authorization: `ApiKey ${this.apiKey}`,
       },
     });
 
+    if (config?.retryConfig) {
+      makeRetriable(axiosInstance, config);
+    }
+
+    this.http = axiosInstance;
+
     this.subscribers = new Subscribers(this.http);
     this.environments = new Environments(this.http);
     this.events = new Events(this.http);

packages/node/src/lib/retries.spec.ts

@@ -0,0 +1,144 @@
+import nock from 'nock';
+import { Novu } from '../index';
+
+const BACKEND_URL = 'http://example.com';
+const TOPICS_PATH = '/v1/topics';
+const TRIGGER_PATH = '/v1/events/trigger';
+
+jest.setTimeout(15000);
+
+const allEqual = (arr: Array<any>) => arr.every((val) => val === arr[0]);
+
+class NetworkError extends Error {
+  constructor(public code: string) {
+    super('Network Error');
+  }
+}
+
+describe('Novu Node.js package - Retries and idempotency key', () => {
+  afterEach(() => {
+    nock.cleanAll();
+    nock.enableNetConnect();
+  });
+
+  const novu = new Novu('fake-key', {
+    backendUrl: BACKEND_URL,
+    retryConfig: {
+      retryMax: 3,
+      waitMax: 1,
+      waitMin: 1,
+    },
+  });
+
+  it('should retry trigger', async () => {
+    // prepare backend api mock endpoints:
+    const idepotencyKeys: string[] = [];
+
+    nock(BACKEND_URL)
+      .post(TRIGGER_PATH)
+      .times(3)
+      .reply(function (_url, _body) {
+        idepotencyKeys.push(this.req.getHeader('idempotency-key') as string);
+
+        return [500, { message: 'Server Exception' }];
+      });
+
+    nock(BACKEND_URL)
+      .post(TRIGGER_PATH)
+      .reply(201, { acknowledged: true, transactionId: '1003' });
+
+    const result = await novu.trigger('fake-workflow', {
+      to: { subscriberId: '123' },
+      payload: {},
+    });
+
+    expect(allEqual(idepotencyKeys)).toBeTruthy();
+    expect(result.status).toEqual(201);
+    expect(result.request.headers['idempotency-key']).toBeDefined();
+  });
+
+  it('should retry and regenerate idempotency key on status 422', async () => {
+    const idepotencyKeys: string[] = [];
+
+    nock(BACKEND_URL)
+      .post(TRIGGER_PATH)
+      .times(3)
+      .reply(function (_url, _body) {
+        idepotencyKeys.push(this.req.getHeader('idempotency-key') as string);
+
+        return [422, { message: 'Unprocessable Content' }];
+      });
+
+    nock(BACKEND_URL)
+      .post(TRIGGER_PATH)
+      .reply(201, { acknowledged: true, transactionId: '1003' });
+
+    const result = await novu.trigger('fake-workflow', {
+      to: { subscriberId: '123' },
+      payload: {},
+    });
+
+    expect(allEqual(idepotencyKeys)).toBeFalsy();
+    expect(result.status).toEqual(201);
+    expect(result.request.headers['idempotency-key']).toBeDefined();
+  });
+
+  it('should retry getting topics list', async () => {
+    nock(BACKEND_URL)
+      .get(TOPICS_PATH)
+      .times(3)
+      .reply(500, { message: 'Server Exception' });
+
+    nock(BACKEND_URL).get(TOPICS_PATH).reply(200, [{}, {}]);
+
+    const result = await novu.topics.list({});
+
+    expect(result.status).toEqual(200);
+    expect(result.request.headers['idempotency-key']).toBeUndefined();
+  });
+
+  it('should retry on various errors until it reach successfull response', async () => {
+    nock(BACKEND_URL)
+      .get(TOPICS_PATH)
+      .reply(429, { message: 'Too many requests' });
+
+    nock(BACKEND_URL)
+      .get(TOPICS_PATH)
+      .reply(408, { message: 'Request Timeout' });
+
+    nock(BACKEND_URL)
+      .get(TOPICS_PATH)
+      .replyWithError(new NetworkError('ECONNRESET'));
+
+    nock(BACKEND_URL)
+      .get(TOPICS_PATH)
+      .replyWithError(new NetworkError('ETIMEDOUT'));
+
+    nock(BACKEND_URL)
+      .get(TOPICS_PATH)
+      .replyWithError(new NetworkError('ECONNREFUSED'));
+
+    nock(BACKEND_URL)
+      .get(TOPICS_PATH)
+      .reply(504, { message: 'Gateway timeout' });
+
+    nock(BACKEND_URL)
+      .get(TOPICS_PATH)
+      .reply(422, { message: 'Unprocessable Content' });
+
+    nock(BACKEND_URL).get(TOPICS_PATH).reply(200, [{}, {}]);
+
+    const novuClient = new Novu('fake-key', {
+      backendUrl: BACKEND_URL,
+      retryConfig: {
+        initialDelay: 0,
+        waitMin: 1,
+        waitMax: 1,
+        retryMax: 7,
+      },
+    });
+
+    const result = await novuClient.topics.list({});
+    expect(result.status).toEqual(200);
+  });
+});

packages/node/src/lib/retries.ts

@@ -0,0 +1,96 @@
+import { AxiosError, AxiosInstance } from 'axios';
+import axiosRetry, { isNetworkError } from 'axios-retry';
+import { v4 as uuid } from 'uuid';
+import { INovuConfiguration } from './novu.interface';
+
+const NON_IDEMPOTENT_METHODS = ['post', 'patch'];
+
+export function makeRetriable(
+  axios: AxiosInstance,
+  config?: INovuConfiguration
+) {
+  axios.interceptors.request.use((axiosConfig) => {
+    // don't attach idempotency key for idempotent methods
+    if (
+      axiosConfig.method &&
+      !NON_IDEMPOTENT_METHODS.includes(axiosConfig.method)
+    ) {
+      return axiosConfig;
+    }
+
+    const idempotencyKey = axiosConfig.headers['Idempotency-Key'];
+    // that means intercepted request is retried, so don't generate new idempotency key
+    if (idempotencyKey) {
+      return axiosConfig;
+    }
+
+    axiosConfig.headers['Idempotency-Key'] = uuid();
+
+    return axiosConfig;
+  });
+
+  const retryConfig = config?.retryConfig || {};
+  const retries = retryConfig.retryMax || 0;
+  const minDelay = retryConfig.waitMin || 1;
+  const maxDelay = retryConfig.waitMax || 30;
+  const initialDelay = retryConfig.initialDelay || minDelay;
+  const retryCondition = retryConfig.retryCondition || defaultRetryCondition;
+
+  function backoff(retryCount: number) {
+    if (retryCount === 1) {
+      return initialDelay;
+    }
+
+    const delay = retryCount * minDelay;
+    if (delay > maxDelay) {
+      return maxDelay;
+    }
+
+    return delay;
+  }
+
+  axiosRetry(axios, {
+    retries,
+    retryCondition,
+    retryDelay(retryCount) {
+      return backoff(retryCount) * 1000; // return delay in milliseconds
+    },
+    onRetry(_retryCount, error, requestConfig) {
+      if (
+        error.response?.status === 422 &&
+        requestConfig.headers &&
+        requestConfig.method &&
+        NON_IDEMPOTENT_METHODS.includes(requestConfig.method)
+      ) {
+        requestConfig.headers['Idempotency-Key'] = uuid();
+      }
+    },
+  });
+}
+
+const RETRIABLE_HTTP_CODES = [408, 429, 422];
+
+export function defaultRetryCondition(err: AxiosError): boolean {
+  // retry on TCP/IP error codes like ECONNRESET
+  if (isNetworkError(err)) {
+    return true;
+  }
+
+  if (err.code === 'ECONNABORTED') {
+    return false;
+  }
+
+  if (!err.response) {
+    return true;
+  }
+
+  if (err.response.status >= 500 && err.response.status <= 599) {
+    return true;
+  }
+
+  if (RETRIABLE_HTTP_CODES.includes(err.response.status)) {
+    return true;
+  }
+
+  return false;
+}