import {
	createPinnedIdentityFetcher,
	getPdsForDid,
	type IdentityGetFetcher,
	isCanonicalWebhookDid,
	parseWebhookScopeAtUri,
	readBoundedIdentityJson,
	validateWebhookRecord,
} from '@wispplace/atproto-utils'
import { createLogger } from '@wispplace/observability'
import type { OwnerReconciliationToken, WebhookSnapshotRecord } from './db'
import {
	applyWebhookSnapshotPage,
	beginOwnerReconciliation,
	completeOwnerReconciliation,
	failOwnerReconciliation,
	listFailedWebhookReconciliationOwners,
	listKnownWebhookOwnerDidsPage,
} from './db'
import { assertSafeWebhookUrlSyntax } from './webhook-url'

export type { WebhookSnapshotRecord } from './db'

const logger = createLogger('webhook-service:backfill')
// The option is inert unless the transport's two development-only env gates are set.
const backfillIdentityGet = createPinnedIdentityFetcher({ allowLoopback: true })
const MAX_BACKFILL_PAGES = 20
const MAX_BACKFILL_RECORDS = 1_000
// Twenty bounded decoded pages cannot exceed the 10 MiB owner scan budget.
const MAX_BACKFILL_PAGE_BYTES = 512 * 1024
const MAX_CURSOR_LENGTH = 2_048
const DEFAULT_OWNER_PAGE_SIZE = 50
const MAX_OWNER_PAGE_SIZE = 1_000
const DEFAULT_OWNERS_PER_PASS = 1_000
const MAX_OWNERS_PER_PASS = 10_000
const DEFAULT_OWNER_CONCURRENCY = 1
const MAX_OWNER_CONCURRENCY = 4

type PageHandler = (records: readonly WebhookSnapshotRecord[]) => Promise<void> | void

interface ListRecordsResponse {
	records: WebhookSnapshotRecord[]
	omitted: number
	cursor?: string
}

export interface StartupBackfillResult {
	found: number
	failed: number
	/** Present only when a bounded pass must resume from this keyset cursor. */
	nextOwnerCursor?: string
}

export interface StartupBackfillProgress {
	scanned: number
	found: number
	failed: number
	capped: boolean
	nextOwnerCursor?: string
}

export type OwnerReconciliationStatus = 'scanning' | 'complete' | 'failed'
export type OwnerTransitionCallback = (did: string, status: OwnerReconciliationStatus) => Promise<void> | void

export interface StartupBackfillOptions {
	fetcher?: IdentityGetFetcher
	onOwnerTransition?: OwnerTransitionCallback
	/** Bounded keyset page source; injectable for tests and explicit continuation jobs. */
	listOwnerPage?: (after: string | undefined, limit: number) => Promise<readonly string[]>
	ownerPageSize?: number
	maxOwnersPerPass?: number
	concurrency?: number
	startAfter?: string
	onProgress?: (progress: StartupBackfillProgress) => Promise<void> | void
	signal?: AbortSignal
}

function boundedBackfillOption(value: number | undefined, fallback: number, maximum: number, name: string): number {
	const actual = value ?? fallback
	if (!Number.isSafeInteger(actual) || actual < 1 || actual > maximum) throw new Error(`Invalid ${name}`)
	return actual
}

function assertOwnerPage(page: readonly string[], after: string | undefined): void {
	let previous = after
	for (const did of page) {
		if (!isCanonicalWebhookDid(did) || (previous !== undefined && did <= previous)) {
			throw new ReconciliationInfrastructureFailure()
		}
		previous = did
	}
}

async function mapBounded<T>(
	values: readonly T[],
	concurrency: number,
	worker: (value: T) => Promise<void>,
): Promise<void> {
	let cursor = 0
	const run = async () => {
		while (cursor < values.length) {
			const index = cursor++
			await worker(values[index] as T)
		}
	}
	await Promise.all(Array.from({ length: Math.min(concurrency, values.length) }, run))
}

function invalidPdsResponse(): never {
	throw new Error('Webhook backfill response is invalid')
}

function isPlainObject(value: unknown): value is Record<string, unknown> {
	return value !== null && typeof value === 'object' && !Array.isArray(value)
}

function owns(value: object, key: string): boolean {
	return Object.getOwnPropertyDescriptor(value, key) !== undefined
}

function parseWebhookRecord(value: unknown) {
	const validation = validateWebhookRecord(value)
	// A malformed record is omitted, not loosely repaired. A complete snapshot
	// then authoritatively removes any stale local row for its rkey.
	if (!validation.ok) return null
	try {
		// Keep startup intake aligned with live firehose validation. DNS pinning is
		// deliberately deferred to delivery; literal private targets are rejected now.
		assertSafeWebhookUrlSyntax(validation.record.url)
	} catch {
		return null
	}
	return validation.record
}

function parseOwnedWebhookUri(uri: string, did: string): string {
	if (uri.length > 4_096) invalidPdsResponse()
	const parsed = parseWebhookScopeAtUri(uri)
	if (!parsed || parsed.did !== did || parsed.collection !== 'place.wisp.v2.wh' || !parsed.rkey) invalidPdsResponse()
	if (uri !== `at://${did}/place.wisp.v2.wh/${parsed.rkey}`) invalidPdsResponse()
	return parsed.rkey
}

function parseListRecordsResponse(value: unknown, did: string): ListRecordsResponse {
	if (!isPlainObject(value) || !Array.isArray(value.records) || value.records.length > 100) invalidPdsResponse()
	const hasCursor = owns(value, 'cursor')
	if (
		hasCursor &&
		(typeof value.cursor !== 'string' || value.cursor.length === 0 || value.cursor.length > MAX_CURSOR_LENGTH)
	) {
		invalidPdsResponse()
	}

	const seenRkeys = new Set<string>()
	const records = value.records.map((entry): WebhookSnapshotRecord | null => {
		if (
			!isPlainObject(entry) ||
			typeof entry.uri !== 'string' ||
			typeof entry.cid !== 'string' ||
			entry.cid.length === 0 ||
			entry.cid.length > 256
		) {
			invalidPdsResponse()
		}
		const rkey = parseOwnedWebhookUri(entry.uri, did)
		if (seenRkeys.has(rkey)) invalidPdsResponse()
		seenRkeys.add(rkey)
		const record = parseWebhookRecord(entry.value)
		return record ? { rkey, cid: entry.cid, record } : null
	})
	const cleanRecords = records.filter((record): record is WebhookSnapshotRecord => record !== null)
	return {
		records: cleanRecords,
		omitted: records.length - cleanRecords.length,
		cursor: hasCursor ? (value.cursor as string) : undefined,
	}
}

function listRecordsUrl(pdsUrl: string, did: string, cursor: string | undefined): string {
	const params = new URLSearchParams({
		repo: did,
		collection: 'place.wisp.v2.wh',
		limit: '100',
	})
	if (cursor) params.set('cursor', cursor)
	const url = new URL('xrpc/com.atproto.repo.listRecords', `${pdsUrl}/`)
	url.search = params.toString()
	return url.toString()
}

const MAX_PDS_ERROR_BYTES = 16 * 1024

async function discardResponse(response: Response): Promise<void> {
	try {
		await response.body?.cancel()
	} catch {
		// The peer may already have closed the response.
	}
}

/** Read only a small error body so status classification cannot create a response-body DoS. */
async function readBoundedErrorJson(response: Response, signal?: AbortSignal): Promise<unknown> {
	const reader = response.body?.getReader()
	if (!reader) return undefined
	const chunks: Uint8Array[] = []
	let total = 0
	try {
		for (;;) {
			if (signal?.aborted) {
				await reader.cancel()
				return undefined
			}
			const { done, value } = await reader.read()
			if (done) break
			if (total + value.byteLength > MAX_PDS_ERROR_BYTES) {
				await reader.cancel()
				return undefined
			}
			chunks.push(value)
			total += value.byteLength
		}
	} catch {
		return undefined
	} finally {
		reader.releaseLock()
	}
	const bytes = new Uint8Array(total)
	let offset = 0
	for (const chunk of chunks) {
		bytes.set(chunk, offset)
		offset += chunk.byteLength
	}
	try {
		return JSON.parse(new TextDecoder().decode(bytes))
	} catch {
		return undefined
	}
}

/** Only an explicit repo-absence response is safe to treat as an empty snapshot. */
async function isAuthoritativeRepoAbsence(response: Response, did: string, signal?: AbortSignal): Promise<boolean> {
	if (response.status !== 400 && response.status !== 404) return false
	const body = await readBoundedErrorJson(response, signal)
	if (!body || typeof body !== 'object' || Array.isArray(body)) return false
	const error = (body as Record<string, unknown>).error
	const message = (body as Record<string, unknown>).message
	return error === 'RepoNotFound' || (error === 'InvalidRequest' && message === `Could not find repo: ${did}`)
}

async function resolveBackfillPds(did: string, fetcher: IdentityGetFetcher, signal?: AbortSignal): Promise<string> {
	if (!isCanonicalWebhookDid(did)) throw new Error('Webhook backfill DID is invalid')
	const pdsUrl = await getPdsForDid(did, fetcher, { allowLoopback: true }, { signal })
	if (!pdsUrl) throw new OwnerScanFailure('Webhook backfill has no valid PDS endpoint')
	return pdsUrl
}

/**
 * Fetch one owner's records page by page. Every page is fully parsed and
 * bounded before the callback can apply it. A later failure leaves final owner
 * classification to the reconciliation caller rather than finalizing deletes.
 */
export async function fetchWhRecordPages(
	did: string,
	onPage: PageHandler,
	fetcher: IdentityGetFetcher = backfillIdentityGet,
	signal?: AbortSignal,
): Promise<number> {
	const pdsUrl = await resolveBackfillPds(did, fetcher, signal)
	const seenCursors = new Set<string>()
	const seenRkeys = new Set<string>()
	let cursor: string | undefined
	let total = 0

	for (let page = 0; page < MAX_BACKFILL_PAGES; page++) {
		let response: Response
		try {
			response = await fetcher(listRecordsUrl(pdsUrl, did, cursor), { signal })
		} catch {
			throw new OwnerScanFailure('Webhook backfill PDS request failed')
		}
		if (!response.ok) {
			const authoritativeEmpty = await isAuthoritativeRepoAbsence(response, did, signal)
			// Absence is authoritative only before any snapshot page was consumed.
			// A prior empty/fully omitted page with a cursor is still a partial scan.
			if (authoritativeEmpty && page === 0) return total
			await discardResponse(response)
			throw new OwnerScanFailure('Webhook backfill PDS request failed')
		}
		let data: ListRecordsResponse
		try {
			data = parseListRecordsResponse(await readBoundedIdentityJson(response, MAX_BACKFILL_PAGE_BYTES, signal), did)
		} catch (error) {
			if (error instanceof ReconciliationInfrastructureFailure || error instanceof OwnerScanFailure) throw error
			throw new OwnerScanFailure('Webhook backfill response is invalid')
		}
		if (data.omitted > 0)
			logger.warn(`[backfill] Omitted ${data.omitted} invalid webhook record(s) from a snapshot page`)
		if (total + data.records.length > MAX_BACKFILL_RECORDS)
			throw new OwnerScanFailure('Webhook backfill record limit exceeded')
		for (const record of data.records) {
			if (seenRkeys.has(record.rkey)) throw new OwnerScanFailure('Webhook backfill response is invalid')
			seenRkeys.add(record.rkey)
		}
		await onPage(data.records)
		total += data.records.length
		if (data.cursor === undefined) return total
		if (seenCursors.has(data.cursor)) throw new OwnerScanFailure('Webhook backfill response is invalid')
		seenCursors.add(data.cursor)
		cursor = data.cursor
	}
	throw new OwnerScanFailure('Webhook backfill page limit exceeded')
}

/** Compatibility helper for callers that explicitly need a bounded in-memory snapshot. */
export async function fetchWhRecordsForDid(
	did: string,
	fetcher: IdentityGetFetcher = backfillIdentityGet,
	signal?: AbortSignal,
): Promise<WebhookSnapshotRecord[]> {
	const records: WebhookSnapshotRecord[] = []
	await fetchWhRecordPages(
		did,
		(page) => {
			records.push(...page)
		},
		fetcher,
		signal,
	)
	return records
}

class OwnerScanFailure extends Error {
	constructor(message = 'Webhook backfill owner scan failed') {
		super(message)
	}
}

class ReconciliationInfrastructureFailure extends Error {
	constructor() {
		super('Webhook reconciliation infrastructure failed')
	}
}

async function notifyOwnerTransition(
	did: string,
	status: OwnerReconciliationStatus,
	callback: OwnerTransitionCallback | undefined,
): Promise<void> {
	try {
		await callback?.(did, status)
	} catch {
		throw new ReconciliationInfrastructureFailure()
	}
}

async function markOwnerFailed(
	token: OwnerReconciliationToken,
	did: string,
	callback: OwnerTransitionCallback | undefined,
): Promise<void> {
	try {
		await failOwnerReconciliation(token)
	} catch {
		throw new ReconciliationInfrastructureFailure()
	}
	await notifyOwnerTransition(did, 'failed', callback)
}

async function reconcileOwner(
	did: string,
	fetcher: IdentityGetFetcher,
	onOwnerTransition: OwnerTransitionCallback | undefined,
	signal?: AbortSignal,
): Promise<number> {
	let token: OwnerReconciliationToken
	try {
		token = await beginOwnerReconciliation(did)
	} catch {
		throw new ReconciliationInfrastructureFailure()
	}
	try {
		await notifyOwnerTransition(did, 'scanning', onOwnerTransition)
	} catch {
		try {
			await markOwnerFailed(token, did, onOwnerTransition)
		} catch {
			// The scan remains `scanning` (and therefore unmatchable) if the DB is down.
		}
		throw new ReconciliationInfrastructureFailure()
	}

	let found = 0
	try {
		await fetchWhRecordPages(
			did,
			async (records) => {
				let result: { applied: boolean; upserted: number }
				try {
					result = await applyWebhookSnapshotPage(token, records)
				} catch {
					throw new ReconciliationInfrastructureFailure()
				}
				if (!result.applied) throw new OwnerScanFailure()
				found += result.upserted
			},
			fetcher,
			signal,
		)
	} catch (error) {
		if (error instanceof ReconciliationInfrastructureFailure) throw error
		await markOwnerFailed(token, did, onOwnerTransition)
		throw new OwnerScanFailure()
	}

	let complete: { applied: boolean; deleted: number; complete: boolean }
	try {
		do {
			if (signal?.aborted) throw new OwnerScanFailure()
			complete = await completeOwnerReconciliation(token)
			if (!complete.applied) break
			// Each DB call finalizes one bounded, timeout-protected transaction.
			// Continue only after that durable batch commits, so a restart safely
			// resumes from the remaining rows rather than holding a giant lock.
		} while (!complete.complete)
	} catch (error) {
		if (error instanceof OwnerScanFailure) {
			await markOwnerFailed(token, did, onOwnerTransition)
			throw error
		}
		// A transient finalization failure should become retryable once the DB
		// returns. If that write also fails, the stale-scanning recovery query
		// eventually admits this owner after its grace period.
		try {
			await markOwnerFailed(token, did, onOwnerTransition)
		} catch {
			// Preserve the infrastructure failure without assuming DB availability.
		}
		throw new ReconciliationInfrastructureFailure()
	}
	if (!complete!.applied || !complete!.complete) {
		await markOwnerFailed(token, did, onOwnerTransition)
		throw new OwnerScanFailure()
	}
	await notifyOwnerTransition(did, 'complete', onOwnerTransition)
	return found
}

/** Retry one already-degraded owner. Infrastructure failures remain fatal to the caller. */
async function retryFailedWebhookOwner(did: string, options: StartupBackfillOptions = {}): Promise<boolean> {
	try {
		await reconcileOwner(did, options.fetcher ?? backfillIdentityGet, options.onOwnerTransition, options.signal)
		return true
	} catch (error) {
		if (error instanceof ReconciliationInfrastructureFailure) throw error
		return false
	}
}

export interface ReconciliationRetrySchedulerOptions {
	fetcher?: IdentityGetFetcher
	onOwnerTransition?: OwnerTransitionCallback
	listFailedOwners?: (limit: number) => Promise<readonly string[]>
	now?: () => number
	random?: () => number
	setTimer?: (callback: () => void, delayMs: number) => ReturnType<typeof setTimeout>
	clearTimer?: (timer: ReturnType<typeof setTimeout>) => void
	maxOwners?: number
	concurrency?: number
	baseDelayMs?: number
	maxDelayMs?: number
	discoveryIntervalMs?: number
}

export interface ReconciliationRetryHealth {
	running: boolean
	infrastructureHealthy: boolean
	active: number
	queued: number
	scheduled: number
	lastInfrastructureFailureAt?: number
}

export interface ReconciliationRetryStopOptions {
	timeoutMs?: number
	signal?: AbortSignal
}

export interface ReconciliationRetryScheduler {
	start(): Promise<void>
	/** Stops admission immediately, aborts active network reads, and waits only to its deadline. */
	stop(options?: ReconciliationRetryStopOptions): Promise<{ drained: boolean }>
	readonly health: ReconciliationRetryHealth
}

interface ScheduledOwnerRetry {
	timer: ReturnType<typeof setTimeout>
	dueAt: number
	generation: number
}

interface ActiveOwnerRetry {
	generation: number
	controller: AbortController
}

function schedulerInteger(value: number | undefined, fallback: number, min: number, max: number, name: string): number {
	const resolved = value ?? fallback
	if (!Number.isSafeInteger(resolved) || resolved < min || resolved > max) throw new RangeError(`${name} is invalid`)
	return resolved
}

function schedulerRandom(random: () => number): number {
	try {
		const value = random()
		return Number.isFinite(value) ? Math.min(Math.max(value, 0), 0.999_999_999) : 0.5
	} catch {
		return 0.5
	}
}

const MIN_RECONCILIATION_RETRY_DELAY_MS = 100
const MAX_RECONCILIATION_RETRY_ATTEMPTS = 100
const DEFAULT_RECONCILIATION_STOP_TIMEOUT_MS = 10_000
const MAX_RECONCILIATION_STOP_TIMEOUT_MS = 120_000

class OwnerRetryScheduler implements ReconciliationRetryScheduler {
	readonly #fetcher: IdentityGetFetcher
	readonly #transition: OwnerTransitionCallback | undefined
	readonly #listFailedOwners: (limit: number) => Promise<readonly string[]>
	readonly #now: () => number
	readonly #random: () => number
	readonly #setTimer: (callback: () => void, delayMs: number) => ReturnType<typeof setTimeout>
	readonly #clearTimer: (timer: ReturnType<typeof setTimeout>) => void
	readonly #maxOwners: number
	readonly #concurrency: number
	readonly #baseDelayMs: number
	readonly #maxDelayMs: number
	readonly #discoveryIntervalMs: number
	readonly #attempts = new Map<string, number>()
	readonly #scheduled = new Map<string, ScheduledOwnerRetry>()
	readonly #queued = new Set<string>()
	readonly #active = new Map<string, ActiveOwnerRetry>()
	#discoveryTimer: ReturnType<typeof setTimeout> | undefined
	#generation = 0
	#started = false
	#stopped = false
	#infrastructureHealthy = true
	#lastInfrastructureFailureAt: number | undefined
	#drainWaiters: Array<() => void> = []

	constructor(options: ReconciliationRetrySchedulerOptions) {
		this.#fetcher = options.fetcher ?? backfillIdentityGet
		this.#transition = options.onOwnerTransition
		this.#listFailedOwners = options.listFailedOwners ?? listFailedWebhookReconciliationOwners
		this.#now = options.now ?? Date.now
		this.#random = options.random ?? Math.random
		this.#setTimer = options.setTimer ?? setTimeout
		this.#clearTimer = options.clearTimer ?? clearTimeout
		this.#maxOwners = schedulerInteger(options.maxOwners, 16, 1, 100, 'Reconciliation retry owner limit')
		this.#concurrency = schedulerInteger(options.concurrency, 1, 1, 2, 'Reconciliation retry concurrency')
		this.#baseDelayMs = schedulerInteger(options.baseDelayMs, 5_000, 0, 60_000, 'Reconciliation retry base delay')
		this.#maxDelayMs = schedulerInteger(
			options.maxDelayMs,
			60 * 60_000,
			1_000,
			60 * 60_000,
			'Reconciliation retry max delay',
		)
		this.#discoveryIntervalMs = schedulerInteger(
			options.discoveryIntervalMs,
			60_000,
			1_000,
			60 * 60_000,
			'Reconciliation retry discovery interval',
		)
	}

	get health(): ReconciliationRetryHealth {
		return {
			running: this.#started && !this.#stopped,
			infrastructureHealthy: this.#infrastructureHealthy,
			active: this.#active.size,
			queued: this.#queued.size,
			scheduled: this.#scheduled.size,
			...(this.#lastInfrastructureFailureAt === undefined
				? {}
				: { lastInfrastructureFailureAt: this.#lastInfrastructureFailureAt }),
		}
	}

	async start(): Promise<void> {
		if (this.#started && !this.#stopped) return
		this.#generation++
		this.#started = true
		this.#stopped = false
		await this.#discover(this.#generation)
	}

	async stop(options: ReconciliationRetryStopOptions = {}): Promise<{ drained: boolean }> {
		this.#generation++
		this.#stopped = true
		this.#started = false
		if (this.#discoveryTimer) this.#clearTimer(this.#discoveryTimer)
		this.#discoveryTimer = undefined
		for (const { timer } of this.#scheduled.values()) this.#clearTimer(timer)
		this.#scheduled.clear()
		this.#queued.clear()
		for (const active of this.#active.values()) active.controller.abort()
		if (this.#active.size === 0) return { drained: true }
		const timeoutMs = schedulerInteger(
			options.timeoutMs,
			DEFAULT_RECONCILIATION_STOP_TIMEOUT_MS,
			0,
			MAX_RECONCILIATION_STOP_TIMEOUT_MS,
			'Reconciliation retry stop timeout',
		)
		return new Promise((resolve) => {
			let settled = false
			let onDrain: (() => void) | undefined
			const finish = (drained: boolean) => {
				if (settled) return
				settled = true
				clearTimeout(timer)
				options.signal?.removeEventListener('abort', abort)
				if (onDrain) {
					const index = this.#drainWaiters.indexOf(onDrain)
					if (index >= 0) this.#drainWaiters.splice(index, 1)
				}
				resolve({ drained })
			}
			const abort = () => finish(false)
			const timer = setTimeout(abort, timeoutMs)
			onDrain = () => finish(true)
			if (options.signal?.aborted) abort()
			else options.signal?.addEventListener('abort', abort, { once: true })
			if (!settled) this.#drainWaiters.push(onDrain)
		})
	}

	#reportInfrastructureFailure(): void {
		this.#infrastructureHealthy = false
		this.#lastInfrastructureFailureAt = this.#now()
	}

	#scheduleDiscovery(generation: number): void {
		if (this.#stopped || !this.#started || generation !== this.#generation) return
		this.#discoveryTimer = this.#setTimer(() => {
			this.#discoveryTimer = undefined
			void this.#discover(generation)
		}, this.#discoveryIntervalMs)
	}

	async #discover(generation: number): Promise<void> {
		if (this.#stopped || !this.#started || generation !== this.#generation) return
		try {
			const owners = await this.#listFailedOwners(this.#maxOwners)
			if (this.#stopped || !this.#started || generation !== this.#generation) return
			this.#infrastructureHealthy = true
			for (const ownerDid of owners) this.#scheduleOwner(ownerDid, this.#attempts.get(ownerDid) ?? 0, generation)
		} catch {
			if (generation === this.#generation) this.#reportInfrastructureFailure()
		} finally {
			this.#scheduleDiscovery(generation)
		}
	}

	#delayFor(attempt: number): number {
		const exponent = Math.min(Math.max(attempt, 0), 16)
		const cap = Math.max(
			MIN_RECONCILIATION_RETRY_DELAY_MS,
			Math.min(this.#baseDelayMs * 2 ** exponent, this.#maxDelayMs),
		)
		return Math.max(MIN_RECONCILIATION_RETRY_DELAY_MS, Math.floor(schedulerRandom(this.#random) * cap))
	}

	#rememberAttempt(did: string, attempt: number): void {
		this.#attempts.delete(did)
		this.#attempts.set(did, attempt)
		while (this.#attempts.size > MAX_RECONCILIATION_RETRY_ATTEMPTS) {
			const oldest = this.#attempts.keys().next().value as string
			this.#attempts.delete(oldest)
		}
	}

	#scheduleOwner(did: string, attempt: number, generation: number): void {
		const trackedOwners = this.#scheduled.size + this.#queued.size + this.#active.size
		if (
			this.#stopped ||
			!this.#started ||
			!isCanonicalWebhookDid(did) ||
			this.#scheduled.has(did) ||
			this.#queued.has(did) ||
			this.#active.has(did) ||
			trackedOwners >= this.#maxOwners ||
			generation !== this.#generation
		) {
			return
		}
		const delay = this.#delayFor(attempt)
		const timer = this.#setTimer(() => {
			if (this.#scheduled.get(did)?.generation !== generation) return
			this.#scheduled.delete(did)
			if (this.#stopped || !this.#started || generation !== this.#generation) return
			this.#queued.add(did)
			this.#pump(generation)
		}, delay)
		this.#scheduled.set(did, { timer, dueAt: this.#now() + delay, generation })
	}

	#pump(generation: number): void {
		while (
			!this.#stopped &&
			generation === this.#generation &&
			this.#active.size < this.#concurrency &&
			this.#queued.size > 0
		) {
			const did = this.#queued.values().next().value as string
			this.#queued.delete(did)
			const controller = new AbortController()
			this.#active.set(did, { generation, controller })
			void this.#runOwner(did, generation, controller.signal).then(
				(shouldRetry) => this.#finishOwner(did, generation, shouldRetry),
				() => this.#finishOwner(did, generation, true),
			)
		}
	}

	async #runOwner(did: string, generation: number, signal: AbortSignal): Promise<boolean> {
		try {
			const recovered = await retryFailedWebhookOwner(did, {
				fetcher: this.#fetcher,
				onOwnerTransition: this.#transition,
				signal,
			})
			if (recovered) {
				if (generation === this.#generation) {
					this.#attempts.delete(did)
					this.#infrastructureHealthy = true
				}
				return false
			}
		} catch {
			if (generation === this.#generation) this.#reportInfrastructureFailure()
		}
		if (this.#stopped || !this.#started || generation !== this.#generation) return false
		const nextAttempt = (this.#attempts.get(did) ?? 0) + 1
		this.#rememberAttempt(did, nextAttempt)
		return true
	}

	#finishOwner(did: string, generation: number, shouldRetry: boolean): void {
		if (this.#active.get(did)?.generation !== generation) return
		this.#active.delete(did)
		if (shouldRetry && !this.#stopped && this.#started && generation === this.#generation) {
			this.#scheduleOwner(did, this.#attempts.get(did) ?? 0, generation)
		}
		this.#pump(generation)
		if (this.#active.size === 0) {
			for (const resolve of this.#drainWaiters.splice(0)) resolve()
		}
	}
}

/** Create a bounded, stoppable failed-owner reconciliation retry lifecycle. */
export function createReconciliationRetryScheduler(
	options: ReconciliationRetrySchedulerOptions = {},
): ReconciliationRetryScheduler {
	return new OwnerRetryScheduler(options)
}

/**
 * On startup, reconcile every known owner through a fenced, page-at-a-time
 * scan. One unavailable owner becomes degraded; healthy owners still complete.
 */
export async function runStartupBackfill(
	fetcherOrOptions: IdentityGetFetcher | StartupBackfillOptions = backfillIdentityGet,
	callback?: OwnerTransitionCallback,
): Promise<StartupBackfillResult> {
	const options =
		typeof fetcherOrOptions === 'function'
			? { fetcher: fetcherOrOptions, onOwnerTransition: callback }
			: fetcherOrOptions
	const fetcher = options.fetcher ?? backfillIdentityGet
	const onOwnerTransition = options.onOwnerTransition ?? callback
	const listOwnerPage = options.listOwnerPage ?? listKnownWebhookOwnerDidsPage
	const pageSize = boundedBackfillOption(
		options.ownerPageSize,
		DEFAULT_OWNER_PAGE_SIZE,
		MAX_OWNER_PAGE_SIZE,
		'known owner page size',
	)
	const maxOwners = boundedBackfillOption(
		options.maxOwnersPerPass,
		DEFAULT_OWNERS_PER_PASS,
		MAX_OWNERS_PER_PASS,
		'known owner pass limit',
	)
	const concurrency = boundedBackfillOption(
		options.concurrency,
		DEFAULT_OWNER_CONCURRENCY,
		MAX_OWNER_CONCURRENCY,
		'known owner concurrency',
	)
	let after = options.startAfter
	let scanned = 0
	let found = 0
	let failed = 0
	let capped = false
	const progress = async (nextOwnerCursor?: string) => {
		await options.onProgress?.({ scanned, found, failed, capped, nextOwnerCursor })
	}

	for (;;) {
		if (options.signal?.aborted) break
		const remaining = maxOwners - scanned
		if (remaining <= 0) {
			capped = true
			break
		}
		const requestLimit = Math.min(pageSize, remaining)
		let page: readonly string[]
		try {
			page = await listOwnerPage(after, requestLimit)
			if (page.length > requestLimit) throw new ReconciliationInfrastructureFailure()
			assertOwnerPage(page, after)
		} catch (error) {
			if (error instanceof ReconciliationInfrastructureFailure) throw error
			throw new ReconciliationInfrastructureFailure()
		}
		if (page.length === 0) break
		// The page is the only queued owner set. Workers take at most `concurrency`
		// entries at once, so neither a large DB nor a slow PDS can grow memory.
		await mapBounded(page, concurrency, async (did) => {
			if (options.signal?.aborted) return
			try {
				found += await reconcileOwner(did, fetcher, onOwnerTransition, options.signal)
			} catch (error) {
				if (error instanceof ReconciliationInfrastructureFailure) throw error
				failed++
				logger.warn('[backfill] DID scan failed')
			}
			scanned++
		})
		if (options.signal?.aborted) break
		after = page[page.length - 1]
		if (scanned >= maxOwners) {
			capped = true
			await progress(after)
			break
		}
		// A caller can persist this completed page cursor immediately. A crash
		// may replay at most one completed owner page, never require an unbounded
		// rescan from the beginning.
		await progress(after)
		if (page.length < requestLimit) break
	}

	if (scanned === 0) logger.info('[backfill] No known DIDs to scan')
	logger.info(`[backfill] Done — ${found} webhook record(s) imported, ${failed} DID(s) failed`)
	return capped && after ? { found, failed, nextOwnerCursor: after } : { found, failed }
}