openvidu/openvidu-dashboard/dashboard-ce/backend/test/integration/analytics-query-plan.test.ts

import { writeFileSync } from 'node:fs';

import { generate } from '@openvidu-dashboard/data-generator/generator';
import { type Db, type FindCursor, MongoClient } from 'mongodb';
import { afterAll, beforeAll, describe, expect, inject, it } from 'vitest';

import { Collection, FeatureKey, ParticipantKind } from '@openvidu-dashboard/typings';

import { AnalyticsCache } from '../../src/analytics/analytics-cache.js';
import { CE_ANALYTICS_CATALOGUE, metricFields } from '../../src/analytics/analytics-catalogue.js';
import { AnalyticsFeatureGate } from '../../src/analytics/analytics-gate.js';
import { planSeries, type RollupCollection } from '../../src/analytics/analytics-planner.js';
import { AnalyticsRepository, ROLLUP_INDEX } from '../../src/analytics/analytics-repository.js';
import {
	analyticsBreakdownsQuery,
	analyticsExportQuery,
	analyticsSeriesQuery,
	analyticsSummaryQuery
} from '../../src/analytics/analytics-schemas.js';
import { AnalyticsService } from '../../src/analytics/analytics-service.js';
import { expectIndexBacked, expectIndexes, p95 } from '../support/query-plan.js';
import { expectWithinBudgets } from '../support/latency-budget.js';

/**
 * Every analytics query explained on a database generated by the data generator with five-minute, node and room-sum
 * rollups, the consistency of the rollup sums with the entities, and the p95 of the performance budget, which fails the
 * suite when exceeded. ANALYTICS_PLAN_SCENARIO (S by default), ANALYTICS_PLAN_DAYS (2 by default) and
 * ANALYTICS_PLAN_REPORT (a file that also receives the report) work as in the rooms suite.
 */

const mongoUri = inject('mongoUri');
const scenario = (process.env.ANALYTICS_PLAN_SCENARIO ?? 'S') as 'S' | 'M' | 'L';
const days = Number(process.env.ANALYTICS_PLAN_DAYS ?? '2');
const DATABASE = 'dashboard_test_analytics_plan';
const HOUR = 3_600_000;
const DAY = 24 * HOUR;
const GENERATION_TIMEOUT_MS = scenario === 'S' ? 600_000 : 3_600_000;
const SERIES_METRICS = 'roomsCreated,participantsJoined,connectionSuccessRate,bytesUp,bytesDown';
const BREAKDOWNS = 'sdk,os,browser,connectionType,kind,egressType,egressStatus';

describe.skipIf(mongoUri === '')(`analytics query plan at scenario ${scenario}`, () => {
	let client: MongoClient;
	let db: Db;
	let repository: AnalyticsRepository;
	let service: AnalyticsService;
	let node: string;
	const now = new Date();
	const hour = new Date(Math.floor(now.getTime() / HOUR) * HOUR);
	const report: string[] = [];
	const clock = () => now;

	/** The rollup read of a series request, explained with the bound of the documents in its range. */
	const explainSeries = async (query: Record<string, string>, expected: RollupCollection) => {
		const params = analyticsSeriesQuery(clock).parse(query);
		const plan = planSeries({
			...params,
			now,
			retention: { statsMs: 7 * DAY, entitiesMs: 90 * DAY, rollups5MinMs: 7 * DAY, rollupsHourlyMs: 400 * DAY },
			info: CE_ANALYTICS_CATALOGUE.info
		});
		const from = plan.starts[0] ?? params.from;
		const filter = { scope: params.scope, bucket: { $gte: from, $lt: params.to } };
		const documents = await db.collection(plan.collection).countDocuments(filter);
		const cursor = repository.rollups(
			plan.collection,
			params.scope,
			from,
			params.to,
			metricFields(CE_ANALYTICS_CATALOGUE, params.metrics)
		) as unknown as FindCursor;
		const summary = await expectIndexBacked(db, cursor, { maxDocsExamined: documents });

		expect(plan.collection).toBe(expected);
		expectIndexes(summary, [ROLLUP_INDEX]);
		expect(summary.totalKeysExamined).toBeLessThanOrEqual(documents + 1);

		return documents;
	};

	beforeAll(async () => {
		client = new MongoClient(mongoUri);
		db = client.db(DATABASE);
		await db.dropDatabase();
		await generate(db, {
			scenario,
			days,
			statsMode: 'participant',
			seed: 1,
			retentionDays: { events: 7, stats: 7, tracks: 7, entities: 90 },
			timeseries: true,
			rollups: { fiveMinute: true, nodes: true, roomSums: true }
		});
		repository = new AnalyticsRepository(db);
		service = new AnalyticsService(db, {
			now: clock,
			cache: new AnalyticsCache(0, clock),
			gate: new AnalyticsFeatureGate(() => [FeatureKey.PER_NODE_BREAKDOWN]),
			catalogue: CE_ANALYTICS_CATALOGUE
		});

		const scopes = ((await db.collection(Collection.ANALYTICS_HOURLY).distinct('scope')) as string[]).filter(
			(scope) => scope.startsWith('node:')
		);

		if (scopes[0] === undefined) {
			throw new Error('the generated database has no node scope');
		}

		node = scopes[0];
	}, GENERATION_TIMEOUT_MS);

	afterAll(async () => {
		if (report.length > 0) {
			const text = [`Analytics p95 of 20 runs, scenario ${scenario}, ${String(days)} days`, ...report].join('\n');

			console.log(text);

			if (process.env.ANALYTICS_PLAN_REPORT !== undefined) {
				writeFileSync(process.env.ANALYTICS_PLAN_REPORT, `${text}\n`);
			}
		}

		await db.dropDatabase();
		await client.close();
	});

	it('series: bounded scans of { scope, bucket } in the collection of each granularity', async () => {
		expect(await explainSeries({ metrics: SERIES_METRICS, from: 'now-24h' }, 'analytics_5min')).toBeGreaterThan(0);
		expect(
			await explainSeries({ metrics: SERIES_METRICS, from: 'now-7d', scope: node }, 'analytics_hourly')
		).toBeGreaterThan(0);
		await explainSeries(
			{ metrics: SERIES_METRICS, from: 'now-400d', granularity: '1d', tz: 'Europe/Madrid' },
			'analytics_hourly'
		);
		expect(
			await explainSeries({ metrics: SERIES_METRICS, from: 'now-400d', granularity: '1d' }, 'analytics_daily')
		).toBeGreaterThan(0);
	});

	it('sums of the rollups equal the entity counts over whole hours', async () => {
		const from = new Date(hour.getTime() - Math.min(days, 2) * DAY);
		const summary = await service.summary({ from, to: hour, tz: 'UTC', scope: 'global' });
		const count = (collection: Collection, field: string, filter: Record<string, unknown> = {}) =>
			db.collection(collection).countDocuments({ ...filter, [field]: { $gte: from, $lt: hour } });

		expect(summary.roomsCreated).toBe(await count(Collection.ROOMS, 'createdAt'));
		// Rollups count standard participant sessions only.
		expect(summary.participantSessions).toBe(
			await count(Collection.PARTICIPANTS, 'joinedAt', { kind: ParticipantKind.STANDARD })
		);
		expect(summary.egressStarted).toBe(await count(Collection.EGRESSES, 'startedAt'));
		expect(summary.roomsCreated).toBeGreaterThan(0);
	});

	it('reports the p95 of every budgeted query and fails above its budget', async () => {
		const series = (query: Record<string, string>) => () =>
			service.series(analyticsSeriesQuery(clock).parse(query));
		const rows: [string, number, () => Promise<unknown>][] = [
			['Series, 5m over 24 h', 50, series({ metrics: SERIES_METRICS, from: 'now-24h' })],
			['Series, 1h over 7 d', 50, series({ metrics: SERIES_METRICS, from: 'now-7d', granularity: '1h' })],
			[
				'Series, 1d over 400 d, Europe/Madrid',
				400,
				series({ metrics: SERIES_METRICS, from: 'now-400d', granularity: '1d', tz: 'Europe/Madrid' })
			],
			[
				'Breakdowns, 90 d',
				150,
				() =>
					service.breakdowns(
						analyticsBreakdownsQuery(clock).parse({ dimensions: BREAKDOWNS, from: 'now-90d' })
					)
			],
			['Summary, 90 d', 150, () => service.summary(analyticsSummaryQuery(clock).parse({ from: 'now-90d' }))],
			[
				'CSV, series 5m over 24 h',
				70,
				() =>
					service.export(
						analyticsExportQuery(clock).parse({ kind: 'series', metrics: SERIES_METRICS, from: 'now-24h' })
					)
			]
		];

		report.push('| Query | Budget | p95 |', '| --- | --- | --- |');

		const over: string[] = [];

		for (const [name, budget, run] of rows) {
			const measured = await p95(run);

			report.push(`| ${name} | ${String(budget)} ms | ${measured.toFixed(1)} ms |`);

			if (measured > budget) {
				over.push(`${name}: ${measured.toFixed(1)} ms > ${String(budget)} ms`);
			}
		}

		expectWithinBudgets(over);
	});
});