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 { Effect } from 'effect'
import type Redis from 'ioredis'
import type {
	CacheInvalidationDependenciesForTests,
	CacheInvalidationGapRecoveryDependencies,
	CacheInvalidationMessage,
} from './cache-invalidation'
import { invalidationIo } from './cache-invalidation-runtime'

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

type ReplayResponse = Array<[string, Array<[string, string[]]>]> | null

class FakeRedis {
	readonly xreadArguments: unknown[][] = []
	readonly options: Record<string, unknown> | undefined
	status = 'ready'
	pingCount = 0
	pingError: Error | null = null
	deferredPing: Promise<string> | undefined
	disconnected = false
	private readonly listeners = new Map<string, Array<(...args: unknown[]) => void>>()
	private readonly pendingReads: Array<{ resolve(response: ReplayResponse): void; reject(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 this.deferredPing ?? '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<ReplayResponse> {
		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, resolve, reject } = Promise.withResolvers<ReplayResponse>()
		this.pendingReads.push({ resolve, reject })
		return promise
	}

	deliver(response: ReplayResponse): void {
		const pending = this.pendingReads.shift()
		if (!pending) throw new Error('No pending replay read')
		pending.resolve(response)
	}

	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,
	getCacheInvalidationQueueSizeForTests,
	isCacheInvalidationOverloaded,
} = 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 new Promise<void>((resolve) => setImmediate(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'
		await 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')
			},
		}

		await 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[] = []

		await 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[] = []

		await 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'
		await 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()
		await flushAsyncWork()
		expect(subscriberClient.pingCount).toBe(1)
		scheduler.fireActiveTicks()
		await flushAsyncWork()
		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()
		await flushAsyncWork()
		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()
		await flushAsyncWork()
		scheduler.fireActiveTicks()
		await flushAsyncWork()
		expect(subscriberClient.pingCount).toBe(2)

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

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

		subscriberClient.emit('end')
		scheduler.fireActiveTicks()
		await flushAsyncWork()
		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()
		await flushAsyncWork()
		expect(subscriberClient.pingCount).toBe(1)

		subscriberClient.pingError = new Error('subscriber keepalive failed')
		scheduler.fireActiveTicks()
		await flushAsyncWork()
		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()
		await flushAsyncWork()
		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'
		await 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)
		await 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()
		await flushAsyncWork()
		expect(subscriberClient.pingCount).toBe(1)

		await stopCacheInvalidationSubscriber()
		scheduler.fireActiveTicks()
		await flushAsyncWork()

		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',
		})
	})
})

describe('bounded invalidation dispatch', () => {
	async function start(dependencies?: CacheInvalidationDependenciesForTests) {
		process.env.REDIS_URL = 'redis://cache-invalidation-test'
		process.env.WISP_CACHE_INVALIDATION_QUEUE_CAPACITY = '2'
		await startCacheInvalidationSubscriberForTests(
			(url, options) => new FakeRedis(url, options) as unknown as Redis,
			undefined,
			undefined,
			undefined,
			dependencies ?? testDependencies,
		)
		delete process.env.WISP_CACHE_INVALIDATION_QUEUE_CAPACITY
		await flushAsyncWork()
		const subscriber = redisClients[redisClients.length - 2]!
		const replay = redisClients[redisClients.length - 1]!
		subscriber.emit('ready')
		return { subscriber, replay }
	}
	const send = (subscriber: FakeRedis, message: CacheInvalidationMessage) =>
		subscriber.emit('message', 'wisp:cache-invalidate', JSON.stringify(message))
	const entry = (id: string, action = 'settings'): [string, string[]] => [
		id,
		['action', action, 'did', DID, 'rkey', RKEY],
	]

	test('durable overflow has a finite queue and recovers only through earlier ordered replay', async () => {
		const { subscriber, replay } = await start()
		for (let i = 1; i <= 20; i++) send(subscriber, siteMessage('settings', { streamId: `${i}-0` }))
		expect(getCacheInvalidationQueueSizeForTests()).toBeLessThanOrEqual(2)
		expect(getCacheInvalidationHealthSnapshot()).toMatchObject({ replayState: 'degraded', overloadLatched: false })
		expect(getLastProcessedStreamIdForTests()).toBe('0-0')
		const earlierRead = waitForReplayRead()
		replay.deliver([['wisp:cache-invalidate-stream', [entry('1-0')]]])
		await earlierRead
		expect(getLastProcessedStreamIdForTests()).toBe('1-0')
		expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded')
		const caughtUp = waitForReplayRead()
		replay.deliver([['wisp:cache-invalidate-stream', [entry('20-0')]]])
		await caughtUp
		expect(getLastProcessedStreamIdForTests()).toBe('20-0')
		expect(getCacheInvalidationHealthSnapshot().replayState).toBe('healthy')
		expect(await readFile(cursorFile, 'utf8')).toBe('20-0\n')
	})

	test('legacy overflow latches fail-closed even after healthy replay', async () => {
		const { subscriber, replay } = await start()
		for (let i = 0; i < 20; i++) send(subscriber, siteMessage('settings'))
		expect(getCacheInvalidationQueueSizeForTests()).toBeLessThanOrEqual(2)
		expect(isCacheInvalidationOverloaded()).toBe(true)
		const nextRead = waitForReplayRead()
		replay.deliver(null)
		await nextRead
		expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded')
		expect(isCacheInvalidationOverloaded()).toBe(true)
		const { default: app } = await import('../server')
		const response = await app.request('https://sites.wisp.place/did:plc:test/site/')
		expect(response.status).toBe(503)
		expect(response.headers.get('cache-control')).toBe('no-store')
		expect(await response.text()).toBe('Hosting temporarily unavailable')
		expect((await app.request('https://wisp.place/live')).status).toBe(200)
		expect((await app.request('https://wisp.place/health')).status).toBe(503)
		await stopCacheInvalidationSubscriber()
		await start()
		expect(isCacheInvalidationOverloaded()).toBe(true)
	})

	test('oversized input is never retained or parsed and latches conservatively', async () => {
		const { subscriber } = await start()
		subscriber.emit(
			'message',
			'wisp:cache-invalidate',
			JSON.stringify({ action: 'settings', did: DID, rkey: RKEY, streamId: '1-0', extra: 'x'.repeat(1024 * 1024) }),
		)
		expect(getCacheInvalidationQueueSizeForTests()).toBe(0)
		expect(isCacheInvalidationOverloaded()).toBe(true)
		expect(getCacheInvalidationHealthSnapshot().overflowCount).toBe(1)
	})

	test('malformed bounded input is discarded without a false legacy-overflow latch', async () => {
		const { subscriber } = await start()
		subscriber.emit('message', 'wisp:cache-invalidate', '{bad json')
		expect(getCacheInvalidationQueueSizeForTests()).toBe(0)
		expect(isCacheInvalidationOverloaded()).toBe(false)
		expect(getCacheInvalidationHealthSnapshot().lastErrorAt).not.toBeNull()
	})

	test('partial replay-client acquisition failure disconnects subscriber and cancels timers', async () => {
		process.env.REDIS_URL = 'redis://cache-invalidation-test'
		const scheduler = new FakeHeartbeatScheduler()
		let creations = 0
		await expect(
			startCacheInvalidationSubscriberForTests(
				(url, options) => {
					if (++creations === 2) throw new Error('secret connection string')
					return new FakeRedis(url, options) as unknown as Redis
				},
				undefined,
				{ scheduleInterval: scheduler.schedule },
				{ scheduleInterval: scheduler.schedule },
			),
		).rejects.toBeDefined()
		expect(redisClients[0]?.disconnected).toBe(true)
		expect(scheduler.activeCount).toBe(0)
		expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded')
	})

	test('stop waits for uncancelable accepted tier writes before finishing', async () => {
		const started = Promise.withResolvers<void>()
		const release = Promise.withResolvers<void>()
		const dependencies = {
			...testDependencies,
			storage: {
				async invalidateUpperCaches() {
					started.resolve()
					await release.promise
					return { hotDeleted: 0, warmDeleted: 0, failures: [] }
				},
			},
			hotTier: undefined,
			warmTier: undefined,
		}
		delete dependencies.hotTier
		delete dependencies.warmTier
		const { subscriber, replay } = await start(dependencies)
		send(subscriber, siteMessage('update', { streamId: '1-0' }))
		await started.promise
		for (let i = 2; i <= 20; i++) send(subscriber, siteMessage('settings', { streamId: `${i}-0` }))
		expect(getCacheInvalidationQueueSizeForTests()).toBe(2)
		expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded')
		let stopped = false
		const closing = stopCacheInvalidationSubscriber().then(() => {
			stopped = true
		})
		await flushAsyncWork()
		expect(replay.disconnected).toBe(true)
		expect(stopped).toBe(false)
		release.resolve()
		await closing
		expect(stopped).toBe(true)
	})

	test('cursor persistence failure retains cursor and terminal marker, then retry persists before clearing', async () => {
		const original = process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE
		const impossible = join(cursorFile, 'child')
		await writeFile(cursorFile, 'not a directory')
		process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE = impossible
		markSiteUpdating(DID, RKEY, 'token-a', '1-0')
		try {
			await expect(
				applyForTest(siteMessage('update', { token: 'token-a', streamId: '1-0' }), 'replay'),
			).rejects.toBeDefined()
			expect(getLastProcessedStreamIdForTests()).toBe('0-0')
			expect(isSiteUpdating(DID, RKEY)).toBe(true)
		} finally {
			process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE = original
		}
		await applyForTest(siteMessage('update', { token: 'token-a', streamId: '1-0' }), 'replay')
		expect(getLastProcessedStreamIdForTests()).toBe('1-0')
		expect(isSiteUpdating(DID, RKEY)).toBe(false)
		expect(await readFile(cursorFile, 'utf8')).toBe('1-0\n')
	})

	test('stale subscription ack cannot make a replacement subscriber healthy', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		process.env.REDIS_URL = 'redis://cache-invalidation-test'
		await startCacheInvalidationSubscriberForTests(
			(url, options) => new FakeRedis(url, options) as unknown as Redis,
			undefined,
			undefined,
			{ scheduleInterval: scheduler.schedule, recreateAfterMs: 1 },
		)
		const old = redisClients[0]!
		old.deferredSubscribeAck = true
		old.emit('ready')
		scheduler.fireActiveTicks()
		await new Promise((resolve) => setTimeout(resolve, 2))
		scheduler.fireActiveTicks()
		expect(old.disconnected).toBe(true)
		old.ackSubscribe()
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false)
		redisClients[2]!.emit('ready')
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)
	})

	test('replay storage failure remains degraded until the same entry is retried', async () => {
		const { subscriber, replay } = await start()
		markSiteUpdating(DID, RKEY, 'token-a', '1-0')
		hotFailuresRemaining = 1
		for (let i = 1; i <= 5; i++) send(subscriber, siteMessage('settings', { streamId: `${i}-0` }))
		replay.deliver([
			['wisp:cache-invalidate-stream', [['1-0', ['action', 'update', 'did', DID, 'rkey', RKEY, 'token', 'token-a']]]],
		])
		await flushAsyncWork()
		expect(getLastProcessedStreamIdForTests()).toBe('0-0')
		expect(isSiteUpdating(DID, RKEY)).toBe(true)
		expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded')
		const retryRead = waitForReplayRead()
		await retryRead
		const caughtUp = waitForReplayRead()
		replay.deliver([
			[
				'wisp:cache-invalidate-stream',
				[['1-0', ['action', 'update', 'did', DID, 'rkey', RKEY, 'token', 'token-a']], entry('5-0')],
			],
		])
		await caughtUp
		expect(getLastProcessedStreamIdForTests()).toBe('5-0')
		expect(isSiteUpdating(DID, RKEY)).toBe(false)
		expect(getCacheInvalidationHealthSnapshot().replayState).toBe('healthy')
	})

	test('restart replays safely from the last persisted cursor, not a later live ID', async () => {
		const { subscriber, replay } = await start()
		const nextRead = waitForReplayRead()
		replay.deliver([['wisp:cache-invalidate-stream', [entry('1-0')]]])
		await nextRead
		send(subscriber, siteMessage('settings', { streamId: '9-0' }))
		await stopCacheInvalidationSubscriber()
		const restarted = await start()
		const args = restarted.replay.xreadArguments[0]!
		expect(args[args.length - 1]).toBe('1-0')
		const duplicate = waitForReplayRead()
		restarted.replay.deliver([['wisp:cache-invalidate-stream', [entry('1-0'), entry('2-0')]]])
		await duplicate
		expect(getLastProcessedStreamIdForTests()).toBe('2-0')
		expect(await readFile(cursorFile, 'utf8')).toBe('2-0\n')
	})

	test('a cache callback defect latches rather than entering a network retry loop', async () => {
		const { subscriber } = await start({
			...testDependencies,
			cache: {
				delete() {
					throw new Error('secret defect')
				},
				deletePrefix() {},
			},
		})
		send(subscriber, siteMessage('settings', { streamId: '1-0' }))
		await flushAsyncWork()
		expect(isCacheInvalidationOverloaded()).toBe(true)
		expect(getLastProcessedStreamIdForTests()).toBe('0-0')
		expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded')
	})

	test('an unsettled PING cannot accumulate a command or fiber on each timer tick', async () => {
		const scheduler = new FakeHeartbeatScheduler()
		process.env.REDIS_URL = 'redis://cache-invalidation-test'
		await startCacheInvalidationSubscriberForTests(
			(url, options) => new FakeRedis(url, options) as unknown as Redis,
			undefined,
			{ scheduleInterval: scheduler.schedule },
		)
		const client = redisClients[0]!
		const pong = Promise.withResolvers<string>()
		client.deferredPing = pong.promise
		client.emit('ready')
		for (let i = 0; i < 100; i++) scheduler.fireActiveTicks()
		expect(client.pingCount).toBe(1)
		client.emit('ready')
		scheduler.fireActiveTicks()
		expect(client.pingCount).toBe(1)
		await stopCacheInvalidationSubscriber()
		pong.resolve('PONG')
		await flushAsyncWork()
		expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false)
		expect(isCacheInvalidationOverloaded()).toBe(false)
	})

	test('oversized replay fields latch without advancing or retaining an event', async () => {
		const { replay } = await start()
		replay.deliver([
			[
				'wisp:cache-invalidate-stream',
				[['1-0', ['action', 'settings', 'did', DID, 'rkey', RKEY, 'extra', 'x'.repeat(1024 * 1024)]]],
			],
		])
		await flushAsyncWork()
		expect(isCacheInvalidationOverloaded()).toBe(true)
		expect(getLastProcessedStreamIdForTests()).toBe('0-0')
		expect(getCacheInvalidationQueueSizeForTests()).toBe(0)
	})

	test('fallback tier deletion batches keys and closes the iterator on failure', async () => {
		let closed = false
		const batches: string[][] = []
		const tier = {
			async *listKeys() {
				try {
					for (let i = 0; i < 200; i++) yield `key-${i}`
				} finally {
					closed = true
				}
			},
			async deleteMany(keys: string[]) {
				batches.push(keys)
				if (batches.length === 2) throw new Error('disk unavailable')
			},
		}
		await expect(
			applyCacheInvalidationForTests(siteMessage('update', { streamId: '1-0' }), 'replay', {
				...testDependencies,
				hotTier: tier as unknown as CacheInvalidationDependenciesForTests['hotTier'],
			}),
		).rejects.toThrow('Failed to invalidate file caches')
		expect(batches.map((batch) => batch.length)).toEqual([64, 64])
		expect(closed).toBe(true)
		expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`])
		expect(getLastProcessedStreamIdForTests()).toBe('0-0')
	})

	test('runtime persistence failure is visible in readiness and retains replay cursor/update marker', async () => {
		const { replay } = await start()
		const original = process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE
		await writeFile(cursorFile, 'not a directory')
		process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE = join(cursorFile, 'child')
		markSiteUpdating(DID, RKEY, 'token-a', '1-0')
		const retried = waitForReplayRead()
		replay.deliver([
			['wisp:cache-invalidate-stream', [['1-0', ['action', 'update', 'did', DID, 'rkey', RKEY, 'token', 'token-a']]]],
		])
		try {
			await retried
			expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded')
			expect(getLastProcessedStreamIdForTests()).toBe('0-0')
			expect(isSiteUpdating(DID, RKEY)).toBe(true)
		} finally {
			process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE = original
		}
		const caughtUp = waitForReplayRead()
		replay.deliver([
			['wisp:cache-invalidate-stream', [['1-0', ['action', 'update', 'did', DID, 'rkey', RKEY, 'token', 'token-a']]]],
		])
		await caughtUp
		expect(getCacheInvalidationHealthSnapshot().replayState).toBe('healthy')
		expect(getLastProcessedStreamIdForTests()).toBe('1-0')
		expect(isSiteUpdating(DID, RKEY)).toBe(false)
	})

	test('gap recovery is ordered with live work and does not issue file-tier purges', async () => {
		process.env.REDIS_URL = 'redis://cache-invalidation-test'
		await writeFile(cursorFile, '1-0\n')
		xinfoFirstEntryId = '2-0'
		const began = Promise.withResolvers<void>()
		const release = Promise.withResolvers<void>()
		const read = waitForReplayRead()
		await startCacheInvalidationSubscriberForTests(
			(url, options) => new FakeRedis(url, options) as unknown as Redis,
			{
				clearMutableCaches() {},
				clearUpdatingAndTerminalState: resetUpdatingSitesForTests,
				resetHtmlPrewarmState() {},
				async evictHotTier() {
					began.resolve()
					await release.promise
				},
				validateWarmCacheSourceCids() {},
			},
			undefined,
			undefined,
			testDependencies,
		)
		await began.promise
		redisClients[0]!.emit(
			'message',
			'wisp:cache-invalidate',
			JSON.stringify(siteMessage('updating', { token: 'new', streamId: '3-0' })),
		)
		await flushAsyncWork()
		expect(isSiteUpdating(DID, RKEY)).toBe(false)
		expect(getCacheInvalidationQueueSizeForTests()).toBe(1)
		release.resolve()
		await read
		await flushAsyncWork()
		expect(isSiteUpdating(DID, RKEY)).toBe(true)
		expect(hotPrefixes).toEqual([])
		expect(warmPrefixes).toEqual([])
	})

	test('expected IO errors carry no serializable foreign causes or credentials', async () => {
		const hostile = { message: 'redis://user:secret@private-host', body: { token: 'secret' }, path: '/private/secret' }
		const result = await Effect.runPromise(Effect.result(invalidationIo('xread', () => Promise.reject(hostile))))
		expect(result._tag).toBe('Failure')
		if (result._tag !== 'Failure') throw new Error('Expected IO failure')
		expect(JSON.stringify(result.failure)).not.toContain('secret')
		expect(JSON.stringify(result.failure)).not.toContain('private-host')
		expect('cause' in result.failure).toBe(false)
		expect(result.failure.operation).toBe('xread')
	})

	test('missing-stream recovery preserves a safe discriminator, not the Redis error body', async () => {
		const result = await Effect.runPromise(
			Effect.result(invalidationIo('xinfo', () => Promise.reject(new Error('ERR no such key secret path')))),
		)
		expect(result._tag).toBe('Failure')
		if (result._tag !== 'Failure') throw new Error('Expected IO failure')
		expect(result.failure.missingStream).toBe(true)
		expect(JSON.stringify(result.failure)).not.toContain('secret')
	})
})