Allow event options to be configurable

atrakh · atrakh · commit fe8882e280ff · 2024-10-24T18:00:26.000-07:00
diff --git a/src/component/events.ts b/src/component/events.ts
@@ -7,10 +7,13 @@ import {
   MutationCtx,
 } from "./_generated/server";
 import { internal } from "./_generated/api";
-import { sendEvents } from "../sdk/EventProcessor";
+import {
+  EventProcessorOptions,
+  sendEvents,
+  validateEventProcessorOptions,
+} from "../sdk/EventProcessor";
 import isEqual from "lodash.isequal";
 
-// TODO: Make these configurable.
 export const EVENT_CAPACITY = 1000;
 export const EVENT_BATCH_SIZE = 100;
 export const EVENT_PROCESSING_INTERVAL_SECONDS = 5;
@@ -20,6 +23,9 @@ const eventsOptions = v.optional(
     allAttributesPrivate: v.optional(v.boolean()),
     privateAttributes: v.optional(v.array(v.string())),
     eventsUri: v.optional(v.string()),
+    eventCapacity: v.optional(v.number()),
+    eventBatchSize: v.optional(v.number()),
+    eventProcessingIntervalSeconds: v.optional(v.number()),
   })
 );
 
@@ -31,20 +37,23 @@ export const storeEvents = mutation({
   },
   returns: v.null(),
   handler: async (ctx, { sdkKey, payloads, options }) => {
+    validateEventProcessorOptions(options);
     await handleScheduleProcessing(ctx, {
       sdkKey,
       options,
     });
 
-    const numEvents = (await ctx.db.query("events").take(EVENT_CAPACITY + 1))
+    const eventCapacity = options?.eventCapacity ?? EVENT_CAPACITY;
+
+    const numEvents = (await ctx.db.query("events").take(eventCapacity + 1))
       .length;
 
-    if (numEvents >= EVENT_CAPACITY) {
+    if (numEvents >= eventCapacity) {
       console.warn("Event store is full, dropping events.");
       return;
     }
 
-    const payloadsToStore = payloads.slice(0, EVENT_CAPACITY - numEvents);
+    const payloadsToStore = payloads.slice(0, eventCapacity - numEvents);
     if (payloadsToStore.length !== payloads.length) {
       console.warn(
         `${payloads.length - payloadsToStore.length} events were dropped due to capacity limits.`
@@ -68,13 +77,10 @@ const handleScheduleProcessing = async (
   }: {
     sdkKey: string;
     runImmediately?: boolean;
-    options?: {
-      allAttributesPrivate?: boolean;
-      privateAttributes?: string[];
-      eventsUri?: string;
-    };
+    options?: EventProcessorOptions;
   }
 ) => {
+  validateEventProcessorOptions(options);
   const existingScheduledJob = await ctx.db.query("eventSchedule").first();
   if (existingScheduledJob !== null) {
     const existingSystemJob = await ctx.db.system.get(
@@ -132,8 +138,10 @@ export const processEvents = internalAction({
     options: eventsOptions,
   },
   handler: async (ctx, { sdkKey, options }) => {
+    validateEventProcessorOptions(options);
+    const eventBatchSize = options?.eventBatchSize ?? EVENT_BATCH_SIZE;
     const events = await ctx.runQuery(internal.events.getOldestEvents, {
-      count: EVENT_BATCH_SIZE,
+      count: eventBatchSize,
     });
 
     if (events.length === 0) {
diff --git a/src/sdk/EventProcessor.test.ts b/src/sdk/EventProcessor.test.ts
@@ -1,5 +1,8 @@
 import { describe, expect, test, vi } from "vitest";
-import { EventProcessor } from "./EventProcessor";
+import {
+  EventProcessor,
+  validateEventProcessorOptions,
+} from "./EventProcessor";
 import { convexTest } from "convex-test";
 import schema from "../component/schema";
 import { modules } from "../component/setup.test";
@@ -35,3 +38,82 @@ describe("EventProcessor", () => {
     expect(sendEvents).toHaveBeenCalledOnce();
   });
 });
+
+describe("validateEventProcessorOptions", () => {
+  test("should throw an error if allAttributesPrivate is not a boolean", () => {
+    const options = { allAttributesPrivate: "true" };
+    expect(() => validateEventProcessorOptions(options)).toThrow(
+      new Error("allAttributesPrivate must be a boolean")
+    );
+  });
+
+  test("should throw an error if privateAttributes is not an array", () => {
+    const options = { privateAttributes: "not-an-array" };
+    // @ts-expect-error testing invalid input
+    expect(() => validateEventProcessorOptions(options)).toThrow(
+      new Error("privateAttributes must be an array of strings")
+    );
+  });
+
+  test("should throw an error if eventsUri is not a string", () => {
+    const options = { eventsUri: 12345 };
+    // @ts-expect-error testing invalid input
+    expect(() => validateEventProcessorOptions(options)).toThrow(
+      new Error("eventsUri must be a string")
+    );
+  });
+
+  test("should throw an error if eventsUri is not a valid URL", () => {
+    const options = { eventsUri: "invalid-url" };
+    expect(() => validateEventProcessorOptions(options)).toThrow(
+      new Error("eventsUri must be a valid URL")
+    );
+  });
+
+  test("should throw an error if eventProcessingIntervalSeconds is not a positive number", () => {
+    const options = { eventProcessingIntervalSeconds: -1 };
+    expect(() => validateEventProcessorOptions(options)).toThrow(
+      new Error("eventProcessingIntervalSeconds must be a positive number")
+    );
+  });
+
+  test("should throw an error if eventCapacity is not a positive number", () => {
+    const options = { eventCapacity: 0 };
+    expect(() => validateEventProcessorOptions(options)).toThrow(
+      new Error("eventCapacity must be a positive number")
+    );
+  });
+
+  test("should throw an error if eventCapacity is less than eventBatchSize", () => {
+    const options = { eventCapacity: 10, eventBatchSize: 20 };
+    expect(() => validateEventProcessorOptions(options)).toThrow(
+      new Error("eventCapacity must be greater than or equal to eventBatchSize")
+    );
+  });
+
+  test("should throw an error if eventBatchSize is not a positive number", () => {
+    const options = { eventBatchSize: -1 };
+    expect(() => validateEventProcessorOptions(options)).toThrow(
+      new Error("eventBatchSize must be a positive number")
+    );
+  });
+
+  test("should throw an error if eventBatchSize exceeds 4000", () => {
+    const options = { eventBatchSize: 5000 };
+    expect(() => validateEventProcessorOptions(options)).toThrow(
+      new Error("eventBatchSize must be less than or equal to 4000")
+    );
+  });
+
+  test("should not throw an error if options are valid", () => {
+    const validOptions = {
+      allAttributesPrivate: true,
+      privateAttributes: ["attr1", "attr2"],
+      eventsUri: "https://valid.url",
+      eventProcessingIntervalSeconds: 10,
+      eventCapacity: 100,
+      eventBatchSize: 50,
+    };
+    expect(() => validateEventProcessorOptions(validOptions)).not.toThrow();
+  });
+});
diff --git a/src/sdk/EventProcessor.ts b/src/sdk/EventProcessor.ts
@@ -18,13 +18,15 @@ export class EventProcessor {
   constructor(
     private readonly eventStore: ComponentApi["events"],
     private readonly ctx: RunMutationCtx,
-    private readonly sdkKey: string
+    private readonly sdkKey: string,
+    private readonly options?: EventProcessorOptions
   ) {}
 
   sendEvent(inputEvent: object) {
     return this.ctx.runMutation(this.eventStore.storeEvents, {
       payloads: [JSON.stringify(inputEvent)],
       sdkKey: this.sdkKey,
+      options: this.options,
     });
   }
 
@@ -94,3 +96,94 @@ export const sendEvents = async (
 
   await eventProcessor.flush();
 };
+
+export type EventProcessorOptions = {
+  allAttributesPrivate?: boolean;
+  privateAttributes?: string[];
+
+  // The URL to send events to. If not provided, the default LaunchDarkly
+  // events endpoint will be used. Most users will not need to set this.
+  eventsUri?: string;
+  // The number of events to store in LaunchDarkly, awaiting processing
+  // to be sent to LaunchDarkly. If the number of events exceeds this
+  // value, new events will be dropped.
+  eventCapacity?: number;
+  // The number of events to send to LaunchDarkly in a single batch.
+  eventBatchSize?: number;
+  // How often to process events, in seconds.
+  eventProcessingIntervalSeconds?: number;
+};
+
+export const validateEventProcessorOptions = (
+  options?: EventProcessorOptions
+) => {
+  if (!options) {
+    return;
+  }
+
+  if (
+    options.allAttributesPrivate !== undefined &&
+    typeof options.allAttributesPrivate !== "boolean"
+  ) {
+    throw new Error("allAttributesPrivate must be a boolean");
+  }
+  if (
+    options.privateAttributes !== undefined &&
+    !Array.isArray(options.privateAttributes)
+  ) {
+    throw new Error("privateAttributes must be an array of strings");
+  }
+  if (options.eventsUri !== undefined) {
+    if (typeof options.eventsUri !== "string") {
+      throw new Error("eventsUri must be a string");
+    }
+    try {
+      new URL(options.eventsUri);
+    } catch {
+      throw new Error("eventsUri must be a valid URL");
+    }
+  }
+
+  if (options.eventProcessingIntervalSeconds !== undefined) {
+    if (
+      typeof options.eventProcessingIntervalSeconds !== "number" ||
+      options.eventProcessingIntervalSeconds <= 0
+    ) {
+      throw new Error(
+        "eventProcessingIntervalSeconds must be a positive number"
+      );
+    }
+  }
+
+  if (options.eventCapacity !== undefined) {
+    if (
+      typeof options.eventCapacity !== "number" ||
+      options.eventCapacity <= 0
+    ) {
+      throw new Error("eventCapacity must be a positive number");
+    }
+
+    if (
+      options.eventBatchSize !== undefined &&
+      options.eventCapacity <= options.eventBatchSize
+    ) {
+      throw new Error(
+        "eventCapacity must be greater than or equal to eventBatchSize"
+      );
+    }
+  }
+
+  if (options.eventBatchSize !== undefined) {
+    if (
+      typeof options.eventBatchSize !== "number" ||
+      options.eventBatchSize <= 0
+    ) {
+      throw new Error("eventBatchSize must be a positive number");
+    }
+
+    // Do not allow eventBatchSize to exceed 4000 to respect Convex limits.
+    if (options.eventBatchSize > 4000) {
+      throw new Error("eventBatchSize must be less than or equal to 4000");
+    }
+  }
+};
diff --git a/src/sdk/LaunchDarkly.ts b/src/sdk/LaunchDarkly.ts
@@ -9,7 +9,11 @@ import {
 import { createPlatformInfo } from "./createPlatformInfo";
 import ConvexCrypto from "./crypto";
 import { FeatureStore } from "./FeatureStore";
-import { EventProcessor } from "./EventProcessor";
+import {
+  EventProcessor,
+  EventProcessorOptions,
+  validateEventProcessorOptions,
+} from "./EventProcessor";
 import { RunMutationCtx, RunQueryCtx } from "../component/types";
 import { ComponentApi } from "./useApi";
 
@@ -21,7 +25,7 @@ export class LaunchDarkly extends LDClientImpl {
       application?: LDOptions["application"];
       sendEvents?: boolean;
       LAUNCHDARKLY_SDK_KEY?: string;
-    }
+    } & EventProcessorOptions
   ) {
     const { store, events } = component;
     const logger = new BasicLogger({
@@ -56,7 +60,8 @@ export class LaunchDarkly extends LDClientImpl {
     // We can only send events if the context has a runMutation function.
     // This exists in Convex mutations and actions, but not in queries.
     if ("runMutation" in ctx && sendEvents) {
-      const eventProcessor = new EventProcessor(events, ctx, sdkKey);
+      validateEventProcessorOptions(options);
+      const eventProcessor = new EventProcessor(events, ctx, sdkKey, options);
       // @ts-expect-error We are setting the eventProcessor directly here.
       this.eventProcessor = eventProcessor;
     }
