@segment/analytics-node - npm Package Compare versions

Comparing version 1.3.0 to 2.0.0

dist/cjs/app/analytics-node.js

@@ -15,7 +15,12 @@ "use strict";
 class Analytics extends emitter_1.NodeEmitter {
+    _eventFactory;
+    _isClosed = false;
+    _pendingEvents = 0;
+    _closeAndFlushDefaultTimeout;
+    _publisher;
+    _isFlushing = false;
+    _queue;
+    ready;
     constructor(settings) {
         super();
-        this._isClosed = false;
-        this._pendingEvents = 0;
-        this._isFlushing = false;
         (0, settings_1.validateSettings)(settings);
@@ -38,2 +43,3 @@ this._eventFactory = new event_factory_1.NodeEventFactory();
                 : settings.httpClient ?? new http_client_1.FetchHTTPClient(),
+            oauthSettings: settings.oauthSettings,
         }, this);
@@ -40,0 +46,0 @@ this._publisher = publisher;

dist/cjs/generated/version.js

@@ -5,3 +5,3 @@ "use strict";
 // This file is generated.
-exports.version = '1.3.0';
+exports.version = '2.0.0';
 //# sourceMappingURL=version.js.map

dist/cjs/index.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-exports.FetchHTTPClient = exports.Context = exports.Analytics = void 0;
-var analytics_node_1 = require("./app/analytics-node");
-Object.defineProperty(exports, "Analytics", { enumerable: true, get: function () { return analytics_node_1.Analytics; } });
-var context_1 = require("./app/context");
-Object.defineProperty(exports, "Context", { enumerable: true, get: function () { return context_1.Context; } });
-var http_client_1 = require("./lib/http-client");
-Object.defineProperty(exports, "FetchHTTPClient", { enumerable: true, get: function () { return http_client_1.FetchHTTPClient; } });
+const tslib_1 = require("tslib");
+tslib_1.__exportStar(require("./index.common"), exports);
 // export Analytics as both a named export and a default export (for backwards-compat. reasons)
-const analytics_node_2 = require("./app/analytics-node");
-exports.default = analytics_node_2.Analytics;
+const index_common_1 = require("./index.common");
+exports.default = index_common_1.Analytics;
 //# sourceMappingURL=index.js.map

dist/cjs/lib/abort.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-exports.abortSignalAfterTimeout = exports.AbortSignal = void 0;
+exports.abortSignalAfterTimeout = exports.AbortController = exports.AbortSignal = void 0;
 /**
@@ -13,7 +13,5 @@ * use non-native event emitter for the benefit of non-node runtimes like CF workers.
 class AbortSignal {
-    constructor() {
-        this.onabort = null;
-        this.aborted = false;
-        this.eventEmitter = new analytics_generic_utils_1.Emitter();
-    }
+    onabort = null;
+    aborted = false;
+    eventEmitter = new analytics_generic_utils_1.Emitter();
     toString() {
@@ -47,5 +45,3 @@ return '[object AbortSignal]';
 class AbortController {
-    constructor() {
-        this.signal = new AbortSignal();
-    }
+    signal = new AbortSignal();
     abort() {
@@ -64,2 +60,3 @@ if (this.signal.aborted)
 }
+exports.AbortController = AbortController;
 /**
@@ -66,0 +63,0 @@ * @param timeoutMs - Set a request timeout, after which the request is cancelled.

dist/cjs/lib/http-client.js

@@ -10,2 +10,3 @@ "use strict";
 class FetchHTTPClient {
+    _fetch;
     constructor(fetchFn) {
@@ -20,3 +21,3 @@ this._fetch = fetchFn ?? fetch_1.fetch;
             headers: options.headers,
-            body: JSON.stringify(options.data),
+            body: options.body,
             signal: signal,
@@ -23,0 +24,0 @@ };

dist/cjs/plugins/segmentio/context-batch.js

@@ -8,6 +8,7 @@ "use strict";
 class ContextBatch {
+    id = (0, uuid_1.uuid)();
+    items = [];
+    sizeInBytes = 0;
+    maxEventCount;
     constructor(maxEventCount) {
-        this.id = (0, uuid_1.uuid)();
-        this.items = [];
-        this.sizeInBytes = 0;
         this.maxEventCount = Math.max(1, maxEventCount);
@@ -14,0 +15,0 @@ }

dist/cjs/plugins/segmentio/publisher.js

@@ -8,3 +8,3 @@ "use strict";
 const context_batch_1 = require("./context-batch");
-const base_64_encode_1 = require("../../lib/base-64-encode");
+const token_manager_1 = require("../../lib/token-manager");
 function sleep(timeoutInMs) {
@@ -18,3 +18,16 @@ return new Promise((resolve) => setTimeout(resolve, timeoutInMs));
 class Publisher {
-    constructor({ host, path, maxRetries, flushAt, flushInterval, writeKey, httpRequestTimeout, httpClient, disable, }, emitter) {
+    pendingFlushTimeout;
+    _batch;
+    _flushInterval;
+    _flushAt;
+    _maxRetries;
+    _url;
+    _flushPendingItemsCount;
+    _httpRequestTimeout;
+    _emitter;
+    _disable;
+    _httpClient;
+    _writeKey;
+    _tokenManager;
+    constructor({ host, path, maxRetries, flushAt, flushInterval, writeKey, httpRequestTimeout, httpClient, disable, oauthSettings, }, emitter) {
         this._emitter = emitter;
@@ -24,3 +37,2 @@ this._maxRetries = maxRetries;
         this._flushInterval = flushInterval;
-        this._auth = (0, base_64_encode_1.b64encode)(`${writeKey}:`);
         this._url = (0, create_url_1.tryCreateFormattedUrl)(host ?? 'https://api.segment.io', path ?? '/v1/batch');
@@ -30,2 +42,10 @@ this._httpRequestTimeout = httpRequestTimeout ?? 10000;
         this._httpClient = httpClient;
+        this._writeKey = writeKey;
+        if (oauthSettings) {
+            this._tokenManager = new token_manager_1.TokenManager({
+                ...oauthSettings,
+                httpClient: oauthSettings.httpClient ?? httpClient,
+                maxRetries: oauthSettings.maxRetries ?? maxRetries,
+            });
+        }
     }
@@ -53,3 +73,6 @@ createBatch() {
         if (!pendingItemsCount) {
-            // if number of pending items is 0, there is nothing to flush
+            // if number of pending items is 0, there will never be anything else entering the batch, since the app is closed.
+            if (this._tokenManager) {
+                this._tokenManager.stopPoller();
+            }
             return;
@@ -65,3 +88,10 @@ }
         if (isExpectingNoMoreItems) {
-            this.send(this._batch).catch(noop);
+            this.send(this._batch)
+                .catch(noop)
+                .finally(() => {
+                // stop poller so program can exit ().
+                if (this._tokenManager) {
+                    this._tokenManager.stopPoller();
+                }
+            });
             this.clearBatch();
@@ -140,15 +170,27 @@ }
                 }
+                let authString = undefined;
+                if (this._tokenManager) {
+                    const token = await this._tokenManager.getAccessToken();
+                    if (token && token.access_token) {
+                        authString = `Bearer ${token.access_token}`;
+                    }
+                }
+                const headers = {
+                    'Content-Type': 'application/json',
+                    'User-Agent': 'analytics-node-next/latest',
+                    ...(authString ? { Authorization: authString } : {}),
+                };
                 const request = {
                     url: this._url,
                     method: 'POST',
-                    headers: {
-                        'Content-Type': 'application/json',
-                        Authorization: `Basic ${this._auth}`,
-                        'User-Agent': 'analytics-node-next/latest',
-                    },
-                    data: { batch: events, sentAt: new Date() },
+                    headers: headers,
+                    body: JSON.stringify({
+                        batch: events,
+                        writeKey: this._writeKey,
+                        sentAt: new Date(),
+                    }),
                     httpRequestTimeout: this._httpRequestTimeout,
                 };
                 this._emitter.emit('http_request', {
-                    body: request.data,
+                    body: request.body,
                     method: request.method,
@@ -164,2 +206,10 @@ url: request.url,
                 }
+                else if (this._tokenManager &&
+                    (response.status === 400 ||
+                        response.status === 401 ||
+                        response.status === 403)) {
+                    // Retry with a new OAuth token if we have OAuth data
+                    this._tokenManager.clearToken();
+                    failureReason = new Error(`[${response.status}] ${response.statusText}`);
+                }
                 else if (response.status === 400) {
@@ -166,0 +216,0 @@ // https://segment.com/docs/connections/sources/catalog/libraries/server/http-api/#max-request-size

dist/esm/app/analytics-node.js

@@ -12,7 +12,12 @@ import { bindAll, pTimeout } from '@segment/analytics-core';
 export class Analytics extends NodeEmitter {
+    _eventFactory;
+    _isClosed = false;
+    _pendingEvents = 0;
+    _closeAndFlushDefaultTimeout;
+    _publisher;
+    _isFlushing = false;
+    _queue;
+    ready;
     constructor(settings) {
         super();
-        this._isClosed = false;
-        this._pendingEvents = 0;
-        this._isFlushing = false;
         validateSettings(settings);
@@ -35,2 +40,3 @@ this._eventFactory = new NodeEventFactory();
                 : settings.httpClient ?? new FetchHTTPClient(),
+            oauthSettings: settings.oauthSettings,
         }, this);
@@ -37,0 +43,0 @@ this._publisher = publisher;

dist/esm/generated/version.js

 // This file is generated.
-export const version = '1.3.0';
+export const version = '2.0.0';
 //# sourceMappingURL=version.js.map

dist/esm/index.js

@@ -1,7 +0,5 @@
-export { Analytics } from './app/analytics-node';
-export { Context } from './app/context';
-export { FetchHTTPClient, } from './lib/http-client';
+export * from './index.common';
 // export Analytics as both a named export and a default export (for backwards-compat. reasons)
-import { Analytics } from './app/analytics-node';
+import { Analytics } from './index.common';
 export default Analytics;
 //# sourceMappingURL=index.js.map

dist/esm/lib/abort.js

@@ -10,7 +10,5 @@ /**
 export class AbortSignal {
-    constructor() {
-        this.onabort = null;
-        this.aborted = false;
-        this.eventEmitter = new Emitter();
-    }
+    onabort = null;
+    aborted = false;
+    eventEmitter = new Emitter();
     toString() {
@@ -42,6 +40,4 @@ return '[object AbortSignal]';
  */
-class AbortController {
-    constructor() {
-        this.signal = new AbortSignal();
-    }
+export class AbortController {
+    signal = new AbortSignal();
     abort() {
@@ -48,0 +44,0 @@ if (this.signal.aborted)

dist/esm/lib/http-client.js

@@ -7,2 +7,3 @@ import { abortSignalAfterTimeout } from './abort';
 export class FetchHTTPClient {
+    _fetch;
     constructor(fetchFn) {
@@ -17,3 +18,3 @@ this._fetch = fetchFn ?? defaultFetch;
             headers: options.headers,
-            body: JSON.stringify(options.data),
+            body: options.body,
             signal: signal,
@@ -20,0 +21,0 @@ };

dist/esm/plugins/segmentio/context-batch.js

@@ -5,6 +5,7 @@ import { uuid } from '../../lib/uuid';
 export class ContextBatch {
+    id = uuid();
+    items = [];
+    sizeInBytes = 0;
+    maxEventCount;
     constructor(maxEventCount) {
-        this.id = uuid();
-        this.items = [];
-        this.sizeInBytes = 0;
         this.maxEventCount = Math.max(1, maxEventCount);
@@ -11,0 +12,0 @@ }

dist/esm/plugins/segmentio/publisher.js

@@ -5,3 +5,3 @@ import { backoff } from '@segment/analytics-core';
 import { ContextBatch } from './context-batch';
-import { b64encode } from '../../lib/base-64-encode';
+import { TokenManager } from '../../lib/token-manager';
 function sleep(timeoutInMs) {
@@ -15,3 +15,16 @@ return new Promise((resolve) => setTimeout(resolve, timeoutInMs));
 export class Publisher {
-    constructor({ host, path, maxRetries, flushAt, flushInterval, writeKey, httpRequestTimeout, httpClient, disable, }, emitter) {
+    pendingFlushTimeout;
+    _batch;
+    _flushInterval;
+    _flushAt;
+    _maxRetries;
+    _url;
+    _flushPendingItemsCount;
+    _httpRequestTimeout;
+    _emitter;
+    _disable;
+    _httpClient;
+    _writeKey;
+    _tokenManager;
+    constructor({ host, path, maxRetries, flushAt, flushInterval, writeKey, httpRequestTimeout, httpClient, disable, oauthSettings, }, emitter) {
         this._emitter = emitter;
@@ -21,3 +34,2 @@ this._maxRetries = maxRetries;
         this._flushInterval = flushInterval;
-        this._auth = b64encode(`${writeKey}:`);
         this._url = tryCreateFormattedUrl(host ?? 'https://api.segment.io', path ?? '/v1/batch');
@@ -27,2 +39,10 @@ this._httpRequestTimeout = httpRequestTimeout ?? 10000;
         this._httpClient = httpClient;
+        this._writeKey = writeKey;
+        if (oauthSettings) {
+            this._tokenManager = new TokenManager({
+                ...oauthSettings,
+                httpClient: oauthSettings.httpClient ?? httpClient,
+                maxRetries: oauthSettings.maxRetries ?? maxRetries,
+            });
+        }
     }
@@ -50,3 +70,6 @@ createBatch() {
         if (!pendingItemsCount) {
-            // if number of pending items is 0, there is nothing to flush
+            // if number of pending items is 0, there will never be anything else entering the batch, since the app is closed.
+            if (this._tokenManager) {
+                this._tokenManager.stopPoller();
+            }
             return;
@@ -62,3 +85,10 @@ }
         if (isExpectingNoMoreItems) {
-            this.send(this._batch).catch(noop);
+            this.send(this._batch)
+                .catch(noop)
+                .finally(() => {
+                // stop poller so program can exit ().
+                if (this._tokenManager) {
+                    this._tokenManager.stopPoller();
+                }
+            });
             this.clearBatch();
@@ -137,15 +167,27 @@ }
                 }
+                let authString = undefined;
+                if (this._tokenManager) {
+                    const token = await this._tokenManager.getAccessToken();
+                    if (token && token.access_token) {
+                        authString = `Bearer ${token.access_token}`;
+                    }
+                }
+                const headers = {
+                    'Content-Type': 'application/json',
+                    'User-Agent': 'analytics-node-next/latest',
+                    ...(authString ? { Authorization: authString } : {}),
+                };
                 const request = {
                     url: this._url,
                     method: 'POST',
-                    headers: {
-                        'Content-Type': 'application/json',
-                        Authorization: `Basic ${this._auth}`,
-                        'User-Agent': 'analytics-node-next/latest',
-                    },
-                    data: { batch: events, sentAt: new Date() },
+                    headers: headers,
+                    body: JSON.stringify({
+                        batch: events,
+                        writeKey: this._writeKey,
+                        sentAt: new Date(),
+                    }),
                     httpRequestTimeout: this._httpRequestTimeout,
                 };
                 this._emitter.emit('http_request', {
-                    body: request.data,
+                    body: request.body,
                     method: request.method,
@@ -161,2 +203,10 @@ url: request.url,
                 }
+                else if (this._tokenManager &&
+                    (response.status === 400 ||
+                        response.status === 401 ||
+                        response.status === 403)) {
+                    // Retry with a new OAuth token if we have OAuth data
+                    this._tokenManager.clearToken();
+                    failureReason = new Error(`[${response.status}] ${response.statusText}`);
+                }
                 else if (response.status === 400) {
@@ -163,0 +213,0 @@ // https://segment.com/docs/connections/sources/catalog/libraries/server/http-api/#max-request-size

dist/types/app/emitter.d.ts

@@ -17,3 +17,3 @@ import type { CoreEmitterContract } from '@segment/analytics-core';
             headers: Record<string, string>;
-            body: Record<string, any>;
+            body: string;
         }
@@ -20,0 +20,0 @@ ];

dist/types/app/settings.d.ts

 import { HTTPClient, HTTPFetchFn } from '../lib/http-client';
+import { OAuthSettings } from '../lib/types';
 export interface AnalyticsSettings {
@@ -46,4 +47,8 @@ /**
     httpClient?: HTTPFetchFn | HTTPClient;
+    /**
+     * Set up OAuth2 authentication between the client and Segment's endpoints
+     */
+    oauthSettings?: OAuthSettings;
 }
 export declare const validateSettings: (settings: AnalyticsSettings) => void;
 //# sourceMappingURL=settings.d.ts.map

dist/types/generated/version.d.ts

@@ -1,2 +0,2 @@
-export declare const version = "1.3.0";
+export declare const version = "2.0.0";
 //# sourceMappingURL=version.d.ts.map

dist/types/index.d.ts

@@ -1,8 +0,4 @@
-export { Analytics } from './app/analytics-node';
-export { Context } from './app/context';
-export { HTTPClient, FetchHTTPClient, HTTPFetchRequest, HTTPResponse, HTTPFetchFn, HTTPClientRequest, } from './lib/http-client';
-export type { Plugin, GroupTraits, UserTraits, TrackParams, IdentifyParams, AliasParams, GroupParams, PageParams, } from './app/types';
-export type { AnalyticsSettings } from './app/settings';
-import { Analytics } from './app/analytics-node';
+export * from './index.common';
+import { Analytics } from './index.common';
 export default Analytics;
 //# sourceMappingURL=index.d.ts.map

dist/types/lib/abort.d.ts

@@ -22,2 +22,12 @@ /// <reference types="node" />
 /**
+ * This polyfill is only neccessary to support versions of node < 14.17.
+ * Can be removed once node 14 support is dropped.
+ */
+export declare class AbortController {
+    signal: AbortSignal;
+    abort(): void;
+    toString(): string;
+    get [Symbol.toStringTag](): string;
+}
+/**
  * @param timeoutMs - Set a request timeout, after which the request is cancelled.
@@ -24,0 +34,0 @@ */

dist/types/lib/http-client.d.ts

@@ -20,2 +20,11 @@ /**
 /**
+ * This interface is meant to be compatible with the Headers interface.
+ * @link https://developer.mozilla.org/en-US/docs/Web/API/Headers
+ */
+export interface HTTPHeaders {
+    get: (key: string) => string | null;
+    has: (key: string) => boolean;
+    entries: () => IterableIterator<[string, any]>;
+}
+/**
  * This interface is meant to very minimally conform to the Response interface.
@@ -25,2 +34,4 @@ * @link https://developer.mozilla.org/en-US/docs/Web/API/Response
 export interface HTTPResponse {
+    headers?: Record<string, any> | HTTPHeaders;
+    text?: () => Promise<string>;
     status: number;
@@ -48,6 +59,5 @@ statusText: string;
     /**
-     * JSON data to be sent with the request (will be stringified)
-     * @example { "batch": [ ... ]}
+     * Data to be sent with the request
      */
-    data: Record<string, any>;
+    body: string;
     /**
@@ -54,0 +64,0 @@ * Specifies the timeout (in milliseconds) for an HTTP client to get an HTTP response from the server

dist/types/plugins/segmentio/publisher.d.ts

 import type { Context } from '../../app/context';
 import { NodeEmitter } from '../../app/emitter';
 import { HTTPClient } from '../../lib/http-client';
+import { OAuthSettings } from '../../lib/types';
 export interface PublisherProps {
@@ -14,2 +15,3 @@ host?: string;
     httpClient: HTTPClient;
+    oauthSettings?: OAuthSettings;
 }
@@ -25,3 +27,2 @@ /**
     private _maxRetries;
-    private _auth;
     private _url;
@@ -33,3 +34,5 @@ private _flushPendingItemsCount?;
     private _httpClient;
-    constructor({ host, path, maxRetries, flushAt, flushInterval, writeKey, httpRequestTimeout, httpClient, disable, }: PublisherProps, emitter: NodeEmitter);
+    private _writeKey;
+    private _tokenManager;
+    constructor({ host, path, maxRetries, flushAt, flushInterval, writeKey, httpRequestTimeout, httpClient, disable, oauthSettings, }: PublisherProps, emitter: NodeEmitter);
     private createBatch;
@@ -36,0 +39,0 @@ private clearBatch;

package.json

 {
   "name": "@segment/analytics-node",
-  "version": "1.3.0",
+  "version": "2.0.0",
   "main": "./dist/cjs/index.js",
@@ -20,3 +20,3 @@ "module": "./dist/esm/index.js",
   "engines": {
-    "node": ">=14"
+    "node": ">=18"
   },
@@ -43,2 +43,3 @@ "scripts": {
     "buffer": "^6.0.3",
+    "jose": "^5.1.0",
     "node-fetch": "^2.6.7",
@@ -45,0 +46,0 @@ "tslib": "^2.4.1"

README.md

@@ -150,2 +150,28 @@ # @segment/analytics-node
 
+### 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' })
+
+```

src/app/analytics-node.ts

@@ -63,2 +63,3 @@ import { CoreAnalytics, bindAll, pTimeout } from '@segment/analytics-core'
             : settings.httpClient ?? new FetchHTTPClient(),
+        oauthSettings: settings.oauthSettings,
       },
@@ -68,2 +69,3 @@ this as NodeEmitter
     this._publisher = publisher
+
     this.ready = this.register(plugin).then(() => undefined)
@@ -70,0 +72,0 @@

src/app/emitter.ts

@@ -18,3 +18,3 @@ import type { CoreEmitterContract } from '@segment/analytics-core'
       headers: Record<string, string>
-      body: Record<string, any>
+      body: string
     }
@@ -21,0 +21,0 @@ ]

src/app/settings.ts

 import { ValidationError } from '@segment/analytics-core'
 import { HTTPClient, HTTPFetchFn } from '../lib/http-client'
+import { OAuthSettings } from '../lib/types'
 
@@ -48,2 +49,6 @@ export interface AnalyticsSettings {
   httpClient?: HTTPFetchFn | HTTPClient
+  /**
+   * Set up OAuth2 authentication between the client and Segment's endpoints
+   */
+  oauthSettings?: OAuthSettings
 }
@@ -50,0 +55,0 @@

src/generated/version.ts

 // This file is generated.
-export const version = '1.3.0'
+export const version = '2.0.0'

src/index.ts

@@ -1,26 +0,5 @@
-export { Analytics } from './app/analytics-node'
-export { Context } from './app/context'
-export {
-  HTTPClient,
-  FetchHTTPClient,
-  HTTPFetchRequest,
-  HTTPResponse,
-  HTTPFetchFn,
-  HTTPClientRequest,
-} from './lib/http-client'
+export * from './index.common'
 
-export type {
-  Plugin,
-  GroupTraits,
-  UserTraits,
-  TrackParams,
-  IdentifyParams,
-  AliasParams,
-  GroupParams,
-  PageParams,
-} from './app/types'
-export type { AnalyticsSettings } from './app/settings'
-
 // export Analytics as both a named export and a default export (for backwards-compat. reasons)
-import { Analytics } from './app/analytics-node'
+import { Analytics } from './index.common'
 export default Analytics

src/lib/abort.ts

@@ -44,3 +44,3 @@ /**
  */
-class AbortController {
+export class AbortController {
   signal = new AbortSignal()
@@ -47,0 +47,0 @@ abort() {

src/lib/http-client.ts

@@ -25,2 +25,12 @@ import { abortSignalAfterTimeout } from './abort'
 /**
+ * This interface is meant to be compatible with the Headers interface.
+ * @link https://developer.mozilla.org/en-US/docs/Web/API/Headers
+ */
+export interface HTTPHeaders {
+  get: (key: string) => string | null
+  has: (key: string) => boolean
+  entries: () => IterableIterator<[string, any]>
+}
+
+/**
  * This interface is meant to very minimally conform to the Response interface.
@@ -30,2 +40,4 @@ * @link https://developer.mozilla.org/en-US/docs/Web/API/Response
 export interface HTTPResponse {
+  headers?: Record<string, any> | HTTPHeaders
+  text?: () => Promise<string>
   status: number
@@ -54,6 +66,5 @@ statusText: string
   /**
-   * JSON data to be sent with the request (will be stringified)
-   * @example { "batch": [ ... ]}
+   * Data to be sent with the request
    */
-  data: Record<string, any>
+  body: string
   /**
@@ -90,3 +101,3 @@ * Specifies the timeout (in milliseconds) for an HTTP client to get an HTTP response from the server
       headers: options.headers,
-      body: JSON.stringify(options.data),
+      body: options.body,
       signal: signal,
@@ -93,0 +104,0 @@ }

src/plugins/segmentio/publisher.ts

@@ -7,4 +7,5 @@ import { backoff } from '@segment/analytics-core'
 import { NodeEmitter } from '../../app/emitter'
-import { b64encode } from '../../lib/base-64-encode'
 import { HTTPClient, HTTPClientRequest } from '../../lib/http-client'
+import { OAuthSettings } from '../../lib/types'
+import { TokenManager } from '../../lib/token-manager'
 
@@ -32,2 +33,3 @@ function sleep(timeoutInMs: number): Promise<void> {
   httpClient: HTTPClient
+  oauthSettings?: OAuthSettings
 }
@@ -45,3 +47,2 @@
   private _maxRetries: number
-  private _auth: string
   private _url: string
@@ -53,2 +54,4 @@ private _flushPendingItemsCount?: number
   private _httpClient: HTTPClient
+  private _writeKey: string
+  private _tokenManager: TokenManager | undefined
 
@@ -66,2 +69,3 @@ constructor(
       disable,
+      oauthSettings,
     }: PublisherProps,
@@ -74,3 +78,2 @@ emitter: NodeEmitter
     this._flushInterval = flushInterval
-    this._auth = b64encode(`${writeKey}:`)
     this._url = tryCreateFormattedUrl(
@@ -83,2 +86,11 @@ host ?? 'https://api.segment.io',
     this._httpClient = httpClient
+    this._writeKey = writeKey
+
+    if (oauthSettings) {
+      this._tokenManager = new TokenManager({
+        ...oauthSettings,
+        httpClient: oauthSettings.httpClient ?? httpClient,
+        maxRetries: oauthSettings.maxRetries ?? maxRetries,
+      })
+    }
   }
@@ -109,3 +121,6 @@
     if (!pendingItemsCount) {
-      // if number of pending items is 0, there is nothing to flush
+      // if number of pending items is 0, there will never be anything else entering the batch, since the app is closed.
+      if (this._tokenManager) {
+        this._tokenManager.stopPoller()
+      }
       return
@@ -123,3 +138,10 @@ }
     if (isExpectingNoMoreItems) {
-      this.send(this._batch).catch(noop)
+      this.send(this._batch)
+        .catch(noop)
+        .finally(() => {
+          // stop poller so program can exit ().
+          if (this._tokenManager) {
+            this._tokenManager.stopPoller()
+          }
+        })
       this.clearBatch()
@@ -211,11 +233,25 @@ }
 
+        let authString = undefined
+        if (this._tokenManager) {
+          const token = await this._tokenManager.getAccessToken()
+          if (token && token.access_token) {
+            authString = `Bearer ${token.access_token}`
+          }
+        }
+
+        const headers: Record<string, string> = {
+          'Content-Type': 'application/json',
+          'User-Agent': 'analytics-node-next/latest',
+          ...(authString ? { Authorization: authString } : {}),
+        }
+
         const request: HTTPClientRequest = {
           url: this._url,
           method: 'POST',
-          headers: {
-            'Content-Type': 'application/json',
-            Authorization: `Basic ${this._auth}`,
-            'User-Agent': 'analytics-node-next/latest',
-          },
-          data: { batch: events, sentAt: new Date() },
+          headers: headers,
+          body: JSON.stringify({
+            batch: events,
+            writeKey: this._writeKey,
+            sentAt: new Date(),
+          }),
           httpRequestTimeout: this._httpRequestTimeout,
@@ -225,3 +261,3 @@ }
         this._emitter.emit('http_request', {
-          body: request.data,
+          body: request.body,
           method: request.method,
@@ -238,2 +274,13 @@ url: request.url,
           return
+        } else if (
+          this._tokenManager &&
+          (response.status === 400 ||
+            response.status === 401 ||
+            response.status === 403)
+        ) {
+          // Retry with a new OAuth token if we have OAuth data
+          this._tokenManager.clearToken()
+          failureReason = new Error(
+            `[${response.status}] ${response.statusText}`
+          )
         } else if (response.status === 400) {
@@ -240,0 +287,0 @@ // https://segment.com/docs/connections/sources/catalog/libraries/server/http-api/#max-request-size

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

247904

increased by35.84%

Number of package files

192

increased by12.28%

Lines of code

4355

increased by36.01%

Number of lines in readme file

177

increased by17.22%

Worsened metrics

Dependency count

7

increased by16.67%

Dependency changes

• + Addedjose@^5.1.0

• + Addedjose@5.9.3(transitive)