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

packages/node/package.json

@@ -43,19 +43,23 @@
   },
   "dependencies": {
     "@novu/shared": "^0.21.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/retry';

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 { makeRetryable } from './retry';
 
 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) {
+      makeRetryable(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/retry.spec.ts

@@ -0,0 +1,205 @@
+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<string>) => 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 and generate idempotency-key only once for request', async () => {
+    const idempotencyKeys: string[] = [];
+
+    nock(BACKEND_URL)
+      .post(TRIGGER_PATH)
+      .times(3)
+      .reply(function (_url, _body) {
+        idempotencyKeys.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: {},
+    });
+
+    // all idempotency keys are supposed to be same.
+    expect(allEqual(idempotencyKeys)).toBeTruthy();
+    expect(result.status).toEqual(201);
+    expect(result.request.headers['idempotency-key']).toBeDefined();
+  });
+
+  it('should generate different idempotency-key for each request', async () => {
+    nock(BACKEND_URL)
+      .post(TRIGGER_PATH)
+      .reply(500, { message: 'Server Exception' });
+
+    nock(BACKEND_URL)
+      .post(TRIGGER_PATH)
+      .times(10)
+      .reply(201, { acknowledged: true, transactionId: '1003' });
+
+    const idempotencyKeys: string[] = [];
+
+    for (let i = 0; i < 10; i++) {
+      const result = await novu.trigger('fake-workflow', {
+        to: { subscriberId: '123' },
+        payload: {},
+      });
+
+      idempotencyKeys.push(result.request?.headers['idempotency-key']);
+    }
+
+    expect(allEqual(idempotencyKeys)).toEqual(false);
+  });
+
+  it('should retry on status 422 and regenerate idempotency-key for every retry', async () => {
+    const idempotencyKeys: string[] = [];
+
+    nock(BACKEND_URL)
+      .post(TRIGGER_PATH)
+      .times(3)
+      .reply(function (_url, _body) {
+        idempotencyKeys.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: {},
+    });
+
+    // idempotency key should be regenerated for every retry for http status 422.
+    expect(allEqual(idempotencyKeys)).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 fail after reaching max retries', async () => {
+    nock(BACKEND_URL)
+      .get(TOPICS_PATH)
+      .times(4)
+      .reply(500, { message: 'Server Exception' });
+
+    nock(BACKEND_URL).get(TOPICS_PATH).reply(200, [{}, {}]);
+
+    await expect(novu.topics.list({})).rejects.toMatchObject({
+      response: { status: 500 },
+    });
+  });
+
+  const NON_RECOVERABLE_ERRORS: Array<[number, string]> = [
+    [400, 'Bad Request'],
+    [401, 'Unauthorized'],
+    [403, 'Forbidden'],
+    [404, 'Not Found'],
+    [405, 'Method not allowed'],
+    [413, 'Payload Too Large'],
+    [414, 'URI Too Long'],
+    [415, 'Unsupported Media Type'],
+  ];
+
+  test.each<[number, string]>(NON_RECOVERABLE_ERRORS)(
+    'should not retry on non-recoverable %i error',
+    async (status, message) => {
+      nock(BACKEND_URL).get(TOPICS_PATH).times(3).reply(status, { message });
+      nock(BACKEND_URL).get(TOPICS_PATH).reply(200, [{}, {}]);
+
+      await expect(novu.topics.list({})).rejects.toMatchObject({
+        response: { status },
+      });
+    }
+  );
+
+  it('should retry on various errors until it reach successful 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/retry.ts

@@ -0,0 +1,94 @@
+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'];
+const IDEMPOTENCY_KEY = 'Idempotency-Key'; // header key
+
+const DEFAULT_RETRY_MAX = 0;
+const DEFAULT_WAIT_MIN = 1;
+const DEFAULT_WAIT_MAX = 30;
+
+export function makeRetryable(
+  axios: AxiosInstance,
+  config?: INovuConfiguration
+) {
+  axios.interceptors.request.use((axiosConfig) => {
+    if (
+      axiosConfig.method &&
+      NON_IDEMPOTENT_METHODS.includes(axiosConfig.method)
+    ) {
+      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 || DEFAULT_RETRY_MAX;
+  const minDelay = retryConfig.waitMin || DEFAULT_WAIT_MIN;
+  const maxDelay = retryConfig.waitMax || DEFAULT_WAIT_MAX;
+  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 RETRYABLE_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.response &&
+    err.response.status >= 500 &&
+    err.response.status <= 599
+  ) {
+    return true;
+  }
+
+  if (err.response && RETRYABLE_HTTP_CODES.includes(err.response.status)) {
+    return true;
+  }
+
+  return false;
+}