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

import { writeFileSync } from 'node:fs';

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

import { Collection, type IngressDocument } from '@openvidu-dashboard/typings';

import { rankedPipeline } from '../../src/api/ranked.js';
import { INGRESS_LIST_PROJECTION } from '../../src/ingress/ingress-mappers.js';
import { activityRange, ingressesListPlan } from '../../src/ingress/ingress-query.js';
import { ingressesListQuery } from '../../src/ingress/ingress-schemas.js';
import { IngressService } from '../../src/ingress/ingress-service.js';
import { expectIdLookup, expectIndexBacked, expectIndexes, p95 } from '../support/query-plan.js';
import { expectWithinBudgets } from '../support/latency-budget.js';

/**
 * Every ingress query explained, filter by filter, on a database generated by the data generator, plus
 * the p95 of the performance budget, which fails the suite when exceeded. INGRESS_PLAN_SCENARIO (S by default),
 * INGRESS_PLAN_DAYS (2 by default) and INGRESS_PLAN_REPORT (a file that also receives the report) work as in the
 * rooms suite.
 */

const mongoUri = inject('mongoUri');
const scenario = (process.env.INGRESS_PLAN_SCENARIO ?? 'S') as 'S' | 'M' | 'L';
const days = Number(process.env.INGRESS_PLAN_DAYS ?? '2');
const DATABASE = 'dashboard_test_ingress_plan';
const LIMIT = 100;
const DAY = 86_400_000;
const GENERATION_TIMEOUT_MS = scenario === 'S' ? 600_000 : 3_600_000;

describe.skipIf(mongoUri === '')(`ingress query plan at scenario ${scenario}`, () => {
	let client: MongoClient;
	let db: Db;
	let ctx: QueryContext;
	let service: IngressService;
	let sample: IngressDocument;
	let namePrefix: string;
	const now = new Date();
	const report: string[] = [];

	const listParams = (query: Record<string, string>) =>
		ingressesListQuery(() => now).parse({ from: ctx.range.from.toISOString(), limit: String(LIMIT), ...query });
	const raw = () => db.collection(Collection.INGRESSES);
	const typed = () => db.collection<IngressDocument>(Collection.INGRESSES);
	/**
	 * The first page of a list request as the service ranks it, examining at most the ingresses of each branch of the
	 * activity range: every branch fetches its own matches, so an ingress in two branches is examined twice, and the
	 * status, input type and name filters apply to the union.
	 */
	const explainList = async (query: Record<string, string>) => {
		const params = listParams(query);
		const plan = ingressesListPlan(params);

		if (plan.kind !== 'ranked') {
			throw new Error('expected a ranked plan');
		}

		const request = { ...params, match: plan.match, rank: plan.rank, projection: INGRESS_LIST_PROJECTION };
		const branches = (activityRange(params.from, params.to) as { $or: Document[] }).$or;
		const counts = await Promise.all(branches.map((branch) => raw().countDocuments(branch)));

		return expectIndexBacked(
			db,
			{ collection: Collection.INGRESSES, pipeline: rankedPipeline({ ...request, cursor: undefined, now }, now) },
			{ maxDocsExamined: counts.reduce((sum, count) => sum + count, 0) }
		);
	};

	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
		});
		ctx = await queryContext(db);
		service = new IngressService(db, { totalMaxRangeMs: 7 * DAY, now: () => new Date() });

		const found = await db
			.collection<IngressDocument>(Collection.INGRESSES)
			.findOne({ _id: ctx.samples.ingressId });

		if (found === null) {
			throw new Error('the generated database has no sample ingress');
		}

		if (found.nameLower === undefined) {
			throw new Error('the sample ingress has no nameLower');
		}

		sample = found;
		namePrefix = found.nameLower.slice(0, 3);
	}, GENERATION_TIMEOUT_MS);

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

			console.log(text);

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

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

	it('lists: the activity range as three index-backed branches, at most the ingresses in it', async () => {
		expectIndexes(await explainList({}), ['createdAt_id', 'lastSessionAt_id', 'status_createdAt']);

		for (const query of [
			{ inputType: 'whip' },
			{ status: 'publishing' },
			{ q: namePrefix },
			{ sort: 'totalDuration' },
			{ sort: 'sessions' },
			{ sort: 'lastSession', order: 'asc' }
		]) {
			expect((await explainList(query)).collscan).toBe(false);
		}
	});

	it('id, detail and summary: _id, status_createdAt and createdAt_id', async () => {
		expectIdLookup(await expectIndexBacked(db, typed().find({ _id: sample._id }).limit(1), { maxDocsExamined: 1 }));
		expect(ingressesListPlan(listParams({ q: sample._id }))).toEqual({ kind: 'id', filter: { _id: sample._id } });
		expectIndexes(
			await expectIndexBacked(db, raw().find({ status: 'publishing' }, { projection: { _id: 1 } }), {
				maxDocsExamined: () => Number.MAX_SAFE_INTEGER
			}),
			['status_createdAt']
		);

		const created = await expectIndexBacked(
			db,
			{
				collection: Collection.INGRESSES,
				pipeline: [
					{ $match: activityRange(new Date(now.getTime() - DAY), now) },
					{ $group: { _id: '$inputType', n: { $sum: 1 } } }
				]
			},
			{ maxDocsExamined: () => Number.MAX_SAFE_INTEGER, collscanThreshold: 0 }
		);

		expect(created.collscan).toBe(false);
	});

	it('reports the p95 of every budgeted query and fails above its budget', async () => {
		// The metric sorts accept up to the ranked-sort limit of the service.
		const week = new Date(now.getTime() - 7 * DAY).toISOString();
		const rows: [string, number, () => Promise<unknown>][] = [
			['List, status all (100 rows)', 100, () => service.list(listParams({}))],
			['List, publishing', 100, () => service.list(listParams({ status: 'publishing' }))],
			['List, input type', 100, () => service.list(listParams({ inputType: 'url' }))],
			['List, name prefix', 100, () => service.list(listParams({ q: namePrefix }))],
			[
				'List, ranked by total duration (7 d)',
				200,
				() => service.list(listParams({ sort: 'totalDuration', from: week }))
			],
			['List, ranked by sessions (7 d)', 200, () => service.list(listParams({ sort: 'sessions', from: week }))],
			[
				'List, ranked by last session (7 d)',
				200,
				() => service.list(listParams({ sort: 'lastSession', from: week }))
			],
			[
				'List, created over 30 days',
				200,
				() => service.list(listParams({ from: new Date(now.getTime() - 30 * DAY).toISOString() }))
			],
			['Summary (24 h)', 100, () => service.summary({ from: new Date(now.getTime() - DAY), to: now })]
		];

		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);
	});
});