/**
 * Bounded, cancellation-aware control flow for firehose backfills.
 *
 * Network/storage operations still need to observe the supplied AbortSignal.
 * These helpers guarantee that cancellation admits no new pages or items and
 * that a broken upstream cursor cannot create an infinite scan.
 */

export type HydrantRepoRow = { did?: string }

export type HydrantPageFetcher = (
	cursor: string | undefined,
	limit: number,
	signal: AbortSignal,
) => Promise<HydrantRepoRow[]>

export interface HydrantScanOptions {
	pageSize?: number
	maxPages?: number
	maxDids?: number
	signal: AbortSignal
}

export class BackfillAbortedError extends Error {
	constructor() {
		super('Backfill cancelled')
		this.name = 'BackfillAbortedError'
	}
}

export class HydrantPaginationError extends Error {
	constructor(readonly code: 'INVALID_PAGE' | 'REPEATED_CURSOR' | 'PAGE_LIMIT' | 'DID_LIMIT') {
		super(`Hydrant pagination failed: ${code}`)
		this.name = 'HydrantPaginationError'
	}
}

export function throwIfBackfillAborted(signal: AbortSignal): void {
	if (signal.aborted) throw new BackfillAbortedError()
}

export async function scanHydrantDids(fetchPage: HydrantPageFetcher, options: HydrantScanOptions): Promise<string[]> {
	const pageSize = options.pageSize ?? 1000
	const maxPages = options.maxPages ?? 10_000
	const maxDids = options.maxDids ?? 5_000_000
	if (!Number.isSafeInteger(pageSize) || pageSize < 1) throw new RangeError('pageSize must be a positive integer')
	if (!Number.isSafeInteger(maxPages) || maxPages < 1) throw new RangeError('maxPages must be a positive integer')
	if (!Number.isSafeInteger(maxDids) || maxDids < 1) throw new RangeError('maxDids must be a positive integer')

	const dids = new Set<string>()
	let cursor: string | undefined

	for (let pageNumber = 0; pageNumber < maxPages; pageNumber++) {
		throwIfBackfillAborted(options.signal)
		let rows: HydrantRepoRow[]
		try {
			rows = await fetchPage(cursor, pageSize, options.signal)
		} catch (error) {
			if (options.signal.aborted) throw new BackfillAbortedError()
			throw error
		}
		throwIfBackfillAborted(options.signal)
		if (!Array.isArray(rows)) throw new HydrantPaginationError('INVALID_PAGE')
		if (rows.length === 0) return [...dids]

		for (const row of rows) {
			if (typeof row?.did !== 'string' || row.did.length === 0) continue
			dids.add(row.did)
			if (dids.size > maxDids) throw new HydrantPaginationError('DID_LIMIT')
		}

		const lastDid = rows[rows.length - 1]?.did
		if (!lastDid || rows.length < pageSize) return [...dids]
		if (lastDid === cursor) throw new HydrantPaginationError('REPEATED_CURSOR')
		cursor = lastDid
	}

	throw new HydrantPaginationError('PAGE_LIMIT')
}

/**
 * Run a fixed worker pool. Aborting admits no new items and waits for already
 * admitted work to settle, so shutdown can safely close shared dependencies.
 */
export async function runCancellableWindow<T>(
	items: readonly T[],
	concurrency: number,
	signal: AbortSignal,
	worker: (item: T, signal: AbortSignal) => Promise<void>,
): Promise<void> {
	if (!Number.isSafeInteger(concurrency) || concurrency < 1) {
		throw new RangeError('concurrency must be a positive integer')
	}

	let nextIndex = 0
	const runWorker = async () => {
		while (!signal.aborted) {
			const index = nextIndex
			if (index >= items.length) return
			nextIndex++
			await worker(items[index] as T, signal)
		}
	}

	const workers = Array.from({ length: Math.min(concurrency, items.length) }, () => runWorker())
	const results = await Promise.allSettled(workers)
	const failed = results.find((result): result is PromiseRejectedResult => result.status === 'rejected')
	if (failed) throw failed.reason
}