New analytics data beacon for new analytics ingestion service

src/build-options.js

@@ -10,6 +10,7 @@ import { assign } from './ponyfills/objects.js';
  * @property {string} environment
  * @property {number} [version]
  * @property {string} [endpoint]
+ * @property {string} [ingestionEndpoint]
  * @property {boolean} [autoConfirm]
  * @property {RequestHooks} [hooks]
  * @property {*} [store]
@@ -43,6 +44,9 @@ export function buildOptions(options) {
 
   opts.version = opts.version || 1;
   opts.endpoint = ((opts.endpoint && opts.endpoint.replace(/\/$/, '')) || 'https://participants.evolv.ai') + '/v' + opts.version;
+  if (opts.ingestionEndpoint) {
+    opts.ingestionEndpoint = opts.ingestionEndpoint.replace(/\/$/, '');
+  }
 
   opts.hooks = assign({}, opts.hooks);
 

src/index.js

@@ -54,6 +54,28 @@ function EvolvClient(opts) {
   };
 
   const contextBeacon = new Beacon(options.endpoint + '/' + options.environment + '/data', context, beaconOptions);
+  let ingestionBeacon = null;
+  if (options.ingestionEndpoint) {
+    try {
+      ingestionBeacon = new Beacon(options.ingestionEndpoint + '/' + options.environment + '/data', context, beaconOptions);
+    } catch (err) {
+      console.log('Evolv: Ingestion beacon initialization failed');
+      console.log(err);
+    }
+  }
+
+  function safeIngestionBeaconCall(action) {
+    if (!ingestionBeacon) {
+      return;
+    }
+
+    try {
+      action();
+    } catch (err) {
+      console.log('Evolv: Ingestion beacon error');
+      console.log(err);
+    }
+  }
 
   /**
    * The context against which the key predicates will be evaluated.
@@ -377,27 +399,39 @@ function EvolvClient(opts) {
     /*eslint no-unused-vars: ["error", { "argsIgnorePattern": "ctx" }]*/
     waitFor(context, CONTEXT_INITIALIZED, function (type, ctx) {
       contextBeacon.emit(type, context.remoteContext);
+      safeIngestionBeaconCall(function() {
+        ingestionBeacon.emit(type, context.remoteContext);
+      });
     });
     waitFor(context, CONTEXT_VALUE_ADDED, function (type, key, value, local) {
       if (local) {
         return;
       }
 
       contextBeacon.emit(type, {key: key, value: value});
+      safeIngestionBeaconCall(function() {
+        ingestionBeacon.emit(type, {key: key, value: value});
+      });
     });
     waitFor(context, CONTEXT_VALUE_CHANGED, function (type, key, value, before, local) {
       if (local) {
         return;
       }
 
       contextBeacon.emit(type, {key: key, value: value});
+      safeIngestionBeaconCall(function() {
+        ingestionBeacon.emit(type, {key: key, value: value});
+      });
     });
     waitFor(context, CONTEXT_VALUE_REMOVED, function (type, key, local) {
       if (local) {
         return;
       }
 
       contextBeacon.emit(type, {key: key});
+      safeIngestionBeaconCall(function() {
+        ingestionBeacon.emit(type, {key: key});
+      });
     });
 
     if (options.autoConfirm) {
@@ -414,6 +448,9 @@ function EvolvClient(opts) {
    */
   this.flush = function() {
     contextBeacon.flush();
+    safeIngestionBeaconCall(function() {
+      ingestionBeacon.flush();
+    });
   };
 
   /**
@@ -423,6 +460,9 @@ function EvolvClient(opts) {
    */
   this.allowEvents = function() {
     contextBeacon.unblockAndFlush();
+    safeIngestionBeaconCall(function() {
+      ingestionBeacon.unblockAndFlush();
+    });
   };
 
   /**

src/tests/index.test.js

@@ -738,6 +738,256 @@ describe('Evolv client integration tests', () => {
       expect(messages[16]).to.be.a.message("context.value.changed", "keys.active", []);
       expect(messages[17]).to.be.a.message("context.value.changed", "variants.active", []);
     });
+
+    it('should send beacon payloads to both existing and ingestion endpoints', async () => {
+      const uid = 123;
+      const environment = '579b106c73';
+      const endpoint = 'https://participants.evolv.ai/';
+      const ingestionEndpoint = 'https://analytics-data.evolvdev.com';
+      const version = 2;
+      const existingPayloads = [];
+      const ingestionPayloads = [];
+
+      xhrMock.get(`${endpoint}v${version}/${environment}/${uid}/configuration.json`, (req, res) => {
+        return res.status(200).body(JSON.stringify({
+          _published: 1584475383.3865728,
+          _client: {},
+          _experiments: []
+        }));
+      });
+
+      xhrMock.get(`${endpoint}v${version}/${environment}/${uid}/allocations`, (req, res) => {
+        return res.status(200).body(JSON.stringify([]));
+      });
+
+      xhrMock.post(`${endpoint}v${version}/${environment}/data`, (req, res) => {
+        existingPayloads.push(JSON.parse(req.body()));
+        return res.status(202);
+      });
+
+      xhrMock.post(`${ingestionEndpoint}/${environment}/data`, (req, res) => {
+        ingestionPayloads.push(JSON.parse(req.body()));
+        return res.status(202);
+      });
+
+      const client = new Evolv({
+        environment,
+        endpoint,
+        ingestionEndpoint,
+        version
+      });
+
+      client.initialize(uid, { remote: true }, { local: true });
+      client.flush();
+
+      await new Promise((resolve) => setTimeout(resolve, 5));
+
+      expect(existingPayloads.length).to.be.greaterThan(0);
+      expect(ingestionPayloads.length).to.equal(existingPayloads.length);
+      expect(ingestionPayloads[0]).to.eql(existingPayloads[0]);
+    });
+
+    it('should keep existing beacon path working when ingestion endpoint fails', async () => {
+      const uid = 123;
+      const environment = '579b106c73';
+      const endpoint = 'https://participants.evolv.ai/';
+      const ingestionEndpoint = 'https://analytics-data.evolvdev.com';
+      const version = 2;
+      const existingPayloads = [];
+      let ingestionAttempts = 0;
+
+      xhrMock.get(`${endpoint}v${version}/${environment}/${uid}/configuration.json`, (req, res) => {
+        return res.status(200).body(JSON.stringify({
+          _published: 1584475383.3865728,
+          _client: {},
+          _experiments: []
+        }));
+      });
+
+      xhrMock.get(`${endpoint}v${version}/${environment}/${uid}/allocations`, (req, res) => {
+        return res.status(200).body(JSON.stringify([]));
+      });
+
+      xhrMock.post(`${endpoint}v${version}/${environment}/data`, (req, res) => {
+        existingPayloads.push(JSON.parse(req.body()));
+        return res.status(202);
+      });
+
+      xhrMock.post(`${ingestionEndpoint}/${environment}/data`, (req, res) => {
+        ingestionAttempts += 1;
+        return res.status(500).body('boom');
+      });
+
+      const client = new Evolv({
+        environment,
+        endpoint,
+        ingestionEndpoint,
+        version
+      });
+
+      client.initialize(uid, { remote: true }, { local: true });
+      client.flush();
+
+      await new Promise((resolve) => setTimeout(resolve, 5));
+
+      expect(existingPayloads.length).to.be.greaterThan(0);
+      expect(ingestionAttempts).to.be.greaterThan(0);
+    });
+
+    it('should not send to ingestion endpoint when ingestionEndpoint is not configured', async () => {
+      const uid = 123;
+      const environment = '579b106c73';
+      const endpoint = 'https://participants.evolv.ai/';
+      const ingestionEndpoint = 'https://analytics-data.evolvdev.com';
+      const version = 2;
+      const existingPayloads = [];
+      const ingestionPayloads = [];
+
+      xhrMock.get(`${endpoint}v${version}/${environment}/${uid}/configuration.json`, (req, res) => {
+        return res.status(200).body(JSON.stringify({
+          _published: 1584475383.3865728,
+          _client: {},
+          _experiments: []
+        }));
+      });
+
+      xhrMock.get(`${endpoint}v${version}/${environment}/${uid}/allocations`, (req, res) => {
+        return res.status(200).body(JSON.stringify([]));
+      });
+
+      xhrMock.post(`${endpoint}v${version}/${environment}/data`, (req, res) => {
+        existingPayloads.push(JSON.parse(req.body()));
+        return res.status(202);
+      });
+
+      xhrMock.post(`${ingestionEndpoint}/${environment}/data`, (req, res) => {
+        ingestionPayloads.push(JSON.parse(req.body()));
+        return res.status(202);
+      });
+
+      const client = new Evolv({
+        environment,
+        endpoint,
+        version
+      });
+
+      client.initialize(uid, { remote: true }, { local: true });
+      client.flush();
+
+      await new Promise((resolve) => setTimeout(resolve, 5));
+
+      expect(existingPayloads.length).to.be.greaterThan(0);
+      expect(ingestionPayloads.length).to.equal(0);
+    });
+
+    it('should block and unblock both beacons with bufferEvents', async () => {
+      const uid = 123;
+      const environment = '579b106c73';
+      const endpoint = 'https://participants.evolv.ai/';
+      const ingestionEndpoint = 'https://analytics-data.evolvdev.com';
+      const version = 2;
+      const existingPayloads = [];
+      const ingestionPayloads = [];
+
+      xhrMock.get(`${endpoint}v${version}/${environment}/${uid}/configuration.json`, (req, res) => {
+        return res.status(200).body(JSON.stringify({
+          _published: 1584475383.3865728,
+          _client: {},
+          _experiments: []
+        }));
+      });
+
+      xhrMock.get(`${endpoint}v${version}/${environment}/${uid}/allocations`, (req, res) => {
+        return res.status(200).body(JSON.stringify([]));
+      });
+
+      xhrMock.post(`${endpoint}v${version}/${environment}/data`, (req, res) => {
+        existingPayloads.push(JSON.parse(req.body()));
+        return res.status(202);
+      });
+
+      xhrMock.post(`${ingestionEndpoint}/${environment}/data`, (req, res) => {
+        ingestionPayloads.push(JSON.parse(req.body()));
+        return res.status(202);
+      });
+
+      const client = new Evolv({
+        environment,
+        endpoint,
+        ingestionEndpoint,
+        version,
+        bufferEvents: true
+      });
+
+      client.initialize(uid, { remote: true }, { local: true });
+      await new Promise((resolve) => setTimeout(resolve, 5));
+      expect(existingPayloads.length).to.equal(0);
+      expect(ingestionPayloads.length).to.equal(0);
+
+      client.allowEvents();
+      await new Promise((resolve) => setTimeout(resolve, 5));
+      expect(existingPayloads.length).to.be.greaterThan(0);
+      expect(ingestionPayloads.length).to.equal(existingPayloads.length);
+    });
+
+    it('should flush queued remote context updates to both endpoints', async () => {
+      const uid = 123;
+      const environment = '579b106c73';
+      const endpoint = 'https://participants.evolv.ai/';
+      const ingestionEndpoint = 'https://analytics-data.evolvdev.com';
+      const version = 2;
+      const existingPayloads = [];
+      const ingestionPayloads = [];
+
+      xhrMock.get(`${endpoint}v${version}/${environment}/${uid}/configuration.json`, (req, res) => {
+        return res.status(200).body(JSON.stringify({
+          _published: 1584475383.3865728,
+          _client: {},
+          _experiments: []
+        }));
+      });
+
+      xhrMock.get(`${endpoint}v${version}/${environment}/${uid}/allocations`, (req, res) => {
+        return res.status(200).body(JSON.stringify([]));
+      });
+
+      xhrMock.post(`${endpoint}v${version}/${environment}/data`, (req, res) => {
+        existingPayloads.push(JSON.parse(req.body()));
+        return res.status(202);
+      });
+
+      xhrMock.post(`${ingestionEndpoint}/${environment}/data`, (req, res) => {
+        ingestionPayloads.push(JSON.parse(req.body()));
+        return res.status(202);
+      });
+
+      const client = new Evolv({
+        environment,
+        endpoint,
+        ingestionEndpoint,
+        version
+      });
+
+      client.initialize(uid, { remote: true }, { local: true });
+      client.flush();
+      await new Promise((resolve) => setTimeout(resolve, 5));
+
+      existingPayloads.length = 0;
+      ingestionPayloads.length = 0;
+
+      client.context.set('new.remote.key', 'value');
+      client.flush();
+      await new Promise((resolve) => setTimeout(resolve, 5));
+
+      expect(existingPayloads.length).to.be.greaterThan(0);
+      expect(ingestionPayloads.length).to.equal(existingPayloads.length);
+      expect(existingPayloads[0].messages.some(function(msg) {
+        return msg.type === 'context.value.added' && msg.payload.key === 'new.remote.key';
+      })).to.be.true;
+      expect(ingestionPayloads[0].messages.some(function(msg) {
+        return msg.type === 'context.value.added' && msg.payload.key === 'new.remote.key';
+      })).to.be.true;
+    });
   });
 
   describe('prevent beacon', () => {

src/types.d.ts

@@ -32,6 +32,7 @@ declare interface SubscribablePromise<T = any> extends Promise<T> {
 export interface EvolvClientOptions {
   environment: string;
   endpoint?: string;
+  ingestionEndpoint?: string;
   lazyUid?: boolean;
   requireConsent?: boolean;
   js?: boolean;