// Package connector_webhooks implements the generic inbound webhook dispatcher
// (NCRM-INT-04 / #338). All inbound provider payloads arrive at
// POST /webhooks/:namespace/:path and are dispatched through the capability
// registry without any provider-name branch.
package connector_webhooks

import (
	"context"
	"crypto/sha256"
	"crypto/subtle"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/url"
	"os"
	"strings"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/google/uuid"
	contractconstants "github.com/pingcap-inc/nexuscrm-contract/pkg/constants"

	connector_admission "github.com/pingcap-inc/nexuscrm-connectors/features/connector/admission"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/capabilities"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inbound"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/ingressreceipts"
	connector_packstate "github.com/pingcap-inc/nexuscrm-connectors/features/connector/packstate"
	"github.com/pingcap-inc/nexuscrm-connectors/features/namedcredential"
	"github.com/pingcap-inc/nexuscrm-connectors/features/webhooks"
)

// maxInboundWebhookBodyBytes is the maximum allowed request body size for
// inbound webhook payloads (1 MiB). This is a compile-time constant, not an
// env-driven parameter — size is bounded before signature verification to
// prevent OOM reads of adversarial payloads.
const maxInboundWebhookBodyBytes int64 = 1 << 20

func resolvedWebhookSecretEqual(left, right webhooks.ResolvedWebhookSecret) bool {
	if strings.TrimSpace(left.Algorithm) != strings.TrimSpace(right.Algorithm) {
		return false
	}
	leftHash := strings.TrimSpace(left.SecretValueHash)
	rightHash := strings.TrimSpace(right.SecretValueHash)
	if leftHash != "" || rightHash != "" {
		return leftHash != "" && leftHash == rightHash
	}
	return len(left.Secret) == len(right.Secret) && subtle.ConstantTimeCompare(left.Secret, right.Secret) == 1
}

func (d *Dispatcher) refreshWebhookSecretForDispatch(
	ctx context.Context,
	c *gin.Context,
	timing *webhookDispatchTiming,
	namespace, secretName, packInstallID string,
	rawBody []byte,
	ackDropEndpoint bool,
	stage string,
) (*webhooks.ResolvedWebhookSecret, bool) {
	freshSecret, refreshErr := d.refreshWebhookSecret(ctx, namespace, secretName, packInstallID, ackDropEndpoint)
	if refreshErr == nil {
		return freshSecret, true
	}
	if errors.Is(refreshErr, webhooks.ErrWebhookSecretIncarnationChanged) {
		log.Printf("connector_webhooks: cached secret refresh observed pack incarnation change for namespace=%q", namespace)
		d.evictWebhookIngressSnapshot(namespace)
		timing.mark(stage, "changed")
		d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
			reason:       "secret_refresh_pack_incarnation_changed",
			err:          refreshErr,
			verification: webhookIngressUnverified,
		})
		return nil, false
	}
	if errors.Is(refreshErr, webhooks.ErrWebhookSecretNotAvailable) {
		midApply, midApplyErr := d.packTransient(ctx, namespace)
		if midApplyErr != nil {
			timing.mark(stage, "error")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "secret_refresh_mid_apply_check_failed",
				err:          midApplyErr,
				verification: webhookIngressUnverified,
			})
			return nil, false
		}
		if !midApply {
			d.evictWebhookIngressSnapshot(namespace)
			timing.mark(stage, "rejected")
			c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
			return nil, false
		}
		timing.mark(stage, "unavailable")
		d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
			reason:       "secret_refresh_pack_mid_apply",
			err:          refreshErr,
			verification: webhookIngressUnverified,
		})
		return nil, false
	}
	log.Printf("connector_webhooks: cached secret refresh failed for namespace=%q: %v", namespace, refreshErr)
	timing.mark(stage, "error")
	d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
		reason:       "secret_refresh_failed",
		err:          refreshErr,
		verification: webhookIngressUnverified,
	})
	return nil, false
}

// Dispatcher is the generic inbound webhook handler. It has no knowledge of
// any specific connector provider. All provider-specific logic (signature
// canonicalization, envelope parsing) lives behind registered capabilities.
//
// All fields are injectable so tests can drive the full pipeline with fakes
// without a real database.
type Dispatcher struct {
	resolvePack                 func(ctx context.Context, namespace string) (*connector_packstate.ConnectorPackRow, bool, error)
	packTransient               func(ctx context.Context, namespace string) (bool, error)
	resolveSecret               func(ctx context.Context, namespace, name, packInstallID string) (*webhooks.ResolvedWebhookSecret, error)
	registry                    *connectorcapabilities.Registry
	routes                      connector_inbound_routing.RouteResolver
	routeCache                  *materializerRouteCache
	parserSingleflight          *materializerParserSingleflight
	ingressCache                *webhookIngressSnapshotCache
	requireLivePackAuthority    bool
	enqueueBatch                func(ctx context.Context, namespace, packInstallID, deliveryID string, events []connector_events.BatchEvent) ([]string, []bool, error)
	resolveCompanion            func(ctx context.Context, namespace, flowDeveloperName string) (CompanionFlowIncarnation, bool, error)
	checkConsent                func(ctx context.Context, companionPackName, companionPackInstallID, namespace, providerPackVersion string) error
	companionSingleflight       *materializerCompanionSingleflight
	consentSingleflight         *materializerConsentSingleflight
	createReceipt               func(ctx context.Context, in connector_ingress_receipts.VerifiedReceiptInput) (*connector_ingress_receipts.Receipt, bool, error)
	receiptRouterWake           func()
	receiptsEnabled             func() bool
	admissionGate               connector_admission.Gate
	maxBodyBytes                int64
	verificationGate            *webhookIngressCapacityGate
	endpointGate                *webhookEndpointFairGate
	ingressUnavailablePolicy    webhookIngressUnavailablePolicy
	ingressAckDropEndpointCount int
	ackDropTimeout              time.Duration
}

// NewDispatcher wires the production Dispatcher from a live *sql.DB, cipher,
// capability registry, connector-events service, and route resolver. Pass
// connectorcapabilities.Default() as registry after calling
// connectorcapabilities.BindDefaultCapabilityGate at bootstrap.
//
// routes is used to snapshot the RouteVersion at enqueue time so backlog events
// are pinned to the immutable route version in effect when they arrived.
// A nil routes panics at construction time — pass
// connector_inbound_routing.NewDBRouteResolver(db) in production.
func NewDispatcher(
	db *sql.DB,
	cipher *namedcredential.SecretCipher,
	registry *connectorcapabilities.Registry,
	events *connector_events.Service,
	routes connector_inbound_routing.RouteResolver,
) *Dispatcher {
	if routes == nil {
		panic("connector_webhooks.NewDispatcher: routes RouteResolver must not be nil")
	}
	ackDropEndpoints := strings.Split(os.Getenv(envConnectorWebhookAckDropEndpoints), ",")
	dispatcher := &Dispatcher{
		resolvePack: func(ctx context.Context, ns string) (*connector_packstate.ConnectorPackRow, bool, error) {
			return connector_packstate.ResolveConnectorPackByNamespace(ctx, db, ns)
		},
		packTransient: func(ctx context.Context, ns string) (bool, error) {
			return connector_packstate.IsConnectorPackMidApply(ctx, db, ns)
		},
		resolveSecret: func(ctx context.Context, ns, name, packInstallID string) (*webhooks.ResolvedWebhookSecret, error) {
			return webhooks.ResolveWebhookSecretForVerification(ctx, db, cipher, ns, name, packInstallID)
		},
		registry:           registry,
		routes:             routes,
		routeCache:         newMaterializerRouteCache(defaultMaterializerRouteCacheTTL),
		parserSingleflight: newMaterializerParserSingleflight(),
		ingressCache:       newWebhookIngressSnapshotCache(defaultWebhookIngressSnapshotCacheTTL),
		enqueueBatch:       events.EnqueueBatch,
		resolveCompanion: func(ctx context.Context, ns, flowDeveloperName string) (CompanionFlowIncarnation, bool, error) {
			return ResolveCompanionFlowIncarnation(ctx, db, ns, flowDeveloperName)
		},
		checkConsent: func(ctx context.Context, companionPackName, companionPackInstallID, ns, providerPackVersion string) error {
			return CheckCompanionConsent(ctx, db, companionPackName, companionPackInstallID, ns, providerPackVersion)
		},
		companionSingleflight:       newMaterializerCompanionSingleflight(),
		consentSingleflight:         newMaterializerConsentSingleflight(),
		createReceipt:               connector_ingress_receipts.NewService(db, cipher, nil, nil).CreateVerifiedReceipt,
		receiptsEnabled:             connectorVerifiedIngressReceiptsEnabledFromEnv,
		admissionGate:               connector_admission.NewSQLGate(db),
		maxBodyBytes:                maxInboundWebhookBodyBytes,
		verificationGate:            newWebhookVerificationGateFromEnv(os.Getenv),
		endpointGate:                newWebhookEndpointFairGateFromEnv(os.Getenv),
		ingressUnavailablePolicy:    webhookIngressUnavailablePolicyFromEndpoints(ackDropEndpoints),
		ingressAckDropEndpointCount: webhookIngressAckDropEndpointCount(ackDropEndpoints),
		ackDropTimeout:              webhookIngressAckDropTimeoutFromEnv(os.Getenv),
	}
	dispatcher.logIngressPolicyReadback()
	return dispatcher
}

// RegisterRoutes mounts the dispatcher on g. Callers MUST NOT attach auth
// middleware to g — HMAC signature verification is the trust boundary.
//
// The trailing segment is a catch-all (*path), not a single :path segment,
// because a connector pack's webhook secret Name may itself contain slashes —
// it is only lowercased/trimmed at intake (packs.NewWebhookSecretComponent
// applies no charset restriction), e.g. "orders/default". A single-segment
// route would let such a secret pass apply/verify yet be unreachable at runtime.
func (d *Dispatcher) RegisterRoutes(g *gin.RouterGroup) {
	g.POST("/:namespace/*path", d.Dispatch)
}

// Dispatch is the main handler. It processes one inbound webhook request through
// the fail-closed pipeline:
// Internal-unavailable outcomes use one dual contract: default endpoints return
// retriable 503, while explicitly allowlisted disposable endpoints return 204
// and drop. The request-URI trust failure is the sole fail-closed 503 exception.
//
//  1. Bounded body read (413 if exceeded — OOM guard before any verification).
//  2. Pack active check (404 if genuinely missing/inactive; internal-unavailable if in a transient
//     install/removal window — needed for 404 vs 401 distinction because resolveSecret
//     collapses inactive pack to ErrWebhookSecretNotAvailable).
//  3. Secret resolution (401 if genuinely missing/inactive; internal-unavailable if
//     server-side cipher/DB/decrypt resolution fails).
//  4. Signature verifier resolution (401 if capability genuinely unavailable;
//     internal-unavailable if the capability gate hit an internal/transient error).
//  5. Signature verification (401 on bad sig).
//  6. Verified-endpoint capacity gate (internal-unavailable under bounded ingress pressure;
//     default responses include Retry-After, and waiters are round-robin by endpoint).
//  7. Admission gate (internal-unavailable while namespace/channel is quiescing or paused;
//     default responses include Retry-After).
//  8. Envelope parser resolution (internal-unavailable on server misconfiguration).
//  9. Envelope parsing (400 — malformed client payload).
//  10. Reject the whole batch if any event has an empty IdempotencyKey (internal-unavailable —
//     would otherwise silently dedup distinct events), else enqueue each
//     parsed event (internal-unavailable on enqueue failure).
//
// 11. 202 Accepted with {"eventIds": [...]}.
//
// Error responses are opaque: the internal cause is logged server-side only;
// secret bytes, headers, and body are never echoed.
func (d *Dispatcher) Dispatch(c *gin.Context) {
	// Canonicalize namespace to lowercase once at route entry so that all
	// downstream lookups — pack DB query (case-insensitive via utf8mb4_unicode_ci
	// collation), secret DB query, and the in-memory capability Registry (a Go map,
	// which IS case-sensitive) — all operate on a consistent key.
	namespace := strings.ToLower(strings.TrimSpace(c.Param("namespace")))
	// *path captures a leading slash and may itself contain slashes (webhook
	// secret Names can contain '/'); strip the leading slash to recover the name,
	// then canonicalize with the SAME rules the webhook-secret surface uses at
	// authoring and lookup (strings.ToLower + strings.TrimSpace — internal '/'
	// preserved). Using the canonical form for BOTH resolveSecret and
	// scopeIdempotencyKey ensures case/whitespace aliases of one secret path
	// (e.g. "Events" vs "events") resolve the same secret AND produce the same
	// endpoint-scoped idempotency key, so replay protection cannot be bypassed
	// via path aliases.
	secretName := strings.ToLower(strings.TrimSpace(strings.TrimPrefix(c.Param("path"), "/")))
	requestID := webhookIngressRequestID(c)
	packInstallID := ""
	ctx := c.Request.Context()
	connectorWebhookDispatchInflight.Inc()
	defer connectorWebhookDispatchInflight.Dec()
	ackDropEndpoint := d.isIngressAckDropEndpoint(namespace, secretName)
	var readDeadlineController *http.ResponseController
	readDeadlineSet := false
	clearReadDeadline := func() {}
	if ackDropEndpoint {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, d.ingressAckDropTimeout())
		defer cancel()
		c.Request = c.Request.WithContext(ctx)
		if deadline, ok := ctx.Deadline(); ok {
			readDeadlineController = http.NewResponseController(c.Writer)
			if err := readDeadlineController.SetReadDeadline(deadline); err != nil {
				reason := "set_failed"
				if errors.Is(err, http.ErrNotSupported) {
					reason = "unsupported"
				}
				log.Printf("connector_webhooks: ack-drop body read deadline unavailable reason=%s", reason)
			} else {
				readDeadlineSet = true
				readDeadlineCleared := false
				clearReadDeadline = func() {
					if readDeadlineCleared {
						return
					}
					readDeadlineCleared = true
					if clearErr := readDeadlineController.SetReadDeadline(time.Time{}); clearErr != nil {
						log.Printf("connector_webhooks: ack-drop body read deadline clear failed reason=clear_failed")
					}
				}
				defer clearReadDeadline()
			}
		}
	}
	// The same absolute allowlisted deadline bounds both direct Go body reads and
	// ctx-aware server dependencies. An upstream break-glass 204 remains the only
	// hard guarantee that no body is read before acknowledgement.
	receiptMode := d.verifiedIngressReceiptsEnabled()
	timing := newWebhookDispatchTiming(receiptMode)
	var verifiedEndpointTelemetry *verifiedWebhookEndpointTelemetry
	defer func() {
		status := c.Writer.Status()
		timing.finish(status)
		verifiedEndpointTelemetry.finish(status)
		if status == http.StatusServiceUnavailable {
			logWebhookIngress503(c, requestID, namespace, packInstallID, timing)
		}
	}()

	var releasePreVerification func()
	if receiptMode {
		var verification webhookAdmissionDecision
		releasePreVerification, verification = d.tryAcquireReceiptVerification(ctx)
		timing.mark("verification_gate", webhookStageOutcomeOK(verification.Allow))
		if !verification.Allow {
			d.logAdmissionDecision(namespace, verification)
			retryAfter := verification.RetryAfter
			if retryAfter <= 0 {
				retryAfter = d.receiptRetryAfter()
			}
			d.respondIngressUnavailable(c, namespace, secretName, nil, webhookIngressUnavailableResponse{
				reason:          verification.Reason,
				err:             errors.New(verification.Reason),
				verification:    webhookIngressUnverified,
				retryAfter:      retryAfter,
				recordNoReceipt: true,
			})
			return
		}
		d.logAdmissionDecision(namespace, verification)
		preVerificationLease := newWebhookIngressCapacityLease(releasePreVerification)
		releasePreVerification = preVerificationLease.releaseRef
		ctx = withWebhookIngressCapacityLease(ctx, preVerificationLease)
		c.Request = c.Request.WithContext(ctx)
		defer func() {
			if releasePreVerification != nil {
				releasePreVerification()
			}
		}()
	}

	// Step 1: bounded body read — must happen before any work to prevent OOM.
	c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, d.maxBodyBytes)
	rawBody, err := io.ReadAll(c.Request.Body)
	clearReadDeadline()
	if err != nil {
		var maxErr *http.MaxBytesError
		if errors.As(err, &maxErr) {
			// Oversized payload — reject before any verification work.
			timing.mark("body", "error")
			c.JSON(http.StatusRequestEntityTooLarge, gin.H{"error": "payload too large"})
			return
		}
		if ackDropEndpoint && readDeadlineSet && (errors.Is(ctx.Err(), context.DeadlineExceeded) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, os.ErrDeadlineExceeded)) {
			timing.mark("body", "deadline")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "body_read_deadline_exceeded",
				err:          err,
				verification: webhookIngressUnverified,
			})
			return
		}
		// Other read errors (client disconnect, truncated body) are client-side.
		log.Printf("connector_webhooks: read body namespace=%q: %v", namespace, err)
		timing.mark("body", "error")
		c.JSON(http.StatusBadRequest, gin.H{"error": "bad request"})
		return
	}
	timing.mark("body", "ok")
	// Step 2: pack active check — resolveSecret collapses inactive→ErrWebhookSecretNotAvailable
	// so we need an explicit check here to produce 404 (not 401) for inactive/missing packs.
	// A transient pack lifecycle state is internal-unavailable rather than terminal 404:
	// default endpoints retry, while allowlisted disposable endpoints acknowledge and drop.
	//
	// The resolved row is kept for Step 8: its InstallID (the pack incarnation ID) is reused
	// for both route resolution and enqueue so that both operations are bound to the same pack
	// incarnation.  Two separate DB reads (one here, one at Step 8) would open a race window
	// where an admin reinstalling the pack mid-dispatch could authorize with pack-A credentials
	// but enqueue events under pack-A' (B) — a security boundary violation.  Caching the
	// Step-2 result eliminates that window.
	packRow, ok, err := d.resolveWebhookPack(ctx, namespace, receiptMode, ackDropEndpoint)
	if err != nil {
		log.Printf("connector_webhooks: resolvePack namespace=%q: transient DB/infra fault; internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop: %v", namespace, err)
		timing.mark("pack", "error")
		d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
			reason:       "pack_resolve_failed",
			err:          err,
			verification: webhookIngressUnverified,
		})
		return
	}
	if !ok {
		transient, terr := d.packTransient(ctx, namespace)
		if terr != nil {
			log.Printf("connector_webhooks: packTransient namespace=%q: transient DB/infra fault; internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop: %v", namespace, terr)
			timing.mark("pack", "error")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "pack_transient_check_failed",
				err:          terr,
				verification: webhookIngressUnverified,
			})
			return
		}
		if transient {
			// Pack is mid install/removal (setup-apply / reapply window):
			// controlled-unavailable, not a terminal missing-pack response.
			timing.mark("pack", "unavailable")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "pack_mid_apply",
				verification: webhookIngressUnverified,
			})
			return
		}
		timing.mark("pack", "not_found")
		c.JSON(http.StatusNotFound, gin.H{"error": "connector pack not available"})
		return
	}
	timing.mark("pack", "ok")

	// Cache the pack incarnation identity from Step 2. All subsequent steps
	// (secret resolution, capability gate, route resolution, enqueue) must use
	// this same packInstallID so that an admin reinstall between any two steps is
	// detected (Steps 3/4 via internal-unavailable incarnation-change sentinels)
	// or fails at the Step-8 FK / row-lookup boundary under the same dual contract.
	packInstallID = packRow.InstallID

	// Step 3: secret resolution. secretName is the :path segment — selects which
	// webhook secret within the namespace to use for verification.
	// packRow.InstallID is threaded through so the resolver can validate that the
	// currently-active pack still matches the incarnation resolved at Step 2. A
	// A mismatch (admin reinstall in the Step-2→Step-3 window) surfaces as
	// ErrWebhookSecretIncarnationChanged → internal-unavailable, never terminal 401.
	secret, secretFromCache, err := d.resolveWebhookSecret(ctx, namespace, secretName, packInstallID, receiptMode, ackDropEndpoint)
	if err != nil {
		// ErrWebhookSecretIncarnationChanged: pack was reinstalled between Step 2
		// and Step 3. Default endpoints retry and re-resolve the active incarnation;
		// allowlisted disposable endpoints acknowledge and drop.
		if errors.Is(err, webhooks.ErrWebhookSecretIncarnationChanged) {
			log.Printf("connector_webhooks: pack incarnation changed between Step 2 and Step 3 for namespace=%q; internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop", namespace)
			d.evictWebhookIngressSnapshot(namespace)
			timing.mark("secret", "changed")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "secret_pack_incarnation_changed",
				err:          err,
				verification: webhookIngressUnverified,
			})
			return
		}
		// Distinguish a genuinely missing/inactive secret (terminal 401) from a
		// server-side resolution failure — cipher misconfig, DB outage, decrypt
		// error (ErrWebhookSecretLookupFailed) — which is internal-unavailable, not
		// terminal bad credentials. Default endpoints retry; allowlisted disposable
		// endpoints acknowledge and drop.
		//
		// ErrWebhookSecretNotAvailable has an additional sub-case: the pack may
		// have transitioned active→installing/removing between Step 2's pack
		// check and this internal resolution (ResolveConnectorPackByNamespace is
		// re-queried inside ResolveWebhookSecretForVerification, and the second
		// query can find the pack in a transient state that collapses to
		// ErrWebhookSecretNotAvailable). In that case the 401 would be wrong —
		// the credentials are valid but the pack is mid-apply. Apply the same
		// internal-unavailable logic used in Steps 2 and 4.
		if errors.Is(err, webhooks.ErrWebhookSecretNotAvailable) {
			midApply, maErr := d.packTransient(ctx, namespace)
			if maErr != nil {
				log.Printf("connector_webhooks: mid-apply check after secret miss for namespace=%q: transient DB/infra fault; internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop: %v", namespace, maErr)
				timing.mark("secret", "error")
				d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
					reason:       "secret_mid_apply_check_failed",
					err:          maErr,
					verification: webhookIngressUnverified,
				})
				return
			}
			if midApply {
				// Pack is mid install/removal: the secret miss is transient, not terminal.
				timing.mark("secret", "unavailable")
				d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
					reason:       "secret_pack_mid_apply",
					verification: webhookIngressUnverified,
				})
				return
			}
			timing.mark("secret", "not_found")
			c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
			return
		}
		log.Printf("connector_webhooks: secret resolution failed for namespace=%q: %v", namespace, err)
		timing.mark("secret", "error")
		d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
			reason:       "secret_resolve_failed",
			err:          err,
			verification: webhookIngressUnverified,
		})
		return
	}
	timing.mark("secret", "ok")

	// Step 4: signature verifier — capability not registered or gate not satisfied.
	// packInstallID is forwarded so the gate can detect a pack reinstall between
	// Step 2 and now, returning ErrCapabilityIncarnationChanged as internal-unavailable (not 401).
	verifier, err := d.resolveWebhookSignatureVerifier(ctx, namespace, packInstallID, receiptMode, ackDropEndpoint)
	if err != nil {
		// ErrCapabilityIncarnationChanged: pack was reinstalled between Step 2 and
		// Step 4. Default endpoints retry and re-resolve the active incarnation;
		// allowlisted disposable endpoints acknowledge and drop.
		if errors.Is(err, connectorcapabilities.ErrCapabilityIncarnationChanged) {
			log.Printf("connector_webhooks: pack incarnation changed between Step 2 and Step 4 for namespace=%q; internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop", namespace)
			d.evictWebhookIngressSnapshot(namespace)
			timing.mark("signature_capability", "changed")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "signature_capability_pack_incarnation_changed",
				err:          err,
				verification: webhookIngressUnverified,
			})
			return
		}
		// ErrCapabilityNotAvailable has two sub-cases:
		//   (a) capability genuinely not declared → terminal 401.
		//   (b) pack exists but is mid-apply (blank PackInstallId, hash drift, or
		//       non-active status) — IsCapabilityDeclaredByConnectorPack returns
		//       false (→ same sentinel) because its SQL WHERE requires a fully-
		//       applied steady row. A webhook arriving during setup-apply /
		//       reapply is internal-unavailable, not terminal 401.
		// Non-ErrCapabilityNotAvailable errors (DB error, gate not bound) are
		// internal-unavailable under the dual response contract.
		if errors.Is(err, connectorcapabilities.ErrCapabilityNotAvailable) {
			midApply, maErr := d.packTransient(ctx, namespace)
			if maErr != nil {
				log.Printf("connector_webhooks: mid-apply check after verifier miss for namespace=%q: transient DB/infra fault; internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop: %v", namespace, maErr)
				timing.mark("signature_capability", "error")
				d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
					reason:       "signature_capability_mid_apply_check_failed",
					err:          maErr,
					verification: webhookIngressUnverified,
				})
				return
			}
			if midApply {
				// Pack is present but not yet in a fully-applied steady state.
				timing.mark("signature_capability", "unavailable")
				d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
					reason:       "signature_capability_pack_mid_apply",
					verification: webhookIngressUnverified,
				})
				return
			}
			timing.mark("signature_capability", "not_found")
			c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
			return
		}
		log.Printf("connector_webhooks: signature-verifier resolution failed for namespace=%q: %v", namespace, err)
		timing.mark("signature_capability", "error")
		d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
			reason:       "signature_capability_resolve_failed",
			err:          err,
			verification: webhookIngressUnverified,
		})
		return
	}
	timing.mark("signature_capability", "ok")

	// Step 5: signature verification.
	requestURI, err := resolveWebhookRequestURI(c.Request)
	if err != nil {
		log.Printf("connector_webhooks: cannot resolve raw request target for namespace=%q: %v", namespace, err)
		timing.mark("verify", "error")
		d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
			reason:       "request_uri_resolve_failed",
			err:          err,
			verification: webhookIngressUnverified,
			failClosed:   true,
		})
		return
	}
	requestURL, err := resolveWebhookRequestURL(c.Request)
	if err != nil {
		log.Printf("connector_webhooks: cannot reconstruct trusted request URL for namespace=%q: %v", namespace, err)
		requestURL = ""
	}
	headers := flattenHeaders(c.Request.Header)
	verifierInput := &connectorcapabilities.VerifierInput{
		Headers:                      headers,
		RawBody:                      rawBody,
		Secret:                       secret.Secret,
		Algorithm:                    secret.Algorithm,
		SecretName:                   secretName,
		Method:                       c.Request.Method,
		RequestURI:                   requestURI,
		RequestURL:                   requestURL,
		EnableVerifiedIngressReceipt: receiptMode,
	}
	verifyErr := verifier.Verify(ctx, verifierInput)
	// A cached rejection may mean the provider has already rotated to a new
	// secret. A cached success without a delivery identity would fall through to
	// the legacy inline path, which has no receipt guard to reject a rotated or
	// deactivated cached credential. Live-refresh exactly those two cases.
	refreshCachedSecret := secretFromCache &&
		(errors.Is(verifyErr, connectorcapabilities.ErrInvalidSignature) ||
			(verifyErr == nil && receiptMode && strings.TrimSpace(verifierInput.DeliveryIdentity) == ""))
	if refreshCachedSecret {
		freshSecret, ok := d.refreshWebhookSecretForDispatch(
			ctx, c, timing, namespace, secretName, packInstallID, rawBody, ackDropEndpoint, "verify",
		)
		if !ok {
			return
		}
		secretChanged := !resolvedWebhookSecretEqual(*secret, *freshSecret)
		secret = freshSecret
		if secretChanged {
			verifierInput.Secret = freshSecret.Secret
			verifierInput.Algorithm = freshSecret.Algorithm
			verifierInput.DeliveryIdentity = ""
			verifierInput.ReceiptHeaderAllowlist = nil
			verifyErr = verifier.Verify(ctx, verifierInput)
		}
	}
	if verifyErr != nil {
		// Per the SignatureVerifierHandler contract, ErrInvalidSignature (and only
		// it) is a definitive bad signature → terminal 401. Any other error is a
		// server-side fault (handler bug / transient dependency) → internal-unavailable,
		// never the terminal bad-credentials response. Default endpoints retry;
		// allowlisted disposable endpoints acknowledge and drop.
		if errors.Is(verifyErr, connectorcapabilities.ErrInvalidSignature) {
			timing.mark("verify", "rejected")
			c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
			return
		}
		log.Printf("connector_webhooks: signature verifier fault for namespace=%q: %v", namespace, verifyErr)
		timing.mark("verify", "error")
		d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
			reason:       "signature_verify_failed",
			err:          verifyErr,
			verification: webhookIngressUnverified,
		})
		return
	}
	timing.mark("verify", "ok")
	if receiptMode {
		// The public pre-verification gate has now covered all untrusted work and
		// can be released. A separate fair gate covers the authenticated remainder
		// so a burst from one configured endpoint cannot crowd out another.
		if releasePreVerification != nil {
			releasePreVerification()
			releasePreVerification = nil
		}
	}

	// Endpoint telemetry starts only after the shared pre-verification capacity
	// token has been released. The opaque label is minted only for a verified
	// endpoint, so untrusted paths cannot create metric series.
	verifiedEndpointTelemetry = startVerifiedWebhookEndpointTelemetry(namespace, secretName, receiptMode, timing.start)

	if receiptMode {
		endpoint := canonicalWebhookEndpoint(namespace, secretName)
		releaseEndpoint, endpointDecision := d.tryAcquireVerifiedEndpoint(ctx, endpoint)
		timing.mark("endpoint_gate", webhookStageOutcomeOK(endpointDecision.Allow))
		if !endpointDecision.Allow {
			d.logEndpointAdmissionDecision(namespace, endpoint, endpointDecision)
			retryAfter := endpointDecision.RetryAfter
			if retryAfter <= 0 {
				retryAfter = d.receiptRetryAfter()
			}
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:          endpointDecision.Reason,
				err:             errors.New(endpointDecision.Reason),
				verification:    webhookIngressVerified,
				retryAfter:      retryAfter,
				recordNoReceipt: true,
			})
			return
		}
		d.logEndpointAdmissionDecision(namespace, endpoint, endpointDecision)
		defer releaseEndpoint()
	}

	admissionOK := d.checkAdmission(ctx, c, namespace, secretName, packInstallID, "webhook:"+secretName, rawBody)
	timing.mark("sql_admission", webhookStageOutcomeOK(admissionOK))
	if !admissionOK {
		return
	}

	if receiptMode && strings.TrimSpace(verifierInput.DeliveryIdentity) != "" {
		if d.createReceipt == nil {
			timing.mark("receipt_insert", "error")
			d.respondReceiptUnavailable(c, namespace, secretName, rawBody, "receipt_service_unwired", fmt.Errorf("receipt service is not wired"))
			return
		}
		receipt, created, err := d.createReceipt(ctx, connector_ingress_receipts.VerifiedReceiptInput{
			Namespace:              namespace,
			InstalledPackID:        packInstallID,
			PackVersion:            packRow.PackVersion,
			PackPayloadHash:        packRow.PayloadHash,
			SecretName:             secretName,
			SecretAlgorithm:        secret.Algorithm,
			SecretValueHash:        secret.SecretValueHash,
			VerifierCapabilityKind: string(connectorcapabilities.SignatureVerifier),
			ParserCapabilityKind:   string(connectorcapabilities.EnvelopeParser),
			RequestMethod:          c.Request.Method,
			RequestURI:             requestURI,
			RequestURL:             requestURL,
			DeliveryIdentity:       verifierInput.DeliveryIdentity,
			RawBody:                rawBody,
			Headers:                c.Request.Header,
			HeaderAllowlist:        verifierInput.ReceiptHeaderAllowlist,
			PurgeAfter:             connector_ingress_receipts.DefaultPurgeAfter(time.Now()),
		})
		if err != nil {
			reason := "receipt_create_failed"
			if errors.Is(err, connector_ingress_receipts.ErrCipherRequired) {
				reason = "receipt_cipher_unavailable"
			}
			if errors.Is(err, connector_ingress_receipts.ErrReceiptGuardFailed) {
				// The atomic receipt guard is the live validation boundary for
				// cached secrets that produced a delivery identity. Resolve the
				// current secret only on guard failure: a rotation/deactivation
				// becomes terminal 401, while unrelated pack/capability drift
				// remains controlled-unavailable. Reconcile at most once per
				// cooldown window and retain the last snapshot on refresh failure;
				// provider retries must not turn one stale guard into a cold-load
				// storm.
				releaseRefreshVerification, refreshDecision := d.tryAcquireReceiptVerification(ctx)
				timing.mark("receipt_guard_refresh_gate", webhookStageOutcomeOK(refreshDecision.Allow))
				d.logAdmissionDecision(namespace, refreshDecision)
				refreshGuard := refreshDecision.Allow
				if refreshGuard {
					refreshGuard = d.ingressCache == nil || d.ingressCache.beginReceiptGuardRefresh(namespace)
					if !refreshGuard {
						releaseRefreshVerification()
					} else {
						refreshLease := newWebhookIngressCapacityLease(releaseRefreshVerification)
						defer refreshLease.releaseRef()
						ctx = withWebhookIngressCapacityLease(ctx, refreshLease)
					}
				}
				if refreshGuard {
					freshSecret, ok := d.refreshWebhookSecretForDispatch(
						ctx, c, timing, namespace, secretName, packInstallID, rawBody, ackDropEndpoint, "receipt_insert",
					)
					if !ok {
						return
					}
					if !resolvedWebhookSecretEqual(*secret, *freshSecret) {
						freshVerifierInput := *verifierInput
						freshVerifierInput.Secret = freshSecret.Secret
						freshVerifierInput.Algorithm = freshSecret.Algorithm
						freshVerifierInput.DeliveryIdentity = ""
						freshVerifierInput.ReceiptHeaderAllowlist = nil
						switch freshVerifyErr := verifier.Verify(ctx, &freshVerifierInput); {
						case errors.Is(freshVerifyErr, connectorcapabilities.ErrInvalidSignature):
							timing.mark("receipt_insert", "rejected")
							c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
							return
						case freshVerifyErr != nil:
							timing.mark("receipt_insert", "error")
							d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
								reason:       "signature_verify_failed",
								err:          freshVerifyErr,
								verification: webhookIngressUnverified,
							})
							return
						}
					}
					if _, _, refreshErr := d.refreshWebhookPack(ctx, namespace, ackDropEndpoint); refreshErr != nil {
						log.Printf("connector_webhooks: receipt guard pack refresh failed namespace=%q: %v", namespace, refreshErr)
					}
				}
				reason = "receipt_guard_stale"
			}
			timing.mark("receipt_insert", "error")
			d.respondReceiptUnavailable(c, namespace, secretName, rawBody, reason, err)
			return
		}
		timing.mark("receipt_insert", "ok")
		if created && d.receiptRouterWake != nil {
			d.receiptRouterWake()
		}
		c.JSON(http.StatusAccepted, gin.H{"receiptId": receipt.ID})
		return
	}

	// Step 6: envelope parser — signature verified, so the request is legitimate.
	// packInstallID is forwarded so the gate can detect a pack reinstall between
	// Step 2 and now, returning ErrCapabilityIncarnationChanged as internal-unavailable.
	// This closes the incarnation race: a delivery authenticated against pack
	// incarnation A at Steps 2-5 cannot silently be parsed with incarnation B's
	// envelope mapping after a reinstall/reapply.
	parser, err := d.registry.ResolveEnvelopeParser(ctx, namespace, packInstallID)
	if err != nil {
		// ErrCapabilityIncarnationChanged: pack was reinstalled between Step 2 and
		// Step 6. Treat it as controlled-unavailable; retry-mode re-resolves the new
		// active incarnation, while explicitly configured best-effort endpoints drop.
		if errors.Is(err, connectorcapabilities.ErrCapabilityIncarnationChanged) {
			log.Printf("connector_webhooks: pack incarnation changed between Step 2 and Step 6 for namespace=%q; controlled-unavailable", namespace)
			d.evictWebhookIngressSnapshot(namespace)
			timing.mark("parser_capability", "changed")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "parser_capability_pack_incarnation_changed",
				err:          err,
				verification: webhookIngressVerified,
			})
			return
		}
		// ErrCapabilityNotAvailable has two sub-cases (mirrors Step 4):
		//   (a) pack exists but is mid-apply (blank PackInstallId, hash drift, or
		//       non-active status) → internal-unavailable. Default endpoints retry;
		//       allowlisted disposable endpoints acknowledge and drop.
		//   (b) steady-state pack genuinely declares no envelope parser → 500.
		//       Unlike Step 4 (which returns 401 for a missing verifier), here the
		//       signature is already verified, so a missing parser is a server-side
		//       pack-config gap, not an auth failure.
		if errors.Is(err, connectorcapabilities.ErrCapabilityNotAvailable) {
			midApply, maErr := d.packTransient(ctx, namespace)
			if maErr != nil {
				log.Printf("connector_webhooks: mid-apply check after parser miss for namespace=%q: transient DB/infra fault; internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop: %v", namespace, maErr)
				timing.mark("parser_capability", "error")
				d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
					reason:       "parser_capability_mid_apply_check_failed",
					err:          maErr,
					verification: webhookIngressVerified,
				})
				return
			}
			if midApply {
				// Pack is present but not yet in a fully-applied steady state:
				// controlled-unavailable after the request has been verified.
				timing.mark("parser_capability", "unavailable")
				d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
					reason:       "parser_capability_pack_mid_apply",
					verification: webhookIngressVerified,
				})
				return
			}
			// Steady-state pack declares no envelope parser. Signature already
			// verified → pack-config gap (CONFIG-DRIFT), not a programming invariant.
			// Default endpoints retry; allowlisted disposable endpoints acknowledge and drop.
			log.Printf("connector_webhooks: CONFIG-DRIFT namespace=%q: pack declares no envelope parser in steady state (likely incomplete pack install); internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop; admin should register the parser capability", namespace)
			timing.mark("parser_capability", "not_found")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "parser_capability_missing",
				err:          err,
				verification: webhookIngressVerified,
			})
			return
		}
		log.Printf("connector_webhooks: ResolveEnvelopeParser namespace=%q: %v", namespace, err)
		timing.mark("parser_capability", "error")
		d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
			reason:       "parser_capability_resolve_failed",
			err:          err,
			verification: webhookIngressVerified,
		})
		return
	}
	timing.mark("parser_capability", "ok")

	// Step 7: parse the envelope.
	//
	// Error classification (per NCRM-INT-11 classification contract):
	//   ErrMalformedPayload    — composite JSON at a typed-field path (client sent wrong shape)
	//                           → terminal 400. Provider must fix its payload; retries will fail.
	//   ErrParserConfigDrift   — typed-field path resolved to absent/null/empty (pack path
	//                           mis-configured) → internal-unavailable plus structured log.
	//   any other error        — server-side parser fault → internal-unavailable.
	events, err := parser.Parse(ctx, &connectorcapabilities.ParserInput{
		Payload:    rawBody,
		SecretName: secretName,
	})
	if err != nil {
		if errors.Is(err, connectorcapabilities.ErrParserConfigDrift) {
			log.Printf("connector_webhooks: CONFIG-DRIFT namespace=%q: parser typed-field path resolved to no scalar value (likely pack mis-configured path); internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop; admin should fix the path: %v", namespace, err)
			timing.mark("parse", "config_drift")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "parser_config_drift",
				err:          err,
				verification: webhookIngressVerified,
			})
			return
		}
		if errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
			timing.mark("parse", "rejected")
			c.JSON(http.StatusBadRequest, gin.H{"error": "malformed webhook payload"})
			return
		}
		log.Printf("connector_webhooks: envelope parser fault for namespace=%q: %v", namespace, err)
		timing.mark("parse", "error")
		d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
			reason:       "parser_failed",
			err:          err,
			verification: webhookIngressVerified,
		})
		return
	}

	// Validate the whole batch BEFORE enqueueing anything. Dedup is scoped to
	// (InstalledPackId, IdempotencyKey), so an empty key collides with every
	// other keyless event from this pack and folds them onto one row — accepting
	// it would silently drop distinct deliveries while still returning 202. A
	// parser that cannot emit a stable key is a capability-contract violation.
	//
	// Classification: an empty IdempotencyKey from the parser is the same
	// CONFIG-DRIFT class as empty SourceID/EventType — likely a mis-configured
	// IdempotencyKeyPath that resolved to absent/null/empty. Step 7's ErrParserConfigDrift
	// branch already catches this when emitEvent returns the sentinel; this check is
	// defense-in-depth for parsers that return a ConnectorEvent with an empty key
	// without wrapping ErrParserConfigDrift (e.g. third-party or future parsers).
	// Treat this as internal-unavailable and enqueue nothing rather than partially
	// accepting the batch. Default endpoints retry after an admin fix; allowlisted
	// disposable endpoints acknowledge and drop.
	for _, ev := range events {
		if strings.TrimSpace(ev.IdempotencyKey) == "" {
			log.Printf("connector_webhooks: CONFIG-DRIFT namespace=%q: parser returned event with empty IdempotencyKey (likely mis-configured IdempotencyKeyPath); internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop; admin should fix the path", namespace)
			timing.mark("parse", "config_drift")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "parser_empty_idempotency_key",
				verification: webhookIngressVerified,
			})
			return
		}
	}
	timing.mark("parse", "ok")

	// Step 8: enqueue each event.
	//
	// Route version snapshot (option 3 / R7 fix): use the packInstallID from the
	// packRow resolved at Step 2.  Both route resolution and enqueue are bound to
	// the same pack incarnation — the single DB read from Step 2 is the single
	// source of truth for this entire dispatch.  A second resolveActivePack call
	// here would open a race window: an admin reinstalling the pack between Step 2
	// and Step 8 could result in the webhook being authorized with pack-A's
	// credentials but enqueued under pack-A' (B) — a security boundary violation.
	// The Step-2 packRow is reused verbatim; no second DB read.
	//
	// In the rare case the pack is reinstalled mid-dispatch (between Step 2's
	// resolvePack and the final Enqueue call), Enqueue may fail with a
	// foreign-key or row-lookup error and use the dual internal-unavailable contract.
	//
	// If no route is declared for (packInstallID, eventKind), ErrRouteNotDeclared
	// uses the dual internal-unavailable contract. No durable row is created;
	// default endpoints retry within the provider budget, while allowlisted
	// disposable endpoints acknowledge and drop.
	//
	// The dedup key is scoped to the webhook endpoint (secretName) — a connector
	// pack may declare multiple named webhook secrets, and all of a pack's
	// endpoints share one envelope parser resolved by namespace, so without
	// endpoint scoping two endpoints that reuse a provider delivery ID would
	// collapse onto one (InstalledPackId, IdempotencyKey) row.

	batch, deliveryID, failureReason, ok := d.prepareInlineDispatchBatch(ctx, namespace, packInstallID, packRow.PackVersion, secretName, events, timing)
	if !ok {
		d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
			reason:       failureReason,
			verification: webhookIngressVerified,
		})
		return
	}

	// No events to enqueue (parser intentionally emitted none — legitimate no-op
	// delivery). Skip the DB round-trip and acknowledge directly.
	if len(batch) == 0 {
		timing.mark("enqueue", "skipped")
		c.JSON(http.StatusAccepted, gin.H{"eventIds": []string{}})
		return
	}

	enqueuedIDs, _, err := d.enqueueBatch(ctx, namespace, packInstallID, deliveryID, batch)
	if err != nil {
		var quiesced *connector_admission.QuiescedError
		if errors.As(err, &quiesced) {
			retryAfter := quiesced.Fence.RetryAfterSeconds
			if retryAfter <= 0 {
				retryAfter = 120
			}
			mode := strings.TrimSpace(quiesced.Fence.Mode)
			if mode == "" {
				mode = connector_admission.ModeQuiescing
			}
			log.Printf("connector_webhooks: transaction admission rejected enqueue namespace=%q mode=%q epoch=%d; retrying delivery", namespace, mode, quiesced.Fence.Epoch)
			timing.mark("enqueue", "quiesced")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "enqueue_admission_not_allowed:" + mode,
				err:          err,
				verification: webhookIngressVerified,
				retryAfter:   time.Duration(retryAfter) * time.Second,
				// A committed transaction fence has rejected materialization after
				// verification. Never let an allowlisted best-effort endpoint turn
				// that retry contract into a 204 acknowledgement: no Event or
				// WorkItem exists and the provider must retry after the fence lifts.
				failClosed: true,
			})
			return
		}
		if errors.Is(err, connector_events.ErrPackIncarnationChanged) {
			log.Printf("connector_webhooks: pack incarnation changed at enqueue for namespace=%q: %v", namespace, err)
			d.evictWebhookIngressSnapshot(namespace)
			timing.mark("enqueue", "error")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "enqueue_pack_or_companion_state_changed",
				err:          err,
				verification: webhookIngressVerified,
			})
			return
		}
		if errors.Is(err, connector_events.ErrCompanionConsentMissing) ||
			errors.Is(err, connector_events.ErrCompanionVersionUnsatisfied) {
			// Companion consent/version is a business/configuration failure, not the
			// typed pack-incarnation lifecycle case above. Default endpoints retry;
			// explicitly allowlisted disposable endpoints acknowledge and drop.
			log.Printf("connector_webhooks: companion state unavailable at enqueue namespace=%q; default endpoints retry, allowlisted disposable endpoints ack-drop: %v", namespace, err)
			timing.mark("enqueue", "error")
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "enqueue_pack_or_companion_state_changed",
				err:          err,
				verification: webhookIngressVerified,
			})
			return
		}
		log.Printf("connector_webhooks: enqueueBatch namespace=%q: transient DB/infra write fault; internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop: %v", namespace, err)
		timing.mark("enqueue", "error")
		d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
			reason:       "enqueue_failed",
			err:          err,
			verification: webhookIngressVerified,
		})
		return
	}
	timing.mark("enqueue", "ok")
	ids := enqueuedIDs

	// Step 9: acknowledge.
	c.JSON(http.StatusAccepted, gin.H{"eventIds": ids})
}

type inlineDispatchResolvedRoute struct {
	routeVersion      int64
	flowDeveloperName string
}

func (d *Dispatcher) prepareInlineDispatchBatch(
	ctx context.Context,
	namespace string,
	packInstallID string,
	packVersion string,
	secretName string,
	events []connectorcapabilities.ConnectorEvent,
	timing *webhookDispatchTiming,
) ([]connector_events.BatchEvent, string, string, bool) {
	// Phase A: validate all parser-emitted events before enqueuing any of them.
	// Validation and enqueue must not interleave: if event N validates but event
	// N+1 fails, we cannot have already durably enqueued event N (provider received
	// no response yet; internal-unavailable is safe for the whole batch). Separating the
	// phases ensures the enqueue side-effect only happens when ALL events pass
	// validation.
	validated := make([]validatedConnectorEvent, 0, len(events))

	// Request-scoped ROUTE-VERSION memo. ResolveFlow depends only on
	// (namespace, packInstallID, eventKind); within THIS request namespace and
	// packInstallID are fixed (packInstallID is the Step-2 packRow incarnation —
	// the single source of truth for this entire dispatch), so the only variable
	// is eventKind. A single provider delivery commonly carries N events of one
	// eventKind (for example, a provider-specific batch for one event type), so
	// without memoization the handler runs N identical ResolveFlow SELECTs serially
	// before enqueue. Memoizing the (version, flowDeveloperName) result per
	// eventKind collapses that to one route SELECT per distinct kind.
	//
	// Lifetime is strictly this HTTP request: the map is a local declared inside
	// this helper and never escapes. Promoting it to process/package scope would
	// be a stale-route hazard — routes change between requests, and snapshotting
	// to the first-seen version for one delivery already matches the existing
	// "Step-2 packRow is the SoT for this entire dispatch" invariant.
	//
	// Deliberately NOT memoized: the companion incarnation + consent gate. Unlike
	// ResolveFlow (which reads the provider pack's own route rows under the fixed
	// Step-2 packInstallID), ResolveCompanionFlowIncarnation reads the LIVE active
	// install of a SEPARATE companion app-pack from _System_InstalledPack and
	// CheckCompanionConsent validates it against the live _System_InstalledPackDependency
	// row on every call. The dispatcher intentionally rebinds the companion
	// incarnation per event at enqueue time: if the companion app-pack is reborn
	// while a large delivery is still validating, later same-kind events must pick
	// up the new active install, not a stale snapshot from the first event. Caching
	// it would pin a stale companion install and dead-letter those later events
	// (accepted-then-terminal). So companion resolution stays per-event below.
	routeMemo := make(map[string]inlineDispatchResolvedRoute, len(events))

	for _, ev := range events {
		preparedEvent, failureReason, ok := d.prepareInlineDispatchEvent(ctx, namespace, packInstallID, packVersion, secretName, ev, routeMemo, timing)
		if !ok {
			return nil, "", failureReason, false
		}
		validated = append(validated, preparedEvent)
	}
	timing.mark("route_validate", "ok")

	// Phase B preparation: derive one DeliveryId from the validated events'
	// scoped idempotency keys, then build the all-or-nothing enqueue batch. The
	// shared helper keeps delivery-id derivation and batch field mapping aligned.
	batch, deliveryID := buildConnectorEventBatch(validated)
	for i := range batch {
		// Inline dispatch has exactly one delivery per enqueue call, so preserve
		// its historical batch-level DeliveryID argument. Receipt-router batches
		// keep per-event DeliveryID because they can contain multiple deliveries.
		batch[i].DeliveryID = ""
	}
	return batch, deliveryID, "", true
}

func (d *Dispatcher) prepareInlineDispatchEvent(
	ctx context.Context,
	namespace string,
	packInstallID string,
	packVersion string,
	secretName string,
	ev connectorcapabilities.ConnectorEvent,
	routeMemo map[string]inlineDispatchResolvedRoute,
	timing *webhookDispatchTiming,
) (validatedConnectorEvent, string, bool) {
	// Guard: blank EventType from the parser would produce a routeVersion=0
	// sentinel + a durable enqueue with EventKind="" — the row is committed but
	// unprocessable and unreplayable because EventKind is the routing key.
	// A parser emitting blank EventType is a server-side contract violation
	// (pack envelope-mapping config is misconfigured or has a transient bug),
	// NOT a client-malformed payload. Returning 400 would make valid signed
	// deliveries terminally fail before durable enqueue. Use internal-unavailable:
	// default endpoints retry after the pack admin fixes the parser config, while
	// allowlisted disposable endpoints acknowledge and drop.
	if strings.TrimSpace(ev.EventType) == "" {
		log.Printf("connector_webhooks: parser for namespace=%q emitted blank EventType (idempotencyKey=%q); parser contract violation; internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop", namespace, ev.IdempotencyKey)
		timing.mark("route_validate", "error")
		return validatedConnectorEvent{}, "parser_blank_event_type", false
	}

	// Route lookup, memoized per eventKind for this delivery. A memo hit reuses
	// the (version, flowDeveloperName) a prior same-kind event already resolved —
	// the inputs (namespace, packInstallID, eventKind) are request-fixed, so a
	// re-resolution returns the identical result while paying for a redundant
	// SELECT inside the HTTP handler.
	var routeVersion int64
	var flowDeveloperName string
	if cached, ok := routeMemo[ev.EventType]; ok {
		routeVersion = cached.routeVersion
		flowDeveloperName = cached.flowDeveloperName
	} else {
		var routeErr error
		routeVersion, flowDeveloperName, routeErr = d.routes.ResolveFlow(ctx, namespace, packInstallID, ev.EventType)
		if routeErr != nil {
			// Both ErrRouteNotDeclared and transient resolver errors are
			// internal-unavailable — never a durable enqueue. Enqueuing an
			// ErrRouteNotDeclared delivery with a RouteVersion=0 sentinel would
			// 202-accept it and then deterministically fail-terminal at the worker,
			// with no provider retry and no recovery even after the route is later
			// declared (the snapshot would be permanently 0). Default endpoints retry
			// and can resolve a later route; allowlisted disposable endpoints acknowledge
			// and drop. No accepted-then-terminal row is created in either case.
			if errors.Is(routeErr, connector_inbound_routing.ErrRouteNotDeclared) {
				log.Printf("connector_webhooks: no route declared for namespace=%q eventType=%q; internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop", namespace, ev.EventType)
			} else {
				log.Printf("connector_webhooks: route resolution failed for namespace=%q eventType=%q: %v", namespace, ev.EventType, routeErr)
			}
			timing.mark("route_validate", "error")
			if errors.Is(routeErr, connector_inbound_routing.ErrRouteNotDeclared) {
				return validatedConnectorEvent{}, "route_not_declared", false
			}
			return validatedConnectorEvent{}, "route_resolve_failed", false
		}
		// routeVersion is now guaranteed >= 1 (ResolveFlow returns the latest live
		// version for a declared route).
		routeMemo[ev.EventType] = inlineDispatchResolvedRoute{routeVersion: routeVersion, flowDeveloperName: flowDeveloperName}
	}

	// Companion incarnation snapshot + consent gate (§4 of design). NOT memoized:
	// this reads the live active install of the companion app-pack and validates
	// consent against the live dependency row, so it must rebind per event — a
	// companion pack reborn mid-delivery must be picked up by later same-kind
	// events rather than reusing a stale snapshot.
	var companionPackInstallID, companionFlowVersionID string
	if d.resolveCompanion != nil {
		companionIncarnation, isCrossPack, companionErr := d.resolveCompanion(ctx, namespace, flowDeveloperName)
		if companionErr != nil {
			// FlowDefinition missing, companion not active, or blank PackInstallId.
			// All use the same dual internal-unavailable contract as ErrRouteNotDeclared.
			log.Printf("connector_webhooks: companion resolution failed for namespace=%q eventType=%q flowName=%q: %v", namespace, ev.EventType, flowDeveloperName, companionErr)
			timing.mark("route_validate", "error")
			return validatedConnectorEvent{}, "companion_resolve_failed", false
		}
		if isCrossPack {
			// Cross-pack: check consent and version constraint.
			if consentErr := d.checkConsent(ctx, companionIncarnation.CompanionPackName, companionIncarnation.CompanionPackInstallID, namespace, packVersion); consentErr != nil {
				log.Printf("connector_webhooks: companion consent check failed for namespace=%q eventType=%q companion=%q: %v", namespace, ev.EventType, companionIncarnation.CompanionPackName, consentErr)
				timing.mark("route_validate", "error")
				return validatedConnectorEvent{}, "companion_consent_failed", false
			}
			companionPackInstallID = companionIncarnation.CompanionPackInstallID
			companionFlowVersionID = companionIncarnation.CompanionFlowVersionID
		}
		// Same-pack: companionPackInstallID/companionFlowVersionID stay empty — unchanged path.
	}

	// Validate parser populated required typed fields. An empty ObjectKind or
	// SourceID is a parser-bug class violation (server-side fault) and therefore
	// internal-unavailable. Default endpoints retry; allowlisted disposable
	// endpoints acknowledge and drop.
	if strings.TrimSpace(ev.ObjectKind) == "" || strings.TrimSpace(ev.SourceID) == "" {
		log.Printf("connector_webhooks: parser for namespace=%q produced event with empty typed fields (evType=%q objKind=%q srcID=%q); parser contract violation; internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop", namespace, ev.EventType, ev.ObjectKind, ev.SourceID)
		timing.mark("route_validate", "error")
		return validatedConnectorEvent{}, "parser_empty_typed_fields", false
	}
	if (strings.TrimSpace(ev.SourceLaneObjectKind) == "") != (strings.TrimSpace(ev.SourceLaneID) == "") {
		log.Printf("connector_webhooks: parser for namespace=%q produced event with partial source lane override (evType=%q laneKind=%q laneID=%q); parser contract violation; internal unavailable; default endpoints retry, allowlisted disposable endpoints ack-drop", namespace, ev.EventType, ev.SourceLaneObjectKind, ev.SourceLaneID)
		timing.mark("route_validate", "error")
		return validatedConnectorEvent{}, "parser_partial_source_lane", false
	}

	// Validate that the parser-emitted payload is in the shape the worker
	// accepts (JSON object or null). Anything else — invalid JSON, scalar, array
	// — cannot be unmarshaled by buildFlowInputs and would produce an
	// accepted-then-terminal queue row (provider already received 202, will not
	// retry, delivery is lost). Use internal-unavailable: default endpoints retry;
	// allowlisted disposable endpoints acknowledge and drop.
	if validateErr := validateParserPayload(ev.Payload); validateErr != nil {
		log.Printf("connector_webhooks: parser produced unprocessable payload (ns=%q evType=%q): %v",
			namespace, ev.EventType, validateErr)
		timing.mark("route_validate", "error")
		return validatedConnectorEvent{}, "parser_payload_invalid", false
	}

	// Encode the typed metadata + user payload into the cross-process queue
	// envelope. The worker decodes this envelope to recover ObjectKind, SourceID,
	// and the original user payload bytes (no reserved-key injection).
	envBytes, encErr := connector_inbound.EncodeQueueEnvelope(ev.ObjectKind, ev.SourceID, json.RawMessage(ev.Payload))
	if encErr != nil {
		log.Printf("connector_webhooks: failed to encode queue envelope for namespace=%q evType=%q: %v", namespace, ev.EventType, encErr)
		timing.mark("route_validate", "error")
		return validatedConnectorEvent{}, "queue_envelope_encode_failed", false
	}

	return validatedConnectorEvent{
		ev:                     ev,
		routeVersion:           routeVersion,
		envBytes:               envBytes,
		scopedKey:              scopeIdempotencyKey(secretName, ev.IdempotencyKey),
		companionPackInstallID: companionPackInstallID,
		companionFlowVersionID: companionFlowVersionID,
	}, "", true
}

func (d *Dispatcher) checkAdmission(ctx context.Context, c *gin.Context, namespace, secretName, packInstallID, channel string, rawBody []byte) bool {
	if d.admissionGate == nil {
		return true
	}
	decision, err := d.admissionGate.Decide(ctx, namespace, packInstallID, channel)
	if err != nil {
		var schemaUnavailable *connector_admission.SchemaUnavailableError
		if errors.As(err, &schemaUnavailable) {
			log.Printf("connector_webhooks: admission schema unavailable for namespace=%q channel=%q capability=%q: %v", namespace, channel, connector_admission.CapabilityUnavailable, err)
			d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
				reason:       "admission_schema_unavailable",
				err:          err,
				verification: webhookIngressVerified,
				retryAfter:   120 * time.Second,
			})
			return false
		}
		log.Printf("connector_webhooks: admission gate failed for namespace=%q channel=%q: %v", namespace, channel, err)
		d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
			reason:       "admission_gate_failed",
			err:          err,
			verification: webhookIngressVerified,
		})
		return false
	}
	if decision.Allow {
		return true
	}
	retryAfter := decision.RetryAfterSeconds
	if retryAfter <= 0 {
		retryAfter = 120
	}
	mode := strings.TrimSpace(decision.Mode)
	if mode == "" {
		mode = "paused"
	}
	log.Printf("connector_webhooks: admission paused namespace=%q channel=%q mode=%q retryAfter=%d reason=%q", namespace, channel, mode, retryAfter, decision.Reason)
	d.respondIngressUnavailable(c, namespace, secretName, rawBody, webhookIngressUnavailableResponse{
		reason:       "admission_not_allowed:" + mode,
		verification: webhookIngressVerified,
		retryAfter:   time.Duration(retryAfter) * time.Second,
	})
	return false
}

// flattenHeaders converts http.Header (map[string][]string) to map[string]string
// by taking the first value for each header name. This is the canonical form
// expected by SignatureVerifierHandler implementations.
func flattenHeaders(h http.Header) map[string]string {
	out := make(map[string]string, len(h))
	for k, vs := range h {
		if len(vs) > 0 {
			out[k] = vs[0]
		}
	}
	return out
}

func resolveWebhookRequestURI(r *http.Request) (string, error) {
	requestTarget := r.RequestURI
	if requestTarget == "" && r.URL != nil {
		requestTarget = r.URL.RequestURI()
	}
	if requestTarget == "" {
		return "", fmt.Errorf("request target is empty")
	}
	if !strings.HasPrefix(requestTarget, "/") {
		return "", fmt.Errorf("request target must be origin-form path/query, got %q", requestTarget)
	}
	return requestTarget, nil
}

func resolveWebhookRequestURL(r *http.Request) (string, error) {
	requestTarget, err := resolveWebhookRequestURI(r)
	if err != nil {
		return "", err
	}

	if envName, rawBase, ok := configuredWebhookBaseURL(); ok {
		return joinWebhookBaseURL(envName, rawBase, requestTarget)
	}

	if hasForwardingHeader(r.Header) {
		return "", fmt.Errorf("request URL requires WEBHOOK_BASE_URL or API_BASE_URL when forwarded headers are present")
	}

	host := r.Host
	if strings.TrimSpace(host) == "" {
		return "", fmt.Errorf("request host is empty")
	}

	scheme := "http"
	if r.TLS != nil {
		scheme = "https"
	}
	return scheme + "://" + host + requestTarget, nil
}

func hasForwardingHeader(h http.Header) bool {
	for name := range h {
		lowerName := strings.ToLower(name)
		if lowerName == "forwarded" || strings.HasPrefix(lowerName, "x-forwarded-") {
			return true
		}
	}
	return false
}

func configuredWebhookBaseURL() (string, string, bool) {
	for _, envName := range []string{"WEBHOOK_BASE_URL", "API_BASE_URL"} {
		if value := strings.TrimSpace(os.Getenv(envName)); value != "" {
			return envName, value, true
		}
	}
	return "", "", false
}

func joinWebhookBaseURL(envName, rawBase, requestTarget string) (string, error) {
	parsed, err := url.Parse(rawBase)
	if err != nil {
		return "", fmt.Errorf("%s is not a valid URL: %w", envName, err)
	}
	if parsed.Scheme != "http" && parsed.Scheme != "https" {
		return "", fmt.Errorf("%s must use http or https scheme", envName)
	}
	if parsed.Host == "" {
		return "", fmt.Errorf("%s must include a host", envName)
	}
	if parsed.User != nil {
		return "", fmt.Errorf("%s must not include userinfo", envName)
	}
	if parsed.RawQuery != "" || parsed.Fragment != "" {
		return "", fmt.Errorf("%s must not include query or fragment", envName)
	}

	base := strings.TrimRight(parsed.String(), "/")
	return base + requestTarget, nil
}

// scopeIdempotencyKey binds a parser's raw idempotency key to the webhook
// endpoint (the webhook-secret name taken from the request path) so the
// (InstalledPackId, IdempotencyKey) dedup is per-endpoint. A connector pack may
// declare multiple named webhook secrets, and all of a pack's endpoints resolve
// the same envelope parser by namespace; the parser only sees the body, so it
// cannot scope the key itself — the dispatcher injects the endpoint here.
//
// The composite is hashed so the stored key stays within the
// _System_ConnectorEvent.IdempotencyKey VARCHAR(255) bound regardless of how
// long the endpoint name or raw key are (the connector-events layer likewise
// keeps keys bounded — cf. the "replay:"+id scheme). The hash is deterministic,
// so a genuine redelivery of the same event to the same endpoint still maps to
// the same key and dedups; the "ep:" prefix namespaces it away from other key
// schemes (e.g. "replay:").
func scopeIdempotencyKey(secretName, rawKey string) string {
	sum := sha256.Sum256([]byte(secretName + "\x00" + rawKey))
	return "ep:" + hex.EncodeToString(sum[:])
}

const connectorWebhookIngressFailureReasonKey = "connector_webhooks.ingress_failure_reason"

// webhookIngressRequestID preserves the host request ID when the dispatcher is
// mounted in the API host, and supplies one for direct mounts and tests.
func webhookIngressRequestID(c *gin.Context) string {
	requestID := c.GetString("request_id")
	requestContext := c.Request.Context()
	if requestID == "" {
		requestID, _ = requestContext.Value(contractconstants.ContextKeyRequestID).(string)
	}
	if requestID == "" {
		requestID = c.GetHeader("X-Request-ID")
	}
	if requestID == "" {
		requestID = uuid.NewString()
	}
	c.Set("request_id", requestID)
	c.Header("X-Request-ID", requestID)
	if contextRequestID, _ := requestContext.Value(contractconstants.ContextKeyRequestID).(string); contextRequestID != requestID {
		c.Request = c.Request.WithContext(context.WithValue(requestContext, contractconstants.ContextKeyRequestID, requestID))
	}
	return requestID
}

func markConnectorWebhookIngressFailure(c *gin.Context, reason string) {
	if c == nil || strings.TrimSpace(reason) == "" {
		return
	}
	c.Set(connectorWebhookIngressFailureReasonKey, strings.TrimSpace(reason))
}

func logWebhookIngress503(c *gin.Context, requestID, namespace, packInstallID string, timing *webhookDispatchTiming) {
	stage, outcome := timing.lastStageOutcome()
	reason := c.GetString(connectorWebhookIngressFailureReasonKey)
	if reason == "" {
		reason = stage + "_" + outcome
	}
	sum := sha256.Sum256([]byte(reason))
	path := ""
	method := ""
	if c != nil && c.Request != nil {
		method = c.Request.Method
		if c.Request.URL != nil {
			path = c.Request.URL.Path
		}
	}
	classification := "ingress_503_before_queue"
	enqueueAttempted := false
	eventExpectedAbsent := true
	if stage == "enqueue" {
		classification = "ingress_503_during_enqueue"
		enqueueAttempted = true
		eventExpectedAbsent = false
	}
	log.Printf("connector_webhooks: %s request_id=%q method=%q path=%q namespace=%q installed_pack_id=%q stage=%q outcome=%q reason=%q error_fingerprint=%q enqueue_attempted=%t expected_event_absent=%t",
		classification,
		requestID,
		method,
		path,
		namespace,
		packInstallID,
		stage,
		outcome,
		reason,
		hex.EncodeToString(sum[:8]),
		enqueueAttempted,
		eventExpectedAbsent,
	)
}

// validateParserPayload ensures parser-emitted Payload bytes are in the shape
// the worker's buildFlowInputs accepts: a JSON object (typed record fields) or
// a JSON null (explicit "no payload" e.g. delete events). Anything else —
// invalid JSON, scalar string/number/bool, JSON array — cannot be unmarshaled
// into map[string]any by the worker and would be durably queued but
// permanently terminal-failed. Use internal-unavailable instead of 202:
// default endpoints retry, while allowlisted disposable endpoints acknowledge and drop.
func validateParserPayload(p []byte) error {
	if len(p) == 0 {
		return fmt.Errorf("parser payload is empty (nil []byte); must be JSON object or null")
	}
	var v interface{}
	if err := json.Unmarshal(p, &v); err != nil {
		return fmt.Errorf("parser payload is not valid JSON: %w", err)
	}
	switch v.(type) {
	case nil: // JSON null — explicitly allowed (delete events, tombstones, etc.)
		return nil
	case map[string]interface{}: // JSON object — primary worker-contract shape
		return nil
	default:
		return fmt.Errorf("parser payload must be JSON object or null; got %T", v)
	}
}