import { createHash } from 'node:crypto'
import { DELETED_SITE_RECORD_CID } from '@wispplace/constants'
import type { SiteCache, SiteSettingsCache } from '@wispplace/database'
import { createLogger } from '@wispplace/observability'
import postgres from 'postgres'
import { config } from '../config'

const logger = createLogger('firehose-service')

const sql = postgres(config.databaseUrl, {
	max: 10,
	idle_timeout: 20,
	connect_timeout: 10,
})

/**
 * Dedicated pool for advisory locks.
 *
 * A site-write lock is held for the entire duration of a site sync, which
 * includes minutes of blob downloads. Holding those long-lived locks on the
 * main query pool would starve ordinary queries (the connection-pooling
 * starvation class of bug). Each slot is a one-connection postgres client, so
 * it remains pinned for the holder and can be physically closed on quarantine.
 * While a slot is held its session is kept alive by lock heartbeats (see
 * withReservedSiteWriteLock) so proxies cannot idle-kill it mid-sync.
 */
const LOCK_POOL_SIZE = 10
/** A reserve must never wait forever for a leaked or wedged lock holder. */
export const LOCK_POOL_RESERVE_TIMEOUT_MS = 30_000
/**
 * Grace period, in seconds, before a retiring lock-pool root client
 * force-destroys its sockets. postgres.js defers sub-1KiB protocol writes
 * through `setImmediate` and schedules writes without re-checking the socket,
 * so its immediate-destroy path (`end({ timeout: 0 })`) can race or follow a
 * cleared socket and crash the process with an uncaught `socket.write`
 * TypeError outside every Promise rejection (observed at postgres@3.4.9
 * connection.js:255 on Bun 1.3.14; no fixed release exists upstream). A
 * positive timeout lets the driver's graceful end path settle queued writes
 * first, while `end` still resolves within the bound because postgres.js
 * force-destroys sockets after the timeout. Quarantined advisory-lock sessions
 * therefore still physically close.
 */
export const LOCK_POOL_RETIRE_TIMEOUT_SECONDS = 1
/**
 * Bound on how long a blocked advisory-lock acquisition may wait. Proxies in
 * front of PostgreSQL idle-kill sessions after 60s without traffic, and a
 * blocked acquisition cannot heartbeat on its own session, so any wait beyond
 * that bound would die as CONNECTION_CLOSED anyway; failing at 45s keeps the
 * failure a clean, retryable lock timeout.
 */
export const LOCK_ACQUIRE_LOCK_TIMEOUT = '45s'
/**
 * Interval between liveness probes sent on a reserved advisory-lock session
 * while the lock callback runs. Proxies in front of PostgreSQL idle-kill
 * sessions after 60s without traffic, and a lock callback legitimately stays
 * idle for minutes while blobs download; a probe well below that bound keeps
 * the session (and therefore the session-scoped advisory lock) alive.
 */
export const LOCK_HEARTBEAT_INTERVAL_MS = 15_000
/**
 * A liveness probe that cannot complete within this bound means the lock
 * session (and with it the advisory lock) must be presumed lost.
 */
export const LOCK_HEARTBEAT_TIMEOUT_MS = 10_000

interface SiteWriteLockQuery<T = unknown> extends PromiseLike<T> {
	/** postgres.js exposes this on pending queries to send a cancel request. */
	cancel?: () => unknown
}

export interface SiteWriteLockReservedClient {
	(strings: TemplateStringsArray, ...values: unknown[]): SiteWriteLockQuery<unknown>
	release(): void
}

export interface SiteWriteLockPoolClient {
	reserve(): Promise<SiteWriteLockReservedClient>
	end(options: { timeout: number }): Promise<void>
}

const createSiteWriteLockPoolClient = (): SiteWriteLockPoolClient =>
	postgres(config.databaseUrl, {
		max: 1,
		// A reserved session can be intentionally idle while files download. Do
		// not let the driver drop its session-scoped advisory lock in that gap.
		idle_timeout: 0,
		max_lifetime: null,
		connect_timeout: 10,
	}) as unknown as SiteWriteLockPoolClient

export interface SiteWriteLockConnection {
	(strings: TemplateStringsArray, ...values: unknown[]): SiteWriteLockQuery<unknown>
	release(): void
	close(options: { timeout: number }): Promise<void>
}

interface LockPoolWaiter {
	resolve(connection: SiteWriteLockConnection): void
	reject(error: Error): void
	active: boolean
	deadline: number
	timer: ReturnType<typeof setTimeout>
	settled: boolean
	signal?: AbortSignal
	abortListener?: () => void
}

interface SiteWriteLockPoolOptions {
	/** Number of one-connection postgres clients kept in this pool. */
	size?: number
	/** Maximum time spent waiting for a reserved slot/session. */
	reserveTimeoutMs?: number
	/** Test seam; production creates a max=1 postgres client. */
	createClient?: () => SiteWriteLockPoolClient
}

/**
 * A bounded pool of reserved postgres sessions.
 *
 * The object returned to a lock holder is a small forwarding wrapper. It is
 * deliberately not the postgres client itself: postgres.js owns its `release`
 * method and its root client can be ended when a session is quarantined.
 */
interface LockPoolSlot {
	client: SiteWriteLockPoolClient
	endPromise: Promise<void> | null
}

interface LockPoolDirectActivation {
	settled: boolean
	reject(error: Error): void
	abortListener?: () => void
}

export class SiteWriteLockPool {
	private readonly slots = new Set<LockPoolSlot>()
	private readonly retiring = new Set<LockPoolSlot>()
	private readonly available: LockPoolSlot[] = []
	private readonly waiters: LockPoolWaiter[] = []
	private readonly assignedWaiters = new Set<LockPoolWaiter>()
	private readonly directActivations = new Set<LockPoolDirectActivation>()
	private readonly size: number
	private readonly reserveTimeoutMs: number
	private readonly createClientFactory: () => SiteWriteLockPoolClient
	private ending: Promise<void> | null = null
	private closed = false

	constructor(options: SiteWriteLockPoolOptions = {}) {
		const size = options.size ?? LOCK_POOL_SIZE
		this.size = Number.isSafeInteger(size) && size > 0 ? size : LOCK_POOL_SIZE
		const reserveTimeoutMs = options.reserveTimeoutMs ?? LOCK_POOL_RESERVE_TIMEOUT_MS
		this.reserveTimeoutMs =
			Number.isFinite(reserveTimeoutMs) && reserveTimeoutMs >= 0 ? reserveTimeoutMs : LOCK_POOL_RESERVE_TIMEOUT_MS
		this.createClientFactory = options.createClient ?? createSiteWriteLockPoolClient
	}

	reserve(signal?: AbortSignal): Promise<SiteWriteLockConnection> {
		if (this.closed) return Promise.reject(new Error('Site write lock pool is closed'))
		if (signal?.aborted) return Promise.reject(abortReason(signal))

		const deadline = Date.now() + this.reserveTimeoutMs
		const slot = this.available.pop()
		if (slot) return this.activateForCaller(slot, deadline, signal)

		if (this.slots.size < this.size) {
			try {
				return this.activateForCaller(this.createSlot(), deadline, signal)
			} catch (error) {
				return Promise.reject(error instanceof Error ? error : new Error('Failed to create lock client'))
			}
		}

		return new Promise<SiteWriteLockConnection>((resolve, reject) => {
			const waiter = {} as LockPoolWaiter
			waiter.resolve = resolve
			waiter.reject = reject
			waiter.active = true
			waiter.deadline = deadline
			waiter.settled = false
			waiter.signal = signal
			waiter.timer = this.startWaiterTimer(waiter)
			if (signal) {
				waiter.abortListener = () => this.abortWaiter(waiter, abortReason(signal))
				signal.addEventListener('abort', waiter.abortListener, { once: true })
			}
			this.waiters.push(waiter)
			if (signal?.aborted) this.abortWaiter(waiter, abortReason(signal))
		})
	}

	end(options: { timeout: number }): Promise<void> {
		if (this.ending) return this.ending

		this.closed = true
		const closeError = new Error('Site write lock pool is closed')
		for (const waiter of this.waiters.splice(0)) this.settleWaiter(waiter, closeError)
		for (const waiter of this.assignedWaiters) {
			this.assignedWaiters.delete(waiter)
			this.settleWaiter(waiter, closeError)
		}
		for (const activation of this.directActivations) activation.reject(closeError)

		// Keep every slot tracked until its physical root client has ended. This
		// includes available, leased, and already-retiring/quarantined sessions.
		const slots = [...this.slots]
		this.ending = Promise.all(slots.map((slot) => this.retire(slot, options, false))).then(() => undefined)
		return this.ending
	}

	private createSlot(): LockPoolSlot {
		const slot = { client: this.createClientFactory(), endPromise: null }
		this.slots.add(slot)
		return slot
	}

	private activateForCaller(
		slot: LockPoolSlot,
		deadline: number,
		signal?: AbortSignal,
	): Promise<SiteWriteLockConnection> {
		return new Promise<SiteWriteLockConnection>((resolve, reject) => {
			const activation: LockPoolDirectActivation = {
				settled: false,
				reject: (error) => {
					if (activation.settled) return
					activation.settled = true
					if (signal && activation.abortListener) signal.removeEventListener('abort', activation.abortListener)
					activation.abortListener = undefined
					reject(error)
				},
			}
			this.directActivations.add(activation)
			if (signal) {
				activation.abortListener = () => activation.reject(abortReason(signal))
				signal.addEventListener('abort', activation.abortListener, { once: true })
			}
			void this.activate(slot, deadline, signal).then(
				(connection) => {
					this.directActivations.delete(activation)
					if (signal && activation.abortListener) signal.removeEventListener('abort', activation.abortListener)
					if (activation.settled || this.closed) {
						void connection.close({ timeout: LOCK_POOL_RETIRE_TIMEOUT_SECONDS }).catch(() => undefined)
						return
					}
					activation.settled = true
					resolve(connection)
				},
				(error) => {
					this.directActivations.delete(activation)
					if (signal && activation.abortListener) signal.removeEventListener('abort', activation.abortListener)
					if (activation.settled) return
					activation.settled = true
					reject(error instanceof Error ? error : new Error('Failed to reserve lock client'))
				},
			)
		})
	}

	private activate(slot: LockPoolSlot, deadline: number, signal?: AbortSignal): Promise<SiteWriteLockConnection> {
		const timeoutMs = Math.max(0, deadline - Date.now())
		return this.reserveClient(slot.client, timeoutMs, signal)
			.then((reserved) => {
				if (this.closed || this.retiring.has(slot)) {
					try {
						reserved.release()
					} catch {
						// The root client is being retired below.
					}
					void this.retire(slot, { timeout: LOCK_POOL_RETIRE_TIMEOUT_SECONDS }).catch(() => undefined)
					throw new Error('Site write lock pool is closed')
				}
				return this.reservation(slot, reserved)
			})
			.catch((error) => {
				// Retirement is tracked separately so a reserve timeout remains
				// bounded even when closing the physical root is slow.
				if (!this.retiring.has(slot))
					void this.retire(slot, { timeout: LOCK_POOL_RETIRE_TIMEOUT_SECONDS }).catch(() => undefined)
				throw error
			})
	}

	private reserveClient(
		client: SiteWriteLockPoolClient,
		timeoutMs: number,
		signal?: AbortSignal,
	): Promise<SiteWriteLockReservedClient> {
		if (signal?.aborted) return Promise.reject(abortReason(signal))
		let pending: Promise<SiteWriteLockReservedClient>
		try {
			pending = Promise.resolve(client.reserve())
		} catch (error) {
			return Promise.reject(error)
		}

		return new Promise<SiteWriteLockReservedClient>((resolve, reject) => {
			let settled = false
			const timer = setTimeout(() => {
				if (settled) return
				settled = true
				cleanup()
				reject(new Error('Timed out reserving site write lock connection'))
				releaseWhenReady()
			}, timeoutMs)
			;(timer as unknown as { unref?: () => void }).unref?.()
			const cleanup = () => {
				clearTimeout(timer)
				if (signal) signal.removeEventListener('abort', onAbort)
			}
			const releaseWhenReady = () => {
				// A driver reserve can still resolve after our bound. Release that
				// wrapper rather than leaving the physical session pinned.
				void pending.then(
					(reserved) => {
						try {
							reserved.release()
						} catch {
							// The root client is being retired by activate's rejection path.
						}
					},
					() => undefined,
				)
			}
			const onAbort = () => {
				if (settled) return
				settled = true
				cleanup()
				reject(abortReason(signal as AbortSignal))
				releaseWhenReady()
			}
			if (signal) signal.addEventListener('abort', onAbort, { once: true })
			pending.then(
				(reserved) => {
					if (settled) return
					settled = true
					cleanup()
					resolve(reserved)
				},
				(error) => {
					if (settled) return
					settled = true
					cleanup()
					reject(error)
				},
			)
			if (signal?.aborted) onAbort()
		})
	}

	private reservation(slot: LockPoolSlot, reserved: SiteWriteLockReservedClient): SiteWriteLockConnection {
		let finished = false
		// Forward queries through a new function object. Do not attach lifecycle
		// methods to either postgres.js object; the driver uses those methods to
		// manage its own connection queues.
		const connection = ((strings: TemplateStringsArray, ...values: unknown[]) =>
			reserved(strings, ...values)) as SiteWriteLockConnection

		connection.release = () => {
			if (finished) return
			finished = true
			try {
				reserved.release()
			} catch (error) {
				// A release failure leaves session state uncertain just like an unlock
				// failure. Quarantine this one slot and continue dispatching waiters.
				void this.retire(slot, { timeout: LOCK_POOL_RETIRE_TIMEOUT_SECONDS }).catch(() => undefined)
				logger.error('[DB] Failed to release reserved site-write connection', undefined, {
					errorKind: databaseErrorKind(error),
				})
				return
			}
			this.returnSlot(slot)
		}

		connection.close = async (options): Promise<void> => {
			if (finished) return
			finished = true
			await this.retire(slot, options)
		}

		return connection
	}

	private startWaiterTimer(waiter: LockPoolWaiter): ReturnType<typeof setTimeout> {
		const remaining = Math.max(0, waiter.deadline - Date.now())
		const timer = setTimeout(() => this.expireWaiter(waiter), remaining)
		;(timer as unknown as { unref?: () => void }).unref?.()
		return timer
	}

	private clearWaiterAbortListener(waiter: LockPoolWaiter): void {
		if (waiter.signal && waiter.abortListener) waiter.signal.removeEventListener('abort', waiter.abortListener)
		waiter.abortListener = undefined
	}

	private settleWaiter(waiter: LockPoolWaiter, error: Error): void {
		if (waiter.settled) return
		waiter.active = false
		waiter.settled = true
		clearTimeout(waiter.timer)
		this.clearWaiterAbortListener(waiter)
		waiter.reject(error)
	}

	private abortWaiter(waiter: LockPoolWaiter, error: Error): void {
		if (waiter.settled) return
		const index = this.waiters.indexOf(waiter)
		if (index >= 0) this.waiters.splice(index, 1)
		this.assignedWaiters.delete(waiter)
		this.settleWaiter(waiter, error)
	}

	private expireWaiter(waiter: LockPoolWaiter): void {
		if (!waiter.active) return
		const index = this.waiters.indexOf(waiter)
		if (index >= 0) this.waiters.splice(index, 1)
		this.settleWaiter(waiter, new Error('Timed out waiting for a site write lock connection'))
	}

	private takeWaiter(clearTimer: boolean): LockPoolWaiter | undefined {
		while (this.waiters.length > 0) {
			const waiter = this.waiters.shift()
			if (!waiter?.active) continue
			waiter.active = false
			if (clearTimer) clearTimeout(waiter.timer)
			return waiter
		}
		return undefined
	}

	private returnSlot(slot: LockPoolSlot): void {
		if (this.closed || this.retiring.has(slot)) return
		const waiter = this.takeWaiter(true)
		if (waiter) {
			this.assignSlot(slot, waiter)
			return
		}
		this.available.push(slot)
	}

	private assignSlot(slot: LockPoolSlot, waiter: LockPoolWaiter): void {
		clearTimeout(waiter.timer)
		if (waiter.deadline <= Date.now()) {
			this.settleWaiter(waiter, new Error('Timed out waiting for a site write lock connection'))
			void this.retire(slot, { timeout: LOCK_POOL_RETIRE_TIMEOUT_SECONDS }).catch(() => undefined)
			return
		}
		this.assignedWaiters.add(waiter)
		void this.activate(slot, waiter.deadline, waiter.signal).then(
			(connection) => {
				this.assignedWaiters.delete(waiter)
				if (waiter.settled || this.closed) {
					this.clearWaiterAbortListener(waiter)
					void connection.close({ timeout: LOCK_POOL_RETIRE_TIMEOUT_SECONDS }).catch(() => undefined)
					return
				}
				waiter.settled = true
				this.clearWaiterAbortListener(waiter)
				waiter.resolve(connection)
			},
			(error) => {
				this.assignedWaiters.delete(waiter)
				if (waiter.settled) return
				this.settleWaiter(waiter, error instanceof Error ? error : new Error('Failed to reserve lock client'))
				this.dispatch()
			},
		)
	}

	private removeAvailableSlot(slot: LockPoolSlot): void {
		const index = this.available.indexOf(slot)
		if (index >= 0) this.available.splice(index, 1)
	}

	private retire(slot: LockPoolSlot, options: { timeout: number }, dispatch = true): Promise<void> {
		if (slot.endPromise) return slot.endPromise
		this.retiring.add(slot)
		this.removeAvailableSlot(slot)
		slot.endPromise = Promise.resolve()
			.then(() => slot.client.end(options))
			.finally(() => {
				this.slots.delete(slot)
				this.retiring.delete(slot)
				if (dispatch && !this.closed) this.dispatch()
			})
		return slot.endPromise
	}

	private dispatch(): void {
		if (this.closed) return

		while (this.waiters.length > 0) {
			const waiter = this.takeWaiter(false)
			if (!waiter) continue

			let slot = this.available.pop()
			if (!slot) {
				if (this.slots.size >= this.size) {
					// No slot is currently available. Put this live waiter back without
					// resetting its original deadline.
					waiter.active = true
					this.waiters.unshift(waiter)
					return
				}
				try {
					slot = this.createSlot()
				} catch (error) {
					waiter.reject(error instanceof Error ? error : new Error('Failed to create lock client'))
					continue
				}
			}

			this.assignSlot(slot, waiter)
		}
	}
}

const lockPool = new SiteWriteLockPool()

/**
 * The unified per-site write-lock key. Shared verbatim with other cache
 * writers so all writers to a site's cache mutually exclude.
 */
function siteWriteLockKey(did: string, rkey: string): string {
	return `site-write:${did}:${rkey}`
}

/**
 * Derive the signed 63-bit PostgreSQL advisory-lock id without converting it
 * through JavaScript Number, which would lose high bits above 2^53 - 1.
 */
export function siteWriteLockId(did: string, rkey: string): bigint {
	const hash = createHash('sha256').update(siteWriteLockKey(did, rkey)).digest('hex')
	return BigInt(`0x${hash.substring(0, 16)}`) & 0x7fffffffffffffffn
}

function databaseErrorKind(error: unknown): string {
	if (!(error instanceof Error)) return 'UnknownError'
	return error.constructor.name || 'Error'
}

function abortReason(signal: AbortSignal): Error {
	return signal.reason instanceof Error ? signal.reason : new Error('Site write lock aborted')
}

/**
 * Execute one lock-acquisition query with postgres.js cancellation. The query
 * starts only after the preflight check, and an abort rejects immediately even
 * when a minimal test seam does not expose `.cancel()`; the caller then closes
 * the reserved session instead of returning a possibly lock-owning connection.
 */
function runCancellableLockQuery<T>(start: () => SiteWriteLockQuery<T>, signal?: AbortSignal): Promise<T> {
	if (signal?.aborted) return Promise.reject(abortReason(signal))
	let query: SiteWriteLockQuery<T>
	try {
		query = start()
	} catch (error) {
		return Promise.reject(error)
	}
	return new Promise<T>((resolve, reject) => {
		let settled = false
		const cleanup = () => {
			if (signal) signal.removeEventListener('abort', onAbort)
		}
		const finish = (callback: () => void) => {
			if (settled) return
			settled = true
			cleanup()
			callback()
		}
		const onAbort = () => {
			if (settled) return
			try {
				const cancellation = query.cancel?.()
				if (cancellation) void Promise.resolve(cancellation).catch(() => undefined)
			} catch {
				// The reserved connection is closed by the caller below.
			}
			finish(() => reject(abortReason(signal as AbortSignal)))
		}
		if (signal) signal.addEventListener('abort', onAbort, { once: true })
		Promise.resolve(query).then(
			(value) => finish(() => resolve(value)),
			(error) => finish(() => reject(error)),
		)
		if (signal?.aborted) onAbort()
	})
}

/** postgres.js decodes a boolean result as `true`; anything else is unsafe. */
function advisoryUnlockReturnedTrue(result: unknown): boolean {
	if (!Array.isArray(result) || result.length === 0) return false
	const row = result[0]
	if (typeof row !== 'object' || row === null) return false
	const value =
		(row as { unlocked?: unknown; pg_advisory_unlock?: unknown }).unlocked ??
		(row as { pg_advisory_unlock?: unknown }).pg_advisory_unlock
	return value === true
}

export class SiteWriteLockAcquisitionError extends Error {
	readonly errorKind: string

	constructor(cause: unknown) {
		super('Failed to acquire site-write lock')
		this.name = 'SiteWriteLockAcquisitionError'
		this.errorKind = databaseErrorKind(cause)
	}
}

/**
 * The advisory-lock session died while the lock callback was running, so the
 * session-scoped lock was released by the server and mutual exclusion can no
 * longer be guaranteed. The callback receives an aborting signal the moment the
 * loss is detected; this error means its writes after that point may have raced
 * with another writer and the work should be retried.
 */
export class SiteWriteLockLostError extends Error {
	readonly errorKind: string

	constructor(cause: unknown) {
		super('Site write lock session was lost while the callback ran')
		this.name = 'SiteWriteLockLostError'
		this.errorKind = databaseErrorKind(cause)
	}
}

/** Liveness-probe tuning for a held site-write lock; production uses the defaults. */
export interface LockHeartbeatOptions {
	/** Interval between probes; defaults to {@link LOCK_HEARTBEAT_INTERVAL_MS}. */
	intervalMs?: number
	/** A probe slower than this bound presumes the lock session is lost. */
	timeoutMs?: number
}

interface LockHeartbeatMonitor {
	stop(): void
}

/**
 * Probe the reserved session on a fixed interval while a lock callback runs.
 * Each probe is an ordinary query on the same session; anything but a prompt
 * success means the session — and with it the session-scoped advisory lock —
 * must be presumed lost. Probes are crash-safe only because the patched
 * postgres.js rejects queries on a closed socket instead of throwing inside a
 * `setImmediate` write callback (see patches/postgres@3.4.9.patch).
 */
function startLockHeartbeat(
	conn: SiteWriteLockConnection,
	options: LockHeartbeatOptions,
	onLost: (error: unknown) => void,
): LockHeartbeatMonitor {
	const intervalMs = options.intervalMs ?? LOCK_HEARTBEAT_INTERVAL_MS
	const timeoutMs = options.timeoutMs ?? LOCK_HEARTBEAT_TIMEOUT_MS
	let timer: ReturnType<typeof setInterval> | null = null
	let probing = false
	let stopped = false
	const tick = () => {
		if (stopped || probing) return
		probing = true
		const timeout = new Promise<never>((_resolve, reject) => {
			const handle = setTimeout(() => reject(new Error('Site write lock heartbeat timed out')), timeoutMs)
			;(handle as unknown as { unref?: () => void }).unref?.()
		})
		// A probe that loses the race must not surface as unhandled later.
		void timeout.catch(() => undefined)
		void Promise.race([runCancellableLockQuery(() => conn`SELECT 1`), timeout]).then(
			() => {
				probing = false
			},
			(error) => {
				stopped = true
				if (timer) clearInterval(timer)
				onLost(error)
			},
		)
	}
	timer = setInterval(tick, intervalMs)
	;(timer as unknown as { unref?: () => void }).unref?.()
	return {
		stop: () => {
			stopped = true
			if (timer) clearInterval(timer)
			timer = null
		},
	}
}

/**
 * Run `fn` on an already-reserved connection while holding the per-site lock.
 * `fn` receives a signal that aborts when the caller aborts or when the lock
 * session is lost, so long downloads can stop instead of writing without
 * mutual exclusion. The connection is released after normal
 * acquisition/callback paths. An unlock failure or a detected session loss
 * closes its physical session so a potentially lock-owning backend cannot be
 * reused or permanently consume a pool slot.
 */
export async function withReservedSiteWriteLock<T>(
	conn: SiteWriteLockConnection,
	did: string,
	rkey: string,
	fn: (signal: AbortSignal) => Promise<T>,
	signal?: AbortSignal,
	heartbeat: LockHeartbeatOptions = {},
): Promise<T> {
	const lockId = siteWriteLockId(did, rkey)
	let held = false
	let lockLost = false
	let connectionSafeToRelease = true
	const lockLoss = new AbortController()
	const safetySignal = signal ? AbortSignal.any([signal, lockLoss.signal]) : lockLoss.signal
	try {
		try {
			await runCancellableLockQuery(() => conn`SET lock_timeout = '45s'`, signal)
			await runCancellableLockQuery(() => conn`SELECT pg_advisory_lock(${lockId}::bigint)`, signal)
			held = true
		} catch (error) {
			if (signal?.aborted) {
				// Cancellation can race a server-side lock grant. Never return this
				// session to the pool unless it has been physically closed.
				connectionSafeToRelease = false
				try {
					await conn.close({ timeout: LOCK_POOL_RETIRE_TIMEOUT_SECONDS })
				} catch (closeError) {
					logger.error('[DB] Failed to close aborted site-write lock connection', undefined, {
						did,
						rkey,
						errorKind: databaseErrorKind(closeError),
					})
				}
			}
			logger.error('[DB] Failed to acquire site-write lock', undefined, {
				did,
				rkey,
				errorKind: databaseErrorKind(error),
			})
			throw new SiteWriteLockAcquisitionError(error)
		}

		const monitor = startLockHeartbeat(conn, heartbeat, (error) => {
			lockLost = true
			lockLoss.abort(error instanceof Error ? error : new Error('Site write lock session was lost'))
		})
		try {
			if (signal?.aborted) throw abortReason(signal)
			const callback = Promise.resolve(fn(safetySignal))
			// The callback can outlive the lock; observe its eventual failure
			// without letting it surface as an unhandled rejection.
			void callback.catch((callbackError) => {
				if (lockLost) {
					logger.warn('[DB] Lock callback failed after lock session loss', {
						did,
						rkey,
						errorKind: databaseErrorKind(callbackError),
					})
				}
			})
			const lost = new Promise<never>((_resolve, reject) => {
				lockLoss.signal.addEventListener('abort', () => reject(new SiteWriteLockLostError(lockLoss.signal.reason)), {
					once: true,
				})
			})
			// Fail fast on lock loss: waiting for a callback that no longer runs
			// under mutual exclusion would hide a fencing breach.
			return await Promise.race([callback, lost])
		} finally {
			monitor.stop()
			if (held) {
				if (lockLost) {
					// The session — and with it the server-side advisory lock — is
					// gone. Skip the pointless unlock query and retire the session.
					connectionSafeToRelease = false
					try {
						await conn.close({ timeout: LOCK_POOL_RETIRE_TIMEOUT_SECONDS })
					} catch (closeError) {
						logger.error('[DB] Failed to close lost site-write lock connection', undefined, {
							did,
							rkey,
							errorKind: databaseErrorKind(closeError),
						})
					}
				} else {
					let unlockFailure: unknown
					try {
						const result = await conn`SELECT pg_advisory_unlock(${lockId}::bigint) AS unlocked`
						if (!advisoryUnlockReturnedTrue(result)) unlockFailure = new Error('Advisory lock was not released')
					} catch (error) {
						unlockFailure = error
					}

					if (unlockFailure) {
						// Advisory locks are session-scoped. A thrown unlock or a false
						// result means this session must never return to the reusable pool.
						connectionSafeToRelease = false
						logger.error('[DB] Failed to release site write lock; quarantining connection', undefined, {
							did,
							rkey,
							errorKind: databaseErrorKind(unlockFailure),
						})
						try {
							await conn.close({ timeout: LOCK_POOL_RETIRE_TIMEOUT_SECONDS })
						} catch (closeError) {
							// Cleanup must not replace a callback failure or expose driver details.
							logger.error('[DB] Failed to close quarantined site-write lock connection', undefined, {
								did,
								rkey,
								errorKind: databaseErrorKind(closeError),
							})
						}
					}
				}
			}
		}
	} finally {
		if (connectionSafeToRelease) {
			try {
				conn.release()
			} catch (error) {
				// Keep the callback/acquisition outcome intact if client cleanup fails.
				logger.error('[DB] Failed to release site-write lock connection', undefined, {
					did,
					rkey,
					errorKind: databaseErrorKind(error),
				})
			}
		}
	}
}

/**
 * Run `fn` while holding the per-site write lock, serializing all cache writers
 * for `${did}/${rkey}` across drivers and instances.
 *
 * Uses a blocking acquire bounded by lock_timeout so a stuck holder cannot wedge
 * the queue forever. An acquisition timeout or database error fails closed: the
 * caller receives an error and can retry without writing concurrently.
 *
 * While `fn` runs, the lock session is kept alive with periodic liveness
 * probes (proxies idle-kill idle sessions at 60s) and `fn` receives a signal
 * that aborts on caller abort or on lock-session loss, so a lost advisory lock
 * stops the protected work instead of silently writing without exclusion.
 */
export async function withSiteWriteLock<T>(
	did: string,
	rkey: string,
	fn: (signal: AbortSignal) => Promise<T>,
	signal?: AbortSignal,
	heartbeat: LockHeartbeatOptions = {},
): Promise<T> {
	let conn: SiteWriteLockConnection

	try {
		conn = await lockPool.reserve(signal)
	} catch (error) {
		logger.error('[DB] Failed to reserve site-write lock connection', undefined, {
			did,
			rkey,
			errorKind: databaseErrorKind(error),
		})
		throw new SiteWriteLockAcquisitionError(error)
	}

	return await withReservedSiteWriteLock(conn, did, rkey, fn, signal, heartbeat)
}

// Read functions

export async function getSiteCache(did: string, rkey: string): Promise<SiteCache | null> {
	const result = await sql<SiteCache[]>`
    SELECT did, rkey, record_cid, file_cids, cached_at, updated_at, cold_synced
    FROM site_cache
    WHERE did = ${did} AND rkey = ${rkey}
    LIMIT 1
  `
	return result[0] || null
}

export async function getSiteSettingsCache(did: string, rkey: string): Promise<SiteSettingsCache | null> {
	const result = await sql<SiteSettingsCache[]>`
    SELECT did, rkey, record_cid, directory_listing, spa_mode, custom_404, index_files, clean_urls, headers, cached_at, updated_at
    FROM site_settings_cache
    WHERE did = ${did} AND rkey = ${rkey}
    LIMIT 1
  `
	return result[0] || null
}

/**
 * List all known DIDs from all DID-bearing tables.
 * Missing tables are skipped to keep bootstrapping resilient.
 */
export async function listAllKnownDids(): Promise<string[]> {
	const sources: Array<{ name: string; fetch: () => Promise<Array<{ did: string }>> }> = [
		{
			name: 'site_cache',
			fetch: () => sql<Array<{ did: string }>>`
        SELECT DISTINCT did
        FROM site_cache
        WHERE did IS NOT NULL AND did <> ''
      `,
		},
		{
			name: 'site_settings_cache',
			fetch: () => sql<Array<{ did: string }>>`
        SELECT DISTINCT did
        FROM site_settings_cache
        WHERE did IS NOT NULL AND did <> ''
      `,
		},
		{
			name: 'domains',
			fetch: () => sql<Array<{ did: string }>>`
        SELECT DISTINCT did
        FROM domains
        WHERE did IS NOT NULL AND did <> ''
      `,
		},
		{
			name: 'custom_domains',
			fetch: () => sql<Array<{ did: string }>>`
        SELECT DISTINCT did
        FROM custom_domains
        WHERE did IS NOT NULL AND did <> ''
      `,
		},
		{
			name: 'supporter',
			fetch: () => sql<Array<{ did: string }>>`
        SELECT DISTINCT did
        FROM supporter
        WHERE did IS NOT NULL AND did <> ''
      `,
		},
	]
	const dids = new Set<string>()

	for (const source of sources) {
		try {
			const rows = await source.fetch()
			for (const row of rows) {
				if (typeof row.did === 'string' && row.did.length > 0) {
					dids.add(row.did)
				}
			}
		} catch {
			logger.warn(`[DB] Skipping DID source table ${source.name}`)
		}
	}

	return [...dids].sort()
}

// Write functions

export async function upsertSiteCache(
	did: string,
	rkey: string,
	recordCid: string,
	fileCids: Record<string, string>,
	// The firehose owns the S3 cold tier, so it always marks the row synced once
	// it has finished writing files. Defaults to true to keep existing call sites
	// (and the contract that this function is only called after S3 writes) intact.
	coldSynced = true,
): Promise<void> {
	logger.debug(`[DB] upsertSiteCache starting for ${did}/${rkey}`)
	try {
		await sql`
      INSERT INTO site_cache (did, rkey, record_cid, file_cids, cached_at, updated_at, cold_synced)
      VALUES (${did}, ${rkey}, ${recordCid}, ${sql.json(fileCids ?? {})}, EXTRACT(EPOCH FROM NOW()), EXTRACT(EPOCH FROM NOW()), ${coldSynced})
      ON CONFLICT (did, rkey)
      DO UPDATE SET
        record_cid = EXCLUDED.record_cid,
        file_cids = EXCLUDED.file_cids,
        updated_at = EXTRACT(EPOCH FROM NOW()),
        cold_synced = EXCLUDED.cold_synced
    `
		logger.debug(`[DB] upsertSiteCache completed for ${did}/${rkey}`)
	} catch (err) {
		logger.error('[DB] upsertSiteCache error', err, { did, rkey })
		throw err
	}
}

/** Keep an empty durable manifest so hosting can distinguish a confirmed delete from a projection/storage outage. */
export async function markSiteCacheDeleted(did: string, rkey: string): Promise<void> {
	await upsertSiteCache(did, rkey, DELETED_SITE_RECORD_CID, {})
}

export async function upsertSiteSettingsCache(
	did: string,
	rkey: string,
	recordCid: string,
	settings: {
		directoryListing: boolean
		spaMode?: string
		custom404?: string
		indexFiles?: string[]
		cleanUrls: boolean
		headers?: Array<{ name: string; value: string; path?: string }>
	},
): Promise<void> {
	const directoryListing = settings.directoryListing ?? false
	const spaMode = settings.spaMode ?? null
	const custom404 = settings.custom404 ?? null
	const cleanUrls = settings.cleanUrls ?? true

	const indexFiles = settings.indexFiles ?? []
	const headers = settings.headers ?? []

	logger.debug(`[DB] upsertSiteSettingsCache starting for ${did}/${rkey}`, {
		directoryListing,
		spaMode,
		custom404,
		indexFiles,
		cleanUrls,
		headers,
	})

	try {
		await sql`
      INSERT INTO site_settings_cache (did, rkey, record_cid, directory_listing, spa_mode, custom_404, index_files, clean_urls, headers, cached_at, updated_at)
      VALUES (
        ${did},
        ${rkey},
        ${recordCid},
        ${directoryListing},
        ${spaMode},
        ${custom404},
        ${sql.json(indexFiles)},
        ${cleanUrls},
        ${sql.json(headers)},
        EXTRACT(EPOCH FROM NOW()),
        EXTRACT(EPOCH FROM NOW())
      )
      ON CONFLICT (did, rkey)
      DO UPDATE SET
        record_cid = EXCLUDED.record_cid,
        directory_listing = EXCLUDED.directory_listing,
        spa_mode = EXCLUDED.spa_mode,
        custom_404 = EXCLUDED.custom_404,
        index_files = EXCLUDED.index_files,
        clean_urls = EXCLUDED.clean_urls,
        headers = EXCLUDED.headers,
        updated_at = EXTRACT(EPOCH FROM NOW())
    `
		logger.debug(`[DB] upsertSiteSettingsCache completed for ${did}/${rkey}`)
	} catch (err) {
		logger.error('[DB] upsertSiteSettingsCache error', err, { did, rkey })
		throw err
	}
}

export async function deleteSiteSettingsCache(did: string, rkey: string): Promise<void> {
	await sql`DELETE FROM site_settings_cache WHERE did = ${did} AND rkey = ${rkey}`
}

export async function isSupporter(did: string): Promise<boolean> {
	const rows = await sql`SELECT 1 FROM supporter WHERE did = ${did} LIMIT 1`
	return rows.length > 0
}

export async function closeDatabase(): Promise<void> {
	await Promise.all([sql.end({ timeout: 5 }), lockPool.end({ timeout: 5 })])
	logger.info('[DB] Database connections closed')
}

export { sql }