add dispatch unit tests and minor refactoring (#602)

* fix bug around invokeCallback

Co-authored-by: Seth Silesky <[email protected]>

.changeset/brown-islands-judge.md

@@ -0,0 +1,5 @@
+---
+'@segment/analytics-core': patch
+---
+
+Fix bug where delay and pTimeout are coupled

packages/core/src/analytics/__tests__/dispatch.test.ts

@@ -0,0 +1,125 @@
+/**
+ * Properly type mocked functions to make it easy to do assertions
+ * for example, myModule.mock.calls[0] will have the typed parameters instead of any.
+ *
+ * TODO: share with rest of project
+ */
+type JestMockedFn<Fn> = Fn extends (...args: infer Args) => infer ReturnT
+  ? jest.Mock<ReturnT, Args>
+  : never
+
+const isOnline = jest.fn().mockReturnValue(true)
+const isOffline = jest.fn().mockReturnValue(false)
+
+jest.mock('../../connection', () => ({
+  isOnline,
+  isOffline,
+}))
+
+const fetcher: JestMockedFn<typeof import('node-fetch')['default']> = jest.fn()
+jest.mock('node-fetch', () => fetcher)
+
+const invokeCallback: JestMockedFn<
+  typeof import('../../callback')['invokeCallback']
+> = jest.fn()
+jest.mock('../../callback', () => ({
+  invokeCallback: invokeCallback,
+}))
+
+import { EventQueue } from '../../queue/event-queue'
+import { Emitter } from '../../emitter'
+import { dispatch, getDelay } from '../dispatch'
+import { PriorityQueue } from '../../priority-queue'
+import { CoreContext } from '../../context'
+
+let emitter!: Emitter
+let queue!: EventQueue
+const dispatchSingleSpy = jest.spyOn(EventQueue.prototype, 'dispatchSingle')
+const dispatchSpy = jest.spyOn(EventQueue.prototype, 'dispatch')
+const screenCtxMatcher = expect.objectContaining<Partial<CoreContext>>({
+  event: { type: 'screen' },
+})
+describe('Dispatch', () => {
+  beforeEach(() => {
+    jest.resetAllMocks()
+    dispatchSingleSpy.mockImplementationOnce((ctx) => Promise.resolve(ctx))
+    invokeCallback.mockImplementationOnce((ctx) => Promise.resolve(ctx))
+    dispatchSpy.mockImplementationOnce((ctx) => Promise.resolve(ctx))
+    queue = new EventQueue(new PriorityQueue(4, []))
+    queue.isEmpty = jest.fn().mockReturnValue(false)
+    emitter = new Emitter()
+  })
+
+  it('should not dispatch if client is currently offline and retries are *disabled* for the main event queue', async () => {
+    isOnline.mockReturnValue(false)
+    isOffline.mockReturnValue(true)
+
+    const ctx = await dispatch({ type: 'screen' }, queue, emitter, {
+      retryQueue: false,
+    })
+    expect(ctx).toEqual(screenCtxMatcher)
+    const called = Boolean(
+      dispatchSingleSpy.mock.calls.length || dispatchSpy.mock.calls.length
+    )
+    expect(called).toBeFalsy()
+  })
+
+  it('should be allowed to dispatch if client is currently offline and retries are *enabled* for the main event queue', async () => {
+    isOnline.mockReturnValue(false)
+    isOffline.mockReturnValue(true)
+
+    await dispatch({ type: 'screen' }, queue, emitter, {
+      retryQueue: true,
+    })
+    const called = Boolean(
+      dispatchSingleSpy.mock.calls.length || dispatchSpy.mock.calls.length
+    )
+    expect(called).toBeTruthy()
+  })
+
+  it('should call dispatchSingle correctly if queue is empty', async () => {
+    queue.isEmpty = jest.fn().mockReturnValue(true)
+    await dispatch({ type: 'screen' }, queue, emitter)
+    expect(dispatchSingleSpy).toBeCalledWith(screenCtxMatcher)
+    expect(dispatchSpy).not.toBeCalled()
+  })
+
+  it('should call dispatch correctly if queue has items', async () => {
+    await dispatch({ type: 'screen' }, queue, emitter)
+    expect(dispatchSpy).toBeCalledWith(screenCtxMatcher)
+    expect(dispatchSingleSpy).not.toBeCalled()
+  })
+
+  it('should only call invokeCallback if callback is passed', async () => {
+    await dispatch({ type: 'screen' }, queue, emitter)
+    expect(invokeCallback).not.toBeCalled()
+
+    const cb = jest.fn()
+    await dispatch({ type: 'screen' }, queue, emitter, { callback: cb })
+    expect(invokeCallback).toBeCalledTimes(1)
+  })
+  it('should call invokeCallback with correct args', async () => {
+    const cb = jest.fn()
+    await dispatch({ type: 'screen' }, queue, emitter, {
+      callback: cb,
+    })
+    expect(dispatchSpy).toBeCalledWith(screenCtxMatcher)
+    expect(invokeCallback).toBeCalledTimes(1)
+    const [ctx, _cb] = invokeCallback.mock.calls[0]
+    expect(ctx).toEqual(screenCtxMatcher)
+    expect(_cb).toBe(cb)
+  })
+})
+
+describe(getDelay, () => {
+  it('should calculate the amount of time to delay before invoking the callback', () => {
+    const aShortTimeAgo = Date.now() - 200
+    const timeout = 5000
+    expect(Math.round(getDelay(aShortTimeAgo, timeout))).toBe(4800)
+  })
+
+  it('should have a sensible default', () => {
+    const aShortTimeAgo = Date.now() - 200
+    expect(Math.round(getDelay(aShortTimeAgo))).toBe(100)
+  })
+})

packages/core/src/analytics/dispatch.ts

@@ -12,6 +12,12 @@ type DispatchOptions = {
   retryQueue?: boolean
 }
 
+/* The amount of time in ms to wait before invoking the callback. */
+export const getDelay = (startTimeInEpochMS: number, timeoutInMS?: number) => {
+  const elapsedTime = Date.now() - startTimeInEpochMS
+  // increasing the timeout increases the delay by almost the same amount -- this is weird legacy behavior.
+  return Math.max((timeoutInMS ?? 300) - elapsedTime, 0)
+}
 /**
  * Push an event into the dispatch queue and invoke any callbacks.
  *
@@ -40,15 +46,12 @@ export async function dispatch(
   } else {
     dispatched = await queue.dispatch(ctx)
   }
-  const elapsedTime = Date.now() - startTime
-  const timeoutInMs = options?.timeout
 
   if (options?.callback) {
     dispatched = await invokeCallback(
       dispatched,
       options.callback,
-      Math.max((timeoutInMs ?? 300) - elapsedTime, 0),
-      timeoutInMs
+      getDelay(startTime, options.timeout)
     )
   }
   if (options?.debug) {

packages/core/src/callback/__tests__/index.test.ts

@@ -19,37 +19,37 @@ describe(invokeCallback, () => {
   })
 
   // Fixes GitHub issue: https://github.com/segmentio/analytics-next/issues/409
-  // A.JS classic waited for the timeout before invoking callback,
+  // A.JS classic waited for the timeout/delay before invoking callback,
   // so keep same behavior in A.JS next.
-  it('calls the callback after a timeout', async () => {
+  it('calls the callback after a delay', async () => {
     const ctx = new CoreContext({
       type: 'track',
     })
 
     const fn = jest.fn()
-    const timeout = 100
+    const delay = 100
 
     const startTime = Date.now()
-    const returned = await invokeCallback(ctx, fn, timeout)
+    const returned = await invokeCallback(ctx, fn, delay)
     const endTime = Date.now()
 
     expect(fn).toHaveBeenCalled()
-    expect(endTime - startTime).toBeGreaterThanOrEqual(timeout - 1)
+    expect(endTime - startTime).toBeGreaterThanOrEqual(delay - 1)
     expect(returned).toBe(ctx)
   })
 
-  it('ignores the callback after a timeout', async () => {
+  it('ignores the callback if it takes too long to resolve', async () => {
     const ctx = new CoreContext({
       type: 'track',
     })
 
     const slow = (_ctx: CoreContext): Promise<void> => {
       return new Promise((resolve) => {
-        setTimeout(resolve, 200)
+        setTimeout(resolve, 1100)
       })
     }
 
-    const returned = await invokeCallback(ctx, slow, 0, 50)
+    const returned = await invokeCallback(ctx, slow, 0)
     expect(returned).toBe(ctx)
 
     const logs = returned.logs()

packages/core/src/callback/index.ts

@@ -22,14 +22,14 @@ function sleep(timeoutInMs: number): Promise<void> {
 }
 
 /**
- * @param delayTimeout - The amount of time in ms to wait before invoking the callback.
- * @param timeout - The maximum amount of time in ms to allow the callback to run for.
+ * @param ctx
+ * @param callback - the function to invoke
+ * @param delay - aka "timeout". The amount of time in ms to wait before invoking the callback.
  */
 export function invokeCallback(
   ctx: CoreContext,
   callback: Callback,
-  delayTimeout: number,
-  timeout?: number
+  delay: number
 ): Promise<CoreContext> {
   const cb = () => {
     try {
@@ -40,9 +40,9 @@ export function invokeCallback(
   }
 
   return (
-    sleep(delayTimeout)
+    sleep(delay)
       // pTimeout ensures that the callback can't cause the context to hang
-      .then(() => pTimeout(cb(), timeout ?? 1000))
+      .then(() => pTimeout(cb(), 1000))
       .catch((err) => {
         ctx?.log('warn', 'Callback Error', { error: err })
         ctx?.stats?.increment('callback_error')