import { afterEach, describe, expect, test } from 'bun:test'
import { MAX_JETSTREAM_SUBSCRIPTION_URL_BYTES } from './admission'
import { buildJetstreamSubscriptionUrl, JetstreamClient, parseJetstreamEvent } from './jetstream'

const DID = 'did:plc:aaaaaaaaaaaaaaaaaaaaaaaa'
const REV = '3lq6x5f2abcde'

class FakeWebSocket {
	static instances: FakeWebSocket[] = []
	readonly url: string
	onopen: (() => void) | null = null
	onclose: (() => void) | null = null
	onerror: (() => void) | null = null
	onmessage: ((message: { data: unknown }) => void) | null = null
	closeImmediately = true
	closeRequested = false
	#closed = false

	constructor(url: string) {
		this.url = url
		FakeWebSocket.instances.push(this)
	}

	open(): void {
		this.onopen?.()
	}

	message(value: unknown): void {
		this.onmessage?.({ data: value })
	}

	close(): void {
		this.closeRequested = true
		if (this.closeImmediately) this.finishClose()
	}

	finishClose(): void {
		if (this.#closed) return
		this.#closed = true
		this.onclose?.()
	}
}

const realWebSocket = globalThis.WebSocket

function event(timeUs: number) {
	return JSON.stringify({
		did: DID,
		time_us: timeUs,
		kind: 'commit',
		commit: { rev: REV, operation: 'create', collection: 'app.bsky.feed.post', rkey: 'one', record: {} },
	})
}

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

afterEach(() => {
	globalThis.WebSocket = realWebSocket
	FakeWebSocket.instances = []
})

describe('JetstreamClient durable batch acknowledgement and protocol quarantine', () => {
	test('does not advance the cursor until the batch commits', async () => {
		globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket
		let release: (() => void) | undefined
		const committed = new Promise<void>((resolve) => {
			release = resolve
		})
		const batches: number[][] = []
		const client = new JetstreamClient({
			url: 'wss://relay.example/subscribe',
			cursor: 10,
			onEvents: (events) => {
				batches.push(events.map((value) => value.time_us))
				return committed
			},
		})
		client.start()
		const socket = FakeWebSocket.instances[0]
		expect(socket).toBeDefined()
		socket?.open()
		socket?.message(event(11))
		await flush()
		expect(client.cursor).toBe(10)
		release?.()
		await flush()
		expect(batches).toEqual([[11]])
		expect(client.cursor).toBe(11)
		client.destroy()
	})

	test('hands the queued events to one batch and advances to its last event', async () => {
		globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket
		const batches: number[][] = []
		let release: (() => void) | undefined
		const firstBatch = new Promise<void>((resolve) => {
			release = resolve
		})
		const client = new JetstreamClient({
			url: 'wss://relay.example/subscribe',
			cursor: 50,
			onEvents: (events) => {
				batches.push(events.map((value) => value.time_us))
				return batches.length === 1 ? firstBatch : Promise.resolve()
			},
		})
		client.start()
		const socket = FakeWebSocket.instances[0]
		expect(socket).toBeDefined()
		socket?.open()
		socket?.message(event(51))
		await flush()
		// These arrive while the first batch is still committing.
		socket?.message(event(52))
		socket?.message(event(53))
		await flush()
		expect(batches).toEqual([[51]])
		release?.()
		await flush()
		// One transaction, one cursor write, both remaining events.
		expect(batches).toEqual([[51], [52, 53]])
		expect(client.cursor).toBe(53)
		expect(client.queued).toBe(0)
		client.destroy()
	})

	test('never hands over more than the batch bound at once', async () => {
		globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket
		const sizes: number[] = []
		const client = new JetstreamClient({
			url: 'wss://relay.example/subscribe',
			cursor: 80,
			batchMax: 2,
			maxQueue: 8,
			onEvents: async (events) => {
				sizes.push(events.length)
			},
		})
		client.start()
		const socket = FakeWebSocket.instances[0]
		expect(socket).toBeDefined()
		socket?.open()
		for (const time of [81, 82, 83, 84, 85]) socket?.message(event(time))
		await flush()
		expect(Math.max(...sizes)).toBeLessThanOrEqual(2)
		expect(client.cursor).toBe(85)
		client.destroy()
	})

	test('stays healthy while a bounded queue drains, and stops when progress stalls', async () => {
		globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket
		let release: (() => void) | undefined
		const held = new Promise<void>((resolve) => {
			release = resolve
		})
		const client = new JetstreamClient({
			url: 'wss://relay.example/subscribe',
			cursor: 60,
			maxQueue: 2,
			batchMax: 1,
			onEvents: () => held,
			progressStaleMs: 60_000,
		})
		client.start()
		const socket = FakeWebSocket.instances[0]
		if (!socket) throw new Error('missing test socket')
		socket.closeImmediately = false
		socket.open()
		socket.message(event(61))
		socket.message(event(62))
		await flush()
		// Backpressure closed the socket; the retained queue is still real work.
		expect(client.isConnected).toBe(false)
		expect(client.failureKind).toBe('queue')
		release?.()
		await flush()
		expect(client.cursor).toBe(62)
		expect(client.isProgressing).toBe(true)

		const stalled = new JetstreamClient({
			url: 'wss://relay.example/subscribe',
			onEvents: async () => undefined,
			progressStaleMs: 1,
		})
		// Never acknowledged anything and not connected: not serving.
		expect(stalled.isProgressing).toBe(false)
		client.destroy()
		stalled.destroy()
	})

	test('retries a failed batch in place instead of replaying the stream', async () => {
		globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket
		let failures = 0
		const client = new JetstreamClient({
			url: 'wss://relay.example/subscribe',
			cursor: 70,
			batchRetryDelayMs: 1,
			onEvents: async () => {
				if (failures === 0) {
					failures++
					throw new Error('transient')
				}
			},
		})
		client.start()
		const socket = FakeWebSocket.instances[0]
		expect(socket).toBeDefined()
		socket?.open()
		socket?.message(event(71))
		socket?.message(event(72))
		await flush()
		await Bun.sleep(10)
		await flush()
		expect(failures).toBe(1)
		// The failure never reached the stream: no replay connection, nothing dropped.
		expect(FakeWebSocket.instances.length).toBe(1)
		expect(client.cursor).toBe(72)
		expect(client.queued).toBe(0)
		client.destroy()
	})

	test('quarantines malformed bytes, ignores late frames, then resets only after an acked replay', async () => {
		globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket
		const client = new JetstreamClient({
			url: 'wss://relay.example/subscribe',
			cursor: 40,
			onEvents: async () => undefined,
			reconnectMinMs: 1,
			reconnectMaxMs: 1,
			reconnectMaxExponent: 0,
		})
		client.start()
		const socket = FakeWebSocket.instances[0]
		expect(socket).toBeDefined()
		if (!socket) throw new Error('missing test socket')
		socket.closeImmediately = false
		socket.open()
		socket.message('{not json')
		await flush()
		expect(client.cursor).toBe(40)
		expect(client.isQuarantined).toBe(true)
		expect(client.protocolFailureCount).toBe(1)
		expect(client.failureKind).toBe('protocol')
		socket.message(event(41))
		await flush()
		expect(client.queued).toBe(0)
		socket.finishClose()
		await Bun.sleep(5)
		expect(FakeWebSocket.instances.length).toBeGreaterThanOrEqual(2)
		const replay = FakeWebSocket.instances[FakeWebSocket.instances.length - 1]
		replay?.open()
		replay?.message(event(41))
		await flush()
		expect(client.cursor).toBe(41)
		expect(client.isQuarantined).toBe(false)
		expect(client.protocolFailureCount).toBe(0)
		client.destroy()
	})

	test('retains a full bounded queue, drains it in order, then reconnects from the latest ack', async () => {
		globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket
		let releaseFirst: (() => void) | undefined
		const first = new Promise<void>((resolve) => {
			releaseFirst = resolve
		})
		const processed: number[] = []
		const client = new JetstreamClient({
			url: 'wss://relay.example/subscribe',
			cursor: 10,
			maxQueue: 2,
			batchMax: 1,
			reconnectMinMs: 1,
			reconnectMaxMs: 1,
			reconnectMaxExponent: 0,
			onEvents: async (events) => {
				for (const value of events) processed.push(value.time_us)
				if (events[0]?.time_us === 11) await first
			},
		})
		client.start()
		const socket = FakeWebSocket.instances[0]
		socket?.open()
		socket?.message(event(11))
		socket?.message(event(12))
		await flush()
		expect(client.queued).toBe(2)
		expect(client.failureKind).toBe('queue')
		expect(client.isConnected).toBe(false)
		expect(FakeWebSocket.instances).toHaveLength(1)
		// A frame delivered after the close is ignored locally and will be replayed
		// from the durable cursor on the replacement connection.
		socket?.message(event(13))
		releaseFirst?.()
		await flush()
		expect(processed).toEqual([11, 12])
		expect(client.cursor).toBe(12)
		await Bun.sleep(5)
		expect(FakeWebSocket.instances.length).toBeGreaterThanOrEqual(2)
		const replay = FakeWebSocket.instances[FakeWebSocket.instances.length - 1]
		expect(replay?.url).toContain('cursor=12')
		replay?.open()
		replay?.message(event(13))
		await flush()
		expect(processed).toEqual([11, 12, 13])
		client.destroy()
	})

	test('replays from the last durable batch when a batch keeps failing', async () => {
		globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket
		const client = new JetstreamClient({
			url: 'wss://relay.example/subscribe',
			cursor: 20,
			reconnectMinMs: 1,
			reconnectMaxMs: 1,
			reconnectMaxExponent: 0,
			// This exercises stream-level failure, so give the batch no retries.
			batchMaxAttempts: 1,
			onEvents: async () => {
				throw new Error('durable batch failed')
			},
		})
		client.start()
		const socket = FakeWebSocket.instances[0]
		expect(socket).toBeDefined()
		if (!socket) throw new Error('missing test socket')
		socket.closeImmediately = false
		socket.open()
		socket.message(event(21))
		await flush()
		expect(client.failureKind).toBe('handler')
		expect(client.queued).toBe(0)
		expect(client.cursor).toBe(20)
		socket.finishClose()
		await Bun.sleep(5)
		expect(FakeWebSocket.instances.length).toBeGreaterThanOrEqual(2)
		expect(FakeWebSocket.instances[FakeWebSocket.instances.length - 1]?.url).toContain('cursor=20')
		client.destroy()
	})

	test('rejects malformed relay fields and caller query injection', () => {
		expect(parseJetstreamEvent({ did: DID, time_us: 1, kind: 'commit', commit: { rev: 'not-a-tid' } })).toBeNull()
		expect(parseJetstreamEvent({ did: DID, time_us: 1.5, kind: 'identity' })).toBeNull()
		expect(
			() =>
				new JetstreamClient({
					url: 'wss://relay.example/subscribe?wantedDids=did:plc:evil',
					onEvents: async () => undefined,
				}),
		).toThrow('Unsafe Jetstream relay URL')
	})

	test('uses exact encoded URL bytes at the subscription hard boundary', () => {
		const options = { url: 'wss://relay.example/', wantedDids: [DID] }
		const cursor = Number.MAX_SAFE_INTEGER
		const base = buildJetstreamSubscriptionUrl(options, cursor)
		const fill = MAX_JETSTREAM_SUBSCRIPTION_URL_BYTES - Buffer.byteLength(base)
		expect(fill).toBeGreaterThan(0)
		const boundary = buildJetstreamSubscriptionUrl(
			{ ...options, url: `wss://relay.example/${'a'.repeat(fill)}` },
			cursor,
		)
		expect(Buffer.byteLength(boundary)).toBe(MAX_JETSTREAM_SUBSCRIPTION_URL_BYTES)
		expect(() =>
			buildJetstreamSubscriptionUrl({ ...options, url: `wss://relay.example/${'a'.repeat(fill + 1)}` }, cursor),
		).toThrow('subscription URL limit')
	})
})