import { describe, expect, mock, test } from 'bun:test'
import { MAX_IDENTITY_JSON_BYTES } from '@wispplace/atproto-utils'

let knownDids: string[] = []
const began: Array<{ ownerDid: string; generation: number }> = []
const appliedPages: Array<{ ownerDid: string; records: ReadonlyArray<{ rkey: string }> }> = []
const completed: Array<{ ownerDid: string; generation: number }> = []
const failed: Array<{ ownerDid: string; generation: number }> = []
const transitions: Array<{ did: string; status: string }> = []
let beginFailure = false

mock.module('@wispplace/observability', () => ({
	createLogger: () => ({
		debug: () => undefined,
		error: () => undefined,
		info: () => undefined,
		warn: () => undefined,
	}),
}))

mock.module('./db', () => ({
	listKnownWebhookOwnerDidsPage: async (after?: string, limit = 100) =>
		knownDids.filter((did) => after === undefined || did > after).slice(0, limit),
	listFailedWebhookReconciliationOwners: async () => [],
	beginOwnerReconciliation: async (ownerDid: string) => {
		if (beginFailure) throw new Error('database credentials must remain private')
		const token = { ownerDid, generation: began.length + 1 }
		began.push(token)
		return token
	},
	applyWebhookSnapshotPage: async (token: { ownerDid: string }, records: ReadonlyArray<{ rkey: string }>) => {
		appliedPages.push({ ownerDid: token.ownerDid, records })
		return { applied: true, upserted: records.length }
	},
	completeOwnerReconciliation: async (token: { ownerDid: string; generation: number }) => {
		completed.push(token)
		return { applied: true, deleted: 0, complete: true }
	},
	failOwnerReconciliation: async (token: { ownerDid: string; generation: number }) => {
		failed.push(token)
	},
}))

const { createReconciliationRetryScheduler, fetchWhRecordPages, fetchWhRecordsForDid, runStartupBackfill } =
	await import('./backfill')

const DID = 'did:web:example.com'
const OTHER_DID = 'did:web:other.example'
const validRecord = {
	$type: 'place.wisp.v2.wh',
	createdAt: '2025-01-01T00:00:00.000Z',
	enabled: true,
	scope: { aturi: 'at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa' },
	url: 'https://receiver.example/hook',
}

const reset = () => {
	knownDids = []
	began.length = 0
	appliedPages.length = 0
	completed.length = 0
	failed.length = 0
	transitions.length = 0
	beginFailure = false
}
const recordUri = (did: string, rkey: string): string => `at://${did}/place.wisp.v2.wh/${rkey}`
const didDocument = (endpoint: string): Response =>
	new Response(JSON.stringify({ service: [{ id: '#atproto_pds', serviceEndpoint: endpoint }] }))

const pdsFetcher =
	(listResponse: Response) =>
	async (url: string): Promise<Response> =>
		url === 'https://example.com/.well-known/did.json' ? didDocument('https://pds.example') : listResponse

describe('webhook startup backfill PDS transport and snapshot validation', () => {
	test('rejects a private PDS endpoint before any PDS list request', async () => {
		reset()
		const requests: string[] = []
		await expect(
			fetchWhRecordsForDid(DID, async (url) => {
				requests.push(url)
				return didDocument('https://169.254.169.254')
			}),
		).rejects.toThrow('no valid PDS endpoint')
		expect(requests).toEqual(['https://example.com/.well-known/did.json'])
	})

	test('uses the injected pinned GET path for every complete PDS page and encodes cursors', async () => {
		reset()
		const requests: string[] = []
		const records = await fetchWhRecordsForDid(DID, async (url) => {
			requests.push(url)
			if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
			const parsed = new URL(url)
			if (parsed.searchParams.get('cursor') === 'next cursor') {
				return new Response(JSON.stringify({ records: [{ uri: recordUri(DID, 'two'), cid: 'b', value: validRecord }] }))
			}
			return new Response(
				JSON.stringify({
					records: [{ uri: recordUri(DID, 'one'), cid: 'a', value: validRecord }],
					cursor: 'next cursor',
				}),
			)
		})

		expect(records.map((record) => record.rkey)).toEqual(['one', 'two'])
		expect(requests).toHaveLength(3)
		expect(requests[1]).toContain('https://pds.example/xrpc/com.atproto.repo.listRecords?')
		expect(requests[2]).toContain('cursor=next+cursor')
	})

	test('rejects non-canonical URIs, malformed cursors, duplicate page keys, and non-success pages', async () => {
		reset()
		const invalidResponses = [
			{ records: [{ uri: `at://${OTHER_DID}/place.wisp.v2.wh/one`, cid: 'a', value: validRecord }] },
			{ records: [], cursor: 7 },
		]
		for (const body of invalidResponses) {
			await expect(fetchWhRecordsForDid(DID, pdsFetcher(new Response(JSON.stringify(body))))).rejects.toThrow(
				'Webhook backfill',
			)
		}
		await expect(fetchWhRecordsForDid(DID, pdsFetcher(new Response(null, { status: 404 })))).rejects.toThrow(
			'Webhook backfill PDS request failed',
		)
		await expect(
			fetchWhRecordPages(
				DID,
				() => undefined,
				async (url) => {
					if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
					const cursor = new URL(url).searchParams.get('cursor')
					return new Response(
						JSON.stringify(
							cursor
								? { records: [{ uri: recordUri(DID, 'one'), cid: 'b', value: validRecord }] }
								: { records: [{ uri: recordUri(DID, 'one'), cid: 'a', value: validRecord }], cursor: 'next' },
						),
					)
				},
			),
		).rejects.toThrow('Webhook backfill')
	})

	test('omits malformed records while applying valid rows and completing the owner snapshot', async () => {
		reset()
		knownDids = [DID]
		const result = await runStartupBackfill(async (url) => {
			if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
			return new Response(
				JSON.stringify({
					records: [
						{ uri: recordUri(DID, 'valid'), cid: 'a', value: validRecord },
						{ uri: recordUri(DID, 'malformed'), cid: 'b', value: { $type: 'place.wisp.v2.wh', enabled: 'yes' } },
					],
				}),
			)
		})
		expect(result).toEqual({ found: 1, failed: 0 })
		expect(appliedPages.map((page) => page.records.map((record) => record.rkey))).toEqual([['valid']])
		expect(completed).toEqual([{ ownerDid: DID, generation: 1 }])
	})

	test('treats an explicit repo-absence response as an authoritative empty snapshot', async () => {
		reset()
		knownDids = [DID]
		const result = await runStartupBackfill({
			fetcher: async (url) => {
				if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
				return new Response(JSON.stringify({ error: 'RepoNotFound', message: 'Repo not found' }), { status: 400 })
			},
			onOwnerTransition: (did, status) => {
				transitions.push({ did, status })
			},
		})

		expect(result).toEqual({ found: 0, failed: 0 })
		expect(appliedPages).toEqual([])
		expect(completed).toEqual([{ ownerDid: DID, generation: 1 }])
		expect(failed).toEqual([])
		expect(transitions).toEqual([
			{ did: DID, status: 'scanning' },
			{ did: DID, status: 'complete' },
		])
	})

	test('treats the PDS InvalidRequest repo-absence form as an authoritative empty snapshot', async () => {
		reset()
		knownDids = [DID]
		const result = await runStartupBackfill({
			fetcher: async (url) => {
				if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
				return new Response(JSON.stringify({ error: 'InvalidRequest', message: `Could not find repo: ${DID}` }), {
					status: 400,
				})
			},
		})

		expect(result).toEqual({ found: 0, failed: 0 })
		expect(completed).toEqual([{ ownerDid: DID, generation: 1 }])
		expect(failed).toEqual([])
	})

	test('does not prune a partial snapshot when repo absence arrives after a page', async () => {
		reset()
		knownDids = [DID]
		const result = await runStartupBackfill({
			fetcher: async (url) => {
				if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
				if (!new URL(url).searchParams.has('cursor')) {
					return new Response(
						JSON.stringify({
							records: [{ uri: recordUri(DID, 'one'), cid: 'a', value: validRecord }],
							cursor: 'next',
						}),
					)
				}
				return new Response(JSON.stringify({ error: 'RepoNotFound', message: 'Repo not found' }), { status: 400 })
			},
		})

		expect(result).toEqual({ found: 0, failed: 1 })
		expect(appliedPages.map((page) => page.records.map((record) => record.rkey))).toEqual([['one']])
		expect(completed).toEqual([])
		expect(failed).toEqual([{ ownerDid: DID, generation: 1 }])
	})

	test('does not treat an empty cursor page followed by repo absence as complete', async () => {
		reset()
		knownDids = [DID]
		const result = await runStartupBackfill({
			fetcher: async (url) => {
				if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
				if (!new URL(url).searchParams.has('cursor')) {
					return new Response(JSON.stringify({ records: [], cursor: 'next' }))
				}
				return new Response(JSON.stringify({ error: 'RepoNotFound', message: 'Repo not found' }), { status: 400 })
			},
		})

		expect(result).toEqual({ found: 0, failed: 1 })
		expect(appliedPages).toEqual([{ ownerDid: DID, records: [] }])
		expect(completed).toEqual([])
		expect(failed).toEqual([{ ownerDid: DID, generation: 1 }])
	})

	test('keeps unknown client errors and gateway failures retryable', async () => {
		for (const response of [
			new Response(JSON.stringify({ error: 'InvalidRequest', message: 'bad request' }), { status: 400 }),
			new Response(JSON.stringify({ error: 'BadGateway' }), { status: 502 }),
		]) {
			reset()
			knownDids = [DID]
			const result = await runStartupBackfill({
				fetcher: async (url) =>
					url === 'https://example.com/.well-known/did.json' ? didDocument('https://pds.example') : response,
			})
			expect(result).toEqual({ found: 0, failed: 1 })
			expect(failed).toEqual([{ ownerDid: DID, generation: 1 }])
			expect(completed).toEqual([])
		}
	})

	test('omits unsafe endpoint rows while completing the authoritative owner snapshot', async () => {
		reset()
		const unsafeUrls = [
			'https://127.0.0.1/internal',
			'https://user:password@receiver.example/hook',
			'https://receiver.example/hook#fragment',
			'http://receiver.example/hook',
		]
		for (const url of unsafeUrls) {
			const records = await fetchWhRecordsForDid(
				DID,
				pdsFetcher(
					new Response(
						JSON.stringify({ records: [{ uri: recordUri(DID, 'unsafe'), cid: 'a', value: { ...validRecord, url } }] }),
					),
				),
			)
			expect(records).toEqual([])
		}

		knownDids = [DID]
		const result = await runStartupBackfill(async (url) => {
			if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
			return new Response(
				JSON.stringify({
					records: [
						{ uri: recordUri(DID, 'unsafe'), cid: 'a', value: { ...validRecord, url: 'https://127.0.0.1/internal' } },
					],
				}),
			)
		})
		expect(result).toEqual({ found: 0, failed: 0 })
		expect(appliedPages).toEqual([{ ownerDid: DID, records: [] }])
		expect(completed).toEqual([{ ownerDid: DID, generation: 1 }])
	})

	test('contains an oversized PDS response and marks only that owner degraded', async () => {
		reset()
		knownDids = [DID]
		let calls = 0
		const result = await runStartupBackfill(
			async (url) => {
				calls++
				if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
				return new Response('x'.repeat(MAX_IDENTITY_JSON_BYTES + 1))
			},
			(did, status) => {
				transitions.push({ did, status })
			},
		)

		expect(result).toEqual({ found: 0, failed: 1 })
		expect(calls).toBe(2)
		expect(appliedPages).toEqual([])
		expect(completed).toEqual([])
		expect(failed).toEqual([{ ownerDid: DID, generation: 1 }])
		expect(transitions).toEqual([
			{ did: DID, status: 'scanning' },
			{ did: DID, status: 'failed' },
		])
	})

	test('continues reconciling healthy owners after another owner fails', async () => {
		reset()
		knownDids = [DID, OTHER_DID]
		const result = await runStartupBackfill({
			fetcher: async (url) => {
				if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
				if (url === 'https://other.example/.well-known/did.json') return didDocument('https://pds.example')
				const repo = new URL(url).searchParams.get('repo')
				if (repo === OTHER_DID) return new Response(null, { status: 503 })
				return new Response(JSON.stringify({ records: [{ uri: recordUri(DID, 'one'), cid: 'a', value: validRecord }] }))
			},
			onOwnerTransition: (did, status) => {
				transitions.push({ did, status })
			},
		})

		expect(result).toEqual({ found: 1, failed: 1 })
		expect(appliedPages.map((page) => page.ownerDid)).toEqual([DID])
		expect(completed).toEqual([{ ownerDid: DID, generation: 1 }])
		expect(failed).toEqual([{ ownerDid: OTHER_DID, generation: 2 }])
		expect(transitions).toEqual([
			{ did: DID, status: 'scanning' },
			{ did: DID, status: 'complete' },
			{ did: OTHER_DID, status: 'scanning' },
			{ did: OTHER_DID, status: 'failed' },
		])
	})

	test('uses keyset pages, a fixed queue, and a continuation cursor instead of materializing all owners', async () => {
		reset()
		const thirdDid = 'did:web:third.example'
		const owners = [DID, OTHER_DID, thirdDid]
		const requests: Array<{ after: string | undefined; limit: number }> = []
		const listOwnerPage = async (after: string | undefined, limit: number) => {
			requests.push({ after, limit })
			return owners.filter((did) => after === undefined || did > after).slice(0, limit)
		}
		const fetcher = async (url: string) => {
			if (url.includes('/.well-known/did.json')) return didDocument('https://pds.example')
			return new Response(JSON.stringify({ records: [] }))
		}

		const first = await runStartupBackfill({
			fetcher,
			listOwnerPage,
			maxOwnersPerPass: 2,
			ownerPageSize: 1,
		})
		expect(first).toEqual({ found: 0, failed: 0, nextOwnerCursor: OTHER_DID })
		expect(requests).toEqual([
			{ after: undefined, limit: 1 },
			{ after: DID, limit: 1 },
		])
		// The source never receives a larger request than the fixed queue page.
		expect(requests.every((request) => request.limit === 1)).toBe(true)

		const continuation = await runStartupBackfill({
			fetcher,
			listOwnerPage,
			maxOwnersPerPass: 2,
			ownerPageSize: 1,
			startAfter: first.nextOwnerCursor,
		})
		expect(continuation).toEqual({ found: 0, failed: 0 })
		expect(began.map((token) => token.ownerDid)).toEqual([DID, OTHER_DID, thirdDid])
	})

	test('continues a >1000-owner bounded pass without materializing the owner set', async () => {
		reset()
		const owners = Array.from({ length: 1_001 }, (_, index) => `did:web:owner${String(index).padStart(4, '0')}.example`)
		const requestedLimits: number[] = []
		const listOwnerPage = async (after: string | undefined, limit: number) => {
			requestedLimits.push(limit)
			return owners.filter((did) => after === undefined || did > after).slice(0, limit)
		}
		const fetcher = async (url: string) =>
			url.includes('/.well-known/did.json')
				? didDocument('https://pds.example')
				: new Response(JSON.stringify({ records: [] }))

		const first = await runStartupBackfill({ fetcher, listOwnerPage })
		expect(first).toEqual({ found: 0, failed: 0, nextOwnerCursor: owners[999] })
		expect(Math.max(...requestedLimits)).toBe(50)
		expect(began).toHaveLength(1_000)

		const resumed = await runStartupBackfill({ fetcher, listOwnerPage, startAfter: first.nextOwnerCursor })
		expect(resumed).toEqual({ found: 0, failed: 0 })
		expect(began).toHaveLength(1_001)
		expect(began[began.length - 1]?.ownerDid).toBe(owners[1_000])
	})

	test('treats reconciliation infrastructure failures as fatal rather than degrading every owner', async () => {
		reset()
		knownDids = [DID]
		beginFailure = true
		await expect(runStartupBackfill(async () => new Response('{}'))).rejects.toThrow(
			'Webhook reconciliation infrastructure failed',
		)
		expect(failed).toEqual([])
	})
})

interface FakeTimer {
	callback: () => void
	delayMs: number
	cleared: boolean
	fired: boolean
}

function createFakeTimers(): {
	timers: FakeTimer[]
	clearTimer: (timer: ReturnType<typeof setTimeout>) => void
	setTimer: (callback: () => void, delayMs: number) => ReturnType<typeof setTimeout>
	fire: (delayMs: number) => void
} {
	const timers: FakeTimer[] = []
	return {
		timers,
		setTimer: (callback, delayMs) => {
			const timer = { callback, delayMs, cleared: false, fired: false }
			timers.push(timer)
			return timer as unknown as ReturnType<typeof setTimeout>
		},
		clearTimer: (timer) => {
			;(timer as unknown as FakeTimer).cleared = true
		},
		fire: (delayMs) => {
			const timer = timers.find((entry) => entry.delayMs === delayMs && !entry.cleared && !entry.fired)
			if (!timer) throw new Error(`No pending ${delayMs}ms timer`)
			timer.fired = true
			timer.callback()
		},
	}
}

function deferred<T>(): { promise: Promise<T>; resolve: (value: T) => void } {
	let resolve: ((value: T) => void) | undefined
	const promise = new Promise<T>((done) => {
		resolve = done
	})
	return { promise, resolve: (value) => resolve?.(value) }
}

describe('failed owner reconciliation retry scheduler', () => {
	test('recovers a transiently failed owner through the injected transition callback', async () => {
		reset()
		const timers = createFakeTimers()
		const retryTransitions: Array<{ did: string; status: string }> = []
		const completedRetry = deferred<void>()
		const scheduler = createReconciliationRetryScheduler({
			baseDelayMs: 1_000,
			clearTimer: timers.clearTimer,
			discoveryIntervalMs: 10_000,
			listFailedOwners: async () => [DID],
			onOwnerTransition: (did, status) => {
				retryTransitions.push({ did, status })
				if (status === 'complete') completedRetry.resolve()
			},
			random: () => 0.5,
			setTimer: timers.setTimer,
			fetcher: async (url) => {
				if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
				return new Response(
					JSON.stringify({ records: [{ uri: recordUri(DID, 'retry'), cid: 'a', value: validRecord }] }),
				)
			},
		})

		await scheduler.start()
		expect(scheduler.health.scheduled).toBe(1)
		timers.fire(500)
		await completedRetry.promise
		expect(appliedPages.map((page) => page.records.map((record) => record.rkey))).toEqual([['retry']])
		expect(retryTransitions).toEqual([
			{ did: DID, status: 'scanning' },
			{ did: DID, status: 'complete' },
		])
		expect(scheduler.health.infrastructureHealthy).toBe(true)
		await scheduler.stop()
	})

	test('uses capped exponential full-jitter retries and limits tracked failed owners', async () => {
		reset()
		const timers = createFakeTimers()
		const failedRetry = deferred<void>()
		const retryTimerCreated = deferred<void>()
		const scheduler = createReconciliationRetryScheduler({
			baseDelayMs: 1_000,
			clearTimer: timers.clearTimer,
			discoveryIntervalMs: 10_000,
			listFailedOwners: async () => [DID, OTHER_DID],
			maxOwners: 1,
			onOwnerTransition: (_did, status) => {
				if (status === 'failed') failedRetry.resolve()
			},
			random: () => 0.5,
			setTimer: (callback, delayMs) => {
				if (delayMs === 1_000) retryTimerCreated.resolve()
				return timers.setTimer(callback, delayMs)
			},
			fetcher: async (url) => {
				if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
				return new Response(null, { status: 503 })
			},
		})

		await scheduler.start()
		expect(scheduler.health.scheduled).toBe(1)
		timers.fire(500)
		await failedRetry.promise
		await retryTimerCreated.promise
		expect(scheduler.health.scheduled).toBe(1)
		expect(timers.timers.some((timer) => timer.delayMs === 1_000 && !timer.fired && !timer.cleared)).toBe(true)
		await scheduler.stop()
	})

	test('cancels pending timers and drains an in-flight owner scan on stop', async () => {
		reset()
		const timers = createFakeTimers()
		let resolvePage: ((response: Response) => void) | undefined
		let requestSignal: AbortSignal | undefined
		const pageRequested = deferred<void>()
		const ownerFinished = deferred<void>()
		const scheduler = createReconciliationRetryScheduler({
			baseDelayMs: 1_000,
			clearTimer: timers.clearTimer,
			discoveryIntervalMs: 10_000,
			listFailedOwners: async () => [DID],
			onOwnerTransition: (_did, status) => {
				if (status === 'failed') ownerFinished.resolve()
			},
			random: () => 0.5,
			setTimer: timers.setTimer,
			fetcher: async (url, options) => {
				if (url === 'https://example.com/.well-known/did.json') return didDocument('https://pds.example')
				requestSignal = options?.signal
				return new Promise<Response>((resolve) => {
					resolvePage = resolve
					pageRequested.resolve()
				})
			},
		})

		await scheduler.start()
		timers.fire(500)
		await pageRequested.promise
		expect(scheduler.health.active).toBe(1)
		const shutdownDeadline = new AbortController()
		const stopped = scheduler.stop({ signal: shutdownDeadline.signal })
		expect(requestSignal?.aborted).toBe(true)
		shutdownDeadline.abort()
		expect(await stopped).toEqual({ drained: false })
		resolvePage?.(new Response(null, { status: 503 }))
		await ownerFinished.promise
		// The first bounded stop returned at the global deadline. A subsequent
		// local stop observes the already-unwinding worker and waits for cleanup.
		expect(await scheduler.stop()).toEqual({ drained: true })
		expect(scheduler.health).toMatchObject({ active: 0, queued: 0, running: false, scheduled: 0 })
	})
})