import { fromUint8Array } from "@atcute/repo";
import { createBobbinClient } from "#lib/api/client.ts";
import { jsonGet, rawGet } from "#lib/api/_request.ts";
import { nsidOf, recordKeyOf, type Nsid, type Rkey } from "#lib/api/syntax.ts";
import type { Fetch } from "#lib/fetch.ts";

const IMPORT_PREFIX = "org.tangled.";

export interface LiarRecord {
	collection: Nsid;
	rkey: Rkey;
	value: { $type?: string };
}

export const fetchLiarRepo = async (
	liarPdsUrl: string,
	did: string,
	{ fetch, signal }: { fetch?: Fetch; signal?: AbortSignal } = {}
): Promise<ReadableStream<Uint8Array>> => {
	const ctx = createBobbinClient({ serviceUrl: liarPdsUrl, fetch });
	const response = await rawGet(ctx, "com.atproto.sync.getRepo", { did }, { signal });
	if (!response.body) throw new Error("The liar PDS sent an empty repo.");
	return response.body;
};

export const listLiarBlobs = async (
	liarPdsUrl: string,
	did: string,
	{ fetch, signal }: { fetch?: Fetch; signal?: AbortSignal } = {}
): Promise<string[]> => {
	const ctx = createBobbinClient({ serviceUrl: liarPdsUrl, fetch });
	const cids: string[] = [];
	let cursor: string | undefined;
	const seenCursors = new Set<string>();
	do {
		const page = await jsonGet<{ cids: string[]; cursor?: string }>(
			ctx,
			"com.atproto.sync.listBlobs",
			{ did, limit: 1000, cursor },
			{ signal }
		);
		cids.push(...page.cids);
		cursor = page.cids.length === 0 ? undefined : page.cursor;
		if (cursor !== undefined) {
			if (seenCursors.has(cursor)) {
				throw new Error("The liar PDS keeps returning the same page of blobs.");
			}
			seenCursors.add(cursor);
		}
	} while (cursor);
	return cids;
};

export const fetchLiarBlob = async (
	liarPdsUrl: string,
	did: string,
	cid: string,
	{ fetch, signal }: { fetch?: Fetch; signal?: AbortSignal } = {}
): Promise<Blob> => {
	const ctx = createBobbinClient({ serviceUrl: liarPdsUrl, fetch });
	const response = await rawGet(ctx, "com.atproto.sync.getBlob", { did, cid }, { signal });
	return response.blob();
};

export const readLiarRecords = async (
	stream: ReadableStream<Uint8Array>
): Promise<LiarRecord[]> => {
	// the streamed reader drops a block that two identical records share
	const car = new Uint8Array(await new Response(stream).arrayBuffer());
	let entries;
	try {
		entries = [...fromUint8Array(car)];
	} catch (cause) {
		// a truncated car must not end in an unmark over records we never wrote
		throw new Error("The repo from the liar PDS is incomplete.", { cause });
	}
	return entries
		.filter((entry) => entry.collection.startsWith(IMPORT_PREFIX))
		.map((entry) => ({
			collection: nsidOf(entry.collection),
			rkey: recordKeyOf(entry.rkey),
			value: entry.record as LiarRecord["value"]
		}));
};