A few fixes and tests

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

@@ -1,9 +1,11 @@
+import { sleep } from '@segment/analytics-core'
 import { HTTPResponse } from '../lib/http-client'
 import { TokenManager, TokenManagerProps } from '../lib/token-manager'
 import {
   TestFetchClient,
   createTestAnalytics,
 } from './test-helpers/create-test-analytics'
+import { createError } from './test-helpers/factories'
 import { resolveCtx } from './test-helpers/resolve-ctx'
 
 const privateKey = Buffer.from(`-----BEGIN PRIVATE KEY-----
@@ -41,6 +43,9 @@ const timestamp = new Date()
 const oauthTestClient = new TestFetchClient()
 const oauthFetcher = jest.spyOn(oauthTestClient, 'makeRequest')
 
+const tapiTestClient = new TestFetchClient()
+const tapiFetcher = jest.spyOn(tapiTestClient, 'makeRequest')
+
 const getTokenManagerProps = () => {
   const tokenManagerProps = {
     httpClient: oauthTestClient,
@@ -106,9 +111,7 @@ describe('OAuth Success', () => {
 
     await analytics.closeAndFlush()
   })
-})
-describe('OAuth Retry Success', () => {
-  it('track event with OAuth', async () => {
+  it('track event with OAuth after retry', async () => {
     const analytics = createTestAnalytics({
       tokenManagerProps: getTokenManagerProps(),
     })
@@ -143,35 +146,180 @@ describe('OAuth Retry Success', () => {
 
     await analytics.closeAndFlush()
   })
-})
 
-describe('OAuth Failure', () => {
-  it('surfaces error', async () => {
+  it('delays appropriately on 429 error', async () => {
     const analytics = createTestAnalytics({
       tokenManagerProps: getTokenManagerProps(),
     })
-    oauthFetcher.mockReturnValue(createOAuthError({ status: 425 }))
-
-    const eventName = 'Test Event'
+    const retryTime = Date.now() + 250
+    oauthFetcher
+      .mockReturnValueOnce(
+        createOAuthError({
+          status: 429,
+          headers: { 'X-RateLimit-Reset': retryTime },
+        })
+      )
+      .mockReturnValue(
+        createOAuthSuccess({
+          access_token: 'token',
+          expires_in: 100,
+        })
+      )
 
     analytics.track({
-      event: eventName,
+      event: 'Test Event',
       anonymousId: 'unknown',
       userId: 'known',
       timestamp: timestamp,
     })
+    const ctx1 = await resolveCtx(analytics, 'track') // forces exception to be thrown
+    expect(ctx1.event.type).toEqual('track')
+    await analytics.closeAndFlush()
+    expect(retryTime).toBeLessThan(Date.now())
+  })
+})
+
+describe('OAuth Failure', () => {
+  it('surfaces error after retries', async () => {
+    const analytics = createTestAnalytics({
+      tokenManagerProps: getTokenManagerProps(),
+    })
 
-    const ctx1 = await resolveCtx(analytics, 'track')
+    oauthFetcher.mockReturnValue(createOAuthError({ status: 500 }))
 
-    expect(ctx1.event.type).toEqual('track')
-    expect(ctx1.event.event).toEqual(eventName)
-    expect(ctx1.event.properties).toEqual({})
-    expect(ctx1.event.anonymousId).toEqual('unknown')
-    expect(ctx1.event.userId).toEqual('known')
-    expect(ctx1.event.timestamp).toEqual(timestamp)
+    const eventName = 'Test Event'
 
-    expect(oauthFetcher).toHaveBeenCalledTimes(3)
+    try {
+      analytics.track({
+        event: eventName,
+        anonymousId: 'unknown',
+        userId: 'known',
+        timestamp: timestamp,
+      })
 
-    await analytics.closeAndFlush()
+      const ctx1 = await resolveCtx(analytics, 'track')
+
+      expect(ctx1.event.type).toEqual('track')
+      expect(ctx1.event.event).toEqual(eventName)
+      expect(ctx1.event.properties).toEqual({})
+      expect(ctx1.event.anonymousId).toEqual('unknown')
+      expect(ctx1.event.userId).toEqual('known')
+      expect(ctx1.event.timestamp).toEqual(timestamp)
+
+      expect(oauthFetcher).toHaveBeenCalledTimes(3)
+
+      await analytics.closeAndFlush()
+
+      throw new Error('fail')
+    } catch (err: any) {
+      expect(err.reason).toEqual(new Error('[500] Foo'))
+      expect(err.code).toMatch(/delivery_failure/)
+    }
+  })
+
+  it('surfaces error after failing immediately', async () => {
+    const logger = jest.fn()
+    const analytics = createTestAnalytics({
+      tokenManagerProps: getTokenManagerProps(),
+    }).on('error', (err) => {
+      logger(err)
+    })
+
+    oauthFetcher.mockReturnValue(createOAuthError({ status: 400 }))
+
+    try {
+      analytics.track({
+        event: 'Test Event',
+        anonymousId: 'unknown',
+        userId: 'known',
+        timestamp: timestamp,
+      })
+
+      const ctx1 = await resolveCtx(analytics, 'track') // forces exception to be thrown
+      expect(ctx1.event.type).toEqual('track')
+      await analytics.closeAndFlush()
+
+      expect(logger).toHaveBeenCalledWith('foo')
+      throw new Error('fail')
+    } catch (err: any) {
+      expect(err.reason).toEqual(new Error('[400] Foo'))
+      expect(err.code).toMatch(/delivery_failure/)
+    }
+  })
+
+  it('handles a bad key', async () => {
+    const props = getTokenManagerProps()
+    props.clientKey = Buffer.from('Garbage')
+    const analytics = createTestAnalytics({
+      tokenManagerProps: props,
+    })
+
+    try {
+      analytics.track({
+        event: 'Test Event',
+        anonymousId: 'unknown',
+        userId: 'known',
+        timestamp: timestamp,
+      })
+      await analytics.closeAndFlush()
+      const ctx1 = await resolveCtx(analytics, 'track') // forces exception to be thrown
+      expect(ctx1.event.type).toEqual('track')
+      throw new Error('fail')
+    } catch (err: any) {
+      expect(err.reason).toEqual(
+        new Error(
+          'secretOrPrivateKey must be an asymmetric key when using RS256'
+        )
+      )
+    }
+  })
+
+  describe('TAPI rejection', () => {
+    it('surfaces error', async () => {
+      const analytics = createTestAnalytics({
+        tokenManagerProps: getTokenManagerProps(),
+        httpClient: tapiTestClient,
+      })
+      const eventName = 'Test Event'
+
+      oauthFetcher.mockReturnValue(
+        createOAuthSuccess({
+          access_token: 'token',
+          expires_in: 100,
+        })
+      )
+      tapiFetcher.mockReturnValue(
+        createError({
+          status: 400,
+          statusText:
+            '{"success":false,"message":"malformed JSON","code":"invalid_request"}',
+        })
+      )
+
+      try {
+        analytics.track({
+          event: eventName,
+          anonymousId: 'unknown',
+          userId: 'known',
+          timestamp: timestamp,
+        })
+
+        const ctx1 = await resolveCtx(analytics, 'track')
+
+        expect(ctx1.event.type).toEqual('track')
+        expect(ctx1.event.event).toEqual(eventName)
+        expect(ctx1.event.properties).toEqual({})
+        expect(ctx1.event.anonymousId).toEqual('unknown')
+        expect(ctx1.event.userId).toEqual('known')
+        expect(ctx1.event.timestamp).toEqual(timestamp)
+
+        expect(oauthFetcher).toHaveBeenCalledTimes(1)
+
+        await analytics.closeAndFlush()
+        throw new Error('fail')
+      } catch (err: any) {
+        expect(err.code).toBe('delivery_failure')
+      }
+    })
   })
 })

packages/node/src/lib/__tests__/token-manager.test.ts

@@ -116,13 +116,13 @@ test('OAuth rate limit', async () => {
     .mockReturnValueOnce(
       createOAuthError({
         status: 429,
-        headers: { 'X-RateLimit-Reset': Date.now() + 1000 },
+        headers: { 'X-RateLimit-Reset': Date.now() + 250 },
       })
     )
     .mockReturnValueOnce(
       createOAuthError({
         status: 429,
-        headers: { 'X-RateLimit-Reset': Date.now() + 1000 },
+        headers: { 'X-RateLimit-Reset': Date.now() + 500 },
       })
     )
     .mockReturnValue(
@@ -135,11 +135,11 @@ test('OAuth rate limit', async () => {
   const tokenManager = getTokenManager()
 
   const tokenPromise = tokenManager.getAccessToken()
-  await sleep(250)
+  await sleep(25)
   expect(fetcher).toHaveBeenCalledTimes(1)
   await sleep(250)
   expect(fetcher).toHaveBeenCalledTimes(2)
-  await sleep(350)
+  await sleep(250)
   expect(fetcher).toHaveBeenCalledTimes(3)
 
   const token = await tokenPromise

packages/node/src/lib/http-client.ts

@@ -27,7 +27,7 @@ export interface HTTPFetchRequest {
  */
 export interface HTTPResponse {
   headers: Record<string, any>
-  body: string | ReadableStream<Uint8Array> | null
+  json(): Promise<any>
   status: number
   statusText: string
 }

packages/node/src/lib/token-manager.ts

@@ -113,7 +113,7 @@ export class TokenManager {
       if (response.status === 200) {
         let body: any
         try {
-          body = await response.body // TODO: Replace with actual method to get body - needs discussion since different HTTP clients expose this differently (buffers, streams, strings, objects)
+          body = await response.json() //body?.getReader().read() // TODO: Replace with actual method to get body - needs discussion since different HTTP clients expose this differently (buffers, streams, strings, objects)
         } catch (err) {
           // Errors reading the body (not parsing) are likely networking issues, we can retry
           retryCount++
@@ -127,7 +127,7 @@ export class TokenManager {
         }
         let token: AccessToken
         try {
-          const parsedBody = JSON.parse(body)
+          const parsedBody = /*JSON.parse(*/ body //)
           // TODO: validate JSON
           token = parsedBody
           this.tokenEmitter.emit('access_token', { token })
@@ -161,10 +161,9 @@ export class TokenManager {
         )
         if (isFinite(rateLimitResetTime)) {
           timeUntilRefreshInMs =
-            (rateLimitResetTime - Date.now()) / 2 +
-            this.clockSkewInSeconds * 1000
+            rateLimitResetTime - Date.now() + this.clockSkewInSeconds * 1000
         } else {
-          timeUntilRefreshInMs = 60 * 1000
+          timeUntilRefreshInMs = 5 * 1000
         }
       } else if ([400, 401, 415].includes(response.status)) {
         // Unrecoverable errors