fix: apply userInfo on processing to prevent concurrency issues (#660)

packages/core/src/__tests__/__helpers__/mockSegmentStore.ts

@@ -16,15 +16,15 @@ import type {
 import { createCallbackManager } from './utils';
 import { createGetter } from '../../storage/helpers';
 
-type Data = {
+export type StoreData = {
   isReady: boolean;
   context?: DeepPartial<Context>;
   settings: SegmentAPIIntegrations;
   userInfo: UserInfoState;
   deepLinkData: DeepLinkData;
 };
 
-const INITIAL_VALUES: Data = {
+const INITIAL_VALUES: StoreData = {
   isReady: true,
   context: undefined,
   settings: {
@@ -50,14 +50,14 @@ export function createMockStoreGetter<T>(fn: () => T) {
 }
 
 export class MockSegmentStore implements Storage {
-  private data: Data;
-  private initialData: Data;
+  private data: StoreData;
+  private initialData: StoreData;
 
   reset = () => {
     this.data = JSON.parse(JSON.stringify(this.initialData));
   };
 
-  constructor(initialData?: Partial<Data>) {
+  constructor(initialData?: Partial<StoreData>) {
     this.data = { ...INITIAL_VALUES, ...initialData };
     this.initialData = JSON.parse(
       JSON.stringify({ ...INITIAL_VALUES, ...initialData })

packages/core/src/__tests__/__helpers__/setupSegmentClient.ts

@@ -0,0 +1,49 @@
+import { SegmentClient } from '../../analytics';
+import { UtilityPlugin } from '../../plugin';
+import { PluginType, SegmentEvent } from '../../types';
+import { getMockLogger } from './mockLogger';
+import { MockSegmentStore, StoreData } from './mockSegmentStore';
+
+jest.mock('../../uuid');
+
+jest
+  .spyOn(Date.prototype, 'toISOString')
+  .mockReturnValue('2010-01-01T00:00:00.000Z');
+
+export const createTestClient = (storeData?: Partial<StoreData>) => {
+  const store = new MockSegmentStore({
+    isReady: true,
+    ...storeData,
+  });
+
+  const clientArgs = {
+    config: {
+      writeKey: 'mock-write-key',
+      autoAddSegmentDestination: false,
+    },
+    logger: getMockLogger(),
+    store: store,
+  };
+
+  const client = new SegmentClient(clientArgs);
+
+  class ObservablePlugin extends UtilityPlugin {
+    type = PluginType.after;
+  }
+
+  const mockPlugin = new ObservablePlugin();
+  jest.spyOn(mockPlugin, 'execute');
+
+  client.add({ plugin: mockPlugin });
+
+  return {
+    client,
+    store,
+    plugin: mockPlugin as UtilityPlugin,
+    expectEvent: (event: Partial<SegmentEvent>) => {
+      return expect(mockPlugin.execute).toHaveBeenCalledWith(
+        expect.objectContaining(event)
+      );
+    },
+  };
+};

packages/core/src/__tests__/methods/identify.test.ts

@@ -1,12 +1,5 @@
-import { SegmentClient } from '../../analytics';
-import { getMockLogger } from '../__helpers__/mockLogger';
-import { MockSegmentStore } from '../__helpers__/mockSegmentStore';
-
-jest.mock('../../uuid');
-
-jest
-  .spyOn(Date.prototype, 'toISOString')
-  .mockReturnValue('2010-01-01T00:00:00.000Z');
+import { EventType, SegmentEvent } from '../../types';
+import { createTestClient } from '../__helpers__/setupSegmentClient';
 
 describe('methods #identify', () => {
   const initialUserInfo = {
@@ -17,40 +10,28 @@ describe('methods #identify', () => {
     userId: 'current-user-id',
     anonymousId: 'very-anonymous',
   };
-  const store = new MockSegmentStore({
+  const { store, client, expectEvent } = createTestClient({
     userInfo: initialUserInfo,
   });
 
-  const clientArgs = {
-    config: {
-      writeKey: 'mock-write-key',
-    },
-    logger: getMockLogger(),
-    store: store,
-  };
-
   beforeEach(() => {
     store.reset();
     jest.clearAllMocks();
   });
 
   it('adds the identify event correctly', async () => {
-    const client = new SegmentClient(clientArgs);
-    jest.spyOn(client, 'process');
-
     await client.identify('new-user-id', { name: 'Mary', age: 30 });
 
-    const expectedEvent = {
+    const expectedEvent: Partial<SegmentEvent> = {
       traits: {
         name: 'Mary',
         age: 30,
       },
       userId: 'new-user-id',
-      type: 'identify',
+      type: EventType.IdentifyEvent,
     };
 
-    expect(client.process).toHaveBeenCalledTimes(1);
-    expect(client.process).toHaveBeenCalledWith(expectedEvent);
+    expectEvent(expectedEvent);
 
     expect(client.userInfo.get()).toEqual({
       ...initialUserInfo,
@@ -60,19 +41,15 @@ describe('methods #identify', () => {
   });
 
   it('does not update user traits when there are no new ones provided', async () => {
-    const client = new SegmentClient(clientArgs);
-    jest.spyOn(client, 'process');
-
     await client.identify('new-user-id');
 
     const expectedEvent = {
       traits: initialUserInfo.traits,
       userId: 'new-user-id',
-      type: 'identify',
+      type: EventType.IdentifyEvent,
     };
 
-    expect(client.process).toHaveBeenCalledTimes(1);
-    expect(client.process).toHaveBeenCalledWith(expectedEvent);
+    expectEvent(expectedEvent);
 
     expect(client.userInfo.get()).toEqual({
       ...initialUserInfo,
@@ -81,19 +58,17 @@ describe('methods #identify', () => {
   });
 
   it('does not update userId when userId is undefined', async () => {
-    const client = new SegmentClient(clientArgs);
     jest.spyOn(client, 'process');
 
     await client.identify(undefined, { name: 'Mary' });
 
     const expectedEvent = {
       traits: { name: 'Mary', age: 30 },
       userId: 'current-user-id',
-      type: 'identify',
+      type: EventType.IdentifyEvent,
     };
 
-    expect(client.process).toHaveBeenCalledTimes(1);
-    expect(client.process).toHaveBeenCalledWith(expectedEvent);
+    expectEvent(expectedEvent);
     expect(client.userInfo.get()).toEqual({
       ...initialUserInfo,
       traits: {
@@ -104,45 +79,34 @@ describe('methods #identify', () => {
   });
 
   it('does not persist identity traits accross events', async () => {
-    const client = new SegmentClient(clientArgs);
-    jest.spyOn(client, 'process');
-    // @ts-ignore accessing the internal timeline to check the processed events
-    jest.spyOn(client.timeline, 'process');
-
     await client.identify('new-user-id', { name: 'Mary', age: 30 });
 
-    const expectedEvent = {
+    const expectedEvent: Partial<SegmentEvent> = {
       traits: {
         name: 'Mary',
         age: 30,
       },
       userId: 'new-user-id',
-      type: 'identify',
+      type: EventType.IdentifyEvent,
     };
-
-    expect(client.process).toHaveBeenCalledTimes(1);
-    expect(client.process).toHaveBeenCalledWith(expectedEvent);
+    expectEvent(expectedEvent);
 
     expect(client.userInfo.get()).toEqual({
       ...initialUserInfo,
       userId: 'new-user-id',
       traits: expectedEvent.traits,
     });
 
-    client.track('track event');
-
-    // Await all promises
-    await new Promise(process.nextTick);
+    await client.track('track event');
 
-    // @ts-ignore
-    expect(client.timeline.process).toHaveBeenLastCalledWith({
+    expectEvent({
       anonymousId: 'very-anonymous',
       event: 'track event',
       integrations: {},
       messageId: 'mocked-uuid',
       properties: {},
       timestamp: '2010-01-01T00:00:00.000Z',
-      type: 'track',
+      type: EventType.TrackEvent,
       userId: 'new-user-id',
     });
   });

packages/core/src/analytics.ts

@@ -15,7 +15,8 @@ import {
 } from './events';
 import type { Logger } from './logger';
 import type { DestinationPlugin, PlatformPlugin, Plugin } from './plugin';
-import { InjectContext } from './plugins/Context';
+import { InjectContext } from './plugins/InjectContext';
+import { InjectUserInfo } from './plugins/InjectUserInfo';
 import { SegmentDestination } from './plugins/SegmentDestination';
 import {
   createGetter,
@@ -83,20 +84,10 @@ export class SegmentClient {
 
   private onContextLoadedCallback: OnContextLoadCallback | undefined;
 
-  get platformPlugins() {
-    const plugins: PlatformPlugin[] = [];
-
-    // add context plugin as well as it's platform specific internally.
-    // this must come first.
-    plugins.push(new InjectContext());
-
-    // setup lifecycle if desired
-    if (this.config.trackAppLifecycleEvents) {
-      // todo: more plugins!
-    }
-
-    return plugins;
-  }
+  private readonly platformPlugins: PlatformPlugin[] = [
+    new InjectUserInfo(),
+    new InjectContext(),
+  ];
 
   // Watchables
   /**
@@ -118,7 +109,7 @@ export class SegmentClient {
   /**
    * Access or subscribe to user info (anonymousId, userId, traits)
    */
-  readonly userInfo: Watchable<UserInfoState>;
+  readonly userInfo: Watchable<UserInfoState> & Settable<UserInfoState>;
 
   readonly deepLinkData: Watchable<DeepLinkData>;
 
@@ -192,6 +183,7 @@ export class SegmentClient {
 
     this.userInfo = {
       get: this.store.userInfo.get,
+      set: this.store.userInfo.set,
       onChange: this.store.userInfo.onChange,
     };
 
@@ -386,12 +378,12 @@ export class SegmentClient {
   }
 
   async process(incomingEvent: SegmentEvent) {
-    const userData = await this.store.userInfo.get(true);
-    const event = applyRawEventData(incomingEvent, userData);
+    const event = applyRawEventData(incomingEvent);
     if (this.store.isReady.get() === true) {
-      this.timeline.process(event);
+      return this.timeline.process(event);
     } else {
       this.pendingEvents.push(event);
+      return event;
     }
   }
 
@@ -492,18 +484,9 @@ export class SegmentClient {
   }
 
   async identify(userId?: string, userTraits?: UserTraits) {
-    const userData = await this.store.userInfo.set((state) => ({
-      ...state,
-      userId: userId ?? state.userId,
-      traits: {
-        ...state.traits,
-        ...userTraits,
-      },
-    }));
-
     const event = createIdentifyEvent({
-      userId: userData.userId,
-      userTraits: userData.traits,
+      userId: userId,
+      userTraits: userTraits,
     });
 
     await this.process(event);
@@ -524,11 +507,6 @@ export class SegmentClient {
     const { anonymousId, userId: previousUserId } =
       await this.store.userInfo.get(true);
 
-    await this.store.userInfo.set((state) => ({
-      ...state,
-      userId: newUserId,
-    }));
-
     const event = createAliasEvent({
       anonymousId,
       userId: previousUserId,

packages/core/src/events.ts

@@ -11,7 +11,6 @@ import {
   AliasEventType,
   EventType,
   SegmentEvent,
-  UserInfoState,
 } from './types';
 
 export const createTrackEvent = ({
@@ -78,24 +77,11 @@ export const createAliasEvent = ({
   previousId: userId || anonymousId,
 });
 
-const isAliasEvent = (event: SegmentEvent): event is AliasEventType =>
-  event.type === EventType.AliasEvent;
-const isIdentifyEvent = (event: SegmentEvent): event is AliasEventType =>
-  event.type === EventType.IdentifyEvent;
-
-export const applyRawEventData = (
-  event: SegmentEvent,
-  userInfo: UserInfoState
-): SegmentEvent => {
+export const applyRawEventData = (event: SegmentEvent): SegmentEvent => {
   return {
     ...event,
-    anonymousId: userInfo.anonymousId,
     messageId: getUUID(),
     timestamp: new Date().toISOString(),
     integrations: event.integrations ?? {},
-    userId:
-      isAliasEvent(event) || isIdentifyEvent(event)
-        ? event.userId
-        : userInfo.userId,
   };
 };