pi-ext

import assert from "node:assert/strict";
import { existsSync } from "node:fs";
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
import { test } from "vitest";
import { Browser, executeOperations } from "../browser.ts";
import { type Frame, TMP_PREFIX } from "../lib.ts";

test("wait before its labeled command fails before sending anything", async () => {
	const b = new Browser();
	const result = await executeOperations(b, [
		{
			kind: "wait",
			after: "nav",
			event: "Page.loadEventFired",
			sessionId: "session",
		},
		{
			kind: "cdp",
			method: "Page.navigate",
			label: "nav",
			sessionId: "session",
			params: { url: "about:blank" },
		},
	]);
	assert.match(result.failure || "", /wait must follow a labeled cdp command/);
	assert.deepEqual(result.output, []);
});

test("batch failure identifies failed operation and skipped operations", async () => {
	const b = new Browser();
	const sent: string[] = [];
	b.send = async (method): Promise<Frame> => {
		sent.push(method);
		return method === "Browser.fail"
			? { id: sent.length, error: { code: -1, message: "expected failure" } }
			: { id: sent.length, result: {} };
	};
	const result = await executeOperations(b, [
		{ kind: "cdp", method: "Browser.getVersion" },
		{ kind: "cdp", method: "Browser.fail" },
		{ kind: "status" },
	]);
	assert.equal(result.failure, "expected failure");
	assert.deepEqual(result.failedOperation, { operation: 1, kind: "cdp" });
	assert.deepEqual(result.skipped, [{ operation: 2, kind: "status" }]);
	assert.deepEqual(sent, ["Browser.getVersion", "Browser.fail"]);
	await b.close();
});

test("multiple waits sharing an after label fail before sending", async () => {
	const b = new Browser();
	let sent = 0;
	b.send = async (): Promise<Frame> => {
		sent++;
		return { id: sent, result: {} };
	};
	const result = await executeOperations(b, [
		{ kind: "cdp", method: "Browser.getVersion", label: "ready" },
		{
			kind: "wait",
			after: "ready",
			event: "Page.loadEventFired",
			sessionId: "s",
		},
		{
			kind: "wait",
			after: "ready",
			event: "Page.frameStoppedLoading",
			sessionId: "s",
		},
	]);
	assert.match(result.failure || "", /multiple waits use after label: ready/);
	assert.deepEqual(result.failedOperation, { operation: 2, kind: "wait" });
	assert.deepEqual(result.skipped, []);
	assert.equal(sent, 0);
	await b.close();
});

test("multiple batch waits match their events and release their observers", async () => {
	const b = new Browser();
	let id = 0;
	b.send = async (): Promise<Frame> => {
		b.onMessage(
			JSON.stringify({
				method: "Page.loadEventFired",
				sessionId: "s",
				params: { number: ++id },
			}),
		);
		return { id, result: {} };
	};
	const result = await executeOperations(b, [
		{ kind: "cdp", method: "Page.navigate", sessionId: "s", label: "first" },
		{
			kind: "wait",
			after: "first",
			event: "Page.loadEventFired",
			sessionId: "s",
		},
		{ kind: "cdp", method: "Page.navigate", sessionId: "s", label: "second" },
		{
			kind: "wait",
			after: "second",
			event: "Page.loadEventFired",
			sessionId: "s",
		},
	]);
	assert.equal(result.failure, undefined);
	assert.deepEqual(
		[result.output[1], result.output[3]].map(
			(item) => (item as { wait: { seq: number } }).wait.seq,
		),
		[1, 2],
	);
	assert.equal(b.watchers.length, 0);
});

test("event overflow spools exact messages and reports path", async () => {
	const b = new Browser();
	b.spoolDir = await mkdtemp(TMP_PREFIX);
	try {
		for (let i = 0; i < 230; i++)
			b.onMessage(
				JSON.stringify({
					method: "Log.entryAdded",
					sessionId: "s",
					params: { text: `entry-${i}` },
				}),
			);
		const res = await executeOperations(b, []);
		assert.ok(res.overflow.length > 0);
		assert.ok(res.events.length < 230);
		assert.match(await readFile(res.overflow[0], "utf8"), /entry-0/);
		b.onMessage(
			JSON.stringify({
				method: "Log.entryAdded",
				sessionId: "s",
				params: { text: "x".repeat(18000) },
			}),
		);
		const big = await executeOperations(b, []);
		assert.equal(big.events.at(-1)?.frame, undefined);
		assert.match(await readFile(big.overflow[0], "utf8"), /"text":"xxx/);
		const path = big.overflow[0];
		await b.close();
		assert.equal(existsSync(path), false);
	} finally {
		await b.close();
		await rm(b.spoolDir, { recursive: true, force: true });
	}
});

test("labeled small replies remain referencable; large replies spool unchanged", async () => {
	const b = new Browser();
	b.spoolDir = await mkdtemp(TMP_PREFIX);
	const small: Frame = { id: 1, result: { value: 42 } };
	const large: Frame = { id: 2, result: { data: "x".repeat(512 * 1024) } };
	b.send = async (method) => (method === "Browser.getVersion" ? small : large);
	try {
		const result = await executeOperations(b, [
			{ kind: "cdp", method: "Browser.getVersion", label: "small" },
			{ kind: "cdp", method: "Browser.getHistograms", label: "large" },
		]);
		assert.equal(result.failure, undefined);
		assert.deepEqual((result.output[0] as { raw: Frame }).raw, small);
		assert.deepEqual(b.results.get("small"), small);
		assert.equal(b.results.has("large"), false);
		const output = (
			result.output[1] as {
				raw: { path: string; bytes: number; preview: string };
			}
		).raw;
		assert.equal(await readFile(output.path, "utf8"), JSON.stringify(large));
		assert.equal(output.bytes, Buffer.byteLength(JSON.stringify(large)));
		assert.equal(output.preview, JSON.stringify(large).slice(0, 512));
	} finally {
		await b.close();
	}
});

test("payload spooling expires old event files absent from this response", async () => {
	const b = new Browser();
	b.spoolDir = await mkdtemp(TMP_PREFIX);
	const response: Frame = {
		id: 1,
		result: { data: "r".repeat(40 * 1024) },
	};
	b.send = async (): Promise<Frame> => response;
	try {
		b.onMessage(
			JSON.stringify({
				method: "Log.entryAdded",
				sessionId: "s",
				params: { text: "e".repeat(16 * 1024 * 1024 - 128 * 1024) },
			}),
		);
		await b.spoolWrites;
		const eventPath = b.events[0]?.spooled;
		assert.ok(eventPath);
		b.cursor = b.seq;
		const result = await executeOperations(b, [
			{ kind: "cdp", method: "Browser.getVersion" },
			{ kind: "cdp", method: "Browser.getVersion" },
			{ kind: "cdp", method: "Browser.getVersion" },
			{ kind: "cdp", method: "Browser.getVersion" },
		]);
		assert.deepEqual(result.overflow, []);
		assert.equal(JSON.stringify(result).includes(eventPath), false);
		const { images: _images, ...text } = result;
		const resultSpool = await b.spoolPayload(JSON.stringify(text));
		assert.equal(existsSync(eventPath), false);
		assert.equal(existsSync(resultSpool), true);
	} finally {
		await b.close();
		await rm(b.spoolDir, { recursive: true, force: true });
	}
});

test("replay reports event files evicted between batches as lost", async () => {
	const b = new Browser();
	b.spoolDir = await mkdtemp(TMP_PREFIX);
	try {
		b.onMessage(
			JSON.stringify({
				method: "Log.entryAdded",
				sessionId: "s",
				params: { text: "e".repeat(9 * 1024 * 1024) },
			}),
		);
		await b.spoolWrites;
		const eventPath = b.events[0]?.spooled;
		assert.ok(eventPath);
		b.cursor = b.seq;
		await executeOperations(b, []);
		await b.spoolPayload("p".repeat(9 * 1024 * 1024));
		assert.equal(existsSync(eventPath), false);
		const replay = await executeOperations(b, [], undefined, 0);
		assert.deepEqual(replay.overflow, []);
		assert.equal(replay.events[0]?.lost, true);
		assert.equal(replay.events[0]?.spooled, undefined);
	} finally {
		await b.close();
		await rm(b.spoolDir, { recursive: true, force: true });
	}
});

test("replayed event spool paths survive response quota exhaustion", async () => {
	const b = new Browser();
	b.spoolDir = await mkdtemp(TMP_PREFIX);
	const eventData = "e".repeat(2 * 1024 * 1024);
	const response: Frame = {
		id: 1,
		result: { data: "r".repeat(15 * 1024 * 1024) },
	};
	b.send = async (): Promise<Frame> => response;
	try {
		b.onMessage(
			JSON.stringify({
				method: "Log.entryAdded",
				sessionId: "s",
				params: { text: eventData },
			}),
		);
		await b.spoolWrites;
		const eventPath = b.events[0]?.spooled;
		assert.ok(eventPath);
		b.cursor = b.seq;
		const result = await executeOperations(
			b,
			[{ kind: "cdp", method: "Browser.getVersion" }],
			undefined,
			0,
		);
		assert.match(result.failure || "", /available spool quota/);
		assert.equal(result.events[0]?.spooled, eventPath);
		assert.equal(existsSync(eventPath), true);
	} finally {
		await b.close();
		await rm(b.spoolDir, { recursive: true, force: true });
	}
});

test("queued event spooling remains protected when a batch starts", async () => {
	const b = new Browser();
	b.spoolDir = await mkdtemp(TMP_PREFIX);
	const response: Frame = {
		id: 1,
		result: { data: "r".repeat(15 * 1024 * 1024) },
	};
	b.send = async (): Promise<Frame> => response;
	try {
		b.onMessage(
			JSON.stringify({
				method: "Log.entryAdded",
				sessionId: "s",
				params: { text: "e".repeat(2 * 1024 * 1024) },
			}),
		);
		const result = await executeOperations(b, [
			{ kind: "cdp", method: "Browser.getVersion" },
		]);
		assert.match(result.failure || "", /available spool quota/);
		const eventPath = result.events[0]?.spooled;
		assert.ok(eventPath);
		assert.equal(existsSync(eventPath), true);
	} finally {
		await b.close();
		await rm(b.spoolDir, { recursive: true, force: true });
	}
});

test("current batch spool paths survive quota exhaustion", async () => {
	const b = new Browser();
	b.spoolDir = await mkdtemp(TMP_PREFIX);
	const frame: Frame = { id: 1, result: { data: "x".repeat(9 * 1024 * 1024) } };
	let calls = 0;
	b.send = async (): Promise<Frame> => ({ ...frame, id: ++calls });
	try {
		const result = await executeOperations(b, [
			{ kind: "cdp", method: "Browser.getVersion", label: "first" },
			{ kind: "cdp", method: "Browser.getVersion", label: "second" },
			{ kind: "status" },
		]);
		assert.match(result.failure || "", /available spool quota/);
		assert.deepEqual(result.failedOperation, { operation: 1, kind: "cdp" });
		assert.deepEqual(result.skipped, [{ operation: 2, kind: "status" }]);
		const first = (result.output[0] as { raw: { path: string } }).raw.path;
		assert.equal(existsSync(first), true);
		assert.equal(
			(await readFile(first, "utf8")).length > 9 * 1024 * 1024,
			true,
		);
	} finally {
		await b.close();
		await rm(b.spoolDir, { recursive: true, force: true });
	}
});

test("video size cap does not throw in event handler", async () => {
	const b = new Browser();
	b.spoolDir = await mkdtemp(TMP_PREFIX);
	try {
		let stopped = false;
		b.video = {
			sessionId: "s",
			size: 256 * 1024 * 1024,
			droppedFrames: 0,
			file: "ignored",
			process: { stdin: { writable: false } } as never,
			done: Promise.resolve(0),
			timer: setTimeout(() => {}, 100),
		};
		b.stopVideo = async () => {
			stopped = true;
			b.video = undefined;
			return "ignored";
		};
		assert.doesNotThrow(() =>
			b.onMessage(
				JSON.stringify({
					method: "Page.screencastFrame",
					sessionId: "s",
					params: { data: Buffer.from("abc").toString("base64"), sessionId: 1 },
				}),
			),
		);
		await new Promise((resolve) => setImmediate(resolve));
		assert.equal(stopped, true);
	} finally {
		await b.close();
		await rm(b.spoolDir, { recursive: true, force: true });
	}
});

test("concurrent video finalization and close share encoder completion", async () => {
	const b = new Browser();
	b.spoolDir = await mkdtemp(TMP_PREFIX);
	const file = `${b.spoolDir}/capture.webm`;
	await writeFile(file, "partial");
	let finish!: (code: number) => void;
	const done = new Promise<number>((resolve) => {
		finish = resolve;
	});
	let ends = 0;
	b.video = {
		file,
		sessionId: "s",
		size: 5,
		droppedFrames: 0,
		timer: setTimeout(() => {}, 5000),
		process: {
			stdin: {
				end: () => {
					ends++;
				},
			},
		} as never,
		done,
	};
	try {
		const a = b.stopVideo();
		const second = b.stopVideo();
		const closing = b.close();
		assert.equal(ends, 1);
		finish(0);
		assert.equal(await a, file);
		assert.equal(await second, file);
		await closing;
		assert.equal(ends, 1);
		assert.equal(b.closed, true);
	} finally {
		await b.close();
		await rm(b.spoolDir, { recursive: true, force: true });
	}
});

test("concurrent trace stops share one Tracing.end and one stream", async () => {
	const b = new Browser();
	b.spoolDir = await mkdtemp(TMP_PREFIX);
	const file = `${b.spoolDir}/trace.json`;
	let complete!: (v: Record<string, unknown>) => void;
	let ends = 0,
		reads = 0,
		closes = 0;
	b.trace = {
		file,
		timer: setTimeout(() => {}, 5000),
		done: new Promise((resolve) => {
			complete = resolve as typeof complete;
		}),
		finish: (v) => complete(v),
	};
	b.send = async (method, params): Promise<Frame> => {
		if (method === "Tracing.end") ends++;
		if (method === "IO.close") closes++;
		if (method === "IO.read") {
			assert.deepEqual(params, { handle: "trace-stream", size: 1024 * 1024 });
			return {
				id: 1,
				result: reads++
					? {
							data: Buffer.from("[]}").toString("base64"),
							base64Encoded: true,
							eof: true,
						}
					: { data: '{"traceEvents":' },
			};
		}
		return { id: 1, result: {} };
	};
	try {
		const a = b.stopTrace();
		const second = b.stopTrace();
		await new Promise((resolve) => setImmediate(resolve));
		complete({ stream: "trace-stream" });
		assert.equal(await a, file);
		assert.equal(await second, file);
		assert.equal(ends, 1);
		assert.equal(reads, 2);
		assert.equal(closes, 1);
		assert.deepEqual(JSON.parse(await readFile(file, "utf8")), {
			traceEvents: [],
		});
	} finally {
		await b.close();
		await rm(b.spoolDir, { recursive: true, force: true });
	}
});

test("failed trace export closes the CDP stream and removes incomplete file", async () => {
	const b = new Browser();
	b.spoolDir = await mkdtemp(TMP_PREFIX);
	const file = `${b.spoolDir}/incomplete.json`;
	let closed = 0,
		reads = 0;
	b.trace = {
		file,
		timer: setTimeout(() => {}, 5000),
		done: Promise.resolve({ stream: "trace-stream" }),
		finish: () => {},
	};
	b.send = async (method): Promise<Frame> => {
		if (method === "IO.close") closed++;
		if (method === "IO.read")
			return reads++
				? { id: 1, error: { code: -1, message: "stream lost" } }
				: { id: 1, result: { data: '{"traceEvents":' } };
		return { id: 1, result: {} };
	};
	try {
		await assert.rejects(b.stopTrace(), /stream lost/);
		assert.equal(closed, 1);
		assert.equal(existsSync(file), false);
	} finally {
		await b.close();
		await rm(b.spoolDir, { recursive: true, force: true });
	}
});