Adds batching support to AnalyticsNode (#640)

* add buffered segmentio node plugin

* adds batching node plugin

* integrate node batching plugin with AnalyticsNode

* update tests for new plugin

* move PublishingProps alongside Publisher and alias ConfigureNodePluginProps

* treat fetch !response.ok as error when setting delivery failure reason

* remove retry attempt count from ContextBatch

* remove unused import from http-integration.test.ts

* remove extract-promise-parts from analytics-next

* delete no longer needed file

* simplify errors

* add optional data object

* fix test

* change `it` tests back to `test`

Co-authored-by: Seth Silesky <5115498+silesky@users.noreply.github.com>

Author: chrisradek

packages/browser/src/browser/__tests__/analytics-pre-init.integration.test.ts

@@ -3,7 +3,7 @@ import unfetch from 'unfetch'
 import { Analytics } from '../../core/analytics'
 import { Context } from '../../core/context'
 import * as Factory from '../../test-helpers/factories'
-import { sleep } from '../../test-helpers/sleep'
+import { sleep } from '../../lib/sleep'
 import { setGlobalCDNUrl } from '../../lib/parse-cdn'
 import { User } from '../../core/user'
 

packages/browser/src/browser/__tests__/standalone-analytics.test.ts

@@ -5,7 +5,7 @@ import { snippet } from '../../tester/__fixtures__/segment-snippet'
 import { install, AnalyticsStandalone } from '../standalone-analytics'
 import unfetch from 'unfetch'
 import { PersistedPriorityQueue } from '../../lib/priority-queue/persisted'
-import { sleep } from '../../test-helpers/sleep'
+import { sleep } from '../../lib/sleep'
 import * as Factory from '../../test-helpers/factories'
 
 const track = jest.fn()

packages/browser/src/core/buffer/__tests__/index.test.ts

@@ -7,7 +7,7 @@ import {
 } from '..'
 import { Analytics } from '../../analytics'
 import { Context } from '../../context'
-import { sleep } from '@/test-helpers/sleep'
+import { sleep } from '@/lib/sleep'
 import { User } from '../../user'
 
 describe('PreInitMethodCallBuffer', () => {

…ckages/browser/src/test-helpers/sleep.ts packages/browser/src/lib/sleep.tspackages/browser/src/test-helpers/sleep.ts renamed to packages/browser/src/lib/sleep.ts packages/browser/src/test-helpers/sleep.ts renamed to packages/browser/src/lib/sleep.ts

@@ -11,6 +11,7 @@ export type EmittedDeliveryFailureError<Ctx extends CoreContext> = {
   code: 'delivery_failure'
   message: string
   ctx: Ctx
+  data?: any
 }
 
 /**

packages/core/src/emitter/interface.ts

@@ -6,6 +6,7 @@ export * from './events/interfaces'
 export * from './events'
 export * from './callback'
 export * from './priority-queue'
+export { backoff } from './priority-queue/backoff'
 export * from './context'
 export * from './queue/event-queue'
 export * from './analytics'

packages/core/src/index.ts

@@ -1,6 +1,6 @@
 import { createSuccess } from './test-helpers/factories'
 
-const fetcher = jest.fn().mockReturnValue(createSuccess())
+const fetcher = jest.fn()
 jest.mock('node-fetch', () => fetcher)
 
 import { AnalyticsNode, NodeSegmentEvent } from '../app/analytics-node'
@@ -19,8 +19,12 @@ describe('Ability for users to exit without losing events', () => {
   let ajs!: AnalyticsNode
   beforeEach(async () => {
     jest.resetAllMocks()
+    fetcher.mockReturnValue(createSuccess())
     ajs = new AnalyticsNode({
       writeKey: 'abc123',
+      batchSettings: {
+        maxEventsInBatch: 1,
+      },
     })
   })
   const _helpers = {

packages/node/src/__tests__/graceful-shutdown-integration.test.ts

@@ -3,7 +3,6 @@ jest.mock('node-fetch', () => fetcher)
 
 import { createSuccess } from './test-helpers/factories'
 import { AnalyticsNode } from '..'
-import { NodeSegmentEvent } from '../app/analytics-node'
 import { resolveCtx } from './test-helpers/resolve-ctx'
 
 const myDate = new Date('2016')
@@ -35,7 +34,7 @@ describe('Analytics Node', () => {
     const call1 = calls[0]
     const [url, httpRes] = call1
     expect(httpRes.method).toBe('POST')
-    expect(url).toBe('https://api.segment.io/v1/identify')
+    expect(url).toBe('https://api.segment.io/v1/batch')
     expect(httpRes.headers).toMatchInlineSnapshot(
       {
         Authorization: expect.stringContaining('Basic'),
@@ -49,39 +48,47 @@ describe('Analytics Node', () => {
       }
     `
     )
-    const body: NodeSegmentEvent = JSON.parse(httpRes.body)
+    const body = JSON.parse(httpRes.body)
     expect(body).toMatchInlineSnapshot(
       {
-        messageId: expect.any(String),
-        _metadata: {
-          nodeVersion: expect.any(String),
-        },
+        batch: [
+          {
+            messageId: expect.any(String),
+            _metadata: {
+              nodeVersion: expect.any(String),
+            },
 
-        context: {
-          library: {
-            version: expect.any(String),
+            context: {
+              library: {
+                version: expect.any(String),
+              },
+            },
           },
-        },
+        ],
       },
       `
       Object {
-        "_metadata": Object {
-          "nodeVersion": Any<String>,
-        },
-        "context": Object {
-          "library": Object {
-            "name": "analytics-node-next",
-            "version": Any<String>,
+        "batch": Array [
+          Object {
+            "_metadata": Object {
+              "nodeVersion": Any<String>,
+            },
+            "context": Object {
+              "library": Object {
+                "name": "AnalyticsNode",
+                "version": Any<String>,
+              },
+            },
+            "integrations": Object {},
+            "messageId": Any<String>,
+            "timestamp": "2016-01-01T00:00:00.000Z",
+            "traits": Object {
+              "foo": "bar",
+            },
+            "type": "identify",
+            "userId": "my_user_id",
           },
-        },
-        "integrations": Object {},
-        "messageId": Any<String>,
-        "timestamp": "2016-01-01T00:00:00.000Z",
-        "traits": Object {
-          "foo": "bar",
-        },
-        "type": "identify",
-        "userId": "my_user_id",
+        ],
       }
     `
     )
@@ -92,7 +99,7 @@ describe('Analytics Node', () => {
     ajs.track({ event: 'bar', userId: 'foo', properties: { foo: 'bar' } })
     await resolveCtx(ajs, 'track')
     expect(fetcher).toHaveBeenCalledWith(
-      'https://api.segment.io/v1/track',
+      'https://api.segment.io/v1/batch',
       expect.anything()
     )
   })
@@ -101,7 +108,7 @@ describe('Analytics Node', () => {
     ajs.page({ name: 'page', anonymousId: 'foo' })
     await resolveCtx(ajs, 'page')
     expect(fetcher).toHaveBeenCalledWith(
-      'https://api.segment.io/v1/page',
+      'https://api.segment.io/v1/batch',
       expect.anything()
     )
   })
@@ -110,7 +117,7 @@ describe('Analytics Node', () => {
     ajs.group({ groupId: 'group', anonymousId: 'foo' })
     await resolveCtx(ajs, 'group')
     expect(fetcher).toHaveBeenCalledWith(
-      'https://api.segment.io/v1/group',
+      'https://api.segment.io/v1/batch',
       expect.anything()
     )
   })
@@ -119,7 +126,7 @@ describe('Analytics Node', () => {
     ajs.alias({ userId: 'alias', previousId: 'previous' })
     await resolveCtx(ajs, 'alias')
     expect(fetcher).toHaveBeenCalledWith(
-      'https://api.segment.io/v1/alias',
+      'https://api.segment.io/v1/batch',
       expect.anything()
     )
   })
@@ -128,7 +135,7 @@ describe('Analytics Node', () => {
     ajs.screen({ name: 'screen', anonymousId: 'foo' })
     await resolveCtx(ajs, 'screen')
     expect(fetcher).toHaveBeenCalledWith(
-      'https://api.segment.io/v1/screen',
+      'https://api.segment.io/v1/batch',
       expect.anything()
     )
   })

packages/node/src/__tests__/http-integration.test.ts

@@ -8,6 +8,7 @@ import { testPlugin } from './test-helpers/test-plugin'
 import { createSuccess, createError } from './test-helpers/factories'
 
 const writeKey = 'foo'
+jest.setTimeout(10000)
 
 beforeEach(() => {
   jest.resetAllMocks()
@@ -22,7 +23,7 @@ describe('Initialization', () => {
     await analytics.ready
 
     const ajsNodeXt = analytics.queue.plugins.find(
-      (xt) => xt.name === 'analytics-node-next'
+      (xt) => xt.name === 'Segment.io'
     )
     expect(ajsNodeXt).toBeDefined()
     expect(ajsNodeXt?.isLoaded()).toBeTruthy()
@@ -46,25 +47,25 @@ describe('Error handling', () => {
     expect(() => analytics.track({} as any)).toThrowError(/event/i)
   })
 
-  test('http delivery errors are accessed through the emitter', (done) => {
+  it('should emit on an error', async () => {
     const analytics = new AnalyticsNode({
       writeKey,
+      batchSettings: {
+        maxAttempts: 1,
+      },
     })
     fetcher.mockReturnValue(
       createError({ statusText: 'Service Unavailable', status: 503 })
     )
-
-    analytics.track({ event: 'foo', userId: 'sup' })
-    analytics.on('error', (emittedErr) => {
-      if (emittedErr.code !== 'http_delivery') {
-        return done.fail('error code incorrect')
-      }
-      expect(emittedErr.message).toMatch(/segment/)
-      expect(emittedErr.code).toMatch(/http/)
-      expect(emittedErr.response.status).toBe(503)
-      done()
-    })
-    expect.assertions(3)
+    try {
+      const promise = resolveCtx(analytics, 'track')
+      analytics.track({ event: 'foo', userId: 'sup' })
+      await promise
+      throw new Error('fail')
+    } catch (err: any) {
+      expect(err.message).toMatch(/fail/)
+      expect(err.code).toMatch(/delivery_failure/)
+    }
   })
 })
 

packages/node/src/__tests__/integration.test.ts

@@ -17,10 +17,8 @@ import {
   pTimeout,
 } from '@segment/analytics-core'
 import { AnalyticsNodeSettings, validateSettings } from './settings'
-import { analyticsNode } from './plugin'
-
 import { version } from '../../package.json'
-import { NodeEmittedError } from './emitted-errors'
+import { configureNodePlugin } from '../plugins/segmentio'
 
 // create a derived class since we may want to add node specific things to Context later
 export class NodeContext extends CoreContext {}
@@ -41,7 +39,7 @@ export interface NodeSegmentEventOptions {
 /**
  * Map of emitter event names to method args.
  */
-type NodeEmitterEvents = CoreEmitterContract<NodeContext, NodeEmittedError> & {
+type NodeEmitterEvents = CoreEmitterContract<NodeContext> & {
   initialize: [AnalyticsNodeSettings]
   call_after_close: [NodeSegmentEvent] // any event that did not get dispatched due to close
   drained: []
@@ -85,9 +83,15 @@ export class AnalyticsNode
     validateSettings(settings)
     this._eventFactory = new EventFactory()
     this.queue = new EventQueue(new NodePriorityQueue(3))
+    const batchSettings = settings.batchSettings || {}
 
     this.ready = this.register(
-      analyticsNode({ writeKey: settings.writeKey }, this)
+      configureNodePlugin({
+        writeKey: settings.writeKey,
+        maxAttempts: batchSettings.maxAttempts ?? 4,
+        maxEventsInBatch: batchSettings.maxEventsInBatch ?? 15,
+        maxWaitTimeInMs: batchSettings.maxWaitTimeInMs ?? 1000,
+      })
     )
       .then(() => undefined)
       .catch((err) => {