diff --git a/front/lib/api/assistant/observability/active_users_metrics.test.ts b/front/lib/api/assistant/observability/active_users_metrics.test.ts new file mode 100644 index 000000000000..08f9d960822d --- /dev/null +++ b/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(); + 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 = { + 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); + }); +}); diff --git a/front/lib/api/assistant/observability/active_users_metrics.ts b/front/lib/api/assistant/observability/active_users_metrics.ts index a68b94065ee1..c023dec2033e 100644 --- a/front/lib/api/assistant/observability/active_users_metrics.ts +++ b/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; +interface CompositeKey { + day: number; + user: string; } interface ActiveUsersAggs { - by_day?: estypes.AggregationsMultiBucketAggregateBase; + 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(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>(); + let afterKey: CompositeKey | undefined; + let buckets: UserDayBucket[]; + + do { + const result = await searchAnalytics(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( - 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>(); - 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(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(