[front] fix: paginate active user metrics query

front/lib/api/assistant/observability/active_users_metrics.test.ts

@@ -0,0 +1,85 @@
+import { fetchActiveUsersMetrics } from "@app/lib/api/assistant/observability/active_users_metrics";
+import { searchAnalytics } from "@app/lib/api/elasticsearch";
+import { createResourceTest } from "@app/tests/utils/generic_resource_tests";
+import { Ok } from "@app/types/shared/result";
+import type { estypes } from "@elastic/elasticsearch";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@app/lib/api/elasticsearch", async (importActual) => {
+  const actual =
+    await importActual<typeof import("@app/lib/api/elasticsearch")>();
+  return { ...actual, searchAnalytics: vi.fn() };
+});
+
+interface UserDayBucket {
+  key: { day: number; user: string };
+  doc_count: number;
+}
+
+function esResponse(
+  buckets: UserDayBucket[],
+  afterKey?: { day: number; user: string }
+) {
+  const response: estypes.SearchResponse<never> = {
+    took: 1,
+    timed_out: false,
+    _shards: { total: 1, successful: 1, failed: 0, skipped: 0 },
+    hits: { hits: [] },
+    aggregations: {
+      by_user_day: {
+        buckets,
+        ...(afterKey ? { after_key: afterKey } : {}),
+      },
+    },
+  };
+
+  return new Ok(response);
+}
+
+describe("fetchActiveUsersMetrics", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it("combines paginated user-day buckets before computing active users", async () => {
+    const { workspace } = await createResourceTest({ role: "admin" });
+    const firstDay = Date.UTC(2026, 7, 6);
+    const secondDay = Date.UTC(2026, 7, 7);
+
+    vi.mocked(searchAnalytics)
+      .mockResolvedValueOnce(
+        esResponse(
+          [
+            { key: { day: firstDay, user: "user-1" }, doc_count: 1 },
+            { key: { day: secondDay, user: "user-1" }, doc_count: 1 },
+          ],
+          { day: secondDay, user: "user-1" }
+        )
+      )
+      .mockResolvedValueOnce(
+        esResponse([{ key: { day: secondDay, user: "user-2" }, doc_count: 1 }])
+      );
+
+    const result = await fetchActiveUsersMetrics(
+      workspace,
+      "2026-08-06",
+      "2026-08-07"
+    );
+
+    expect(result.isOk()).toBe(true);
+    if (result.isOk()) {
+      expect(
+        result.value.map(({ date, dau, wau, mau }) => ({
+          date,
+          dau,
+          wau,
+          mau,
+        }))
+      ).toEqual([
+        { date: "2026-08-06", dau: 1, wau: 1, mau: 1 },
+        { date: "2026-08-07", dau: 2, wau: 2, mau: 2 },
+      ]);
+    }
+    expect(searchAnalytics).toHaveBeenCalledTimes(2);
+  });
+});

front/lib/api/assistant/observability/active_users_metrics.ts

@@ -1,5 +1,4 @@
 import {
-  bucketsToArray,
   formatDateFromMillis,
   searchAnalytics,
 } from "@app/lib/api/elasticsearch";
@@ -23,25 +22,29 @@ export type GetWorkspaceActiveUsersResponse = {
   points: ActiveUsersMetricsPoint[];
 };
 
-interface UserBucket {
-  key: string;
+interface UserDayBucket {
+  key: {
+    day: number;
+    user: string;
+  };
   doc_count: number;
 }
 
-interface DayBucket {
-  key: number;
-  key_as_string: string;
-  doc_count: number;
-  users?: estypes.AggregationsMultiBucketAggregateBase<UserBucket>;
+interface CompositeKey {
+  day: number;
+  user: string;
 }
 
 interface ActiveUsersAggs {
-  by_day?: estypes.AggregationsMultiBucketAggregateBase<DayBucket>;
+  by_user_day?: {
+    after_key?: CompositeKey;
+    buckets: UserDayBucket[];
+  };
 }
 
 const WAU_WINDOW_DAYS = 7;
 const MAU_WINDOW_DAYS = 28;
-const MAX_USERS_PER_DAY = 10000;
+const COMPOSITE_AGG_SIZE = 10000;
 const MS_PER_DAY = 24 * 60 * 60 * 1000;
 
 /**
@@ -70,10 +73,8 @@ function computeRollingActiveUsers(
  * Fetches DAU/WAU/MAU metrics for the given time range.
  *
  * Strategy:
- * 1. Fetch user IDs per day using a single ES query with terms aggregation
+ * 1. Fetch all (day, user ID) pairs with a paginated composite aggregation
  * 2. Compute rolling WAU (7-day) and MAU (30-day) windows on the server
- *
- * This approach is efficient (single ES query) and accurate for typical workspace sizes.
  */
 export async function fetchActiveUsersMetrics(
   workspace: LightWorkspaceType,
@@ -105,57 +106,64 @@ export async function fetchActiveUsersMetrics(
     },
   };
 
-  const result = await searchAnalytics<never, ActiveUsersAggs>(query, {
-    aggregations: {
-      by_day: {
-        date_histogram: {
-          field: "timestamp",
-          calendar_interval: "day",
-          time_zone: timezone,
-        },
-        aggs: {
-          users: {
-            terms: {
-              field: "user_id",
-              size: MAX_USERS_PER_DAY,
-            },
+  const usersByDay = new Map<number, Set<string>>();
+  let afterKey: CompositeKey | undefined;
+  let buckets: UserDayBucket[];
+
+  do {
+    const result = await searchAnalytics<never, ActiveUsersAggs>(query, {
+      aggregations: {
+        by_user_day: {
+          composite: {
+            size: COMPOSITE_AGG_SIZE,
+            sources: [
+              {
+                day: {
+                  date_histogram: {
+                    field: "timestamp",
+                    calendar_interval: "day",
+                    time_zone: timezone,
+                  },
+                },
+              },
+              { user: { terms: { field: "user_id" } } },
+            ],
+            ...(afterKey ? { after: afterKey } : {}),
           },
         },
       },
-    },
-    size: 0,
-  });
-
-  if (result.isErr()) {
-    const status =
-      result.error.statusCode !== undefined
-        ? `, HTTP ${result.error.statusCode}`
-        : "";
-
-    return new Err(
-      new Error(
-        `Elasticsearch query failed (${result.error.type}${status}): ${result.error.message}`,
-        { cause: result.error }
-      )
-    );
-  }
+      size: 0,
+    });
 
-  const dayBuckets = bucketsToArray<DayBucket>(
-    result.value.aggregations?.by_day?.buckets
-  );
+    if (result.isErr()) {
+      const status =
+        result.error.statusCode !== undefined
+          ? `, HTTP ${result.error.statusCode}`
+          : "";
+
+      return new Err(
+        new Error(
+          `Elasticsearch query failed (${result.error.type}${status}): ${result.error.message}`,
+          { cause: result.error }
+        )
+      );
+    }
 
-  // Build a map of timestamp -> Set of user IDs
-  const usersByDay = new Map<number, Set<string>>();
-  const sortedTimestamps: number[] = [];
+    const aggregation = result.value.aggregations?.by_user_day;
+    buckets = aggregation?.buckets ?? [];
+    for (const bucket of buckets) {
+      const users = usersByDay.get(bucket.key.day);
+      if (users) {
+        users.add(bucket.key.user);
+      } else {
+        usersByDay.set(bucket.key.day, new Set([bucket.key.user]));
+      }
+    }
 
-  for (const bucket of dayBuckets) {
-    const users = bucketsToArray<UserBucket>(bucket.users?.buckets);
-    const userSet = new Set(users.map((u) => u.key));
-    usersByDay.set(bucket.key, userSet);
-    sortedTimestamps.push(bucket.key);
-  }
+    afterKey = aggregation?.after_key;
+  } while (afterKey !== undefined && buckets.length > 0);
 
-  sortedTimestamps.sort((a, b) => a - b);
+  const sortedTimestamps = [...usersByDay.keys()].sort((a, b) => a - b);
 
   // Collect timestamps in the requested range for membership counting.
   const requestedTimestamps = sortedTimestamps.filter(