feat: Add support for per-context summary events. (#126)

Author: kinyoklion

src/EventProcessor.js

@@ -1,5 +1,5 @@
 const EventSender = require('./EventSender');
-const EventSummarizer = require('./EventSummarizer');
+const MultiEventSummarizer = require('./MultiEventSummarizer');
 const ContextFilter = require('./ContextFilter');
 const errors = require('./errors');
 const messages = require('./messages');
@@ -17,8 +17,8 @@ function EventProcessor(
   const processor = {};
   const eventSender = sender || EventSender(platform, environmentId, options);
   const mainEventsUrl = utils.appendUrlPath(options.eventsUrl, '/events/bulk/' + environmentId);
-  const summarizer = EventSummarizer();
   const contextFilter = ContextFilter(options);
+  const summarizer = MultiEventSummarizer(contextFilter);
   const samplingInterval = options.samplingInterval;
   const eventCapacity = options.eventCapacity;
   const flushInterval = options.flushInterval;
@@ -117,17 +117,19 @@ function EventProcessor(
     }
   };
 
-  processor.flush = function() {
+  processor.flush = async function() {
     if (disabled) {
       return Promise.resolve();
     }
     const eventsToSend = queue;
-    const summary = summarizer.getSummary();
-    summarizer.clearSummary();
-    if (summary) {
-      summary.kind = 'summary';
-      eventsToSend.push(summary);
-    }
+    const summaries = summarizer.getSummaries();
+
+    summaries.forEach(summary => {
+      if (Object.keys(summary.features).length) {
+        eventsToSend.push(summary);
+      }
+    });
+
     if (diagnosticsAccumulator) {
       // For diagnostic events, we record how many events were in the queue at the last flush (since "how
       // many events happened to be in the queue at the moment we decided to send a diagnostic event" would

src/EventSummarizer.js

@@ -89,6 +89,7 @@ function EventSummarizer() {
           startDate,
           endDate,
           features: flagsOut,
+          kind: 'summary',
         };
   };
 

src/MultiEventSummarizer.js

@@ -0,0 +1,60 @@
+const canonicalize = require('./canonicalize');
+const EventSummarizer = require('./EventSummarizer');
+
+/**
+ * Construct a multi-event summarizer. This summarizer produces a summary event for each unique context.
+ * @param {{filter: (context: any) => any}} contextFilter
+ */
+function MultiEventSummarizer(contextFilter) {
+  let summarizers = {};
+  let contexts = {};
+
+  /**
+   * Summarize the given event.
+   * @param {{
+   *   kind: string,
+   *   context?: any,
+   * }} event
+   */
+  function summarizeEvent(event) {
+    if (event.kind === 'feature') {
+      const key = canonicalize(event.context);
+      if (!key) {
+        return;
+      }
+
+      let summarizer = summarizers[key];
+      if (!summarizer) {
+        summarizers[key] = EventSummarizer();
+        summarizer = summarizers[key];
+        contexts[key] = event.context;
+      }
+
+      summarizer.summarizeEvent(event);
+    }
+  }
+
+  /**
+   * Get the summaries of the events that have been summarized.
+   * @returns {any[]}
+   */
+  function getSummaries() {
+    const summarizersToFlush = summarizers;
+    const contextsForSummaries = contexts;
+
+    summarizers = {};
+    contexts = {};
+    return Object.entries(summarizersToFlush).map(([key, summarizer]) => {
+      const summary = summarizer.getSummary();
+      summary.context = contextFilter.filter(contextsForSummaries[key]);
+      return summary;
+    });
+  }
+
+  return {
+    summarizeEvent,
+    getSummaries,
+  };
+}
+
+module.exports = MultiEventSummarizer;

src/__tests__/EventProcessor-test.js

@@ -485,6 +485,73 @@ describe.each([
     });
   });
 
+  it('generates separate summary events for different contexts', async () => {
+    await withProcessorAndSender(defaultConfig, async (ep, mockEventSender) => {
+      const context1 = { key: 'user1', kind: 'user' };
+      const context2 = { key: 'user2', kind: 'user' };
+
+      // Create feature events for two different contexts
+      const event1 = {
+        kind: 'feature',
+        creationDate: 1000,
+        context: context1,
+        key: 'flag1',
+        version: 11,
+        variation: 1,
+        value: 'value1',
+        default: 'default1',
+        trackEvents: false,
+      };
+
+      const event2 = {
+        kind: 'feature',
+        creationDate: 1000,
+        context: context2,
+        key: 'flag2',
+        version: 22,
+        variation: 2,
+        value: 'value2',
+        default: 'default2',
+        trackEvents: false,
+      };
+
+      ep.enqueue(event1);
+      ep.enqueue(event2);
+      await ep.flush();
+
+      expect(mockEventSender.calls.length()).toEqual(1);
+      const output = (await mockEventSender.calls.take()).events;
+
+      // Should have two summary events, one for each context
+      expect(output.length).toEqual(2);
+
+      // Find the summary event for each context
+      const summary1 = output.find(e => e.context.key === 'user1');
+      const summary2 = output.find(e => e.context.key === 'user2');
+
+      // Verify each summary event has the correct context and flag data
+      expect(summary1).toBeDefined();
+      expect(summary1.context).toEqual(context1);
+      expect(summary1.features).toEqual({
+        flag1: {
+          contextKinds: ['user'],
+          default: 'default1',
+          counters: [{ version: 11, variation: 1, value: 'value1', count: 1 }],
+        },
+      });
+
+      expect(summary2).toBeDefined();
+      expect(summary2.context).toEqual(context2);
+      expect(summary2.features).toEqual({
+        flag2: {
+          contextKinds: ['user'],
+          default: 'default2',
+          counters: [{ version: 22, variation: 2, value: 'value2', count: 1 }],
+        },
+      });
+    });
+  });
+
   describe('interaction with diagnostic events', () => {
     it('sets eventsInLastBatch on flush', async () => {
       const e0 = { kind: 'custom', creationDate: 1000, context: eventContext, key: 'key0' };
@@ -525,5 +592,50 @@ describe.each([
         expect(diagnosticAccumulator.getProps().droppedEvents).toEqual(2);
       });
     });
+
+    it('filters context in summary events', async () => {
+      const event = {
+        kind: 'feature',
+        creationDate: 1000,
+        context: eventContext,
+        key: 'flagkey',
+        version: 11,
+        variation: 1,
+        value: 'value',
+        default: 'default',
+        trackEvents: true,
+      };
+
+      // Configure with allAttributesPrivate set to true
+      const config = { ...defaultConfig, allAttributesPrivate: true };
+
+      const sender = MockEventSender();
+      const ep = EventProcessor(platform, config, envId, null, null, sender);
+      try {
+        ep.enqueue(event);
+        await ep.flush();
+
+        expect(sender.calls.length()).toEqual(1);
+        const output = (await sender.calls.take()).events;
+        expect(output.length).toEqual(2);
+
+        // Verify the feature event has filtered context
+        checkFeatureEvent(output[0], event, false, filteredContext);
+
+        // Verify the summary event has filtered context
+        const summaryEvent = output[1];
+        checkSummaryEvent(summaryEvent);
+        expect(summaryEvent.context).toEqual(filteredContext);
+        expect(summaryEvent.features).toEqual({
+          flagkey: {
+            contextKinds: ['user'],
+            default: 'default',
+            counters: [{ version: 11, variation: 1, value: 'value', count: 1 }],
+          },
+        });
+      } finally {
+        ep.stop();
+      }
+    });
   });
 });

src/__tests__/LDClient-test.js

@@ -672,7 +672,9 @@ describe('LDClient', () => {
           await client.waitForInitialization(5);
         });
 
-        expect(eventsServer.requests.length()).toEqual(1);
+        // Flushing is an async operation, so we cannot ensure that the requests are made by
+        // the time we reach this point. If we await the nextRequest(), then it will catch
+        // whatever was flushed.
         const req = await eventsServer.nextRequest();
         const data = JSON.parse(req.body);
         expect(data.length).toEqual(1);

src/__tests__/MultiEventSummarizer-test.js

@@ -0,0 +1,148 @@
+const MultiEventSummarizer = require('../MultiEventSummarizer');
+const ContextFilter = require('../ContextFilter');
+
+function makeEvent(key, version, variation, value, defaultVal, context) {
+  return {
+    kind: 'feature',
+    creationDate: 1000,
+    key: key,
+    version: version,
+    context: context,
+    variation: variation,
+    value: value,
+    default: defaultVal,
+  };
+}
+
+describe('given a multi-event summarizer and context filter', () => {
+  let summarizer;
+  let contextFilter;
+
+  beforeEach(() => {
+    contextFilter = ContextFilter(false, []);
+    summarizer = MultiEventSummarizer(contextFilter);
+  });
+
+  it('creates new summarizer for new context hash', async () => {
+    const context = { kind: 'user', key: 'user1' };
+    const event = { kind: 'feature', context };
+
+    summarizer.summarizeEvent(event);
+
+    const summaries = await summarizer.getSummaries();
+    expect(summaries).toHaveLength(1);
+  });
+
+  it('uses existing summarizer for same context hash', async () => {
+    const context = { kind: 'user', key: 'user1' };
+    const event1 = { kind: 'feature', context, value: 'value1' };
+    const event2 = { kind: 'feature', context, value: 'value2' };
+
+    summarizer.summarizeEvent(event1);
+    summarizer.summarizeEvent(event2);
+
+    const summaries = await summarizer.getSummaries();
+    expect(summaries).toHaveLength(1);
+  });
+
+  it('ignores non-feature events', async () => {
+    const context = { kind: 'user', key: 'user1' };
+    const event = { kind: 'identify', context };
+
+    summarizer.summarizeEvent(event);
+
+    const summaries = await summarizer.getSummaries();
+    expect(summaries).toHaveLength(0);
+  });
+
+  it('handles multiple different contexts', async () => {
+    const context1 = { kind: 'user', key: 'user1' };
+    const context2 = { kind: 'user', key: 'user2' };
+    const event1 = { kind: 'feature', context: context1 };
+    const event2 = { kind: 'feature', context: context2 };
+
+    summarizer.summarizeEvent(event1);
+    summarizer.summarizeEvent(event2);
+
+    const summaries = await summarizer.getSummaries();
+    expect(summaries).toHaveLength(2);
+  });
+
+  it('automatically clears summaries when summarized', async () => {
+    const context = { kind: 'user', key: 'user1' };
+    const event = { kind: 'feature', context };
+
+    summarizer.summarizeEvent(event);
+
+    const summariesA = await summarizer.getSummaries();
+    const summariesB = await summarizer.getSummaries();
+    expect(summariesA).toHaveLength(1);
+    expect(summariesB).toHaveLength(0);
+  });
+
+  it('increments counters for feature events across multiple contexts', async () => {
+    const context1 = { kind: 'user', key: 'user1' };
+    const context2 = { kind: 'user', key: 'user2' };
+
+    // Events for context1 (using values 100-199)
+    const event1 = makeEvent('key1', 11, 1, 100, 111, context1);
+    const event2 = makeEvent('key1', 11, 2, 150, 111, context1);
+    const event3 = makeEvent('key2', 22, 1, 199, 222, context1);
+
+    // Events for context2 (using values 200-299)
+    const event4 = makeEvent('key1', 11, 1, 200, 211, context2);
+    const event5 = makeEvent('key1', 11, 2, 250, 211, context2);
+    const event6 = makeEvent('key2', 22, 1, 299, 222, context2);
+
+    summarizer.summarizeEvent(event1);
+    summarizer.summarizeEvent(event2);
+    summarizer.summarizeEvent(event3);
+    summarizer.summarizeEvent(event4);
+    summarizer.summarizeEvent(event5);
+    summarizer.summarizeEvent(event6);
+
+    const summaries = await summarizer.getSummaries();
+    expect(summaries).toHaveLength(2);
+
+    // Sort summaries by context key to make assertions consistent
+    summaries.sort((a, b) => a.context.key.localeCompare(b.context.key));
+
+    // Verify first context's summary (user1, values 100-199)
+    const summary1 = summaries[0];
+    summary1.features.key1.counters.sort((a, b) => a.value - b.value);
+    expect(summary1.features).toEqual({
+      key1: {
+        contextKinds: ['user'],
+        default: 111,
+        counters: [
+          { value: 100, variation: 1, version: 11, count: 1 },
+          { value: 150, variation: 2, version: 11, count: 1 },
+        ],
+      },
+      key2: {
+        contextKinds: ['user'],
+        default: 222,
+        counters: [{ value: 199, variation: 1, version: 22, count: 1 }],
+      },
+    });
+
+    // Verify second context's summary (user2, values 200-299)
+    const summary2 = summaries[1];
+    summary2.features.key1.counters.sort((a, b) => a.value - b.value);
+    expect(summary2.features).toEqual({
+      key1: {
+        contextKinds: ['user'],
+        default: 211,
+        counters: [
+          { value: 200, variation: 1, version: 11, count: 1 },
+          { value: 250, variation: 2, version: 11, count: 1 },
+        ],
+      },
+      key2: {
+        contextKinds: ['user'],
+        default: 222,
+        counters: [{ value: 299, variation: 1, version: 22, count: 1 }],
+      },
+    });
+  });
+});