import { describe, expect, it, jest } from "@jest/globals";
import { WorkerPipeline } from "../../../src/worker/worker.pipeline.ts";
import type { Monitor } from "../../../src/domain/monitors/monitor.type.ts";
import type { Job } from "../../../src/domain/jobs/job.type.ts";
import type { Check } from "../../../src/domain/checks/check.type.ts";
import type { StatusChangeResult } from "../../../src/types/network.ts";
import { createMockLogger } from "../../helpers/createMockLogger.ts";
import { NETWORK_ERROR } from "../../../src/types/network.ts";

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

const makeMonitor = (overrides?: Partial<Monitor>): Monitor =>
	({
		id: "m1",
		teamId: "team",
		type: "http",
		interval: 60000,
		status: "up",
		...overrides,
	}) as Monitor;

const makeJob = (overrides?: Partial<Job>): Job => ({
	id: "check:m1",
	type: "check",
	refId: "m1",
	isActive: true,
	nextScheduledAt: 1000,
	intervalMs: 60000,
	lockedBy: null,
	lockedUntil: null,
	pendingChecks: [],
	runCount: 0,
	failCount: 0,
	lastFinishedAt: null,
	lastFailReason: null,
	...overrides,
});

// An evaluate row as ingest leaves it: one pending entry per stored check
const makeEvaluateJob = (...checkIds: string[]): Job =>
	makeJob({
		id: "evaluate:m1",
		type: "evaluate",
		intervalMs: null,
		pendingChecks: checkIds.map((checkId, i) => ({ checkId, createdAt: 1000 + i })),
	});

const makeCheck = (overrides?: Partial<Check>): Check =>
	({
		id: "check-1",
		metadata: { monitorId: "mon-1", teamId: "team-1", type: "http" },
		status: true,
		statusCode: 200,
		responseTime: 100,
		message: "OK",
		createdAt: "2026-01-01T00:00:00.000Z",
		updatedAt: "2026-01-01T00:00:00.000Z",
		...overrides,
	}) as Check;

const makeStatusChange = (overrides: Partial<StatusChangeResult> & { status?: string }): StatusChangeResult => {
	const { status = "up", ...rest } = overrides;
	return {
		monitor: { id: "m1", status } as any,
		statusChanged: false,
		prevStatus: "up",
		...rest,
	} as StatusChangeResult;
};

// A maintenance window straddling "now" — active per isWindowActive.
const activeWindow = () => {
	const now = Date.now();
	return { active: true, start: new Date(now - 1000).toISOString(), end: new Date(now + 1000).toISOString(), repeat: 0 };
};

const createPipeline = (overrides?: Record<string, any>) => {
	const deps = {
		logger: createMockLogger(),
		monitorsRepository: {
			findByIdLean: jest.fn<any>().mockResolvedValue(makeMonitor()),
			updateById: jest.fn<any>().mockResolvedValue({}),
			findDockerTlsKeyById: jest.fn<any>().mockResolvedValue(null),
		},
		maintenanceWindowsRepository: { findByMonitorId: jest.fn<any>().mockResolvedValue([]) },
		checksRepository: { findUnevaluatedByMonitorId: jest.fn<any>().mockResolvedValue([]) },
		jobsRepository: {
			upsertEvaluate: jest.fn<any>().mockResolvedValue(true),
			pullEvaluated: jest.fn<any>().mockResolvedValue(true),
		},
		checkService: {
			toCheck: jest.fn<any>().mockReturnValue({ id: "check-1" }),
			createChecks: jest.fn<any>().mockResolvedValue([]),
		},
		providerRegistry: { probe: jest.fn<any>().mockResolvedValue({ monitorId: "m1", status: true, code: 200, message: "OK" }) },
		proxyResolver: { resolve: jest.fn<any>().mockResolvedValue(undefined) },
		bufferService: { addToBuffer: jest.fn<any>(), addDockerLogToBuffer: jest.fn<any>() },
		dockerLogsService: { buildDockerLogs: jest.fn<any>().mockResolvedValue([]) },
		egressService: { assessAfterFailure: jest.fn<any>().mockResolvedValue(null) },
		statusService: { updateMonitorStatus: jest.fn<any>().mockResolvedValue(makeStatusChange({})) },
		dispatcher: { dispatch: jest.fn<any>().mockResolvedValue(undefined) },
		...overrides,
	};
	const pipeline = new WorkerPipeline(deps as any);
	return { pipeline, deps };
};

// ── Stage 1: check job ───────────────────────────────────────────────────────

describe("WorkerPipeline", () => {
	describe("handleCheck", () => {
		it("loads the monitor by refId and produces a check for it", async () => {
			const { pipeline, deps } = createPipeline();

			await pipeline.handleCheck(makeJob());

			expect(deps.monitorsRepository.findByIdLean).toHaveBeenCalledWith("m1");
			expect(deps.providerRegistry.probe).toHaveBeenCalled();
			expect(deps.bufferService.addToBuffer).toHaveBeenCalledWith({ id: "check-1" });
		});

		it("does nothing for a job without a refId", async () => {
			const { pipeline, deps } = createPipeline();

			await pipeline.handleCheck(makeJob({ refId: null }));

			expect(deps.monitorsRepository.findByIdLean).not.toHaveBeenCalled();
			expect(deps.providerRegistry.probe).not.toHaveBeenCalled();
		});

		it("does nothing when the monitor no longer exists", async () => {
			const { pipeline, deps } = createPipeline({
				monitorsRepository: { findByIdLean: jest.fn<any>().mockResolvedValue(null) },
			});

			await pipeline.handleCheck(makeJob());

			expect(deps.providerRegistry.probe).not.toHaveBeenCalled();
		});
	});

	describe("produce", () => {
		it("throws when monitor id is missing", async () => {
			const { pipeline } = createPipeline();
			await expect(pipeline.produce({} as Monitor)).rejects.toThrow("No monitor id");
		});

		// ── maintenance gate ──────────────────────────────────────────────────

		it("skips the check and flips status to 'maintenance' when in a maintenance window", async () => {
			const { pipeline, deps } = createPipeline({
				maintenanceWindowsRepository: { findByMonitorId: jest.fn<any>().mockResolvedValue([activeWindow()]) },
			});

			const result = await pipeline.produce(makeMonitor({ status: "up" }));

			expect(result).toBeNull();
			expect(deps.monitorsRepository.updateById).toHaveBeenCalledWith("m1", "team", { status: "maintenance", statusWindow: [] });
			expect(deps.providerRegistry.probe).not.toHaveBeenCalled();
			expect(deps.bufferService.addToBuffer).not.toHaveBeenCalled();
		});

		it("does not re-write status when already in maintenance", async () => {
			const { pipeline, deps } = createPipeline({
				maintenanceWindowsRepository: { findByMonitorId: jest.fn<any>().mockResolvedValue([activeWindow()]) },
			});

			const result = await pipeline.produce(makeMonitor({ status: "maintenance" }));

			expect(result).toBeNull();
			expect(deps.monitorsRepository.updateById).not.toHaveBeenCalled();
		});

		it("moves a monitor out of maintenance explicitly once the window is over", async () => {
			const monitor = makeMonitor({ status: "maintenance" });
			const { pipeline, deps } = createPipeline({
				monitorsRepository: {
					updateById: jest.fn<any>().mockResolvedValue({ ...monitor, status: "initializing" }),
					findDockerTlsKeyById: jest.fn<any>().mockResolvedValue(null),
				},
			});

			await pipeline.produce(monitor);

			// The exit used to be a side effect of evaluating the resulting check, which the degraded-egress
			// short-circuit skips. Left implicit, a monitor could sit in "maintenance" for a whole egress outage.
			expect(deps.monitorsRepository.updateById).toHaveBeenCalledWith("m1", "team", { status: "initializing" });
			expect(deps.providerRegistry.probe).toHaveBeenCalled();
		});

		it("does not touch the status of a monitor that was not in maintenance", async () => {
			const { pipeline, deps } = createPipeline();

			await pipeline.produce(makeMonitor({ status: "up" }));

			expect(deps.monitorsRepository.updateById).not.toHaveBeenCalled();
		});

		it("proceeds with the check when the maintenance window is inactive", async () => {
			const { pipeline, deps } = createPipeline({
				maintenanceWindowsRepository: { findByMonitorId: jest.fn<any>().mockResolvedValue([{ ...activeWindow(), active: false }]) },
			});

			await pipeline.produce(makeMonitor());

			expect(deps.providerRegistry.probe).toHaveBeenCalled();
		});

		// ── proxy resolution ──────────────────────────────────────────────────

		it("passes the resolved proxy url into probe", async () => {
			const { pipeline, deps } = createPipeline({
				proxyResolver: { resolve: jest.fn<any>().mockResolvedValue("http://proxy.example.com:8080") },
			});
			const monitor = makeMonitor();

			await pipeline.produce(monitor);

			expect(deps.providerRegistry.probe).toHaveBeenCalledWith(monitor, { proxyUrl: "http://proxy.example.com:8080" });
		});

		it("still produces a check when the resolver returns undefined", async () => {
			const { pipeline, deps } = createPipeline();
			const monitor = makeMonitor();

			const result = await pipeline.produce(monitor);

			expect(deps.providerRegistry.probe).toHaveBeenCalledWith(monitor, { proxyUrl: undefined });
			expect(result).toEqual({ status: expect.objectContaining({ monitorId: "m1" }), check: { id: "check-1" } });
		});

		// ── docker tls key resolution ─────────────────────────────────────────

		it("fetches the stored key for a docker monitor with a key set and passes it in the context", async () => {
			const { pipeline, deps } = createPipeline({
				monitorsRepository: { updateById: jest.fn<any>(), findDockerTlsKeyById: jest.fn<any>().mockResolvedValue("v1.abc123.iv.tag.data") },
			});
			const monitor = makeMonitor({ type: "docker", url: "tcp://host", dockerTlsKeySet: true });

			await pipeline.produce(monitor);

			expect(deps.monitorsRepository.findDockerTlsKeyById).toHaveBeenCalledWith("m1");
			expect(deps.providerRegistry.probe).toHaveBeenCalledWith(monitor, { proxyUrl: undefined, dockerTlsKey: "v1.abc123.iv.tag.data" });
		});

		it("does not fetch a key for a docker monitor without one set", async () => {
			const { pipeline, deps } = createPipeline();
			const monitor = makeMonitor({ type: "docker", url: "unix:///var/run/docker.sock", dockerTlsKeySet: false });

			await pipeline.produce(monitor);

			expect(deps.monitorsRepository.findDockerTlsKeyById).not.toHaveBeenCalled();
			expect(deps.providerRegistry.probe).toHaveBeenCalledWith(monitor, { proxyUrl: undefined, dockerTlsKey: undefined });
		});

		it("does not fetch a key for non-docker monitors even if the flag is set", async () => {
			const { pipeline, deps } = createPipeline();

			await pipeline.produce(makeMonitor({ type: "http", dockerTlsKeySet: true }));

			expect(deps.monitorsRepository.findDockerTlsKeyById).not.toHaveBeenCalled();
		});

		it("passes undefined when the flag is set but the repository has no key", async () => {
			const { pipeline, deps } = createPipeline();
			const monitor = makeMonitor({ type: "docker", url: "tcp://host", dockerTlsKeySet: true });

			await pipeline.produce(monitor);

			expect(deps.monitorsRepository.findDockerTlsKeyById).toHaveBeenCalledWith("m1");
			expect(deps.providerRegistry.probe).toHaveBeenCalledWith(monitor, { proxyUrl: undefined, dockerTlsKey: undefined });
		});

		// ── acquire / record ──────────────────────────────────────────────────

		it("throws when the network response is null", async () => {
			const { pipeline } = createPipeline({
				providerRegistry: { probe: jest.fn<any>().mockResolvedValue(null) },
			});

			await expect(pipeline.produce(makeMonitor())).rejects.toThrow("No network response");
		});

		it("returns null, warns, and does not buffer when toCheck yields nothing", async () => {
			const { pipeline, deps } = createPipeline({
				checkService: { toCheck: jest.fn<any>().mockReturnValue(null) },
			});

			const result = await pipeline.produce(makeMonitor());

			expect(result).toBeNull();
			expect(deps.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("No check could be built") }));
			expect(deps.bufferService.addToBuffer).not.toHaveBeenCalled();
		});

		it("buffers the built check and returns the status and check on success", async () => {
			const status = { monitorId: "m1", status: false, code: 500, message: "Error" };
			const check = { id: "check-1" };
			const { pipeline, deps } = createPipeline({
				providerRegistry: { probe: jest.fn<any>().mockResolvedValue(status) },
				checkService: { toCheck: jest.fn<any>().mockReturnValue(check) },
			});

			const result = await pipeline.produce(makeMonitor());

			expect(deps.bufferService.addToBuffer).toHaveBeenCalledWith(check);
			expect(result).toEqual({ status, check });
		});

		it("builds and buffers docker logs for a docker status", async () => {
			const status = { type: "docker", monitorId: "m1", teamId: "team", status: true, code: 200, message: "OK", payload: {} };
			const dockerLogs = [{ id: "docker-log-1" }, { id: "docker-log-2" }];
			const { pipeline, deps } = createPipeline({
				providerRegistry: { probe: jest.fn<any>().mockResolvedValue(status) },
				dockerLogsService: { buildDockerLogs: jest.fn<any>().mockResolvedValue(dockerLogs) },
			});

			await pipeline.produce(makeMonitor({ type: "docker" }));

			expect(deps.dockerLogsService.buildDockerLogs).toHaveBeenCalledWith(status);
			expect(deps.bufferService.addDockerLogToBuffer).toHaveBeenNthCalledWith(1, dockerLogs[0]);
			expect(deps.bufferService.addDockerLogToBuffer).toHaveBeenNthCalledWith(2, dockerLogs[1]);
		});

		it("does not build docker logs for a non-docker status", async () => {
			const { pipeline, deps } = createPipeline();

			await pipeline.produce(makeMonitor());

			expect(deps.dockerLogsService.buildDockerLogs).not.toHaveBeenCalled();
		});

		// ── egress self-check ─────────────────────────────────────────────────

		// A transport failure: nothing came back from the target, so the code is the network-error sentinel.
		const failingStatus = { monitorId: "m1", status: false, code: NETWORK_ERROR, message: "Timeout" };

		it("does not consult the egress service for a successful check and leaves the field unset", async () => {
			const check: Record<string, unknown> = { id: "check-1" };
			const { pipeline, deps } = createPipeline({
				checkService: { toCheck: jest.fn<any>().mockReturnValue(check) },
			});

			await pipeline.produce(makeMonitor());

			expect(deps.egressService.assessAfterFailure).not.toHaveBeenCalled();
			expect(check).not.toHaveProperty("egressStatus");
		});

		it("does not consult egress when an HTTP response was received", async () => {
			for (const status of [
				{ monitorId: "m1", status: false, code: 500, message: "Internal Server Error" },
				{ monitorId: "m1", status: false, code: 200, message: "Content mismatch" },
			]) {
				const check: Record<string, unknown> = { id: "check-1" };
				const { pipeline, deps } = createPipeline({
					providerRegistry: { probe: jest.fn<any>().mockResolvedValue(status) },
					checkService: { toCheck: jest.fn<any>().mockReturnValue(check) },
					egressService: { assessAfterFailure: jest.fn<any>().mockResolvedValue("degraded") },
				});

				await pipeline.produce(makeMonitor());

				expect(deps.egressService.assessAfterFailure).not.toHaveBeenCalled();
				expect(check).not.toHaveProperty("egressStatus");
				expect(deps.bufferService.addToBuffer).toHaveBeenCalledWith(check);
			}
		});

		it("consults egress for a timeout or network error", async () => {
			const check: Record<string, unknown> = { id: "check-1" };
			const { pipeline, deps } = createPipeline({
				providerRegistry: { probe: jest.fn<any>().mockResolvedValue(failingStatus) },
				checkService: { toCheck: jest.fn<any>().mockReturnValue(check) },
				egressService: { assessAfterFailure: jest.fn<any>().mockResolvedValue("degraded") },
			});

			await pipeline.produce(makeMonitor());

			expect(deps.egressService.assessAfterFailure).toHaveBeenCalledTimes(1);
			expect(check.egressStatus).toBe("degraded");
		});

		it("never consults egress for a docker daemon on a local socket", async () => {
			for (const url of ["unix:///var/run/docker.sock", "/var/run/docker.sock"]) {
				const check: Record<string, unknown> = { id: "check-1" };
				const { pipeline, deps } = createPipeline({
					providerRegistry: { probe: jest.fn<any>().mockResolvedValue({ ...failingStatus, type: "docker" }) },
					checkService: { toCheck: jest.fn<any>().mockReturnValue(check) },
					egressService: { assessAfterFailure: jest.fn<any>().mockResolvedValue("degraded") },
				});

				await pipeline.produce(makeMonitor({ type: "docker", url }));

				// Local IPC never leaves the instance, so it cannot fail through egress.
				expect(deps.egressService.assessAfterFailure).not.toHaveBeenCalled();
				expect(check).not.toHaveProperty("egressStatus");
				expect(deps.bufferService.addToBuffer).toHaveBeenCalledWith(check);
			}
		});

		it("consults egress for a hardware check and for a docker daemon reached over TCP", async () => {
			for (const monitor of [
				makeMonitor({ type: "hardware", url: "http://capture.example.com:59232/api/v1/metrics" }),
				makeMonitor({ type: "docker", url: "tcp://docker.example.com:2376" }),
			]) {
				const check: Record<string, unknown> = { id: "check-1" };
				const { pipeline, deps } = createPipeline({
					providerRegistry: { probe: jest.fn<any>().mockResolvedValue({ ...failingStatus, type: monitor.type }) },
					checkService: { toCheck: jest.fn<any>().mockReturnValue(check) },
					egressService: { assessAfterFailure: jest.fn<any>().mockResolvedValue("degraded") },
				});

				await pipeline.produce(monitor);

				// Both requests leave the instance, so our own loss of egress can be the cause.
				expect(deps.egressService.assessAfterFailure).toHaveBeenCalledTimes(1);
				expect(check.egressStatus).toBe("degraded");
			}
		});

		it("does not consult egress when the provider reports that the peer answered", async () => {
			// gRPC NOT_SERVING and an authoritative DNS NXDOMAIN both carry NETWORK_ERROR, so only this flag separates
			// them from a connection failure.
			const check: Record<string, unknown> = { id: "check-1" };
			const { pipeline, deps } = createPipeline({
				providerRegistry: { probe: jest.fn<any>().mockResolvedValue({ ...failingStatus, peerResponded: true }) },
				checkService: { toCheck: jest.fn<any>().mockReturnValue(check) },
				egressService: { assessAfterFailure: jest.fn<any>().mockResolvedValue("degraded") },
			});

			await pipeline.produce(makeMonitor({ type: "dns" }));

			expect(deps.egressService.assessAfterFailure).not.toHaveBeenCalled();
			expect(check).not.toHaveProperty("egressStatus");
		});

		it("consults egress when the provider reports that nothing answered, despite an HTTP-shaped code", async () => {
			const check: Record<string, unknown> = { id: "check-1" };
			const { pipeline, deps } = createPipeline({
				providerRegistry: { probe: jest.fn<any>().mockResolvedValue({ ...failingStatus, code: 503, peerResponded: false }) },
				checkService: { toCheck: jest.fn<any>().mockReturnValue(check) },
				egressService: { assessAfterFailure: jest.fn<any>().mockResolvedValue("degraded") },
			});

			await pipeline.produce(makeMonitor({ type: "grpc" }));

			// An explicit report from the provider outranks the HTTP-status heuristic.
			expect(deps.egressService.assessAfterFailure).toHaveBeenCalledTimes(1);
			expect(check.egressStatus).toBe("degraded");
		});

		it("flags a failing check as degraded when the egress service reports degraded egress", async () => {
			const check: Record<string, unknown> = { id: "check-1" };
			const { pipeline, deps } = createPipeline({
				providerRegistry: { probe: jest.fn<any>().mockResolvedValue(failingStatus) },
				checkService: { toCheck: jest.fn<any>().mockReturnValue(check) },
				egressService: { assessAfterFailure: jest.fn<any>().mockResolvedValue("degraded") },
			});

			const result = await pipeline.produce(makeMonitor());

			expect(deps.egressService.assessAfterFailure).toHaveBeenCalledTimes(1);
			expect(check.egressStatus).toBe("degraded");
			expect(deps.bufferService.addToBuffer).toHaveBeenCalledWith(expect.objectContaining({ egressStatus: "degraded" }));
			expect(result?.check.egressStatus).toBe("degraded");
		});

		// ── status window across a spell ─────────────────────────────────────
		// Matching the maintenance gate: cleared once as the monitor stops being evaluated, so the results either
		// side of a spell are never adjacent.

		it("clears the status window on the first degraded check of a spell", async () => {
			const { pipeline, deps } = createPipeline({
				providerRegistry: { probe: jest.fn<any>().mockResolvedValue(failingStatus) },
				egressService: { assessAfterFailure: jest.fn<any>().mockResolvedValue("degraded") },
			});

			await pipeline.produce(makeMonitor({ statusWindow: [true, true, true, false, false] }));

			expect(deps.monitorsRepository.updateById).toHaveBeenCalledWith("m1", "team", { statusWindow: [] });
		});

		it("does not re-write the status window for later degraded checks in the same spell", async () => {
			const { pipeline, deps } = createPipeline({
				providerRegistry: { probe: jest.fn<any>().mockResolvedValue(failingStatus) },
				egressService: { assessAfterFailure: jest.fn<any>().mockResolvedValue("degraded") },
			});

			await pipeline.produce(makeMonitor({ statusWindow: [] }));

			expect(deps.monitorsRepository.updateById).not.toHaveBeenCalled();
		});

		it("leaves the status window alone when the probe found egress fine", async () => {
			const { pipeline, deps } = createPipeline({
				providerRegistry: { probe: jest.fn<any>().mockResolvedValue(failingStatus) },
				egressService: { assessAfterFailure: jest.fn<any>().mockResolvedValue("ok") },
			});

			await pipeline.produce(makeMonitor({ statusWindow: [true, true, true, false, false] }));

			expect(deps.monitorsRepository.updateById).not.toHaveBeenCalled();
		});

		it("flags a failing check as ok when the probe found egress fine", async () => {
			const check: Record<string, unknown> = { id: "check-1" };
			const { pipeline } = createPipeline({
				providerRegistry: { probe: jest.fn<any>().mockResolvedValue(failingStatus) },
				checkService: { toCheck: jest.fn<any>().mockReturnValue(check) },
				egressService: { assessAfterFailure: jest.fn<any>().mockResolvedValue("ok") },
			});

			await pipeline.produce(makeMonitor());

			expect(check.egressStatus).toBe("ok");
		});

		it("leaves the field unset on a failing check when the egress check is disabled", async () => {
			const check: Record<string, unknown> = { id: "check-1" };
			const { pipeline, deps } = createPipeline({
				providerRegistry: { probe: jest.fn<any>().mockResolvedValue(failingStatus) },
				checkService: { toCheck: jest.fn<any>().mockReturnValue(check) },
			});

			await pipeline.produce(makeMonitor());

			expect(deps.egressService.assessAfterFailure).toHaveBeenCalledTimes(1);
			expect(check).not.toHaveProperty("egressStatus");
		});
	});

	// ── Stage 2: ingestion ───────────────────────────────────────────────────

	describe("ingestChecks", () => {
		it("stores the batch, then arms the evaluate stage once per distinct monitor", async () => {
			const { pipeline, deps } = createPipeline();
			const batch = [
				makeCheck({ id: "c1", metadata: { monitorId: "mon-1", teamId: "team-1", type: "http" } }),
				makeCheck({ id: "c2", metadata: { monitorId: "mon-1", teamId: "team-1", type: "http" } }),
				makeCheck({ id: "c3", metadata: { monitorId: "mon-2", teamId: "team-1", type: "http" } }),
			];

			await pipeline.ingestChecks(batch);

			const createdAt = Date.parse("2026-01-01T00:00:00.000Z");
			expect(deps.checkService.createChecks).toHaveBeenCalledWith(batch);
			expect(deps.jobsRepository.upsertEvaluate).toHaveBeenCalledTimes(2);
			expect(deps.jobsRepository.upsertEvaluate).toHaveBeenCalledWith(
				"mon-1",
				[
					{ checkId: "c1", createdAt },
					{ checkId: "c2", createdAt },
				],
				expect.any(Number)
			);
			expect(deps.jobsRepository.upsertEvaluate).toHaveBeenCalledWith("mon-2", [{ checkId: "c3", createdAt }], expect.any(Number));
		});

		it("retries a failed arm on the next ingest without re-storing the checks", async () => {
			const { pipeline, deps } = createPipeline();
			deps.jobsRepository.upsertEvaluate.mockRejectedValueOnce(new Error("arm failed"));

			await pipeline.ingestChecks([makeCheck({ id: "c1" })]);
			expect(deps.checkService.createChecks).toHaveBeenCalledTimes(1);
			expect(deps.logger.error).toHaveBeenCalledWith(expect.objectContaining({ method: "ingestChecks" }));

			await pipeline.ingestChecks([]); // nothing new, only the retry runs
			expect(deps.checkService.createChecks).toHaveBeenCalledTimes(1);
			expect(deps.jobsRepository.upsertEvaluate).toHaveBeenCalledTimes(2);
			expect(deps.jobsRepository.upsertEvaluate).toHaveBeenLastCalledWith(
				"mon-1",
				[{ checkId: "c1", createdAt: expect.any(Number) }],
				expect.any(Number)
			);

			await pipeline.ingestChecks([]); // nothing left to retry
			expect(deps.jobsRepository.upsertEvaluate).toHaveBeenCalledTimes(2);
		});

		it("neither writes nor arms for an empty batch", async () => {
			const { pipeline, deps } = createPipeline();

			await pipeline.ingestChecks([]);

			expect(deps.checkService.createChecks).not.toHaveBeenCalled();
			expect(deps.jobsRepository.upsertEvaluate).not.toHaveBeenCalled();
		});

		it("does not arm the evaluate stage when the check write fails", async () => {
			const { pipeline, deps } = createPipeline();
			deps.checkService.createChecks.mockRejectedValue(new Error("DB write failed"));

			await expect(pipeline.ingestChecks([makeCheck()])).rejects.toThrow("DB write failed");

			expect(deps.jobsRepository.upsertEvaluate).not.toHaveBeenCalled();
		});
	});

	// ── Stage 3: evaluation ──────────────────────────────────────────────────

	describe("handleEvaluate", () => {
		it("loads the row's pending checks, dispatches each, and pulls its id after dispatch", async () => {
			const job = makeEvaluateJob("c1");
			const { pipeline, deps } = createPipeline({
				checksRepository: { findUnevaluatedByMonitorId: jest.fn<any>().mockResolvedValue([makeCheck({ id: "c1" })]) },
			});

			await pipeline.handleEvaluate(job);

			expect(deps.checksRepository.findUnevaluatedByMonitorId).toHaveBeenCalledWith("m1", job.pendingChecks);
			expect(deps.statusService.updateMonitorStatus).toHaveBeenCalledTimes(1);
			expect(deps.dispatcher.dispatch).toHaveBeenCalledTimes(1);
			expect(deps.jobsRepository.pullEvaluated).toHaveBeenCalledWith(job.id, ["c1"]);
			// The monitor cursor is gone: nothing writes lastEvaluatedAt
			expect(deps.monitorsRepository.updateById).not.toHaveBeenCalled();
		});

		it("does nothing when the row has no pending checks", async () => {
			const { pipeline, deps } = createPipeline();

			await pipeline.handleEvaluate(makeEvaluateJob());

			expect(deps.monitorsRepository.findByIdLean).not.toHaveBeenCalled();
			expect(deps.checksRepository.findUnevaluatedByMonitorId).not.toHaveBeenCalled();
		});

		it("drops every pending id for a deleted monitor", async () => {
			const job = makeEvaluateJob("c1", "c2");
			const { pipeline, deps } = createPipeline({
				monitorsRepository: { findByIdLean: jest.fn<any>().mockResolvedValue(null) },
			});

			await pipeline.handleEvaluate(job);

			expect(deps.statusService.updateMonitorStatus).not.toHaveBeenCalled();
			expect(deps.jobsRepository.pullEvaluated).toHaveBeenCalledWith(job.id, ["c1", "c2"]);
		});

		it("stops applying checks once the lease is lost so a second claimer does not double-apply", async () => {
			const job = makeEvaluateJob("c1", "c2", "c3");
			const { pipeline, deps } = createPipeline({
				checksRepository: {
					findUnevaluatedByMonitorId: jest.fn<any>().mockResolvedValue([makeCheck({ id: "c1" }), makeCheck({ id: "c2" }), makeCheck({ id: "c3" })]),
				},
				jobsRepository: {
					upsertEvaluate: jest.fn<any>(),
					pullEvaluated: jest.fn<any>().mockResolvedValueOnce(true).mockResolvedValueOnce(false), // lease lost after c2
				},
			});

			await pipeline.handleEvaluate(job);

			expect(deps.statusService.updateMonitorStatus).toHaveBeenCalledTimes(2);
			expect(deps.dispatcher.dispatch).toHaveBeenCalledTimes(2);
			expect(deps.jobsRepository.pullEvaluated).toHaveBeenCalledTimes(2); // no trailing sweep either
			expect(deps.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("Lost lease") }));
		});

		it("pulls ids whose check no longer exists so they do not sit on the row forever", async () => {
			const job = makeEvaluateJob("c1", "c-gone");
			const { pipeline, deps } = createPipeline({
				checksRepository: { findUnevaluatedByMonitorId: jest.fn<any>().mockResolvedValue([makeCheck({ id: "c1" })]) }, // c-gone was cleaned up
			});

			await pipeline.handleEvaluate(job);

			expect(deps.statusService.updateMonitorStatus).toHaveBeenCalledTimes(1);
			expect(deps.jobsRepository.pullEvaluated).toHaveBeenCalledWith(job.id, ["c1"]);
			expect(deps.jobsRepository.pullEvaluated).toHaveBeenCalledWith(job.id, ["c-gone"]);
		});

		it("reads the monitor once and threads the post-write monitor across the backlog", async () => {
			const job = makeEvaluateJob("c1", "c2");
			const initialMonitor = makeMonitor();
			const postWriteMonitor = makeMonitor({ status: "down" }); // what updateStatusWindowAndChecks would return after check c1
			const { pipeline, deps } = createPipeline({
				monitorsRepository: { findByIdLean: jest.fn<any>().mockResolvedValue(initialMonitor), updateById: jest.fn<any>() },
				checksRepository: { findUnevaluatedByMonitorId: jest.fn<any>().mockResolvedValue([makeCheck({ id: "c1" }), makeCheck({ id: "c2" })]) },
				statusService: { updateMonitorStatus: jest.fn<any>().mockResolvedValue(makeStatusChange({ monitor: postWriteMonitor })) },
			});

			await pipeline.handleEvaluate(job);

			// The monitor is read once for the whole backlog, not once per check
			expect(deps.monitorsRepository.findByIdLean).toHaveBeenCalledTimes(1);
			expect(deps.statusService.updateMonitorStatus).toHaveBeenCalledTimes(2);
			// First check evaluates against the freshly-read monitor; the second against the post-write monitor from the first
			expect(deps.statusService.updateMonitorStatus.mock.calls[0][1]).toBe(initialMonitor);
			expect(deps.statusService.updateMonitorStatus.mock.calls[1][1]).toBe(postWriteMonitor);
		});
	});

	describe("evaluateCheck", () => {
		it("threads check and monitor through to the status service", async () => {
			const { pipeline, deps } = createPipeline();
			const check = makeCheck({ status: true, statusCode: 200, message: "OK" });
			const monitor = makeMonitor({ status: "up" });

			await pipeline.evaluateCheck(check, monitor);

			expect(deps.statusService.updateMonitorStatus).toHaveBeenCalledWith(check, monitor);
		});

		it("returns an evaluation that opens an incident on a down transition", async () => {
			const check = makeCheck({ status: false, statusCode: 500, message: "Error" });
			const statusChange = makeStatusChange({ monitor: makeMonitor({ status: "down" }), statusChanged: true, prevStatus: "up" });
			const { pipeline } = createPipeline({
				statusService: { updateMonitorStatus: jest.fn<any>().mockResolvedValue(statusChange) },
			});

			const result = await pipeline.evaluateCheck(check, makeMonitor());

			expect(result).toMatchObject({
				monitor: statusChange.monitor,
				check,
				decision: { transition: "status_down" },
			});
		});

		it("returns an evaluation that resolves an incident on recovery", async () => {
			const statusChange = makeStatusChange({ monitor: makeMonitor({ status: "up" }), statusChanged: true, prevStatus: "down" });
			const { pipeline } = createPipeline({
				statusService: { updateMonitorStatus: jest.fn<any>().mockResolvedValue(statusChange) },
			});

			const result = await pipeline.evaluateCheck(makeCheck({ status: true, statusCode: 200, message: "OK" }), makeMonitor({ status: "down" }));

			expect(result.decision).toEqual({ transition: "status_up" });
		});

		it("produces a no-op decision when status did not change", async () => {
			const { pipeline } = createPipeline();

			const result = await pipeline.evaluateCheck(makeCheck({ status: true, statusCode: 200, message: "OK" }), makeMonitor());

			expect(result.decision).toEqual({ transition: null });
		});

		// ── degraded egress ───────────────────────────────────────────────────

		it("short-circuits a check flagged with degraded egress without touching monitor status", async () => {
			const check = makeCheck({ status: false, statusCode: 500, message: "Error", egressStatus: "degraded" });
			const monitor = makeMonitor({ status: "up" });
			const { pipeline, deps } = createPipeline();

			const result = await pipeline.evaluateCheck(check, monitor);

			expect(deps.statusService.updateMonitorStatus).not.toHaveBeenCalled();
			expect(result.monitor).toBe(monitor);
			expect(result).not.toHaveProperty("statusChange");
			expect(result.decision).toEqual({ transition: null });
			expect(deps.logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("egress was degraded") }));
		});

		it("evaluates a check flagged with egress ok as normal", async () => {
			const check = makeCheck({ status: false, statusCode: 500, message: "Error", egressStatus: "ok" });
			const { pipeline, deps } = createPipeline();

			await pipeline.evaluateCheck(check, makeMonitor());

			expect(deps.statusService.updateMonitorStatus).toHaveBeenCalledWith(check, expect.anything());
		});
	});

	// The decision table, driven through evaluateCheck with the status service stubbed to return each transition.

	describe("decision", () => {
		const decide = async (statusChange: StatusChangeResult) => {
			const { pipeline } = createPipeline({
				statusService: { updateMonitorStatus: jest.fn<any>().mockResolvedValue(statusChange) },
			});
			const result = await pipeline.evaluateCheck(makeCheck({ status: true, statusCode: 200, message: "OK" }), makeMonitor());
			return result.decision;
		};

		it("does nothing when statusChanged is false", async () => {
			const decision = await decide(makeStatusChange({ status: "down", statusChanged: false, prevStatus: "up" }));
			expect(decision).toEqual({ transition: null });
		});

		it("reports status_down when the monitor goes down", async () => {
			const decision = await decide(makeStatusChange({ status: "down", statusChanged: true, prevStatus: "up" }));
			expect(decision).toEqual({ transition: "status_down" });
		});

		it("reports threshold_breach when the monitor is breached", async () => {
			const decision = await decide(makeStatusChange({ status: "breached", statusChanged: true, prevStatus: "up" }));
			expect(decision).toEqual({ transition: "threshold_breach" });
		});

		it("reports status_up when the monitor recovers from down", async () => {
			const decision = await decide(makeStatusChange({ status: "up", statusChanged: true, prevStatus: "down" }));
			expect(decision).toEqual({ transition: "status_up" });
		});

		it("reports threshold_resolved when the monitor recovers from breached", async () => {
			const decision = await decide(makeStatusChange({ status: "up", statusChanged: true, prevStatus: "breached" }));
			expect(decision).toEqual({ transition: "threshold_resolved" });
		});

		it("reports no transition for unhandled status changes", async () => {
			const decision = await decide(makeStatusChange({ status: "paused", statusChanged: true, prevStatus: "up" }));
			expect(decision).toEqual({ transition: null });
		});

		it("carries the threshold breaches onto a breached decision", async () => {
			const thresholdBreaches = { cpu: true, memory: false, disk: true, temp: false };
			const decision = await decide(makeStatusChange({ status: "breached", statusChanged: true, prevStatus: "up", thresholdBreaches }));
			expect(decision).toEqual({ transition: "threshold_breach", thresholdBreaches });
		});

		it("leaves threshold breaches off a down decision even when the status change carries them", async () => {
			const thresholdBreaches = { cpu: true, memory: false, disk: false, temp: false };
			const decision = await decide(makeStatusChange({ status: "down", statusChanged: true, prevStatus: "up", thresholdBreaches }));
			expect(decision).not.toHaveProperty("thresholdBreaches");
		});

		it("leaves threshold breaches off a no-op decision", async () => {
			const thresholdBreaches = { cpu: true, memory: false, disk: false, temp: false };
			const decision = await decide(makeStatusChange({ status: "breached", statusChanged: false, prevStatus: "breached", thresholdBreaches }));
			expect(decision).not.toHaveProperty("thresholdBreaches");
		});

		// StatusService only sets thresholdBreaches for hardware monitors; a docker breach must not leave the key behind holding undefined.
		it("leaves threshold breaches off a breached decision when the status change carries none", async () => {
			const decision = await decide(makeStatusChange({ status: "breached", statusChanged: true, prevStatus: "up" }));
			expect(decision).toEqual({ transition: "threshold_breach" });
		});

		// A transition must not leak into the next decision the same pipeline makes, whether that is a no-op
		// evaluation or a degraded-egress short-circuit.
		it("starts every decision clean after a transition on the same pipeline", async () => {
			const updateMonitorStatus = jest
				.fn<any>()
				.mockResolvedValueOnce(makeStatusChange({ status: "down", statusChanged: true, prevStatus: "up" }))
				.mockResolvedValueOnce(makeStatusChange({ status: "down", statusChanged: false, prevStatus: "down" }));
			const { pipeline } = createPipeline({ statusService: { updateMonitorStatus } });
			const failing = makeCheck({ status: false, statusCode: 500, message: "Error" });

			const first = await pipeline.evaluateCheck(failing, makeMonitor());
			const second = await pipeline.evaluateCheck(failing, makeMonitor({ status: "down" }));
			const degraded = await pipeline.evaluateCheck(makeCheck({ ...failing, egressStatus: "degraded" }), makeMonitor({ status: "down" }));

			expect(first.decision).toEqual({ transition: "status_down" });
			const clean = { transition: null };
			expect(second.decision).toEqual(clean);
			expect(degraded.decision).toEqual(clean);
		});
	});
});