/**
 * Tests for the fire-and-forget Unix socket emitter (ADR-0010).
 *
 * Test categories:
 *   - Frame round-trip: events reach a local echo server
 *   - session_id stability: same value across emits and reconnects
 *   - Fire-and-forget: returns null quickly when daemon is down
 *   - Reconnect: re-dials transparently after server restart
 *   - Error after close: returns Error for closed emitter
 *   - Validation: caller-bug errors for bad inputs
 *   - Frame size: oversized frames return an Error
 *   - defaultSocketPath: env-var and OS-default resolution
 */

import { unlinkSync } from "node:fs";
import { createServer } from "node:net";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import {
	defaultSocketPath,
	type EmitEvent,
	Emitter,
	MAX_FRAME_SIZE,
	SUPPORTED_FRAME_VERSION,
} from "./emitter.js";

// ─── Helpers ────────────────────────────────────────────────────────────────

/** Monotonic counter so every tempSockPath call returns a unique path. */
let _sockSeq = 0;

/**
 * A unique socket path for each call (pid + counter + suffix).
 *
 * On macOS, os.tmpdir() returns a long path under /var/folders/… that can
 * exceed the AF_UNIX 104-byte limit. Use /tmp directly on darwin to keep
 * paths well under the limit.
 */
function tempSockPath(suffix: string): string {
	const base = process.platform === "darwin" ? "/tmp" : tmpdir();
	return join(base, `ar-${process.pid}-${++_sockSeq}-${suffix}.sock`);
}

/**
 * Minimal echo server that reads length-prefixed frames and collects the JSON
 * body of each frame. Returns the socket path and a way to read received
 * frames.
 *
 * `ready` resolves once the server is actually listening. Tests that emit
 * immediately after construction should await it to avoid a race between the
 * first connect and the server's bind completing.
 */
function startEchoServer(sockPath: string): {
	frames: () => Promise<string[]>;
	stop: () => Promise<void>;
	ready: Promise<void>;
} {
	const received: string[] = [];
	// Track open client sockets so stop() can forcefully destroy them,
	// ensuring the emitter sees the connection die immediately.
	const openSockets = new Set<import("node:net").Socket>();

	const server = createServer((socket) => {
		openSockets.add(socket);
		socket.once("close", () => openSockets.delete(socket));

		let buf = Buffer.alloc(0);
		socket.on("data", (chunk: Buffer) => {
			buf = Buffer.concat([buf, chunk]);
			// Parse as many complete frames as possible.
			while (buf.length >= 4) {
				const len = buf.readUInt32BE(0);
				if (buf.length < 4 + len) {
					break;
				}
				received.push(buf.subarray(4, 4 + len).toString("utf8"));
				buf = buf.subarray(4 + len);
			}
		});
	});

	// Unlink any stale socket file from a previous (possibly crashed) test run
	// so server.listen() doesn't get EADDRINUSE.
	try {
		unlinkSync(sockPath);
	} catch {
		// Socket file didn't exist — fine.
	}

	const ready = new Promise<void>((resolve, reject) => {
		server.once("listening", resolve);
		server.once("error", reject);
	});
	server.listen(sockPath);

	return {
		frames: () =>
			new Promise<string[]>((resolve) => {
				// Give the OS a tick to deliver any in-flight data.
				setTimeout(() => resolve([...received]), 10);
			}),
		stop: () =>
			new Promise<void>((resolve, reject) => {
				// Destroy all open client sockets so the emitter sees the connection
				// die immediately rather than waiting for a graceful FIN.
				for (const s of openSockets) {
					s.destroy();
				}
				server.close((err) => {
					try {
						unlinkSync(sockPath);
					} catch {
						// Already gone — fine.
					}
					err ? reject(err) : resolve();
				});
			}),
		ready,
	};
}

/** Wait up to `ms` for `predicate()` to return true, polling every 5 ms. */
async function waitFor(
	predicate: () => boolean | Promise<boolean>,
	ms = 500,
): Promise<void> {
	const deadline = Date.now() + ms;
	while (Date.now() < deadline) {
		if (await predicate()) {
			return;
		}
		await new Promise((r) => setTimeout(r, 5));
	}
	throw new Error(`waitFor timed out after ${ms}ms`);
}

const GOOD_EVENT: EmitEvent = {
	channel: "test-channel",
	tool: { name: "my_tool" },
	decision: "allowed",
};

// ─── Tests ──────────────────────────────────────────────────────────────────

describe("Emitter — validation errors (caller bugs)", () => {
	it("returns an error for empty channel", async () => {
		const e = new Emitter({ socketPath: tempSockPath("noop") });
		const err = await e.emit({ ...GOOD_EVENT, channel: "" });
		expect(err).toBeInstanceOf(Error);
		expect(err?.message).toMatch(/missing channel/);
		e.close();
	});

	it("returns an error for empty tool.name", async () => {
		const e = new Emitter({ socketPath: tempSockPath("noop") });
		const err = await e.emit({ ...GOOD_EVENT, tool: { name: "" } });
		expect(err).toBeInstanceOf(Error);
		expect(err?.message).toMatch(/missing tool\.name/);
		e.close();
	});

	it("returns an error for invalid decision", async () => {
		const e = new Emitter({ socketPath: tempSockPath("noop") });
		const err = await e.emit({
			...GOOD_EVENT,
			// @ts-expect-error testing invalid decision value
			decision: "maybe",
		});
		expect(err).toBeInstanceOf(Error);
		expect(err?.message).toMatch(/invalid decision/);
		e.close();
	});

	it("returns an error for malformed input JSON", async () => {
		const e = new Emitter({ socketPath: tempSockPath("noop") });
		const err = await e.emit({ ...GOOD_EVENT, input: "{bad json}" });
		expect(err).toBeInstanceOf(Error);
		expect(err?.message).toMatch(/input is not valid JSON/);
		e.close();
	});

	it("returns an error for malformed output JSON", async () => {
		const e = new Emitter({ socketPath: tempSockPath("noop") });
		const err = await e.emit({ ...GOOD_EVENT, output: "[unclosed" });
		expect(err).toBeInstanceOf(Error);
		expect(err?.message).toMatch(/output is not valid JSON/);
		e.close();
	});

	it("returns an error for input containing a non-finite number", async () => {
		// 1e400 overflows float64 to Infinity; JSON.parse accepts it but the
		// daemon's RFC 8785 canonicaliser rejects it — catch it here as a
		// caller-bug Error rather than letting it become a silent drop.
		const e = new Emitter({ socketPath: tempSockPath("noop") });
		const err = await e.emit({ ...GOOD_EVENT, input: '{"n":1e400}' });
		expect(err).toBeInstanceOf(Error);
		expect(err?.message).toMatch(/input is not valid JSON/);
		e.close();
	});

	it("returns an error for output containing a non-finite number", async () => {
		const e = new Emitter({ socketPath: tempSockPath("noop") });
		const err = await e.emit({ ...GOOD_EVENT, output: "[1e400]" });
		expect(err).toBeInstanceOf(Error);
		expect(err?.message).toMatch(/output is not valid JSON/);
		e.close();
	});

	it("returns an error after close", async () => {
		const sockPath = tempSockPath("closed");
		const server = startEchoServer(sockPath);
		await server.ready;
		const e = new Emitter({ socketPath: sockPath });
		e.close();
		const err = await e.emit(GOOD_EVENT);
		expect(err).toBeInstanceOf(Error);
		expect(err?.message).toMatch(/closed/);
		await server.stop();
	});
});

describe("Emitter — frame round-trip", () => {
	let sockPath: string;
	let server: ReturnType<typeof startEchoServer>;
	let emitter: Emitter;

	beforeEach(async () => {
		sockPath = tempSockPath("roundtrip");
		server = startEchoServer(sockPath);
		await server.ready;
		emitter = new Emitter({ socketPath: sockPath });
	});

	afterEach(async () => {
		emitter.close();
		await server.stop();
	});

	it("delivers a frame to the server", async () => {
		const err = await emitter.emit(GOOD_EVENT);
		expect(err).toBeNull();

		await waitFor(async () => (await server.frames()).length > 0);

		const frames = await server.frames();
		expect(frames).toHaveLength(1);
		const f = JSON.parse(frames[0] ?? "{}");
		expect(f.v).toBe(SUPPORTED_FRAME_VERSION);
		expect(f.channel).toBe("test-channel");
		expect(f.tool.name).toBe("my_tool");
		expect(f.decision).toBe("allowed");
	});

	it("frame contains session_id matching emitter.sessionId", async () => {
		await emitter.emit(GOOD_EVENT);
		await waitFor(async () => (await server.frames()).length > 0);

		const frames = await server.frames();
		const f = JSON.parse(frames[0] ?? "{}");
		expect(f.session_id).toBe(emitter.sessionId);
	});

	it("ts_emit is RFC3339Nano UTC", async () => {
		await emitter.emit(GOOD_EVENT);
		await waitFor(async () => (await server.frames()).length > 0);

		const frames = await server.frames();
		const f = JSON.parse(frames[0] ?? "{}");
		// e.g. "2026-05-07T12:34:56.789000000Z"
		expect(f.ts_emit).toMatch(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z$/);
		expect(f.ts_emit.endsWith("Z")).toBe(true);
	});

	it("optional tool.server is included when provided", async () => {
		await emitter.emit({
			...GOOD_EVENT,
			tool: { server: "mcp-server", name: "read_file" },
		});
		await waitFor(async () => (await server.frames()).length > 0);

		const frames = await server.frames();
		const f = JSON.parse(frames[0] ?? "{}");
		expect(f.tool.server).toBe("mcp-server");
		expect(f.tool.name).toBe("read_file");
	});

	it("tool.server is absent when not provided", async () => {
		await emitter.emit(GOOD_EVENT);
		await waitFor(async () => (await server.frames()).length > 0);

		const frames = await server.frames();
		const f = JSON.parse(frames[0] ?? "{}");
		expect(f.tool).not.toHaveProperty("server");
	});

	it("input and output are forwarded as raw JSON values (not double-encoded)", async () => {
		await emitter.emit({
			...GOOD_EVENT,
			input: '{"key":"value"}',
			output: "[1,2,3]",
		});
		await waitFor(async () => (await server.frames()).length > 0);

		const frames = await server.frames();
		const f = JSON.parse(frames[0] ?? "{}");
		expect(f.input).toEqual({ key: "value" });
		expect(f.output).toEqual([1, 2, 3]);
	});

	it("input and output are byte-exact verbatim (whitespace preserved)", async () => {
		// The daemon canonicalises (RFC 8785) and hashes the raw frame bytes,
		// so any whitespace, key-order, or number-formatting changes here would
		// silently change the receipt hash. Pin the verbatim guarantee.
		const rawInput = '{ "b": 2,\n  "a": 1 }';
		const rawOutput = "[\t1.0,\n2,\n3 ]";
		await emitter.emit({
			...GOOD_EVENT,
			input: rawInput,
			output: rawOutput,
		});
		await waitFor(async () => (await server.frames()).length > 0);

		const raw = (await server.frames())[0] ?? "";
		expect(raw).toContain(`"input":${rawInput}`);
		expect(raw).toContain(`"output":${rawOutput}`);
	});

	it("preserves '$' sequences in input/output (no String.replace pattern interpretation)", async () => {
		// String.prototype.replace treats '$&', '$1', etc. as backreferences when
		// the replacement is a string. We use the function form to avoid that —
		// pin it with a payload that would be mangled if we ever regressed.
		const rawInput = '{"price":"$100","ref":"$1$&$\'"}';
		await emitter.emit({ ...GOOD_EVENT, input: rawInput });
		await waitFor(async () => (await server.frames()).length > 0);

		const raw = (await server.frames())[0] ?? "";
		expect(raw).toContain(`"input":${rawInput}`);
	});

	it("omits input and output when not provided", async () => {
		await emitter.emit(GOOD_EVENT);
		await waitFor(async () => (await server.frames()).length > 0);

		const frames = await server.frames();
		const f = JSON.parse(frames[0] ?? "{}");
		expect(f).not.toHaveProperty("input");
		expect(f).not.toHaveProperty("output");
	});

	it("includes error field when provided", async () => {
		await emitter.emit({ ...GOOD_EVENT, error: "tool failed" });
		await waitFor(async () => (await server.frames()).length > 0);

		const frames = await server.frames();
		const f = JSON.parse(frames[0] ?? "{}");
		expect(f.error).toBe("tool failed");
	});

	it("sends multiple frames in order", async () => {
		for (const decision of ["allowed", "denied", "pending"] as const) {
			await emitter.emit({ ...GOOD_EVENT, decision });
		}
		await waitFor(async () => (await server.frames()).length >= 3);

		const frames = await server.frames();
		expect(frames).toHaveLength(3);
		const decisions = frames.map((raw) => JSON.parse(raw).decision);
		expect(decisions).toEqual(["allowed", "denied", "pending"]);
	});
});

describe("Emitter — session_id stability", () => {
	it("session_id is stable across multiple emits", async () => {
		const sockPath = tempSockPath("session");
		const server = startEchoServer(sockPath);
		await server.ready;
		const emitter = new Emitter({ socketPath: sockPath });

		await emitter.emit(GOOD_EVENT);
		await emitter.emit(GOOD_EVENT);
		await waitFor(async () => (await server.frames()).length >= 2);

		const frames = await server.frames();
		const ids = frames.map((raw) => JSON.parse(raw).session_id);
		expect(ids[0]).toBe(ids[1]);
		expect(ids[0]).toBe(emitter.sessionId);

		emitter.close();
		await server.stop();
	});

	it("host-supplied sessionId is used and stable", async () => {
		const sockPath = tempSockPath("supplied-session");
		const server = startEchoServer(sockPath);
		await server.ready;
		const emitter = new Emitter({
			socketPath: sockPath,
			sessionId: "my-session-42",
		});

		expect(emitter.sessionId).toBe("my-session-42");

		await emitter.emit(GOOD_EVENT);
		await waitFor(async () => (await server.frames()).length > 0);

		const frames = await server.frames();
		const f = JSON.parse(frames[0] ?? "{}");
		expect(f.session_id).toBe("my-session-42");

		emitter.close();
		await server.stop();
	});

	it("empty sessionId falls back to a generated UUID", () => {
		const emitter = new Emitter({
			socketPath: tempSockPath("empty-session"),
			sessionId: "",
		});
		expect(emitter.sessionId).toMatch(
			/^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/,
		);
		emitter.close();
	});
});

describe("Emitter — fire-and-forget when daemon is down", () => {
	it("returns null quickly when no daemon is listening", async () => {
		const sockPath = tempSockPath("down");
		const emitter = new Emitter({ socketPath: sockPath });

		const start = Date.now();
		const err = await emitter.emit(GOOD_EVENT);
		const elapsed = Date.now() - start;

		expect(err).toBeNull();
		// Must return within 50ms (dial timeout is 25ms + overhead).
		expect(elapsed).toBeLessThan(150);

		emitter.close();
	});

	it("logs a drop when daemon is down (debugLog is called)", async () => {
		const sockPath = tempSockPath("drop-log");
		const drops: Array<{ message: string; attrs: Record<string, string> }> = [];
		const emitter = new Emitter({
			socketPath: sockPath,
			debugLog: (message, attrs) => drops.push({ message, attrs }),
		});

		await emitter.emit(GOOD_EVENT);

		expect(drops).toHaveLength(1);
		expect(drops[0]?.message).toMatch(/dropped event/);
		expect(drops[0]?.attrs.stage).toBe("dial");

		emitter.close();
	});
});

describe("Emitter — reconnect after daemon restart", () => {
	it("re-dials after the server stops and restarts", async () => {
		const sockPath = tempSockPath("reconnect");
		const server1 = startEchoServer(sockPath);
		await server1.ready;
		const emitter = new Emitter({ socketPath: sockPath });

		// First emit reaches server1.
		await emitter.emit(GOOD_EVENT);
		await waitFor(async () => (await server1.frames()).length > 0);
		expect(await server1.frames()).toHaveLength(1);

		// Stop server1 and emit — connection is broken, drop silently.
		await server1.stop();
		const errAfterStop = await emitter.emit(GOOD_EVENT);
		expect(errAfterStop).toBeNull();

		// Restart on the same socket path, then emit again — should re-dial.
		const server2 = startEchoServer(sockPath);
		await server2.ready;

		const errAfterRestart = await emitter.emit(GOOD_EVENT);
		expect(errAfterRestart).toBeNull();

		await waitFor(async () => (await server2.frames()).length > 0);
		expect(await server2.frames()).toHaveLength(1);

		emitter.close();
		await server2.stop();
	});
});

describe("Emitter — frame size limit", () => {
	it("returns an error for oversized frames", async () => {
		const sockPath = tempSockPath("oversize");
		const emitter = new Emitter({ socketPath: sockPath });
		// Construct a JSON string that, when marshalled into the full frame, exceeds 1 MiB.
		// A 1 MiB payload string alone is sufficient.
		const bigInput = JSON.stringify("x".repeat(MAX_FRAME_SIZE));
		const err = await emitter.emit({ ...GOOD_EVENT, input: bigInput });
		expect(err).toBeInstanceOf(Error);
		expect(err?.message).toMatch(/frame too large/);
		emitter.close();
	});
});

describe("defaultSocketPath", () => {
	it("respects AGENTRECEIPTS_SOCKET env var", () => {
		const original = process.env.AGENTRECEIPTS_SOCKET;
		process.env.AGENTRECEIPTS_SOCKET = "/custom/path.sock";
		try {
			expect(defaultSocketPath()).toBe("/custom/path.sock");
		} finally {
			if (original === undefined) {
				delete process.env.AGENTRECEIPTS_SOCKET;
			} else {
				process.env.AGENTRECEIPTS_SOCKET = original;
			}
		}
	});

	it("returns a non-empty path on this platform (darwin or linux)", () => {
		const original = process.env.AGENTRECEIPTS_SOCKET;
		delete process.env.AGENTRECEIPTS_SOCKET;
		try {
			const p = defaultSocketPath();
			// macOS and Linux both have defaults; only 'other' platforms return "".
			const os = process.platform;
			if (os === "darwin" || os === "linux") {
				expect(p).not.toBe("");
				expect(p).toMatch(/events\.sock$/);
			}
		} finally {
			if (original !== undefined) {
				process.env.AGENTRECEIPTS_SOCKET = original;
			}
		}
	});
});

describe("Emitter — constructor", () => {
	it("throws when socketPath is an empty string and no env override is set", () => {
		// An explicit empty socketPath replicates the same failure code path
		// that occurs on platforms where defaultSocketPath() returns "".
		// (??-fallback only triggers on null/undefined, so "" reaches the
		// !socketPath guard.) The constructor must reject this with an
		// actionable error rather than silently producing an unusable emitter.
		const original = process.env.AGENTRECEIPTS_SOCKET;
		delete process.env.AGENTRECEIPTS_SOCKET;
		try {
			expect(() => new Emitter({ socketPath: "" })).toThrow(
				/no default socket path/,
			);
		} finally {
			if (original !== undefined) {
				process.env.AGENTRECEIPTS_SOCKET = original;
			}
		}
	});

	it("does not throw when given a valid explicit socket path", () => {
		expect(() => new Emitter({ socketPath: "/tmp/test.sock" })).not.toThrow();
	});
});

describe("Emitter — socket-error robustness", () => {
	// Reach into the private conn to simulate Node firing an 'error' event on
	// the underlying socket (peer reset, daemon crash, EPIPE). Without a
	// permanent error listener the host process would crash via Node's
	// unhandled-'error' rule. These tests pin that contract.
	function getConn(emitter: Emitter): import("node:net").Socket | null {
		// @ts-expect-error accessing private conn for test assertions
		return emitter.conn;
	}

	it("surviving an 'error' event on a live conn does not crash the process", async () => {
		const sockPath = tempSockPath("err-survive");
		const server = startEchoServer(sockPath);
		await server.ready;
		const drops: string[] = [];
		const emitter = new Emitter({
			socketPath: sockPath,
			debugLog: (_msg, attrs) => {
				if (attrs.stage) drops.push(attrs.stage);
			},
		});

		// Establish a live connection.
		await emitter.emit(GOOD_EVENT);
		await waitFor(async () => (await server.frames()).length > 0);

		const conn = getConn(emitter);
		expect(conn).not.toBeNull();
		// At least one listener must be attached for 'error' so the next
		// peer-side mishap cannot crash the host.
		expect(conn?.listenerCount("error")).toBeGreaterThan(0);

		// Simulate a peer reset. Without the permanent listener this would
		// throw an unhandled 'error' and tear the process down.
		conn?.emit("error", new Error("synthetic ECONNRESET"));

		// Listener should have logged the drop and cleared the conn.
		expect(drops).toContain("socket");
		expect(getConn(emitter)).toBeNull();

		emitter.close();
		await server.stop();
	});

	it("re-dials transparently after a socket error event", async () => {
		const sockPath = tempSockPath("err-redial");
		const server = startEchoServer(sockPath);
		await server.ready;
		const emitter = new Emitter({ socketPath: sockPath });

		// First emit dials and delivers.
		await emitter.emit(GOOD_EVENT);
		await waitFor(async () => (await server.frames()).length > 0);

		// Inject a synthetic error on the live conn — the permanent listener
		// must clear it without throwing.
		getConn(emitter)?.emit("error", new Error("synthetic peer reset"));

		// Next emit should re-dial and deliver another frame to the SAME
		// echo server (it never went away).
		await emitter.emit(GOOD_EVENT);
		await waitFor(async () => (await server.frames()).length >= 2);
		expect(await server.frames()).toHaveLength(2);

		emitter.close();
		await server.stop();
	});

	it("is safe to close while a dial is in flight", async () => {
		// Connect to a server that exists, then call close() before the
		// connect callback has fired. The freshly connected socket must be
		// destroyed and not retained.
		const sockPath = tempSockPath("close-during-dial");
		const server = startEchoServer(sockPath);
		await server.ready;
		const emitter = new Emitter({ socketPath: sockPath });

		const emitPromise = emitter.emit(GOOD_EVENT);
		// Close immediately. Depending on timing this may or may not race
		// the connect callback, but neither outcome must crash or leak.
		emitter.close();

		const result = await emitPromise;
		// Either null (silently dropped) or the closed Error — both acceptable.
		if (result !== null) {
			expect(result.message).toMatch(/closed/);
		}

		await server.stop();
	});
});
