import { afterAll, afterEach, beforeEach, describe, expect, test } from 'bun:test'
import { readFile, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import type Redis from 'ioredis'
import type {
	CacheInvalidationDependenciesForTests,
	CacheInvalidationGapRecoveryDependencies,
	CacheInvalidationMessage,
} from './cache-invalidation'

const DID = 'did:plc:test'
const RKEY = 'site'
const cursorFile = join(tmpdir(), `wisp-cache-invalidation-${process.pid}.lastid`)
const originalCursorFile = process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE
const originalRedisUrl = process.env.REDIS_URL

const hotPrefixes: string[] = []
const warmPrefixes: string[] = []
const cacheDeletes: Array<[string, string]> = []
const cachePrefixDeletes: Array<[string, string]> = []
const htmlWarmupResets: Array<[string, string]> = []
const redisClients: FakeRedis[] = []
let hotFailuresRemaining = 0
let xinfoFirstEntryId: string | null = null
let xinfoResponse: unknown | null = null
let xinfoError: Error | null = null
let notifyReplayRead: (() => void) | null = null

class FakeRedis {
	readonly xreadArguments: unknown[][] = []
	readonly options: Record<string, unknown> | undefined
	status = 'ready'
	pingCount = 0
	pingError: Error | null = null
	disconnected = false
	private readonly listeners = new Map<string, Array<(...args: unknown[]) => void>>()
	private readonly pendingReads: Array<(reason?: unknown) => void> = []

	constructor(...args: unknown[]) {
		this.options = args[1] as Record<string, unknown> | undefined
		redisClients.push(this)
	}

	on(event: string, listener: (...args: unknown[]) => void): this {
		const listeners = this.listeners.get(event) ?? []
		listeners.push(listener)
		this.listeners.set(event, listeners)
		return this
	}

	emit(event: string, ...args: unknown[]): void {
		for (const listener of this.listeners.get(event) ?? []) {
			listener(...args)
		}
	}

	deferredSubscribeAck = false
	subscribeError: Error | null = null
	subscribeCalls: string[] = []
	private subscribeCallback: ((error: Error | null) => void) | null = null

	subscribe(channel: string, callback: (error: Error | null) => void): void {
		this.subscribeCalls.push(channel)
		if (this.deferredSubscribeAck) {
			this.subscribeCallback = callback
			return
		}
		callback(this.subscribeError)
	}

	ackSubscribe(): void {
		this.subscribeCallback?.(this.subscribeError)
		this.subscribeCallback = null
	}

	async ping(): Promise<string> {
		this.pingCount += 1
		if (this.pingError) throw this.pingError
		return 'PONG'
	}

	async xinfo(..._args: unknown[]): Promise<unknown> {
		if (xinfoError) throw xinfoError
		if (xinfoResponse !== null) return xinfoResponse
		return xinfoFirstEntryId ? ['first-entry', [xinfoFirstEntryId, []]] : []
	}

	xread(...args: unknown[]): Promise<null> {
		this.xreadArguments.push(args)
		// A real ioredis client rejects commands after 'end'; mirror that so a
		// replay loop reaching XREAD after disconnect cannot hang on a promise
		// that never settles.
		if (this.disconnected) {
			return Promise.reject(new Error('Connection is closed.'))
		}
		notifyReplayRead?.()
		notifyReplayRead = null
		const { promise, reject } = Promise.withResolvers<null>()
		this.pendingReads.push(reject)
		return promise
	}

	disconnect(): void {
		this.disconnected = true
		const error = new Error('Redis disconnected')
		for (const reject of this.pendingReads) {
			reject(error)
		}
		this.pendingReads.length = 0
	}
}

/**
 * Captures heartbeat interval requests instead of running real timers. Each
 * scheduled entry stays fireable until its cancel handle is called, mirroring a
 * real clearInterval without needing fake time.
 */
class FakeHeartbeatScheduler {
	readonly requestedIntervalMs: number[] = []
	cancelledCount = 0
	private readonly entries: Array<{ intervalMs: number; cancelled: boolean; tick: () => void }> = []

	schedule = (callback: () => void, intervalMs: number): (() => void) => {
		const entry = { intervalMs, cancelled: false, tick: callback }
		this.entries.push(entry)
		this.requestedIntervalMs.push(intervalMs)
		return () => {
			entry.cancelled = true
			this.cancelledCount += 1
		}
	}

	get activeCount(): number {
		return this.entries.filter((entry) => !entry.cancelled).length
	}

	fireActiveTicks(): void {
		for (const entry of [...this.entries]) {
			if (!entry.cancelled) entry.tick()
		}
	}
}

const fakeHotTier = {
	async deletePrefix(prefix: string): Promise<number> {
		hotPrefixes.push(prefix)
		if (hotFailuresRemaining > 0) {
			hotFailuresRemaining -= 1
			throw new Error('hot tier unavailable')
		}
		return 3
	},
}

const fakeWarmTier = {
	async deletePrefix(prefix: string): Promise<number> {
		warmPrefixes.push(prefix)
		return 2
	},
}

const fakeCache = {
	delete(namespace: string, key: string): void {
		cacheDeletes.push([namespace, key])
	},
	deletePrefix(namespace: string, prefix: string): void {
		cachePrefixDeletes.push([namespace, prefix])
	},
}

// Keep replay cursor test writes out of the repository while exercising the
// real persistence queue.
process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE = cursorFile

const {
	applyCacheInvalidationForTests,
	clearSiteUpdating,
	compareStreamIds,
	getCacheInvalidationHealthSnapshot,
	getLastProcessedStreamIdForTests,
	getUpdatingSiteCountForTests,
	isSiteUpdating,
	markSiteUpdating,
	parseCacheInvalidationMessage,
	parseCacheInvalidationStreamEntry,
	resetCacheInvalidationReplayForTests,
	stopCacheInvalidationSubscriber,
	resetUpdatingSitesForTests,
	startCacheInvalidationSubscriberForTests,
} = await import('./cache-invalidation')

const testDependencies: CacheInvalidationDependenciesForTests = {
	hotTier: fakeHotTier as unknown as CacheInvalidationDependenciesForTests['hotTier'],
	warmTier: fakeWarmTier as unknown as CacheInvalidationDependenciesForTests['warmTier'],
	cache: fakeCache as unknown as CacheInvalidationDependenciesForTests['cache'],
	resetSiteHtmlHotCacheWarmup(did: string, rkey: string): void {
		htmlWarmupResets.push([did, rkey])
	},
}

function applyForTest(parsed: CacheInvalidationMessage, source: 'pubsub' | 'replay' = 'pubsub'): Promise<void> {
	return applyCacheInvalidationForTests(parsed, source, testDependencies)
}

function siteMessage(
	action: Extract<CacheInvalidationMessage['action'], 'updating' | 'update' | 'delete' | 'settings'>,
	overrides: Partial<CacheInvalidationMessage> = {},
): CacheInvalidationMessage {
	return { did: DID, rkey: RKEY, action, ...overrides }
}

async function flushAsyncWork(): Promise<void> {
	await Promise.resolve()
	await Promise.resolve()
	await Promise.resolve()
}

function waitForReplayRead(): Promise<void> {
	return new Promise((resolve) => {
		notifyReplayRead = resolve
	})
}

beforeEach(async () => {
	await resetCacheInvalidationReplayForTests()
	await rm(cursorFile, { force: true })
	resetUpdatingSitesForTests()
	hotPrefixes.length = 0
	warmPrefixes.length = 0
	cacheDeletes.length = 0
	cachePrefixDeletes.length = 0
	htmlWarmupResets.length = 0
	redisClients.length = 0
	hotFailuresRemaining = 0
	xinfoFirstEntryId = null
	xinfoResponse = null
	xinfoError = null
	notifyReplayRead = null
})

afterEach(async () => {
	await stopCacheInvalidationSubscriber()
	for (const client of redisClients) {
		client.disconnect()
	}
	redisClients.length = 0
})

afterAll(async () => {
	await stopCacheInvalidationSubscriber()
	for (const client of redisClients) {
		client.disconnect()
	}
	redisClients.length = 0
	await resetCacheInvalidationReplayForTests()
	await rm(cursorFile, { force: true })
	if (originalCursorFile === undefined) {
		delete process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE
	} else {
		process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE = originalCursorFile
	}
	if (originalRedisUrl === undefined) {
		delete process.env.REDIS_URL
	} else {
		process.env.REDIS_URL = originalRedisUrl
	}
})

describe('cache invalidation updating state', () => {
	test('stale token cannot clear a newer update', () => {
		markSiteUpdating(DID, RKEY, 'token-a')
		markSiteUpdating(DID, RKEY, 'token-b')

		expect(clearSiteUpdating(DID, RKEY, 'token-a')).toBe(false)
		expect(isSiteUpdating(DID, RKEY)).toBe(true)
	})

	test('matching token clears the active update', () => {
		markSiteUpdating(DID, RKEY, 'token-a')

		expect(clearSiteUpdating(DID, RKEY, 'token-a')).toBe(true)
		expect(isSiteUpdating(DID, RKEY)).toBe(false)
	})

	test('unversioned clear remains backward compatible', () => {
		markSiteUpdating(DID, RKEY, 'token-a')

		expect(clearSiteUpdating(DID, RKEY)).toBe(true)
		expect(isSiteUpdating(DID, RKEY)).toBe(false)
	})

	test('a newer terminal stream clears an older marker despite a different token', () => {
		markSiteUpdating(DID, RKEY, 'token-a', '1-0')

		expect(clearSiteUpdating(DID, RKEY, 'token-b', '2-0')).toBe(true)
		expect(isSiteUpdating(DID, RKEY)).toBe(false)
	})

	test('an older terminal stream cannot clear a newer marker despite a different token', () => {
		markSiteUpdating(DID, RKEY, 'token-b', '2-0')

		expect(clearSiteUpdating(DID, RKEY, 'token-a', '1-0')).toBe(false)
		expect(isSiteUpdating(DID, RKEY)).toBe(true)
	})

	test('legacy unversioned token mismatch cannot clear a marker', () => {
		markSiteUpdating(DID, RKEY, 'token-a')

		expect(clearSiteUpdating(DID, RKEY, 'token-b')).toBe(false)
		expect(isSiteUpdating(DID, RKEY)).toBe(true)
	})

	test('legacy token matching remains compatible when only the start has a stream id', () => {
		markSiteUpdating(DID, RKEY, 'token-a', '2-0')

		expect(clearSiteUpdating(DID, RKEY, 'token-a')).toBe(true)
		expect(isSiteUpdating(DID, RKEY)).toBe(false)
	})

	test('tokenized final update invalidates when the updating marker is absent', async () => {
		await applyForTest(siteMessage('update', { token: 'token-a' }))

		expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(isSiteUpdating(DID, RKEY)).toBe(false)
	})

	test('tokenized final update clears a matching marker and invalidates', async () => {
		markSiteUpdating(DID, RKEY, 'token-a')

		await applyForTest(siteMessage('update', { token: 'token-a' }))

		expect(isSiteUpdating(DID, RKEY)).toBe(false)
		expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`])
	})

	test('tokenized final update keeps a newer marker and still invalidates', async () => {
		markSiteUpdating(DID, RKEY, 'token-b')

		await applyForTest(siteMessage('update', { token: 'token-a' }))

		expect(isSiteUpdating(DID, RKEY)).toBe(true)
		expect(clearSiteUpdating(DID, RKEY, 'token-b')).toBe(true)
		expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`])
	})

	test('marking a new site prunes expired updating entries that were never requested', () => {
		const originalNow = Date.now
		let now = 1_000_000
		Date.now = () => now

		try {
			markSiteUpdating(DID, 'expired-a')
			markSiteUpdating(DID, 'expired-b')

			now += 10 * 60 * 1000 + 1
			markSiteUpdating(DID, 'active')

			expect(getUpdatingSiteCountForTests()).toBe(1)
			expect(isSiteUpdating(DID, 'active')).toBe(true)
			expect(isSiteUpdating(DID, 'expired-a')).toBe(false)
			expect(isSiteUpdating(DID, 'expired-b')).toBe(false)
		} finally {
			Date.now = originalNow
		}
	})

	test('updating state is capped even when entries have not expired', () => {
		for (let i = 0; i < 10_005; i++) {
			markSiteUpdating(DID, `site-${i}`)
		}

		expect(getUpdatingSiteCountForTests()).toBe(10_000)
		expect(isSiteUpdating(DID, 'site-0')).toBe(false)
		expect(isSiteUpdating(DID, 'site-5')).toBe(true)
		expect(isSiteUpdating(DID, 'site-10004')).toBe(true)
	})
})

describe('cache invalidation actions', () => {
	test('settings only clears settings-dependent memory caches', async () => {
		markSiteUpdating(DID, RKEY, 'active-update')

		await applyForTest(siteMessage('settings'))

		expect(hotPrefixes).toEqual([])
		expect(warmPrefixes).toEqual([])
		expect(cacheDeletes).toEqual([
			['redirectRules', `${DID}:${RKEY}`],
			['settings', `${DID}:${RKEY}`],
		])
		expect(cachePrefixDeletes).toEqual([
			['siteFiles', `${DID}:${RKEY}:`],
			['sourceCidMismatches', `${DID}:${RKEY}:`],
		])
		expect(htmlWarmupResets).toEqual([])
		expect(isSiteUpdating(DID, RKEY)).toBe(true)
	})

	test('site update purges file tiers and related memory caches', async () => {
		await applyForTest(siteMessage('update'))

		expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(cacheDeletes).toEqual([
			['redirectRules', `${DID}:${RKEY}`],
			['settings', `${DID}:${RKEY}`],
			['siteCache', `${DID}:${RKEY}`],
		])
		expect(cachePrefixDeletes).toEqual([
			['siteFiles', `${DID}:${RKEY}:`],
			['sourceCidMismatches', `${DID}:${RKEY}:`],
		])
		expect(htmlWarmupResets).toEqual([[DID, RKEY]])
	})

	test('site delete purges file tiers and related memory caches', async () => {
		await applyForTest(siteMessage('delete'))

		expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(cacheDeletes).toEqual([
			['redirectRules', `${DID}:${RKEY}`],
			['settings', `${DID}:${RKEY}`],
			['siteCache', `${DID}:${RKEY}`],
		])
		expect(cachePrefixDeletes).toEqual([
			['siteFiles', `${DID}:${RKEY}:`],
			['sourceCidMismatches', `${DID}:${RKEY}:`],
		])
		expect(htmlWarmupResets).toEqual([[DID, RKEY]])
	})

	test('uses the TieredStorage upper-cache invalidation fence when no tiers are injected', async () => {
		const upperCachePrefixes: string[] = []
		const productionStyleDependencies: CacheInvalidationDependenciesForTests = {
			storage: {
				async invalidateUpperCaches(prefix: string) {
					upperCachePrefixes.push(prefix)
					return { hotDeleted: 3, warmDeleted: 2, failures: [] }
				},
			},
			cache: fakeCache,
			resetSiteHtmlHotCacheWarmup(did: string, rkey: string): void {
				htmlWarmupResets.push([did, rkey])
			},
		}

		await applyCacheInvalidationForTests(siteMessage('update'), 'pubsub', productionStyleDependencies)

		expect(upperCachePrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(hotPrefixes).toEqual([])
		expect(warmPrefixes).toEqual([])
		expect(htmlWarmupResets).toEqual([[DID, RKEY]])
	})
})

describe('cache invalidation replay', () => {
	test('a higher live stream id cannot skip an earlier replay event for another site', async () => {
		const liveTarget = { did: DID, rkey: 'live-site', action: 'update' as const, streamId: '2-0' }
		const earlierReplayTarget = { did: DID, rkey: 'replay-site', action: 'update' as const, streamId: '1-0' }

		await applyForTest(liveTarget, 'pubsub')
		expect(getLastProcessedStreamIdForTests()).toBe('0-0')

		await applyForTest(earlierReplayTarget, 'replay')
		expect(getLastProcessedStreamIdForTests()).toBe('1-0')

		await applyForTest(liveTarget, 'replay')
		expect(getLastProcessedStreamIdForTests()).toBe('2-0')
		expect(hotPrefixes).toEqual([`${DID}/live-site/`, `${DID}/replay-site/`, `${DID}/live-site/`])
	})

	test('a live terminal update suppresses an older replayed updating marker', async () => {
		const terminal = siteMessage('update', { token: 'token-a', streamId: '2-0' })
		const olderUpdating = siteMessage('updating', { token: 'token-a', streamId: '1-0' })

		await applyForTest(terminal, 'pubsub')
		await applyForTest(olderUpdating, 'replay')

		expect(isSiteUpdating(DID, RKEY)).toBe(false)
		expect(getLastProcessedStreamIdForTests()).toBe('1-0')
		expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`])

		// Replay still reapplies the terminal event idempotently and advances only in order.
		await applyForTest(terminal, 'replay')
		expect(getLastProcessedStreamIdForTests()).toBe('2-0')
	})

	test('an older tokenless delete retains a newer live updating marker but still purges stale caches', async () => {
		await applyForTest(siteMessage('updating', { token: 'token-new', streamId: '2-0' }), 'pubsub')
		await applyForTest(siteMessage('delete', { streamId: '1-0' }), 'replay')

		expect(isSiteUpdating(DID, RKEY)).toBe(true)
		expect(getLastProcessedStreamIdForTests()).toBe('1-0')
		expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`])
	})

	test('settings and terminal invalidations remain independent across out-of-order delivery', async () => {
		await applyForTest(siteMessage('settings', { streamId: '2-0' }), 'pubsub')
		await applyForTest(siteMessage('delete', { streamId: '1-0' }), 'replay')

		// A newer settings event cannot suppress an older terminal file purge.
		expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(getLastProcessedStreamIdForTests()).toBe('1-0')

		const otherTerminal = { did: DID, rkey: 'mixed-site', action: 'update' as const, streamId: '3-0' }
		const olderSettings = { did: DID, rkey: 'mixed-site', action: 'settings' as const, streamId: '2-0' }
		await applyForTest(otherTerminal, 'pubsub')
		cacheDeletes.length = 0
		cachePrefixDeletes.length = 0
		await applyForTest(olderSettings, 'replay')

		// Settings cleanup is also safe after a terminal event for the same site.
		expect(cacheDeletes).toEqual([
			['redirectRules', `${DID}:mixed-site`],
			['settings', `${DID}:mixed-site`],
		])
		expect(cachePrefixDeletes).toEqual([
			['siteFiles', `${DID}:mixed-site:`],
			['sourceCidMismatches', `${DID}:mixed-site:`],
		])
		expect(getLastProcessedStreamIdForTests()).toBe('2-0')
	})

	test('tier failure does not advance the replay cursor and retries the same event', async () => {
		hotFailuresRemaining = 1
		const event = siteMessage('update', { streamId: '3-0' })

		await expect(applyForTest(event, 'replay')).rejects.toThrow('Failed to invalidate file caches')
		expect(getLastProcessedStreamIdForTests()).toBe('0-0')
		expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`])
		// Warm cleanup still ran despite the hot-tier failure.
		expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`])

		await applyForTest(event, 'replay')
		expect(getLastProcessedStreamIdForTests()).toBe('3-0')
		expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`, `${DID}/${RKEY}/`])
		expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`, `${DID}/${RKEY}/`])
	})

	test('TieredStorage upper-cache failures also retain the replay cursor for retry', async () => {
		const failure = new Error('hot tier unavailable')
		let attempts = 0
		const dependencies: CacheInvalidationDependenciesForTests = {
			storage: {
				async invalidateUpperCaches() {
					attempts++
					return attempts === 1
						? { hotDeleted: 0, warmDeleted: 2, failures: [{ tier: 'hot' as const, reason: failure }] }
						: { hotDeleted: 3, warmDeleted: 2, failures: [] }
				},
			},
			cache: fakeCache,
			resetSiteHtmlHotCacheWarmup(did: string, rkey: string): void {
				htmlWarmupResets.push([did, rkey])
			},
		}
		const event = siteMessage('update', { streamId: '3-0' })

		await expect(applyCacheInvalidationForTests(event, 'replay', dependencies)).rejects.toThrow(
			'Failed to invalidate file caches',
		)
		expect(getLastProcessedStreamIdForTests()).toBe('0-0')

		await applyCacheInvalidationForTests(event, 'replay', dependencies)
		expect(attempts).toBe(2)
		expect(getLastProcessedStreamIdForTests()).toBe('3-0')
	})

	test('tier failure retains a matching updating marker until the retry succeeds', async () => {
		markSiteUpdating(DID, RKEY, 'token-a', '3-0')
		hotFailuresRemaining = 1
		const event = siteMessage('update', { token: 'token-a', streamId: '3-0' })

		await expect(applyForTest(event, 'replay')).rejects.toThrow('Failed to invalidate file caches')
		expect(isSiteUpdating(DID, RKEY)).toBe(true)
		expect(getLastProcessedStreamIdForTests()).toBe('0-0')

		await applyForTest(event, 'replay')
		expect(isSiteUpdating(DID, RKEY)).toBe(false)
		expect(getLastProcessedStreamIdForTests()).toBe('3-0')
	})
})

describe('cache invalidation message parsing', () => {
	test('message parsing preserves token', () => {
		expect(
			parseCacheInvalidationMessage(JSON.stringify({ did: DID, rkey: RKEY, action: 'update', token: 'token-a' })),
		).toEqual({
			did: DID,
			rkey: RKEY,
			action: 'update',
			token: 'token-a',
		})
	})

	test('message parsing preserves stream id', () => {
		expect(
			parseCacheInvalidationMessage(
				JSON.stringify({ did: DID, rkey: RKEY, action: 'update', token: 'token-a', streamId: '1713811200000-2' }),
			),
		).toEqual({
			did: DID,
			rkey: RKEY,
			action: 'update',
			token: 'token-a',
			streamId: '1713811200000-2',
		})
	})

	test('stream entry parsing reconstructs invalidation messages', () => {
		expect(
			parseCacheInvalidationStreamEntry('1713811200000-5', [
				'did',
				DID,
				'rkey',
				RKEY,
				'action',
				'updating',
				'token',
				'token-a',
				'ts',
				'1713811200000',
			]),
		).toEqual({
			did: DID,
			rkey: RKEY,
			action: 'updating',
			token: 'token-a',
			streamId: '1713811200000-5',
		})
	})

	test('domain invalidation message parsing preserves domain keys', () => {
		expect(
			parseCacheInvalidationMessage(
				JSON.stringify({
					action: 'domain',
					domain: 'example.wisp.place',
					domainKind: 'wisp',
					streamId: '1713811200000-6',
				}),
			),
		).toEqual({
			action: 'domain',
			domain: 'example.wisp.place',
			domainKind: 'wisp',
			streamId: '1713811200000-6',
		})
	})

	test('domain stream entry parsing reconstructs domain invalidation messages', () => {
		expect(
			parseCacheInvalidationStreamEntry('1713811200000-7', [
				'action',
				'domain',
				'domain',
				'example.com',
				'domainKind',
				'custom',
				'customDomainId',
				'abc123',
				'ts',
				'1713811200000',
			]),
		).toEqual({
			action: 'domain',
			domain: 'example.com',
			domainKind: 'custom',
			customDomainId: 'abc123',
			streamId: '1713811200000-7',
		})
	})

	test('stream ids sort by timestamp and sequence', () => {
		expect(compareStreamIds('1713811200000-1', '1713811200000-2')).toBeLessThan(0)
		expect(compareStreamIds('1713811200001-0', '1713811200000-999')).toBeGreaterThan(0)
		expect(compareStreamIds('1713811200001-3', '1713811200001-3')).toBe(0)
	})
})

describe('cache invalidation replay health', () => {
	test('stopping disconnects a blocking replay read promptly', async () => {
		process.env.REDIS_URL = 'redis://cache-invalidation-test'
		startCacheInvalidationSubscriberForTests(
			(redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis,
		)
		await flushAsyncWork()

		expect(redisClients).toHaveLength(2)
		const [subscriber, replay] = redisClients
		expect(replay?.xreadArguments).toHaveLength(1)
		expect(replay?.options?.blockingTimeout).toEqual(expect.any(Number))
		expect(replay?.options?.socketTimeout as number).toBeGreaterThan(replay?.options?.blockingTimeout as number)

		await stopCacheInvalidationSubscriber()

		expect(subscriber?.disconnected).toBe(true)
		expect(replay?.disconnected).toBe(true)
		expect(getCacheInvalidationHealthSnapshot()).toMatchObject({
			subscriberConnected: false,
			replayConnected: false,
			replayState: 'stopped',
			retrying: false,
		})
	})

	test('recovers a retained-stream gap by clearing local state before replaying from zero', async () => {
		process.env.REDIS_URL = 'redis://cache-invalidation-test'
		xinfoFirstEntryId = '2-0'
		await writeFile(cursorFile, '1-0\n')
		markSiteUpdating(DID, RKEY, 'stale-update', '1-0')
		const recoveryOrder: string[] = []
		const replayRead = waitForReplayRead()
		const recovery: CacheInvalidationGapRecoveryDependencies = {
			clearMutableCaches() {
				recoveryOrder.push('cache-manager')
			},
			clearUpdatingAndTerminalState() {
				recoveryOrder.push('site-state')
				resetUpdatingSitesForTests()
			},
			resetHtmlPrewarmState() {
				recoveryOrder.push('html-prewarm')
			},
			async evictHotTier() {
				recoveryOrder.push('hot-tier')
			},
			async validateWarmCacheSourceCids() {
				recoveryOrder.push('warm-cid-validation')
			},
		}

		startCacheInvalidationSubscriberForTests(
			(redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis,
			recovery,
		)
		await replayRead

		expect(recoveryOrder).toEqual(['cache-manager', 'site-state', 'html-prewarm', 'hot-tier', 'warm-cid-validation'])
		expect(isSiteUpdating(DID, RKEY)).toBe(false)
		expect(getLastProcessedStreamIdForTests()).toBe('0-0')
		expect(await readFile(cursorFile, 'utf8')).toBe('0-0\n')
		const replayReadArguments = redisClients[1]?.xreadArguments[0] ?? []
		expect(replayReadArguments[replayReadArguments.length - 1]).toBe('0-0')

		const health = getCacheInvalidationHealthSnapshot()
		expect(health.replayState).toBe('starting')
		expect(health.gapCount).toBe(1)
		expect(health.lastGapAt).not.toBeNull()
		expect(health.lastGapRecoveryAt).not.toBeNull()
		expect(health.lastErrorAt).not.toBeNull()

		await stopCacheInvalidationSubscriber()
	})

	test('retries malformed XINFO without clearing state or resetting the cursor', async () => {
		process.env.REDIS_URL = 'redis://cache-invalidation-test'
		xinfoResponse = ['first-entry', ['not-a-stream-id', []]]
		await writeFile(cursorFile, '1-0\n')
		const recoveryOrder: string[] = []

		startCacheInvalidationSubscriberForTests(
			(redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis,
			{
				clearMutableCaches: () => {
					recoveryOrder.push('cache-manager')
				},
			},
		)
		await flushAsyncWork()

		expect(recoveryOrder).toEqual([])
		expect(getLastProcessedStreamIdForTests()).toBe('1-0')
		expect(redisClients[1]?.xreadArguments).toEqual([])
		expect(getCacheInvalidationHealthSnapshot()).toMatchObject({
			replayState: 'degraded',
			gapCount: 0,
			lastGapAt: null,
			lastGapRecoveryAt: null,
			retrying: true,
		})

		await stopCacheInvalidationSubscriber()
	})

	test('retries an XINFO Redis error without clearing state or resetting the cursor', async () => {
		process.env.REDIS_URL = 'redis://cache-invalidation-test'
		xinfoError = new Error('Redis temporarily unavailable')
		await writeFile(cursorFile, '1-0\n')
		const recoveryOrder: string[] = []

		startCacheInvalidationSubscriberForTests(
			(redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis,
			{
				clearMutableCaches: () => {
					recoveryOrder.push('cache-manager')
				},
			},
		)
		await flushAsyncWork()

		expect(recoveryOrder).toEqual([])
		expect(getLastProcessedStreamIdForTests()).toBe('1-0')
		expect(redisClients[1]?.xreadArguments).toEqual([])
		expect(getCacheInvalidationHealthSnapshot()).toMatchObject({
			replayState: 'degraded',
			gapCount: 0,
			lastGapAt: null,
			lastGapRecoveryAt: null,
			retrying: true,
		})

		await stopCacheInvalidationSubscriber()
	})
})

describe('cache invalidation subscriber heartbeat', () => {
	async function startHeartbeatSubscriber(scheduler: FakeHeartbeatScheduler): Promise<FakeRedis> {
		process.env.REDIS_URL = 'redis://cache-invalidation-test'
		startCacheInvalidationSubscriberForTests(
			(redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis,
			undefined,
			{ scheduleInterval: scheduler.schedule },
		)
		// Park the replay loop in its blocking read before any lifecycle changes,
		// matching the rhythm of the replay-health tests above.
		await flushAsyncWork()
		const subscriberClient = redisClients[0]
		if (!subscriberClient) throw new Error('subscriber client was not created')
		return subscriberClient
	}

	test('requests the 60 second production keepalive interval', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		const subscriberClient = await startHeartbeatSubscriber(scheduler)
		subscriberClient.emit('ready')

		expect(scheduler.requestedIntervalMs).toEqual([60_000])
		expect(subscriberClient.pingCount).toBe(0)

		await stopCacheInvalidationSubscriber()
	})

	test('pings the subscriber on each heartbeat interval tick while connected', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		const subscriberClient = await startHeartbeatSubscriber(scheduler)
		subscriberClient.emit('ready')

		scheduler.fireActiveTicks()
		expect(subscriberClient.pingCount).toBe(1)
		scheduler.fireActiveTicks()
		expect(subscriberClient.pingCount).toBe(2)
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)

		await stopCacheInvalidationSubscriber()
	})

	test('repeated ready events replace the heartbeat interval instead of stacking', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		const subscriberClient = await startHeartbeatSubscriber(scheduler)
		subscriberClient.emit('ready')
		subscriberClient.emit('ready')
		subscriberClient.emit('ready')

		expect(scheduler.requestedIntervalMs).toEqual([60_000, 60_000, 60_000])
		expect(scheduler.cancelledCount).toBe(2)
		expect(scheduler.activeCount).toBe(1)
		scheduler.fireActiveTicks()
		expect(subscriberClient.pingCount).toBe(1)
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)

		await stopCacheInvalidationSubscriber()
	})

	test('heartbeat stops on close/end and restarts on reconnect ready', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		const subscriberClient = await startHeartbeatSubscriber(scheduler)
		subscriberClient.emit('ready')
		scheduler.fireActiveTicks()
		scheduler.fireActiveTicks()
		expect(subscriberClient.pingCount).toBe(2)

		subscriberClient.emit('close')
		scheduler.fireActiveTicks()
		expect(subscriberClient.pingCount).toBe(2)
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false)

		subscriberClient.emit('ready')
		scheduler.fireActiveTicks()
		scheduler.fireActiveTicks()
		expect(subscriberClient.pingCount).toBe(4)
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)

		subscriberClient.emit('end')
		scheduler.fireActiveTicks()
		expect(subscriberClient.pingCount).toBe(4)
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false)

		await stopCacheInvalidationSubscriber()
	})

	test('a failed keepalive ping recreates the wedged subscriber and fences stale clients', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		const subscriberClient = await startHeartbeatSubscriber(scheduler)
		subscriberClient.emit('ready')
		scheduler.fireActiveTicks()
		expect(subscriberClient.pingCount).toBe(1)

		subscriberClient.pingError = new Error('subscriber keepalive failed')
		scheduler.fireActiveTicks()
		await flushAsyncWork()

		expect(getCacheInvalidationHealthSnapshot().lastErrorAt).not.toBeNull()
		expect(getCacheInvalidationHealthSnapshot().subscriberRecreations).toBe(1)
		expect(subscriberClient.disconnected).toBe(true)
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false)

		// The replacement client restores health on ready + SUBSCRIBE ack, and
		// the heartbeat continues against the new client.
		const replacement = redisClients[2]
		if (!replacement) throw new Error('replacement subscriber was not created')
		replacement.emit('ready')
		scheduler.fireActiveTicks()
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)
		expect(replacement.pingCount).toBe(1)

		// A stale client's late events are generation-fenced: a close from the
		// old client cannot flip the new client's health.
		subscriberClient.emit('close')
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)

		await stopCacheInvalidationSubscriber()
	})

	test('autoResubscribe is disabled so each ready issues its own acked SUBSCRIBE', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		const subscriberClient = await startHeartbeatSubscriber(scheduler)
		expect(subscriberClient.options?.autoResubscribe).toBe(false)

		subscriberClient.emit('ready')
		subscriberClient.emit('ready')
		expect(subscriberClient.subscribeCalls).toEqual(['wisp:cache-invalidate', 'wisp:cache-invalidate'])
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)

		await stopCacheInvalidationSubscriber()
	})

	test('ready alone is not connected until the SUBSCRIBE ack arrives', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		const subscriberClient = await startHeartbeatSubscriber(scheduler)
		subscriberClient.deferredSubscribeAck = true
		subscriberClient.emit('ready')

		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false)

		subscriberClient.ackSubscribe()
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)

		await stopCacheInvalidationSubscriber()
	})

	test('a failed SUBSCRIBE ack leaves the subscriber unconnected and records an error', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		const subscriberClient = await startHeartbeatSubscriber(scheduler)
		subscriberClient.subscribeError = new Error('NOPERM denied')
		subscriberClient.emit('ready')

		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false)
		expect(getCacheInvalidationHealthSnapshot().lastErrorAt).not.toBeNull()

		await stopCacheInvalidationSubscriber()
	})

	test('supervisor recreates a subscriber that stays ready without a SUBSCRIBE ack', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		const supervisorScheduler = new FakeHeartbeatScheduler()
		process.env.REDIS_URL = 'redis://cache-invalidation-test'
		startCacheInvalidationSubscriberForTests(
			(redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis,
			undefined,
			{ scheduleInterval: scheduler.schedule },
			{ scheduleInterval: supervisorScheduler.schedule, recreateAfterMs: 1 },
		)
		await flushAsyncWork()
		const subscriberClient = redisClients[0]
		if (!subscriberClient) throw new Error('subscriber client was not created')
		subscriberClient.deferredSubscribeAck = true
		subscriberClient.emit('ready')

		// recreateAfterMs=1: the first supervisor tick records the unhealthy
		// start; a tick after the bound recreates the client.
		supervisorScheduler.fireActiveTicks()
		await new Promise((resolve) => setTimeout(resolve, 2))
		supervisorScheduler.fireActiveTicks()
		await flushAsyncWork()
		expect(getCacheInvalidationHealthSnapshot().subscriberRecreations).toBe(1)
		expect(subscriberClient.disconnected).toBe(true)

		const replacement = redisClients[2]
		if (!replacement) throw new Error('replacement subscriber was not created')
		replacement.emit('ready')
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)

		await stopCacheInvalidationSubscriber()
	})

	test('start-once: a repeated full start is ignored and creates no duplicate clients', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		const subscriberClient = await startHeartbeatSubscriber(scheduler)
		startCacheInvalidationSubscriberForTests(
			(redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis,
			undefined,
			{ scheduleInterval: scheduler.schedule },
		)
		await flushAsyncWork()
		expect(redisClients).toHaveLength(2)

		subscriberClient.emit('ready')
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)
		await stopCacheInvalidationSubscriber()
	})

	test('stop-once: concurrent stops share one idempotent teardown', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		const subscriberClient = await startHeartbeatSubscriber(scheduler)
		subscriberClient.emit('ready')

		const [first, second] = [stopCacheInvalidationSubscriber(), stopCacheInvalidationSubscriber()]
		await Promise.all([first, second])
		await stopCacheInvalidationSubscriber()

		expect(subscriberClient.disconnected).toBe(true)
		expect(redisClients[1]?.disconnected).toBe(true)
		expect(getCacheInvalidationHealthSnapshot()).toMatchObject({
			subscriberConnected: false,
			replayConnected: false,
			replayState: 'stopped',
		})
	})

	test('stopping the subscriber cancels the heartbeat so no further pings fire', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		const subscriberClient = await startHeartbeatSubscriber(scheduler)
		subscriberClient.emit('ready')
		scheduler.fireActiveTicks()
		expect(subscriberClient.pingCount).toBe(1)

		await stopCacheInvalidationSubscriber()
		scheduler.fireActiveTicks()

		expect(subscriberClient.pingCount).toBe(1)
		expect(scheduler.activeCount).toBe(0)
		expect(subscriberClient.disconnected).toBe(true)
		expect(redisClients[1]?.disconnected).toBe(true)
		expect(getCacheInvalidationHealthSnapshot()).toMatchObject({
			subscriberConnected: false,
			replayConnected: false,
			replayState: 'stopped',
		})
	})
})