import { createHash, createHmac, randomUUID } from 'node:crypto'
import { createLogger } from '@wispplace/observability'
import { MAX_ACTIVE_WEBHOOK_SUBSCRIPTIONS } from './admission'
import { isValidAtprotoRevision } from './atproto-revision'
import type {
	ClaimedWebhookDelivery,
	CurrentWebhookSubscription,
	NewWebhookDeliveryEventRow,
	NewWebhookDeliveryOutboxRow,
	WebhookEntry,
} from './db'
import * as database from './db'
import type { EventKind } from './matcher'
import * as redis from './redis'
import {
	discardWebhookResponse,
	pinnedWebhookFetch,
	type WebhookResolver,
	type WebhookTransport,
	WebhookUrlError,
} from './webhook-url'

const logger = createLogger('webhook-service:delivery')

export const MAX_WEBHOOK_DELIVERY_ENQUEUE_BATCH = 1_000
/** Kept exported for callers; it is the exact registry admission ceiling. */
export const MAX_WEBHOOK_DELIVERY_FANOUT = MAX_ACTIVE_WEBHOOK_SUBSCRIPTIONS
const MAX_EVENT_FIELD_LENGTH = 2_048
const MAX_PAYLOAD_BYTES = 512 * 1024
const MAX_PAYLOAD_NODES = 10_000
/** Bound allocation and one DB round trip even for a valid high-fanout event. */
export const MAX_WEBHOOK_DELIVERY_ENQUEUE_CHUNK_BYTES = 4 * 1024 * 1024
export const MAX_WEBHOOK_DELIVERY_EVENT_BYTES = 16 * 1024 * 1024

export interface WebhookPayload {
	id: string
	event: EventKind
	did: string
	collection: string
	rkey: string
	cid?: string
	record?: unknown
	timestamp: string
}

/** Source identity is deliberately independent of direct/backlink consumer name. */
export interface WebhookDeliveryEvent {
	relay: string
	timeUs: number
	rev: string
	operation: EventKind
	did: string
	collection: string
	rkey: string
	cid?: string
	record?: unknown
}

export interface EnqueueWebhookDeliveriesResult {
	enqueued: number
	deduplicated: number
}

export type WebhookDeliveryInputErrorKind =
	| 'invalid_event'
	| 'invalid_subscription'
	| 'payload_invalid'
	| 'payload_too_large'
	| 'fanout_limit'

/** A deterministic bad input may be quarantined/acknowledged by intake; DB errors must still retry. */
export class WebhookDeliveryInputError extends Error {
	constructor(
		public readonly kind: WebhookDeliveryInputErrorKind,
		message: string,
	) {
		super(message)
		this.name = 'WebhookDeliveryInputError'
	}
}

function boundedString(value: unknown, maximum = MAX_EVENT_FIELD_LENGTH): value is string {
	return typeof value === 'string' && value.length > 0 && value.length <= maximum
}

function canonicalize(value: unknown, depth = 0, state = { nodes: 0 }): unknown {
	state.nodes++
	if (depth > 32 || state.nodes > MAX_PAYLOAD_NODES) {
		throw new WebhookDeliveryInputError('payload_invalid', 'Webhook payload is invalid')
	}
	if (value === null || typeof value === 'boolean') return value
	if (typeof value === 'string') {
		// Avoid allocating/serializing an unbounded individual string before the
		// final UTF-8 byte cap below.
		if (value.length > MAX_PAYLOAD_BYTES) {
			throw new WebhookDeliveryInputError('payload_too_large', 'Webhook payload is too large')
		}
		return value
	}
	if (typeof value === 'number') {
		if (!Number.isFinite(value)) throw new WebhookDeliveryInputError('payload_invalid', 'Webhook payload is invalid')
		return value
	}
	if (Array.isArray(value)) {
		if (value.length > MAX_PAYLOAD_NODES) {
			throw new WebhookDeliveryInputError('payload_invalid', 'Webhook payload is invalid')
		}
		return value.map((entry) => canonicalize(entry, depth + 1, state))
	}
	if (typeof value === 'object') {
		const input = value as Record<string, unknown>
		const keys = Object.keys(input)
		if (keys.length > MAX_PAYLOAD_NODES) {
			throw new WebhookDeliveryInputError('payload_invalid', 'Webhook payload is invalid')
		}
		const output: Record<string, unknown> = {}
		for (const key of keys.sort()) {
			if (key.length > 512) throw new WebhookDeliveryInputError('payload_invalid', 'Webhook payload is invalid')
			output[key] = canonicalize(input[key], depth + 1, state)
		}
		return output
	}
	throw new WebhookDeliveryInputError('payload_invalid', 'Webhook payload is invalid')
}

/** JSON with sorted object keys, so stored bytes and HMAC input are stable. */
export function canonicalWebhookJson(value: unknown): string {
	let encoded: string
	try {
		encoded = JSON.stringify(canonicalize(value))
	} catch {
		throw new WebhookDeliveryInputError('payload_invalid', 'Webhook payload is invalid')
	}
	if (typeof encoded !== 'string' || new TextEncoder().encode(encoded).byteLength > MAX_PAYLOAD_BYTES) {
		throw new WebhookDeliveryInputError('payload_too_large', 'Webhook payload is too large')
	}
	return encoded
}

function normalizedRelayIdentity(relay: string): { canonical: string; hash: string } {
	if (!boundedString(relay)) throw new WebhookDeliveryInputError('invalid_event', 'Webhook event is invalid')
	let parsed: URL
	try {
		parsed = new URL(relay)
	} catch {
		throw new WebhookDeliveryInputError('invalid_event', 'Webhook event is invalid')
	}
	if ((parsed.protocol !== 'ws:' && parsed.protocol !== 'wss:') || !parsed.hostname)
		throw new WebhookDeliveryInputError('invalid_event', 'Webhook event is invalid')
	parsed.username = ''
	parsed.password = ''
	parsed.search = ''
	parsed.hash = ''
	parsed.hostname = parsed.hostname.toLowerCase().replace(/\.$/, '')
	const canonical = `${parsed.protocol}//${parsed.host}${parsed.pathname || '/'}`
	return {
		canonical,
		hash: createHash('sha256').update(`wisp-jetstream-relay/v1\0${canonical}`).digest('hex'),
	}
}

function timestampForTimeUs(timeUs: number): string {
	if (!Number.isSafeInteger(timeUs) || timeUs < 0)
		throw new WebhookDeliveryInputError('invalid_event', 'Webhook event is invalid')
	const date = new Date(timeUs / 1_000)
	if (Number.isNaN(date.getTime())) throw new WebhookDeliveryInputError('invalid_event', 'Webhook event is invalid')
	return date.toISOString()
}

function assertEvent(event: WebhookDeliveryEvent): void {
	if (!event || typeof event !== 'object')
		throw new WebhookDeliveryInputError('invalid_event', 'Webhook event is invalid')
	if (event.operation !== 'create' && event.operation !== 'update' && event.operation !== 'delete')
		throw new WebhookDeliveryInputError('invalid_event', 'Webhook event is invalid')
	if (
		!isValidAtprotoRevision(event.rev) ||
		!boundedString(event.did) ||
		!boundedString(event.collection) ||
		!boundedString(event.rkey) ||
		(event.cid !== undefined && !boundedString(event.cid, 1_024))
	) {
		throw new WebhookDeliveryInputError('invalid_event', 'Webhook event is invalid')
	}
	timestampForTimeUs(event.timeUs)
	normalizedRelayIdentity(event.relay)
}

/** One immutable event body is shared across every subscription delivery. */
export function createWebhookDeliveryEventId(event: WebhookDeliveryEvent): string {
	assertEvent(event)
	const identity = canonicalWebhookJson({
		version: 1,
		relay: normalizedRelayIdentity(event.relay).canonical,
		timeUs: event.timeUs,
		rev: event.rev,
		operation: event.operation,
		did: event.did,
		collection: event.collection,
		rkey: event.rkey,
		cid: event.cid ?? null,
	})
	return `whe_v1_${createHash('sha256').update(identity).digest('hex')}`
}

/** A versioned deterministic Idempotency-Key for one subscription delivery. */
export function createWebhookDeliveryId(event: WebhookDeliveryEvent, ownerDid: string, webhookRkey: string): string {
	assertEvent(event)
	if (!boundedString(ownerDid) || !boundedString(webhookRkey, 1_024))
		throw new WebhookDeliveryInputError('invalid_subscription', 'Webhook subscription is invalid')
	const identity = canonicalWebhookJson({
		version: 1,
		relay: normalizedRelayIdentity(event.relay).canonical,
		timeUs: event.timeUs,
		rev: event.rev,
		operation: event.operation,
		did: event.did,
		collection: event.collection,
		rkey: event.rkey,
		cid: event.cid ?? null,
		ownerDid,
		webhookRkey,
	})
	return `whd_v1_${createHash('sha256').update(identity).digest('hex')}`
}

/** Do not copy a webhook signing secret into an event payload/outbox row. */
function payloadRecord(event: WebhookDeliveryEvent): unknown {
	if (event.record === undefined) return undefined
	if (
		event.collection !== 'place.wisp.v2.wh' ||
		!event.record ||
		typeof event.record !== 'object' ||
		Array.isArray(event.record)
	) {
		return event.record
	}
	const copy = { ...(event.record as Record<string, unknown>) }
	delete copy.secret
	return copy
}

interface SubscriptionDeliveryMetadata {
	fingerprint: string
	signingMode: NewWebhookDeliveryOutboxRow['signingMode']
	secretId?: string
}

/** Validate only fields needed by the outbox; never materialize a record snapshot. */
function subscriptionDeliveryMetadata(
	entry: WebhookEntry,
	fingerprintForRecord: (record: WebhookEntry['record']) => string,
): SubscriptionDeliveryMetadata {
	const record = entry.record
	if (!record || !boundedString(record.url) || !record.scope || !boundedString(record.scope.aturi)) {
		throw new WebhookDeliveryInputError('invalid_subscription', 'Webhook subscription is invalid')
	}
	const signingMode: NewWebhookDeliveryOutboxRow['signingMode'] = record.secretId
		? 'secret_id'
		: record.secret
			? 'record_secret'
			: 'none'
	try {
		return { fingerprint: fingerprintForRecord(record), signingMode, secretId: record.secretId }
	} catch {
		throw new WebhookDeliveryInputError('invalid_subscription', 'Webhook subscription is invalid')
	}
}

export interface EnqueueWebhookDeliveriesOptions {
	/** Test seam; production uses the transactional DB bulk writer. */
	enqueueOutbox?: (
		event: NewWebhookDeliveryEventRow,
		rows: readonly NewWebhookDeliveryOutboxRow[],
		ensureEvent?: boolean,
	) => Promise<{ enqueued: number; deduplicated: number }>
	/** Test seam; production uses the same fingerprint function as send-time revocation. */
	subscriptionFingerprint?: (record: WebhookEntry['record']) => string
}

function encodedBytes(value: string): number {
	return new TextEncoder().encode(value).byteLength
}

function estimatedOutboxInputBytes(
	entry: WebhookEntry,
	metadata: SubscriptionDeliveryMetadata,
	eventId: string,
): number {
	// This covers every text parameter in the UNNEST row. No record snapshot or
	// payload body is copied into a subscription-specific row.
	return (
		encodedBytes(entry.ownerDid) +
		encodedBytes(entry.rkey) +
		encodedBytes(entry.record.url) +
		encodedBytes(metadata.secretId ?? '') +
		encodedBytes(metadata.signingMode) +
		encodedBytes(metadata.fingerprint) +
		encodedBytes(eventId) +
		80
	)
}

export async function enqueueWebhookDeliveries(
	entries: readonly WebhookEntry[],
	event: WebhookDeliveryEvent,
	options: EnqueueWebhookDeliveriesOptions = {},
): Promise<EnqueueWebhookDeliveriesResult> {
	assertEvent(event)
	const enqueueOutbox = options.enqueueOutbox ?? database.enqueueWebhookDeliveryOutbox
	const fingerprintForRecord = options.subscriptionFingerprint ?? database.webhookSubscriptionFingerprint
	if (entries.length > MAX_WEBHOOK_DELIVERY_FANOUT) {
		throw new WebhookDeliveryInputError('fanout_limit', 'Webhook delivery fanout is too large')
	}
	if (entries.length === 0) return { enqueued: 0, deduplicated: 0 }

	const relay = normalizedRelayIdentity(event.relay)
	const timestamp = timestampForTimeUs(event.timeUs)
	const record = payloadRecord(event)
	const eventId = createWebhookDeliveryEventId(event)
	const payload: WebhookPayload = {
		id: eventId,
		event: event.operation,
		did: event.did,
		collection: event.collection,
		rkey: event.rkey,
		...(event.cid === undefined ? {} : { cid: event.cid }),
		...(record === undefined ? {} : { record }),
		timestamp,
	}
	const payloadJson = canonicalWebhookJson(payload)
	const eventRow: NewWebhookDeliveryEventRow = {
		eventId,
		payloadJson,
		sourceRelayId: relay.hash,
		sourceTimeUs: event.timeUs,
		sourceRevision: event.rev,
		sourceOperation: event.operation,
	}

	// Preflight all subscription snapshots before committing any chunk. A
	// deterministic fanout/input breach is quarantined as one event, never
	// partially delivered. The estimates bound both the entire event and every
	// typed-UNNEST parameter batch.
	const rowMetadata: SubscriptionDeliveryMetadata[] = []
	const rowInputBytes: number[] = []
	let eventInputBytes = encodedBytes(payloadJson)
	for (const entry of entries) {
		if (!boundedString(entry.ownerDid) || !boundedString(entry.rkey, 1_024)) {
			throw new WebhookDeliveryInputError('invalid_subscription', 'Webhook subscription is invalid')
		}
		const metadata = subscriptionDeliveryMetadata(entry, fingerprintForRecord)
		const estimated = estimatedOutboxInputBytes(entry, metadata, eventId)
		eventInputBytes += estimated
		if (eventInputBytes > MAX_WEBHOOK_DELIVERY_EVENT_BYTES) {
			throw new WebhookDeliveryInputError('fanout_limit', 'Webhook delivery fanout is too large')
		}
		rowMetadata.push(metadata)
		rowInputBytes.push(estimated)
	}

	let enqueued = 0
	let deduplicated = 0
	let offset = 0
	// A large valid fanout commits deterministic bounded chunks. A later DB
	// failure leaves earlier stable IDs safely dedupable on replay; intake only
	// acknowledges after the whole function resolves.
	while (offset < entries.length) {
		let end = offset
		let chunkInputBytes = 0
		while (end < entries.length && end - offset < MAX_WEBHOOK_DELIVERY_ENQUEUE_BATCH) {
			const estimated = rowInputBytes[end]
			if (estimated === undefined) throw new Error('Invalid webhook delivery estimate')
			if (end > offset && chunkInputBytes + estimated > MAX_WEBHOOK_DELIVERY_ENQUEUE_CHUNK_BYTES) break
			chunkInputBytes += estimated
			end++
		}
		const rows: NewWebhookDeliveryOutboxRow[] = []
		for (let index = offset; index < end; index++) {
			const entry = entries[index]
			if (!entry) throw new Error('Invalid webhook delivery entry')
			const deliveryId = createWebhookDeliveryId(event, entry.ownerDid, entry.rkey)
			const metadata = rowMetadata[index]
			if (!metadata) throw new Error('Invalid webhook delivery metadata')
			rows.push({
				deliveryId,
				eventId,
				ownerDid: entry.ownerDid,
				webhookRkey: entry.rkey,
				targetUrl: entry.record.url,
				secretId: metadata.secretId,
				signingMode: metadata.signingMode,
				subscriptionFingerprint: metadata.fingerprint,
			})
		}
		const result = await enqueueOutbox(eventRow, rows, offset === 0)
		enqueued += result.enqueued
		deduplicated += result.deduplicated
		offset = end
	}
	return { enqueued, deduplicated }
}

/** HMAC over exactly the canonical payload bytes stored in the outbox. */
export function signWebhookBody(secret: string, body: string): string {
	return `sha256=${createHmac('sha256', secret).update(body, 'utf8').digest('hex')}`
}

function syntheticLegacyRevision(seed: string): string {
	const alphabet = '234567abcdefghijklmnopqrstuvwxyz'
	const bytes = createHash('sha256').update(seed).digest()
	let result = ''
	for (let index = 0; index < 13; index++) result += alphabet[(bytes[index] ?? 0) % alphabet.length]
	return result
}

/**
 * Legacy API compatibility during rolling deploys. New intake must call
 * enqueueWebhookDeliveries with source relay/time/rev before cursor advancement.
 */
export async function deliverWebhook(
	entry: WebhookEntry,
	eventDid: string,
	eventCollection: string,
	eventRkey: string,
	eventKind: EventKind,
	eventCid?: string,
	eventRecord?: unknown,
): Promise<void> {
	const timeUs = Date.now() * 1_000
	await enqueueWebhookDeliveries([entry], {
		relay: 'wss://legacy.webhook-service.invalid/',
		timeUs,
		rev: isValidAtprotoRevision(eventCid)
			? (eventCid as string)
			: syntheticLegacyRevision(`${timeUs}:${eventCid ?? ''}`),
		operation: eventKind,
		did: eventDid,
		collection: eventCollection,
		rkey: eventRkey,
		cid: eventCid,
		record: eventRecord,
	})
}

export type DeliveryFailureKind =
	| 'subscription_unavailable'
	| 'subscription_changed'
	| 'secret_unavailable'
	| 'invalid_target'
	| 'blocked_target'
	| 'dns'
	| 'timeout'
	| 'network'
	| 'redirect'
	| 'request_too_large'
	| 'response_too_large'
	| 'http_4xx'
	| 'http_5xx'
	| 'internal'

interface AttemptResult {
	status: number
	retryAfterMs?: number
}

interface DeliveryFailure {
	kind: DeliveryFailureKind
	transient: boolean
	httpStatus?: number
	retryAfterMs?: number
}

function boundedOption(value: number | undefined, fallback: number, minimum: number, maximum: number): number {
	if (value === undefined) return fallback
	if (!Number.isFinite(value)) return fallback
	return Math.max(minimum, Math.min(maximum, Math.floor(value)))
}

/** Narrow persistence seam keeps worker shutdown behavior testable without a live DB. */
export type WebhookDeliveryWorkerPersistence = Pick<
	typeof database,
	| 'cancelWebhookDeliveryForSubscriptionChange'
	| 'claimWebhookDeliveryOutbox'
	| 'getCurrentWebhookSubscription'
	| 'getWebhookInlineSecret'
	| 'getWebhookSecretToken'
	| 'markWebhookDeliverySucceeded'
	| 'renewWebhookDeliveryLease'
	| 'rescheduleWebhookDelivery'
	| 'runWebhookMaintenance'
>

export interface WebhookDeliveryWorkerOptions {
	concurrency?: number
	batchSize?: number
	pollIntervalMs?: number
	leaseMs?: number
	maxAttempts?: number
	retryBaseMs?: number
	retryMaxMs?: number
	requestTimeoutMs?: number
	maxRequestBytes?: number
	maxResponseBytes?: number
	maxRedirects?: number
	maintenanceIntervalMs?: number
	deliveredRetentionMs?: number
	deadLetterRetentionMs?: number
	resolver?: WebhookResolver
	transport?: WebhookTransport
	/** Development-only request escape; webhook-url independently enforces its env gates. */
	allowLoopback?: boolean
	random?: () => number
	workerId?: string
	/** Test seam; production uses the normal durable DB module. */
	persistence?: WebhookDeliveryWorkerPersistence
}

interface BoundedWorkerOptions {
	concurrency: number
	batchSize: number
	pollIntervalMs: number
	leaseMs: number
	maxAttempts: number
	retryBaseMs: number
	retryMaxMs: number
	requestTimeoutMs: number
	maxRequestBytes: number
	maxResponseBytes: number
	maxRedirects: number
	maintenanceIntervalMs: number
	deliveredRetentionMs: number
	deadLetterRetentionMs: number
	resolver?: WebhookResolver
	transport?: WebhookTransport
	allowLoopback: boolean
	random: () => number
	workerId: string
	persistence: WebhookDeliveryWorkerPersistence
}

function boundWorkerOptions(options: WebhookDeliveryWorkerOptions): BoundedWorkerOptions {
	const retryBaseMs = boundedOption(options.retryBaseMs, 1_000, 100, 60_000)
	const retryMaxMs = boundedOption(options.retryMaxMs, 5 * 60_000, retryBaseMs, 60 * 60_000)
	const requestTimeoutMs = boundedOption(options.requestTimeoutMs, 10_000, 100, 30_000)
	// A claim covers one HTTP attempt only. Keep enough room for response
	// cancellation and the terminal DB update; retries are rescheduled, not held.
	const leaseMs = boundedOption(
		options.leaseMs,
		Math.max(60_000, requestTimeoutMs + 5_000),
		requestTimeoutMs + 5_000,
		10 * 60_000,
	)
	return {
		concurrency: boundedOption(options.concurrency, 4, 1, 32),
		batchSize: boundedOption(options.batchSize, 32, 1, 256),
		pollIntervalMs: boundedOption(options.pollIntervalMs, 1_000, 100, 60_000),
		leaseMs,
		maxAttempts: boundedOption(options.maxAttempts, 12, 1, 50),
		retryBaseMs,
		retryMaxMs,
		requestTimeoutMs,
		maxRequestBytes: boundedOption(options.maxRequestBytes, MAX_PAYLOAD_BYTES, 1, MAX_PAYLOAD_BYTES),
		maxResponseBytes: boundedOption(options.maxResponseBytes, 64 * 1024, 0, 512 * 1024),
		maxRedirects: boundedOption(options.maxRedirects, 3, 0, 5),
		maintenanceIntervalMs: boundedOption(options.maintenanceIntervalMs, 60_000, 10_000, 60 * 60_000),
		deliveredRetentionMs: boundedOption(
			options.deliveredRetentionMs,
			7 * 24 * 60 * 60_000,
			60_000,
			365 * 24 * 60 * 60_000,
		),
		deadLetterRetentionMs: boundedOption(
			options.deadLetterRetentionMs,
			30 * 24 * 60 * 60_000,
			60_000,
			365 * 24 * 60 * 60_000,
		),
		resolver: options.resolver,
		transport: options.transport,
		allowLoopback: options.allowLoopback === true,
		random: options.random ?? Math.random,
		workerId: boundedString(options.workerId, 256) ? options.workerId : `webhook-worker-${randomUUID()}`,
		persistence: options.persistence ?? database,
	}
}
function parseRetryAfter(value: string | null, now = Date.now()): number | undefined {
	if (!value) return undefined
	if (/^\d+$/.test(value)) return Math.min(Number(value) * 1_000, 60 * 60_000)
	const date = Date.parse(value)
	if (Number.isNaN(date)) return undefined
	return Math.max(0, Math.min(date - now, 60 * 60_000))
}

function classifyHttpStatus(status: number, retryAfterMs?: number): DeliveryFailure {
	if (status >= 200 && status < 300) return { kind: 'internal', transient: false, httpStatus: status }
	if (status >= 500 || status === 408 || status === 409 || status === 423 || status === 425 || status === 429) {
		return { kind: 'http_5xx', transient: true, httpStatus: status, retryAfterMs }
	}
	if (status >= 400 && status < 500) return { kind: 'http_4xx', transient: false, httpStatus: status }
	// Redirects should have been followed/rejected by the pinned transport; an
	// unexpected final 3xx is a target configuration problem, not a retry storm.
	return { kind: 'redirect', transient: false, httpStatus: status }
}

function failureFromError(error: unknown): DeliveryFailure {
	if (error instanceof WebhookUrlError) {
		switch (error.kind) {
			case 'invalid_url':
				return { kind: 'invalid_target', transient: false }
			case 'blocked_destination':
				return { kind: 'blocked_target', transient: false }
			case 'dns':
				return { kind: 'dns', transient: true }
			case 'timeout':
				return { kind: 'timeout', transient: true }
			case 'network':
				return { kind: 'network', transient: true }
			case 'redirect':
				return { kind: 'redirect', transient: false }
			case 'request_too_large':
				return { kind: 'request_too_large', transient: false }
			case 'response_too_large':
				return { kind: 'response_too_large', transient: false }
		}
	}
	return { kind: 'network', transient: true }
}

function retryDelayMs(attemptNumber: number, options: BoundedWorkerOptions): number {
	const exponent = Math.min(Math.max(0, attemptNumber - 1), 20)
	const maximum = Math.min(options.retryMaxMs, options.retryBaseMs * 2 ** exponent)
	const random = Math.max(0, Math.min(1, options.random()))
	return Math.max(1, Math.floor(maximum * (0.5 + random * 0.5)))
}

interface DeliveryAuditEvent {
	ownerDid: string
	rkey: string
	eventKind: EventKind
	eventDid: string
	eventCollection: string
	eventRkey: string
	cid?: string
	status: 'ok' | 'failed' | 'dead_letter'
	deliveredAt: string
}

function auditFromPayload(row: ClaimedWebhookDelivery, status: DeliveryAuditEvent['status']): DeliveryAuditEvent {
	let payload: Partial<WebhookPayload> = {}
	try {
		const parsed = JSON.parse(row.payloadBody) as unknown
		if (parsed && typeof parsed === 'object') payload = parsed as Partial<WebhookPayload>
	} catch {
		// A corrupt row is handled as a delivery failure without exposing contents.
	}
	return {
		ownerDid: row.ownerDid,
		rkey: row.webhookRkey,
		eventKind:
			payload.event === 'create' || payload.event === 'update' || payload.event === 'delete'
				? payload.event
				: row.sourceOperation === 'create' || row.sourceOperation === 'update' || row.sourceOperation === 'delete'
					? row.sourceOperation
					: 'update',
		eventDid: typeof payload.did === 'string' ? payload.did : '[unknown]',
		eventCollection: typeof payload.collection === 'string' ? payload.collection : '[unknown]',
		eventRkey: typeof payload.rkey === 'string' ? payload.rkey : '[unknown]',
		cid: typeof payload.cid === 'string' ? payload.cid : undefined,
		status,
		deliveredAt: new Date().toISOString(),
	}
}

function persistDeliveryAudit(audit: DeliveryAuditEvent): Promise<void> {
	// Audit/notification failure must never turn a durably completed delivery
	// into an unhandled rejection or cause a second POST.
	return Promise.all([
		Promise.resolve()
			.then(() => database.insertEventLog(audit))
			.catch(() => logger.warn('[delivery] audit persistence failed')),
		Promise.resolve()
			.then(() =>
				redis.publishWebhookEvent({
					...audit,
					url: '[redacted]',
					status: audit.status === 'ok' ? 'ok' : 'failed',
				}),
			)
			.catch(() => logger.warn('[delivery] audit publish failed')),
	]).then(() => undefined)
}

async function postDelivery(
	row: ClaimedWebhookDelivery,
	secret: string | undefined,
	options: BoundedWorkerOptions,
	signal: AbortSignal,
): Promise<AttemptResult> {
	const headers: Record<string, string> = {
		'Content-Type': 'application/json',
		'User-Agent': 'wisp.place-webhook/2.0',
		'Idempotency-Key': row.deliveryId,
	}
	if (secret) headers['X-Webhook-Signature'] = signWebhookBody(secret, row.payloadBody)
	const response = await pinnedWebhookFetch(row.targetUrl, {
		method: 'POST',
		headers,
		body: row.payloadBody,
		timeoutMs: options.requestTimeoutMs,
		maxRequestBytes: options.maxRequestBytes,
		maxResponseBytes: options.maxResponseBytes,
		maxRedirects: options.maxRedirects,
		resolver: options.resolver,
		transport: options.transport,
		allowLoopback: options.allowLoopback,
		signal,
	})
	const result = { status: response.status, retryAfterMs: parseRetryAfter(response.headers.get('retry-after')) }
	await discardWebhookResponse(response)
	return result
}

async function mapBounded<T>(
	values: readonly T[],
	concurrency: number,
	operation: (value: T) => Promise<void>,
): Promise<void> {
	let next = 0
	const runners = Array.from({ length: Math.min(concurrency, values.length) }, async () => {
		while (true) {
			const index = next++
			const value = values[index]
			if (value === undefined) return
			await operation(value)
		}
	})
	await Promise.all(runners)
}

/** Maintains ownership during one bounded HTTP attempt and aborts on lease loss. */
class DeliveryLeaseHeartbeat {
	private readonly controller = new AbortController()
	private timer: ReturnType<typeof setTimeout> | undefined
	private stopped = false
	private refreshPromise: Promise<void> | undefined

	constructor(
		private readonly row: ClaimedWebhookDelivery,
		private readonly leaseMs: number,
		private readonly renewLease: WebhookDeliveryWorkerPersistence['renewWebhookDeliveryLease'],
	) {}

	get signal(): AbortSignal {
		return this.controller.signal
	}

	async acquire(): Promise<boolean> {
		if (this.stopped || this.controller.signal.aborted) return false
		try {
			const renewed = await this.renewLease(this.row.deliveryId, this.row.leaseToken, this.leaseMs)
			return renewed && !this.stopped && !this.controller.signal.aborted
		} catch {
			return false
		}
	}

	start(): void {
		this.schedule()
	}

	private schedule(): void {
		if (this.stopped || this.controller.signal.aborted || this.refreshPromise) return
		this.timer = setTimeout(() => this.refresh(), Math.max(1_000, Math.floor(this.leaseMs / 3)))
	}

	private refresh(): void {
		if (this.stopped || this.controller.signal.aborted || this.refreshPromise) return
		const pending = this.refreshOnce()
		this.refreshPromise = pending
		void pending.finally(() => {
			if (this.refreshPromise === pending) this.refreshPromise = undefined
			this.schedule()
		})
	}

	private async refreshOnce(): Promise<void> {
		try {
			const renewed = await this.renewLease(this.row.deliveryId, this.row.leaseToken, this.leaseMs)
			if (!renewed && !this.stopped) this.abort(new Error('Webhook delivery lease lost'))
		} catch {
			// Failing closed avoids a POST after DB ownership becomes uncertain.
			if (!this.stopped) this.abort(new Error('Webhook delivery lease lost'))
		}
	}

	/** Abort a pinned HTTP attempt and prevent any later lease refresh. */
	abort(reason = new Error('Webhook delivery stopped')): void {
		this.stop()
		if (!this.controller.signal.aborted) this.controller.abort(reason)
	}

	/** Stop timers and wait for an already-issued refresh before DB teardown. */
	async stopAndDrain(): Promise<void> {
		this.stop()
		await this.refreshPromise?.catch(() => undefined)
	}

	stop(): void {
		this.stopped = true
		if (this.timer) clearTimeout(this.timer)
		this.timer = undefined
	}
}

export class WebhookDeliveryWorker {
	private readonly options: BoundedWorkerOptions
	private started = false
	private stopping = false
	private running = false
	private timer: ReturnType<typeof setTimeout> | undefined
	private readonly inFlight = new Set<Promise<void>>()
	private readonly activeLeases = new Set<DeliveryLeaseHeartbeat>()
	private readonly auditTasks = new Set<Promise<void>>()
	private readonly activityWaiters = new Set<() => void>()
	private lastMaintenanceAt = 0

	constructor(options: WebhookDeliveryWorkerOptions = {}) {
		this.options = boundWorkerOptions(options)
	}

	start(): void {
		if (this.started || this.stopping) return
		this.started = true
		this.schedule(0)
	}

	private schedule(delayMs: number): void {
		if (!this.started || this.stopping) return
		this.timer = setTimeout(() => {
			void this.tick()
		}, delayMs)
	}

	private async tick(): Promise<void> {
		try {
			await this.runOnce()
		} catch {
			// The queue remains durable. Keep the message safe and let the next tick retry.
			logger.error('[delivery] worker pass failed')
		} finally {
			this.schedule(this.options.pollIntervalMs)
		}
	}

	private notifyActivity(): void {
		for (const resolve of this.activityWaiters) resolve()
		this.activityWaiters.clear()
	}

	private async waitForShutdownWork(deadline: number): Promise<boolean> {
		while (this.running || this.inFlight.size > 0 || this.auditTasks.size > 0) {
			const remaining = deadline - Date.now()
			if (remaining <= 0) return false
			let resolveActivity: (() => void) | undefined
			const activity = new Promise<void>((resolve) => {
				resolveActivity = resolve
				this.activityWaiters.add(resolve)
			})
			if (!this.running && this.inFlight.size === 0 && this.auditTasks.size === 0) {
				if (resolveActivity) this.activityWaiters.delete(resolveActivity)
				return true
			}
			await Promise.race([activity, new Promise<void>((resolve) => setTimeout(resolve, remaining))])
			if (resolveActivity) this.activityWaiters.delete(resolveActivity)
		}
		return true
	}

	async runOnce(): Promise<number> {
		if (this.running || this.stopping) return 0
		this.running = true
		try {
			const leaseToken = `${this.options.workerId}:${randomUUID()}`
			// Do not prefetch leases: rows behind a slow concurrent request could
			// expire before their first POST and be claimed by another process.
			const rows = await this.options.persistence.claimWebhookDeliveryOutbox(
				leaseToken,
				Math.min(this.options.batchSize, this.options.concurrency),
				this.options.leaseMs,
			)
			if (this.stopping) return 0
			await mapBounded(rows, this.options.concurrency, async (row) => {
				if (this.stopping) return
				const promise = this.process(row)
				this.inFlight.add(promise)
				this.notifyActivity()
				try {
					await promise
				} finally {
					this.inFlight.delete(promise)
					this.notifyActivity()
				}
			})
			if (!this.stopping) await this.maintainIfDue()
			return rows.length
		} finally {
			this.running = false
			this.notifyActivity()
		}
	}

	private async maintainIfDue(): Promise<void> {
		if (Date.now() - this.lastMaintenanceAt < this.options.maintenanceIntervalMs) return
		this.lastMaintenanceAt = Date.now()
		try {
			await this.options.persistence.runWebhookMaintenance({
				deliveredRetentionMs: this.options.deliveredRetentionMs,
				deadLetterRetentionMs: this.options.deadLetterRetentionMs,
			})
		} catch {
			logger.error('[delivery] maintenance failed')
		}
	}

	private recordDeliveryAudit(audit: DeliveryAuditEvent): void {
		if (this.stopping) return
		const task = persistDeliveryAudit(audit)
		this.auditTasks.add(task)
		this.notifyActivity()
		void task.finally(() => {
			this.auditTasks.delete(task)
			this.notifyActivity()
		})
	}

	private async process(row: ClaimedWebhookDelivery): Promise<void> {
		if (this.stopping) return
		let current: CurrentWebhookSubscription | null
		try {
			current = await this.options.persistence.getCurrentWebhookSubscription(row.ownerDid, row.webhookRkey)
		} catch {
			if (!this.stopping) await this.retry(row, { kind: 'subscription_unavailable', transient: true })
			return
		}
		if (this.stopping) return
		if (!current || current.url !== row.targetUrl || current.fingerprint !== row.subscriptionFingerprint) {
			if (
				!this.stopping &&
				(await this.options.persistence.cancelWebhookDeliveryForSubscriptionChange(row.deliveryId, row.leaseToken))
			) {
				const audit = auditFromPayload(row, 'failed')
				this.recordDeliveryAudit(audit)
			}
			return
		}

		let secret: string | undefined
		try {
			if (current.signingMode === 'secret_id') {
				const token = current.secretId
					? await this.options.persistence.getWebhookSecretToken(row.ownerDid, current.secretId)
					: null
				if (!token) throw new Error('secret unavailable')
				secret = token
			} else if (current.signingMode === 'record_secret') {
				const token = await this.options.persistence.getWebhookInlineSecret(row.ownerDid, row.webhookRkey)
				if (!token) throw new Error('secret unavailable')
				secret = token
			}
		} catch {
			if (!this.stopping) await this.retry(row, { kind: 'secret_unavailable', transient: true })
			return
		}
		if (this.stopping) return

		const lease = new DeliveryLeaseHeartbeat(
			row,
			this.options.leaseMs,
			this.options.persistence.renewWebhookDeliveryLease,
		)
		this.activeLeases.add(lease)
		try {
			// A DB timeout/lost lease before this point means do not send at all.
			if (!(await lease.acquire()) || this.stopping || lease.signal.aborted) return
			lease.start()
			if (this.stopping || lease.signal.aborted) return
			try {
				const result = await postDelivery(row, secret, this.options, lease.signal)
				if (this.stopping || lease.signal.aborted) return
				if (result.status >= 200 && result.status < 300) {
					if (
						await this.options.persistence.markWebhookDeliverySucceeded(row.deliveryId, row.leaseToken, result.status)
					) {
						const audit = auditFromPayload(row, 'ok')
						this.recordDeliveryAudit(audit)
					}
					return
				}
				await this.retry(row, classifyHttpStatus(result.status, result.retryAfterMs))
			} catch (error) {
				if (!this.stopping && !lease.signal.aborted) await this.retry(row, failureFromError(error))
			}
		} finally {
			await lease.stopAndDrain()
			this.activeLeases.delete(lease)
			this.notifyActivity()
		}
	}

	private async retry(row: ClaimedWebhookDelivery, failure: DeliveryFailure): Promise<void> {
		if (this.stopping) return
		const attemptNumber = row.attemptCount + 1
		const deadLetter = !failure.transient || attemptNumber >= this.options.maxAttempts
		const delay =
			failure.retryAfterMs === undefined
				? retryDelayMs(attemptNumber, this.options)
				: Math.min(this.options.retryMaxMs, failure.retryAfterMs)
		if (this.stopping) return
		const updated = await this.options.persistence.rescheduleWebhookDelivery(row.deliveryId, row.leaseToken, {
			nextAttemptAt: new Date(Date.now() + (deadLetter ? 0 : delay)).toISOString(),
			errorKind: failure.kind,
			httpStatus: failure.httpStatus,
			deadLetter,
		})
		if (updated && deadLetter && !this.stopping) {
			const audit = auditFromPayload(row, 'dead_letter')
			this.recordDeliveryAudit(audit)
		}
	}

	private abortActiveAttempts(): void {
		for (const lease of this.activeLeases) lease.abort()
	}

	/** Stop admission, grace-drain, then abort pinned HTTP attempts and heartbeat refreshes. */
	async stop(timeoutMs = this.options.requestTimeoutMs + 5_000): Promise<boolean> {
		this.stopping = true
		this.started = false
		if (this.timer) clearTimeout(this.timer)
		this.timer = undefined
		const timeout = boundedOption(timeoutMs, this.options.requestTimeoutMs + 5_000, 1, 10 * 60_000)
		const deadline = Date.now() + timeout
		// Reserve a bounded cleanup interval even for a short caller deadline.
		const cleanupBudget = Math.min(1_000, Math.max(25, Math.floor(timeout / 4)))
		const graceful = await this.waitForShutdownWork(Math.max(Date.now(), deadline - cleanupBudget))
		if (graceful) return true

		this.abortActiveAttempts()
		// Processes await stopAndDrain in their finally block, so a true result means
		// no heartbeat callback or pinned HTTP attempt remains before DB/Redis close.
		return this.waitForShutdownWork(deadline)
	}

	/** Graceful public drain; unlike stop it never aborts a healthy delivery. */
	async drain(timeoutMs = this.options.requestTimeoutMs + 5_000): Promise<boolean> {
		const deadline = Date.now() + boundedOption(timeoutMs, this.options.requestTimeoutMs + 5_000, 1, 10 * 60_000)
		return this.waitForShutdownWork(deadline)
	}
}

let defaultWorker: WebhookDeliveryWorker | undefined

export function startWebhookDeliveryWorker(options: WebhookDeliveryWorkerOptions = {}): WebhookDeliveryWorker {
	if (!defaultWorker) defaultWorker = new WebhookDeliveryWorker(options)
	defaultWorker.start()
	return defaultWorker
}

export async function stopWebhookDeliveryWorker(timeoutMs?: number): Promise<boolean> {
	if (!defaultWorker) return true
	const worker = defaultWorker
	defaultWorker = undefined
	return worker.stop(timeoutMs)
}

export async function drainWebhookDeliveryWorker(timeoutMs?: number): Promise<boolean> {
	return defaultWorker ? defaultWorker.drain(timeoutMs) : true
}