import { onceAsync, remainingShutdownTimeout, settleBeforeDeadline } from '@wispplace/graceful-shutdown'
import { createLogger, shutdownGrafanaExporters } from '@wispplace/observability'
import { config } from './config'
import {
	createReconciliationRetryScheduler,
	type ReconciliationRetryScheduler,
	runStartupBackfill,
} from './lib/backfill'
import * as database from './lib/db'
import * as delivery from './lib/delivery'
import {
	drainFirehose,
	getFirehoseHealth,
	refreshRegistryFromDatabase,
	refreshRegistryOwnerFromDatabase,
	startFirehose,
	stopFirehose,
} from './lib/firehose'
import { isWebhookServingReady } from './lib/readiness'
import { closeRedisPublisher, getRedisPublisherHealth } from './lib/redis'
import { canCloseSharedClients } from './lib/shutdown-policy'

const logger = createLogger('webhook-service')

type LifecyclePhase = 'starting' | 'reconciling-live' | 'live' | 'stopping' | 'stopped' | 'failed'

type HealthServer = ReturnType<typeof Bun.serve>

let healthServer: HealthServer | null = null
let startupPromise: Promise<void> | null = null
let phase: LifecyclePhase = 'starting'
let startupBackfill = { found: 0, failed: 0 }
let reconciliationRetryScheduler: ReconciliationRetryScheduler | null = null
let reconciliationSchedulerStartPromise: Promise<void> | null = null
let lastReconciliationSuccessAt: number | undefined
let initialBackfillRunning = false
let initialBackfillPromise: Promise<void> | null = null
// The bounded DB registry bootstrap can still be in flight before the detached
// initial PDS reconciliation promise exists.
let registryBootstrapPromise: Promise<void> | null = null
// Delivery/firehose startup has asynchronous cursor and transport work too.
let intakeStartupPromise: Promise<void> | null = null
let backfillInfrastructureHealthy = true
let backgroundBackfillStopped = false
let reconciliationAbortController: AbortController | null = null
let reconciliationGeneration = 0
let initialBackfillContinuationPending = false
let initialBackfillPasses = 0
let initialBackfillScanned = 0
let lastInitialBackfillCompletedAt: number | undefined
let initialBackfillRetryAttempt = 0

function cursorAgeSeconds(value: number | undefined): number | undefined {
	return value === undefined ? undefined : boundedNumber(Math.floor(value / 1000))
}

function boundedNumber(value: unknown): number {
	return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 ? Math.min(value, 1_000_000) : 0
}

async function reconciliationHealth(): Promise<{
	scanning: number
	failed: number
	retryInfrastructureHealthy: boolean
}> {
	const retryHealth = reconciliationRetryScheduler?.health
	// A pending or stopped scheduler is not ready. This prevents the
	// reconciling-live phase from briefly masking a stuck scheduler startup.
	const retryInfrastructureHealthy = retryHealth?.running === true && retryHealth.infrastructureHealthy
	const api = database as typeof database & {
		getWebhookOwnerReconciliationHealth?: () => Promise<{ scanning: number; failed: number }>
	}
	if (!api.getWebhookOwnerReconciliationHealth) {
		return { scanning: 0, failed: startupBackfill.failed, retryInfrastructureHealthy }
	}
	try {
		const status = await api.getWebhookOwnerReconciliationHealth()
		return {
			scanning: boundedNumber(status.scanning),
			failed: boundedNumber(status.failed),
			retryInfrastructureHealthy,
		}
	} catch {
		// A DB outage makes readiness degraded without disclosing its error.
		return { scanning: 1, failed: Math.max(1, startupBackfill.failed), retryInfrastructureHealthy: false }
	}
}

async function healthResponse(): Promise<Response> {
	const intake = getFirehoseHealth()
	const redis = getRedisPublisherHealth()
	const reconciliation = await reconciliationHealth()
	// A single owner may be unavailable or still retrying without making the
	// whole service unable to serve other tenants. Keep those counts visible in
	// the response, but gate readiness only on lifecycle, intake, and shared
	// infrastructure health.
	const healthy = isWebhookServingReady({
		phase,
		intakeHealthy: intake.healthy,
		backfillInfrastructureHealthy,
		retryInfrastructureHealthy: reconciliation.retryInfrastructureHealthy,
	})
	const body = {
		status: healthy ? 'healthy' : 'degraded',
		phase,
		// Counts and booleans only. Do not expose records, DIDs, URLs, cursors, or errors.
		backfill: {
			running: initialBackfillRunning,
			continuationPending: initialBackfillContinuationPending,
			passes: boundedNumber(initialBackfillPasses),
			scanned: boundedNumber(initialBackfillScanned),
			infrastructureHealthy: backfillInfrastructureHealthy,
			found: boundedNumber(startupBackfill.found),
			failed: reconciliation.failed,
			scanning: reconciliation.scanning,
			retryInfrastructureHealthy: reconciliation.retryInfrastructureHealthy,
			lastSuccessAt: lastReconciliationSuccessAt,
			lastInitialCompletionAt: lastInitialBackfillCompletedAt,
		},
		redis: {
			configured: redis.configured,
			connected: redis.connected,
			dropped: boundedNumber(redis.dropped),
		},
		intake: {
			connected: intake.connected,
			// Socket state and durable progress differ under bounded backpressure.
			progressing: intake.directProgressing && intake.backlinkProgressing && intake.registryProgressing,
			started: intake.started,
			queued: Math.min(intake.queued, config.intakeQueueMax),
			directConnected: intake.directConnected,
			backlinkConnected: intake.backlinkConnected,
			// Relay replay is finite: a cursor age approaching it loses events.
			cursorAgeSeconds: {
				direct: cursorAgeSeconds(intake.cursorAgeMs.direct),
				backlink: cursorAgeSeconds(intake.cursorAgeMs.backlink),
				registry: cursorAgeSeconds(intake.cursorAgeMs.registry),
			},
			admissionLimits: intake.admissionLimits,
			streamFailures: intake.streamFailures,
		},
	}
	return Response.json(body, { status: healthy ? 200 : 503, headers: { 'Cache-Control': 'no-store' } })
}

/** The only HTTP listener. It intentionally has no diagnostic/admin routes. */
export function startHealthServer(): HealthServer {
	if (healthServer) return healthServer
	healthServer = Bun.serve({
		hostname: config.healthHost,
		port: config.healthPort,
		async fetch(request) {
			const url = new URL(request.url)
			if (request.method === 'GET' && url.pathname === '/health') return healthResponse()
			if (request.method === 'GET' && url.pathname === '/live') {
				return Response.json(
					{ status: phase === 'stopping' || phase === 'stopped' || phase === 'failed' ? 'stopping' : 'live' },
					{ status: phase === 'failed' ? 503 : 200 },
				)
			}
			return new Response('Not Found', { status: 404 })
		},
	})
	return healthServer
}

async function startDeliveryWorker(): Promise<void> {
	const api = delivery as typeof delivery & { startWebhookDeliveryWorker?: () => Promise<void> | void }
	if (!api.startWebhookDeliveryWorker) throw new Error('Durable webhook delivery worker is unavailable')
	await api.startWebhookDeliveryWorker()
}

async function drainAndStopDeliveryWorker(timeoutMs = config.shutdownTimeoutMs): Promise<boolean> {
	const api = delivery as typeof delivery & {
		drainWebhookDeliveryWorker?: (timeoutMs?: number) => Promise<boolean>
		stopWebhookDeliveryWorker?: (timeoutMs?: number) => Promise<boolean> | boolean
	}
	// stop() owns the bounded grace period and then aborts active pinned attempts.
	// Do not run a separate full drain first or it can consume the abort budget.
	if (api.stopWebhookDeliveryWorker) return (await api.stopWebhookDeliveryWorker(timeoutMs)) !== false
	return api.drainWebhookDeliveryWorker ? await api.drainWebhookDeliveryWorker(timeoutMs) : true
}

function waitForBackfillRetry(signal: AbortSignal, attempt: number): Promise<void> {
	const exponent = Math.min(attempt, config.initialBackfillRetryMaxExponent)
	const ceiling = Math.min(config.initialBackfillRetryMaxMs, config.initialBackfillRetryMinMs * 2 ** exponent)
	const delay = Math.max(1, Math.floor(Math.random() * ceiling))
	return new Promise((resolve) => {
		const timer = setTimeout(resolve, delay)
		const abort = () => {
			clearTimeout(timer)
			resolve()
		}
		if (signal.aborted) abort()
		else signal.addEventListener('abort', abort, { once: true })
	})
}

async function closeBeforeDeadline(
	task: Promise<void>,
	deadline: number,
	timedOutMessage: string,
	failedMessage: string,
): Promise<void> {
	const result = await settleBeforeDeadline(task, deadline)
	if (result.status === 'timed-out') logger.warn(timedOutMessage)
	else if (result.status === 'rejected') logger.warn(failedMessage)
}

function ensureReconciliationRetryScheduler(
	onOwnerTransition: (did: string, status: 'scanning' | 'complete' | 'failed') => Promise<void>,
): Promise<void> {
	if (reconciliationRetryScheduler || backgroundBackfillStopped) return Promise.resolve()
	if (reconciliationSchedulerStartPromise) return reconciliationSchedulerStartPromise
	const scheduler = createReconciliationRetryScheduler({ onOwnerTransition })
	reconciliationSchedulerStartPromise = scheduler
		.start()
		.then(async () => {
			if (backgroundBackfillStopped) {
				await scheduler.stop({ timeoutMs: config.shutdownTimeoutMs })
				return
			}
			reconciliationRetryScheduler = scheduler
		})
		.finally(() => {
			reconciliationSchedulerStartPromise = null
		})
	return reconciliationSchedulerStartPromise
}

async function launchInitialBackfill(
	onOwnerTransition: (did: string, status: 'scanning' | 'complete' | 'failed') => Promise<void>,
): Promise<void> {
	const signal = reconciliationAbortController?.signal
	const generation = reconciliationGeneration
	if (!signal) {
		initialBackfillRunning = false
		return
	}
	const active = () => !backgroundBackfillStopped && !signal.aborted && reconciliationGeneration === generation
	initialBackfillRunning = true
	let startAfter: string | undefined
	let continuationLoaded = false
	let completedScanned = 0
	try {
		while (active()) {
			try {
				// Loading the durable continuation is part of the retryable controller,
				// not a pre-flight await that can wedge the detached lifecycle promise.
				if (!continuationLoaded) {
					startAfter = await database.loadWebhookBackfillContinuation()
					if (!active()) return
					continuationLoaded = true
				}
				let passScanned = 0
				const result = await runStartupBackfill({
					onOwnerTransition,
					startAfter,
					signal,
					onProgress: async (progress) => {
						if (!active()) return
						// The source reports cumulative pass progress. Replace the in-flight
						// portion rather than adding it on every page callback.
						passScanned = boundedNumber(progress.scanned)
						initialBackfillScanned = Math.min(1_000_000, completedScanned + passScanned)
						if (progress.nextOwnerCursor) {
							// Persist after each fully reconciled keyset page. If a later page
							// fails, replay starts at this durable boundary, never at process start.
							await database.saveWebhookBackfillContinuation(progress.nextOwnerCursor)
							if (!active()) return
							startAfter = progress.nextOwnerCursor
							initialBackfillContinuationPending = true
						} else {
							initialBackfillContinuationPending = progress.capped
						}
					},
				})
				if (!active()) return
				initialBackfillPasses++
				completedScanned = Math.min(1_000_000, completedScanned + passScanned)
				initialBackfillScanned = completedScanned
				initialBackfillRetryAttempt = 0
				backfillInfrastructureHealthy = true
				startupBackfill = {
					found: Math.min(1_000_000, startupBackfill.found + boundedNumber(result.found)),
					failed: Math.min(1_000_000, startupBackfill.failed + boundedNumber(result.failed)),
				}
				if (!active()) return
				if (result.nextOwnerCursor) {
					initialBackfillContinuationPending = true
					await database.saveWebhookBackfillContinuation(result.nextOwnerCursor)
					if (!active()) return
					startAfter = result.nextOwnerCursor
					continue
				}
				await ensureReconciliationRetryScheduler(onOwnerTransition)
				if (!active()) return
				// Clear only after the last page and post-scan state are durable.
				await database.clearWebhookBackfillContinuation()
				if (!active()) return
				initialBackfillContinuationPending = false
				lastInitialBackfillCompletedAt = Date.now()
				phase = 'live'
				return
			} catch {
				if (!active()) return
				// Keep live registry intake running and retry the bounded current page.
				// Do not abandon an infrastructure-failed initial reconciliation.
				backfillInfrastructureHealthy = false
				initialBackfillContinuationPending = true
				try {
					// Failed/stale owners continue independently as soon as the DB is reachable.
					await ensureReconciliationRetryScheduler(onOwnerTransition)
					if (!active()) return
				} catch {
					// The bounded initial retry below also recovers scheduler bootstrap.
				}
				if (!active()) return
				await waitForBackfillRetry(signal, initialBackfillRetryAttempt++)
			}
		}
	} finally {
		initialBackfillRunning = false
	}
}

async function runMain(): Promise<void> {
	startHealthServer()
	phase = 'starting'
	const onOwnerTransition = async (did: string, status: 'scanning' | 'complete' | 'failed') => {
		if (backgroundBackfillStopped) return
		if (status === 'complete') lastReconciliationSuccessAt = Date.now()
		// Reconciliation transitions swap only this owner's bounded active rows.
		// A full registry scan per owner is both stale-prone and O(owners × rows).
		await refreshRegistryOwnerFromDatabase(did)
	}
	// Start durable live registry intake immediately from its cursor (or retained
	// cursor zero). Reconciliation runs afterwards under DB live-event fences.
	const bootstrap = refreshRegistryFromDatabase()
	registryBootstrapPromise = bootstrap
	try {
		await bootstrap
	} finally {
		if (registryBootstrapPromise === bootstrap) registryBootstrapPromise = null
	}
	if (backgroundBackfillStopped) return
	const intakeStartup = (async () => {
		await startDeliveryWorker()
		if (backgroundBackfillStopped) {
			await drainAndStopDeliveryWorker(config.shutdownTimeoutMs)
			return
		}
		await startFirehose()
		if (backgroundBackfillStopped) stopFirehose()
	})()
	intakeStartupPromise = intakeStartup
	try {
		await intakeStartup
	} finally {
		if (intakeStartupPromise === intakeStartup) intakeStartupPromise = null
	}
	if (backgroundBackfillStopped) return
	reconciliationAbortController = new AbortController()
	void ensureReconciliationRetryScheduler(onOwnerTransition).catch(() => {
		if (!backgroundBackfillStopped) backfillInfrastructureHealthy = false
	})
	initialBackfillRunning = true // Set before exposing reconciling-live readiness.
	phase = 'reconciling-live'
	initialBackfillPromise = launchInitialBackfill(onOwnerTransition).catch(() => {
		// A final guard for programming errors outside the retry controller.
		backfillInfrastructureHealthy = false
		initialBackfillRunning = false
	})
	void initialBackfillPromise
	logger.info('Webhook service started')
}

export function main(): Promise<void> {
	if (!startupPromise) {
		startupPromise = runMain().catch(async (error) => {
			phase = 'failed'
			await shutdown('startup failure')
			throw error
		})
	}
	return startupPromise
}

/**
 * Idempotent, ordered shutdown used by signals and tests. Returns false if a
 * bounded cleanup left a producer active. Signal handlers use
 * that result as a generic non-zero exit status instead of closing shared clients
 * underneath the outstanding work.
 */
const runShutdown = onceAsync(async (signal: string): Promise<boolean> => {
	const shutdownDeadline = Date.now() + config.shutdownTimeoutMs
	// During early startup the registry snapshot is the active reconciliation
	// producer; afterwards the detached PDS backfill owns that role.
	const initialBackfill = initialBackfillPromise ?? registryBootstrapPromise ?? intakeStartupPromise
	phase = 'stopping'
	backgroundBackfillStopped = true
	reconciliationGeneration++
	const abortController = reconciliationAbortController
	reconciliationAbortController = null
	const scheduler = reconciliationRetryScheduler
	const schedulerStart = reconciliationSchedulerStartPromise
	reconciliationRetryScheduler = null
	// Begin cancellation now, but do not let PDS retry draining delay stopping
	// intake. It is awaited at a bounded point after accepted relay work drains.
	const schedulerStop = scheduler?.stop({
		timeoutMs: remainingShutdownTimeout(shutdownDeadline),
		...(abortController ? { signal: abortController.signal } : {}),
	})
	abortController?.abort()
	logger.info(`Webhook service stopping (${signal})`)

	// Stop the unauthenticated HTTP surface first, then stop accepting relay work.
	if (healthServer) {
		try {
			healthServer.stop(true)
		} catch {
			// Listener may already be stopped by a concurrent signal.
		}
		healthServer = null
	}
	stopFirehose()
	const initialBackfillResult = await settleBeforeDeadline(initialBackfill, shutdownDeadline)
	const initialBackfillSettled = initialBackfillResult.status !== 'timed-out'
	if (!initialBackfillSettled) logger.warn('Initial reconciliation did not settle before shutdown deadline')

	let intakeDrained = false
	const intakeDrain = drainFirehose(remainingShutdownTimeout(shutdownDeadline))
	const intakeDrainResult = await settleBeforeDeadline(intakeDrain, shutdownDeadline)
	if (intakeDrainResult.status === 'timed-out') {
		logger.warn('Webhook intake drain timed out; unfinished events remain behind their cursor')
	} else if (intakeDrainResult.status === 'rejected') {
		logger.warn('Webhook intake drain failed; unfinished events remain behind their cursor')
	} else {
		intakeDrained = intakeDrainResult.value
		if (!intakeDrained) logger.warn('Webhook intake drain timed out; unfinished events remain behind their cursor')
	}

	let deliveryStopped = false
	const deliveryStop = drainAndStopDeliveryWorker(remainingShutdownTimeout(shutdownDeadline))
	const deliveryStopResult = await settleBeforeDeadline(deliveryStop, shutdownDeadline)
	if (deliveryStopResult.status === 'timed-out') {
		logger.warn('Webhook delivery worker abort cleanup did not settle before shutdown deadline')
	} else if (deliveryStopResult.status === 'rejected') {
		logger.warn('Webhook delivery worker did not stop cleanly')
	} else {
		deliveryStopped = deliveryStopResult.value
		if (!deliveryStopped) logger.warn('Webhook delivery worker abort cleanup did not settle before shutdown deadline')
	}
	let schedulerDrained = true
	if (schedulerStop) {
		const schedulerStopResult = await settleBeforeDeadline(schedulerStop, shutdownDeadline)
		if (schedulerStopResult.status === 'timed-out') {
			schedulerDrained = false
			logger.warn('Reconciliation retry scheduler did not drain before shutdown deadline')
		} else if (schedulerStopResult.status === 'rejected') {
			schedulerDrained = false
			logger.warn('Reconciliation retry scheduler did not stop cleanly')
		} else {
			schedulerDrained = schedulerStopResult.value.drained
			if (!schedulerDrained) logger.warn('Reconciliation retry scheduler did not drain before shutdown deadline')
		}
	}
	// A scheduler can still be constructing when shutdown begins. Its startup
	// promise includes the stopped-path cleanup in ensureReconciliationRetryScheduler.
	const schedulerStartResult = await settleBeforeDeadline(schedulerStart, shutdownDeadline)
	if (schedulerStartResult.status === 'timed-out') {
		schedulerDrained = false
		logger.warn('Reconciliation retry scheduler startup did not settle before shutdown deadline')
	} else if (schedulerStartResult.status === 'rejected') {
		schedulerDrained = false
		logger.warn('Reconciliation retry scheduler startup did not stop cleanly')
	}
	const safeToCloseSharedClients = canCloseSharedClients({
		initialBackfillSettled,
		intakeDrained,
		schedulerDrained,
		deliveryStopped,
	})
	if (safeToCloseSharedClients) {
		await closeBeforeDeadline(
			closeRedisPublisher(),
			shutdownDeadline,
			'Redis publisher did not close before shutdown deadline',
			'Redis publisher did not close cleanly',
		)
		await closeBeforeDeadline(
			database.closeDatabase(),
			shutdownDeadline,
			'Database did not close before shutdown deadline',
			'Database did not close cleanly',
		)
		await closeBeforeDeadline(
			shutdownGrafanaExporters(),
			shutdownDeadline,
			'Observability did not close before shutdown deadline',
			'Observability did not close cleanly',
		)
	} else {
		// A callback ignored cancellation or its bounded drain. Leave all shared
		// clients open until process exit rather than close them beneath that work.
		logger.warn('Skipping shared-client close because webhook shutdown is incomplete')
	}
	phase = safeToCloseSharedClients ? 'stopped' : 'failed'
	return safeToCloseSharedClients
})

export function shutdown(signal = 'shutdown'): Promise<boolean> {
	return runShutdown(signal)
}

if (import.meta.main) {
	const exitAfterShutdown = (signal: string) => {
		void shutdown(signal)
			.then((safeToCloseSharedClients) => process.exit(safeToCloseSharedClients ? 0 : 1))
			.catch(() => process.exit(1))
	}
	process.once('SIGINT', () => exitAfterShutdown('SIGINT'))
	process.once('SIGTERM', () => exitAfterShutdown('SIGTERM'))
	void main().catch(() => {
		// Startup errors are intentionally not serialized: they can contain URLs or credentials.
		process.exit(1)
	})
}