/**
 * Cache invalidation subscriber
 *
 * Uses Redis pub/sub for low-latency invalidation and a Redis stream for replay.
 * When a site is updated/deleted, clears the hosting-service's local caches
 * (tiered storage hot+warm tiers, redirect rules) so stale data isn't served.
 *
 * Also tracks sites that are actively being downloaded ('updating' action) so
 * the serving layer can show a "site updating" page instead of stale/partial content.
 */

import { existsSync, readFileSync } from 'node:fs'
import { mkdir, rename, writeFile } from 'node:fs/promises'
import { basename, dirname, resolve } from 'node:path'
import {
	type CacheInvalidationMessage,
	DEFAULT_CACHE_INVALIDATION_CHANNEL,
	DEFAULT_CACHE_INVALIDATION_STREAM,
	decodeCacheInvalidationMessage,
} from '@wispplace/constants'
import { createLogger } from '@wispplace/observability'
import type { StorageTier } from '@wispplace/tiered-storage'
import { Effect, Exit, Scope, Stream } from 'effect'
import Redis from 'ioredis'
import {
	InvalidationIoError,
	type InvalidationRuntime,
	invalidationIo,
	makeInvalidationRuntime,
} from './cache-invalidation-runtime'
import { cache } from './cache-manager'
import { resetAllHtmlHotCacheWarmups, resetSiteHtmlHotCacheWarmup } from './html-prewarm'
import { hotTier, storage, warmTier } from './storage'

const logger = createLogger('cache-invalidation')
const CHANNEL = DEFAULT_CACHE_INVALIDATION_CHANNEL
const STREAM = process.env.WISP_CACHE_INVALIDATION_STREAM || DEFAULT_CACHE_INVALIDATION_STREAM
const STREAM_BLOCK_MS = parsePositiveInt(process.env.WISP_CACHE_INVALIDATION_BLOCK_MS, 5000)
const STREAM_BATCH_COUNT = parsePositiveInt(process.env.WISP_CACHE_INVALIDATION_BATCH_COUNT, 100)
const STREAM_BLOCKING_GRACE_MS = parsePositiveInt(process.env.WISP_CACHE_INVALIDATION_BLOCKING_GRACE_MS, 1000)
const STREAM_SOCKET_TIMEOUT_MS = Math.max(
	parsePositiveInt(
		process.env.WISP_CACHE_INVALIDATION_SOCKET_TIMEOUT_MS,
		STREAM_BLOCK_MS + STREAM_BLOCKING_GRACE_MS + 10_000,
	),
	STREAM_BLOCK_MS + STREAM_BLOCKING_GRACE_MS + 1000,
)
const SUBSCRIBER_RECREATE_AFTER_MS = 180_000

function getCursorFile(): string {
	return (
		process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE ||
		resolve(process.env.CACHE_DIR || './cache/sites', '..', 'cache-invalidation.lastid')
	)
}
const STREAM_ID_PATTERN = /^\d+-\d+$/

export type { CacheInvalidationMessage } from '@wispplace/constants'

// Sites currently being downloaded by the firehose-service.
// Maps `${did}/${rkey}` → current update token and timestamp.
// Used to show an "updating" page instead of serving stale files.
const UPDATING_TTL_MS = 10 * 60 * 1000 // 10 minutes safety timeout
const MAX_UPDATING_SITES = parsePositiveInt(process.env.WISP_MAX_UPDATING_SITES, 10_000)
type UpdatingSiteState = { since: number; token?: string; streamId?: string }
type TerminalSiteState = { since: number; streamId: string }

const updatingSites = new Map<string, UpdatingSiteState>()
// This local, bounded high-water mark prevents live terminal events from being
// undone by older replayed starts. Replay remains the only durable cursor owner.
const latestTerminalSiteStreamIds = new Map<string, TerminalSiteState>()

type InvalidationCache = Pick<typeof cache, 'delete' | 'deletePrefix'>
type UpperCacheInvalidator = Pick<typeof storage, 'invalidateUpperCaches'>

export interface CacheInvalidationDependenciesForTests {
	/** Use the real TieredStorage upper-cache fence in production or inject it in focused tests. */
	storage?: UpperCacheInvalidator
	/** Supplying either tier keeps deterministic legacy injected-tier tests on the direct fallback. */
	hotTier?: StorageTier
	warmTier?: StorageTier
	cache?: InvalidationCache
	resetSiteHtmlHotCacheWarmup?: typeof resetSiteHtmlHotCacheWarmup
}

interface CacheInvalidationDependencies {
	storage: UpperCacheInvalidator | undefined
	hotTier: StorageTier
	warmTier: StorageTier | undefined
	cache: InvalidationCache
	resetSiteHtmlHotCacheWarmup: typeof resetSiteHtmlHotCacheWarmup
}

function getCacheInvalidationDependencies(
	overrides?: CacheInvalidationDependenciesForTests,
): CacheInvalidationDependencies {
	const hasInjectedFileTiers = overrides !== undefined && ('hotTier' in overrides || 'warmTier' in overrides)
	return {
		storage: hasInjectedFileTiers ? undefined : (overrides?.storage ?? storage),
		hotTier: overrides?.hotTier ?? hotTier,
		warmTier: overrides?.warmTier ?? warmTier,
		cache: overrides?.cache ?? cache,
		resetSiteHtmlHotCacheWarmup: overrides?.resetSiteHtmlHotCacheWarmup ?? resetSiteHtmlHotCacheWarmup,
	}
}

/**
 * Recovery hooks for a Redis stream retention gap. The warm-cache validation
 * hook is intentionally separate: file serving validates entries against the
 * current source CID, so recovery must not wipe up to 10GB of warm disk data.
 */
export interface CacheInvalidationGapRecoveryDependencies {
	clearMutableCaches?: () => void | Promise<void>
	clearUpdatingAndTerminalState?: () => void | Promise<void>
	resetHtmlPrewarmState?: () => void | Promise<void>
	evictHotTier?: () => void | Promise<void>
	validateWarmCacheSourceCids?: () => void | Promise<void>
}

interface ResolvedCacheInvalidationGapRecoveryDependencies {
	clearMutableCaches: () => void | Promise<void>
	clearUpdatingAndTerminalState: () => void | Promise<void>
	resetHtmlPrewarmState: () => void | Promise<void>
	evictHotTier: () => void | Promise<void>
	validateWarmCacheSourceCids: () => void | Promise<void>
}

function clearUpdatingAndTerminalState(): void {
	updatingSites.clear()
	latestTerminalSiteStreamIds.clear()
}

function getCacheInvalidationGapRecoveryDependencies(
	overrides?: CacheInvalidationGapRecoveryDependencies,
): ResolvedCacheInvalidationGapRecoveryDependencies {
	return {
		clearMutableCaches: overrides?.clearMutableCaches ?? (() => cache.clearAll()),
		clearUpdatingAndTerminalState: overrides?.clearUpdatingAndTerminalState ?? clearUpdatingAndTerminalState,
		resetHtmlPrewarmState: overrides?.resetHtmlPrewarmState ?? resetAllHtmlHotCacheWarmups,
		evictHotTier: overrides?.evictHotTier ?? (() => hotTier.clear()),
		// File serving validates retained warm entries against the source CID. This
		// seam permits deployments to perform an additional validation pass before
		// replay resumes, without an unsafe broad warm-tier wipe.
		validateWarmCacheSourceCids: overrides?.validateWarmCacheSourceCids ?? (() => undefined),
	}
}

let cacheInvalidationGapRecoveryDependencies = getCacheInvalidationGapRecoveryDependencies()

function getSiteInvalidationKey(did: string, rkey: string): string {
	return `${did}/${rkey}`
}

function pruneExpiredUpdatingSites(now = Date.now()): void {
	for (const [key, state] of updatingSites) {
		if (now - state.since > UPDATING_TTL_MS) {
			updatingSites.delete(key)
		}
	}
}

function pruneExpiredTerminalSiteStreamIds(now = Date.now()): void {
	for (const [key, state] of latestTerminalSiteStreamIds) {
		if (now - state.since > UPDATING_TTL_MS) {
			latestTerminalSiteStreamIds.delete(key)
		}
	}
}

function enforceUpdatingSitesLimit(): void {
	while (updatingSites.size > MAX_UPDATING_SITES) {
		const oldestKey = updatingSites.keys().next().value
		if (oldestKey === undefined) return
		updatingSites.delete(oldestKey)
	}
}

function enforceTerminalSiteStreamIdsLimit(): void {
	while (latestTerminalSiteStreamIds.size > MAX_UPDATING_SITES) {
		const oldestKey = latestTerminalSiteStreamIds.keys().next().value
		if (oldestKey === undefined) return
		latestTerminalSiteStreamIds.delete(oldestKey)
	}
}

function recordTerminalSiteStreamId(did: string, rkey: string, streamId: string | undefined): void {
	const normalized = normalizeStreamId(streamId)
	if (!normalized) return

	const now = Date.now()
	pruneExpiredTerminalSiteStreamIds(now)
	const key = getSiteInvalidationKey(did, rkey)
	const previous = latestTerminalSiteStreamIds.get(key)
	if (previous && compareStreamIds(normalized, previous.streamId) < 0) return

	latestTerminalSiteStreamIds.delete(key)
	latestTerminalSiteStreamIds.set(key, { since: now, streamId: normalized })
	enforceTerminalSiteStreamIdsLimit()
}

function isUpdatingSupersededByTerminal(did: string, rkey: string, streamId: string | undefined): boolean {
	const normalized = normalizeStreamId(streamId)
	if (!normalized) return false

	pruneExpiredTerminalSiteStreamIds()
	const terminal = latestTerminalSiteStreamIds.get(getSiteInvalidationKey(did, rkey))
	return terminal !== undefined && compareStreamIds(terminal.streamId, normalized) >= 0
}

export function isSiteUpdating(did: string, rkey: string): boolean {
	const key = getSiteInvalidationKey(did, rkey)
	const state = updatingSites.get(key)
	if (state === undefined) return false
	if (Date.now() - state.since > UPDATING_TTL_MS) {
		// Firehose must have crashed; remove the stale entry
		updatingSites.delete(key)
		return false
	}
	return true
}

export function markSiteUpdating(did: string, rkey: string, token?: string, streamId?: string): void {
	const now = Date.now()
	const key = getSiteInvalidationKey(did, rkey)
	pruneExpiredUpdatingSites(now)
	// Refresh insertion order so the cap evicts the oldest active update.
	updatingSites.delete(key)
	updatingSites.set(key, { since: now, token, streamId: normalizeStreamId(streamId) })
	enforceUpdatingSitesLimit()
}

function canClearSiteUpdating(did: string, rkey: string, token?: string, streamId?: string): boolean {
	const state = updatingSites.get(getSiteInvalidationKey(did, rkey))
	if (!state) return false

	const terminalStreamId = normalizeStreamId(streamId)
	if (state.streamId && terminalStreamId) {
		// Comparable stream IDs are authoritative. A later terminal event wins even
		// if its updating hint was missed or carried a different token.
		return compareStreamIds(terminalStreamId, state.streamId) >= 0
	}

	// Without comparable stream IDs, retain the legacy token guard. This keeps
	// tokenized producers interoperable when only one side carries a stream ID.
	return !(token && state.token && state.token !== token)
}

export function clearSiteUpdating(did: string, rkey: string, token?: string, streamId?: string): boolean {
	if (!canClearSiteUpdating(did, rkey, token, streamId)) return false
	updatingSites.delete(getSiteInvalidationKey(did, rkey))
	return true
}

export function resetUpdatingSitesForTests(): void {
	clearUpdatingAndTerminalState()
}

export function getUpdatingSiteCountForTests(): number {
	return updatingSites.size
}

export interface CacheInvalidationHealthSnapshot {
	subscriberConnected: boolean
	replayConnected: boolean
	replayState: 'stopped' | 'starting' | 'healthy' | 'degraded' | 'gap'
	cursor: string
	lastEventAt: number | null
	lastErrorAt: number | null
	lastGapAt: number | null
	gapCount: number
	lastGapRecoveryAt: number | null
	retrying: boolean
	subscriberRecreations: number
	overloadLatched?: boolean
	overflowCount?: number
}

export interface CacheInvalidationRedisOptions {
	maxRetriesPerRequest: number
	enableReadyCheck: boolean
	blockingTimeout?: number
	blockingTimeoutGrace?: number
	socketTimeout?: number
	autoResubscribe?: boolean
}

export type CacheInvalidationRedisFactory = (redisUrl: string, options: CacheInvalidationRedisOptions) => Redis

/**
 * Schedules the repeating subscriber keepalive PING and returns a cancel handle.
 * Production uses an unref'd interval; focused tests inject a fake that records
 * the requested delay and fires ticks manually instead of waiting on real time.
 */
export type CacheInvalidationHeartbeatIntervalScheduler = (callback: () => void, intervalMs: number) => () => void

/** Overrides for the subscriber keepalive PING; focused tests inject a fake scheduler. */
export interface CacheInvalidationSubscriberHeartbeatOptions {
	scheduleInterval?: CacheInvalidationHeartbeatIntervalScheduler
}

export interface CacheInvalidationSubscriberSupervisorOptions {
	scheduleInterval?: CacheInvalidationHeartbeatIntervalScheduler
	recreateAfterMs?: number
}

const defaultRedisClientFactory: CacheInvalidationRedisFactory = (redisUrl, options) => new Redis(redisUrl, options)

// The keepalive must never hold the event loop open during shutdown.
const scheduleRealHeartbeatInterval: CacheInvalidationHeartbeatIntervalScheduler = (callback, intervalMs) => {
	const timer = setInterval(callback, intervalMs)
	timer.unref()
	return () => clearInterval(timer)
}

let runtimeScope: Scope.Closeable | undefined
let runtime: InvalidationRuntime | undefined
let startPromise: Promise<void> | undefined
let stopPromise: Promise<void> | undefined
let replayCursorIsEstablished = false
let lastProcessedStreamId = '0-0'
let cacheInvalidationHealth: CacheInvalidationHealthSnapshot = {
	subscriberConnected: false,
	replayConnected: false,
	replayState: 'stopped',
	cursor: lastProcessedStreamId,
	lastEventAt: null,
	lastErrorAt: null,
	lastGapAt: null,
	gapCount: 0,
	lastGapRecoveryAt: null,
	retrying: false,
	subscriberRecreations: 0,
}

export function getCacheInvalidationHealthSnapshot(): CacheInvalidationHealthSnapshot {
	return { ...cacheInvalidationHealth, cursor: lastProcessedStreamId }
}

function setReplayState(state: CacheInvalidationHealthSnapshot['replayState']): void {
	cacheInvalidationHealth.replayState = state
}

function recordCacheInvalidationError(): void {
	cacheInvalidationHealth.lastErrorAt = Date.now()
}

function singleLineLogValue(value: string, maxLength = 256): string {
	return Array.from(value, (character) => {
		const code = character.charCodeAt(0)
		return code <= 0x1f || code === 0x7f || character === '\u2028' || character === '\u2029' ? ' ' : character
	})
		.join('')
		.slice(0, maxLength)
}

function safeCursorFileBasename(cursorFile: string): string {
	const name = basename(cursorFile)
	return /^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/.test(name) ? name : '<unsafe-basename>'
}

function cursorLogContext(cursorFile: string): { cursorFile: string; cursorFileConfigured: boolean } {
	return {
		cursorFile: safeCursorFileBasename(cursorFile),
		cursorFileConfigured: Boolean(process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE),
	}
}

function errorKind(error: unknown): string {
	if (error instanceof Error && error.name) return singleLineLogValue(error.name, 80)
	return 'UnknownError'
}

function siteLogContext(did: string, rkey: string): { did: string; rkey: string } {
	return { did: singleLineLogValue(did), rkey: singleLineLogValue(rkey) }
}

function invalidationLogContext(parsed: CacheInvalidationMessage): Record<string, string | boolean | undefined> {
	if (parsed.action === 'domain') {
		return {
			action: parsed.action,
			domain: parsed.domain === undefined ? undefined : singleLineLogValue(parsed.domain),
			domainKind: parsed.domainKind,
			hasCustomDomainId: Boolean(parsed.customDomainId),
		}
	}

	return {
		action: parsed.action,
		...siteLogContext(parsed.did ?? '(missing)', parsed.rkey ?? '(missing)'),
		tokenized: Boolean(parsed.token),
	}
}

function parsePositiveInt(value: string | undefined, fallback: number): number {
	if (!value) return fallback
	const parsed = Number.parseInt(value, 10)
	return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback
}

function normalizeStreamId(streamId: string | undefined): string | undefined {
	if (!streamId) return undefined
	const trimmed = streamId.trim()
	return STREAM_ID_PATTERN.test(trimmed) ? trimmed : undefined
}

function parseStreamIdParts(streamId: string): [bigint, bigint] {
	const normalized = normalizeStreamId(streamId)
	if (!normalized) {
		throw new Error('Invalid Redis stream ID')
	}

	const [msRaw, seqRaw] = normalized.split('-') as [string, string]
	return [BigInt(msRaw), BigInt(seqRaw)]
}

export function compareStreamIds(a: string, b: string): number {
	const [aMs, aSeq] = parseStreamIdParts(a)
	const [bMs, bSeq] = parseStreamIdParts(b)
	if (aMs < bMs) return -1
	if (aMs > bMs) return 1
	if (aSeq < bSeq) return -1
	if (aSeq > bSeq) return 1
	return 0
}

function loadCursorFromDisk(): void {
	replayCursorIsEstablished = false
	const cursorFile = getCursorFile()
	if (!existsSync(cursorFile)) return

	try {
		const stored = normalizeStreamId(readFileSync(cursorFile, 'utf8'))
		if (!stored) {
			logger.warn('[CacheInvalidation] Invalid cursor file ignored', {
				operation: 'cursor-load',
				...cursorLogContext(cursorFile),
			})
			return
		}
		lastProcessedStreamId = stored
		replayCursorIsEstablished = true
		logger.info('[CacheInvalidation] Replay cursor loaded', {
			operation: 'cursor-load',
			streamId: stored,
			...cursorLogContext(cursorFile),
		})
	} catch (err) {
		recordCacheInvalidationError()
		logger.error('[CacheInvalidation] Cursor load failed', undefined, {
			operation: 'cursor-load',
			...cursorLogContext(cursorFile),
			errorKind: errorKind(err),
		})
	}
}

const persistCursor = Effect.fnUntraced(function* (streamId: string) {
	const cursorFile = getCursorFile()
	yield* invalidationIo('cursor-persist', async () => {
		await mkdir(dirname(cursorFile), { recursive: true })
		const temporary = `${cursorFile}.tmp`
		await writeFile(temporary, `${streamId}\n`, 'utf8')
		await rename(temporary, cursorFile)
	})
})

const advanceStreamCursor = Effect.fnUntraced(function* (streamId: string | undefined) {
	const normalized = normalizeStreamId(streamId)
	if (!normalized || compareStreamIds(normalized, lastProcessedStreamId) <= 0) return
	yield* persistCursor(normalized)
	lastProcessedStreamId = normalized
	replayCursorIsEstablished = true
})

function shouldSkipReplayMessage(streamId: string | undefined): boolean {
	const normalized = normalizeStreamId(streamId)
	if (!normalized) return false
	return compareStreamIds(normalized, lastProcessedStreamId) <= 0
}

type XInfoField = { found: boolean; value: unknown }

function findXInfoField(streamInfo: unknown[], wantedField: string): XInfoField {
	for (let index = 0; index < streamInfo.length - 1; index += 2) {
		if (streamInfo[index] === wantedField) return { found: true, value: streamInfo[index + 1] }
	}
	return { found: false, value: undefined }
}

function parseXInfoLength(value: unknown): number | undefined {
	if (typeof value === 'number') return value
	if (typeof value === 'string' && /^\d+$/.test(value)) return Number(value)
	return undefined
}

function parseXInfoFirstEntry(value: unknown): string | undefined {
	if (value === null) return undefined
	if (!Array.isArray(value) || typeof value[0] !== 'string') {
		throw new Error('Malformed Redis XINFO first-entry response')
	}
	const firstStreamId = normalizeStreamId(value[0])
	if (!firstStreamId) throw new Error('Malformed Redis XINFO first-entry ID')
	return firstStreamId
}

function extractFirstStreamEntryId(streamInfo: unknown): string | undefined {
	if (!Array.isArray(streamInfo)) throw new Error('Malformed Redis XINFO STREAM response')
	if (streamInfo.length === 0) return undefined

	const firstEntry = findXInfoField(streamInfo, 'first-entry')
	if (firstEntry.found) return parseXInfoFirstEntry(firstEntry.value)

	if (parseXInfoLength(findXInfoField(streamInfo, 'length').value) === 0) return undefined
	throw new Error('Malformed Redis XINFO STREAM response')
}

const recoverReplayCursorGap = Effect.fnUntraced(function* (firstStreamId: string) {
	const priorCursor = lastProcessedStreamId
	cacheInvalidationHealth.lastGapAt = Date.now()
	cacheInvalidationHealth.gapCount += 1
	cacheInvalidationHealth.retrying = false
	setReplayState('degraded')
	recordCacheInvalidationError()
	logger.warn('[CacheInvalidation] Replay retention gap detected; starting fail-closed recovery', {
		operation: 'replay-gap-detected',
		priorCursor,
		firstRetainedStreamId: firstStreamId,
		gapCount: cacheInvalidationHealth.gapCount,
	})

	const recovery = cacheInvalidationGapRecoveryDependencies
	// All local state must be cleared before the cursor can move. A failure in
	// any step leaves the old cursor intact and makes the loop retry recovery.
	yield* invalidationIo('gap-clearMutableCaches', async () => recovery.clearMutableCaches())
	yield* invalidationIo('gap-clearUpdatingAndTerminalState', async () => recovery.clearUpdatingAndTerminalState())
	yield* invalidationIo('gap-resetHtmlPrewarmState', async () => recovery.resetHtmlPrewarmState())
	yield* invalidationIo('gap-evictHotTier', async () => recovery.evictHotTier())
	yield* invalidationIo('gap-validateWarmCacheSourceCids', async () => recovery.validateWarmCacheSourceCids())

	// Persist the reset before exposing it in memory. If this fails, retain the
	// old cursor and retry rather than replaying with an unrecorded safe point.
	yield* persistCursor('0-0')
	lastProcessedStreamId = '0-0'
	replayCursorIsEstablished = false
	cacheInvalidationHealth.lastGapRecoveryAt = Date.now()
	setReplayState('starting')
	logger.info('[CacheInvalidation] Replay retention gap recovered; replaying retained stream', {
		operation: 'replay-gap-recovered',
		priorCursor,
		firstRetainedStreamId: firstStreamId,
		resetCursor: lastProcessedStreamId,
		gapCount: cacheInvalidationHealth.gapCount,
	})
})

const ensureReplayCursorIsRetained = Effect.fnUntraced(function* (client: Redis) {
	// A first boot has no established cursor to prove was behind a trimmed range.
	// Once replay has a cursor, a cursor that predates first-entry needs a local
	// cache reset before replay can safely resume from the retained history.
	if (!replayCursorIsEstablished) return true

	const streamInfo = yield* invalidationIo('xinfo', () => client.xinfo('STREAM', STREAM)).pipe(
		Effect.catch((error) => {
			if (error.missingStream) return Effect.succeed([])
			return Effect.fail(error)
		}),
	)

	const firstStreamId = yield* Effect.try({
		try: () => extractFirstStreamEntryId(streamInfo),
		catch: () => new InvalidationIoError({ operation: 'xinfo-decode' }),
	})
	if (!firstStreamId || compareStreamIds(lastProcessedStreamId, firstStreamId) >= 0) {
		return true
	}

	yield* recoverReplayCursorGap(firstStreamId)
	return true
})

export function parseCacheInvalidationMessage(message: string): CacheInvalidationMessage | null {
	const parsed = decodeCacheInvalidationMessage(message)
	if (!parsed) return null
	parsed.streamId = normalizeStreamId(typeof parsed.streamId === 'string' ? parsed.streamId : undefined)
	return parsed
}

export function parseCacheInvalidationStreamEntry(streamId: string, fields: string[]): CacheInvalidationMessage | null {
	const payload: Record<string, string> = {}
	for (let index = 0; index < fields.length - 1; index += 2) {
		payload[fields[index]!] = fields[index + 1]!
	}
	delete payload.ts
	return parseCacheInvalidationMessage(JSON.stringify({ ...payload, streamId }))
}

const invalidateTier = (tier: StorageTier, prefix: string) => {
	if (tier.deletePrefix) return invalidationIo('tier-delete-prefix', () => tier.deletePrefix!(prefix))
	return Stream.fromAsyncIterable(
		tier.listKeys(prefix),
		() => new InvalidationIoError({ operation: 'tier-list' }),
	).pipe(
		Stream.grouped(64),
		Stream.mapEffect((keys) =>
			invalidationIo('tier-delete-many', async () => {
				await tier.deleteMany([...keys])
				return keys.length
			}),
		),
		Stream.runFold(
			() => 0,
			(total, count) => total + count,
		),
	)
}

const invalidateSiteFileTiers = Effect.fnUntraced(function* (
	did: string,
	rkey: string,
	dependencies: CacheInvalidationDependencies,
) {
	const prefix = `${did}/${rkey}/`
	const logFailure = (tier: string, reason: unknown) =>
		logger.error('[CacheInvalidation] Tier invalidation failed', undefined, {
			operation: 'site-tier-invalidate',
			tier,
			...siteLogContext(did, rkey),
			errorKind: errorKind(reason),
		})
	// Retain the shared promotion fence: direct hot/warm deletion is only an injected-test fallback.
	if (dependencies.storage) {
		const result = yield* invalidationIo('invalidate-upper-caches', () =>
			dependencies.storage!.invalidateUpperCaches(prefix),
		)
		for (const failure of result.failures) logFailure(failure.tier, failure.reason)
		return {
			hotDeleted: result.hotDeleted,
			warmDeleted: result.warmDeleted,
			failures: result.failures.map((failure) => failure.reason),
		}
	}
	const tiers: Array<{ tier: StorageTier; name: 'hot' | 'warm' }> = [{ tier: dependencies.hotTier, name: 'hot' }]
	if (dependencies.warmTier) tiers.push({ tier: dependencies.warmTier, name: 'warm' })
	// Attempt both tiers even if one fails; replay retries partial purges idempotently.
	const results = yield* Effect.forEach(tiers, ({ tier }) => Effect.result(invalidateTier(tier, prefix)), {
		concurrency: 2,
	})
	let hotDeleted = 0
	let warmDeleted = 0
	const failures: unknown[] = []
	for (const [index, result] of results.entries()) {
		const { name } = tiers[index]!
		if (result._tag === 'Success') {
			if (name === 'hot') hotDeleted = result.success
			else warmDeleted = result.success
		} else {
			logFailure(name, result.failure)
			failures.push(result.failure)
		}
	}
	return { hotDeleted, warmDeleted, failures }
})

const advanceReplayCursor = (source: 'pubsub' | 'replay', streamId: string | undefined) =>
	source === 'replay' ? advanceStreamCursor(streamId) : Effect.void

const applyCacheInvalidation = Effect.fnUntraced(function* (
	parsed: CacheInvalidationMessage,
	source: 'pubsub' | 'replay',
	dependencies = getCacheInvalidationDependencies(),
) {
	const { did, rkey, action, token, streamId } = parsed
	cacheInvalidationHealth.lastEventAt = Date.now()

	// Pub/sub is only a low-latency hint. It must not move the durable cursor because
	// a later live message can otherwise skip earlier ordered stream entries.
	if (source === 'replay' && shouldSkipReplayMessage(streamId)) {
		logger.info('[CacheInvalidation] Duplicate replay event skipped', {
			operation: 'event-skip-duplicate',
			source,
			streamId,
			...invalidationLogContext(parsed),
		})
		return
	}

	logger.info('[CacheInvalidation] Invalidation event received', {
		operation: 'event-received',
		source,
		streamId,
		...invalidationLogContext(parsed),
	})

	if (action === 'domain') {
		applyDomainCacheInvalidation(parsed, dependencies.cache)
		yield* advanceReplayCursor(source, streamId)
		return
	}

	if (!did || !rkey) {
		logger.warn('[CacheInvalidation] Site invalidation missing identifier', {
			operation: 'site-event-invalid',
			source,
			streamId,
			action,
			hasDid: Boolean(did),
			hasRkey: Boolean(rkey),
		})
		yield* advanceReplayCursor(source, streamId)
		return
	}

	if (action === 'updating') {
		if (isUpdatingSupersededByTerminal(did, rkey, streamId)) {
			logger.info('[CacheInvalidation] Superseded updating marker ignored', {
				operation: 'updating-marker-skip',
				source,
				streamId,
				...siteLogContext(did, rkey),
			})
			yield* advanceReplayCursor(source, streamId)
			return
		}

		markSiteUpdating(did, rkey, token, streamId)
		yield* advanceReplayCursor(source, streamId)
		logger.info('[CacheInvalidation] Site marked updating', {
			operation: 'updating-marker-set',
			source,
			streamId,
			...siteLogContext(did, rkey),
			tokenized: Boolean(token),
		})
		return
	}

	if (action === 'settings') {
		// Settings affect routing and fallback paths, but do not change stored site files.
		dependencies.cache.delete('redirectRules', `${did}:${rkey}`)
		dependencies.cache.delete('settings', `${did}:${rkey}`)
		dependencies.cache.deletePrefix('siteFiles', `${did}:${rkey}:`)
		dependencies.cache.deletePrefix('sourceCidMismatches', `${did}:${rkey}:`)
		yield* advanceReplayCursor(source, streamId)
		return
	}

	// Do not remove a matching updating marker until every file-tier purge has
	// succeeded. A replay retry must keep serving the updating state, not stale
	// or partially refreshed cache content, while a terminal invalidation fails.
	const canClearUpdating = canClearSiteUpdating(did, rkey, token, streamId)
	if (!canClearUpdating && action === 'update' && token) {
		// The marker may be absent after a restart/TTL expiry, or belong to a newer
		// update. In either case, final invalidation is still required and idempotent.
		logger.info('[CacheInvalidation] Updating marker absent; invalidating anyway', {
			operation: 'updating-marker-missing',
			reason: 'missing-or-replaced',
			source,
			streamId,
			...siteLogContext(did, rkey),
		})
	} else if (!canClearUpdating && streamId) {
		// A newer start may stay visible, but terminal cache cleanup is still safe.
		logger.info('[CacheInvalidation] Updating marker not clearable; invalidating anyway', {
			operation: 'updating-marker-missing',
			reason: 'newer-marker',
			source,
			streamId,
			...siteLogContext(did, rkey),
		})
	}

	const { hotDeleted, warmDeleted, failures } = yield* invalidateSiteFileTiers(did, rkey, dependencies)

	logger.info('[CacheInvalidation] Site file tiers invalidated', {
		operation: 'site-tier-invalidate',
		hotDeleted,
		warmDeleted,
		source,
		streamId,
		...siteLogContext(did, rkey),
	})

	dependencies.cache.delete('redirectRules', `${did}:${rkey}`)
	dependencies.cache.delete('settings', `${did}:${rkey}`)
	dependencies.cache.delete('siteCache', `${did}:${rkey}`)
	dependencies.cache.deletePrefix('siteFiles', `${did}:${rkey}:`)
	dependencies.cache.deletePrefix('sourceCidMismatches', `${did}:${rkey}:`)
	dependencies.resetSiteHtmlHotCacheWarmup(did, rkey)

	if (failures.length > 0) {
		recordCacheInvalidationError()
		return yield* new InvalidationIoError({ operation: 'invalidate-tiers' })
	}

	yield* advanceReplayCursor(source, streamId)
	if (canClearUpdating) {
		clearSiteUpdating(did, rkey, token, streamId)
	}

	// Only a completed terminal invalidation can supersede an older replayed start.
	recordTerminalSiteStreamId(did, rkey, streamId)
})

export async function applyCacheInvalidationForTests(
	parsed: CacheInvalidationMessage,
	source: 'pubsub' | 'replay' = 'pubsub',
	dependencies?: CacheInvalidationDependenciesForTests,
): Promise<void> {
	await Effect.runPromise(applyCacheInvalidation(parsed, source, getCacheInvalidationDependencies(dependencies)))
}

export function getLastProcessedStreamIdForTests(): string {
	return lastProcessedStreamId
}

export async function resetCacheInvalidationReplayForTests(): Promise<void> {
	await stopCacheInvalidationSubscriber()
	lastProcessedStreamId = '0-0'
	replayCursorIsEstablished = false
	cacheInvalidationHealth = initialHealth()
	cacheInvalidationHealth.overloadLatched = false
	cacheInvalidationHealth.overflowCount = 0
}

function applyDomainCacheInvalidation(parsed: CacheInvalidationMessage, invalidationCache: InvalidationCache): void {
	const domain = parsed.domain?.trim().toLowerCase()
	if (!domain) return

	if (parsed.domainKind !== 'custom') {
		invalidationCache.delete('domains', domain)
	}

	if (parsed.domainKind !== 'wisp') {
		invalidationCache.delete('customDomains', domain)
		if (parsed.customDomainId) {
			invalidationCache.delete('customDomains', parsed.customDomainId)
			invalidationCache.delete('customDomains', `hash:${parsed.customDomainId}`)
		}
	}

	logger.info('[CacheInvalidation] Domain lookup cache invalidated', {
		operation: 'domain-cache-clear',
		...invalidationLogContext({ ...parsed, domain }),
	})
}

function initialHealth(): CacheInvalidationHealthSnapshot {
	return {
		subscriberConnected: false,
		replayConnected: false,
		replayState: 'starting',
		cursor: lastProcessedStreamId,
		lastEventAt: null,
		lastErrorAt: null,
		lastGapAt: null,
		gapCount: 0,
		lastGapRecoveryAt: null,
		retrying: false,
		subscriberRecreations: 0,
		overloadLatched: cacheInvalidationHealth.overloadLatched ?? false,
		overflowCount: cacheInvalidationHealth.overflowCount ?? 0,
	}
}

export function isCacheInvalidationOverloaded(): boolean {
	return cacheInvalidationHealth.overloadLatched === true
}

export function getCacheInvalidationQueueSizeForTests(): number {
	return runtime?.queued() ?? 0
}

function boundedConfiguration(value: string | undefined, fallback: number, maximum: number): number {
	const parsed = Number(value)
	return Number.isSafeInteger(parsed) && parsed > 0 && parsed <= maximum ? parsed : fallback
}

/** Process integration boundary. Resources are acquired once and owned by one scope. */
export function startCacheInvalidationSubscriber(
	gapRecoveryOverrides?: CacheInvalidationGapRecoveryDependencies,
): Promise<void> {
	return startCacheInvalidationSubscriberForTests(defaultRedisClientFactory, gapRecoveryOverrides)
}

export function startCacheInvalidationSubscriberForTests(
	factory: CacheInvalidationRedisFactory,
	gapRecoveryOverrides?: CacheInvalidationGapRecoveryDependencies,
	heartbeatOptions?: CacheInvalidationSubscriberHeartbeatOptions,
	supervisorOptions?: CacheInvalidationSubscriberSupervisorOptions,
	dependencies?: CacheInvalidationDependenciesForTests,
): Promise<void> {
	if (startPromise) return startPromise
	if (runtimeScope || stopPromise) return Promise.resolve()
	const redisUrl = process.env.REDIS_URL
	if (!redisUrl) {
		cacheInvalidationHealth.replayState = 'stopped'
		return Promise.resolve()
	}
	cacheInvalidationHealth = initialHealth()
	cacheInvalidationGapRecoveryDependencies = getCacheInvalidationGapRecoveryDependencies(gapRecoveryOverrides)
	loadCursorFromDisk()
	const scope = Scope.makeUnsafe()
	runtimeScope = scope
	startPromise = Effect.runPromise(
		Scope.provide(scope)(
			makeInvalidationRuntime({
				redisUrl,
				factory,
				channel: CHANNEL,
				stream: STREAM,
				blockMs: STREAM_BLOCK_MS,
				batchCount: STREAM_BATCH_COUNT,
				blockingGraceMs: STREAM_BLOCKING_GRACE_MS,
				socketTimeoutMs: STREAM_SOCKET_TIMEOUT_MS,
				capacity: boundedConfiguration(process.env.WISP_CACHE_INVALIDATION_QUEUE_CAPACITY, 1024, 65_536),
				maxMessageBytes: boundedConfiguration(
					process.env.WISP_CACHE_INVALIDATION_MAX_MESSAGE_BYTES,
					1024 * 1024,
					1024 * 1024,
				),
				heartbeatScheduler: heartbeatOptions?.scheduleInterval ?? scheduleRealHeartbeatInterval,
				supervisorScheduler: supervisorOptions?.scheduleInterval ?? scheduleRealHeartbeatInterval,
				recreateAfterMs: supervisorOptions?.recreateAfterMs ?? SUBSCRIBER_RECREATE_AFTER_MS,
				health: cacheInvalidationHealth,
				parse: parseCacheInvalidationMessage,
				parseEntry: parseCacheInvalidationStreamEntry,
				apply: (message, source) =>
					applyCacheInvalidation(message, source, getCacheInvalidationDependencies(dependencies)),
				ensureRetained: ensureReplayCursorIsRetained,
				advance: advanceStreamCursor,
				cursor: () => lastProcessedStreamId,
				compare: compareStreamIds,
				error: (operation, error) => {
					recordCacheInvalidationError()
					logger.error('[CacheInvalidation] Invalidation operation failed', undefined, {
						operation,
						errorKind: errorKind(error),
					})
				},
			}),
		),
	)
		.then((acquired) => {
			runtime = acquired
		})
		.catch(async (error: unknown) => {
			await Effect.runPromise(Scope.close(scope, Exit.void))
			runtimeScope = undefined
			cacheInvalidationHealth.replayState = 'degraded'
			throw error
		})
		.finally(() => {
			startPromise = undefined
		})
	return startPromise
}

export function stopCacheInvalidationSubscriber(): Promise<void> {
	if (stopPromise) return stopPromise
	if (!runtimeScope) return Promise.resolve()
	const scope = runtimeScope
	stopPromise = (async () => {
		await startPromise?.catch(() => undefined)
		await Effect.runPromise(Scope.close(scope, Exit.void))
		runtime = undefined
		runtimeScope = undefined
	})().finally(() => {
		stopPromise = undefined
	})
	return stopPromise
}