import { execFile } from "node:child_process";
import { randomUUID } from "node:crypto";
import { copyFile, mkdir, mkdtemp, rm } from "node:fs/promises";
import { join, resolve } from "node:path";
import { pathToFileURL } from "node:url";
import { promisify } from "node:util";
import type { RowDataPacket } from "mysql2/promise";
import mysql from "mysql2/promise";
import {
	GenericContainer,
	type StartedTestContainer,
	Wait,
} from "testcontainers";
import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";

const exec = promisify(execFile);
const migrations = [
	"0027_catalog_packages.sql",
	"0028_history_snapshots.sql",
	"0029_admin_table_views.sql",
];
let maria: StartedTestContainer | undefined;
let redisContainer: StartedTestContainer | undefined;
let connection: mysql.Connection | undefined;
let appDb: typeof import("@/lib/db").db | undefined;
let appRedis: typeof import("@/lib/redis").redis;
let commands: typeof import("@/features/catalog/server/bulk-offers");
let cache: typeof import("@/lib/cache");
let migrationRoot: string | undefined;
let databaseUrl: string;

async function rows(query: string) {
	if (!connection) throw Error("Integration database is not connected");
	const [result] = await connection.query<RowDataPacket[]>(query);
	return result;
}

async function migrate(...args: string[]) {
	if (!migrationRoot) throw Error("Migration fixture is not ready");
	// Only explicit test variables: the copied loader cannot see checkout .env files.
	return exec(
		process.execPath,
		[
			"--import",
			pathToFileURL(resolve("node_modules/tsx/dist/loader.mjs")).href,
			join(migrationRoot, "scripts/apply-migrations.ts"),
			...args,
		],
		{
			cwd: migrationRoot,
			env: {
				PATH: process.env.PATH,
				SystemRoot: process.env.SystemRoot,
				DATABASE_URL: databaseUrl,
				NODE_ENV: "test",
			},
			timeout: 30_000,
		},
	);
}

beforeAll(async () => {
	// No fixed names, ports, external URLs, shared volumes or container reuse.
	const databasePassword = randomUUID();
	const redisPassword = randomUUID();
	maria = await new GenericContainer("mariadb:11.4.5")
		.withEnvironment({
			MARIADB_ROOT_PASSWORD: randomUUID(),
			MARIADB_DATABASE: "integration",
			MARIADB_USER: "integration",
			MARIADB_PASSWORD: databasePassword,
		})
		.withExposedPorts(3306)
		.withHealthCheck({
			test: ["CMD", "healthcheck.sh", "--connect", "--innodb_initialized"],
			interval: 1000,
			timeout: 5000,
			retries: 60,
			startPeriod: 1000,
		})
		.withWaitStrategy(Wait.forHealthCheck())
		.withStartupTimeout(120_000)
		.start();
	databaseUrl = `mysql://integration:${databasePassword}@${maria.getHost()}:${maria.getMappedPort(3306)}/integration`;
	connection = await mysql.createConnection(databaseUrl);
	redisContainer = await new GenericContainer("redis:7.4.2-alpine")
		.withCommand(["redis-server", "--requirepass", redisPassword])
		.withExposedPorts(6379)
		.withWaitStrategy(Wait.forLogMessage("Ready to accept connections"))
		.withStartupTimeout(60_000)
		.start();
	process.env.DATABASE_URL = databaseUrl;
	process.env.REDIS_URL = `redis://:${redisPassword}@${redisContainer.getHost()}:${redisContainer.getMappedPort(6379)}/0`;
	delete process.env.SKIP_ENV_VALIDATION;
	Object.assign(process.env, { NODE_ENV: "test" });
	process.env.HOTEL_NAME = "Integration";

	await connection.query(
		"CREATE TABLE catalog_pages (id INT PRIMARY KEY, caption VARCHAR(255) NOT NULL) ENGINE=InnoDB",
	);
	await connection.query(
		"CREATE TABLE catalog_items (id INT PRIMARY KEY, catalog_name VARCHAR(255) NOT NULL, page_id VARCHAR(25) NOT NULL, cost_credits INT NOT NULL, cost_points INT NOT NULL, points_type INT NOT NULL) ENGINE=InnoDB",
	);
	await connection.query(
		"CREATE TABLE admin_audit_log (id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, action VARCHAR(191) NOT NULL DEFAULT '', target VARCHAR(191) NOT NULL DEFAULT '', target_id INT NULL, details TEXT NULL, `before` TEXT NULL, `after` TEXT NULL, diff TEXT NULL, ip_address VARCHAR(45) NULL, created_at VARCHAR(64) NOT NULL DEFAULT '', updated_at VARCHAR(64) NULL) ENGINE=InnoDB",
	);
	// The emulator owns core tables. Exercise the real CMS migration CLI over this baseline.
	migrationRoot = await mkdtemp(join(resolve("integration"), ".migration-"));
	await mkdir(join(migrationRoot, "scripts"));
	await mkdir(join(migrationRoot, "drizzle/migrations"), { recursive: true });
	for (const file of [
		"apply-migrations.ts",
		"load-env.ts",
		"db-url.ts",
		"sql-statements.ts",
	]) {
		await copyFile(
			resolve("scripts", file),
			join(migrationRoot, "scripts", file),
		);
	}
	for (const file of migrations)
		await copyFile(
			resolve("drizzle/migrations", file),
			join(migrationRoot, "drizzle/migrations", file),
		);
	await migrate();
	appDb = (await import("@/lib/db")).db;
	appRedis = (await import("@/lib/redis")).redis;
	if (!appRedis) throw Error("Redis must be enabled in integration tests");
	await appRedis.ping();
	commands = await import("@/features/catalog/server/bulk-offers");
	cache = await import("@/lib/cache");
});

afterAll(async () => {
	// Settle every cleanup so a failed setup or close cannot leak the other container.
	const cleanup = await Promise.allSettled([
		appDb?.$client.end(),
		appRedis?.quit(),
		connection?.end(),
	]);
	const stops = await Promise.allSettled([
		maria?.stop(),
		redisContainer?.stop(),
	]);
	if (migrationRoot) {
		const target = resolve(migrationRoot);
		if (
			!target.startsWith(`${resolve("integration")}\\.migration-`) &&
			!target.startsWith(`${resolve("integration")}/.migration-`)
		)
			throw Error("Unsafe migration fixture path");
		await rm(target, { recursive: true, force: true });
	}
	for (const result of [...cleanup, ...stops])
		if (result.status === "rejected") throw result.reason;
});

beforeEach(async () => {
	if (!connection) throw Error("Integration database is not connected");
	await connection.query("DROP TRIGGER IF EXISTS reject_second_history");
	await connection.query("DELETE FROM admin_audit_log");
	await connection.query("DELETE FROM catalog_items");
	await connection.query("DELETE FROM catalog_pages");
	await connection.query(
		"INSERT INTO catalog_pages VALUES (1, 'Original'), (2, 'Destination')",
	);
	await connection.query(
		"INSERT INTO catalog_items VALUES (1, 'Chair', '1', 10, 0, 0), (2, 'Table', '1', 20, 0, 0)",
	);
});

const input = {
	ids: [1, 2],
	changes: { costCredits: { mode: "add" as const, value: 5 } },
};

describe("MariaDB migrations and catalog transactions", () => {
	it("runs the actual migration CLI twice without duplicate tracking or lost data", async () => {
		const before = await rows("SELECT * FROM cms_migrations ORDER BY id");
		expect(before.map((row) => row.migration)).toEqual(
			migrations.map((file) => file.replace(/\.sql$/, "")),
		);
		await connection?.query(
			"INSERT INTO website_admin_table_views VALUES (1, '/admin/catalog', 'Saved', '{}')",
		);
		expect((await migrate()).stdout).toContain(
			"All migrations already applied",
		);
		expect(await rows("SELECT * FROM cms_migrations ORDER BY id")).toEqual(
			before,
		);
		expect(await rows("SELECT name FROM website_admin_table_views")).toEqual([
			{ name: "Saved" },
		]);
		expect((await migrate("--status")).stdout).toContain(
			"3/3 applied, 0 pending",
		);
		const columns = await rows(
			"SELECT COLUMN_NAME, DATA_TYPE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='admin_audit_log' AND COLUMN_NAME IN ('before','after') ORDER BY COLUMN_NAME",
		);
		expect(columns.map((row) => row.DATA_TYPE)).toEqual([
			"mediumtext",
			"mediumtext",
		]);
	});

	it("commits both offers and history, then restores them through the real undo command", async () => {
		const preview = await commands.previewBulkOffersCommand(input);
		const result = await commands.applyBulkOffersCommand(
			input,
			preview.fingerprint,
			7,
		);
		expect(result.changedCount).toBe(2);
		expect(result.historyIds).toHaveLength(2);
		expect(
			(await rows("SELECT cost_credits FROM catalog_items ORDER BY id")).map(
				(row) => row.cost_credits,
			),
		).toEqual([15, 25]);
		expect(
			await rows("SELECT target, action FROM admin_audit_log ORDER BY id"),
		).toEqual([
			{ target: "catalog_offer", action: "history_update" },
			{ target: "catalog_offer", action: "history_update" },
		]);
		await commands.undoBulkOffersCommand(result.historyIds, 7);
		expect(
			(await rows("SELECT cost_credits FROM catalog_items ORDER BY id")).map(
				(row) => row.cost_credits,
			),
		).toEqual([10, 20]);
		expect(await rows("SELECT id FROM admin_audit_log")).toHaveLength(4);
	});

	it("rolls back earlier offer updates and history when the second history insert fails", async () => {
		await connection?.query(
			"CREATE TRIGGER reject_second_history BEFORE INSERT ON admin_audit_log FOR EACH ROW BEGIN IF NEW.target_id=2 THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='forced history failure'; END IF; END",
		);
		const preview = await commands.previewBulkOffersCommand(input);
		await expect(
			commands.applyBulkOffersCommand(input, preview.fingerprint, 7),
		).rejects.toThrow();
		expect(
			(await rows("SELECT cost_credits FROM catalog_items ORDER BY id")).map(
				(row) => row.cost_credits,
			),
		).toEqual([10, 20]);
		expect(await rows("SELECT id FROM admin_audit_log")).toHaveLength(0);
	});

	it("serializes competing applications of one preview and rejects the stale contender", async () => {
		const preview = await commands.previewBulkOffersCommand(input);
		const results = await Promise.allSettled([
			commands.applyBulkOffersCommand(input, preview.fingerprint, 7),
			commands.applyBulkOffersCommand(input, preview.fingerprint, 8),
		]);
		expect(
			results.filter((result) => result.status === "fulfilled"),
		).toHaveLength(1);
		const rejected = results.find((result) => result.status === "rejected");
		expect(rejected?.status === "rejected" && rejected.reason.message).toMatch(
			/selection.*changed|preview/i,
		);
		expect(
			(await rows("SELECT cost_credits FROM catalog_items ORDER BY id")).map(
				(row) => row.cost_credits,
			),
		).toEqual([15, 25]);
		expect(await rows("SELECT id FROM admin_audit_log")).toHaveLength(2);
	});
});

describe("Redis application cache", () => {
	it("writes to real Redis with expiry and reads it after memory invalidation", async () => {
		const key = `integration:${randomUUID()}:catalog`;
		let fetches = 0;
		const fetch = async () => ({ revision: ++fetches });
		expect(await cache.cached(key, 60_000, fetch)).toEqual({ revision: 1 });
		expect(await appRedis?.get(key)).toBe('{"revision":1}');
		expect(await appRedis?.ttl(key)).toBeGreaterThan(0);
		cache.invalidateMemory(key);
		expect(await cache.cached(key, 60_000, fetch)).toEqual({ revision: 1 });
		expect(fetches).toBe(1);
		await appRedis?.del(key);
		cache.invalidateMemory(key);
		expect(await cache.cached(key, 60_000, fetch)).toEqual({ revision: 2 });
	});

	it("keeps two independently named cache entries isolated during invalidation", async () => {
		const first = `integration:${randomUUID()}:first`;
		const second = `integration:${randomUUID()}:second`;
		await cache.cached(first, 60_000, async () => "first");
		await cache.cached(second, 60_000, async () => "second");
		await appRedis?.del(first);
		cache.invalidateMemory(first);
		cache.invalidateMemory(second);
		expect(await cache.cached(first, 60_000, async () => "updated")).toBe(
			"updated",
		);
		expect(await cache.cached(second, 60_000, async () => "wrong")).toBe(
			"second",
		);
		expect(await appRedis?.get(second)).toBe('"second"');
	});
});