OAuth2 support (#935)

Co-authored-by: Christopher Radek <christopher.radek@segment.com>
Co-authored-by: Seth Silesky <5115498+silesky@users.noreply.github.com>

.changeset/empty-avocados-juggle.md

@@ -0,0 +1,5 @@
+---
+'@segment/analytics-node': major
+---
+
+Removing support for Node.js 14 and 16 as they are EOL

.changeset/hot-eyes-run.md

@@ -0,0 +1,15 @@
+---
+'@segment/analytics-node': major
+---
+
+Support for Segment OAuth2
+
+OAuth2 must be enabled from the Segment dashboard. You will need a PEM format
+private/public key pair.  Once you've uploaded your public key, you will need
+the OAuth Client Id, the Key Id, and your private key.  You can set these in
+the new OAuthSettings type and provide it in your Analytics configuration.
+
+Note: This introduces a breaking change only if you have implemented a custom
+HTTPClient.  HTTPClientRequest `data: Record<string, any>` has changed to 
+`body: string`. Processing data into a string now occurs before calls to
+`makeRequest`

.github/workflows/ci.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        node-version: [14, 16, 18]
+        node-version: [18]
     steps:
       - uses: actions/checkout@v3
       - uses: actions/setup-node@v3

packages/node-integration-tests/src/cloudflare-tests/index.test.ts

@@ -91,6 +91,68 @@ describe('Analytics in Cloudflare workers', () => {
             },
           ],
           "sentAt": StringMatching /\\^\\\\d\\{4\\}-\\\\d\\{2\\}-\\\\d\\{2\\}T\\\\d\\{2\\}:\\\\d\\{2\\}:\\\\d\\{2\\}\\\\\\.\\\\d\\{3\\}Z\\$/,
+          "writeKey": "__TEST__",
+        },
+      ]
+    `
+    )
+  })
+
+  it('can send an oauth event', async () => {
+    const batches: any[] = []
+    const oauths: any[] = []
+    mockSegmentServer.on('batch', (batch) => {
+      batches.push(batch)
+    })
+
+    mockSegmentServer.on('token', (oauth) => {
+      oauths.push(oauth)
+    })
+
+    const worker = await unstable_dev(
+      joinPath(__dirname, 'workers', 'send-oauth-event.ts'),
+      {
+        experimental: {
+          disableExperimentalWarning: true,
+        },
+        bundle: true,
+      }
+    )
+    const response = await worker.fetch('http://localhost')
+    await response.text()
+    await worker.stop()
+
+    console.log(batches)
+
+    expect(oauths[0]).toHaveLength(756)
+
+    expect(batches).toMatchInlineSnapshot(
+      batches.map(() => snapshotMatchers.getReqBody(1)),
+      `
+      [
+        {
+          "batch": [
+            {
+              "_metadata": {
+                "jsRuntime": "cloudflare-worker",
+              },
+              "context": {
+                "library": {
+                  "name": "@segment/analytics-node",
+                  "version": Any<String>,
+                },
+              },
+              "event": "some-event",
+              "integrations": {},
+              "messageId": Any<String>,
+              "properties": {},
+              "timestamp": StringMatching /\\^\\\\d\\{4\\}-\\\\d\\{2\\}-\\\\d\\{2\\}T\\\\d\\{2\\}:\\\\d\\{2\\}:\\\\d\\{2\\}\\\\\\.\\\\d\\{3\\}Z\\$/,
+              "type": "track",
+              "userId": "some-user",
+            },
+          ],
+          "sentAt": StringMatching /\\^\\\\d\\{4\\}-\\\\d\\{2\\}-\\\\d\\{2\\}T\\\\d\\{2\\}:\\\\d\\{2\\}:\\\\d\\{2\\}\\\\\\.\\\\d\\{3\\}Z\\$/,
+          "writeKey": "__TEST__",
         },
       ]
     `
@@ -235,6 +297,7 @@ describe('Analytics in Cloudflare workers', () => {
             },
           ],
           "sentAt": StringMatching /\\^\\\\d\\{4\\}-\\\\d\\{2\\}-\\\\d\\{2\\}T\\\\d\\{2\\}:\\\\d\\{2\\}:\\\\d\\{2\\}\\\\\\.\\\\d\\{3\\}Z\\$/,
+          "writeKey": "__TEST__",
         },
       ]
     `
@@ -314,6 +377,7 @@ describe('Analytics in Cloudflare workers', () => {
             },
           ],
           "sentAt": StringMatching /\\^\\\\d\\{4\\}-\\\\d\\{2\\}-\\\\d\\{2\\}T\\\\d\\{2\\}:\\\\d\\{2\\}:\\\\d\\{2\\}\\\\\\.\\\\d\\{3\\}Z\\$/,
+          "writeKey": "__TEST__",
         },
         {
           "batch": [
@@ -359,6 +423,7 @@ describe('Analytics in Cloudflare workers', () => {
             },
           ],
           "sentAt": StringMatching /\\^\\\\d\\{4\\}-\\\\d\\{2\\}-\\\\d\\{2\\}T\\\\d\\{2\\}:\\\\d\\{2\\}:\\\\d\\{2\\}\\\\\\.\\\\d\\{3\\}Z\\$/,
+          "writeKey": "__TEST__",
         },
         {
           "batch": [
@@ -404,6 +469,7 @@ describe('Analytics in Cloudflare workers', () => {
             },
           ],
           "sentAt": StringMatching /\\^\\\\d\\{4\\}-\\\\d\\{2\\}-\\\\d\\{2\\}T\\\\d\\{2\\}:\\\\d\\{2\\}:\\\\d\\{2\\}\\\\\\.\\\\d\\{3\\}Z\\$/,
+          "writeKey": "__TEST__",
         },
       ]
     `

packages/node-integration-tests/src/cloudflare-tests/workers/send-oauth-event.ts

@@ -0,0 +1,54 @@
+/// <reference types="@cloudflare/workers-types" />
+import { Analytics, OAuthSettings } from '@segment/analytics-node'
+
+export default {
+  async fetch(_request: Request, _env: {}, _ctx: ExecutionContext) {
+    const settings: OAuthSettings = {
+      clientId: '__CLIENTID__',
+      // Has to be a valid key to encrypt the JWT, not used for anything else
+      clientKey: `-----BEGIN PRIVATE KEY-----
+MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDVll7uJaH322IN
+PQsH2aOXZJ2r1q+6hpVK1R5JV1p41PUzn8pOxyXFHWB+53dUd4B8qywKS36XQjp0
+VmhR1tQ22znQ9ZCM6y4LGeOJBjAZiFZLcGQNNrDFC0WGWTrK1ZTS2K7p5qy4fIXG
+laNkMXiGGCawkgcHAdOvPTy8m1d9a6YSetYVmBP/tEYN95jPyZFIoHQfkQPBPr9W
+cWPpdEBzasHV5d957akjurPpleDiD5as66UW4dkWXvS7Wu7teCLCyDApcyJKTb2Z
+SXybmWjhIZuctZMAx3wT/GgW3FbkGaW5KLQgBUMzjpL0fCtMatlqckMD92ll1FuK
+R+HnXu05AgMBAAECggEBAK4o2il4GDUh9zbyQo9ZIPLuwT6AZXRED3Igi3ykNQp4
+I6S/s9g+vQaY6LkyBnSiqOt/K/8NBiFSiJWaa5/n+8zrP56qzf6KOlYk+wsdN5Vq
+PWtwLrUzljpl8YAWPEFunNa8hwwE42vfZbnDBKNLT4qQIOQzfnVxQOoQlfj49gM2
+iSrblvsnQTyucFy3UyTeioHbh8q2Xqcxry5WUCOrFDd3IIwATTpLZGw0IPeuFJbJ
+NfBizLEcyJaM9hujQU8PRCRd16MWX+bbYM6Mh4dkT40QXWsVnHBHwsgPdQgDgseF
+Na4ajtHoC0DlwYCXpCm3IzJfKfq/LR2q8NDUgKeF4AECgYEA9nD4czza3SRbzhpZ
+bBoK77CSNqCcMAqyuHB0hp/XX3yB7flF9PIPb2ReO8wwmjbxn+bm7PPz2Uwd2SzO
+pU+FXmyKJr53Jxw/hmDWZCoh42gsGDlVqpmytzsj74KlaYiMyZmEGbD7t/FGfNGV
+LdLDJaHIYxEimFviOTXKCeKvPAECgYEA3d8tv4jdp1uAuRZiU9Z/tfw5mJOi3oXF
+8AdFFDwaPzcTorEAxjrt9X6IjPbLIDJNJtuXYpe+dG6720KyuNnhLhWW9oZEJTwT
+dUgqZ2fTCOS9uH0jSn+ZFlgTWI6UDQXRwE7z8avlhMIrQVmPsttGTo7V6sQVtGRx
+bNj2RSVekTkCgYAJvy4UYLPHS0jWPfSLcfw8vp8JyhBjVgj7gncZW/kIrcP1xYYe
+yfQSU8XmV40UjFfCGz/G318lmP0VOdByeVKtCV3talsMEPHyPqI8E+6DL/uOebYJ
+qUqINK6XKnOgWOY4kvnGillqTQCcry1XQp61PlDOmj7kB75KxPXYrj6AAQKBgQDa
++ixCv6hURuEyy77cE/YT/Q4zYnL6wHjtP5+UKwWUop1EkwG6o+q7wtiul90+t6ah
+1VUCP9X/QFM0Qg32l0PBohlO0pFrVnG17TW8vSHxwyDkds1f97N19BOT8ZR5jebI
+sKPfP9LVRnY+l1BWLEilvB+xBzqMwh2YWkIlWI6PMQKBgGi6TBnxp81lOYrxVRDj
+/3ycRnVDmBdlQKFunvfzUBmG1mG/G0YHeVSUKZJGX7w2l+jnDwIA383FcUeA8X6A
+l9q+amhtkwD/6fbkAu/xoWNl+11IFoxd88y2ByBFoEKB6UVLuCTSKwXDqzEZet7x
+mDyRxq7ohIzLkw8b8buDeuXZ
+-----END PRIVATE KEY-----`,
+      keyId: '__KEYID__',
+      maxRetries: 3,
+      authServer: 'http://localhost:3000',
+      scope: 'tracking_api:write',
+    }
+
+    const analytics = new Analytics({
+      writeKey: '__TEST__',
+      host: 'http://localhost:3000',
+      oauthSettings: settings,
+    })
+
+    analytics.track({ userId: 'some-user', event: 'some-event' })
+
+    await analytics.closeAndFlush()
+    return new Response('ok')
+  },
+}

packages/node-integration-tests/src/server/mock-segment-workers.ts

@@ -20,29 +20,52 @@ function isBatchRequest(req: IncomingMessage) {
   return false
 }
 
+function isTokenRequest(req: IncomingMessage) {
+  if (req.url?.endsWith('/token')) {
+    return true
+  }
+  return false
+}
+
 type BatchHandler = (batch: any) => void
+type TokenHandler = (token: any) => void
 
 export class MockSegmentServer {
   private server: ReturnType<typeof createServer>
   private port: number
   private onBatchHandlers: Set<BatchHandler> = new Set()
+  private onTokenHandlers: Set<TokenHandler> = new Set()
 
   constructor(port: number) {
     this.port = port
     this.server = createServer(async (req, res) => {
-      if (!isBatchRequest(req)) {
+      if (!isBatchRequest(req) && !isTokenRequest(req)) {
         res.writeHead(404, { 'Content-Type': 'application/json' })
         res.end(JSON.stringify({ success: false }))
         return
       }
 
       const text = await getRequestText(req)
-      const batch = JSON.parse(text)
-      this.onBatchHandlers.forEach((handler) => {
-        handler(batch)
-      })
       res.writeHead(200, { 'Content-Type': 'application/json' })
-      res.end(JSON.stringify({ success: true }))
+      if (isBatchRequest(req)) {
+        const batch = JSON.parse(text)
+        this.onBatchHandlers.forEach((handler) => {
+          handler(batch)
+        })
+        res.end(JSON.stringify({ success: true }))
+      } else if (isTokenRequest(req)) {
+        this.onTokenHandlers.forEach((handler) => {
+          handler(text)
+        })
+        res.end(
+          JSON.stringify({
+            access_token: '__TOKEN__',
+            token_type: 'Bearer',
+            scope: 'tracking_api:write',
+            expires_in: 86400,
+          })
+        )
+      }
     })
   }
 
@@ -68,11 +91,19 @@ export class MockSegmentServer {
     })
   }
 
-  on(_event: 'batch', handler: BatchHandler) {
-    this.onBatchHandlers.add(handler)
+  on(_event: 'batch' | 'token', handler: BatchHandler) {
+    if (_event === 'batch') {
+      this.onBatchHandlers.add(handler)
+    } else if (_event === 'token') {
+      this.onTokenHandlers.add(handler)
+    }
   }
 
-  off(_event: 'batch', handler: BatchHandler) {
-    this.onBatchHandlers.delete(handler)
+  off(_event: 'batch' | 'token', handler: BatchHandler) {
+    if (_event === 'batch') {
+      this.onBatchHandlers.delete(handler)
+    } else if (_event === 'token') {
+      this.onTokenHandlers.delete(handler)
+    }
   }
 }

packages/node/README.md

@@ -147,4 +147,30 @@ export default {
 
 ```
 
+### OAuth 2
+In order to guarantee authorized communication between your server environment and Segment's Tracking API, you can [enable OAuth 2 in your Segment workspace](https://segment.com/docs/partners/enable-with-segment/).  To support the non-interactive server environment, the OAuth workflow used is a signed client assertion JWT.  You will need a public and private key pair where the public key is uploaded to the segment dashboard and the private key is kept in your server environment to be used by this SDK. Your server will verify its identity by signing a token request and will receive a token that is used to to authorize all communication with the Segment Tracking API.
 
+You will also need to provide the OAuth Application ID and the public key's ID, both of which are provided in the Segment dashboard.  You should ensure that you are implementing handling for Analytics SDK errors.  Good logging will help distinguish any configuration issues.
+
+```ts
+import { Analytics, OAuthSettings } from '@segment/analytics-node';
+import { readFileSync } from 'fs'
+
+const privateKey = readFileSync('private.pem', 'utf8')
+
+const settings: OAuthSettings = {
+  clientId: '<CLIENT_ID_FROM_DASHBOARD>',
+  clientKey: privateKey,
+  keyId: '<PUB_KEY_ID_FROM_DASHBOARD>',
+}
+
+const analytics = new Analytics({
+  writeKey: '<MY_WRITE_KEY>',
+  oauthSettings: settings,
+})
+
+analytics.on('error', (err) => { console.error(err) })
+
+analytics.track({ userId: 'foo', event: 'bar' })
+
+```

packages/node/package.json

@@ -17,7 +17,7 @@
     "!*.tsbuildinfo"
   ],
   "engines": {
-    "node": ">=14"
+    "node": ">=18"
   },
   "scripts": {
     ".": "yarn run -T turbo run --filter=@segment/analytics-node",
@@ -39,6 +39,7 @@
     "@segment/analytics-core": "1.4.1",
     "@segment/analytics-generic-utils": "1.1.1",
     "buffer": "^6.0.3",
+    "jose": "^5.1.0",
     "node-fetch": "^2.6.7",
     "tslib": "^2.4.1"
   },

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

@@ -28,11 +28,11 @@ describe('Ability for users to exit without losing events', () => {
   })
   const _helpers = {
     getFetchCalls: () =>
-      makeReqSpy.mock.calls.map(([{ url, method, data, headers }]) => ({
+      makeReqSpy.mock.calls.map(([{ url, method, body, headers }]) => ({
         url,
         method,
         headers,
-        data,
+        body,
       })),
     makeTrackCall: (analytics = ajs, cb?: (...args: any[]) => void) => {
       analytics.track({ userId: 'foo', event: 'Thing Updated' }, cb)
@@ -206,7 +206,7 @@ describe('Ability for users to exit without losing events', () => {
       expect(elapsedTime).toBeLessThan(100)
       const calls = _helpers.getFetchCalls()
       expect(calls.length).toBe(1)
-      expect(calls[0].data.batch.length).toBe(2)
+      expect(JSON.parse(calls[0].body).batch.length).toBe(2)
     })
 
     test('should wait to flush if close is called and an event has not made it to the segment.io plugin yet', async () => {
@@ -238,7 +238,7 @@ describe('Ability for users to exit without losing events', () => {
       expect(elapsedTime).toBeLessThan(TRACK_DELAY * 2)
       const calls = _helpers.getFetchCalls()
       expect(calls.length).toBe(1)
-      expect(calls[0].data.batch.length).toBe(2)
+      expect(JSON.parse(calls[0].body).batch.length).toBe(2)
     })
   })
 

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

@@ -20,7 +20,6 @@ const helpers = {
   ) => {
     expect(url).toBe('https://api.segment.io/v1/batch')
     expect(options.headers).toEqual({
-      Authorization: 'Basic Zm9vOg==',
       'Content-Type': 'application/json',
       'User-Agent': 'analytics-node-next/latest',
     })