import { readdir, unlink } from "node:fs/promises";
import { join, relative } from "node:path";
import { pathToFileURL } from "node:url";
import type { BucketSource } from "../server/config.ts";

/**
 * A Koh 3 remote as Kepr uses it (§9.1): read-only on Koh's data, which needs two of its five
 * operations, plus a short-lived download URL so Kepr never streams release files itself (§22.2).
 * Kepr's own files (uploads) live under `KEPR_PREFIX` (June, 2026-09-29); no Koh repo can have
 * that name, since Kepr and Koh both need a `head` for one. Beyond it Kepr touches only builds,
 * which Koh has no way to delete (June, 2026-09-30): it deletes a version's files, rewrites its
 * record without a deleted one, and moves `releases/latest`.
 */
export type Bucket = {
	get(key: string): Promise<Uint8Array | null>;
	list(prefix: string): Promise<string[]>;
	/** The top-level folders' names: one listing, however many keys they hold. */
	folders(): Promise<string[]>;
	/** A key's size in bytes, asked without reading it. */
	size(key: string): Promise<number>;
	/** Every key under `prefix` with its size in bytes, for the console's storage view. */
	sizes(prefix: string): Promise<Map<string, number>>;
	/** A URL that serves `key` for five minutes, as an attachment with this filename and content type. */
	downloadUrl(key: string, filename: string, type: string): string;
	/** A five-minute URL that shows `key` in the page, or null when Kepr must serve the bytes itself. */
	viewUrl(key: string, type: string): string | null;
	/**
	 * Writes one of Kepr's own files, a repo's `releases/latest`, or a version's `release.json` when
	 * one of its files is deleted; any other key is refused.
	 */
	put(key: string, bytes: Uint8Array, type: string): Promise<void>;
	/** Deletes one of Kepr's own files, a release's file or `releases/latest`; nothing else. */
	remove(key: string): Promise<void>;
	/** Copies a key of one repo to another repo: a rename (June, 2026-10-01), and nothing else. */
	copy(from: string, to: string): Promise<void>;
	/** Deletes a key of a repo that a rename has copied to its new name. */
	removeRenamed(key: string): Promise<void>;
	/**
	 * Every key that belongs to one repo (§24): its own, and Kepr's files for it under `.kepr/`.
	 * What deleting a repo has to remove, and nothing of any other repo's.
	 */
	repoKeys(repo: string): Promise<string[]>;
	/** Deletes one of those keys; anything not of that repo is refused. */
	removeRepoKey(repo: string, key: string): Promise<void>;
};

export const KEPR_PREFIX = ".kepr/";

/** `each` over `items`, `atOnce` of them in flight at a time; the results in order. */
export async function inTurn<T, R>(
	items: T[],
	atOnce: number,
	each: (item: T) => Promise<R>,
): Promise<R[]> {
	const results: R[] = [];
	for (let i = 0; i < items.length; i += atOnce) {
		results.push(...(await Promise.all(items.slice(i, i + atOnce).map(each))));
	}
	return results;
}

/** A write Kepr can't make: outside its folder, or refused by the bucket. */
export class BucketWriteError extends Error {}

const DOWNLOAD_SECONDS = 5 * 60;

const RELEASE_WRITE = /^[^/]+\/releases\/(latest|[^/]+\/release\.json)$/;
const RELEASE_FILE = /^[^/]+\/releases\/(latest|[^/]+\/.+)$/;

// A key inside a repo, as a rename moves them: never Kepr's own folder, never `..` or empty parts.
function requireRepoKey(key: string): void {
	if (
		key.startsWith(KEPR_PREFIX) ||
		!key.includes("/") ||
		key.split("/").some((p) => p === ".." || p === "")
	) {
		throw new BucketWriteError(`a rename moves only a repo's keys, not "${key}"`);
	}
}

function requireWritable(key: string, releaseKey: RegExp): void {
	const allowed = key.startsWith(KEPR_PREFIX) || releaseKey.test(key);
	if (!allowed || key.split("/").some((part) => part === ".." || part === "")) {
		throw new BucketWriteError(`Kepr writes only under ${KEPR_PREFIX} and builds, not "${key}"`);
	}
}

// Kepr's own files for one repo: uploaded media, kept tarballs, the hidden list, the rename record.
const keprRepoFolders = (repo: string) => [
	`${KEPR_PREFIX}media/${repo}/`,
	`${KEPR_PREFIX}tarballs/${repo}/`,
];
const keprRepoFiles = (repo: string) => [
	`${KEPR_PREFIX}hidden/${repo}`,
	`${KEPR_PREFIX}renamed/${repo}`,
];

/**
 * Whether `key` belongs to `repo` alone: inside it, or one of the files Kepr keeps for it. An empty
 * name is nothing, and no key may escape through `..` or an empty part.
 */
export function isRepoKey(repo: string, key: string): boolean {
	if (repo === "" || key.split("/").some((part) => part === ".." || part === "")) return false;
	if (key.startsWith(`${repo}/`)) return true;
	if (keprRepoFolders(repo).some((folder) => key.startsWith(folder))) return true;
	return keprRepoFiles(repo).includes(key);
}

/** Every key of one repo: what a bucket lists under it, plus the files Kepr keeps for it. */
async function listRepoKeys(
	list: (prefix: string) => Promise<string[]>,
	repo: string,
): Promise<string[]> {
	if (repo === "") return [];
	const keys = await list(`${repo}/`);
	for (const folder of keprRepoFolders(repo)) keys.push(...(await list(folder)));
	return [...keys, ...keprRepoFiles(repo)].filter((key) => isRepoKey(repo, key));
}

export function openBucket(source: BucketSource): Bucket {
	return source.kind === "s3" ? s3Reader(source) : folderReader(source.path);
}

type S3Source = Extract<BucketSource, { kind: "s3" }>;

/**
 * The console's connection test (§24), made before settings are saved: the key must list the
 * bucket and write Kepr's own test file. Returns what failed, or null.
 */
export async function checkS3Access(source: S3Source): Promise<string | null> {
	const reason = (error: unknown) => (error instanceof Error ? error.message : String(error));
	try {
		// One key is proof enough; listing the whole bucket could take minutes.
		await s3Client(source).list({ maxKeys: 1 });
	} catch (error) {
		return `The key cannot list the bucket: ${reason(error)}`;
	}
	try {
		const stamp = new TextEncoder().encode(new Date().toISOString());
		await s3Reader(source).put(`${KEPR_PREFIX}connection-test`, stamp, "text/plain");
	} catch (error) {
		return `The key cannot write ${KEPR_PREFIX}connection-test: ${reason(error)}`;
	}
	return null;
}

function s3Client(source: S3Source): Bun.S3Client {
	return new Bun.S3Client({
		endpoint: source.endpoint,
		bucket: source.name,
		region: source.region,
		accessKeyId: source.accessKeyId,
		secretAccessKey: source.secretAccessKey,
	});
}

function s3Reader(source: S3Source): Bucket {
	const client = s3Client(source);
	// A refused write is a BucketWriteError, which the console shows and a tarball cache skips.
	const write = async (key: string, run: (file: Bun.S3File) => Promise<unknown>) => {
		try {
			await run(client.file(key));
		} catch (error) {
			throw new BucketWriteError(`The bucket refused ${key}: ${error}`);
		}
	};
	return {
		async get(key) {
			// Through a presigned URL: Bun 1.3.14's S3File reads keep about one copy of every body
			// read, which a first sync of a big repo turns into hundreds of MB; fetch frees its own.
			const response = await fetch(client.presign(key, { method: "GET", expiresIn: 60 }));
			if (response.status === 404) return null;
			if (!response.ok) throw new Error(`S3 GET ${key} failed: HTTP ${response.status}`);
			return new Uint8Array(await response.arrayBuffer());
		},
		async list(prefix) {
			const keys: string[] = [];
			let continuationToken: string | undefined;
			do {
				const page = await client.list({ prefix, continuationToken });
				for (const object of page.contents ?? []) keys.push(object.key);
				continuationToken = page.isTruncated ? page.nextContinuationToken : undefined;
			} while (continuationToken);
			return keys;
		},
		size: (key) => client.size(key),
		async folders() {
			const names: string[] = [];
			let continuationToken: string | undefined;
			do {
				const page = await client.list({ delimiter: "/", continuationToken });
				for (const { prefix } of page.commonPrefixes ?? []) names.push(prefix.slice(0, -1));
				continuationToken = page.isTruncated ? page.nextContinuationToken : undefined;
			} while (continuationToken);
			return names;
		},
		async sizes(prefix) {
			const sizes = new Map<string, number>();
			let continuationToken: string | undefined;
			do {
				const page = await client.list({ prefix, continuationToken });
				for (const object of page.contents ?? []) sizes.set(object.key, object.size ?? 0);
				continuationToken = page.isTruncated ? page.nextContinuationToken : undefined;
			} while (continuationToken);
			return sizes;
		},
		downloadUrl(key, filename, type) {
			// Bun signs these as response-content-disposition and response-content-type (verified, v1.3.14).
			return client.presign(key, {
				method: "GET",
				expiresIn: DOWNLOAD_SECONDS,
				contentDisposition: `attachment; filename="${filename}"`,
				type,
			});
		},
		viewUrl(key, type) {
			return client.presign(key, { method: "GET", expiresIn: DOWNLOAD_SECONDS, type });
		},
		async put(key, bytes, type) {
			requireWritable(key, RELEASE_WRITE);
			await write(key, (file) => file.write(bytes, { type }));
		},
		async remove(key) {
			requireWritable(key, RELEASE_FILE);
			await write(key, (file) => file.delete());
		},
		async copy(from, to) {
			requireRepoKey(from);
			requireRepoKey(to);
			const bytes = await this.get(from);
			if (bytes === null) throw new Error(`${from} is missing`);
			await write(to, (file) => file.write(bytes));
		},
		async removeRenamed(key) {
			requireRepoKey(key);
			await write(key, (file) => file.delete());
		},
		async repoKeys(repo) {
			return listRepoKeys((prefix) => this.list(prefix), repo);
		},
		async removeRepoKey(repo, key) {
			if (!isRepoKey(repo, key)) {
				throw new BucketWriteError(`Kepr deletes only ${repo}'s own keys, not "${key}"`);
			}
			await write(key, (file) => file.delete());
		},
	};
}

function folderReader(root: string): Bucket {
	const list = async (prefix: string) => {
		const entries = await readdir(root, { recursive: true, withFileTypes: true });
		return entries
			.filter((entry) => entry.isFile())
			.map((entry) => relative(root, join(entry.parentPath, entry.name)))
			.filter((key) => key.startsWith(prefix))
			.sort();
	};
	return {
		async get(key) {
			if (key.split("/").includes("..")) throw new Error(`bucket key escapes the folder: ${key}`);
			const file = Bun.file(join(root, key));
			return (await file.exists()) ? await file.bytes() : null;
		},
		list,
		async size(key) {
			if (key.split("/").includes("..")) throw new Error(`bucket key escapes the folder: ${key}`);
			return Bun.file(join(root, key)).size;
		},
		async folders() {
			const entries = await readdir(root, { withFileTypes: true });
			return entries
				.filter((entry) => entry.isDirectory())
				.map((entry) => entry.name)
				.sort();
		},
		async sizes(prefix) {
			return new Map((await list(prefix)).map((key) => [key, Bun.file(join(root, key)).size]));
		},
		// Development and tests only: a file:// URL can't carry a filename, a type or an expiry.
		downloadUrl(key) {
			if (key.split("/").includes("..")) throw new Error(`bucket key escapes the folder: ${key}`);
			return pathToFileURL(join(root, key)).href;
		},
		viewUrl: () => null,
		async put(key, bytes) {
			requireWritable(key, RELEASE_WRITE);
			await Bun.write(join(root, key), bytes);
		},
		async remove(key) {
			requireWritable(key, RELEASE_FILE);
			await unlink(join(root, key)).catch((error: NodeJS.ErrnoException) => {
				if (error.code !== "ENOENT") throw error;
			});
		},
		async copy(from, to) {
			requireRepoKey(from);
			requireRepoKey(to);
			await Bun.write(join(root, to), Bun.file(join(root, from)));
		},
		async removeRenamed(key) {
			requireRepoKey(key);
			await unlink(join(root, key));
		},
		async repoKeys(repo) {
			return listRepoKeys(list, repo);
		},
		async removeRepoKey(repo, key) {
			if (!isRepoKey(repo, key)) {
				throw new BucketWriteError(`Kepr deletes only ${repo}'s own keys, not "${key}"`);
			}
			await unlink(join(root, key)).catch((error: NodeJS.ErrnoException) => {
				if (error.code !== "ENOENT") throw error;
			});
		},
	};
}