feat: Add support for Payload Filtering (#551)

Author: keelerm84

contract-tests/index.js

@@ -27,6 +27,8 @@ app.get('/', (req, res) => {
       'all-flags-with-reasons',
       'tags',
       'big-segments',
+      'filtering',
+      'filtering-strict',
       'user-type',
       'migrations',
       'event-sampling',

contract-tests/sdkClientEntity.js

@@ -26,11 +26,17 @@ export function makeSdkConfig(options, tag) {
   if (options.streaming) {
     cf.streamUri = options.streaming.baseUri;
     cf.streamInitialReconnectDelay = maybeTime(options.streaming.initialRetryDelayMs);
+    if (options.streaming.filter) {
+      cf.payloadFilterKey = options.streaming.filter;
+    }
   }
   if (options.polling) {
     cf.stream = false;
     cf.baseUri = options.polling.baseUri;
     cf.pollInterface = options.polling.pollIntervalMs / 1000;
+    if (options.polling.filter) {
+      cf.payloadFilterKey = options.polling.filter;
+    }
   }
   if (options.events) {
     cf.allAttributesPrivate = options.events.allAttributesPrivate;

packages/shared/common/__tests__/options/ServiceEndpoints.test.ts

@@ -1,4 +1,8 @@
-import ServiceEndpoints from '../../src/options/ServiceEndpoints';
+import ServiceEndpoints, {
+  getEventsUri,
+  getPollingUri,
+  getStreamingUri,
+} from '../../src/options/ServiceEndpoints';
 
 describe.each([
   [
@@ -33,3 +37,26 @@ describe.each([
     expect(endpoints.events).toEqual(expected.eventsUri);
   });
 });
+
+it('applies payload filter to polling and streaming endpoints', () => {
+  const endpoints = new ServiceEndpoints(
+    'https://stream.launchdarkly.com',
+    'https://sdk.launchdarkly.com',
+    'https://events.launchdarkly.com',
+    '/bulk',
+    '/diagnostic',
+    true,
+    'filterKey',
+  );
+
+  expect(getStreamingUri(endpoints, '/all', [])).toEqual(
+    'https://stream.launchdarkly.com/all?filter=filterKey',
+  );
+  expect(getPollingUri(endpoints, '/sdk/latest-all', [])).toEqual(
+    'https://sdk.launchdarkly.com/sdk/latest-all?filter=filterKey',
+  );
+  expect(
+    getPollingUri(endpoints, '/sdk/latest-all', [{ key: 'withReasons', value: 'true' }]),
+  ).toEqual('https://sdk.launchdarkly.com/sdk/latest-all?withReasons=true&filter=filterKey');
+  expect(getEventsUri(endpoints, '/bulk', [])).toEqual('https://events.launchdarkly.com/bulk');
+});

packages/shared/common/src/internal/events/EventSender.ts

@@ -10,7 +10,7 @@ import {
   isHttpRecoverable,
   LDUnexpectedResponseError,
 } from '../../errors';
-import { ClientContext } from '../../options';
+import { ClientContext, getEventsUri } from '../../options';
 import { defaultHeaders, httpErrorMessage, sleep } from '../../utils';
 
 export default class EventSender implements LDEventSender {
@@ -26,19 +26,18 @@ export default class EventSender implements LDEventSender {
     const { basicConfiguration, platform } = clientContext;
     const {
       sdkKey,
-      serviceEndpoints: {
-        events,
-        analyticsEventPath,
-        diagnosticEventPath,
-        includeAuthorizationHeader,
-      },
+      serviceEndpoints: { analyticsEventPath, diagnosticEventPath, includeAuthorizationHeader },
       tags,
     } = basicConfiguration;
     const { crypto, info, requests } = platform;
 
     this.defaultHeaders = defaultHeaders(sdkKey, info, tags, includeAuthorizationHeader);
-    this.eventsUri = `${events}${analyticsEventPath}`;
-    this.diagnosticEventsUri = `${events}${diagnosticEventPath}`;
+    this.eventsUri = getEventsUri(basicConfiguration.serviceEndpoints, analyticsEventPath, []);
+    this.diagnosticEventsUri = getEventsUri(
+      basicConfiguration.serviceEndpoints,
+      diagnosticEventPath,
+      [],
+    );
     this.requests = requests;
     this.crypto = crypto;
   }

packages/shared/common/src/internal/stream/StreamingProcessor.test.ts

@@ -107,6 +107,7 @@ describe('given a stream processor with mock event source', () => {
         platform: basicPlatform,
       },
       '/all',
+      [],
       listeners,
       diagnosticsManager,
       mockErrorHandler,
@@ -142,6 +143,7 @@ describe('given a stream processor with mock event source', () => {
         platform: basicPlatform,
       },
       '/all',
+      [],
       listeners,
       diagnosticsManager,
       mockErrorHandler,

packages/shared/common/src/internal/stream/StreamingProcessor.ts

@@ -9,6 +9,7 @@ import {
 import { LDStreamProcessor } from '../../api/subsystem';
 import { LDStreamingError } from '../../errors';
 import { ClientContext } from '../../options';
+import { getStreamingUri } from '../../options/ServiceEndpoints';
 import { defaultHeaders, httpErrorMessage, shouldRetry } from '../../utils';
 import { DiagnosticsManager } from '../diagnostics';
 import { StreamingErrorHandler } from './types';
@@ -37,6 +38,7 @@ class StreamingProcessor implements LDStreamProcessor {
     sdkKey: string,
     clientContext: ClientContext,
     streamUriPath: string,
+    parameters: { key: string; value: string }[],
     private readonly listeners: Map<EventName, ProcessStreamResponse>,
     private readonly diagnosticsManager?: DiagnosticsManager,
     private readonly errorHandler?: StreamingErrorHandler,
@@ -49,7 +51,11 @@ class StreamingProcessor implements LDStreamProcessor {
     this.headers = defaultHeaders(sdkKey, info, tags);
     this.logger = logger;
     this.requests = requests;
-    this.streamUri = `${basicConfiguration.serviceEndpoints.streaming}${streamUriPath}`;
+    this.streamUri = getStreamingUri(
+      basicConfiguration.serviceEndpoints,
+      streamUriPath,
+      parameters,
+    );
   }
 
   private logConnectionStarted() {

packages/shared/common/src/options/ServiceEndpoints.ts

@@ -2,6 +2,10 @@ function canonicalizeUri(uri: string): string {
   return uri.replace(/\/+$/, '');
 }
 
+function canonicalizePath(path: string): string {
+  return path.replace(/^\/+/, '').replace(/\?$/, '');
+}
+
 /**
  * Specifies the base service URIs used by SDK components.
  */
@@ -11,6 +15,7 @@ export default class ServiceEndpoints {
   public readonly streaming: string;
   public readonly polling: string;
   public readonly events: string;
+  public readonly payloadFilterKey?: string;
 
   /** Valid paths are:
    * /bulk
@@ -36,12 +41,84 @@ export default class ServiceEndpoints {
     analyticsEventPath: string = '/bulk',
     diagnosticEventPath: string = '/diagnostic',
     includeAuthorizationHeader: boolean = true,
+    payloadFilterKey?: string,
   ) {
     this.streaming = canonicalizeUri(streaming);
     this.polling = canonicalizeUri(polling);
     this.events = canonicalizeUri(events);
     this.analyticsEventPath = analyticsEventPath;
     this.diagnosticEventPath = diagnosticEventPath;
     this.includeAuthorizationHeader = includeAuthorizationHeader;
+    this.payloadFilterKey = payloadFilterKey;
   }
 }
+
+function getWithParams(uri: string, parameters: { key: string; value: string }[]) {
+  if (parameters.length === 0) {
+    return uri;
+  }
+
+  const parts = parameters.map(({ key, value }) => `${key}=${value}`);
+  return `${uri}?${parts.join('&')}`;
+}
+
+/**
+ * Get the URI for the streaming endpoint.
+ *
+ * @param endpoints The service endpoints.
+ * @param path The path to the resource, devoid of any query parameters or hrefs.
+ * @param parameters The query parameters. These query parameters must already have the appropriate encoding applied. This function WILL NOT apply it for you.
+ */
+export function getStreamingUri(
+  endpoints: ServiceEndpoints,
+  path: string,
+  parameters: { key: string; value: string }[],
+): string {
+  const canonicalizedPath = canonicalizePath(path);
+
+  const combinedParameters = [...parameters];
+  if (endpoints.payloadFilterKey) {
+    combinedParameters.push({ key: 'filter', value: endpoints.payloadFilterKey });
+  }
+
+  return getWithParams(`${endpoints.streaming}/${canonicalizedPath}`, combinedParameters);
+}
+
+/**
+ * Get the URI for the polling endpoint.
+ *
+ * @param endpoints The service endpoints.
+ * @param path The path to the resource, devoid of any query parameters or hrefs.
+ * @param parameters The query parameters. These query parameters must already have the appropriate encoding applied. This function WILL NOT apply it for you.
+ */
+export function getPollingUri(
+  endpoints: ServiceEndpoints,
+  path: string,
+  parameters: { key: string; value: string }[],
+): string {
+  const canonicalizedPath = canonicalizePath(path);
+
+  const combinedParameters = [...parameters];
+  if (endpoints.payloadFilterKey) {
+    combinedParameters.push({ key: 'filter', value: endpoints.payloadFilterKey });
+  }
+
+  return getWithParams(`${endpoints.polling}/${canonicalizedPath}`, combinedParameters);
+}
+
+/**
+ * Get the URI for the events endpoint.
+ *
+ * @param endpoints The service endpoints.
+ * @param path The path to the resource, devoid of any query parameters or hrefs.
+ * @param parameters The query parameters. These query parameters must already have the appropriate encoding applied. This function WILL NOT apply it for you.
+ */
+export function getEventsUri(
+  endpoints: ServiceEndpoints,
+  path: string,
+  parameters: { key: string; value: string }[],
+): string {
+  const canonicalizedPath = canonicalizePath(path);
+
+  return getWithParams(`${endpoints.events}/${canonicalizedPath}`, parameters);
+}

packages/shared/common/src/options/index.ts

@@ -1,6 +1,14 @@
 import ApplicationTags from './ApplicationTags';
 import ClientContext from './ClientContext';
 import OptionMessages from './OptionMessages';
-import ServiceEndpoints from './ServiceEndpoints';
+import ServiceEndpoints, { getEventsUri, getPollingUri, getStreamingUri } from './ServiceEndpoints';
 
-export { ApplicationTags, OptionMessages, ServiceEndpoints, ClientContext };
+export {
+  ApplicationTags,
+  OptionMessages,
+  ServiceEndpoints,
+  ClientContext,
+  getStreamingUri,
+  getPollingUri,
+  getEventsUri,
+};

packages/shared/common/src/validators.ts

@@ -118,7 +118,7 @@ export class StringMatchingRegex extends Type<string> {
   }
 
   override is(u: unknown): u is string {
-    return !!(u as string).match(this.expression);
+    return typeof u === 'string' && !!(u as string).match(this.expression);
   }
 }
 

packages/shared/mocks/src/streamingProcessor.ts

@@ -25,6 +25,7 @@ export const setupMockStreamingProcessor = (
       sdkKey: string,
       clientContext: ClientContext,
       streamUriPath: string,
+      parameters: { key: string; value: string }[],
       listeners: Map<EventName, ProcessStreamResponse>,
       diagnosticsManager: internal.DiagnosticsManager,
       errorHandler: internal.StreamingErrorHandler,