/**
 * Firehose worker - watches AT Protocol firehose for site changes.
 *
 * A cursor is only durable after all earlier site work has completed (or has
 * been durably handed to the revalidation stream). This gives reconnects and
 * process restarts at-least-once delivery instead of skipping queued writes.
 */

import { IdResolver } from '@atproto/identity'
import { Firehose } from '@atproto/sync'
import { BunFirehose, type CommitEvt, type Event, isBun } from '@wispplace/bun-firehose'
import type { Record as WispFsRecord } from '@wispplace/lexicons/types/place/wisp/fs'
import type { Record as WispSettings } from '@wispplace/lexicons/types/place/wisp/settings'
import { validateRecord as validateSettingsRecord } from '@wispplace/lexicons/types/place/wisp/settings'
import { createLogger } from '@wispplace/observability'
import { config } from '../config'
import {
	enqueueSiteRevalidation,
	recordSiteReconciliation,
	type SiteRevalidationEnqueueResult,
} from './cache-invalidation'
import { handleSettingsDelete, handleSettingsUpdate, handleSiteCreateOrUpdate, handleSiteDelete } from './cache-writer'
import { readDurableCursor, relayFingerprint, saveDurableCursor } from './leader'
import {
	SETTINGS_DELETE_FAILURE_REASON,
	SETTINGS_UPDATE_FAILURE_REASON,
	SITE_DELETE_TOMBSTONE_REASON,
} from './revalidate-queue'
import {
	assertRevalidationActive,
	createRevalidationResourceContext,
	type RevalidationResourceContext,
} from './revalidate-resources'
import {
	DEFAULT_REVALIDATE_DEADLINE_MS,
	DEFAULT_REVALIDATE_TRANSFER_BUDGET_BYTES,
	resolveRevalidateWorkerRuntimeConfig,
} from './revalidate-worker'

const idResolver = new IdResolver({ plcUrl: config.plcDirectoryUrl })
const logger = createLogger('firehose-service')

const STALL_THRESHOLD_MS = 30_000
const DEFAULT_MAX_PENDING_EVENTS = 10_000
const MAX_PENDING_EVENTS = 100_000
const DEFAULT_DRAIN_GRACE_PERIOD_MS = 30_000
const MAX_DRAIN_GRACE_PERIOD_MS = 5 * 60_000
const DEFAULT_MAX_CONCURRENCY = 5
const MAX_CONCURRENCY = 20

// Normal firehose site work uses the same bounded resource settings as replay
// work. Settings records are small, so keep their lock-held operation strict:
// one event cannot occupy a DB lock while waiting for two long PDS requests.
const MAX_SETTINGS_OPERATION_DEADLINE_MS = 30_000
const DEFAULT_SETTINGS_TRANSFER_BUDGET_BYTES = 1024 * 1024
const MAX_SETTINGS_TRANSFER_BUDGET_BYTES = DEFAULT_SETTINGS_TRANSFER_BUDGET_BYTES
const firehoseResourceConfig = resolveRevalidateWorkerRuntimeConfig()
const firehoseOperationDeadlineMs = firehoseResourceConfig.revalidationDeadlineMs ?? DEFAULT_REVALIDATE_DEADLINE_MS
const firehoseTransferBudgetBytes =
	firehoseResourceConfig.transferBudgetBytes ?? DEFAULT_REVALIDATE_TRANSFER_BUDGET_BYTES

function createFilesystemEventResources(upstreamSignal?: AbortSignal): RevalidationResourceContext {
	// Site updates keep per-request fetch timeouts and manifest/file limits, but
	// have no aggregate wall or transfer deadline. A large valid publication must
	// not fail because serial/parallel retries counted its bytes twice.
	return createRevalidationResourceContext(null, null, upstreamSignal)
}

function createSettingsEventResources(upstreamSignal?: AbortSignal): RevalidationResourceContext {
	return createRevalidationResourceContext(
		Math.min(firehoseOperationDeadlineMs, MAX_SETTINGS_OPERATION_DEADLINE_MS),
		Math.min(firehoseTransferBudgetBytes, MAX_SETTINGS_TRANSFER_BUDGET_BYTES),
		upstreamSignal,
	)
}

/** Run one scheduled event with one caller-owned context and always close it. */
export async function runWithFirehoseResourceContext<T>(
	createResources: () => RevalidationResourceContext,
	operation: (resources: RevalidationResourceContext) => Promise<T>,
): Promise<T> {
	const resources = createResources()
	try {
		return await operation(resources)
	} finally {
		resources.close()
	}
}

type FirehoseLifecycle = 'stopped' | 'running' | 'draining'

type FirehoseHandle = {
	destroy: () => void | Promise<void>
}

import { type CursorReservation, isValidFirehoseSeq, OrderedCursorTracker } from './firehose-cursor'
import {
	DurableReplayController,
	type DurableReplayFailureAttempt,
	destroyThenConnect,
	RelayCursorCoordinator,
	type RelayCursorStore,
	RelayFailureBudget,
	RelayGenerationGuard,
} from './firehose-relay'
import { type SchedulerDrainResult, SiteWorkScheduler } from './firehose-scheduler'

export { type CursorReservation, isValidFirehoseSeq, OrderedCursorTracker } from './firehose-cursor'
export {
	DEFAULT_DURABLE_REPLAY_BASE_DELAY_MS,
	DEFAULT_DURABLE_REPLAY_FAILURE_DECAY_MS,
	DEFAULT_DURABLE_REPLAY_MAX_DELAY_MS,
	DEFAULT_DURABLE_REPLAY_MAX_FAILURES,
	DurableReplayBackoff,
	type DurableReplayBackoffOptions,
	DurableReplayController,
	type DurableReplayControllerOptions,
	type DurableReplayFailureAttempt,
	destroyThenConnect,
	durableReplayBackoffCap,
	durableReplayFullJitter,
	type RelayCursorActivation,
	RelayCursorCoordinator,
	type RelayCursorLoad,
	type RelayCursorStore,
	RelayFailureBudget,
	RelayGenerationGuard,
} from './firehose-relay'
export { type SchedulerDrainResult, SiteWorkScheduler } from './firehose-scheduler'

async function settlesWithinGrace(operation: Promise<unknown>, gracePeriodMs: number): Promise<boolean> {
	if (gracePeriodMs <= 0) return false
	return await new Promise((resolve) => {
		const timer = setTimeout(() => resolve(false), gracePeriodMs)
		void operation.then(
			() => {
				clearTimeout(timer)
				resolve(true)
			},
			() => {
				clearTimeout(timer)
				resolve(false)
			},
		)
	})
}

interface StopAndDrainOptions {
	/** Overrides FIREHOSE_DRAIN_GRACE_MS for this stop operation. */
	gracePeriodMs?: number
}

interface FirehoseDrainResult extends SchedulerDrainResult {
	cursor?: number
	pendingCursorEvents: number
	replayRequired: boolean
}

function boundedPositiveInteger(value: number | string | undefined, fallback: number, maximum: number): number {
	const parsed = typeof value === 'number' ? value : Number(value)
	if (!Number.isSafeInteger(parsed) || parsed < 1 || parsed > maximum) return fallback
	return parsed
}

function configuredMaxPendingEvents(): number {
	return boundedPositiveInteger(config.firehoseMaxPendingEvents, DEFAULT_MAX_PENDING_EVENTS, MAX_PENDING_EVENTS)
}

function configuredMaxConcurrency(): number {
	return boundedPositiveInteger(config.firehoseMaxConcurrency, DEFAULT_MAX_CONCURRENCY, MAX_CONCURRENCY)
}

function configuredDrainGracePeriodMs(value?: number): number {
	if (value === undefined) return config.firehoseDrainGraceMs
	const parsed = value
	// Zero is an explicit opt-in to forced teardown and is useful for callers
	// which have their own external deadline.
	if (!Number.isSafeInteger(parsed) || parsed < 0 || parsed > MAX_DRAIN_GRACE_PERIOD_MS) {
		return DEFAULT_DRAIN_GRACE_PERIOD_MS
	}
	return parsed
}

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

function relayLabel(service: string): 'primary' | 'secondary' | 'configured' {
	if (service === config.firehoseService) return 'primary'
	if (service === config.firehoseServiceSecondary) return 'secondary'
	return 'configured'
}

function isDurableRevalidation(result: SiteRevalidationEnqueueResult): boolean {
	return result === 'enqueued' || result === 'deduplicated'
}

const DURABLE_REVALIDATION_CAPACITY_RETRY_BASE_MS = 250
const DURABLE_REVALIDATION_CAPACITY_RETRY_MAX_MS = 5_000

interface DurableRevalidationRetryOptions {
	/** Stop waiting when intake or the owning lifecycle is no longer active. */
	shouldContinue?: () => boolean
	/** Abort the capacity wait when the owning firehose lifecycle stops. */
	signal?: AbortSignal
	/** Timer seam for deterministic capacity/race tests. */
	wait?: (delayMs: number) => Promise<void>
	baseDelayMs?: number
	maxDelayMs?: number
}

async function waitForDurableCapacityRetry(
	delayMs: number,
	wait: ((delayMs: number) => Promise<void>) | undefined,
	signal: AbortSignal | undefined,
): Promise<void> {
	if (!signal) {
		await (wait ? wait(delayMs) : new Promise<void>((resolve) => setTimeout(resolve, delayMs)))
		return
	}
	if (signal.aborted) return

	await new Promise<void>((resolve, reject) => {
		let settled = false
		let timer: ReturnType<typeof setTimeout> | undefined
		const finish = (error?: unknown) => {
			if (settled) return
			settled = true
			if (timer !== undefined) clearTimeout(timer)
			signal.removeEventListener('abort', onAbort)
			if (error === undefined) resolve()
			else reject(error)
		}
		const onAbort = () => finish()
		signal.addEventListener('abort', onAbort, { once: true })
		const waiting = wait
			? wait(delayMs)
			: new Promise<void>((resolveTimer) => {
					timer = setTimeout(resolveTimer, delayMs)
				})
		void waiting.then(
			() => finish(),
			(error) => finish(error),
		)
	})
}

/**
 * Capacity is backpressure, not a failed handoff. Keep the event's handler
 * pending until the independent revalidation consumer frees a slot. The
 * caller's cursor reservation therefore cannot advance while the stream is
 * full, and bounded cursor/scheduler queues naturally stop relay intake.
 */
export async function retryDurableRevalidationUntilAvailable(
	enqueue: () => Promise<SiteRevalidationEnqueueResult>,
	options: DurableRevalidationRetryOptions = {},
): Promise<SiteRevalidationEnqueueResult> {
	const shouldContinue = options.shouldContinue ?? (() => true)
	const signal = options.signal
	const baseDelayMs =
		Number.isSafeInteger(options.baseDelayMs) && (options.baseDelayMs ?? 0) > 0
			? (options.baseDelayMs as number)
			: DURABLE_REVALIDATION_CAPACITY_RETRY_BASE_MS
	const maxDelayMs = Math.max(
		baseDelayMs,
		Number.isSafeInteger(options.maxDelayMs) && (options.maxDelayMs ?? 0) > 0
			? (options.maxDelayMs as number)
			: DURABLE_REVALIDATION_CAPACITY_RETRY_MAX_MS,
	)
	let retries = 0
	const active = () => !signal?.aborted && shouldContinue()

	while (active()) {
		const result = await enqueue()
		if (result !== 'capacity') return result
		if (!active()) return 'unavailable'
		const delayMs = Math.min(maxDelayMs, baseDelayMs * 2 ** Math.min(retries, 30))
		retries++
		await waitForDurableCapacityRetry(delayMs, options.wait, signal)
	}

	return 'unavailable'
}

const durableReplayController = new DurableReplayController()

async function requireDurableRevalidation(
	did: string,
	rkey: string,
	reason: string,
	lifecycleSignal: AbortSignal,
	sourceVersion = '',
): Promise<void> {
	// Do not route a full stream through requestReplay: that path invokes the
	// leader failure callback and stops this process's only consumer. Capacity
	// waits here, while the independently started worker remains alive to drain.
	const result = await retryDurableRevalidationUntilAvailable(
		() => enqueueSiteRevalidation(did, rkey, reason, sourceVersion),
		{
			shouldContinue: () => acceptingEvents,
			signal: lifecycleSignal,
		},
	)
	if (isDurableRevalidation(result)) {
		durableReplayController.recordDurableSuccess()
		return
	}
	throw new Error('Durable revalidation is unavailable')
}

const cursorTracker = new OrderedCursorTracker(configuredMaxPendingEvents())
const siteWorkScheduler = new SiteWorkScheduler(configuredMaxConcurrency())

// Track firehose health and lifecycle.
let lastEventTime = Date.now()
let isConnected = false
const relayFailureBudget = new RelayFailureBudget()
let activeService: string = config.firehoseService
let lifecycle: FirehoseLifecycle = 'stopped'
let acceptingEvents = false
// Each start gets a new signal. Accepted work keeps the old signal, so a
// later start cannot accidentally revive a stopped event's retry or IO.
let lifecycleAbortController = new AbortController()
let firehoseHandle: FirehoseHandle | null = null
let stallWatchdogHandle: ReturnType<typeof setInterval> | null = null
let statusLogHandle: ReturnType<typeof setInterval> | null = null
let activeFailureCallback: (() => void) | undefined
let stopAndDrainPromise: Promise<FirehoseDrainResult> | null = null
let relayTransitionPromise: Promise<void> | null = null
let relayDestroyFailed = false
const relayGenerationGuard = new RelayGenerationGuard()
const relayCursorCoordinator = new RelayCursorCoordinator(relayFingerprint)
const relayCursorStore: RelayCursorStore = {
	read: readDurableCursor,
	save: async (service, cursor) => (await saveDurableCursor(cursor, service)).kind === 'saved',
}
let relaySwitchPromise: Promise<void> | null = null

export function getCurrentSeq(): number | undefined {
	return cursorTracker.resumableCursor
}

export function getActiveService(): string {
	return activeService
}

export function getFirehoseHealth() {
	const draining = lifecycle === 'draining'
	return {
		connected: isConnected,
		lastEventTime,
		timeSinceLastEvent: Date.now() - lastEventTime,
		queueSize: siteWorkScheduler.queuedHandlers,
		activeHandlers: siteWorkScheduler.activeHandlers,
		pendingCursorEvents: cursorTracker.pendingCount,
		lifecycle,
		draining,
		consecutiveFailures: relayFailureBudget.consecutiveFailures,
		healthy: lifecycle === 'running' && isConnected && Date.now() - lastEventTime < 60_000,
	}
}

function isCurrentRelay(relayGeneration: number): boolean {
	return relayGenerationGuard.isCurrent(relayGeneration)
}

function canAcceptRelayEvent(relayGeneration: number): boolean {
	return acceptingEvents && isCurrentRelay(relayGeneration)
}

function isCommitWriteEvent(evt: Event | CommitEvt): evt is CommitEvt {
	return 'event' in evt && (evt.event === 'create' || evt.event === 'update' || evt.event === 'delete')
}

function scheduleCursorWork(
	siteKey: string,
	handler: () => Promise<void>,
	reservation: CursorReservation,
	relayGeneration: number,
	resources?: RevalidationResourceContext,
): void {
	const scheduledHandler = resources ? () => runWithFirehoseResourceContext(() => resources, handler) : handler
	let work: Promise<void>
	try {
		work = siteWorkScheduler.schedule(siteKey, scheduledHandler)
	} catch (error) {
		resources?.close()
		throw error
	}
	void work.then(
		() => {
			// Shutdown preserves the active generation so accepted work can advance
			// the final cursor. Relay replacement invalidates it and replays instead.
			if (isCurrentRelay(relayGeneration)) reservation.complete()
		},
		(error) => {
			if (!isCurrentRelay(relayGeneration)) return
			reservation.fail()
			logger.error('[Firehose] Site work requires replay', undefined, { errorKind: errorKind(error) })
			if (acceptingEvents) requestReplay()
		},
	)
}

async function reconcileFilesystemEvent(
	event: CommitEvt,
	resources?: RevalidationResourceContext,
): Promise<string | undefined> {
	const { did, rkey } = event
	if (event.event === 'delete') {
		await handleSiteDelete(did, rkey, undefined, resources)
		return undefined
	}

	// The CAR record is only a hint. handleSiteCreateOrUpdate fetches the
	// authoritative record after acquiring the site lock, so a delayed C1 cannot
	// commit after a newer C2. A confirmed absence is a no-op; transport or
	// validation failures are raised by the locked writer for revalidation.
	await handleSiteCreateOrUpdate(did, rkey, event.record as WispFsRecord, event.cid?.toString() || '', { resources })
	return undefined
}

async function processFilesystemEvent(
	event: CommitEvt,
	resources: RevalidationResourceContext,
	lifecycleSignal: AbortSignal,
): Promise<void> {
	const { did, rkey } = event
	let recoveryReason: string | undefined
	try {
		logger.debug('[place.wisp.fs] Processing event', { did, rkey, event: event.event })
		assertRevalidationActive(resources)
		recoveryReason = await reconcileFilesystemEvent(event, resources)
		assertRevalidationActive(resources)
		// Record only a completed reconciliation. The monotonic repo revision keeps
		// a replay of this same event from clearing a later DLQ fence.
		if (!(await recordSiteReconciliation(did, rkey, event.rev))) {
			throw new Error('Successful site reconciliation could not be recorded')
		}
	} catch (error) {
		logger.error('[place.wisp.fs] Error handling event', undefined, {
			did,
			rkey,
			event: event.event,
			errorKind: errorKind(error),
		})
		recoveryReason =
			event.event === 'delete' ? SITE_DELETE_TOMBSTONE_REASON : `firehose-processing-failed:${event.event}`
	} finally {
		resources.close()
	}

	if (recoveryReason) await requireDurableRevalidation(did, rkey, recoveryReason, lifecycleSignal, event.rev)
	logger.debug('[place.wisp.fs] Completed event', { did, rkey, event: event.event })
}

async function reconcileSettingsEvent(
	event: CommitEvt,
	resources?: RevalidationResourceContext,
): Promise<string | undefined> {
	const { did, rkey, record, cid } = event
	if (event.event === 'delete') {
		await handleSettingsDelete(did, rkey, undefined, resources)
		return undefined
	}
	if (record && validateSettingsRecord(record).success) {
		await handleSettingsUpdate(did, rkey, record as WispSettings, cid?.toString() || '', { resources })
		return undefined
	}

	logger.warn('[place.wisp.settings] Invalid record requires reconciliation', { did, rkey })
	return SETTINGS_UPDATE_FAILURE_REASON
}

async function processSettingsEvent(
	event: CommitEvt,
	resources: RevalidationResourceContext,
	lifecycleSignal: AbortSignal,
): Promise<void> {
	const { did, rkey } = event
	let recoveryReason: string | undefined
	try {
		recoveryReason = await reconcileSettingsEvent(event, resources)
	} catch (error) {
		logger.error('[place.wisp.settings] Error handling event', undefined, {
			did,
			rkey,
			event: event.event,
			errorKind: errorKind(error),
		})
		recoveryReason = event.event === 'delete' ? SETTINGS_DELETE_FAILURE_REASON : SETTINGS_UPDATE_FAILURE_REASON
	} finally {
		resources.close()
	}

	if (recoveryReason) await requireDurableRevalidation(did, rkey, recoveryReason, lifecycleSignal)
}

function dispatchCommitWork(event: CommitEvt, reservation: CursorReservation, relayGeneration: number): void {
	const { did, collection, rkey } = event
	const siteKey = `${did}/${rkey}`
	const lifecycleSignal = lifecycleAbortController.signal
	if (collection === 'place.wisp.fs') {
		logger.info('[place.wisp.fs] Received event', { did, rkey, event: event.event })
		const resources = createFilesystemEventResources(lifecycleSignal)
		scheduleCursorWork(
			siteKey,
			() => processFilesystemEvent(event, resources, lifecycleSignal),
			reservation,
			relayGeneration,
			resources,
		)
		return
	}
	if (collection === 'place.wisp.settings') {
		const resources = createSettingsEventResources(lifecycleSignal)
		scheduleCursorWork(
			siteKey,
			() => processSettingsEvent(event, resources, lifecycleSignal),
			reservation,
			relayGeneration,
			resources,
		)
		return
	}
	reservation.complete()
}

/**
 * Handle a firehose event. The reservation wait is deliberate: when the
 * bounded cursor queue is full, runtimes which await this handler slow intake
 * instead of allocating unbounded queued work.
 */
async function handleEvent(evt: Event | CommitEvt, relayGeneration: number): Promise<void> {
	if (!canAcceptRelayEvent(relayGeneration)) return
	if (!('seq' in evt) || !isValidFirehoseSeq(evt.seq)) {
		logger.warn('[Firehose] Ignoring event without a valid relay sequence')
		return
	}

	let reservation: CursorReservation | undefined
	try {
		reservation = await cursorTracker.reserve(evt.seq)
		if (!reservation || !canAcceptRelayEvent(relayGeneration)) {
			reservation?.fail()
			return
		}

		lastEventTime = Date.now()
		relayFailureBudget.recordEvent()
		if (!isCommitWriteEvent(evt)) {
			reservation.complete()
			return
		}
		dispatchCommitWork(evt, reservation, relayGeneration)
	} catch (error) {
		if (!isCurrentRelay(relayGeneration)) return
		reservation?.fail()
		logger.error('[Firehose] Unexpected event handling error', undefined, { errorKind: errorKind(error) })
		if (acceptingEvents) requestReplay()
	}
}

function getAlternateService(current: string): string | undefined {
	if (!config.firehoseServiceSecondary) return undefined
	return current === config.firehoseService ? config.firehoseServiceSecondary : config.firehoseService
}

function destroyCurrentRelay(invalidateGeneration = true): Promise<void> {
	if (invalidateGeneration) relayGenerationGuard.invalidate()
	const handle = firehoseHandle
	firehoseHandle = null
	isConnected = false
	if (!handle) return Promise.resolve()

	try {
		return Promise.resolve(handle.destroy()).catch((error) => {
			relayDestroyFailed = true
			logger.error('[Firehose] Relay destroy failed', undefined, { errorKind: errorKind(error) })
			throw error
		})
	} catch (error) {
		relayDestroyFailed = true
		logger.error('[Firehose] Relay destroy failed', undefined, { errorKind: errorKind(error) })
		return Promise.reject(error)
	}
}

/** Serialize destroy -> connect so an async old relay cannot overlap a new one. */
function reconnectRelay(allowDuringRelaySwitch = false): Promise<void> {
	if (relayTransitionPromise) return relayTransitionPromise

	const transition = destroyThenConnect(
		() => destroyCurrentRelay(),
		() => acceptingEvents && (allowDuringRelaySwitch || !relaySwitchPromise),
		() => connect(activeFailureCallback),
	).catch((error) => {
		logger.error('[Firehose] Relay transition failed', undefined, { errorKind: errorKind(error) })
		if (activeFailureCallback) activeFailureCallback()
		else void stopAndDrainFirehose()
	})

	relayTransitionPromise = transition
	void transition.then(() => {
		if (relayTransitionPromise === transition) relayTransitionPromise = null
	})
	return transition
}

function clearRuntimeTimers(): void {
	if (stallWatchdogHandle) {
		clearInterval(stallWatchdogHandle)
		stallWatchdogHandle = null
	}
	if (statusLogHandle) {
		clearInterval(statusLogHandle)
		statusLogHandle = null
	}
}

/**
 * Move between independent relay sequence spaces only after the old completed
 * prefix is durable and the target prefix has been read successfully.
 */
function requestRelayFailover(targetService: string, onFailure?: () => void): void {
	if (!acceptingEvents || relaySwitchPromise || durableReplayController.pending) return

	const sourceService = activeService
	const sourceCursor = getCurrentSeq()
	const priorRelayTransition = relayTransitionPromise
	// Fence callbacks before any await. Accepted site work may finish, but it
	// cannot advance the target relay's cursor generation. The existing relay
	// transition is awaited before destroying again, so its old relay cannot
	// overlap the checkpoint/destroy transition below.
	relayGenerationGuard.invalidate()

	const switching = (async () => {
		if (priorRelayTransition) await priorRelayTransition
		await destroyCurrentRelay(false)
		const activation = await relayCursorCoordinator.switchTo(targetService, sourceCursor, relayCursorStore)
		if (!activation) throw new Error('Durable relay checkpoint is unavailable')
		if (!acceptingEvents) return

		activeService = targetService
		cursorTracker.reset(activation.cursor)
		lastEventTime = Date.now()
		if (activation.missingCheckpoint) {
			logger.warn('[Firehose] Target relay has no checkpoint; starting live without cross-relay replay', {
				relay: relayLabel(targetService),
			})
		}
		await reconnectRelay(true)
	})()
		.catch((error) => {
			logger.error('[Firehose] Relay failover stopped because its checkpoint is uncertain', undefined, {
				from: relayLabel(sourceService),
				to: relayLabel(targetService),
				errorKind: errorKind(error),
			})
			if (onFailure) onFailure()
			else void stopAndDrainFirehose()
		})
		.finally(() => {
			if (relaySwitchPromise === switching) relaySwitchPromise = null
		})

	relaySwitchPromise = switching
}

function handleError(error: Error, onTooManyFailures?: () => void): void {
	if (!acceptingEvents) return

	logger.error('[Firehose] Connection error', undefined, {
		relay: relayLabel(activeService),
		errorKind: errorKind(error),
	})
	if (relayFailureBudget.recordConnectionError() < 3) return

	const alternate = getAlternateService(activeService)
	if (alternate && relayFailureBudget.canFailOver()) {
		logger.warn('[Firehose] Relay failures reached threshold; failing over', {
			from: relayLabel(activeService),
			to: relayLabel(alternate),
		})
		relayFailureBudget.recordFailOver()
		requestRelayFailover(alternate, onTooManyFailures)
		return
	}

	if (alternate) {
		logger.error('[Firehose] Both configured relays are failing', undefined, { relay: relayLabel(activeService) })
	} else {
		logger.warn('[Firehose] Relay failures reached threshold', { relay: relayLabel(activeService) })
	}
	onTooManyFailures?.()
}

function handleStall(onTooManyFailures?: () => void): void {
	if (!acceptingEvents || durableReplayController.pending || relaySwitchPromise) return

	const silenceMs = Date.now() - lastEventTime
	if (silenceMs < STALL_THRESHOLD_MS) return

	if (relayFailureBudget.recordStall() === 1) {
		logger.warn('[Firehose] Relay stalled; reconnecting from the safe cursor', { relay: relayLabel(activeService) })
		const cursor = getCurrentSeq()
		cursorTracker.reset(cursor)
		lastEventTime = Date.now()
		void reconnectRelay()
		return
	}

	const alternate = getAlternateService(activeService)
	if (alternate && relayFailureBudget.canFailOver()) {
		logger.warn('[Firehose] Relay remained stalled; failing over', {
			from: relayLabel(activeService),
			to: relayLabel(alternate),
		})
		relayFailureBudget.recordFailOver()
		requestRelayFailover(alternate, onTooManyFailures)
		return
	}

	logger.error('[Firehose] Relay stalled with no recoverable alternate', undefined, {
		relay: relayLabel(activeService),
	})
	onTooManyFailures?.()
}

function connect(onTooManyFailures: (() => void) | undefined = activeFailureCallback): void {
	if (!acceptingEvents || firehoseHandle) return

	const relayGeneration = relayGenerationGuard.beginConnection()
	const handleRelayEvent = (evt: Event | CommitEvt) => handleEvent(evt, relayGeneration)
	const isCurrentRelay = () => relayGenerationGuard.isCurrent(relayGeneration)
	logger.info('[Firehose] Connecting', { relay: relayLabel(activeService) })
	if (isBun) {
		const bunFirehose = new BunFirehose({
			idResolver,
			service: activeService,
			filterCollections: ['place.wisp.fs', 'place.wisp.settings'],
			handleEvent: handleRelayEvent,
			onError: (error: Error) => {
				if (isCurrentRelay()) handleError(error, onTooManyFailures)
			},
			getCursor: () => getCurrentSeq(),
			onConnect: () => {
				if (!acceptingEvents || !isCurrentRelay()) return
				isConnected = true
				relayFailureBudget.recordConnected()
				// A connected socket is not proof of a usable relay. The failure and
				// cross-relay budgets reset only when handleEvent receives an event.
				logger.info('[Firehose] Connected', { relay: relayLabel(activeService) })
			},
			onDisconnect: () => {
				if (!isCurrentRelay()) return
				isConnected = false
				if (acceptingEvents) logger.warn('[Firehose] Disconnected; relay will reconnect')
			},
		})
		void bunFirehose.start()
		firehoseHandle = { destroy: () => bunFirehose.destroy() }
		return
	}

	const nodeFirehose = new Firehose({
		idResolver,
		service: activeService,
		filterCollections: ['place.wisp.fs', 'place.wisp.settings'],
		handleEvent: handleRelayEvent as (evt: Event) => Promise<void>,
		onError: (error: Error) => {
			if (isCurrentRelay()) handleError(error, onTooManyFailures)
		},
		getCursor: () => getCurrentSeq(),
	})
	isConnected = true
	void nodeFirehose.start()
	firehoseHandle = { destroy: () => nodeFirehose.destroy() }
}

/** Fail closed when bounded durable-recovery replay cannot make progress. */
function failClosedAfterDurableReplayFailures(attempt: DurableReplayFailureAttempt): void {
	logger.error('[Firehose] Durable recovery replay failure budget exhausted', undefined, {
		consecutiveFailures: attempt.consecutiveFailures,
	})
	if (activeFailureCallback) {
		activeFailureCallback()
		return
	}
	void stopAndDrainFirehose()
}

/** Reconnect from the last safe cursor after durable recovery was unavailable. */
function requestReplay(): void {
	if (!acceptingEvents) return

	const attempt = durableReplayController.request(async () => {
		if (!acceptingEvents) return
		const cursor = getCurrentSeq()
		// Discard old reservations after retaining their safe prefix. Replayed
		// events receive fresh reservations; callbacks from old work are ignored.
		cursorTracker.reset(cursor)
		isConnected = false
		await reconnectRelay()
	}, failClosedAfterDurableReplayFailures)
	if (!attempt || attempt.terminal) return

	isConnected = false
	logger.warn('[Firehose] Durable recovery unavailable; replay reconnect scheduled', {
		consecutiveFailures: attempt.consecutiveFailures,
		delayMs: attempt.delayMs,
		delayCapMs: attempt.delayCapMs,
	})
}

function startFirehoseNow(initialCursor: number | undefined, onTooManyFailures?: () => void): void {
	logger.info('[Firehose] Starting', { runtime: isBun ? 'bun' : 'node', maxConcurrency: configuredMaxConcurrency() })
	if (config.firehoseServiceSecondary) logger.info('[Firehose] Secondary relay configured')
	if (initialCursor !== undefined) logger.info('[Firehose] Resuming from a saved cursor')

	stopAndDrainPromise = null
	relayDestroyFailed = false
	lifecycleAbortController = new AbortController()
	activeService = config.firehoseService
	activeFailureCallback = onTooManyFailures
	relayFailureBudget.reset()
	durableReplayController.reset()
	lastEventTime = Date.now()
	relayCursorCoordinator.initialize(activeService, initialCursor)
	cursorTracker.reset(initialCursor)
	acceptingEvents = true
	lifecycle = 'running'
	connect(onTooManyFailures)

	if (statusLogHandle) clearInterval(statusLogHandle)
	statusLogHandle = setInterval(() => {
		logger.info('[Firehose] Status check', {
			queueSize: siteWorkScheduler.queuedHandlers,
			activeHandlers: siteWorkScheduler.activeHandlers,
			pendingCursorEvents: cursorTracker.pendingCount,
		})
	}, 60 * 60_000)

	if (stallWatchdogHandle) clearInterval(stallWatchdogHandle)
	stallWatchdogHandle = setInterval(() => handleStall(onTooManyFailures), STALL_THRESHOLD_MS)
}

/**
 * Start the firehose worker. If a previous leader is still draining, startup is
 * delayed until it has drained successfully instead of overlapping handlers.
 */
export function startFirehose(initialCursor?: number, onTooManyFailures?: () => void): void {
	if (acceptingEvents) {
		logger.warn('[Firehose] Start requested while already running; ignoring duplicate request')
		return
	}

	const priorDrain = stopAndDrainPromise
	if (priorDrain) {
		void priorDrain.then((result) => {
			if (stopAndDrainPromise !== priorDrain) return
			if (result.forced) {
				logger.error('[Firehose] Not restarting after a forced drain', undefined, {
					remainingWork: result.remainingWork,
				})
				return
			}
			startFirehoseNow(initialCursor, onTooManyFailures)
		})
		return
	}

	startFirehoseNow(initialCursor, onTooManyFailures)
}

/**
 * Stop intake immediately, then wait for all accepted site work. The result is
 * explicit when the grace period expires so callers can choose a forced process
 * teardown without pretending that queued work drained.
 */
export function stopAndDrainFirehose(options: StopAndDrainOptions = {}): Promise<FirehoseDrainResult> {
	if (stopAndDrainPromise) return stopAndDrainPromise

	const gracePeriodMs = configuredDrainGracePeriodMs(options.gracePeriodMs)
	logger.info('[Firehose] Stopping intake and draining queued work', { gracePeriodMs })
	acceptingEvents = false
	lifecycle = 'draining'
	lifecycleAbortController.abort(new Error('Firehose lifecycle stopped'))
	isConnected = false
	relayFailureBudget.reset()
	durableReplayController.stop()
	clearRuntimeTimers()
	cursorTracker.stopAccepting()
	// No more reservations can arrive after this point, so the final commit can
	// become eligible for checkpointing as soon as its work completes.
	cursorTracker.sealOpenSequence()
	const replayReconnectPromise = durableReplayController.pending
	const hadRelayTransition = Boolean(
		firehoseHandle || replayReconnectPromise || relayTransitionPromise || relaySwitchPromise,
	)
	const relayDestroyed = destroyCurrentRelay(false)
	const relayStopping = Promise.all([
		relayDestroyed,
		replayReconnectPromise ?? Promise.resolve(),
		relayTransitionPromise ?? Promise.resolve(),
		relaySwitchPromise ?? Promise.resolve(),
	])

	stopAndDrainPromise = (async () => {
		const [workerDrain, relayStopped] = await Promise.all([
			siteWorkScheduler.drain(gracePeriodMs),
			hadRelayTransition ? settlesWithinGrace(relayStopping, gracePeriodMs) : Promise.resolve(true),
		])
		const drain: SchedulerDrainResult =
			workerDrain.forced || !relayStopped || relayDestroyFailed
				? { ...workerDrain, outcome: 'forced', forced: true }
				: workerDrain
		cursorTracker.sealOpenSequence()
		const result: FirehoseDrainResult = {
			...drain,
			cursor: getCurrentSeq(),
			pendingCursorEvents: cursorTracker.pendingCount,
			replayRequired: cursorTracker.pendingCount > 0,
		}
		lifecycle = 'stopped'

		if (result.forced) {
			logger.warn('[Firehose] Drain did not complete safely; forced teardown is required', {
				remainingWork: result.remainingWork,
				activeHandlers: result.activeHandlers,
				pendingCursorEvents: result.pendingCursorEvents,
				relayDestroyFailed,
			})
		} else {
			logger.info('[Firehose] Drain complete', {
				pendingCursorEvents: result.pendingCursorEvents,
				replayRequired: result.replayRequired,
			})
		}
		return result
	})()

	return stopAndDrainPromise
}

/**
 * Backward-compatible stop entry point. Callers that can await it receive the
 * same idempotent drain result; callback-only callers may safely ignore it.
 */
export function stopFirehose(): Promise<FirehoseDrainResult> {
	return stopAndDrainFirehose()
}