/**
 * Leader election for distributed firehose deployments.
 *
 * Only one instance connects to the firehose at a time. Cursor keys are scoped
 * to a non-secret relay fingerprint because relay sequence spaces are local.
 */

import { createHash, randomUUID } from 'node:crypto'
import { performance } from 'node:perf_hooks'
import { createLogger } from '@wispplace/observability'
import Redis from 'ioredis'
import { config } from '../config'

const logger = createLogger('firehose-service')

const LEADER_KEY = 'wisp:firehose-leader'
const CURSOR_KEY_PREFIX = 'wisp:firehose-cursor'
const CURSOR_PATTERN = /^(0|[1-9][0-9]*)$/

// Unique ID for this process instance
const instanceId = randomUUID()

// Lua script: renew leadership only if this instance still owns the key
const RENEW_SCRIPT = `
if redis.call('get', KEYS[1]) == ARGV[1] then
  return redis.call('set', KEYS[1], ARGV[1], 'XX', 'PX', tonumber(ARGV[2]))
else
  return 0
end
`

// Lua script: release leadership only if this instance still owns the key
const RELEASE_SCRIPT = `
if redis.call('get', KEYS[1]) == ARGV[1] then
  return redis.call('del', KEYS[1])
else
  return 0
end
`

const LUA_CURSOR_DECIMAL_HELPERS = `
local maxSafe = '9007199254740991'
local function valid(value)
  if not value or value == '' then return false end
  if value == '0' then return true end
  if not string.match(value, '^[1-9][0-9]*$') then return false end
  if string.len(value) < string.len(maxSafe) then return true end
  return string.len(value) == string.len(maxSafe) and value <= maxSafe
end
local function atLeast(left, right)
  if string.len(left) ~= string.len(right) then return string.len(left) > string.len(right) end
  return left >= right
end
`

/** Atomically keep the greatest cursor in one relay's sequence space. */
export const SAVE_CURSOR_SCRIPT = `
${LUA_CURSOR_DECIMAL_HELPERS}
local current = redis.call('get', KEYS[1])
local next = ARGV[1]
if valid(current) and atLeast(current, next) then
  return 0
end
redis.call('set', KEYS[1], next)
return 1
`

/**
 * Atomically copy the same-relay legacy cursor into the hashed key without
 * ever moving the hashed key backwards. The client verifies the returned
 * cursor with a second read before it removes the legacy key.
 */
export const MIGRATE_CURSOR_SCRIPT = `
${LUA_CURSOR_DECIMAL_HELPERS}
local current = redis.call('get', KEYS[1])
local legacy = redis.call('get', KEYS[2])
if not valid(legacy) then
  return {0, current or ''}
end

local selected = legacy
if valid(current) and atLeast(current, legacy) then selected = current end
if not valid(current) or current ~= selected then redis.call('set', KEYS[1], selected) end
return {1, selected}
`

/**
 * Remove a legacy cursor only when both keys are valid and the hashed cursor
 * is at least as new as the legacy cursor at the instant of deletion.
 *
 * Return values: 1 deleted, 0 already absent, -1 malformed legacy, -2
 * malformed hashed value, -3 legacy became newer and needs another migration,
 * -4 hashed value fell below the cursor that was re-read by the client.
 */
export const DELETE_MIGRATED_CURSOR_SCRIPT = `
${LUA_CURSOR_DECIMAL_HELPERS}
local current = redis.call('get', KEYS[1])
local legacy = redis.call('get', KEYS[2])
local expected = ARGV[1]
if not valid(current) then return -2 end
if not valid(expected) then return -4 end
if legacy and valid(legacy) and not atLeast(current, legacy) then return -3 end
if not atLeast(current, expected) then return -4 end
if not legacy then return 0 end
if not valid(legacy) then return -1 end
return redis.call('del', KEYS[2])
`

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

class AmbiguousLegacyCursorError extends Error {
	constructor(hashedKey: string) {
		super(
			`Ambiguous legacy cursor for a non-root relay path; preseed ${hashedKey} with the correct checkpoint before startup`,
		)
		this.name = 'AmbiguousLegacyCursorError'
	}
}

/** A stable non-secret relay identifier for Redis keys and diagnostic labels. */
export function relayFingerprint(service: string): string {
	let normalized = 'invalid-relay'
	try {
		const url = new URL(service)
		normalized = `${url.protocol}//${url.hostname.toLowerCase()}:${url.port || 'default'}${url.pathname}`
	} catch {
		// Configuration validates relays before this module starts. Keep a stable
		// fallback to avoid ever placing a raw malformed value in a key or log.
	}
	return createHash('sha256').update(normalized).digest('hex')
}

function relayLabel(service: string): string {
	if (service === config.firehoseService) return 'primary'
	if (service === config.firehoseServiceSecondary) return 'secondary'
	return `relay-${relayFingerprint(service)}`
}

function cursorKey(service: string): string {
	return `${CURSOR_KEY_PREFIX}:${relayFingerprint(service)}`
}

/**
 * Previous releases keyed cursors by host. Keep deriving that key so operators
 * can identify old data, but only root relay URLs may import it: URL paths on
 * the same host cannot be disambiguated.
 */
export function legacyCursorKey(service: string): string | undefined {
	try {
		return `${CURSOR_KEY_PREFIX}:${new URL(service).host}`
	} catch {
		return undefined
	}
}

function isRootRelayPath(service: string): boolean {
	try {
		return new URL(service).pathname === '/'
	} catch {
		return false
	}
}

export function isValidCursor(seq: unknown): seq is number {
	return typeof seq === 'number' && Number.isSafeInteger(seq) && seq >= 0
}

/** Parses only decimal nonnegative safe integers. */
export function parseStoredCursor(value: string | null): number | undefined {
	if (!value || !CURSOR_PATTERN.test(value)) return undefined
	const parsed = Number(value)
	return isValidCursor(parsed) ? parsed : undefined
}

/** Missing and corrupt values both mean there is no confirmed checkpoint. */
export function boundedCursorFromStoredValue(value: string | null): number | undefined {
	return parseStoredCursor(value)
}

/** Pure mirror of the atomic Redis cursor rule, used for validation/tests. */
export function shouldAdvanceStoredCursor(value: string | null, next: number): boolean {
	if (!isValidCursor(next)) return false
	const current = parseStoredCursor(value)
	return current === undefined || current < next
}

const MIN_REDIS_OPERATION_TIMEOUT_MS = 100
const MAX_REDIS_OPERATION_TIMEOUT_MS = 5_000

/** Keep every Redis operation short enough to leave a substantial lease margin. */
function redisOperationTimeoutMs(): number {
	const ttl = config.leaderTtlMs
	if (!Number.isSafeInteger(ttl) || ttl < 1) return MIN_REDIS_OPERATION_TIMEOUT_MS
	return Math.max(MIN_REDIS_OPERATION_TIMEOUT_MS, Math.min(MAX_REDIS_OPERATION_TIMEOUT_MS, Math.floor(ttl / 8)))
}

class RedisOperationTimeoutError extends Error {
	constructor(operation: string) {
		super(`${operation} timed out`)
		this.name = 'RedisOperationTimeoutError'
	}
}

function withRedisTimeout<T>(operation: PromiseLike<T>, timeoutMs: number, operationName: string): Promise<T> {
	return new Promise((resolve, reject) => {
		const timer = setTimeout(() => reject(new RedisOperationTimeoutError(operationName)), timeoutMs)
		Promise.resolve(operation).then(
			(value) => {
				clearTimeout(timer)
				resolve(value)
			},
			(error) => {
				clearTimeout(timer)
				reject(error)
			},
		)
	})
}

let redis: Redis | null = null
let redisConnectPromise: Promise<void> | null = null

function getRedis(): Redis {
	if (!redis) {
		if (!config.redisUrl) throw new Error('REDIS_URL is required for leader election')
		const timeoutMs = redisOperationTimeoutMs()
		redis = new Redis(config.redisUrl, {
			// A lease operation must not sit behind ioredis's retry queue. The
			// explicit command timeout below is deliberately much shorter than TTL.
			maxRetriesPerRequest: 0,
			connectTimeout: timeoutMs,
			commandTimeout: timeoutMs,
			enableOfflineQueue: false,
			enableReadyCheck: true,
			lazyConnect: true,
		})
		redis.on('error', (error) => logger.error('[Leader] Redis error', undefined, { errorKind: errorKind(error) }))
	}
	return redis
}

function waitForRedisReady(client: Redis): Promise<void> {
	return new Promise((resolve, reject) => {
		let settled = false
		const finish = (error?: unknown) => {
			if (settled) return
			settled = true
			client.removeListener('ready', onReady)
			client.removeListener('error', onError)
			if (error === undefined) resolve()
			else reject(error)
		}
		const onReady = () => finish()
		const onError = (error: unknown) => finish(error)
		client.once('ready', onReady)
		client.once('error', onError)
	})
}

async function ensureRedisConnected(client: Redis): Promise<void> {
	// Keep the command boundary usable with lightweight test doubles while the
	// real ioredis client always exposes status/connect and is fully bounded.
	if (typeof client.connect !== 'function' || typeof client.status !== 'string') return
	if (client.status === 'ready') return
	if (redisConnectPromise) return redisConnectPromise

	const attempt = (async () => {
		const timeoutMs = redisOperationTimeoutMs()
		if (client.status === 'wait') {
			await withRedisTimeout(client.connect(), timeoutMs, 'Redis connect')
			return
		}
		await withRedisTimeout(waitForRedisReady(client), timeoutMs, 'Redis connect')
	})()
	const tracked = attempt.finally(() => {
		if (redisConnectPromise === tracked) redisConnectPromise = null
	})
	redisConnectPromise = tracked
	return tracked
}

function resetRedisClient(client: Redis): void {
	if (redis !== client) return
	redis = null
	redisConnectPromise = null
	try {
		client.disconnect()
	} catch {
		// The next bounded operation creates a fresh client.
	}
}

async function runRedisCommand<T>(operationName: string, command: (client: Redis) => PromiseLike<T>): Promise<T> {
	const client = getRedis()
	try {
		await ensureRedisConnected(client)
		return await withRedisTimeout(
			Promise.resolve().then(() => command(client)),
			redisOperationTimeoutMs(),
			operationName,
		)
	} catch (error) {
		// A timeout can leave an ioredis command queued behind a dead socket.
		// Fence that client so a later election attempt cannot reuse it.
		if (error instanceof RedisOperationTimeoutError || client.status === 'end' || client.status === 'close') {
			resetRedisClient(client)
		}
		throw error
	}
}

async function tryBecomeLeader(): Promise<boolean> {
	const result = await runRedisCommand('Redis leader acquisition', (client) =>
		client.set(LEADER_KEY, instanceId, 'PX', config.leaderTtlMs, 'NX'),
	)
	return result === 'OK'
}

async function renewLeadership(): Promise<boolean> {
	const result = (await runRedisCommand('Redis leader renewal', (client) =>
		client.eval(RENEW_SCRIPT, 1, LEADER_KEY, instanceId, String(config.leaderTtlMs)),
	)) as string | null
	return result === 'OK'
}

export async function releaseLeadership(): Promise<void> {
	try {
		const deleted = (await runRedisCommand('Redis leader release', (client) =>
			client.eval(RELEASE_SCRIPT, 1, LEADER_KEY, instanceId),
		)) as number
		if (deleted === 1) logger.info('[Leader] Released leader key for standby takeover')
	} catch (error) {
		logger.warn('[Leader] Failed to release leader key', { errorKind: errorKind(error) })
	}
}

export type DurableCursorRead = { kind: 'found'; cursor: number } | { kind: 'missing' } | { kind: 'unavailable' }

export type DurableCursorSave = { kind: 'saved' } | { kind: 'invalid' } | { kind: 'unavailable' }

/**
 * Save a relay-scoped cursor and report whether Redis durably accepted it.
 * The boolean is intentionally separate from the legacy void API so relay
 * failover can refuse to subscribe from an unknown checkpoint.
 */
export async function saveDurableCursor(seq: number, service: string): Promise<DurableCursorSave> {
	if (!isValidCursor(seq)) {
		logger.warn('[Leader] Refused to save an invalid cursor')
		return { kind: 'invalid' }
	}
	try {
		const result = await runRedisCommand('Redis cursor save', (client) =>
			client.eval(SAVE_CURSOR_SCRIPT, 1, cursorKey(service), String(seq)),
		)
		if (result !== 0 && result !== 1 && result !== '0' && result !== '1') {
			throw new Error('Redis cursor save returned an invalid result')
		}
		return { kind: 'saved' }
	} catch (error) {
		logger.warn('[Leader] Failed to save cursor', { errorKind: errorKind(error), relay: relayLabel(service) })
		return { kind: 'unavailable' }
	}
}

/** Backward-compatible best-effort cursor save used by periodic checkpointing. */
export async function saveCursor(seq: number, service: string): Promise<void> {
	await saveDurableCursor(seq, service)
}

export interface CursorRedisClient {
	get(key: string): Promise<string | null>
	eval(script: string, keyCount: number, ...args: Array<string | number>): Promise<unknown>
}

export type CursorReadDetails = {
	read: DurableCursorRead
	migrated: boolean
	corrupt: boolean
}

const MAX_CURSOR_MIGRATION_ATTEMPTS = 3

function redisIntegerResult(value: unknown): number | undefined {
	if (typeof value === 'number') return Number.isSafeInteger(value) ? value : undefined
	if (typeof value !== 'string' || !/^-?[0-9]+$/.test(value)) return undefined
	const parsed = Number(value)
	return Number.isSafeInteger(parsed) ? parsed : undefined
}

function parseCursorMigrationResult(value: unknown): { status: 0 | 1; cursorValue: string } {
	if (!Array.isArray(value) || value.length < 2) throw new Error('Redis cursor migration returned an invalid result')
	const status = redisIntegerResult(value[0])
	if (status !== 0 && status !== 1) throw new Error('Redis cursor migration returned an invalid status')
	const cursorValue = value[1]
	if (typeof cursorValue !== 'string') throw new Error('Redis cursor migration returned an invalid cursor')
	return { status, cursorValue }
}

function parseCursorDeletionResult(value: unknown): number {
	const result = redisIntegerResult(value)
	if (result !== 0 && result !== 1 && result !== -1 && result !== -2 && result !== -3 && result !== -4) {
		throw new Error('Redis legacy cursor deletion returned an invalid result')
	}
	return result
}

async function readCursorWithoutLegacy(client: CursorRedisClient, key: string): Promise<CursorReadDetails> {
	const value = await client.get(key)
	const cursor = parseStoredCursor(value)
	if (cursor !== undefined) return { read: { kind: 'found', cursor }, migrated: false, corrupt: false }
	return { read: { kind: 'missing' }, migrated: false, corrupt: value !== null }
}

/**
 * Read a cursor and migrate the previous same-relay host key for root relay URLs.
 * Non-root URLs cannot safely identify which independent path produced a host-only
 * legacy key, so an existing legacy value requires an operator preseed instead.
 *
 * The migration script atomically takes the maximum of both valid values. We
 * then re-read the hashed key and only run the conditional delete script after
 * that re-read proves a valid cursor at least as new as the migrated value.
 * This keeps a failed read, a malformed concurrent write, or a racing newer
 * legacy value from deleting the only recoverable checkpoint.
 */
export async function readDurableCursorWithRedis(
	client: CursorRedisClient,
	service: string,
): Promise<CursorReadDetails> {
	const key = cursorKey(service)
	const oldKey = legacyCursorKey(service)
	if (!isRootRelayPath(service)) {
		const hashedValue = await client.get(key)
		const hashedCursor = parseStoredCursor(hashedValue)
		if (hashedCursor !== undefined)
			return { read: { kind: 'found', cursor: hashedCursor }, migrated: false, corrupt: false }

		// A host-only legacy key may belong to any independent URL path on this
		// host. Never import it into a path-scoped key. An operator must preseed
		// the hashed key from the correct relay checkpoint before startup.
		if (oldKey && oldKey !== key && (await client.get(oldKey)) !== null) {
			throw new AmbiguousLegacyCursorError(key)
		}
		return { read: { kind: 'missing' }, migrated: false, corrupt: hashedValue !== null }
	}
	if (!oldKey || oldKey === key) return readCursorWithoutLegacy(client, key)

	for (let attempt = 0; attempt < MAX_CURSOR_MIGRATION_ATTEMPTS; attempt++) {
		const migration = parseCursorMigrationResult(await client.eval(MIGRATE_CURSOR_SCRIPT, 2, key, oldKey))
		const migratedCursor = parseStoredCursor(migration.cursorValue)
		if (migration.status === 0) {
			if (migratedCursor !== undefined)
				return { read: { kind: 'found', cursor: migratedCursor }, migrated: false, corrupt: false }
			return { read: { kind: 'missing' }, migrated: false, corrupt: migration.cursorValue !== '' }
		}
		if (migratedCursor === undefined) throw new Error('Redis legacy cursor migration produced an invalid cursor')

		const rereadCursor = parseStoredCursor(await client.get(key))
		if (rereadCursor === undefined || rereadCursor < migratedCursor) {
			throw new Error('Redis cursor changed before legacy migration was confirmed')
		}

		const deletion = parseCursorDeletionResult(
			await client.eval(DELETE_MIGRATED_CURSOR_SCRIPT, 2, key, oldKey, String(rereadCursor)),
		)
		if (deletion === 1 || deletion === 0) {
			return { read: { kind: 'found', cursor: rereadCursor }, migrated: true, corrupt: false }
		}
		if (deletion === -1) {
			// The legacy value changed to malformed data after migration. Keep it so
			// no unverified delete can destroy a potentially recoverable checkpoint.
			return { read: { kind: 'found', cursor: rereadCursor }, migrated: false, corrupt: false }
		}
		if (deletion === -2 || deletion === -4)
			throw new Error('Redis cursor changed before legacy migration was confirmed')
		// A newer legacy value won the race. Re-run the atomic max and confirmation
		// sequence rather than deleting that newer checkpoint.
	}
	throw new Error('Redis legacy cursor migration changed concurrently')
}

/**
 * Distinguish a missing checkpoint from an unavailable durable store. A missing
 * or corrupt value is a confirmed absence and starts the relay live; an
 * unavailable store must not cause a relay subscription from an unknown position.
 */
export async function readDurableCursor(service: string): Promise<DurableCursorRead> {
	try {
		const result = await runRedisCommand('Redis cursor read', (client) => readDurableCursorWithRedis(client, service))
		if (result.migrated) logger.info('[Leader] Migrated same-relay legacy cursor checkpoint')
		if (result.corrupt && result.read.kind === 'missing') {
			logger.warn('[Leader] Ignoring corrupt saved cursor; starting live', { relay: relayLabel(service) })
		}
		return result.read
	} catch (error) {
		if (error instanceof AmbiguousLegacyCursorError) {
			logger.error('[Leader] Ambiguous legacy cursor; preseed the path-scoped cursor before startup', undefined, {
				relay: relayLabel(service),
			})
		} else {
			logger.warn('[Leader] Failed to read cursor', { errorKind: errorKind(error), relay: relayLabel(service) })
		}
		return { kind: 'unavailable' }
	}
}

/** Backward-compatible cursor reader for leader startup. */
export async function readCursor(service: string): Promise<number | undefined> {
	const result = await readDurableCursor(service)
	if (result.kind === 'unavailable') throw new Error('Durable cursor store is unavailable')
	// A confirmed missing or corrupt checkpoint starts from the relay head. Only
	// an unavailable store is an error and must fail closed before subscription.
	return result.kind === 'found' ? result.cursor : undefined
}

export function getLeaderInfo() {
	return { instanceId }
}

function sleep(ms: number, signal: AbortSignal): Promise<void> {
	if (signal.aborted) return Promise.resolve()
	return new Promise((resolve) => {
		const timer = setTimeout(done, ms)
		const onAbort = () => {
			clearTimeout(timer)
			done()
		}
		function done() {
			signal.removeEventListener('abort', onAbort)
			resolve()
		}
		signal.addEventListener('abort', onAbort, { once: true })
	})
}

/** Returns true only when direct firehose work drained safely. */
type LeadershipLossCallback = () => boolean | Promise<boolean>
type LeadershipStartCallback = (
	cursor: number | undefined,
	stepDown: () => Promise<void>,
	isActive: () => boolean,
) => void | Promise<void>

export interface LeaderElectionDependencies {
	tryBecomeLeader(): Promise<boolean>
	renewLeadership(): Promise<boolean>
	releaseLeadership(): Promise<void>
	readCursor(service: string): Promise<number | undefined>
	sleep(ms: number, signal: AbortSignal): Promise<void>
	/** Monotonic clock seam for deterministic lease-deadline tests. */
	now?: () => number
	/** Timer seams keep lease-loss ordering tests deterministic. */
	setTimeout?: (callback: () => void, delayMs: number) => ReturnType<typeof setTimeout>
	clearTimeout?: (timer: ReturnType<typeof setTimeout>) => void
}

const defaultLeaderElectionDependencies: LeaderElectionDependencies = {
	tryBecomeLeader,
	renewLeadership,
	releaseLeadership,
	readCursor,
	sleep,
}

const MIN_LEASE_SAFETY_MARGIN_MS = 100

function leaseSafetyMarginMs(): number {
	// Keep a quarter of the configured TTL as a local safety window. Redis
	// operations are bounded to a smaller fraction, and the lease deadline is
	// enforced independently of renewal response timing.
	return Math.max(MIN_LEASE_SAFETY_MARGIN_MS, Math.floor(config.leaderTtlMs / 4))
}

function leaseValidityWindowMs(): number {
	return Math.max(1, config.leaderTtlMs - leaseSafetyMarginMs())
}

/**
 * Owns one process's Redis lease state. Epochs fence late renewal responses so
 * an old timer cannot modify a later leadership term after a release/reacquire.
 * The monotonic deadline is deliberately separate from renewal failures: a
 * hung renewal cannot keep this process acting as leader past its local safety
 * window.
 */
class LeaderElectionMachine {
	private isLeader = false
	private renewalTimer: ReturnType<typeof setTimeout> | null = null
	private leaseDeadlineTimer: ReturnType<typeof setTimeout> | null = null
	private leaseDeadline: number | null = null
	private renewalInFlightEpoch: number | null = null
	private steppingDown: Promise<void> | null = null
	private leadershipLoss: Promise<boolean> | null = null
	private abortLoss: Promise<void> = Promise.resolve()
	private lostLeadershipDrain: Promise<boolean> | null = null
	private terminalError: Error | null = null
	private leadershipEpoch = 0
	private readonly terminalWake: Promise<void>
	private resolveTerminalWake: (() => void) | null = null

	constructor(
		private readonly onBecomeLeader: LeadershipStartCallback,
		private readonly onLoseLeadership: LeadershipLossCallback,
		private readonly signal: AbortSignal,
		private readonly initialService: string,
		private readonly dependencies: LeaderElectionDependencies = defaultLeaderElectionDependencies,
	) {
		this.terminalWake = new Promise((resolve) => {
			this.resolveTerminalWake = resolve
		})
	}

	async run(): Promise<void> {
		logger.info('[Leader] Starting election loop', { relay: relayLabel(this.initialService) })
		this.signal.addEventListener('abort', this.onAbort, { once: true })
		try {
			while (this.canContinue()) await this.runCycle()
		} finally {
			this.signal.removeEventListener('abort', this.onAbort)
			await this.abortLoss
		}
		if (this.terminalError) throw this.terminalError
	}

	private canContinue(): boolean {
		return !this.signal.aborted && this.terminalError === null
	}

	private async runCycle(): Promise<void> {
		if (this.isLeader) {
			this.checkLeaseDeadline()
			if (!this.isLeader) return
			await Promise.race([this.dependencies.sleep(config.leaderPollIntervalMs, this.signal), this.terminalWake])
			// A fake or delayed clock may advance without allowing a native timer to
			// run. Check again after every election-loop sleep as a second fence.
			this.checkLeaseDeadline()
			return
		}
		await this.tryAcquireLeadership()
	}

	private async tryAcquireLeadership(): Promise<void> {
		if (!(await this.awaitLostLeadershipDrain()) || !this.canContinue()) return

		const won = await this.tryBecomeLeader()
		if (!won) {
			await Promise.race([this.dependencies.sleep(config.leaderPollIntervalMs, this.signal), this.terminalWake])
			return
		}
		if (!this.canContinue()) {
			await this.dependencies.releaseLeadership()
			return
		}
		const started = await this.startLeaderWork()
		if (!started && this.canContinue()) {
			await Promise.race([this.dependencies.sleep(config.leaderPollIntervalMs, this.signal), this.terminalWake])
		}
	}

	private async awaitLostLeadershipDrain(): Promise<boolean> {
		const pendingDrain = this.lostLeadershipDrain
		if (!pendingDrain) return true

		const drained = await pendingDrain
		if (this.lostLeadershipDrain === pendingDrain) this.lostLeadershipDrain = null
		if (this.leadershipLoss === pendingDrain) this.leadershipLoss = null
		if (drained) return true
		this.markTerminal('Leader election stopped after an unsafe leadership drain')
		return false
	}

	private async tryBecomeLeader(): Promise<boolean> {
		try {
			return await this.dependencies.tryBecomeLeader()
		} catch (error) {
			logger.error('[Leader] Election attempt failed', undefined, { errorKind: errorKind(error) })
			return false
		}
	}

	private async startLeaderWork(): Promise<boolean> {
		const epoch = ++this.leadershipEpoch
		this.isLeader = true
		this.leaseDeadline = this.monotonicNow() + leaseValidityWindowMs()
		this.armLeaseDeadline(epoch)
		this.scheduleRenewal(epoch)

		try {
			const cursor = await this.dependencies.readCursor(this.initialService)
			// The lease deadline can fire while a durable cursor read is in flight.
			// Never start work from a term which was fenced during that await.
			if (!this.isActiveEpoch(epoch)) return false
			logger.info('[Leader] Won leadership', {
				relay: relayLabel(this.initialService),
				hasCursor: cursor !== undefined,
			})
			await this.onBecomeLeader(cursor, this.stepDown, () => this.isActiveEpoch(epoch))
		} catch (error) {
			logger.error('[Leader] Failed to start leader work', undefined, { errorKind: errorKind(error) })
			await this.stepDown()
			return false
		}
		return this.isActiveEpoch(epoch)
	}

	private monotonicNow(): number {
		try {
			const supplied = this.dependencies.now?.()
			if (typeof supplied === 'number' && Number.isFinite(supplied)) return supplied
		} catch {
			// Fall back to the process monotonic clock if a test or optional clock
			// implementation fails. Never use wall-clock time for lease deadlines.
		}
		return performance.now()
	}

	private isActiveEpoch(epoch: number): boolean {
		return !this.signal.aborted && this.isLeader && epoch === this.leadershipEpoch
	}

	private leaseHasExpired(): boolean {
		return this.leaseDeadline !== null && this.monotonicNow() >= this.leaseDeadline
	}

	private checkLeaseDeadline(): void {
		if (this.isLeader && this.leaseHasExpired()) this.beginLeaseLoss()
	}

	private armLeaseDeadline(epoch: number): void {
		if (!this.isActiveEpoch(epoch) || this.leaseDeadline === null) return
		if (this.leaseDeadlineTimer) this.clearTimer(this.leaseDeadlineTimer)
		const remaining = this.leaseDeadline - this.monotonicNow()
		if (remaining <= 0) {
			this.beginLeaseLoss()
			return
		}
		this.leaseDeadlineTimer = this.scheduleTimer(
			() => {
				this.leaseDeadlineTimer = null
				if (!this.isActiveEpoch(epoch)) return
				if (this.leaseHasExpired()) this.beginLeaseLoss()
				else this.armLeaseDeadline(epoch)
			},
			Math.max(1, Math.ceil(remaining)),
		)
	}

	private scheduleRenewal(epoch: number): void {
		if (!this.isActiveEpoch(epoch) || this.leaseDeadline === null) return
		if (this.renewalTimer) this.clearTimer(this.renewalTimer)
		const remaining = this.leaseDeadline - this.monotonicNow()
		if (remaining <= 0) {
			this.beginLeaseLoss()
			return
		}
		const configuredInterval = Number.isSafeInteger(config.leaderRenewIntervalMs)
			? Math.max(1, config.leaderRenewIntervalMs)
			: Math.max(1, Math.floor(config.leaderTtlMs / 4))
		const delay = Math.min(configuredInterval, Math.max(1, Math.floor(remaining)))
		this.renewalTimer = this.scheduleTimer(() => {
			this.renewalTimer = null
			void this.renewLeadership(epoch)
		}, delay)
	}

	private async renewLeadership(epoch: number): Promise<void> {
		if (!this.isActiveEpoch(epoch) || this.renewalInFlightEpoch === epoch) return
		if (this.leaseHasExpired()) {
			this.beginLeaseLoss()
			return
		}

		this.renewalInFlightEpoch = epoch
		let renewed = false
		try {
			renewed = await this.tryRenewLeadership()
		} finally {
			if (this.renewalInFlightEpoch === epoch) this.renewalInFlightEpoch = null
		}
		if (!this.isActiveEpoch(epoch)) return
		if (this.leaseHasExpired()) {
			this.beginLeaseLoss()
			return
		}
		if (renewed) {
			// The response arrived before the old deadline. Start the next local
			// safety window from its receipt instead of trusting wall-clock time.
			this.leaseDeadline = this.monotonicNow() + leaseValidityWindowMs()
			this.armLeaseDeadline(epoch)
			this.scheduleRenewal(epoch)
			return
		}
		logger.warn('[Leader] Renewal failed before the lease deadline; retrying')
		this.scheduleRenewal(epoch)
	}

	private async tryRenewLeadership(): Promise<boolean> {
		try {
			return await this.dependencies.renewLeadership()
		} catch (error) {
			logger.error('[Leader] Renewal error', undefined, { errorKind: errorKind(error) })
			return false
		}
	}

	private beginLeaseLoss(): void {
		if (!this.isLeader) return
		logger.warn('[Leader] Lease safety deadline reached; stopping intake without waiting')
		this.relinquishLeadership()
		// If a voluntary step-down is already draining, share that exact promise
		// rather than invoking the stop callback twice.
		const drain = this.startLeadershipLoss()
		this.lostLeadershipDrain = drain
		void drain.then((drained) => {
			if (!drained) this.markTerminal('Leader election stopped after an unsafe leadership drain')
		})
	}

	private startLeadershipLoss(): Promise<boolean> {
		if (!this.leadershipLoss) this.leadershipLoss = this.safelyLoseLeadership()
		return this.leadershipLoss
	}

	private readonly stepDown = (): Promise<void> => {
		if (this.steppingDown) return this.steppingDown
		if (!this.isLeader) return Promise.resolve()

		this.steppingDown = this.drainThenRelease().finally(() => {
			this.steppingDown = null
		})
		return this.steppingDown
	}

	private async drainThenRelease(): Promise<void> {
		logger.warn('[Leader] Stepping down voluntarily after drain')
		// Renewal deliberately remains active during this await. The deadline
		// watchdog can still fence a hung drain or renewal independently.
		const drained = await this.startLeadershipLoss()
		const stillOwned = this.isLeader
		if (!stillOwned) return
		this.relinquishLeadership()
		if (!drained) {
			this.leaveLeaseToExpire()
			return
		}

		const released = await this.safelyReleaseLeadership()
		if (!released) {
			this.leaveLeaseToExpire()
			return
		}
		this.leadershipLoss = null
	}

	private relinquishLeadership(): void {
		this.clearRenewal()
		this.isLeader = false
		this.leadershipEpoch++
		this.leaseDeadline = null
	}

	private leaveLeaseToExpire(): void {
		this.markTerminal('Leader election stopped after a forced leadership drain')
		logger.error('[Leader] Drain was forced; leaving lease to expire instead of releasing it')
	}

	private async safelyReleaseLeadership(): Promise<boolean> {
		try {
			await this.dependencies.releaseLeadership()
			return true
		} catch (error) {
			logger.error('[Leader] Failed to release leader key', undefined, { errorKind: errorKind(error) })
			return false
		}
	}

	private async safelyLoseLeadership(): Promise<boolean> {
		try {
			return await this.onLoseLeadership()
		} catch (error) {
			logger.error('[Leader] Firehose stop failed', undefined, { errorKind: errorKind(error) })
			return false
		}
	}

	private markTerminal(message: string): void {
		if (this.terminalError) return
		this.terminalError = new Error(message)
		this.resolveTerminalWake?.()
		this.resolveTerminalWake = null
	}

	private readonly onAbort = (): void => {
		this.clearRenewal()
		this.leaseDeadline = null
		this.leadershipEpoch++
		if (this.steppingDown) {
			this.abortLoss = this.steppingDown
			return
		}
		if (this.isLeader) {
			this.isLeader = false
			const loss = this.startLeadershipLoss()
			this.abortLoss = loss.then((drained) => {
				if (!drained) this.markTerminal('Leader election stopped after an unsafe shutdown drain')
			})
			return
		}
		if (this.lostLeadershipDrain) this.abortLoss = this.lostLeadershipDrain.then(() => undefined)
	}

	private scheduleTimer(callback: () => void, delayMs: number): ReturnType<typeof setTimeout> {
		return this.dependencies.setTimeout?.(callback, delayMs) ?? setTimeout(callback, delayMs)
	}

	private clearTimer(timer: ReturnType<typeof setTimeout>): void {
		if (this.dependencies.clearTimeout) this.dependencies.clearTimeout(timer)
		else clearTimeout(timer)
	}

	private clearRenewal(): void {
		if (this.renewalTimer) this.clearTimer(this.renewalTimer)
		this.renewalTimer = null
		if (this.leaseDeadlineTimer) this.clearTimer(this.leaseDeadlineTimer)
		this.leaseDeadlineTimer = null
	}
}

/**
 * Run the leader election loop. A voluntary step-down retains renewal until
 * direct firehose work has drained, then releases the lock. A true renewal loss
 * stops intake immediately and relies on replay plus per-site locks for any
 * overlap with the new leader.
 */
export async function runLeaderElection(
	onBecomeLeader: LeadershipStartCallback,
	onLoseLeadership: LeadershipLossCallback,
	signal: AbortSignal,
	initialService: string,
	dependencies: LeaderElectionDependencies = defaultLeaderElectionDependencies,
): Promise<void> {
	await new LeaderElectionMachine(onBecomeLeader, onLoseLeadership, signal, initialService, dependencies).run()
}

export async function closeLeaderRedis(): Promise<void> {
	if (redis) {
		const toClose = redis
		redis = null
		redisConnectPromise = null
		try {
			await withRedisTimeout(toClose.quit(), redisOperationTimeoutMs(), 'Redis close')
		} catch {
			try {
				toClose.disconnect()
			} catch {
				// Closing is best effort after the bounded shutdown attempt.
			}
		}
	}
}