feat: added destination metadata and internal timeline processing for destinatio plugins (segmentio#477)

Author: oscb

packages/core/src/plugin.ts

@@ -40,7 +40,7 @@ export class Plugin {
 }
 
 export class EventPlugin extends Plugin {
-  execute(event: SegmentEvent) {
+  execute(event: SegmentEvent): SegmentEvent | undefined {
     if (event === undefined) {
       return event;
     }
@@ -140,9 +140,33 @@ export class DestinationPlugin extends EventPlugin {
     this.timeline.remove(plugin);
   }
 
-  // find(pluginType: PluginType) {
-  //   // return this.timeline.find(pluginType);
-  // }
+  execute(event: SegmentEvent): SegmentEvent | undefined {
+    // Apply before and enrichment plugins
+    const beforeResult = this.timeline.applyPlugins({
+      type: PluginType.before,
+      event,
+    });
+
+    if (beforeResult === undefined) {
+      return;
+    }
+
+    const enrichmentResult = this.timeline.applyPlugins({
+      type: PluginType.enrichment,
+      event: beforeResult,
+    });
+
+    // Now send the event to the destination by executing the normal flow of an EventPlugin
+    super.execute(enrichmentResult);
+
+    // apply .after plugins
+    let afterResult = this.timeline.applyPlugins({
+      type: PluginType.after,
+      event: enrichmentResult,
+    });
+
+    return afterResult;
+  }
 }
 
 export class UtilityPlugin extends EventPlugin {}

packages/core/src/plugins/DestinationMetadataEnrichment.ts

@@ -0,0 +1,54 @@
+import { DestinationPlugin, UtilityPlugin } from '../plugin';
+import { PluginType, SegmentEvent } from '../types';
+import { SEGMENT_DESTINATION_KEY } from './SegmentDestination';
+
+export class DestinationMetadataEnrichment extends UtilityPlugin {
+  type = PluginType.enrichment;
+
+  execute(event: SegmentEvent): SegmentEvent {
+    const pluginSettings = this.analytics?.settings.get();
+    const plugins = this.analytics?.getPlugins(PluginType.destination);
+
+    if (pluginSettings === undefined) {
+      return event;
+    }
+
+    // Disable all destinations that have a device mode plugin
+    const destinations =
+      plugins?.map((plugin) => (plugin as DestinationPlugin).key) ?? [];
+    const bundled: string[] = [];
+
+    for (const key of destinations) {
+      if (key === SEGMENT_DESTINATION_KEY) {
+        continue;
+      }
+
+      if (key in pluginSettings) {
+        bundled.push(key);
+      }
+    }
+
+    const unbundled: string[] = [];
+    const segmentInfo =
+      (pluginSettings[SEGMENT_DESTINATION_KEY] as Record<string, any>) ?? {};
+    const unbundledIntegrations: string[] =
+      segmentInfo.unbundledIntegrations ?? [];
+
+    for (const integration of unbundledIntegrations) {
+      if (!(integration in bundled)) {
+        unbundled.push(integration);
+      }
+    }
+
+    // User/event defined integrations override the cloud/device mode merge
+    const enrichedEvent: SegmentEvent = {
+      ...event,
+      _metadata: {
+        bundled,
+        unbundled,
+        bundledIds: [],
+      },
+    };
+    return enrichedEvent;
+  }
+}

packages/core/src/plugins/SegmentDestination.ts

@@ -1,53 +1,31 @@
 import { DestinationPlugin } from '../plugin';
-import {
-  PluginType,
-  SegmentAPIIntegrations,
-  SegmentAPISettings,
-  SegmentEvent,
-  UpdateType,
-} from '../types';
+import { PluginType, SegmentEvent } from '../types';
 import { chunk } from '../util';
 import { sendEvents } from '../api';
+import type { SegmentClient } from '../analytics';
+import { DestinationMetadataEnrichment } from './DestinationMetadataEnrichment';
 
 const MAX_EVENTS_PER_BATCH = 100;
+export const SEGMENT_DESTINATION_KEY = 'Segment.io';
 
 export class SegmentDestination extends DestinationPlugin {
   type = PluginType.destination;
 
-  key = 'Segment.io';
+  key = SEGMENT_DESTINATION_KEY;
 
-  update(_: SegmentAPISettings, __: UpdateType) {
-    // this is where analytics-swift initalizes the HTTP client
-    // no need to do this for React Native where we just use the fetch polyfill directly
-    // see flush() below
-  }
+  configure(analytics: SegmentClient): void {
+    super.configure(analytics);
 
-  execute(event: SegmentEvent): SegmentEvent {
-    const pluginSettings = this.analytics?.settings.get();
-    const plugins = this.analytics?.getPlugins(PluginType.destination);
+    // Enrich events with the Destination metadata
+    this.add(new DestinationMetadataEnrichment());
+  }
 
-    // Disable all destinations that have a device mode plugin
-    const deviceModePlugins =
-      plugins?.map((plugin) => (plugin as DestinationPlugin).key) ?? [];
-    const disabledCloudIntegrations: SegmentAPIIntegrations = {};
-    if (pluginSettings !== undefined) {
-      for (const key of deviceModePlugins) {
-        if (key in pluginSettings) {
-          disabledCloudIntegrations[key] = false;
-        }
-      }
+  execute(event: SegmentEvent): SegmentEvent | undefined {
+    const enrichedEvent = super.execute(event);
+    if (enrichedEvent !== undefined) {
+      this.analytics?.queueEvent(enrichedEvent);
     }
-
-    // User/event defined integrations override the cloud/device mode merge
-    const mergedEvent = {
-      ...event,
-      integrations: {
-        ...disabledCloudIntegrations,
-        ...event?.integrations,
-      },
-    };
-    this.analytics?.queueEvent(mergedEvent);
-    return mergedEvent;
+    return enrichedEvent;
   }
 
   async flush() {

packages/core/src/plugins/__tests__/SegmentDestination.test.ts

@@ -1,5 +1,8 @@
 import { EventType, SegmentEvent, TrackEventType } from '../../types';
-import { SegmentDestination } from '../SegmentDestination';
+import {
+  SegmentDestination,
+  SEGMENT_DESTINATION_KEY,
+} from '../SegmentDestination';
 import { SegmentClient } from '../../analytics';
 import { MockSegmentStore } from '../../__tests__/__helpers__/mockSegmentStore';
 import { getMockLogger } from '../../__tests__/__helpers__/mockLogger';
@@ -43,7 +46,7 @@ describe('SegmentDestination', () => {
 
   it('disables device mode plugins to prevent dups', () => {
     const plugin = new SegmentDestination();
-    plugin.analytics = new SegmentClient({
+    const analytics = new SegmentClient({
       ...clientArgs,
       store: new MockSegmentStore({
         settings: {
@@ -53,8 +56,9 @@ describe('SegmentDestination', () => {
         },
       }),
     });
+    plugin.configure(analytics);
 
-    plugin.analytics.getPlugins = jest.fn().mockReturnValue([
+    plugin.analytics!.getPlugins = jest.fn().mockReturnValue([
       {
         key: 'firebase',
         type: 'destination',
@@ -72,14 +76,57 @@ describe('SegmentDestination', () => {
       integrations: {},
     };
 
-    const expectedIntegrations = {
-      firebase: false,
+    const result = plugin.execute(event);
+    expect(result).toEqual({
+      ...event,
+      _metadata: {
+        bundled: ['firebase'],
+        unbundled: [],
+        bundledIds: [],
+      },
+    });
+  });
+
+  it('marks unbundled plugins where the cloud mode is disabled', () => {
+    const plugin = new SegmentDestination();
+    const analytics = new SegmentClient({
+      ...clientArgs,
+      store: new MockSegmentStore({
+        settings: {
+          [SEGMENT_DESTINATION_KEY]: {
+            unbundledIntegrations: ['firebase'],
+          },
+        },
+      }),
+    });
+    plugin.configure(analytics);
+
+    plugin.analytics!.getPlugins = jest.fn().mockReturnValue([
+      {
+        key: 'firebase',
+        type: 'destination',
+      },
+    ]);
+
+    const event: TrackEventType = {
+      anonymousId: '3534a492-e975-4efa-a18b-3c70c562fec2',
+      event: 'Awesome event',
+      type: EventType.TrackEvent,
+      properties: {},
+      timestamp: '2000-01-01T00:00:00.000Z',
+      messageId: '1d1744bf-5beb-41ac-ad7a-943eac33babc',
+      context: { app: { name: 'TestApp' } },
+      integrations: {},
     };
 
     const result = plugin.execute(event);
     expect(result).toEqual({
       ...event,
-      integrations: expectedIntegrations,
+      _metadata: {
+        bundled: [],
+        unbundled: ['firebase'],
+        bundledIds: [],
+      },
     });
   });
 

packages/core/src/types.ts

@@ -27,6 +27,7 @@ interface BaseEventType {
 
   context?: PartialContext;
   integrations?: SegmentAPIIntegrations;
+  _metadata?: DestinationMetadata;
 }
 
 export interface TrackEventType extends BaseEventType {
@@ -256,6 +257,12 @@ export type SegmentAPISettings = {
   integrations: SegmentAPIIntegrations;
 };
 
+export type DestinationMetadata = {
+  bundled: string[];
+  unbundled: string[];
+  bundledIds: string[];
+};
+
 export enum PluginType {
   // Executed before event processing begins.
   'before' = 'before',