import os from 'node:os'
import { SubfsExpansionError } from '@wispplace/atproto-utils'
import { revalidationQuarantineKey, revalidationSiteVersionKey } from '@wispplace/constants'
import { createLogger } from '@wispplace/observability'
import { SafeFetchHttpError } from '@wispplace/safe-fetch'
import { DecompressionLimitError } from '@wispplace/tiered-storage'
import Redis from 'ioredis'
import { config } from '../config'
import {
	AuthoritativeSettingsRecordError,
	AuthoritativeSiteRecordError,
	FileLogicalSizeLimitError,
	fetchSettingsRecord,
	fetchSettingsRecordOutcome,
	fetchSiteRecord,
	fetchSiteRecordOutcome,
	handleSettingsDelete,
	handleSettingsUpdate,
	handleSiteCreateOrUpdate,
	handleSiteDelete,
	SiteBlobBackoffError,
	SiteLogicalQuotaExceededError,
} from './cache-writer'
import {
	isSettingsFailureRevalidationReason,
	isSiteDeleteTombstoneReason,
	SETTINGS_DELETE_FAILURE_REASON,
} from './revalidate-queue'
import {
	assertRevalidationActive,
	createRevalidationResourceContext,
	RevalidationDeadlineError,
	type RevalidationResourceContext,
	TransferBudgetExceededError,
} from './revalidate-resources'

export type { RevalidationResourceContext, TransferByteBudgetLike } from './revalidate-resources'
export {
	createRevalidationResourceContext,
	RevalidationDeadlineError,
	TransferBudgetExceededError,
	TransferByteBudget,
} from './revalidate-resources'

const logger = createLogger('firehose-service')
const consumerName = process.env.WISP_REVALIDATE_CONSUMER || `${os.hostname()}:${process.pid}`

const DEFAULT_BATCH_SIZE = 10
const MAX_BATCH_SIZE = 100
const DEFAULT_CLAIM_IDLE_MS = 60_000
const MIN_CLAIM_IDLE_MS = 1_000
const MAX_CLAIM_IDLE_MS = 60 * 60 * 1000
const DEFAULT_BLOCK_MS = 5_000
const MIN_BLOCK_MS = 1_000
const MAX_BLOCK_MS = 60_000
const DEFAULT_BLOCKING_GRACE_MS = 1_000
const MIN_BLOCKING_GRACE_MS = 100
const MAX_BLOCKING_GRACE_MS = 10_000
const MIN_SOCKET_TIMEOUT_EXTRA_MS = 1_000
const DEFAULT_SOCKET_TIMEOUT_EXTRA_MS = 10_000
const MAX_SOCKET_TIMEOUT_MS = 5 * 60 * 1000
const DEFAULT_FAILURE_BACKOFF_SECONDS = 600
const MAX_FAILURE_BACKOFF_SECONDS = 24 * 60 * 60
const DEFAULT_RECONNECT_MIN_MS = 250
const DEFAULT_RECONNECT_MAX_MS = 30_000
const MAX_RECONNECT_MS = 5 * 60 * 1000

/** Strict poison-message/resource defaults. */
export const DEFAULT_REVALIDATE_MAX_ATTEMPTS = 3
export const DEFAULT_REVALIDATE_DEADLINE_MS = 5 * 60 * 1000
export const DEFAULT_REVALIDATE_TRANSFER_BUDGET_BYTES = 1024 * 1024 * 1024
export const DEFAULT_REVALIDATE_RETRY_BACKOFF_BASE_MS = 60_000
export const DEFAULT_REVALIDATE_RETRY_BACKOFF_MAX_MS = 15 * 60 * 1000
const MAX_REVALIDATE_ATTEMPTS = 20
const MAX_REVALIDATE_DEADLINE_MS = 30 * 60 * 1000
const MAX_REVALIDATE_TRANSFER_BUDGET_BYTES = 10 * 1024 * 1024 * 1024
const MAX_REVALIDATE_BACKOFF_MS = 60 * 60 * 1000

export interface RevalidateWorkerRuntimeConfig {
	batchSize: number
	claimIdleMs: number
	blockMs: number
	blockingGraceMs: number
	socketTimeoutMs: number
	failureBackoffSeconds: number
	reconnectMinMs: number
	reconnectMaxMs: number
	/** Maximum delivery attempts before quarantine. */
	maxAttempts?: number
	/** Per-message wall-clock deadline. */
	revalidationDeadlineMs?: number
	/** Shared inbound transfer-byte budget. */
	transferBudgetBytes?: number
	/** Capped exponential retry delay parameters. */
	retryBackoffBaseMs?: number
	retryBackoffMaxMs?: number
	/** Compatibility aliases for isolated callers. */
	wallDeadlineMs?: number
	maxTransferBytes?: number
	maxDeliveryAttempts?: number
	transferByteBudgetBytes?: number
	retryBaseMs?: number
	retryMaxMs?: number
}

export interface RevalidateWorkerRedisOptions {
	maxRetriesPerRequest: number
	enableReadyCheck: boolean
	blockingTimeout: number
	blockingTimeoutGrace: number
	socketTimeout: number
}

export type RevalidateWorkerRedisFactory = (redisUrl: string, options: RevalidateWorkerRedisOptions) => Redis

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

export function resolveRevalidateWorkerRuntimeConfig(
	environment: Record<string, string | undefined> = process.env,
): RevalidateWorkerRuntimeConfig {
	const batchSize = parseBoundedPositiveInt(
		environment.WISP_REVALIDATE_BATCH_SIZE,
		DEFAULT_BATCH_SIZE,
		1,
		MAX_BATCH_SIZE,
	)
	const claimIdleMs = parseBoundedPositiveInt(
		environment.WISP_REVALIDATE_CLAIM_IDLE_MS,
		DEFAULT_CLAIM_IDLE_MS,
		MIN_CLAIM_IDLE_MS,
		MAX_CLAIM_IDLE_MS,
	)
	const blockMs = parseBoundedPositiveInt(
		environment.WISP_REVALIDATE_BLOCK_MS,
		DEFAULT_BLOCK_MS,
		MIN_BLOCK_MS,
		MAX_BLOCK_MS,
	)
	const blockingGraceMs = parseBoundedPositiveInt(
		environment.WISP_REVALIDATE_BLOCKING_GRACE_MS,
		DEFAULT_BLOCKING_GRACE_MS,
		MIN_BLOCKING_GRACE_MS,
		MAX_BLOCKING_GRACE_MS,
	)
	const minimumSocketTimeoutMs = blockMs + blockingGraceMs + MIN_SOCKET_TIMEOUT_EXTRA_MS
	const socketTimeoutMs = Math.max(
		parseBoundedPositiveInt(
			environment.WISP_REVALIDATE_SOCKET_TIMEOUT_MS,
			blockMs + blockingGraceMs + DEFAULT_SOCKET_TIMEOUT_EXTRA_MS,
			1,
			MAX_SOCKET_TIMEOUT_MS,
		),
		minimumSocketTimeoutMs,
	)
	const failureBackoffSeconds = parseBoundedPositiveInt(
		environment.WISP_REVALIDATE_FAILURE_BACKOFF_SECONDS,
		DEFAULT_FAILURE_BACKOFF_SECONDS,
		1,
		MAX_FAILURE_BACKOFF_SECONDS,
	)
	const reconnectMinMs = parseBoundedPositiveInt(
		environment.WISP_REVALIDATE_RECONNECT_MIN_MS,
		DEFAULT_RECONNECT_MIN_MS,
		10,
		MAX_RECONNECT_MS,
	)
	const reconnectMaxMs = Math.max(
		reconnectMinMs,
		parseBoundedPositiveInt(
			environment.WISP_REVALIDATE_RECONNECT_MAX_MS,
			DEFAULT_RECONNECT_MAX_MS,
			10,
			MAX_RECONNECT_MS,
		),
	)
	const maxAttempts = parseBoundedPositiveInt(
		environment.WISP_REVALIDATE_MAX_ATTEMPTS ??
			environment.WISP_REVALIDATE_MAX_DELIVERIES ??
			environment.FIREHOSE_REVALIDATE_MAX_ATTEMPTS,
		DEFAULT_REVALIDATE_MAX_ATTEMPTS,
		1,
		MAX_REVALIDATE_ATTEMPTS,
	)
	const revalidationDeadlineMs = parseBoundedPositiveInt(
		environment.WISP_REVALIDATE_DEADLINE_MS ??
			environment.WISP_REVALIDATE_WALL_DEADLINE_MS ??
			environment.FIREHOSE_REVALIDATION_DEADLINE_MS,
		DEFAULT_REVALIDATE_DEADLINE_MS,
		1_000,
		MAX_REVALIDATE_DEADLINE_MS,
	)
	const transferBudgetBytes = parseBoundedPositiveInt(
		environment.WISP_REVALIDATE_TRANSFER_BUDGET_BYTES ??
			environment.WISP_REVALIDATE_TRANSFER_BUDGET ??
			environment.FIREHOSE_REVALIDATION_TRANSFER_BUDGET_BYTES,
		DEFAULT_REVALIDATE_TRANSFER_BUDGET_BYTES,
		1,
		MAX_REVALIDATE_TRANSFER_BUDGET_BYTES,
	)
	const retryBackoffBaseMs = parseBoundedPositiveInt(
		environment.WISP_REVALIDATE_RETRY_BACKOFF_BASE_MS ??
			environment.WISP_REVALIDATE_RETRY_BASE_MS ??
			environment.FIREHOSE_REVALIDATE_RETRY_BASE_MS,
		DEFAULT_REVALIDATE_RETRY_BACKOFF_BASE_MS,
		100,
		MAX_REVALIDATE_BACKOFF_MS,
	)
	const retryBackoffMaxMs = Math.max(
		retryBackoffBaseMs,
		parseBoundedPositiveInt(
			environment.WISP_REVALIDATE_RETRY_BACKOFF_MAX_MS ??
				environment.WISP_REVALIDATE_RETRY_MAX_MS ??
				environment.FIREHOSE_REVALIDATE_RETRY_MAX_MS,
			DEFAULT_REVALIDATE_RETRY_BACKOFF_MAX_MS,
			retryBackoffBaseMs,
			MAX_REVALIDATE_BACKOFF_MS,
		),
	)

	// Keep the legacy enumerable shape stable for existing callers while making
	// the strict policy available as ordinary properties. This also avoids
	// accidentally logging the new resource values as part of old diagnostics.
	const result: RevalidateWorkerRuntimeConfig = {
		batchSize,
		claimIdleMs,
		blockMs,
		blockingGraceMs,
		socketTimeoutMs,
		failureBackoffSeconds,
		reconnectMinMs,
		reconnectMaxMs,
	}
	Object.defineProperties(result, {
		maxAttempts: { value: maxAttempts, enumerable: false },
		revalidationDeadlineMs: { value: revalidationDeadlineMs, enumerable: false },
		transferBudgetBytes: { value: transferBudgetBytes, enumerable: false },
		retryBackoffBaseMs: { value: retryBackoffBaseMs, enumerable: false },
		retryBackoffMaxMs: { value: retryBackoffMaxMs, enumerable: false },
		// Friendly aliases accepted by older isolated test harnesses.
		wallDeadlineMs: { value: revalidationDeadlineMs, enumerable: false },
		maxTransferBytes: { value: transferBudgetBytes, enumerable: false },
		maxDeliveryAttempts: { value: maxAttempts, enumerable: false },
		transferByteBudgetBytes: { value: transferBudgetBytes, enumerable: false },
		retryBaseMs: { value: retryBackoffBaseMs, enumerable: false },
		retryMaxMs: { value: retryBackoffMaxMs, enumerable: false },
	})
	return result
}

export function createRevalidateWorkerRedisOptions(
	runtimeConfig: RevalidateWorkerRuntimeConfig,
): RevalidateWorkerRedisOptions {
	return {
		maxRetriesPerRequest: 2,
		enableReadyCheck: true,
		// A normal XREADGROUP BLOCK response lands before this client-side bound.
		// The grace lets ioredis resolve a half-open blocking command instead of
		// leaving the worker (and shutdown) hung forever.
		blockingTimeout: runtimeConfig.blockMs + runtimeConfig.blockingGraceMs,
		blockingTimeoutGrace: runtimeConfig.blockingGraceMs,
		socketTimeout: Math.max(
			runtimeConfig.socketTimeoutMs,
			runtimeConfig.blockMs + runtimeConfig.blockingGraceMs + MIN_SOCKET_TIMEOUT_EXTRA_MS,
		),
	}
}

const revalidateWorkerRuntimeConfig = resolveRevalidateWorkerRuntimeConfig()
const failureBackoffSeconds = revalidateWorkerRuntimeConfig.failureBackoffSeconds

interface RevalidatePolicyConfig {
	maxAttempts: number
	revalidationDeadlineMs: number
	transferBudgetBytes: number
	retryBackoffBaseMs: number
	retryBackoffMaxMs: number
}

function policyConfig(runtime: RevalidateWorkerRuntimeConfig): RevalidatePolicyConfig {
	const values = runtime as RevalidateWorkerRuntimeConfig & {
		wallDeadlineMs?: number
		maxTransferBytes?: number
		maxDeliveryAttempts?: number
		transferByteBudgetBytes?: number
		retryBaseMs?: number
		retryMaxMs?: number
	}
	const maxAttempts = values.maxAttempts ?? values.maxDeliveryAttempts ?? DEFAULT_REVALIDATE_MAX_ATTEMPTS
	const revalidationDeadlineMs =
		values.revalidationDeadlineMs ?? values.wallDeadlineMs ?? DEFAULT_REVALIDATE_DEADLINE_MS
	const transferBudgetBytes =
		values.transferBudgetBytes ??
		values.maxTransferBytes ??
		values.transferByteBudgetBytes ??
		DEFAULT_REVALIDATE_TRANSFER_BUDGET_BYTES
	const retryBackoffBaseMs = values.retryBackoffBaseMs ?? values.retryBaseMs ?? DEFAULT_REVALIDATE_RETRY_BACKOFF_BASE_MS
	const retryBackoffMaxMs = Math.max(
		values.retryBackoffMaxMs ?? values.retryMaxMs ?? DEFAULT_REVALIDATE_RETRY_BACKOFF_MAX_MS,
		retryBackoffBaseMs,
	)
	return { maxAttempts, revalidationDeadlineMs, transferBudgetBytes, retryBackoffBaseMs, retryBackoffMaxMs }
}

export function revalidateRetryBackoffMs(
	attempt: number,
	baseMs = DEFAULT_REVALIDATE_RETRY_BACKOFF_BASE_MS,
	maxMs = DEFAULT_REVALIDATE_RETRY_BACKOFF_MAX_MS,
): number {
	if (!Number.isSafeInteger(attempt) || attempt < 1)
		throw new RangeError('Revalidation attempt must be a positive safe integer')
	if (!Number.isSafeInteger(baseMs) || baseMs < 1 || !Number.isSafeInteger(maxMs) || maxMs < baseMs) {
		throw new RangeError('Revalidation retry backoff bounds are invalid')
	}
	return Math.min(maxMs, baseMs * 2 ** Math.min(attempt - 1, 30))
}

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

let redis: Redis | null = null
let running = false
let loopPromise: Promise<void> | null = null
let cancelLoopRetryWait: (() => void) | null = null
let workerAbortController: AbortController | null = null
let workerGeneration = 0

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

function isActiveWorkerClient(client: Redis): boolean {
	return running && redis === client
}

function getFailureBackoffKey(did: string, rkey: string): string {
	return `revalidate:site:failure-backoff:${did}:${rkey}`
}

function getRetrySuppressionKey(id: string): string {
	return `revalidate:retry:${id}`
}

function parseFields(raw: string[]): Record<string, string> {
	const fields: Record<string, string> = {}
	for (let i = 0; i < raw.length; i += 2) {
		const key = raw[i]
		const value = raw[i + 1]
		if (key) {
			fields[key] = value ?? ''
		}
	}
	return fields
}

export function shouldSkipInvalidationForReason(reason: string): boolean {
	// Rewrite repairs only repopulate `.rewritten/*` HTML variants. They should not
	// flip the whole site into "updating" while the original files remain serveable.
	return reason.startsWith('rewrite-miss')
}

export interface RevalidateWorkerDependencies {
	fetchSettingsRecord: typeof fetchSettingsRecord
	fetchSettingsRecordOutcome?: typeof fetchSettingsRecordOutcome
	fetchSiteRecord: typeof fetchSiteRecord
	fetchSiteRecordOutcome?: typeof fetchSiteRecordOutcome
	handleSettingsDelete: typeof handleSettingsDelete
	handleSettingsUpdate: typeof handleSettingsUpdate
	handleSiteCreateOrUpdate: typeof handleSiteCreateOrUpdate
	handleSiteDelete: typeof handleSiteDelete
}

export interface RevalidateRedisPipeline {
	ttl(key: string): RevalidateRedisPipeline
	exec(): PromiseLike<unknown>
}

export interface RevalidateRedisClient {
	ttl(key: string): PromiseLike<number>
	/** Durable DLQ fence lookup; production ioredis always exposes GET. */
	get?(key: string): PromiseLike<string | null>
	/** Queue commands and execute them in one Redis round trip. */
	pipeline?(): RevalidateRedisPipeline
	/** Atomically XACK then XDEL for the sole documented consumer group. */
	eval(script: string, keyCount: number, ...args: string[]): PromiseLike<unknown>
	set(key: string, value: string, expirationMode: 'EX', ttlSeconds: number): PromiseLike<'OK' | null>
	/** Redis XPENDING exact-id lookup; unavailable only in minimal test seams. */
	xpending?(...args: (string | number)[]): PromiseLike<unknown>
}

export type RevalidationFailureClass = 'permanent' | 'transient'

export interface RevalidationFailureInfo {
	classification: RevalidationFailureClass
	code: string
}

/** Stable, bounded failure used for typed PDS outcomes. */
export class RevalidationProcessingError extends Error {
	constructor(
		readonly code: string,
		readonly classification: RevalidationFailureClass,
	) {
		super(`Revalidation failed: ${code}`)
		this.name = 'RevalidationProcessingError'
	}
}

const COMPLETE_REVALIDATION_SCRIPT = `
local acknowledged = redis.call('XACK', KEYS[1], ARGV[1], ARGV[2])
if acknowledged ~= 1 then return {acknowledged, 0} end
local deleted = redis.call('XDEL', KEYS[1], ARGV[2])
return {acknowledged, deleted}
`

/**
 * Write the DLQ record before touching the source entry. Redis executes the
 * whole script atomically, so a successful XACK/XDEL can never precede the
 * quarantine write.
 */
export const QUARANTINE_REVALIDATION_SCRIPT = `
local dlqId = redis.call('XADD', KEYS[2], 'MAXLEN', '~', ARGV[11], '*',
  'sourceId', ARGV[2],
  'did', ARGV[3],
  'rkey', ARGV[4],
  'reason', ARGV[5],
  'errorCode', ARGV[6],
  'error', ARGV[7],
  'classification', ARGV[9],
  'attempts', ARGV[8],
  'quarantinedAt', ARGV[10])
local fenceVersion = ARGV[12]
if fenceVersion == '' then fenceVersion = redis.call('GET', KEYS[4]) or '' end
redis.call('SET', KEYS[3], fenceVersion)
local acknowledged = redis.call('XACK', KEYS[1], ARGV[1], ARGV[2])
if acknowledged ~= 1 then return {acknowledged, dlqId, 0} end
local deleted = redis.call('XDEL', KEYS[1], ARGV[2])
return {acknowledged, dlqId, deleted}
`

// Legacy cleanup is deliberately much stricter than producer capacity: every
// consumer group must be entirely drained. XINFO, the safety checks, and XTRIM
// execute atomically, so a producer cannot add an entry between observing lag
// zero and trimming. Unknown or old groups fail closed instead of losing entries.
const TRIM_DRAINED_REVALIDATION_SCRIPT = `
local groups = redis.call('XINFO', 'GROUPS', KEYS[1])
local configuredFound = false
for _, group in ipairs(groups) do
  local name = nil
  local pending = nil
  local lag = nil
  for i = 1, #group, 2 do
    if group[i] == 'name' then name = group[i + 1] end
    if group[i] == 'pending' then pending = tonumber(group[i + 1]) end
    if group[i] == 'lag' then lag = tonumber(group[i + 1]) end
  end
  if not name or pending == nil or lag == nil then return 0 end
  if name == ARGV[1] then configuredFound = true end
  if pending ~= 0 or lag ~= 0 then return 0 end
end
if not configuredFound then return 0 end
return redis.call('XTRIM', KEYS[1], 'MAXLEN', '=', 0)
`

const defaultRevalidateWorkerDependencies: RevalidateWorkerDependencies = {
	fetchSettingsRecord,
	fetchSettingsRecordOutcome,
	fetchSiteRecord,
	fetchSiteRecordOutcome,
	handleSettingsDelete,
	handleSettingsUpdate,
	handleSiteCreateOrUpdate,
	handleSiteDelete,
}

/**
 * Complete an entry in one Redis script. If its response is lost, Redis has
 * either not run it (the entry is pending) or run both XACK and XDEL; an
 * acknowledged orphan cannot consume the producer capacity limit.
 */
async function acknowledgeCompletedMessage(redisClient: RevalidateRedisClient, id: string): Promise<void> {
	const result = await redisClient.eval(
		COMPLETE_REVALIDATION_SCRIPT,
		1,
		config.revalidateStream,
		config.revalidateGroup,
		id,
	)
	if (!Array.isArray(result) || typeof result[0] !== 'number' || typeof result[1] !== 'number') {
		throw new Error(`Revalidate completion script returned malformed result for ${id}`)
	}
	const [acknowledged, deleted] = result
	if (acknowledged !== 1) {
		logger.warn(`[Revalidate] Completion invariant failed for ${id}: expected XACK 1, got ${acknowledged}`)
		return
	}
	if (deleted !== 1) throw new Error(`Revalidate completion script failed to XDEL ${id}: got ${deleted}`)
}

/** Best-effort cleanup for old acknowledged-orphan entries from pre-atomic releases. */
export async function trimDrainedRevalidationEntries(redisClient: Pick<RevalidateRedisClient, 'eval'>): Promise<void> {
	try {
		const trimmed = await redisClient.eval(
			TRIM_DRAINED_REVALIDATION_SCRIPT,
			1,
			config.revalidateStream,
			config.revalidateGroup,
		)
		if (typeof trimmed !== 'number' || trimmed < 0) {
			logger.warn('[Revalidate] Drained-stream maintenance returned an invalid result')
		}
	} catch (error) {
		// Older Redis versions may not expose group lag. Maintenance is optional;
		// never substitute a risky fallback trim.
		logger.warn(`[Revalidate] Drained-stream maintenance skipped (${errorKind(error)})`)
	}
}

export interface ParsedRevalidationMessage {
	fields: Record<string, string>
	did?: string
	rkey?: string
	reason: string
	sourceVersion: string
}

function hasPermanentPdsCode(code: string): boolean {
	return code === 'INVALID_RECORD' || code === 'MISSING_CID'
}

/** Classify without echoing hostile response bodies or URLs. */
export function classifyRevalidationError(error: unknown): RevalidationFailureInfo {
	if (error instanceof RevalidationProcessingError) {
		return { classification: error.classification, code: error.code }
	}
	if (error instanceof SiteBlobBackoffError) return { classification: 'transient', code: 'BLOB_500_BACKOFF' }
	if (error instanceof SubfsExpansionError) {
		const permanent = error.code !== 'FETCH_FAILED'
		return { classification: permanent ? 'permanent' : 'transient', code: error.code }
	}
	if (error instanceof AuthoritativeSiteRecordError) {
		return { classification: hasPermanentPdsCode(error.code) ? 'permanent' : 'transient', code: error.code }
	}
	if (error instanceof AuthoritativeSettingsRecordError) {
		return { classification: hasPermanentPdsCode(error.code) ? 'permanent' : 'transient', code: error.code }
	}
	if (error instanceof SafeFetchHttpError) {
		const retryable = error.status === 408 || error.status === 425 || error.status === 429 || error.status >= 500
		return { classification: retryable ? 'transient' : 'permanent', code: `HTTP_${error.status}` }
	}
	if (error instanceof SiteLogicalQuotaExceededError) return { classification: 'permanent', code: 'SITE_LOGICAL_QUOTA' }
	if (error instanceof FileLogicalSizeLimitError) return { classification: 'permanent', code: 'FILE_LOGICAL_SIZE' }
	if (error instanceof DecompressionLimitError) return { classification: 'permanent', code: 'DECOMPRESSION_LIMIT' }
	if (error instanceof TransferBudgetExceededError) return { classification: 'transient', code: error.code }
	if (error instanceof RevalidationDeadlineError) return { classification: 'transient', code: error.code }
	const value = error as { code?: unknown; status?: unknown; statusCode?: unknown }
	if (typeof value?.status === 'number' || typeof value?.statusCode === 'number') {
		const status = Number(value.status ?? value.statusCode)
		if (Number.isSafeInteger(status) && status >= 400 && status <= 599) {
			return {
				classification: status === 408 || status === 425 || status === 429 || status >= 500 ? 'transient' : 'permanent',
				code: `HTTP_${status}`,
			}
		}
	}
	if (typeof value?.code === 'string' && /^[A-Z][A-Z0-9_-]{1,63}$/.test(value.code)) {
		const permanent = ['INVALID_RECORD', 'MISSING_CID', 'INVALID_SUBJECT', 'CYCLE', 'DUPLICATE_PATH'].includes(
			value.code,
		)
		return { classification: permanent ? 'permanent' : 'transient', code: value.code }
	}
	const message = error instanceof Error ? error.message : String(error)
	if (/marked gzip|valid gzip|logical size|exceeds .*limit|invalid (?:record|subject)/i.test(message)) {
		return { classification: 'permanent', code: 'INVALID_PAYLOAD' }
	}
	return {
		classification: 'transient',
		code:
			errorKind(error)
				.replace(/[^A-Za-z0-9_-]/g, '_')
				.slice(0, 64) || 'UNKNOWN',
	}
}

function removeControlCharacters(value: string): string {
	return Array.from(value, (character) => {
		const code = character.charCodeAt(0)
		return code <= 0x1f || code === 0x7f ? ' ' : character
	}).join('')
}

function boundedFailureText(error: unknown): string {
	const message = error instanceof Error ? error.message : String(error)
	return removeControlCharacters(message).slice(0, 512)
}

function boundedField(value: string | undefined, maximum = 2048): string {
	return removeControlCharacters(value ?? '').slice(0, maximum)
}

export interface RevalidateMessagePolicyOptions {
	/** Redis delivery count; `attempt` is accepted as a compatibility alias. */
	deliveryAttempt?: number
	attempt?: number
	runtimeConfig?: RevalidateWorkerRuntimeConfig
	/** Supply a context in tests or when composing a larger operation. */
	resourceContext?: RevalidationResourceContext
	/** Abort active work when the owning worker lifecycle is stopping. */
	upstreamSignal?: AbortSignal
	/** Internal worker calls set this to enforce bounded processing. */
	enforceAttemptPolicy?: boolean
}

export async function quarantineRevalidationMessage(
	redisClient: RevalidateRedisClient,
	message: ParsedRevalidationMessage,
	id: string,
	error: unknown,
	attempt: number,
	classification: RevalidationFailureClass,
	code?: string,
): Promise<void> {
	const resolvedCode = code ?? classifyRevalidationError(error).code
	const result = await redisClient.eval(
		QUARANTINE_REVALIDATION_SCRIPT,
		4,
		config.revalidateStream,
		config.revalidateDlqStream,
		revalidationQuarantineKey(message.did ?? '', message.rkey ?? ''),
		revalidationSiteVersionKey(message.did ?? '', message.rkey ?? ''),
		config.revalidateGroup,
		id,
		boundedField(message.did),
		boundedField(message.rkey),
		boundedField(message.reason),
		boundedField(resolvedCode, 128),
		boundedFailureText(error),
		String(Math.max(1, attempt)),
		classification,
		Date.now().toString(),
		String(config.revalidateDlqStreamMaxLen),
		boundedField(message.sourceVersion),
	)
	if (
		!Array.isArray(result) ||
		typeof result[0] !== 'number' ||
		typeof result[1] !== 'string' ||
		typeof result[2] !== 'number'
	) {
		throw new Error(`Revalidate quarantine script returned malformed result for ${id}`)
	}
	if (result[0] !== 1 || result[2] !== 1) {
		throw new Error(`Revalidate quarantine did not ACK and delete ${id}`)
	}
	logger.warn(`[Revalidate] Quarantined ${id} after ${attempt} attempt(s)`, {
		did: message.did,
		rkey: message.rkey,
		reason: message.reason,
		errorCode: resolvedCode,
		classification,
		dlqId: result[1],
	})
}

type VerifiedSiteRecord = NonNullable<Awaited<ReturnType<typeof fetchSiteRecord>>>

interface PreparedSiteRevalidation {
	did: string
	rkey: string
	reason: string
	isDeleteTombstone: boolean
	failureBackoffKey: string
	record: VerifiedSiteRecord
}

function parseRevalidationMessage(rawFields: string[]): ParsedRevalidationMessage {
	const fields = parseFields(rawFields)
	return {
		fields,
		did: fields.did,
		rkey: fields.rkey,
		reason: fields.reason || 'storage-miss',
		sourceVersion: fields.sourceVersion || '',
	}
}

async function isSiteRepairQuarantined(
	redisClient: RevalidateRedisClient,
	message: ParsedRevalidationMessage,
): Promise<boolean> {
	if (!message.did || !message.rkey || !redisClient.get) return false
	if (isSettingsFailureRevalidationReason(message.reason)) return false
	const fenceVersion = await redisClient.get(revalidationQuarantineKey(message.did, message.rkey))
	if (fenceVersion === null) return false
	return message.sourceVersion === '' || message.sourceVersion <= fenceVersion
}

async function setRetrySuppressionTtl(
	redisClient: Pick<RevalidateRedisClient, 'set'>,
	id: string,
	ttlSeconds: number,
): Promise<void> {
	const suppressionTtlSeconds = Math.max(1, Math.ceil(ttlSeconds))
	if (!Number.isSafeInteger(suppressionTtlSeconds)) {
		throw new Error(`Invalid retry suppression TTL for ${id}`)
	}
	const result = await redisClient.set(getRetrySuppressionKey(id), '1', 'EX', suppressionTtlSeconds)
	if (result !== 'OK') throw new Error(`Failed to persist retry suppression for ${id}`)
}

async function reconcileSettingsRevalidation(
	did: string,
	rkey: string,
	dependencies: RevalidateWorkerDependencies,
	resources: RevalidationResourceContext | undefined,
	strict: boolean,
	isDelete: boolean,
): Promise<boolean> {
	// Settings events are reconciled from the current PDS state. This makes an
	// old failed delete safe if a settings record has since reappeared, and makes
	// an old failed update safe if the record has since been deleted. The lookup
	// is handed to the locked writer rather than performed here: a pre-lock
	// record can become stale while waiting for the advisory lock, and doing both
	// lookups would also charge the same revalidation twice.
	assertRevalidationActive(resources)
	const fetchCurrentSettings: typeof fetchSettingsRecordOutcome =
		dependencies.fetchSettingsRecordOutcome ??
		(async (lookupDid, lookupRkey, pdsEndpoint, lookupResources) => {
			const record = await dependencies.fetchSettingsRecord(lookupDid, lookupRkey, pdsEndpoint, lookupResources)
			return record ? { kind: 'present' as const, ...record } : { kind: 'absent' as const }
		})

	try {
		// Both wrappers reconcile against the current PDS record under the lock.
		// Undefined hints are deliberate: only that post-lock lookup is authoritative.
		if (isDelete) {
			await dependencies.handleSettingsDelete(did, rkey, undefined, resources, {
				resources,
				fetchSettingsRecordOutcome: fetchCurrentSettings,
			})
		} else {
			await dependencies.handleSettingsUpdate(did, rkey, undefined, undefined, {
				resources,
				fetchSettingsRecordOutcome: fetchCurrentSettings,
			})
		}
		assertRevalidationActive(resources)
		return true
	} catch (error) {
		if (!(error instanceof AuthoritativeSettingsRecordError)) throw error
		logger.warn(`[Revalidate] Settings PDS lookup remains retryable for ${did}/${rkey}`, {
			outcome: error.code,
		})
		if (strict)
			throw new RevalidationProcessingError(error.code, hasPermanentPdsCode(error.code) ? 'permanent' : 'transient')
		return false
	}
}

async function handleMissingSiteRecord(
	id: string,
	did: string,
	rkey: string,
	isDeleteTombstone: boolean,
	redisClient: RevalidateRedisClient,
	dependencies: RevalidateWorkerDependencies,
	resources?: RevalidationResourceContext,
): Promise<void> {
	if (!isDeleteTombstone) {
		logger.warn(`[Revalidate] Site record not found on PDS: ${did}/${rkey}`)
		assertRevalidationActive(resources)
		await acknowledgeCompletedMessage(redisClient, id)
		return
	}

	// A failed delete must remain pending until its cache/DB cleanup actually
	// succeeds. `handleSiteDelete` is idempotent, so retrying a partial delete
	// is safe.
	logger.info(`[Revalidate] Applying delete tombstone ${id}: ${did}/${rkey}`)
	await dependencies.handleSiteDelete(did, rkey, undefined, resources)
	assertRevalidationActive(resources)
	logger.info(`[Revalidate] Completed delete tombstone ${id}: ${did}/${rkey}`)
	await acknowledgeCompletedMessage(redisClient, id)
}

async function prepareSiteRevalidation(
	id: string,
	did: string,
	rkey: string,
	reason: string,
	redisClient: RevalidateRedisClient,
	dependencies: RevalidateWorkerDependencies,
	resources: RevalidationResourceContext | undefined,
	strict: boolean,
): Promise<PreparedSiteRevalidation | null> {
	const isDeleteTombstone = isSiteDeleteTombstoneReason(reason)
	const failureBackoffKey = getFailureBackoffKey(did, rkey)
	assertRevalidationActive(resources)
	const activeBackoffTtl = await redisClient.ttl(failureBackoffKey)
	assertRevalidationActive(resources)
	if (activeBackoffTtl > 0 && !isDeleteTombstone) {
		// A storage-miss is a durable repair request. Do not ACK it merely because
		// a previous blob 500 installed a site backoff; XAUTOCLAIM must revisit it.
		if (strict || reason.startsWith('storage-miss')) {
			await setRetrySuppressionTtl(redisClient, id, activeBackoffTtl)
			logger.info(
				`[Revalidate] Keeping ${id} pending: ${did}/${rkey} site backoff active (${activeBackoffTtl}s remaining)`,
			)
			return null
		}
		logger.info(`[Revalidate] Acking ${id}: ${did}/${rkey} site backoff active (${activeBackoffTtl}s remaining)`)
		await acknowledgeCompletedMessage(redisClient, id)
		return null
	}

	const outcome = await (dependencies.fetchSiteRecordOutcome
		? dependencies.fetchSiteRecordOutcome(did, rkey, resources)
		: (() => {
				// Compatibility seam for isolated tests; production always supplies the
				// typed outcome API below.
				return dependencies
					.fetchSiteRecord(did, rkey, resources)
					.then((record) => (record ? { kind: 'present' as const, ...record } : { kind: 'absent' as const }))
			})())
	assertRevalidationActive(resources)
	if (outcome.kind === 'retryable') {
		logger.warn(`[Revalidate] PDS lookup remains retryable for ${did}/${rkey}`, { outcome: outcome.error })
		if (strict)
			throw new RevalidationProcessingError(
				outcome.error,
				hasPermanentPdsCode(outcome.error) ? 'permanent' : 'transient',
			)
		return null
	}
	if (outcome.kind === 'absent') {
		await handleMissingSiteRecord(id, did, rkey, isDeleteTombstone, redisClient, dependencies, resources)
		return null
	}
	const record = { record: outcome.record, cid: outcome.cid }

	if (isDeleteTombstone && activeBackoffTtl > 0) {
		// Still look for an absent record above so a real delete is not hidden by a
		// stale blob backoff. If the record has reappeared, preserve this pending
		// tombstone until the backoff expires rather than ACKing it and losing the
		// required current-record materialization.
		await setRetrySuppressionTtl(redisClient, id, activeBackoffTtl)
		logger.info(
			`[Revalidate] Keeping delete tombstone ${id} pending: ${did}/${rkey} site backoff active (${activeBackoffTtl}s remaining)`,
		)
		return null
	}

	return { did, rkey, reason, isDeleteTombstone, failureBackoffKey, record }
}

async function handleSiteMaterializationFailure(
	id: string,
	state: PreparedSiteRevalidation,
	error: unknown,
	redisClient: RevalidateRedisClient,
	strict: boolean,
	resources?: RevalidationResourceContext,
): Promise<void> {
	if (!(error instanceof SiteBlobBackoffError)) throw error

	const now = Date.now()
	const until = Math.max(error.until, now + 1000)
	const ttlSeconds = Math.max(failureBackoffSeconds, Math.ceil((until - now) / 1000))
	assertRevalidationActive(resources)
	await redisClient.set(state.failureBackoffKey, until.toString(), 'EX', ttlSeconds)
	assertRevalidationActive(resources)

	if (state.isDeleteTombstone || strict || state.reason.startsWith('storage-miss')) {
		// Do not ACK a durable repair whose replacement could not be materialized.
		// Leaving it pending lets the bounded worker retry or quarantine it. The
		// per-message key keeps JUSTID scans from turning this wait into deliveries.
		await setRetrySuppressionTtl(redisClient, id, ttlSeconds)
		logger.warn(`[Revalidate] Blob backoff for ${state.did}/${state.rkey}; keeping ${id} pending`, {
			did: state.did,
			rkey: state.rkey,
			failures: error.failures,
			backoffUntil: new Date(until).toISOString(),
			ttlSeconds,
		})
		return
	}

	logger.warn(`[Revalidate] Blob backoff for ${state.did}/${state.rkey}; acking ${id} and suppressing retries`, {
		did: state.did,
		rkey: state.rkey,
		failures: error.failures,
		backoffUntil: new Date(until).toISOString(),
		ttlSeconds,
	})
	await acknowledgeCompletedMessage(redisClient, id)
}

async function materializeSiteRevalidation(
	id: string,
	state: PreparedSiteRevalidation,
	redisClient: RevalidateRedisClient,
	dependencies: RevalidateWorkerDependencies,
	resources: RevalidationResourceContext | undefined,
	strict: boolean,
): Promise<void> {
	// For storage-miss events, force re-download all files since storage is empty.
	// A failed delete can have removed some blobs before its DB cleanup failed, so
	// a reappeared tombstone also needs a full materialization rather than an
	// incremental diff against a potentially stale ledger.
	const forceDownload = state.isDeleteTombstone || state.reason.startsWith('storage-miss')
	const forceRewriteHtml = state.reason.startsWith('rewrite-miss')
	const skipInvalidation = shouldSkipInvalidationForReason(state.reason)

	try {
		assertRevalidationActive(resources)
		await dependencies.handleSiteCreateOrUpdate(state.did, state.rkey, state.record.record, state.record.cid, {
			skipInvalidation,
			forceDownload,
			forceRewriteHtml,
			resources,
		})
	} catch (error) {
		assertRevalidationActive(resources)
		await handleSiteMaterializationFailure(id, state, error, redisClient, strict, resources)
		return
	}

	assertRevalidationActive(resources)
	logger.info(`[Revalidate] Completed ${id}: ${state.did}/${state.rkey}`)
	await acknowledgeCompletedMessage(redisClient, id)
}

/** Process one stream entry under the optional strict resource/attempt policy. */
export async function processRevalidationMessage(
	id: string,
	rawFields: string[],
	redisClient: RevalidateRedisClient,
	dependencies: RevalidateWorkerDependencies = defaultRevalidateWorkerDependencies,
	options: RevalidateMessagePolicyOptions = {},
): Promise<void> {
	const message = parseRevalidationMessage(rawFields)
	const strict =
		options.enforceAttemptPolicy === true ||
		options.deliveryAttempt !== undefined ||
		options.attempt !== undefined ||
		options.runtimeConfig !== undefined
	const runtime = policyConfig(options.runtimeConfig ?? revalidateWorkerRuntimeConfig)
	const suppliedAttempt = options.deliveryAttempt ?? options.attempt
	const ownedResources = options.resourceContext === undefined
	const resources =
		options.resourceContext ??
		(isSettingsFailureRevalidationReason(message.reason)
			? createRevalidationResourceContext(
					runtime.revalidationDeadlineMs,
					runtime.transferBudgetBytes,
					options.upstreamSignal,
				)
			: // Filesystem repair is bounded by per-request limits, manifest file/size
				// admission, and lifecycle. Retries must not consume a second transfer quota.
				createRevalidationResourceContext(null, null, options.upstreamSignal))
	const lifecycleCancelled = () => options.upstreamSignal?.aborted === true

	if (lifecycleCancelled()) {
		if (ownedResources) resources.close()
		return
	}

	let attempt: number
	try {
		const measuredAttempt =
			suppliedAttempt ?? (strict ? await getDeliveryAttempt(redisClient as unknown as Redis, id) : 1)
		if (lifecycleCancelled()) {
			if (strict && suppliedAttempt === undefined) clearLocalDeliveryAttempt(id, measuredAttempt)
			if (ownedResources) resources.close()
			return
		}
		attempt = Math.max(1, Math.floor(measuredAttempt))
	} catch (error) {
		if (ownedResources) resources.close()
		throw error
	}

	try {
		if (lifecycleCancelled()) {
			if (strict && suppliedAttempt === undefined) clearLocalDeliveryAttempt(id, attempt)
			return
		}
		if (strict && locallyQuarantinedMessages.has(id)) return
		if (strict && attempt > runtime.maxAttempts) {
			await quarantineRevalidationMessage(
				redisClient,
				message,
				id,
				new RevalidationProcessingError('MAX_ATTEMPTS', 'permanent'),
				attempt,
				'permanent',
				'MAX_ATTEMPTS',
			)
			locallyQuarantinedMessages.add(id)
			if (locallyQuarantinedMessages.size > MAX_LOCAL_QUARANTINED_MESSAGES) {
				const first = locallyQuarantinedMessages.values().next().value as string | undefined
				if (first) locallyQuarantinedMessages.delete(first)
			}
			return
		}
		if (!message.did || !message.rkey) {
			logger.warn('[Revalidate] Missing did/rkey in message', { id, fields: message.fields })
			if (strict) {
				await quarantineRevalidationMessage(
					redisClient,
					message,
					id,
					new RevalidationProcessingError('MALFORMED_MESSAGE', 'permanent'),
					attempt,
					'permanent',
					'MALFORMED_MESSAGE',
				)
				locallyQuarantinedMessages.add(id)
			} else {
				await acknowledgeCompletedMessage(redisClient, id)
			}
			return
		}

		if (await isSiteRepairQuarantined(redisClient, message)) {
			logger.info(`[Revalidate] Dropping ${id}: site remains quarantined until a newer firehose event`, {
				did: message.did,
				rkey: message.rkey,
				reason: message.reason,
			})
			await acknowledgeCompletedMessage(redisClient, id)
			return
		}

		logger.info(`[Revalidate] Received message ${id}: ${message.did}/${message.rkey} (${message.reason})`)
		if (isSettingsFailureRevalidationReason(message.reason)) {
			if (
				await reconcileSettingsRevalidation(
					message.did,
					message.rkey,
					dependencies,
					resources,
					strict,
					message.reason === SETTINGS_DELETE_FAILURE_REASON,
				)
			) {
				assertRevalidationActive(resources)
				await acknowledgeCompletedMessage(redisClient, id)
			}
			return
		}

		const state = await prepareSiteRevalidation(
			id,
			message.did,
			message.rkey,
			message.reason,
			redisClient,
			dependencies,
			resources,
			strict,
		)
		if (!state) return
		await materializeSiteRevalidation(id, state, redisClient, dependencies, resources, strict)
	} catch (error) {
		// A lifecycle stop is not a delivery attempt. Do not ACK, quarantine, or
		// install a retry key after the worker has been fenced; the PEL entry must
		// remain pending for the next consumer/leader.
		if (options.upstreamSignal?.aborted) {
			if (strict && suppliedAttempt === undefined) clearLocalDeliveryAttempt(id, attempt)
			return
		}
		if (!strict) throw error
		const failure = classifyRevalidationError(error)
		if (failure.classification === 'permanent' || attempt >= runtime.maxAttempts) {
			await quarantineRevalidationMessage(
				redisClient,
				message,
				id,
				error,
				attempt,
				failure.classification,
				failure.code,
			)
			locallyQuarantinedMessages.add(id)
			if (locallyQuarantinedMessages.size > MAX_LOCAL_QUARANTINED_MESSAGES) {
				const first = locallyQuarantinedMessages.values().next().value as string | undefined
				if (first) locallyQuarantinedMessages.delete(first)
			}
			return
		}
		const delayMs = revalidateRetryBackoffMs(attempt, runtime.retryBackoffBaseMs, runtime.retryBackoffMaxMs)
		const retryTtlSeconds = Math.max(1, Math.ceil(delayMs / 1000))
		await setRetrySuppressionTtl(redisClient, id, retryTtlSeconds)
		logger.warn(`[Revalidate] Retaining ${id} pending after transient failure`, {
			did: message.did,
			rkey: message.rkey,
			reason: message.reason,
			errorCode: failure.code,
			attempt,
			nextRetryInMs: delayMs,
		})
	} finally {
		if (ownedResources) resources.close()
	}
}

const localDeliveryAttempts = new Map<string, number>()
const locallyQuarantinedMessages = new Set<string>()
const MAX_LOCAL_DELIVERY_ATTEMPTS = 20_000
const MAX_LOCAL_QUARANTINED_MESSAGES = 20_000
/** Never let one stale-entry sweep run forever on a malformed/cyclic cursor. */
const MAX_CLAIM_SCAN_ROUNDS = 100
const STREAM_ID_PATTERN = /^\d+-\d+$/

type WorkerActiveCheck = () => boolean

function parseDeliveryCount(value: unknown): number | undefined {
	const row = Array.isArray(value) && Array.isArray(value[0]) ? value[0] : value
	if (!Array.isArray(row)) return undefined
	const count = Number(row[3])
	return Number.isSafeInteger(count) && count >= 1 ? count : undefined
}

function redisText(value: unknown): string | undefined {
	if (typeof value === 'string') return value
	if (value instanceof Uint8Array) return Buffer.from(value).toString()
	return undefined
}

function parseStreamId(value: unknown): string | undefined {
	const id = redisText(value)
	return id && STREAM_ID_PATTERN.test(id) ? id : undefined
}

function parseStreamIdList(value: unknown): string[] {
	if (!Array.isArray(value)) return []
	const ids: string[] = []
	for (const item of value) {
		// JUSTID returns strings, but accepting an entry-shaped item here keeps the
		// scanner safe when a proxy returns the non-JUSTID shape unexpectedly.
		const id = parseStreamId(Array.isArray(item) ? item[0] : item)
		if (id) ids.push(id)
	}
	return ids
}

interface ParsedAutoClaimResponse {
	nextId: string
	claimedIds: string[]
	deletedIds: Set<string>
}

function parseAutoClaimResponse(value: unknown): ParsedAutoClaimResponse | undefined {
	if (!Array.isArray(value)) return undefined
	const nextId = parseStreamId(value[0])
	if (!nextId) return undefined
	return {
		nextId,
		claimedIds: parseStreamIdList(value[1]),
		// Redis 7 adds a third response item containing IDs removed from the PEL
		// because their stream entries were deleted. Older Redis omits it.
		deletedIds: new Set(parseStreamIdList(value[2])),
	}
}

function parseClaimedMessages(value: unknown, requestedIds: ReadonlySet<string>): Array<[string, string[]]> {
	if (!Array.isArray(value)) return []
	const messages: Array<[string, string[]]> = []
	const seen = new Set<string>()
	for (const item of value) {
		if (!Array.isArray(item) || item.length < 2) continue
		const id = parseStreamId(item[0])
		if (!id || !requestedIds.has(id) || seen.has(id)) continue
		const rawFields = item[1]
		if (!Array.isArray(rawFields)) continue
		const fields: string[] = []
		let valid = true
		for (const field of rawFields) {
			const text = redisText(field)
			if (text === undefined) {
				valid = false
				break
			}
			fields.push(text)
		}
		if (!valid) continue
		seen.add(id)
		messages.push([id, fields])
	}
	return messages
}

async function getDeliveryAttempt(client: Redis, id: string): Promise<number> {
	const previous = localDeliveryAttempts.get(id) ?? 0
	let observed: number | undefined
	const xpending = (client as unknown as RevalidateRedisClient).xpending
	if (typeof xpending === 'function') {
		try {
			observed = parseDeliveryCount(
				await xpending.call(client, config.revalidateStream, config.revalidateGroup, id, id, 1),
			)
		} catch {
			// A minimal test seam or an old Redis proxy may not expose XPENDING.
		}
	}
	const attempt = Math.max(observed ?? 0, previous + 1, 1)
	localDeliveryAttempts.set(id, attempt)
	if (localDeliveryAttempts.size > MAX_LOCAL_DELIVERY_ATTEMPTS) {
		const first = localDeliveryAttempts.keys().next().value as string | undefined
		if (first) localDeliveryAttempts.delete(first)
	}
	return attempt
}

function parseRetrySuppressionTtl(value: unknown, id: string): number {
	if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < -2) {
		throw new Error(`Malformed retry suppression TTL for ${id}`)
	}
	return value
}

/**
 * Read the retry keys as one bounded pipeline. The optional direct-command
 * fallback keeps minimal test/proxy seams compatible; production ioredis always
 * exposes `pipeline`, so a stale-entry round performs at most one TTL round trip.
 */
async function readRetrySuppressionTtls(client: Redis, ids: readonly string[]): Promise<number[]> {
	if (ids.length === 0) return []
	if (ids.length > MAX_BATCH_SIZE) throw new Error(`Retry suppression pipeline exceeds ${MAX_BATCH_SIZE} IDs`)

	const redisClient = client as unknown as RevalidateRedisClient
	if (typeof redisClient.pipeline !== 'function') {
		// Keep old/minimal seams working without reintroducing a serial scan. The
		// caller already caps `ids` at MAX_BATCH_SIZE, so this fallback is bounded.
		return await Promise.all(
			ids.map(async (id) => parseRetrySuppressionTtl(await redisClient.ttl(getRetrySuppressionKey(id)), id)),
		)
	}

	const pipeline = redisClient.pipeline()
	if (!pipeline || typeof pipeline.ttl !== 'function' || typeof pipeline.exec !== 'function') {
		throw new Error('Redis retry suppression pipeline is malformed')
	}
	for (const id of ids) pipeline.ttl(getRetrySuppressionKey(id))

	const replies = await pipeline.exec()
	if (!Array.isArray(replies) || replies.length !== ids.length) {
		const replyCount = Array.isArray(replies) ? replies.length : 'invalid'
		throw new Error(`Redis retry suppression pipeline returned ${replyCount} replies for ${ids.length} IDs`)
	}

	const ttls: number[] = []
	for (const [index, reply] of replies.entries()) {
		const id = ids[index]
		if (!id || !Array.isArray(reply) || reply.length !== 2) {
			throw new Error(`Malformed retry suppression pipeline reply for ${id ?? 'unknown'}`)
		}
		const [error, value] = reply
		if (error !== null && error !== undefined) {
			if (error instanceof Error) throw error
			throw new Error(`Retry suppression TTL command failed for ${id}: ${String(error)}`)
		}
		ttls.push(parseRetrySuppressionTtl(value, id))
	}
	return ttls
}

function clearLocalDeliveryAttempt(id: string, attempt: number): void {
	if (localDeliveryAttempts.get(id) === attempt) localDeliveryAttempts.delete(id)
}

async function processMessage(
	id: string,
	rawFields: string[],
	client: Redis,
	runtimeConfig: RevalidateWorkerRuntimeConfig,
	dependencies: RevalidateWorkerDependencies = defaultRevalidateWorkerDependencies,
	upstreamSignal?: AbortSignal,
): Promise<void> {
	// A retry backoff is not a delivery. Check it before reading XPENDING or
	// advancing the local fallback counter so repeated scans cannot consume
	// attempts while the message is intentionally waiting. Lifecycle cancellation
	// is checked around each Redis await so a stop cannot add local attempt debt.
	if (upstreamSignal?.aborted) return
	const retryTtl = parseRetrySuppressionTtl(
		await (client as unknown as RevalidateRedisClient).ttl(getRetrySuppressionKey(id)),
		id,
	)
	if (upstreamSignal?.aborted || retryTtl > 0) return
	const deliveryAttempt = await getDeliveryAttempt(client, id)
	if (upstreamSignal?.aborted) {
		clearLocalDeliveryAttempt(id, deliveryAttempt)
		return
	}
	try {
		await processRevalidationMessage(id, rawFields, client as unknown as RevalidateRedisClient, dependencies, {
			deliveryAttempt,
			runtimeConfig,
			upstreamSignal,
			enforceAttemptPolicy: true,
		})
	} finally {
		if (upstreamSignal?.aborted) clearLocalDeliveryAttempt(id, deliveryAttempt)
	}
}

async function processMessages(
	messages: Array<[string, string[]]>,
	client: Redis,
	runtimeConfig: RevalidateWorkerRuntimeConfig,
	dependencies: RevalidateWorkerDependencies = defaultRevalidateWorkerDependencies,
	isActive: WorkerActiveCheck = () => isActiveWorkerClient(client),
	upstreamSignal?: AbortSignal,
): Promise<void> {
	for (const [id, rawFields] of messages) {
		if (!isActive()) return
		try {
			await processMessage(id, rawFields, client, runtimeConfig, dependencies, upstreamSignal)
		} catch (err) {
			logger.error(`[Revalidate] Failed to process message ${id} (${errorKind(err)})`)
		}
	}
}

async function ensureGroup(client: Redis): Promise<void> {
	if (!isActiveWorkerClient(client)) return
	try {
		await client.xgroup('CREATE', config.revalidateStream, config.revalidateGroup, '0', 'MKSTREAM')
	} catch (err) {
		const error = err instanceof Error ? err : new Error(String(err))
		if (!error.message.includes('BUSYGROUP')) {
			throw error
		}
	}
}

async function claimStaleMessages(
	client: Redis,
	runtimeConfig: RevalidateWorkerRuntimeConfig,
	dependencies: RevalidateWorkerDependencies = defaultRevalidateWorkerDependencies,
	isActive: WorkerActiveCheck = () => isActiveWorkerClient(client),
	upstreamSignal?: AbortSignal,
): Promise<void> {
	let startId = '0-0'
	const visitedCursors = new Set<string>()
	let rounds = 0

	while (isActive() && rounds < MAX_CLAIM_SCAN_ROUNDS) {
		if (visitedCursors.has(startId)) {
			logger.warn(`[Revalidate] Stale-entry scan stopped on repeated cursor ${startId}`)
			break
		}
		visitedCursors.add(startId)
		rounds++

		// JUSTID claims ownership without incrementing the delivery counter. The
		// subsequent XCLAIM below is the only delivery increment for this attempt.
		const rawResponse = await client.xautoclaim(
			config.revalidateStream,
			config.revalidateGroup,
			consumerName,
			runtimeConfig.claimIdleMs,
			startId,
			'COUNT',
			runtimeConfig.batchSize,
			'JUSTID',
		)

		if (!isActive()) return
		const response = parseAutoClaimResponse(rawResponse)
		if (!response) {
			logger.warn('[Revalidate] Stale-entry scan received a malformed XAUTOCLAIM response')
			break
		}

		// Redis honors COUNT, but cap a malformed/proxy response as a final
		// guard against unbounded TTL and XCLAIM work in one scan round.
		const claimedIds: string[] = []
		const seenIds = new Set<string>()
		const scanBatchSize = Number.isSafeInteger(runtimeConfig.batchSize)
			? Math.max(1, Math.min(MAX_BATCH_SIZE, runtimeConfig.batchSize))
			: DEFAULT_BATCH_SIZE
		for (const id of response.claimedIds.slice(0, scanBatchSize)) {
			if (response.deletedIds.has(id) || seenIds.has(id)) continue
			seenIds.add(id)
			claimedIds.push(id)
		}

		const retryTtls = await readRetrySuppressionTtls(client, claimedIds)
		if (!isActive()) return
		const eligibleIds: string[] = []
		for (const [index, id] of claimedIds.entries()) {
			const retryTtl = retryTtls[index]
			if (retryTtl === undefined) throw new Error(`Missing retry suppression TTL for ${id}`)
			if (retryTtl > 0) continue
			eligibleIds.push(id)
		}

		if (eligibleIds.length > 0) {
			// XCLAIM retrieves fields and increments deliveries exactly once for
			// each eligible ID. Deleted entries are simply absent from this reply.
			const claimedMessages = parseClaimedMessages(
				await client.xclaim(config.revalidateStream, config.revalidateGroup, consumerName, 0, ...eligibleIds),
				new Set(eligibleIds),
			)
			if (!isActive()) return
			await processMessages(claimedMessages, client, runtimeConfig, dependencies, isActive, upstreamSignal)
			if (!isActive()) return
		}

		if (response.nextId === '0-0' || response.nextId === startId) break
		startId = response.nextId
	}

	if (rounds >= MAX_CLAIM_SCAN_ROUNDS && isActive()) {
		logger.warn(`[Revalidate] Stale-entry scan reached ${MAX_CLAIM_SCAN_ROUNDS} rounds`)
	}
}

/** @internal Command seam for tests; production uses the worker's active check. */
export async function scanStaleRevalidationMessagesForTests(
	client: Redis,
	runtimeConfig: RevalidateWorkerRuntimeConfig,
	dependencies: RevalidateWorkerDependencies = defaultRevalidateWorkerDependencies,
): Promise<void> {
	await claimStaleMessages(client, runtimeConfig, dependencies, () => true)
}

async function readNewMessages(
	client: Redis,
	runtimeConfig: RevalidateWorkerRuntimeConfig,
	dependencies: RevalidateWorkerDependencies = defaultRevalidateWorkerDependencies,
	upstreamSignal?: AbortSignal,
): Promise<boolean> {
	// BLOCK bounds normal idle reads. ioredis's blockingTimeout resolves a
	// half-open command after BLOCK + grace, while socketTimeout tears down a
	// dead connection instead of letting the worker or shutdown wait forever.
	const response = (await client.xreadgroup(
		'GROUP',
		config.revalidateGroup,
		consumerName,
		'COUNT',
		runtimeConfig.batchSize,
		'BLOCK',
		runtimeConfig.blockMs,
		'STREAMS',
		config.revalidateStream,
		'>',
	)) as [string, Array<[string, string[]]>][] | null

	if (!response || !isActiveWorkerClient(client)) return false

	let received = false
	for (const [, messages] of response) {
		if (messages.length > 0) received = true
		await processMessages(messages, client, runtimeConfig, dependencies, undefined, upstreamSignal)
		if (!isActiveWorkerClient(client)) return false
	}
	return received
}

export function fullJitterDelay(attempt: number, minimumMs: number, maximumMs: number, random = Math.random): number {
	const cap = Math.min(maximumMs, minimumMs * 2 ** Math.min(attempt, 30))
	return Math.floor(random() * (cap + 1))
}

function waitForReconnect(delayMs: number): Promise<void> {
	if (!running) return Promise.resolve()
	return new Promise((resolve) => {
		let settled = false
		const finish = () => {
			if (settled) return
			settled = true
			if (cancelLoopRetryWait === finish) cancelLoopRetryWait = null
			resolve()
		}
		const timer = setTimeout(finish, delayMs)
		cancelLoopRetryWait = () => {
			clearTimeout(timer)
			finish()
		}
	})
}

async function runLoop(
	client: Redis,
	runtimeConfig: RevalidateWorkerRuntimeConfig,
	onHealthyIteration: () => void = () => undefined,
	upstreamSignal?: AbortSignal,
	dependencies: RevalidateWorkerDependencies = defaultRevalidateWorkerDependencies,
): Promise<void> {
	await ensureGroup(client)

	while (isActiveWorkerClient(client)) {
		try {
			await claimStaleMessages(client, runtimeConfig, dependencies, undefined, upstreamSignal)
			const received = await readNewMessages(client, runtimeConfig, dependencies, upstreamSignal)
			if (!isActiveWorkerClient(client)) return
			// A successful XREADGROUP (including a healthy timeout) proves this
			// connection recovered. Reset reconnect backoff while the loop is live,
			// rather than waiting for runLoop to return during shutdown.
			onHealthyIteration()
			if (!received) await trimDrainedRevalidationEntries(client)
		} catch (err) {
			if (!isActiveWorkerClient(client)) break
			throw err
		}
	}
}

export interface RevalidateWorkerTestHooks {
	/** Deterministic backoff source for supervisor tests. */
	random?: () => number
	/** Replace the cancellable production wait in supervisor tests. */
	waitForReconnect?: (delayMs: number) => Promise<void>
	/** Isolated dependencies for lifecycle/cancellation tests. */
	dependencies?: RevalidateWorkerDependencies
}

function startRevalidateWorkerWithFactory(
	redisUrl: string | undefined,
	redisClientFactory: RevalidateWorkerRedisFactory,
	runtimeConfig: RevalidateWorkerRuntimeConfig,
	hooks: RevalidateWorkerTestHooks = {},
): void {
	if (!redisUrl) {
		logger.warn('[Revalidate] REDIS_URL not set; revalidate worker disabled')
		return
	}
	// A timed-out previous loop may still be executing an uncooperative handler.
	// Never start a replacement that could overlap its shared dependencies.
	if (running || loopPromise) return
	running = true
	const generation = ++workerGeneration
	const abortController = new AbortController()
	workerAbortController = abortController
	let attempt = 0
	const supervise = async () => {
		while (running) {
			let client: Redis | null = null
			try {
				client = redisClientFactory(redisUrl, createRevalidateWorkerRedisOptions(runtimeConfig))
				redis = client
				client.on('error', (err) => logger.error(`[Revalidate] Redis error (${errorKind(err)})`))
				client.on('ready', () =>
					logger.info(
						`[Revalidate] Redis connected, stream: ${config.revalidateStream}, group: ${config.revalidateGroup}`,
					),
				)
				await runLoop(
					client,
					runtimeConfig,
					() => {
						attempt = 0
					},
					abortController.signal,
					hooks.dependencies,
				)
				attempt = 0
			} catch (err) {
				if (running) logger.error(`[Revalidate] Redis worker connection failed (${errorKind(err)})`)
			} finally {
				if (redis === client) redis = null
				client?.disconnect()
			}
			if (!running) break
			const delay = fullJitterDelay(attempt++, runtimeConfig.reconnectMinMs, runtimeConfig.reconnectMaxMs, hooks.random)
			await (hooks.waitForReconnect ?? waitForReconnect)(delay)
		}
	}
	const trackedLoop = supervise().finally(() => {
		if (loopPromise === trackedLoop) loopPromise = null
		if (workerGeneration === generation && !redis) {
			running = false
			workerAbortController = null
		}
	})
	loopPromise = trackedLoop
}

export async function startRevalidateWorker(): Promise<void> {
	startRevalidateWorkerWithFactory(config.redisUrl, defaultRedisClientFactory, revalidateWorkerRuntimeConfig)
}

export function startRevalidateWorkerForTests(
	redisClientFactory: RevalidateWorkerRedisFactory,
	redisUrl = 'redis://revalidate-worker-test',
	runtimeConfig: RevalidateWorkerRuntimeConfig = revalidateWorkerRuntimeConfig,
	hooks: RevalidateWorkerTestHooks = {},
): void {
	startRevalidateWorkerWithFactory(redisUrl, redisClientFactory, runtimeConfig, hooks)
}

export interface RevalidateWorkerState {
	running: boolean
	hasRedisClient: boolean
	hasLoop: boolean
}

export function getRevalidateWorkerState(): RevalidateWorkerState {
	return {
		running,
		hasRedisClient: redis !== null,
		hasLoop: loopPromise !== null,
	}
}

/** @internal Backward-compatible test seam. */
export const getRevalidateWorkerStateForTests = getRevalidateWorkerState

export interface RevalidateWorkerStopOptions {
	/** Maximum time to await an in-flight handler before reporting unsafe stop. */
	gracePeriodMs?: number
}

export interface RevalidateWorkerStopResult {
	/** True only when the worker loop settled before the grace period elapsed. */
	stopped: boolean
	/** A timeout means the process must not release shared authority. */
	forced: boolean
}

async function settlesWithinWorkerGrace(work: Promise<void> | null, gracePeriodMs: number): Promise<boolean> {
	if (!work) return true
	if (gracePeriodMs <= 0) return false
	return await new Promise((resolve) => {
		const timer = setTimeout(() => resolve(false), gracePeriodMs)
		void work.then(
			() => {
				clearTimeout(timer)
				resolve(true)
			},
			() => {
				clearTimeout(timer)
				resolve(false)
			},
		)
	})
}

export async function stopRevalidateWorker(
	options: RevalidateWorkerStopOptions = {},
): Promise<RevalidateWorkerStopResult> {
	running = false
	cancelLoopRetryWait?.()

	// Abort active PDS/blob work before disconnecting Redis. The resource context
	// treats this as lifecycle cancellation, not a delivery failure, so the PEL
	// entry remains pending without ACK/XDEL, quarantine, or retry-key writes.
	const generation = workerGeneration
	const abortController = workerAbortController
	abortController?.abort(new Error('Revalidation worker stopping'))

	// Disconnect before awaiting the loop so a BLOCKing XREADGROUP rejects
	// immediately. disconnect() also prevents another reconnect during shutdown.
	const clientToClose = redis
	redis = null
	const loopToWait = loopPromise
	clientToClose?.disconnect()

	const gracePeriodMs = options.gracePeriodMs ?? config.firehoseDrainGraceMs
	const stopped = await settlesWithinWorkerGrace(loopToWait, gracePeriodMs)
	if (!stopped) return { stopped: false, forced: true }
	if (workerGeneration === generation && workerAbortController === abortController) {
		workerAbortController = null
	}
	return { stopped: true, forced: false }
}