import { describe, expect, test } from 'bun:test'
import {
	DurableReplayBackoff,
	DurableReplayController,
	destroyThenConnect,
	durableReplayBackoffCap,
	durableReplayFullJitter,
	OrderedCursorTracker,
	RelayCursorCoordinator,
	RelayFailureBudget,
	RelayGenerationGuard,
	retryDurableRevalidationUntilAvailable,
	runWithFirehoseResourceContext,
	SiteWorkScheduler,
} from './firehose'
import { createRevalidationResourceContext } from './revalidate-resources'

function deferred<T = void>() {
	let resolve!: (value: T | PromiseLike<T>) => void
	const promise = new Promise<T>((resolvePromise) => {
		resolve = resolvePromise
	})
	return { promise, resolve }
}

async function flushMicrotasks(): Promise<void> {
	for (let index = 0; index < 8; index++) await Promise.resolve()
}

describe('normal firehose resource context', () => {
	test('closes the context after scheduled work, including a failed operation', async () => {
		const context = createRevalidationResourceContext(25, 8)
		let closeCalls = 0
		const resources = {
			...context,
			close: () => {
				closeCalls++
				context.close()
			},
		}

		await expect(
			runWithFirehoseResourceContext(
				() => resources,
				async (scheduledResources) => {
					scheduledResources.transferBudget.consume(3)
					throw new Error('scheduled operation failed')
				},
			),
		).rejects.toThrow('scheduled operation failed')
		expect(closeCalls).toBe(1)
		expect(context.transferBudget.consumedBytes).toBe(3)

		await new Promise((resolve) => setTimeout(resolve, 35))
		expect(context.signal.aborted).toBe(false)
	})
})

describe('durable replay backoff', () => {
	test('increases the exponential cap and stops growing at the bound', () => {
		const backoff = new DurableReplayBackoff({
			baseDelayMs: 10,
			maxDelayMs: 25,
			maxConsecutiveFailures: 6,
			random: () => 0.5,
		})

		expect([1, 2, 3, 4].map((attempt) => durableReplayBackoffCap(attempt, 10, 25))).toEqual([10, 20, 25, 25])
		expect([backoff.recordFailure(), backoff.recordFailure(), backoff.recordFailure()]).toMatchObject([
			{ consecutiveFailures: 1, delayCapMs: 10, delayMs: 5 },
			{ consecutiveFailures: 2, delayCapMs: 20, delayMs: 10 },
			{ consecutiveFailures: 3, delayCapMs: 25, delayMs: 13 },
		])
		expect(durableReplayFullJitter(25, () => 0)).toBe(0)
		expect(durableReplayFullJitter(25, () => 1)).toBe(25)
	})

	test('decays a stale failure burst after healthy time', () => {
		let now = 0
		const backoff = new DurableReplayBackoff({
			baseDelayMs: 10,
			failureDecayMs: 100,
			maxConsecutiveFailures: 3,
			now: () => now,
			random: () => 0,
		})

		expect(backoff.recordFailure().consecutiveFailures).toBe(1)
		now = 50
		expect(backoff.recordFailure().consecutiveFailures).toBe(2)
		now = 151
		// One healthy interval makes the next failure a fresh attempt rather
		// than inheriting a stale near-terminal budget.
		expect(backoff.consecutiveFailures).toBe(0)
		expect(backoff.recordFailure()).toMatchObject({ consecutiveFailures: 1, terminal: false })
	})

	test('permits a new replay probe after a terminal budget cools down', async () => {
		let now = 0
		const controller = new DurableReplayController({
			backoff: new DurableReplayBackoff({
				baseDelayMs: 1,
				failureDecayMs: 100,
				maxConsecutiveFailures: 2,
				now: () => now,
				random: () => 0,
			}),
		})
		let reconnects = 0
		const reconnect = async () => {
			reconnects++
		}
		let terminals = 0

		controller.request(reconnect, () => terminals++)
		await controller.pending
		controller.request(reconnect, () => terminals++)
		expect(terminals).toBe(1)
		expect(reconnects).toBe(1)

		now = 101
		const probe = controller.request(reconnect, () => terminals++)
		expect(probe).toMatchObject({ consecutiveFailures: 1, terminal: false })
		await controller.pending
		expect(reconnects).toBe(2)
		expect(terminals).toBe(1)
	})

	test('keeps a capacity failure pending until the revalidation consumer frees space', async () => {
		const outcomes = ['capacity', 'capacity', 'enqueued'] as const
		const delays: number[] = []
		let calls = 0
		const result = await retryDurableRevalidationUntilAvailable(async () => outcomes[calls++] ?? 'unavailable', {
			wait: async (delayMs) => {
				delays.push(delayMs)
			},
			baseDelayMs: 10,
			maxDelayMs: 20,
		})

		expect(result).toBe('enqueued')
		expect(calls).toBe(3)
		expect(delays).toEqual([10, 20])
	})

	test('does not advance the cursor while capacity blocks a durable handoff', async () => {
		const tracker = new OrderedCursorTracker(1)
		const reservation = await tracker.reserve(101)
		expect(reservation).toBeDefined()
		let releaseCapacityWait!: () => void
		let calls = 0
		const handoff = retryDurableRevalidationUntilAvailable(async () => (calls++ === 0 ? 'capacity' : 'enqueued'), {
			wait: async () => await new Promise<void>((resolve) => (releaseCapacityWait = resolve)),
		})
		await flushMicrotasks()

		expect(calls).toBe(1)
		expect(tracker.cursor).toBeUndefined()
		expect(tracker.pendingCount).toBe(1)
		releaseCapacityWait()
		expect(await handoff).toBe('enqueued')
		reservation?.complete()
		tracker.sealOpenSequence()
		expect(tracker.cursor).toBe(101)
	})

	test('cancels a capacity wait when intake stops without another enqueue', async () => {
		let accepting = true
		let calls = 0
		const result = await retryDurableRevalidationUntilAvailable(
			async () => {
				calls++
				return 'capacity'
			},
			{
				shouldContinue: () => accepting,
				wait: async () => {
					accepting = false
				},
			},
		)

		expect(result).toBe('unavailable')
		expect(calls).toBe(1)
	})

	test('aborts a capacity sleep promptly without completing its cursor reservation', async () => {
		const tracker = new OrderedCursorTracker(1)
		const reservation = await tracker.reserve(202)
		const controller = new AbortController()
		const waitStarted = deferred<void>()
		const wait = new Promise<void>(() => {})
		let calls = 0
		const retry = retryDurableRevalidationUntilAvailable(
			async () => {
				calls++
				return 'capacity'
			},
			{
				signal: controller.signal,
				wait: async () => {
					waitStarted.resolve()
					await wait
				},
			},
		)

		await waitStarted.promise
		controller.abort()
		expect(await retry).toBe('unavailable')
		expect(calls).toBe(1)
		expect(tracker.cursor).toBeUndefined()
		expect(tracker.pendingCount).toBe(1)
		// The event owner, not the retry loop, is responsible for deciding whether
		// a stopped reservation should be replayed.
		reservation?.fail()
	})

	test('allows only one replay reconnect at a time', async () => {
		const gate = deferred()
		const controller = new DurableReplayController({
			backoff: new DurableReplayBackoff({ baseDelayMs: 1, random: () => 0 }),
		})
		let reconnects = 0
		const reconnect = async () => {
			reconnects++
			await gate.promise
		}
		const terminal = () => {
			throw new Error('unexpected terminal replay')
		}

		controller.request(reconnect, terminal)
		controller.request(reconnect, terminal)
		await flushMicrotasks()
		expect(reconnects).toBe(1)

		gate.resolve()
		await controller.pending
		await flushMicrotasks()
		expect(reconnects).toBe(1)
	})

	test('resets only when durable recovery succeeds', async () => {
		const controller = new DurableReplayController({
			backoff: new DurableReplayBackoff({ baseDelayMs: 1, random: () => 0 }),
		})
		const reconnect = async () => {}
		const terminal = () => {
			throw new Error('unexpected terminal replay')
		}

		controller.request(reconnect, terminal)
		await controller.pending
		expect(controller.consecutiveFailures).toBe(1)
		controller.recordDurableSuccess()
		expect(controller.consecutiveFailures).toBe(0)
		expect(controller.request(reconnect, terminal)?.consecutiveFailures).toBe(1)
		await controller.pending
	})

	test('calls the terminal callback after the bounded failure count', async () => {
		const controller = new DurableReplayController({
			backoff: new DurableReplayBackoff({ baseDelayMs: 1, maxConsecutiveFailures: 2, random: () => 0 }),
		})
		let reconnects = 0
		let terminals = 0
		const reconnect = async () => {
			reconnects++
		}
		const terminal = () => {
			terminals++
		}

		controller.request(reconnect, terminal)
		await controller.pending
		controller.request(reconnect, terminal)
		controller.request(reconnect, terminal)
		expect(reconnects).toBe(1)
		expect(terminals).toBe(1)
	})

	test('cancels a delayed replay without reconnecting after shutdown', async () => {
		const timers: Array<() => void> = []
		let cleared = 0
		const controller = new DurableReplayController({
			backoff: new DurableReplayBackoff({ baseDelayMs: 50, random: () => 1 }),
			setTimeout: (callback) => {
				timers.push(callback)
				return timers.length as unknown as ReturnType<typeof setTimeout>
			},
			clearTimeout: () => {
				cleared++
			},
		})
		let reconnects = 0
		const pending = controller.request(
			async () => {
				reconnects++
			},
			() => {},
		)
		const replay = controller.pending
		controller.stop()
		await replay
		timers[0]?.()
		await flushMicrotasks()

		expect(pending?.delayMs).toBe(50)
		expect(reconnects).toBe(0)
		expect(cleared).toBe(1)
	})
})

describe('OrderedCursorTracker', () => {
	test('does not advance past an earlier slow sequence when later work finishes first', async () => {
		const tracker = new OrderedCursorTracker(10)
		const slow = await tracker.reserve(41)
		const fast = await tracker.reserve(42)

		expect(slow).toBeDefined()
		expect(fast).toBeDefined()
		fast?.complete()
		expect(tracker.cursor).toBeUndefined()

		slow?.complete()
		// 42 is the open sequence. Its completion is held until another relay
		// sequence arrives or source shutdown seals it.
		expect(tracker.cursor).toBe(41)
		tracker.sealOpenSequence()
		expect(tracker.cursor).toBe(42)
	})

	test('uses only the completed prefix as the crash/resume cursor', async () => {
		const tracker = new OrderedCursorTracker(10, 500)
		const completed = await tracker.reserve(501)
		const failed = await tracker.reserve(502)

		completed?.complete()
		failed?.fail()
		tracker.sealOpenSequence()

		expect(tracker.cursor).toBe(501)
		expect(tracker.resumableCursor).toBe(501)
		expect(tracker.pendingCount).toBe(1)

		// A reconnect discards the failed in-memory reservation and replays it
		// from the durable prefix instead of skipping sequence 502.
		const resumed = new OrderedCursorTracker(10, tracker.resumableCursor)
		const replayed = await resumed.reserve(502)
		replayed?.complete()
		resumed.sealOpenSequence()
		expect(resumed.cursor).toBe(502)
	})

	test('shares a cap-one slot across one commit and seals it before admitting the next sequence', async () => {
		const tracker = new OrderedCursorTracker(1)
		const firstOperation = await tracker.reserve(100)
		const secondOperation = await tracker.reserve(100)

		expect(tracker.pendingCount).toBe(1)
		firstOperation?.complete()
		secondOperation?.complete()

		// The incoming sequence seals completed sequence 100 before checking the
		// cap, so a cap of one cannot deadlock a multi-operation commit.
		const next = await tracker.reserve(101)
		expect(next?.seq).toBe(101)
		expect(tracker.cursor).toBe(100)
	})

	test('applies bounded backpressure until a sealed slow prefix frees capacity', async () => {
		const tracker = new OrderedCursorTracker(1)
		const first = await tracker.reserve(200)
		let secondResolved = false
		const secondPromise = tracker.reserve(201).then((reservation) => {
			secondResolved = true
			return reservation
		})

		await flushMicrotasks()
		expect(secondResolved).toBe(false)

		first?.complete()
		const second = await secondPromise
		expect(second?.seq).toBe(201)
		expect(tracker.cursor).toBe(200)
	})

	test('rejects malformed, duplicate, and rewound sequences without regressing the cursor', async () => {
		const tracker = new OrderedCursorTracker(2, 50)
		expect(await tracker.reserve(-1)).toBeUndefined()
		expect(await tracker.reserve(Number.MAX_SAFE_INTEGER + 1)).toBeUndefined()
		expect(await tracker.reserve(50)).toBeUndefined()
		expect(await tracker.reserve(49)).toBeUndefined()

		const next = await tracker.reserve(51)
		next?.complete()
		tracker.sealOpenSequence()
		expect(tracker.cursor).toBe(51)
		expect(await tracker.reserve(50)).toBeUndefined()
		expect(tracker.cursor).toBe(51)
	})
})

describe('relay lifecycle guards', () => {
	test('does not reset the cross-relay budget for connect-error loops without an event', () => {
		const budget = new RelayFailureBudget()

		for (let relay = 0; relay < 2; relay++) {
			budget.recordConnected()
			for (let error = 0; error < 3; error++) budget.recordConnectionError()
			expect(budget.canFailOver()).toBe(true)
			budget.recordFailOver()
		}

		budget.recordConnected()
		for (let error = 0; error < 3; error++) budget.recordConnectionError()
		expect(budget.canFailOver()).toBe(false)

		budget.recordEvent()
		expect(budget.canFailOver()).toBe(true)
		expect(budget.consecutiveFailures).toBe(0)
	})

	test('serializes delayed failover/replay destruction before replacing a relay', async () => {
		const generations = new RelayGenerationGuard()
		const oldRelay = generations.beginConnection()
		const destroy = deferred()
		let replacement: number | undefined
		const transition = destroyThenConnect(
			async () => {
				generations.invalidate()
				await destroy.promise
			},
			() => true,
			() => {
				replacement = generations.beginConnection()
			},
		)

		expect(generations.isCurrent(oldRelay)).toBe(false)
		expect(replacement).toBeUndefined()
		destroy.resolve()
		await transition
		expect(replacement).toBeDefined()
		expect(generations.isCurrent(replacement ?? -1)).toBe(true)
		expect(generations.isCurrent(oldRelay)).toBe(false)
	})

	test('does not open a replacement when old relay destruction fails', async () => {
		let connections = 0
		await expect(
			destroyThenConnect(
				async () => {
					throw new Error('destroy failed')
				},
				() => true,
				() => {
					connections++
				},
			),
		).rejects.toThrow('destroy failed')
		expect(connections).toBe(0)
	})

	test('does not reconnect after a delayed shutdown destroy', async () => {
		const destroy = deferred()
		let connections = 0
		const shutdown = destroyThenConnect(
			async () => await destroy.promise,
			() => false,
			() => {
				connections++
			},
		)

		destroy.resolve()
		await shutdown
		expect(connections).toBe(0)
	})
})

describe('SiteWorkScheduler', () => {
	test('keeps each site ordered while allowing a different site to run concurrently and drain', async () => {
		const scheduler = new SiteWorkScheduler(2)
		const firstGate = deferred()
		const otherGate = deferred()
		const order: string[] = []

		const first = scheduler.schedule('did:example/a', async () => {
			order.push('a1:start')
			await firstGate.promise
			order.push('a1:end')
		})
		const second = scheduler.schedule('did:example/a', async () => {
			order.push('a2:start')
		})
		const other = scheduler.schedule('did:example/b', async () => {
			order.push('b1:start')
			await otherGate.promise
			order.push('b1:end')
		})

		await flushMicrotasks()
		expect(order).toEqual(['a1:start', 'b1:start'])
		expect(scheduler.activeHandlers).toBe(2)
		expect(scheduler.queuedHandlers).toBe(3)

		const draining = scheduler.drain(1_000)
		firstGate.resolve()
		await first
		await flushMicrotasks()
		expect(order).toEqual(['a1:start', 'b1:start', 'a1:end', 'a2:start'])

		otherGate.resolve()
		await Promise.all([second, other])
		await expect(draining).resolves.toMatchObject({
			outcome: 'drained',
			forced: false,
			remainingWork: 0,
			activeHandlers: 0,
		})
	})

	test('rejects failed work while preserving order for a later event on the same site', async () => {
		const scheduler = new SiteWorkScheduler(1)
		const order: string[] = []
		const failed = scheduler.schedule('did:example/a', async () => {
			order.push('first')
			throw new Error('transient write failure')
		})
		const later = scheduler.schedule('did:example/a', async () => {
			order.push('second')
		})

		await expect(failed).rejects.toThrow('transient write failure')
		await expect(later).resolves.toBeUndefined()
		expect(order).toEqual(['first', 'second'])
	})

	test('reports an explicit forced outcome without waiting for an uncooperative handler', async () => {
		const scheduler = new SiteWorkScheduler(1)
		const gate = deferred()
		const work = scheduler.schedule('did:example/a', async () => {
			await gate.promise
		})

		await flushMicrotasks()
		await expect(scheduler.drain(0)).resolves.toMatchObject({
			outcome: 'forced',
			forced: true,
			remainingWork: 1,
			activeHandlers: 1,
		})

		gate.resolve()
		await work
		await expect(scheduler.drain(0)).resolves.toMatchObject({
			outcome: 'drained',
			forced: false,
		})
	})
})

describe('relay-scoped durable cursors', () => {
	test('persists A before B, loads B, and reloads A on return', async () => {
		const durable = new Map<string, number>([
			['relay-a', 41],
			['relay-b', 87],
		])
		const saves: Array<[string, number]> = []
		const store = {
			read: async (service: string) => {
				const cursor = durable.get(service)
				return cursor === undefined ? { kind: 'missing' as const } : { kind: 'found' as const, cursor }
			},
			save: async (service: string, cursor: number) => {
				saves.push([service, cursor])
				durable.set(service, cursor)
				return true
			},
		}
		const cursors = new RelayCursorCoordinator((service) => service)
		expect(cursors.initialize('relay-a', 50)).toBe(50)

		const switchedToB = await cursors.switchTo('relay-b', 52, store)
		expect(switchedToB).toEqual({ cursor: 87, missingCheckpoint: false })
		expect(saves).toEqual([['relay-a', 52]])

		cursors.recordActiveCursor(91)
		const switchedBackToA = await cursors.switchTo('relay-a', 91, store)
		expect(switchedBackToA).toEqual({ cursor: 52, missingCheckpoint: false })
		expect(saves).toEqual([
			['relay-a', 52],
			['relay-b', 91],
		])
		expect(cursors.knownCursor('relay-a')).toBe(52)
		expect(cursors.knownCursor('relay-b')).toBe(91)
	})

	test('starts live when durable storage confirms a target cursor is missing', async () => {
		const cursors = new RelayCursorCoordinator((service) => service)
		const saves: Array<[string, number]> = []
		expect(cursors.initialize('relay-a', undefined)).toBeUndefined()
		const activation = await cursors.switchTo('relay-b', undefined, {
			read: async () => ({ kind: 'missing' }),
			save: async (service, cursor) => {
				saves.push([service, cursor])
				return true
			},
		})
		expect(saves).toEqual([])
		expect(activation).toEqual({ cursor: undefined, missingCheckpoint: true })
	})

	test('refuses a target transition when durable state is unavailable', async () => {
		const cursors = new RelayCursorCoordinator((service) => service)
		cursors.initialize('relay-a', 10)
		const activation = await cursors.switchTo('relay-b', 11, {
			read: async () => ({ kind: 'unavailable' }),
			save: async () => true,
		})
		expect(activation).toBeUndefined()
	})
})