package connector_webhooks

import (
	"context"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/gin-gonic/gin"

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/ingressreceipts"
)

const (
	envConnectorVerifiedIngressReceipts             = "NEXUSCRM_CONNECTOR_VERIFIED_INGRESS_RECEIPTS"
	envConnectorWebhookVerificationLimit            = "NEXUSCRM_CONNECTOR_WEBHOOK_VERIFICATION_LIMIT"
	envConnectorWebhookVerificationWaitLimit        = "NEXUSCRM_CONNECTOR_WEBHOOK_VERIFICATION_WAIT_LIMIT"
	envConnectorWebhookVerificationWaitMilliseconds = "NEXUSCRM_CONNECTOR_WEBHOOK_VERIFICATION_WAIT_MS"
	envConnectorWebhookRetryAfterSeconds            = "NEXUSCRM_CONNECTOR_WEBHOOK_RETRY_AFTER_SECONDS"
	envConnectorWebhookAckDropEndpoints             = "NEXUSCRM_CONNECTOR_WEBHOOK_ACK_DROP_ENDPOINTS"
	envConnectorWebhookAckDropTimeoutMilliseconds   = "NEXUSCRM_CONNECTOR_WEBHOOK_ACK_DROP_TIMEOUT_MS"

	defaultConnectorWebhookVerificationLimit            = 64
	defaultConnectorWebhookVerificationWaitMilliseconds = 250
	defaultConnectorWebhookRetryAfterSecs               = 5
	maxConnectorWebhookVerificationLimit                = 4096
	maxConnectorWebhookVerificationWaitMilliseconds     = 5000
	maxConnectorWebhookRetryAfterSecs                   = 3600
	defaultConnectorWebhookAckDropTimeoutMilliseconds   = 750
	minConnectorWebhookAckDropTimeoutMilliseconds       = 1
	maxConnectorWebhookAckDropTimeoutMilliseconds       = 4000
)

type webhookAdmissionDecision struct {
	Allow      bool
	Reason     string
	RetryAfter time.Duration
}

type webhookIngressVerificationState string

const (
	webhookIngressUnverified webhookIngressVerificationState = "unverified"
	webhookIngressVerified   webhookIngressVerificationState = "verified"
)

type webhookIngressUnavailableDisposition uint8

const (
	webhookIngressRetry webhookIngressUnavailableDisposition = iota
	webhookIngressAckDrop
)

type webhookIngressUnavailablePolicy func(namespace, secretName string) webhookIngressUnavailableDisposition

type webhookIngressUnavailableResponse struct {
	reason          string
	err             error
	verification    webhookIngressVerificationState
	retryAfter      time.Duration
	recordNoReceipt bool
	// failClosed is reserved for protocol/trust failures that must never be
	// acknowledged as a business-approved drop.
	failClosed bool
}

// webhookIngressCapacityGate bounds public pre-auth work: body reads, pack and
// secret resolution, and signature verification. Keep this independent from
// endpoint fairness so an invalid-signature flood cannot consume DB or crypto
// capacity by minting arbitrary endpoint names.
type webhookIngressCapacityGate struct {
	tokens      chan struct{}
	waitTokens  chan struct{}
	retryAfter  time.Duration
	waitTimeout time.Duration
}

// webhookIngressCapacityLease keeps one verification-gate token alive while
// caller-independent snapshot work is still running. The request owns the
// initial reference; each shared load it starts owns an additional reference.
// This lets a timed-out request return without either leaking uncounted DB work
// or waiting indefinitely for a resolver that ignores cancellation.
type webhookIngressCapacityLease struct {
	mu      sync.Mutex
	refs    int
	release func()
}

type webhookIngressCapacityLeaseContextKey struct{}

func newWebhookIngressCapacityLease(release func()) *webhookIngressCapacityLease {
	if release == nil {
		release = func() {}
	}
	return &webhookIngressCapacityLease{
		refs:    1,
		release: release,
	}
}

func (l *webhookIngressCapacityLease) retain() func() {
	if l == nil {
		return func() {}
	}
	l.mu.Lock()
	if l.refs == 0 || l.release == nil {
		l.mu.Unlock()
		return func() {}
	}
	l.refs++
	l.mu.Unlock()

	var once sync.Once
	return func() {
		once.Do(l.releaseRef)
	}
}

func (l *webhookIngressCapacityLease) releaseRef() {
	if l == nil {
		return
	}
	var release func()
	l.mu.Lock()
	if l.refs > 0 {
		l.refs--
		if l.refs == 0 {
			release = l.release
			l.release = nil
		}
	}
	l.mu.Unlock()
	if release != nil {
		release()
	}
}

func withWebhookIngressCapacityLease(ctx context.Context, lease *webhookIngressCapacityLease) context.Context {
	if ctx == nil || lease == nil {
		return ctx
	}
	return context.WithValue(ctx, webhookIngressCapacityLeaseContextKey{}, lease)
}

func retainWebhookIngressCapacityLease(ctx context.Context) func() {
	if ctx == nil {
		return func() {}
	}
	lease, _ := ctx.Value(webhookIngressCapacityLeaseContextKey{}).(*webhookIngressCapacityLease)
	return lease.retain()
}

// webhookEndpointFairGate starts only after a delivery has authenticated. It
// retains the same global active and waiting bounds while round-robining the
// bounded wait queue across trusted endpoint identities.
type webhookEndpointFairGate struct {
	limit       int
	waitLimit   int
	retryAfter  time.Duration
	waitTimeout time.Duration

	mu            sync.Mutex
	active        int
	activeByKey   map[string]int
	waiting       int
	waitingByKey  map[string]int
	queues        map[string][]*webhookEndpointWaiter
	roundRobinKey []string
}

type webhookEndpointWaiter struct {
	key                string
	ready              chan struct{}
	rejected           chan struct{}
	queued             bool
	granted            bool
	rejectedByFairness bool
}

type webhookEndpointWaiterState uint8

const (
	webhookEndpointWaiterCanceled webhookEndpointWaiterState = iota
	webhookEndpointWaiterGranted
	webhookEndpointWaiterRejected
)

func newWebhookVerificationGateFromEnv(getenv func(string) string) *webhookIngressCapacityGate {
	if getenv == nil {
		getenv = os.Getenv
	}
	limit, waitLimit, waitTimeout, retryAfter := webhookVerificationGateConfigFromEnv(getenv)
	return newWebhookIngressCapacityGateWithWait(limit, retryAfter, waitLimit, waitTimeout)
}

func newWebhookEndpointFairGateFromEnv(getenv func(string) string) *webhookEndpointFairGate {
	if getenv == nil {
		getenv = os.Getenv
	}
	limit, waitLimit, waitTimeout, retryAfter := webhookVerificationGateConfigFromEnv(getenv)
	return newWebhookEndpointFairGateWithWait(limit, retryAfter, waitLimit, waitTimeout)
}

func webhookIngressUnavailablePolicyFromEnv(getenv func(string) string) webhookIngressUnavailablePolicy {
	if getenv == nil {
		getenv = os.Getenv
	}
	return webhookIngressUnavailablePolicyFromEndpoints(strings.Split(getenv(envConnectorWebhookAckDropEndpoints), ","))
}

func webhookIngressUnavailablePolicyFromEndpoints(endpoints []string) webhookIngressUnavailablePolicy {
	configured := webhookIngressAckDropEndpointSet(endpoints)
	if len(configured) == 0 {
		return nil
	}
	return func(namespace, secretName string) webhookIngressUnavailableDisposition {
		if _, ok := configured[canonicalWebhookEndpoint(namespace, secretName)]; ok {
			return webhookIngressAckDrop
		}
		return webhookIngressRetry
	}
}

func webhookIngressAckDropEndpointCount(endpoints []string) int {
	return len(webhookIngressAckDropEndpointSet(endpoints))
}

func webhookIngressAckDropEndpointSet(endpoints []string) map[string]struct{} {
	configured := make(map[string]struct{})
	for _, rawEndpoint := range endpoints {
		namespace, secretName, ok := strings.Cut(strings.TrimSpace(rawEndpoint), "/")
		if !ok {
			continue
		}
		canonical := canonicalWebhookEndpoint(namespace, secretName)
		if canonical != "" {
			configured[canonical] = struct{}{}
		}
	}
	return configured
}

func canonicalWebhookEndpoint(namespace, secretName string) string {
	namespace = strings.ToLower(strings.TrimSpace(namespace))
	secretName = strings.ToLower(strings.TrimSpace(strings.TrimPrefix(secretName, "/")))
	if namespace == "" || secretName == "" {
		return ""
	}
	return namespace + "/" + secretName
}

// SetIngressAckDropEndpoints replaces the local controlled-unavailable policy.
// Call it during bootstrap, before serving requests. An empty list restores the
// default retry behavior.
func (d *Dispatcher) SetIngressAckDropEndpoints(endpoints []string) {
	if d == nil {
		return
	}
	d.ingressUnavailablePolicy = webhookIngressUnavailablePolicyFromEndpoints(endpoints)
	d.ingressAckDropEndpointCount = webhookIngressAckDropEndpointCount(endpoints)
	d.logIngressPolicyReadback()
}

type webhookIngressPolicyReadback struct {
	verifiedIngressReceipts bool
	verificationLimit       int
	verificationWaitLimit   int
	verificationWaitMillis  int64
	retryAfterSeconds       int64
	ackDropEndpointCount    int
	ackDropTimeoutMillis    int64
}

func (d *Dispatcher) ingressPolicyReadback() webhookIngressPolicyReadback {
	readback := webhookIngressPolicyReadback{
		verifiedIngressReceipts: d.verifiedIngressReceiptsEnabled(),
		ackDropTimeoutMillis:    d.ingressAckDropTimeout().Milliseconds(),
	}
	if d == nil {
		return readback
	}
	readback.ackDropEndpointCount = d.ingressAckDropEndpointCount
	if d.verificationGate != nil {
		readback.verificationLimit = cap(d.verificationGate.tokens)
		readback.verificationWaitLimit = cap(d.verificationGate.waitTokens)
		readback.verificationWaitMillis = d.verificationGate.waitTimeout.Milliseconds()
		readback.retryAfterSeconds = int64(d.verificationGate.retryAfter / time.Second)
	}
	return readback
}

func (d *Dispatcher) logIngressPolicyReadback() {
	readback := d.ingressPolicyReadback()
	log.Printf("connector_webhooks: ingress_policy_readback verified_ingress_receipts=%t verification_limit=%d verification_wait_limit=%d verification_wait_ms=%d retry_after_seconds=%d ack_drop_endpoint_count=%d ack_drop_timeout_ms=%d",
		readback.verifiedIngressReceipts,
		readback.verificationLimit,
		readback.verificationWaitLimit,
		readback.verificationWaitMillis,
		readback.retryAfterSeconds,
		readback.ackDropEndpointCount,
		readback.ackDropTimeoutMillis,
	)
}

func webhookIngressAckDropTimeoutFromEnv(getenv func(string) string) time.Duration {
	if getenv == nil {
		getenv = os.Getenv
	}
	raw := strings.TrimSpace(getenv(envConnectorWebhookAckDropTimeoutMilliseconds))
	milliseconds, err := strconv.Atoi(raw)
	if raw == "" || err != nil || milliseconds < minConnectorWebhookAckDropTimeoutMilliseconds || milliseconds > maxConnectorWebhookAckDropTimeoutMilliseconds {
		milliseconds = defaultConnectorWebhookAckDropTimeoutMilliseconds
	}
	return time.Duration(milliseconds) * time.Millisecond
}

func (d *Dispatcher) ingressAckDropTimeout() time.Duration {
	if d == nil || d.ackDropTimeout < time.Duration(minConnectorWebhookAckDropTimeoutMilliseconds)*time.Millisecond || d.ackDropTimeout > time.Duration(maxConnectorWebhookAckDropTimeoutMilliseconds)*time.Millisecond {
		return time.Duration(defaultConnectorWebhookAckDropTimeoutMilliseconds) * time.Millisecond
	}
	return d.ackDropTimeout
}

func (d *Dispatcher) isIngressAckDropEndpoint(namespace, secretName string) bool {
	return d != nil && d.ingressUnavailablePolicy != nil && d.ingressUnavailablePolicy(namespace, secretName) == webhookIngressAckDrop
}

func webhookVerificationGateConfigFromEnv(getenv func(string) string) (limit, waitLimit int, waitTimeout, retryAfter time.Duration) {
	limit = positiveIntFromEnv(getenv, envConnectorWebhookVerificationLimit, defaultConnectorWebhookVerificationLimit, 1, maxConnectorWebhookVerificationLimit)
	waitLimit = positiveIntFromEnv(getenv, envConnectorWebhookVerificationWaitLimit, limit, 1, maxConnectorWebhookVerificationLimit)
	waitMillis := positiveIntFromEnv(getenv, envConnectorWebhookVerificationWaitMilliseconds, defaultConnectorWebhookVerificationWaitMilliseconds, 0, maxConnectorWebhookVerificationWaitMilliseconds)
	retryAfterSecs := positiveIntFromEnv(getenv, envConnectorWebhookRetryAfterSeconds, defaultConnectorWebhookRetryAfterSecs, 1, maxConnectorWebhookRetryAfterSecs)
	return limit, waitLimit, time.Duration(waitMillis) * time.Millisecond, time.Duration(retryAfterSecs) * time.Second
}

func newWebhookIngressCapacityGateWithWait(limit int, retryAfter time.Duration, waitLimit int, waitTimeout time.Duration) *webhookIngressCapacityGate {
	if limit <= 0 {
		return nil
	}
	if waitLimit < 0 {
		waitLimit = 0
	}
	if retryAfter <= 0 {
		retryAfter = time.Duration(defaultConnectorWebhookRetryAfterSecs) * time.Second
	}
	return &webhookIngressCapacityGate{
		tokens:      make(chan struct{}, limit),
		waitTokens:  make(chan struct{}, waitLimit),
		retryAfter:  retryAfter,
		waitTimeout: waitTimeout,
	}
}

func newWebhookEndpointFairGateWithWait(limit int, retryAfter time.Duration, waitLimit int, waitTimeout time.Duration) *webhookEndpointFairGate {
	if limit <= 0 {
		return nil
	}
	if waitLimit < 0 {
		waitLimit = 0
	}
	if retryAfter <= 0 {
		retryAfter = time.Duration(defaultConnectorWebhookRetryAfterSecs) * time.Second
	}
	return &webhookEndpointFairGate{
		limit:        limit,
		waitLimit:    waitLimit,
		retryAfter:   retryAfter,
		waitTimeout:  waitTimeout,
		activeByKey:  make(map[string]int),
		waitingByKey: make(map[string]int),
		queues:       make(map[string][]*webhookEndpointWaiter),
	}
}

func (d *Dispatcher) tryAcquireAdmission(_ context.Context) (func(), webhookAdmissionDecision) {
	return func() {}, webhookAdmissionDecision{Allow: true, Reason: "admission_unbounded"}
}

func (d *Dispatcher) tryAcquireReceiptVerification(ctx context.Context) (func(), webhookAdmissionDecision) {
	if d == nil || d.verificationGate == nil {
		return func() {}, webhookAdmissionDecision{Allow: true, Reason: "receipt_verification_unwired"}
	}
	return d.verificationGate.tryAcquire(ctx)
}

func (d *Dispatcher) tryAcquireVerifiedEndpoint(ctx context.Context, endpoint string) (func(), webhookAdmissionDecision) {
	if d == nil || d.endpointGate == nil {
		return func() {}, webhookAdmissionDecision{Allow: true, Reason: "receipt_endpoint_unwired"}
	}
	return d.endpointGate.tryAcquire(ctx, endpoint)
}

func (g *webhookIngressCapacityGate) tryAcquire(ctx context.Context) (func(), webhookAdmissionDecision) {
	select {
	case g.tokens <- struct{}{}:
		return func() { <-g.tokens }, webhookAdmissionDecision{Allow: true, Reason: "receipt_verification_capacity"}
	default:
	}
	if g.waitTimeout <= 0 || cap(g.waitTokens) == 0 {
		return func() {}, webhookAdmissionDecision{Allow: false, Reason: "receipt_verification_capacity_full", RetryAfter: g.retryAfter}
	}
	select {
	case g.waitTokens <- struct{}{}:
		defer func() { <-g.waitTokens }()
	default:
		return func() {}, webhookAdmissionDecision{Allow: false, Reason: "receipt_verification_wait_capacity_full", RetryAfter: g.retryAfter}
	}
	timer := time.NewTimer(g.waitTimeout)
	defer stopTimer(timer)
	select {
	case g.tokens <- struct{}{}:
		return func() { <-g.tokens }, webhookAdmissionDecision{Allow: true, Reason: "receipt_verification_capacity_waited"}
	case <-ctx.Done():
		return func() {}, webhookAdmissionDecision{Allow: false, Reason: "receipt_verification_context_done", RetryAfter: g.retryAfter}
	case <-timer.C:
		return func() {}, webhookAdmissionDecision{Allow: false, Reason: "receipt_verification_capacity_full", RetryAfter: g.retryAfter}
	}
}

func (g *webhookEndpointFairGate) tryAcquire(ctx context.Context, endpoint string) (func(), webhookAdmissionDecision) {
	if endpoint == "" {
		endpoint = "default"
	}
	g.mu.Lock()
	if g.active < g.limit && g.waiting == 0 {
		g.active++
		g.activeByKey[endpoint]++
		g.mu.Unlock()
		return func() { g.release(endpoint) }, webhookAdmissionDecision{Allow: true, Reason: "receipt_endpoint_capacity"}
	}
	if g.waitTimeout <= 0 || g.waitLimit == 0 {
		g.mu.Unlock()
		return func() {}, webhookAdmissionDecision{Allow: false, Reason: "receipt_endpoint_capacity_full", RetryAfter: g.retryAfter}
	}
	if g.waiting >= g.waitLimit && !g.reserveWaitSlotForIndependentEndpointLocked(endpoint) {
		g.mu.Unlock()
		return func() {}, webhookAdmissionDecision{Allow: false, Reason: "receipt_endpoint_wait_capacity_full", RetryAfter: g.retryAfter}
	}
	waiter := &webhookEndpointWaiter{
		key:      endpoint,
		ready:    make(chan struct{}, 1),
		rejected: make(chan struct{}, 1),
		queued:   true,
	}
	g.queues[endpoint] = append(g.queues[endpoint], waiter)
	if g.waitingByKey[endpoint] == 0 {
		g.roundRobinKey = append(g.roundRobinKey, endpoint)
	}
	g.waitingByKey[endpoint]++
	g.waiting++
	g.mu.Unlock()

	timer := time.NewTimer(g.waitTimeout)
	defer stopTimer(timer)
	select {
	case <-waiter.ready:
		return func() { g.release(endpoint) }, webhookAdmissionDecision{Allow: true, Reason: "receipt_endpoint_capacity_waited"}
	case <-waiter.rejected:
		return func() {}, webhookAdmissionDecision{Allow: false, Reason: "receipt_endpoint_wait_preempted", RetryAfter: g.retryAfter}
	case <-ctx.Done():
		return g.resolveWaiterAfterAbort(waiter, "receipt_endpoint_context_done")
	case <-timer.C:
		return g.resolveWaiterAfterAbort(waiter, "receipt_endpoint_capacity_full")
	}
}

func (g *webhookEndpointFairGate) resolveWaiterAfterAbort(waiter *webhookEndpointWaiter, canceledReason string) (func(), webhookAdmissionDecision) {
	switch waiter.cancel(g) {
	case webhookEndpointWaiterGranted:
		<-waiter.ready
		return func() { g.release(waiter.key) }, webhookAdmissionDecision{Allow: true, Reason: "receipt_endpoint_capacity_waited"}
	case webhookEndpointWaiterRejected:
		<-waiter.rejected
		return func() {}, webhookAdmissionDecision{Allow: false, Reason: "receipt_endpoint_wait_preempted", RetryAfter: g.retryAfter}
	default:
		return func() {}, webhookAdmissionDecision{Allow: false, Reason: canceledReason, RetryAfter: g.retryAfter}
	}
}

func (w *webhookEndpointWaiter) cancel(g *webhookEndpointFairGate) webhookEndpointWaiterState {
	g.mu.Lock()
	defer g.mu.Unlock()
	if w.granted {
		return webhookEndpointWaiterGranted
	}
	if w.rejectedByFairness {
		return webhookEndpointWaiterRejected
	}
	g.removeWaiterLocked(w)
	return webhookEndpointWaiterCanceled
}

func (g *webhookEndpointFairGate) release(endpoint string) {
	g.mu.Lock()
	g.active--
	g.activeByKey[endpoint]--
	if g.activeByKey[endpoint] == 0 {
		delete(g.activeByKey, endpoint)
	}
	waiter := g.popNextWaiterLocked()
	if waiter != nil {
		g.active++
		g.activeByKey[waiter.key]++
		waiter.granted = true
	}
	g.mu.Unlock()
	if waiter != nil {
		waiter.ready <- struct{}{}
	}
}

func (g *webhookEndpointFairGate) popNextWaiterLocked() *webhookEndpointWaiter {
	for len(g.roundRobinKey) > 0 {
		key := g.roundRobinKey[0]
		g.roundRobinKey = g.roundRobinKey[1:]
		queue := g.queues[key]
		if len(queue) == 0 {
			continue
		}
		waiter := queue[0]
		queue = queue[1:]
		g.waiting--
		g.waitingByKey[key]--
		if len(queue) == 0 {
			delete(g.queues, key)
			delete(g.waitingByKey, key)
		} else {
			g.queues[key] = queue
			g.roundRobinKey = append(g.roundRobinKey, key)
		}
		waiter.queued = false
		return waiter
	}
	return nil
}

func (g *webhookEndpointFairGate) reserveWaitSlotForIndependentEndpointLocked(endpoint string) bool {
	if g.activeByKey[endpoint] > 0 || g.waitingByKey[endpoint] > 0 {
		return false
	}
	var candidate string
	maxWaiting := 0
	for key, count := range g.waitingByKey {
		if key == endpoint || (g.activeByKey[key] == 0 && count <= 1) || count <= maxWaiting {
			continue
		}
		candidate = key
		maxWaiting = count
	}
	if candidate == "" {
		return false
	}
	queue := g.queues[candidate]
	waiter := queue[len(queue)-1]
	waiter.rejectedByFairness = true
	g.removeWaiterLocked(waiter)
	waiter.rejected <- struct{}{}
	return true
}

func (g *webhookEndpointFairGate) removeWaiterLocked(waiter *webhookEndpointWaiter) {
	if waiter == nil || !waiter.queued {
		return
	}
	queue := g.queues[waiter.key]
	for i, candidate := range queue {
		if candidate != waiter {
			continue
		}
		queue = append(queue[:i], queue[i+1:]...)
		g.waiting--
		g.waitingByKey[waiter.key]--
		waiter.queued = false
		if len(queue) == 0 {
			delete(g.queues, waiter.key)
			delete(g.waitingByKey, waiter.key)
			g.removeRoundRobinKeyLocked(waiter.key)
		} else {
			g.queues[waiter.key] = queue
		}
		return
	}
}

func (g *webhookEndpointFairGate) removeRoundRobinKeyLocked(key string) {
	for i := 0; i < len(g.roundRobinKey); i++ {
		if g.roundRobinKey[i] != key {
			continue
		}
		g.roundRobinKey = append(g.roundRobinKey[:i], g.roundRobinKey[i+1:]...)
		return
	}
}

func (d *Dispatcher) respondReceiptUnavailable(c *gin.Context, namespace, secretName string, rawBody []byte, reason string, err error) {
	d.respondReceiptUnavailableWithRetryAfter(c, namespace, secretName, rawBody, reason, err, 0)
}

func (d *Dispatcher) respondReceiptUnavailableWithRetryAfter(c *gin.Context, namespace, secretName string, rawBody []byte, reason string, err error, retryAfter time.Duration) {
	if retryAfter <= 0 {
		retryAfter = d.receiptRetryAfter()
	}
	d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
		reason:          reason,
		err:             err,
		verification:    webhookIngressVerified,
		retryAfter:      retryAfter,
		recordNoReceipt: true,
	})
}

// respondIngressUnavailable is the only internal-unavailable response writer.
// Explicitly allowlisted disposable endpoints acknowledge dependency/overload
// failures with 204; the default contract remains the original 503 response.
func (d *Dispatcher) respondIngressUnavailable(c *gin.Context, namespace, secretName string, rawBody []byte, response webhookIngressUnavailableResponse) {
	reason := strings.TrimSpace(response.reason)
	if reason == "" {
		reason = "unknown"
	}
	verification := response.verification
	if verification == "" {
		verification = webhookIngressUnverified
	}
	if !response.failClosed && d.isIngressAckDropEndpoint(namespace, secretName) {
		c.Writer.Header().Del("Retry-After")
		shedReason := "ack_drop:" + string(verification) + ":" + reason
		markConnectorWebhookIngressFailure(c, shedReason)
		recordConnectorWebhookAckDrop(string(verification), webhookIngressUnavailableMetricReason(reason))
		log.Printf("connector_webhooks: ingress unavailable acknowledged as contract drop namespace=%q endpoint=%q verification=%s reason=%s body_size=%d body_hash=%s error=%v",
			namespace,
			canonicalWebhookEndpoint(namespace, secretName),
			verification,
			reason,
			len(rawBody),
			connector_ingress_receipts.BodyHash(rawBody),
			response.err,
		)
		c.Status(http.StatusNoContent)
		return
	}

	markConnectorWebhookIngressFailure(c, reason)
	if response.retryAfter > 0 {
		c.Header("Retry-After", strconv.Itoa(int(response.retryAfter.Seconds())))
	}
	if response.recordNoReceipt || verification == webhookIngressUnverified {
		recordConnectorWebhookNoReceipt503(reason)
	}
	log.Printf("connector_webhooks: ingress unavailable namespace=%q endpoint=%q verification=%s reason=%s retry_after_sec=%d body_size=%d body_hash=%s error=%v",
		namespace,
		canonicalWebhookEndpoint(namespace, secretName),
		verification,
		reason,
		int(response.retryAfter.Seconds()),
		len(rawBody),
		connector_ingress_receipts.BodyHash(rawBody),
		response.err,
	)
	c.JSON(http.StatusServiceUnavailable, gin.H{"error": "service unavailable"})
}

func webhookIngressUnavailableMetricReason(reason string) string {
	if strings.HasPrefix(reason, "admission_not_allowed:") {
		return "admission_not_allowed"
	}
	switch reason {
	case "receipt_verification_capacity_full",
		"receipt_verification_wait_capacity_full",
		"receipt_verification_context_done",
		"receipt_endpoint_capacity_full",
		"receipt_endpoint_wait_capacity_full",
		"receipt_endpoint_wait_preempted",
		"receipt_endpoint_context_done",
		"body_read_deadline_exceeded",
		"pack_resolve_failed",
		"pack_transient_check_failed",
		"pack_mid_apply",
		"secret_pack_incarnation_changed",
		"secret_mid_apply_check_failed",
		"secret_pack_mid_apply",
		"secret_resolve_failed",
		"secret_refresh_pack_incarnation_changed",
		"secret_refresh_mid_apply_check_failed",
		"secret_refresh_pack_mid_apply",
		"secret_refresh_failed",
		"signature_capability_pack_incarnation_changed",
		"signature_capability_mid_apply_check_failed",
		"signature_capability_pack_mid_apply",
		"signature_capability_resolve_failed",
		"signature_verify_failed",
		"admission_gate_failed",
		"receipt_service_unwired",
		"receipt_create_failed",
		"receipt_cipher_unavailable",
		"receipt_guard_stale",
		"parser_capability_pack_incarnation_changed",
		"parser_capability_mid_apply_check_failed",
		"parser_capability_pack_mid_apply",
		"parser_capability_missing",
		"parser_capability_resolve_failed",
		"parser_config_drift",
		"parser_failed",
		"parser_empty_idempotency_key",
		"parser_blank_event_type",
		"route_not_declared",
		"route_resolve_failed",
		"companion_resolve_failed",
		"companion_consent_failed",
		"parser_empty_typed_fields",
		"parser_partial_source_lane",
		"parser_payload_invalid",
		"queue_envelope_encode_failed",
		"enqueue_pack_or_companion_state_changed",
		"enqueue_failed":
		return reason
	default:
		return "unknown"
	}
}

func (d *Dispatcher) receiptRetryAfter() time.Duration {
	if d != nil && d.verificationGate != nil && d.verificationGate.retryAfter > 0 {
		return d.verificationGate.retryAfter
	}
	return time.Duration(defaultConnectorWebhookRetryAfterSecs) * time.Second
}

func (d *Dispatcher) logAdmissionDecision(namespace string, decision webhookAdmissionDecision) {
	recordConnectorWebhookAdmission(decision)
	log.Printf("connector_webhooks: admission namespace=%q allow=%t reason=%s retry_after_sec=%d",
		namespace,
		decision.Allow,
		decision.Reason,
		int(decision.RetryAfter.Seconds()),
	)
}

func (d *Dispatcher) logEndpointAdmissionDecision(namespace, endpoint string, decision webhookAdmissionDecision) {
	recordConnectorWebhookAdmission(decision)
	log.Printf("connector_webhooks: endpoint admission namespace=%q endpoint=%q allow=%t reason=%s retry_after_sec=%d",
		namespace,
		endpoint,
		decision.Allow,
		decision.Reason,
		int(decision.RetryAfter.Seconds()),
	)
}

func connectorVerifiedIngressReceiptsEnabledFromEnv() bool {
	switch strings.ToLower(strings.TrimSpace(os.Getenv(envConnectorVerifiedIngressReceipts))) {
	case "1", "true", "yes", "on":
		return true
	default:
		return false
	}
}

func positiveIntFromEnv(getenv func(string) string, key string, defaultValue int, minValue int, maxValue int) int {
	raw := strings.TrimSpace(getenv(key))
	if raw == "" {
		return defaultValue
	}
	n, err := strconv.Atoi(raw)
	if err != nil {
		return defaultValue
	}
	if n < minValue {
		return minValue
	}
	if maxValue > 0 && n > maxValue {
		return maxValue
	}
	return n
}