import { type Client, ClientResponseError, ok } from "@atcute/client";
import { type Did, isRecordKey, type Nsid } from "@atcute/lexicons/syntax";
import {
	type FirehoseCommit,
	handleCommitEvent,
} from "../../firehose/handlers.ts";
import { getNoteByAtUri, recordLanded } from "../../server/db/queries/index.ts";
import { COLLECTIONS } from "../collections.ts";
import { resolvePdsClient } from "./pds-client.ts";
import {
	addError,
	type BackfillReport,
	countIngest,
	countSkip,
	countUnchanged,
	createReport,
	setEndpoint,
} from "./report.ts";

interface BackoffConfig {
	baseMs: number;
	maxMs: number;
	maxRetries: number;
}

const DEFAULT_BACKOFF: BackoffConfig = {
	baseMs: 1000,
	maxMs: 30_000,
	maxRetries: 5,
};
const DEFAULT_THROTTLE_MS = 100;
const PAGE_LIMIT = 100;

export interface BackfillOptions {
	/** Log/count what would be ingested without dispatching to the handler. */
	dryRun?: boolean;
	/** Restrict the crawl to these collection NSIDs (the --collection flag). */
	collections?: string[];
	/** Politeness delay between paginated calls (ms). */
	throttleMs?: number;
	backoff?: BackoffConfig;
	// --- Injection seams (tests) ---
	resolveClient?: (
		did: string,
	) => Promise<{ client: Client; endpoint: string } | null>;
	dispatch?: (commit: FirehoseCommit) => void;
	noteExists?: (uri: string) => boolean;
	sleep?: (ms: number) => Promise<void>;
}

const realSleep = (ms: number): Promise<void> =>
	new Promise((resolve) => setTimeout(resolve, ms));

const defaultNoteExists = (uri: string): boolean =>
	getNoteByAtUri(uri) !== null;

interface ListRecordsPage {
	records: { uri: string; cid: string; value: unknown }[];
	cursor?: string;
}

// One listRecords page with retry. Retries 429/5xx/network with exponential
// backoff; rethrows other 4xx immediately (caller logs and skips the pair).
async function listPage(
	client: Client,
	did: string,
	collection: string,
	cursor: string | undefined,
	reverse: boolean,
	options: BackfillOptions,
): Promise<ListRecordsPage> {
	const cfg = options.backoff ?? DEFAULT_BACKOFF;
	const sleep = options.sleep ?? realSleep;
	let attempt = 0;

	for (;;) {
		try {
			const res = await ok(
				client.get("com.atproto.repo.listRecords", {
					params: {
						repo: did as Did,
						collection: collection as Nsid,
						limit: PAGE_LIMIT,
						reverse,
						...(cursor ? { cursor } : {}),
					},
				}),
			);
			return res as ListRecordsPage;
		} catch (err) {
			const status = err instanceof ClientResponseError ? err.status : 0; // 0 = network/timeout
			// Non-429 client errors aren't worth retrying (e.g. deactivated account → 400).
			const retryable = status === 0 || status === 429 || status >= 500;
			if (!retryable || attempt >= cfg.maxRetries) throw err;

			const delay = Math.min(cfg.baseMs * 2 ** attempt, cfg.maxMs);
			await sleep(delay);
			attempt++;
		}
	}
}

// The uri is whatever the remote PDS returned, so the rkey needs the same
// validation jetstream applies to live commits — otherwise backfill is a second,
// unchecked door into handleCommitEvent.
function rkeyOf(uri: string): string | null {
	const rkey = uri.split("/").pop() ?? "";
	return isRecordKey(rkey) ? rkey : null;
}

// Feed one listRecords row into the firehose handler and classify the *outcome*
// by checking the DB before and after dispatch — the handler silently drops
// records it can't place, so "dispatched" must not be counted as "ingested".
// Revisions keep a parent-missing pre-check for a precise reason; other drops are
// reported generically as "rejected".
function dispatchRow(
	did: string,
	collection: string,
	row: { uri: string; value: unknown },
	report: BackfillReport,
	options: BackfillOptions,
): void {
	// Dry-run can't observe outcomes (nothing is written), so it only inventories
	// what the PDS holds. Skipping the dispatch-dependent parent-missing check here
	// also avoids false positives (the parent notes were never written).
	if (options.dryRun) {
		countIngest(report, did, collection);
		return;
	}

	const rkey = rkeyOf(row.uri);
	if (rkey === null) {
		countSkip(report, did, collection, "invalid-rkey");
		return;
	}

	if (collection === COLLECTIONS.noteRevision) {
		const noteRef = (row.value as { noteRef?: unknown }).noteRef;
		const noteExists = options.noteExists ?? defaultNoteExists;
		if (typeof noteRef !== "string" || !noteExists(noteRef)) {
			countSkip(report, did, collection, "parent-missing");
			return;
		}
	}

	const existedBefore = recordLanded(collection, did, row.uri, row.value);
	const dispatch = options.dispatch ?? handleCommitEvent;
	dispatch({
		did,
		collection,
		rkey,
		operation: "create",
		record: row.value,
	});

	if (!recordLanded(collection, did, row.uri, row.value)) {
		// Dispatched but not present afterwards: the handler declined it.
		countSkip(report, did, collection, "rejected");
	} else if (existedBefore) {
		countUnchanged(report, did, collection);
	} else {
		countIngest(report, did, collection);
	}
}

// Paginate one (DID, collection) pair to exhaustion, dispatching each record.
async function crawlCollection(
	client: Client,
	did: string,
	collection: string,
	reverse: boolean,
	report: BackfillReport,
	options: BackfillOptions,
): Promise<void> {
	const sleep = options.sleep ?? realSleep;
	const throttleMs = options.throttleMs ?? DEFAULT_THROTTLE_MS;
	let cursor: string | undefined;

	do {
		let page: ListRecordsPage;
		try {
			page = await listPage(client, did, collection, cursor, reverse, options);
		} catch (err) {
			const status = err instanceof ClientResponseError ? err.status : 0;
			// 4xx (other than 429): the PDS reachable but rejected — skip this pair,
			// not a fatal DID failure. Network/5xx exhaustion is fatal.
			const fatal = status === 0 || status === 429 || status >= 500;
			addError(report, did, `${collection}: ${String(err)}`, fatal);
			return;
		}

		for (const row of page.records) {
			dispatchRow(did, collection, row, report, options);
		}

		cursor = page.cursor;
		// Stop on an absent cursor or an empty page (defends against PDSes that
		// echo a stable cursor forever).
		if (cursor && page.records.length > 0) {
			await sleep(throttleMs);
		} else {
			cursor = undefined;
		}
	} while (cursor);
}

interface Stage {
	collections: string[];
	reverse: boolean;
}

// Cross-collection ordering matters, for two reasons. References: revisions point
// at notes, notes/memberships point at wikis. Authorization: handleNote and
// handleRevision both call canDidEditWiki, which reads memberships — so
// memberships must land before notes or every non-owner's content is dropped.
// Crawl every DID through each stage before advancing.
const STAGES: Stage[] = [
	{ collections: [COLLECTIONS.wiki], reverse: false },
	{ collections: [COLLECTIONS.membership], reverse: false },
	{ collections: [COLLECTIONS.note], reverse: false },
	// reverse: true → ascending TID (oldest first) so the diff chain applies in order.
	{ collections: [COLLECTIONS.noteRevision], reverse: true },
	{
		collections: [COLLECTIONS.memberRequest, COLLECTIONS.bookmark],
		reverse: false,
	},
];

export async function runBackfill(
	dids: string[],
	options: BackfillOptions = {},
): Promise<BackfillReport> {
	const report = createReport();
	const resolveClient = options.resolveClient ?? resolvePdsClient;
	const allowed = options.collections ? new Set(options.collections) : null;

	// Resolve every PDS up front; a DID with no reachable PDS is fatal but doesn't
	// abort the others.
	const clients = new Map<string, Client>();
	for (const did of dids) {
		const resolved = await resolveClient(did);
		if (!resolved) {
			setEndpoint(report, did, null);
			addError(report, did, "could not resolve PDS endpoint", true);
			continue;
		}
		setEndpoint(report, did, resolved.endpoint);
		clients.set(did, resolved.client);
	}

	for (const stage of STAGES) {
		for (const [did, client] of clients) {
			for (const collection of stage.collections) {
				if (allowed && !allowed.has(collection)) continue;
				await crawlCollection(
					client,
					did,
					collection,
					stage.reverse,
					report,
					options,
				);
			}
		}
	}

	return report;
}