processor management

Author: OmerKat

src/modules/analytics.ts

@@ -68,17 +68,6 @@ export const createAnalyticsModule = ({
     return analyticsSharedState.sessionContext;
   };
 
-  const track = (params: TrackEventParams) => {
-    if (!enabled || analyticsSharedState.requestsQueue.length >= maxQueueSize) {
-      return;
-    }
-    const intrinsicData = getEventIntrinsicData();
-    analyticsSharedState.requestsQueue.push({
-      ...params,
-      ...intrinsicData,
-    });
-  };
-
   const batchRequestFallback = async (events: AnalyticsApiRequestData[]) => {
     await axiosClient.request({
       method: "POST",
@@ -131,20 +120,35 @@ export const createAnalyticsModule = ({
     window.addEventListener("visibilitychange", onVisibilityChange);
   }
 
-  // start analytics processor only if it's the first instance and analytics is enabled //
-  if (enabled) {
+  const startProcessing = () => {
+    if (!enabled) return;
     startAnalyticsProcessor(flush, {
       throttleTime,
       batchSize,
     });
-  }
+  };
+
+  const track = (params: TrackEventParams) => {
+    if (!enabled || analyticsSharedState.requestsQueue.length >= maxQueueSize) {
+      return;
+    }
+    const intrinsicData = getEventIntrinsicData();
+    analyticsSharedState.requestsQueue.push({
+      ...params,
+      ...intrinsicData,
+    });
+    startProcessing();
+  };
 
   const cleanup = () => {
     if (typeof window === "undefined") return;
     window.removeEventListener("visibilitychange", onVisibilityChange);
     stopAnalyticsProcessor();
   };
 
+  // start the flusing process ///
+  startProcessing();
+
   return {
     track,
     cleanup,
@@ -163,11 +167,16 @@ async function startAnalyticsProcessor(
   }
 ) {
   if (analyticsSharedState.isProcessing) {
+    // only one instance of the analytics processor can be running at a time //
     return;
   }
   analyticsSharedState.isProcessing = true;
+
   const { throttleTime = 1000, batchSize = 30 } = options ?? {};
-  while (analyticsSharedState.isProcessing) {
+  while (
+    analyticsSharedState.isProcessing &&
+    analyticsSharedState.requestsQueue.length > 0
+  ) {
     const requests = analyticsSharedState.requestsQueue.splice(0, batchSize);
     if (requests.length > 0) {
       try {
@@ -179,6 +188,7 @@ async function startAnalyticsProcessor(
     }
     await new Promise((resolve) => setTimeout(resolve, throttleTime));
   }
+  analyticsSharedState.isProcessing = false;
 }
 
 function getEventIntrinsicData(): TrackEventIntrinsicData {

tests/unit/analytics.test.ts

@@ -6,6 +6,8 @@ import {
   TrackEventData,
 } from "../../src/index.ts";
 import { getSharedInstance } from "../../src/utils/sharedInstance.ts";
+import { User } from "../../src/modules/auth.types.ts";
+import { AxiosInstance } from "axios";
 
 describe("Analytics Module", () => {
   let base44: ReturnType<typeof createClient>;
@@ -19,26 +21,45 @@ describe("Analytics Module", () => {
   const serverUrl = "https://api.base44.com";
 
   beforeEach(() => {
-    base44 = createClient({
-      serverUrl,
-      appId,
-    });
+    vi.mock("../../src/utils/axios-client.ts", () => ({
+      createAxiosClient: vi.fn().mockImplementation(
+        () =>
+          ({
+            request: vi.fn().mockResolvedValue({
+              status: 200,
+              data: {
+                message: "success",
+              },
+            }),
+          } as unknown as AxiosInstance)
+      ),
+    }));
     sharedState = getSharedInstance("analytics", () => ({
       requestsQueue: [],
       isProcessing: false,
-      sessionContext: {
-        user_id: "test-user-id",
-      },
-      config: {
-        enabled: true,
-        maxQueueSize: 1000,
-        throttleTime: 1000,
-        batchSize: 30,
-      },
+      sessionContext: {},
+      config: {},
     }));
+    sharedState.isProcessing = false;
+    sharedState.requestsQueue = [];
+    sharedState.sessionContext = {
+      user_id: "test-user-id",
+    };
+    sharedState.config = {
+      enabled: true,
+      maxQueueSize: 1000,
+      throttleTime: 1000,
+      batchSize: 2,
+    };
+
+    base44 = createClient({
+      serverUrl,
+      appId,
+    });
   });
 
   afterEach(() => {
+    vi.clearAllMocks();
     base44.cleanup();
     sharedState = null;
   });
@@ -47,17 +68,38 @@ describe("Analytics Module", () => {
     expect(base44.analytics).toBeDefined();
     expect(sharedState).toBeDefined();
     expect(sharedState?.requestsQueue).toBeDefined();
-    expect(sharedState?.isProcessing).toBe(true);
+    expect(sharedState?.isProcessing).toBe(false);
   });
 
   test("should track an event", () => {
-    vi.spyOn(base44.analytics, "track").mockImplementation(() => {
-      console.log("track called");
-    });
+    vi.spyOn(base44.analytics, "track");
 
     base44.analytics.track({ eventName: "test-event" });
+    expect(sharedState?.isProcessing).toBe(true);
     expect(base44.analytics.track).toHaveBeenCalledWith({
       eventName: "test-event",
     });
   });
+
+  test("should track multiple events", async () => {
+    vi.useFakeTimers();
+
+    for (let i = 0; i < 5; i++) {
+      base44.analytics.track({ eventName: `test-event ${i}` });
+    }
+
+    expect(sharedState?.isProcessing).toBe(true);
+    expect(sharedState?.requestsQueue.length).toBe(4);
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(sharedState?.requestsQueue.length).toBe(2);
+    // add another event while processing to mix things up
+    base44.analytics.track({ eventName: `test-event 5` });
+
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(sharedState?.requestsQueue.length).toBe(1);
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(sharedState?.requestsQueue.length).toBe(0);
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(sharedState?.isProcessing).toBe(false);
+  });
 });