package connector_webhooks

import (
	"bytes"
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"net/http/httptest"
	"strconv"
	"strings"
	"testing"
	"time"

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

	connector_admission "github.com/pingcap-inc/nexuscrm-connectors/features/connector/admission"
	connectorcapabilities "github.com/pingcap-inc/nexuscrm-connectors/features/connector/capabilities"
	connector_events "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	connector_inbound "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inbound"
	connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	connector_packstate "github.com/pingcap-inc/nexuscrm-connectors/features/connector/packstate"
	"github.com/pingcap-inc/nexuscrm-connectors/features/webhooks"
)

func init() {
	gin.SetMode(gin.TestMode)
}

func TestBuildConnectorEventBatchCarriesSourceLaneOverride(t *testing.T) {
	batch, deliveryID := buildConnectorEventBatch([]validatedConnectorEvent{
		{
			ev: connectorcapabilities.ConnectorEvent{
				EventType:            "workflow.action.execute",
				ObjectKind:           "workflow-actions",
				SourceID:             "contact-123",
				SourceLaneObjectKind: "shared-parent",
				SourceLaneID:         "parent-001",
				IdempotencyKey:       "callback-1",
			},
			envBytes:     []byte(`{"objectKind":"workflow-actions","sourceID":"contact-123","payload":{"parent_id":"parent-001"}}`),
			scopedKey:    "scoped-callback-1",
			routeVersion: 1,
		},
	})
	if deliveryID == "" {
		t.Fatal("deliveryID is empty")
	}
	if len(batch) != 1 {
		t.Fatalf("len(batch) = %d, want 1", len(batch))
	}
	if batch[0].SourceLaneObjectKind != "shared-parent" || batch[0].SourceLaneID != "parent-001" {
		t.Fatalf("source lane = %q/%q, want shared-parent/parent-001", batch[0].SourceLaneObjectKind, batch[0].SourceLaneID)
	}
	if string(batch[0].Payload) == "" {
		t.Fatal("payload is empty")
	}
}

// ── fakes ────────────────────────────────────────────────────────────────────

// stubRouteResolverAlways returns a fixed (version, flow) for any namespace/packInstallID/eventKind.
// Use version=1 for tests that exercise the happy path; use 0 for "no route" tests.
type stubRouteResolverAlways struct {
	version int64
	flow    string
	err     error
}

func (r *stubRouteResolverAlways) ResolveFlow(_ context.Context, _, _, _ string) (int64, string, error) {
	return r.version, r.flow, r.err
}
func (r *stubRouteResolverAlways) ResolveFlowByVersion(_ context.Context, _, _ string, _ int64) (string, error) {
	return r.flow, r.err
}
func (r *stubRouteResolverAlways) ResolveExecutionSpecByVersion(_ context.Context, _, _ string, _ int64) (connector_inbound_routing.RouteExecutionSpec, error) {
	if r.err != nil {
		return connector_inbound_routing.RouteExecutionSpec{}, r.err
	}
	return connector_inbound_routing.NormalizeRouteExecutionSpec(connector_inbound_routing.RouteExecutionSpec{FlowDeveloperName: r.flow})
}

// noopRoutes is a convenience instance — returns version=1 and no error so tests
// that don't care about routing still get a valid RouteVersion on queue rows.
var noopRoutes connector_inbound_routing.RouteResolver = &stubRouteResolverAlways{version: 1, flow: "SomeHandler"}

type fakeVerifier struct {
	calls           int
	err             error
	got             *connectorcapabilities.VerifierInput
	requestID       string
	deliveryID      string
	headerAllowlist []string
}

func (v *fakeVerifier) Verify(ctx context.Context, input *connectorcapabilities.VerifierInput) error {
	v.calls++
	v.requestID, _ = ctx.Value(contractconstants.ContextKeyRequestID).(string)
	if v.deliveryID != "" {
		input.DeliveryIdentity = v.deliveryID
	}
	if len(v.headerAllowlist) > 0 {
		input.ReceiptHeaderAllowlist = append([]string(nil), v.headerAllowlist...)
	}
	v.got = input
	return v.err
}

type sequenceSecretVerifier struct {
	calls           int
	deliveryID      string
	expectedSecrets []string
}

func (v *sequenceSecretVerifier) Verify(_ context.Context, input *connectorcapabilities.VerifierInput) error {
	if v.deliveryID != "" {
		input.DeliveryIdentity = v.deliveryID
	}
	expected := ""
	if v.calls < len(v.expectedSecrets) {
		expected = v.expectedSecrets[v.calls]
	}
	v.calls++
	if expected != "" && !bytes.Equal(input.Secret, []byte(expected)) {
		return errBadSig
	}
	return nil
}

type fakeParser struct {
	calls    int
	events   []connectorcapabilities.ConnectorEvent
	err      error
	gotInput *connectorcapabilities.ParserInput
}

type fakeAdmissionGate struct {
	decision     connector_admission.Decision
	err          error
	calls        int
	gotNamespace string
	gotPackID    string
	gotChannel   string
}

func (g *fakeAdmissionGate) Decide(_ context.Context, namespace, packInstallID, channel string) (connector_admission.Decision, error) {
	g.calls++
	g.gotNamespace = namespace
	g.gotPackID = packInstallID
	g.gotChannel = channel
	if g.err != nil {
		return connector_admission.Decision{}, g.err
	}
	return g.decision, nil
}

func (p *fakeParser) Parse(_ context.Context, input *connectorcapabilities.ParserInput) ([]connectorcapabilities.ConnectorEvent, error) {
	p.calls++
	p.gotInput = input
	return p.events, p.err
}

// ── helpers ───────────────────────────────────────────────────────────────────

const (
	testNamespace      = "__test_noop__"
	testOtherNamespace = "__test_other__"
	testSecretName     = "default"
	testPackInstallID  = "install-test"
)

// errBadSig / errParseFail wrap the capability-contract sentinels (as a real
// handler would, via fmt.Errorf("...: %w", ...)) so the dispatcher maps them to
// the terminal client codes 401 / 400. A non-sentinel handler error is a server
// fault → retriable 503 (see TestDispatch_VerifierServerFault_503 /
// TestDispatch_ParserServerFault_503).
var errBadSig = fmt.Errorf("verify acme: %w", connectorcapabilities.ErrInvalidSignature)
var errParseFail = fmt.Errorf("parse acme: %w", connectorcapabilities.ErrMalformedPayload)

// stubSecret returns a fixed resolved secret for any namespace/name/packInstallID.
func stubSecret(alg string, secret []byte) func(context.Context, string, string, string) (*webhooks.ResolvedWebhookSecret, error) {
	return func(_ context.Context, _, _, _ string) (*webhooks.ResolvedWebhookSecret, error) {
		return &webhooks.ResolvedWebhookSecret{Algorithm: alg, Secret: secret, SecretValueHash: "secret-value-hash"}, nil
	}
}

// stubPackActive returns an active-looking pack row with a fixed install ID for
// any namespace.  The InstallID ("install-test") is used by Dispatch as the
// per-incarnation identity at both route resolution and enqueue — tests that do
// not care about which specific InstallID flows through can rely on this default.
func stubPackActive() func(context.Context, string) (*connector_packstate.ConnectorPackRow, bool, error) {
	return stubPackActiveWithInstallID(testPackInstallID)
}

// stubPackActiveWithInstallID is like stubPackActive but lets callers pin a
// specific InstallID.  Use in tests that need to assert which packInstallID
// propagates to ResolveFlow / Enqueue (e.g. the Step-2/8 single-resolve invariant).
func stubPackActiveWithInstallID(installID string) func(context.Context, string) (*connector_packstate.ConnectorPackRow, bool, error) {
	return func(_ context.Context, ns string) (*connector_packstate.ConnectorPackRow, bool, error) {
		return &connector_packstate.ConnectorPackRow{Namespace: ns, InstallID: installID}, true, nil
	}
}

// stubPackMissing returns (nil, false, nil) always — simulates missing/inactive pack.
func stubPackMissing() func(context.Context, string) (*connector_packstate.ConnectorPackRow, bool, error) {
	return func(_ context.Context, _ string) (*connector_packstate.ConnectorPackRow, bool, error) {
		return nil, false, nil
	}
}

// stubSecretMissing returns ErrWebhookSecretNotAvailable for any lookup.
func stubSecretMissing() func(context.Context, string, string, string) (*webhooks.ResolvedWebhookSecret, error) {
	return func(_ context.Context, _, _, _ string) (*webhooks.ResolvedWebhookSecret, error) {
		return nil, webhooks.ErrWebhookSecretNotAvailable
	}
}

// buildRegistry constructs a registry with a stubGate that always returns the
// provided boolean, and registers verifier + parser for the given namespace.
func buildRegistry(ns string, allow bool, verifier connectorcapabilities.SignatureVerifierHandler, parser connectorcapabilities.EnvelopeParserHandler) *connectorcapabilities.Registry {
	return buildRegistryForPackInstallID(ns, testPackInstallID, allow, verifier, parser)
}

func buildRegistryForPackInstallID(ns string, packInstallID string, allow bool, verifier connectorcapabilities.SignatureVerifierHandler, parser connectorcapabilities.EnvelopeParserHandler) *connectorcapabilities.Registry {
	gate := func(_ context.Context, _ string, _ connectorcapabilities.CapabilityKind, _ string) (bool, error) {
		return allow, nil
	}
	reg := connectorcapabilities.NewRegistry(gate)
	if verifier != nil {
		_ = reg.UpsertCapabilityForPackInstall(ns, connectorcapabilities.SignatureVerifier, packInstallID, verifier)
	}
	if parser != nil {
		_ = reg.UpsertCapabilityForPackInstall(ns, connectorcapabilities.EnvelopeParser, packInstallID, parser)
	}
	return reg
}

// postRequest sends a POST to /webhooks/:namespace/:path and returns the recorder.
func postRequest(d *Dispatcher, namespace, secretName, body string) *httptest.ResponseRecorder {
	return postRequestTarget(d, "/webhooks/"+namespace+"/"+secretName, body, nil)
}

func postRequestTarget(d *Dispatcher, target, body string, headers http.Header) *httptest.ResponseRecorder {
	engine := gin.New()
	grp := engine.Group("/webhooks")
	d.RegisterRoutes(grp)

	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, target, strings.NewReader(body))
	if req.URL.IsAbs() {
		req.RequestURI = req.URL.RequestURI()
	}
	if req.Header.Get("Content-Type") == "" {
		req.Header.Set("Content-Type", "application/json")
	}
	for k, values := range headers {
		for _, value := range values {
			req.Header.Add(k, value)
		}
	}
	engine.ServeHTTP(w, req)
	return w
}

func captureStandardLogs(t *testing.T) *bytes.Buffer {
	t.Helper()
	var logs bytes.Buffer
	oldWriter := log.Writer()
	oldFlags := log.Flags()
	oldPrefix := log.Prefix()
	log.SetOutput(&logs)
	log.SetFlags(0)
	log.SetPrefix("")
	t.Cleanup(func() {
		log.SetOutput(oldWriter)
		log.SetFlags(oldFlags)
		log.SetPrefix(oldPrefix)
	})
	return &logs
}

func newTemplateVerifier(t *testing.T) connectorcapabilities.SignatureVerifierHandler {
	return newTemplateVerifierWithTemplate(t, "{Method}\n{RequestURI}\n{RawBody}\n{Timestamp}")
}

func newFullURLTemplateVerifier(t *testing.T) connectorcapabilities.SignatureVerifierHandler {
	return newTemplateVerifierWithTemplate(t, "{Method}\n{RequestURL}\n{RawBody}\n{Timestamp}")
}

func newTemplateVerifierWithTemplate(t *testing.T, signingTemplate string) connectorcapabilities.SignatureVerifierHandler {
	t.Helper()
	verifier, err := connector_inbound.NewSignatureVerifier(connector_inbound.SignatureSchemeConfig{
		Scheme:            connector_inbound.SchemeHMACSHA256Template,
		SigningTemplate:   signingTemplate,
		TimestampHeader:   "X-Test-Timestamp",
		SignatureHeader:   "X-Test-Signature",
		SignatureEncoding: connector_inbound.EncodingHex,
		SkewWindow:        5 * time.Minute,
		SecretName:        testSecretName,
	})
	if err != nil {
		t.Fatalf("NewSignatureVerifier: %v", err)
	}
	return verifier
}

func signedWebhookHeaders(body, secret []byte, method, requestURI string, ts time.Time) http.Header {
	timestamp := strconv.FormatInt(ts.UnixMilli(), 10)
	mac := hmac.New(sha256.New, secret)
	mac.Write([]byte(method + "\n" + requestURI + "\n" + string(body) + "\n" + timestamp))

	headers := http.Header{}
	headers.Set("X-Test-Timestamp", timestamp)
	headers.Set("X-Test-Signature", hex.EncodeToString(mac.Sum(nil)))
	return headers
}

func dispatcherWithVerifier(verifier connectorcapabilities.SignatureVerifierHandler, secret []byte) *Dispatcher {
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-signed", Payload: []byte(`{}`), IdempotencyKey: "signed-evt"},
		},
	}
	return &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", secret),
		registry:      buildRegistry(testNamespace, true, verifier, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "evt-id"
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}
}

// ── test cases ────────────────────────────────────────────────────────────────

// TestDispatch_ValidSigned_202_EnqueuesOnce: happy path — 202, verifier called,
// parser called, one event enqueued, verifier received Secret+Algorithm from resolveSecret.
func TestDispatch_ValidSigned_202_EnqueuesOnce(t *testing.T) {
	verifier := &fakeVerifier{}
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-test-1", Payload: []byte(`{}`), IdempotencyKey: "evt_1"},
		},
	}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("mysecret")),
		registry:      buildRegistry(testNamespace, true, verifier, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "evt-id-1"
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{"event":"contact.created"}`)

	if w.Code != http.StatusAccepted {
		t.Fatalf("status = %d, want 202", w.Code)
	}
	if verifier.calls != 1 {
		t.Fatalf("verifier.calls = %d, want 1", verifier.calls)
	}
	if parser.calls != 1 {
		t.Fatalf("parser.calls = %d, want 1", parser.calls)
	}
	if enqueueCalls != 1 {
		t.Fatalf("enqueueCalls = %d, want 1", enqueueCalls)
	}
	if verifier.got == nil || string(verifier.got.Secret) != "mysecret" {
		t.Fatalf("verifier.got.Secret = %q, want %q", verifier.got.Secret, "mysecret")
	}
	if verifier.got.Algorithm != "hmac-sha256" {
		t.Fatalf("verifier.got.Algorithm = %q, want hmac-sha256", verifier.got.Algorithm)
	}
	// The dispatcher must thread the canonical (lower-cased) secretName through
	// both VerifierInput.SecretName and ParserInput.SecretName so capability
	// handlers can perform per-endpoint logic.
	if verifier.got.SecretName != testSecretName {
		t.Fatalf("verifier.got.SecretName = %q, want %q", verifier.got.SecretName, testSecretName)
	}
	if parser.gotInput == nil || parser.gotInput.SecretName != testSecretName {
		var gotSN string
		if parser.gotInput != nil {
			gotSN = parser.gotInput.SecretName
		}
		t.Fatalf("parser.gotInput.SecretName = %q, want %q", gotSN, testSecretName)
	}
	var resp map[string]interface{}
	if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
		t.Fatalf("decode response: %v", err)
	}
	ids, ok := resp["eventIds"]
	if !ok {
		t.Fatalf("response missing eventIds: %v", resp)
	}
	if len(ids.([]interface{})) != 1 {
		t.Fatalf("eventIds len = %d, want 1", len(ids.([]interface{})))
	}
}

func TestDispatch_AdmissionGatePaused_503RetryAfter_NoEnqueue(t *testing.T) {
	verifier := &fakeVerifier{}
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-test-1", Payload: []byte(`{}`), IdempotencyKey: "evt_1"},
		},
	}
	gate := &fakeAdmissionGate{
		decision: connector_admission.Decision{
			Allow:             false,
			Mode:              connector_admission.ModeQuiescing,
			RetryAfterSeconds: 90,
			Reason:            "deploy",
		},
	}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("mysecret")),
		registry:      buildRegistry(testNamespace, true, verifier, parser),
		admissionGate: gate,
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			return []string{"evt-id-1"}, []bool{false}, nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{"event":"contact.created"}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503", w.Code)
	}
	if got := w.Header().Get("Retry-After"); got != "90" {
		t.Fatalf("Retry-After = %q, want 90", got)
	}
	if verifier.calls != 1 {
		t.Fatalf("verifier.calls = %d, want 1", verifier.calls)
	}
	if parser.calls != 0 {
		t.Fatalf("parser.calls = %d, want 0; gate must stop before parser/enqueue work", parser.calls)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
	if gate.calls != 1 {
		t.Fatalf("gate.calls = %d, want 1", gate.calls)
	}
	if gate.gotNamespace != testNamespace || gate.gotPackID != testPackInstallID || gate.gotChannel != "webhook:"+testSecretName {
		t.Fatalf("gate inputs = (%q, %q, %q), want (%q, %q, %q)",
			gate.gotNamespace, gate.gotPackID, gate.gotChannel,
			testNamespace, testPackInstallID, "webhook:"+testSecretName)
	}
}

func TestDispatch_AdmissionSchemaUnavailable_503_NoEnqueue(t *testing.T) {
	parser := &fakeParser{events: []connectorcapabilities.ConnectorEvent{{
		EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "schema-unavailable", Payload: []byte(`{}`), IdempotencyKey: "schema-unavailable",
	}}}
	gate := &fakeAdmissionGate{err: &connector_admission.SchemaUnavailableError{Namespace: testNamespace}}
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("mysecret")),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, parser),
		admissionGate: gate,
		enqueueBatch: func(context.Context, string, string, string, []connector_events.BatchEvent) ([]string, []bool, error) {
			t.Fatal("enqueueBatch must not run while admission schema is unavailable")
			return nil, nil, nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{"event":"contact.created"}`)
	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503", w.Code)
	}
	if got := w.Header().Get("Retry-After"); got != "120" {
		t.Fatalf("Retry-After = %q, want 120", got)
	}
}

func TestDispatch_TransactionFenceQuiescedAfterFastAdmission_503RetryAfter(t *testing.T) {
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-fence-race", Payload: []byte(`{}`), IdempotencyKey: "evt-fence-race"},
		},
	}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("mysecret")),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			return nil, nil, &connector_admission.QuiescedError{Fence: connector_admission.Fence{
				Namespace:         testNamespace,
				Mode:              connector_admission.ModeQuiescing,
				RetryAfterSeconds: 75,
				Epoch:             12,
			}}
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}
	// A transaction fence rejects materialization after signature verification;
	// even an explicitly allowlisted disposable endpoint must receive a retryable
	// response because acknowledging here would lose the delivery permanently.
	d.ingressUnavailablePolicy = ackDropPolicyFor(testNamespace, testSecretName)

	w := postRequest(d, testNamespace, testSecretName, `{"event":"contact.created"}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503", w.Code)
	}
	if got := w.Header().Get("Retry-After"); got != "75" {
		t.Fatalf("Retry-After = %q, want 75", got)
	}
	if enqueueCalls != 1 {
		t.Fatalf("enqueueCalls = %d, want 1 after fast admission", enqueueCalls)
	}
}

func TestDispatch_BadSignatureSkipsAdmissionGate(t *testing.T) {
	gate := &fakeAdmissionGate{decision: connector_admission.Allow()}
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("mysecret")),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{err: errBadSig}, &fakeParser{}),
		admissionGate: gate,
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			t.Fatal("enqueueBatch must not be called for bad signature")
			return nil, nil, nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{"event":"contact.created"}`)

	if w.Code != http.StatusUnauthorized {
		t.Fatalf("status = %d, want 401", w.Code)
	}
	if gate.calls != 0 {
		t.Fatalf("gate.calls = %d, want 0; admission gate must run only after signature verification", gate.calls)
	}
}

func TestDispatch_HubSpotGenericContactMergeRoutesToContactMergeFlow(t *testing.T) {
	parser, err := connector_inbound.NewEnvelopeParser("hubspot-crm-connector", connector_inbound.EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []connector_inbound.EnvelopeMappingRule{
			{
				MatchAll: []connector_inbound.MatchCondition{
					{Path: ".subscriptionType", Equals: "object.merge"},
					{Path: ".objectTypeId", Equals: "0-1"},
				},
				EventTypeLiteral:   "contact.merge",
				NamespaceFromCtx:   true,
				ObjectKind:         "contacts",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
				PayloadProjection: []connector_inbound.PayloadField{
					{Name: "objectId", Path: ".objectId", Mode: connector_inbound.PayloadFieldModeScalarString},
					{Name: "primaryObjectId", Path: ".primaryObjectId", Mode: connector_inbound.PayloadFieldModeScalarString},
					{Name: "mergedObjectIds", Path: ".mergedObjectIds", Mode: connector_inbound.PayloadFieldModeScalarStringArray},
					{Name: "newObjectId", Path: ".newObjectId", Mode: connector_inbound.PayloadFieldModeScalarString},
					{Name: "numberOfPropertiesMoved", Path: ".numberOfPropertiesMoved"},
				},
			},
			{
				MatchAll: []connector_inbound.MatchCondition{
					{Path: ".subscriptionType", Equals: "object.propertyChange"},
					{Path: ".objectTypeId", Equals: "0-1"},
				},
				EventTypeLiteral:   "contact.propertychange",
				NamespaceFromCtx:   true,
				ObjectKind:         "contacts",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
		},
	})
	if err != nil {
		t.Fatalf("NewEnvelopeParser: %v", err)
	}

	routes := &capturingRouteResolver{
		version: 9,
		flow:    "HubSpot_Contact_Merge_Reconcile",
	}
	var enqueued struct {
		namespace     string
		packInstallID string
		events        []connector_events.BatchEvent
	}
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("mysecret")),
		registry:      buildRegistry("hubspot-crm-connector", true, &fakeVerifier{}, parser),
		enqueueBatch: func(_ context.Context, namespace, packID, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueued.namespace = namespace
			enqueued.packInstallID = packID
			enqueued.events = append([]connector_events.BatchEvent(nil), events...)
			ids := make([]string, len(events))
			for i := range ids {
				ids[i] = "evt-merge"
			}
			return ids, make([]bool, len(events)), nil
		},
		routes:       routes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	body := `[{
		"subscriptionType":"object.merge",
		"objectTypeId":"0-1",
		"eventId":"merge-event-1",
		"objectId":228295732538,
		"primaryObjectId":228298315601,
		"mergedObjectIds":[228298327066],
		"newObjectId":228295732538,
		"numberOfPropertiesMoved":7
	}]`
	w := postRequest(d, "hubspot-crm-connector", testSecretName, body)

	if w.Code != http.StatusAccepted {
		t.Fatalf("status = %d body=%s, want 202", w.Code, w.Body.String())
	}
	if routes.gotNamespace != "hubspot-crm-connector" || routes.gotEventKind != "contact.merge" {
		t.Fatalf("route lookup = (%q,%q), want (hubspot-crm-connector,contact.merge)", routes.gotNamespace, routes.gotEventKind)
	}
	if enqueued.namespace != "hubspot-crm-connector" || enqueued.packInstallID != testPackInstallID {
		t.Fatalf("enqueue target = (%q,%q), want (hubspot-crm-connector,%q)", enqueued.namespace, enqueued.packInstallID, testPackInstallID)
	}
	if len(enqueued.events) != 1 {
		t.Fatalf("enqueued events = %d, want 1", len(enqueued.events))
	}
	gotEvent := enqueued.events[0]
	if gotEvent.EventKind != "contact.merge" {
		t.Fatalf("enqueued eventKind = %q, want contact.merge", gotEvent.EventKind)
	}
	if gotEvent.RouteVersion != 9 {
		t.Fatalf("routeVersion = %d, want 9", gotEvent.RouteVersion)
	}
	if !strings.HasPrefix(gotEvent.IdempotencyKey, "ep:") {
		t.Fatalf("idempotencyKey = %q, want scoped ep: key", gotEvent.IdempotencyKey)
	}

	env, err := connector_inbound.DecodeQueueEnvelope(gotEvent.Payload)
	if err != nil {
		t.Fatalf("DecodeQueueEnvelope: %v", err)
	}
	if env.ObjectKind != "contacts" || env.SourceID != "228295732538" {
		t.Fatalf("queue envelope typed fields = (%q,%q)", env.ObjectKind, env.SourceID)
	}
	var payload map[string]any
	if err := json.Unmarshal(env.Payload, &payload); err != nil {
		t.Fatalf("payload unmarshal: %v", err)
	}
	if payload["primaryObjectId"] != "228298315601" || payload["newObjectId"] != "228295732538" {
		t.Fatalf("payload canonical IDs = %#v", payload)
	}
	merged, ok := payload["mergedObjectIds"].([]any)
	if !ok || len(merged) != 1 || merged[0] != "228298327066" {
		t.Fatalf("mergedObjectIds = %#v", payload["mergedObjectIds"])
	}
}

// TestDispatch_OversizedBody_413_NoEnqueue: body > maxBodyBytes → 413, no enqueue.
func TestDispatch_OversizedBody_413_NoEnqueue(t *testing.T) {
	verifier := &fakeVerifier{}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, verifier, &fakeParser{}),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: 10, // tiny limit
	}

	oversized := bytes.Repeat([]byte("x"), 100)
	engine := gin.New()
	grp := engine.Group("/webhooks")
	d.RegisterRoutes(grp)

	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, "/webhooks/"+testNamespace+"/"+testSecretName, bytes.NewReader(oversized))
	engine.ServeHTTP(w, req)

	if w.Code != http.StatusRequestEntityTooLarge {
		t.Fatalf("status = %d, want 413", w.Code)
	}
	if verifier.calls != 0 {
		t.Fatalf("verifier.calls = %d, want 0 (body must be bounded before verification)", verifier.calls)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

// TestDispatch_UnknownOrInactivePack_404: resolvePack returns (nil,false,nil) → 404.
func TestDispatch_UnknownOrInactivePack_404(t *testing.T) {
	verifier := &fakeVerifier{}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackMissing(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, verifier, &fakeParser{}),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusNotFound {
		t.Fatalf("status = %d, want 404", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
	if verifier.calls != 0 {
		t.Fatalf("verifier.calls = %d, want 0 (verifier must not be called for inactive pack)", verifier.calls)
	}
}

// TestDispatch_MissingSecret_401_NoEnqueue: resolveSecret returns ErrWebhookSecretNotAvailable → 401.
func TestDispatch_MissingSecret_401_NoEnqueue(t *testing.T) {
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecretMissing(),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, &fakeParser{}),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusUnauthorized {
		t.Fatalf("status = %d, want 401", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

// TestDispatch_BadSignature_401_NoEnqueue: verifier returns an error → 401.
func TestDispatch_BadSignature_401_NoEnqueue(t *testing.T) {
	verifier := &fakeVerifier{err: errBadSig}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, verifier, &fakeParser{}),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusUnauthorized {
		t.Fatalf("status = %d, want 401", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

// TestDispatch_VerifierServerFault_503: a NON-sentinel verifier error is a
// server-side fault → retriable 503 (not a terminal 401), so the provider keeps
// retrying a legitimate delivery instead of treating a handler/dependency fault
// as bad credentials.
func TestDispatch_VerifierServerFault_503(t *testing.T) {
	verifier := &fakeVerifier{err: errors.New("verifier dependency timeout")} // non-sentinel
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, verifier, &fakeParser{}),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (non-sentinel verifier error is a retriable server fault)", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

func TestDispatch_503LogsRequestCorrelation(t *testing.T) {
	verifier := &fakeVerifier{err: errors.New("verifier dependency timeout")}
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, verifier, &fakeParser{}),
		enqueueBatch: func(context.Context, string, string, string, []connector_events.BatchEvent) ([]string, []bool, error) {
			return nil, nil, errors.New("enqueue must not run")
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	logs := captureStandardLogs(t)

	requestID := "hubspot-ingress-503-test"
	w := postRequestTarget(d, "/webhooks/"+testNamespace+"/"+testSecretName, `{}`, http.Header{"X-Request-ID": []string{requestID}})

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503", w.Code)
	}
	if got := w.Header().Get("X-Request-ID"); got != requestID {
		t.Fatalf("X-Request-ID = %q, want %q", got, requestID)
	}
	if verifier.requestID != requestID {
		t.Fatalf("verifier context request ID = %q, want %q", verifier.requestID, requestID)
	}
	for _, want := range []string{
		"ingress_503_before_queue",
		`request_id="hubspot-ingress-503-test"`,
		`path="/webhooks/__test_noop__/default"`,
		`installed_pack_id="install-test"`,
		`stage="verify"`,
		`reason="signature_verify_failed"`,
		"error_fingerprint=",
		"enqueue_attempted=false",
		"expected_event_absent=true",
	} {
		if !strings.Contains(logs.String(), want) {
			t.Fatalf("logs missing %q: %s", want, logs.String())
		}
	}
	if got := strings.Count(logs.String(), "ingress_503_"); got != 1 {
		t.Fatalf("ingress 503 correlation log count = %d, want 1: %s", got, logs.String())
	}
}

func TestDispatch_503LogsPreciseRouteReason(t *testing.T) {
	parser := &fakeParser{events: []connectorcapabilities.ConnectorEvent{{
		EventType:      "contact.created",
		ObjectKind:     "contact",
		SourceID:       "source-1",
		Payload:        []byte(`{}`),
		IdempotencyKey: "event-1",
	}}}
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, parser),
		enqueueBatch: func(context.Context, string, string, string, []connector_events.BatchEvent) ([]string, []bool, error) {
			return nil, nil, errors.New("enqueue must not run")
		},
		routes:       &stubRouteResolverAlways{err: connector_inbound_routing.ErrRouteNotDeclared},
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}
	logs := captureStandardLogs(t)

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503", w.Code)
	}
	for _, want := range []string{
		"ingress_503_before_queue",
		`stage="route_validate"`,
		`reason="route_not_declared"`,
		"enqueue_attempted=false",
		"expected_event_absent=true",
	} {
		if !strings.Contains(logs.String(), want) {
			t.Fatalf("logs missing %q: %s", want, logs.String())
		}
	}
}

// TestDispatch_ParserServerFault_503: a NON-sentinel parser error is a
// server-side fault → retriable 503 (not a terminal 400), so a parser defect or
// transient failure does not permanently drop a valid delivery.
func TestDispatch_ParserServerFault_503(t *testing.T) {
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, &fakeParser{err: errors.New("parser dependency timeout")}), // non-sentinel
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (non-sentinel parser error is a retriable server fault)", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

// TestDispatch_ReplaySameIdempotency_202_NoExtraEnqueue: enqueue returns isDuplicate=true
// on the 2nd call → both calls return 202. Dispatcher must not 500 on duplicate.
func TestDispatch_ReplaySameIdempotency_202_NoExtraEnqueue(t *testing.T) {
	verifier := &fakeVerifier{}
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "test.event", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-test", Payload: []byte(`{}`), IdempotencyKey: "idempotent-key"},
		},
	}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, verifier, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			isDup := enqueueCalls > 1
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "dedup-id"
			}
			_dups := make([]bool, len(events))
			for _bi := range events {
				_dups[_bi] = isDup
			}
			return _ids, _dups, nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	// First call
	w1 := postRequest(d, testNamespace, testSecretName, `{}`)
	if w1.Code != http.StatusAccepted {
		t.Fatalf("first call status = %d, want 202", w1.Code)
	}
	// Second call — same idempotency key, isDuplicate=true
	w2 := postRequest(d, testNamespace, testSecretName, `{}`)
	if w2.Code != http.StatusAccepted {
		t.Fatalf("second call status = %d, want 202 (duplicate must not 500)", w2.Code)
	}
	if enqueueCalls != 2 {
		t.Fatalf("enqueueCalls = %d, want 2", enqueueCalls)
	}
}

// TestDispatch_EnvelopeParseError_400: parser returns an error → 400.
func TestDispatch_EnvelopeParseError_400(t *testing.T) {
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, &fakeParser{err: errParseFail}),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusBadRequest {
		t.Fatalf("status = %d, want 400", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

// TestDispatch_CapabilityGateFalse_401: gate returns false → ResolveSignatureVerifier
// returns ErrCapabilityNotAvailable → dispatcher returns 401.
func TestDispatch_CapabilityGateFalse_401(t *testing.T) {
	enqueueCalls := 0
	// buildRegistry with allow=false means gate returns false for all capabilities.
	// We still register handlers; the gate rejection is what matters.
	reg := buildRegistry(testNamespace, false, &fakeVerifier{}, &fakeParser{})
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      reg,
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusUnauthorized {
		t.Fatalf("status = %d, want 401 (gate=false should give 401)", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

// TestDispatch_SecondConnector_SameCodePath: register __test_other__ namespace;
// POST to it returns 202 via the SAME Dispatcher. Proves no per-namespace branch.
func TestDispatch_SecondConnector_SameCodePath(t *testing.T) {
	verifierOther := &fakeVerifier{}
	parserOther := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "issue.created", Namespace: testOtherNamespace, ObjectKind: "issue", SourceID: "src-issue-1", Payload: []byte(`{}`), IdempotencyKey: "other-evt-1"},
		},
	}
	enqueueCalls := 0

	// Registry with BOTH namespaces registered via the same gate.
	gate := func(_ context.Context, _ string, _ connectorcapabilities.CapabilityKind, _ string) (bool, error) {
		return true, nil
	}
	reg := connectorcapabilities.NewRegistry(gate)
	_ = reg.UpsertCapabilityForPackInstall(testOtherNamespace, connectorcapabilities.SignatureVerifier, testPackInstallID, verifierOther)
	_ = reg.UpsertCapabilityForPackInstall(testOtherNamespace, connectorcapabilities.EnvelopeParser, testPackInstallID, parserOther)

	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("other-secret")),
		registry:      reg,
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "other-evt-id"
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testOtherNamespace, "default", `{"event":"issue.created"}`)

	if w.Code != http.StatusAccepted {
		t.Fatalf("status = %d, want 202", w.Code)
	}
	if verifierOther.calls != 1 {
		t.Fatalf("verifierOther.calls = %d, want 1", verifierOther.calls)
	}
	if parserOther.calls != 1 {
		t.Fatalf("parserOther.calls = %d, want 1", parserOther.calls)
	}
	if enqueueCalls != 1 {
		t.Fatalf("enqueueCalls = %d, want 1", enqueueCalls)
	}
}

// TestDispatch_EmptyEvents_202_NoEnqueue: parser returns zero events with no
// error after signature verification. This is the generic no-op parser contract:
// parsers may intentionally skip unmatched/no-op items without turning the
// provider delivery into a retry loop.
func TestDispatch_EmptyEvents_202_NoEnqueue(t *testing.T) {
	verifier := &fakeVerifier{}
	parser := &fakeParser{events: nil} // zero events, nil error
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, verifier, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusAccepted {
		t.Fatalf("status = %d, want 202", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0 (no events to enqueue)", enqueueCalls)
	}
	var resp map[string]interface{}
	if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
		t.Fatalf("decode response: %v", err)
	}
	ids, ok := resp["eventIds"].([]interface{})
	if !ok {
		t.Fatalf("response missing eventIds: %v", resp)
	}
	if len(ids) != 0 {
		t.Fatalf("eventIds len = %d, want 0", len(ids))
	}
}

// TestDispatch_PassesDeliverySequenceInArrayOrder pins the core ordering
// mechanism for issue #2125: the Phase B enqueue loop hands each event its array
// index (0, 1, 2, ...) as DeliverySequence, so the worker claims one delivery's
// events in FIFO order (a link event after the create events it depends on).
// Every other enqueue mock ignores this parameter, so without this assertion the
// index plumbing could regress silently.
func TestDispatch_PassesDeliverySequenceInArrayOrder(t *testing.T) {
	verifier := &fakeVerifier{}
	parser := &fakeParser{events: []connectorcapabilities.ConnectorEvent{
		{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-seq-0", Payload: []byte(`{}`), IdempotencyKey: "seq-evt-0"},
		{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-seq-1", Payload: []byte(`{}`), IdempotencyKey: "seq-evt-1"},
		{EventType: "contact.link", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-seq-2", Payload: []byte(`{}`), IdempotencyKey: "seq-evt-2"},
	}}
	var gotSequences []int
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, verifier, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			for _, e := range events {
				gotSequences = append(gotSequences, e.DeliverySequence)
			}
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "evt-ok"
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusAccepted {
		t.Fatalf("status = %d, want 202", w.Code)
	}
	want := []int{0, 1, 2}
	if len(gotSequences) != len(want) {
		t.Fatalf("enqueue called %d times (sequences=%v), want %d calls", len(gotSequences), gotSequences, len(want))
	}
	for i, got := range gotSequences {
		if got != want[i] {
			t.Fatalf("enqueue call %d: DeliverySequence=%d, want %d (events must enqueue in array order)", i, got, want[i])
		}
	}
}

// TestDispatch_ParserConfigDrift_503_NoEnqueue: signed deliveries are retried
// only when the parser explicitly reports metadata/config drift. The dispatcher
// must not infer drift from ordinary zero-event no-op parses.
func TestDispatch_ParserConfigDrift_503_NoEnqueue(t *testing.T) {
	verifier := &fakeVerifier{}
	parser := &fakeParser{err: fmt.Errorf("no rule matched required delivery shape: %w", connectorcapabilities.ErrParserConfigDrift)}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, verifier, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
	var resp map[string]interface{}
	if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
		t.Fatalf("decode response: %v", err)
	}
	if _, ok := resp["eventIds"]; ok {
		t.Fatalf("response must not acknowledge eventIds for parser config drift: %v", resp)
	}
	if resp["error"] != "service unavailable" {
		t.Fatalf("error = %v, want service unavailable", resp["error"])
	}
}

// TestDispatch_EmptyIdempotencyKey_503_NoEnqueue: a parser that emits an event
// with an empty IdempotencyKey is a CONFIG-DRIFT class error (likely mis-configured
// IdempotencyKeyPath); the dispatcher must return 503 (not 500) and enqueue nothing
// so the provider retries rather than treating it as terminal.
func TestDispatch_EmptyIdempotencyKey_503_NoEnqueue(t *testing.T) {
	verifier := &fakeVerifier{}
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "test.event", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-1", Payload: []byte(`{}`), IdempotencyKey: ""},
		},
	}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, verifier, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "id"
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (empty idempotency key is CONFIG-DRIFT, must be retriable not 500)", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0 (no event enqueued when batch has an empty key)", enqueueCalls)
	}
}

// TestDispatch_SlashInSecretName_RoutesToHandler: a webhook secret Name with a
// slash (e.g. "orders/default") must route to the handler via the *path
// catch-all and reach resolveSecret with the leading slash stripped and the
// inner slash preserved.
func TestDispatch_SlashInSecretName_RoutesToHandler(t *testing.T) {
	var gotName string
	verifier := &fakeVerifier{}
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "e", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-slash-1", Payload: []byte(`{}`), IdempotencyKey: "k1"},
		},
	}
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: func(_ context.Context, _, name, _ string) (*webhooks.ResolvedWebhookSecret, error) {
			gotName = name
			return &webhooks.ResolvedWebhookSecret{Algorithm: "hmac-sha256", Secret: []byte("s")}, nil
		},
		registry: buildRegistry(testNamespace, true, verifier, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "id"
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, "orders/default", `{}`)

	if w.Code != http.StatusAccepted {
		t.Fatalf("status = %d, want 202 (slash-containing secret name must route)", w.Code)
	}
	if gotName != "orders/default" {
		t.Fatalf("resolveSecret name = %q, want %q (leading slash stripped, inner slash preserved)", gotName, "orders/default")
	}
}

// TestDispatch_RepeatedKeyWithinBatch_DedupsSecond: two events sharing one
// IdempotencyKey in a single batch are both forwarded to Enqueue; the queue's
// (InstalledPackId, IdempotencyKey) dedup returns isDuplicate=true for the
// second. The dispatcher must treat the duplicate as success (202), not 500.
func TestDispatch_RepeatedKeyWithinBatch_DedupsSecond(t *testing.T) {
	verifier := &fakeVerifier{}
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "e1", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-dup-1", Payload: []byte(`{}`), IdempotencyKey: "samekey"},
			{EventType: "e2", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-dup-2", Payload: []byte(`{}`), IdempotencyKey: "samekey"},
		},
	}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, verifier, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			// Both events are forwarded in one batch; real queue dedups the second via in-tx SELECT.
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "id"
			}
			_dups := make([]bool, len(events))
			if len(events) > 1 {
				_dups[1] = true
			}
			return _ids, _dups, nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusAccepted {
		t.Fatalf("status = %d, want 202", w.Code)
	}
	if enqueueCalls != 1 {
		t.Fatalf("enqueueCalls = %d, want 1 (batch call: both events forwarded in a single EnqueueBatch call; queue dedups the second)", enqueueCalls)
	}
}

// TestDispatch_SecretLookupFailure_503: a server-side secret-resolution failure
// (cipher/DB/decrypt — ErrWebhookSecretLookupFailed) must return a retriable 503,
// not a terminal 401, so the provider retries rather than dropping the delivery.
func TestDispatch_SecretLookupFailure_503(t *testing.T) {
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: func(_ context.Context, _, _, _ string) (*webhooks.ResolvedWebhookSecret, error) {
			return nil, webhooks.ErrWebhookSecretLookupFailed
		},
		registry: buildRegistry(testNamespace, true, &fakeVerifier{}, &fakeParser{}),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (server-side secret failure must be retriable)", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

// TestDispatch_CapabilityGateInternalError_503: when the capability gate hits an
// internal error (e.g. control-plane DB failure) — distinct from a genuine
// undeclared capability — the dispatcher must return a retriable 503, not 401.
func TestDispatch_CapabilityGateInternalError_503(t *testing.T) {
	gate := func(_ context.Context, _ string, _ connectorcapabilities.CapabilityKind, _ string) (bool, error) {
		return false, errors.New("control-plane DB error")
	}
	reg := connectorcapabilities.NewRegistry(gate)
	_ = reg.UpsertCapabilityForPackInstall(testNamespace, connectorcapabilities.SignatureVerifier, testPackInstallID, &fakeVerifier{})
	_ = reg.UpsertCapabilityForPackInstall(testNamespace, connectorcapabilities.EnvelopeParser, testPackInstallID, &fakeParser{})

	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      reg,
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (capability-gate internal error must be retriable)", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

// TestDispatch_TransientPackState_503: resolvePack returns (nil,false,nil) and
// packTransient returns (true,nil) — pack is mid install/removal. Must return
// a retriable 503, not a terminal 404.
func TestDispatch_TransientPackState_503(t *testing.T) {
	verifier := &fakeVerifier{}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackMissing(),
		packTransient: func(context.Context, string) (bool, error) { return true, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, verifier, &fakeParser{}),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (transient pack state must be retriable, not terminal 404)", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
	if verifier.calls != 0 {
		t.Fatalf("verifier.calls = %d, want 0 (verifier must not be called for transient pack)", verifier.calls)
	}
}

// TestDispatch_TransientCheckError_503: resolvePack returns (nil,false,nil) and
// packTransient returns an error — DB/infra fault during transient check → 503
// (not 500). A transient DB blip must not cause a provider to stop retrying.
func TestDispatch_TransientCheckError_503(t *testing.T) {
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack: stubPackMissing(),
		packTransient: func(context.Context, string) (bool, error) {
			return false, errors.New("transient check db error")
		},
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, &fakeParser{}),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (DB/infra fault during transient check must be retriable, not 500)", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

// TestDispatch_CapabilityGateMidApply_503: resolvePack ok=true (pack resolved
// fine), but the capability gate reports the required capability missing AND the
// mid-apply seam returns true (pack present but not yet steady-applied). Must
// return a retriable 503 rather than the terminal 401 that a
// genuinely-undeclared capability would produce. Covers both gate points:
//
//   - VerifierCapability: gate returns (false,nil) → ResolveSignatureVerifier
//     returns ErrCapabilityNotAvailable (Step 4).
//   - ParserCapability: verifier gate=true so Step 4 passes and the signature is
//     already verified; parser gate=false so ResolveEnvelopeParser returns
//     ErrCapabilityNotAvailable (Step 6). The mid-apply parser miss must likewise
//     be retriable, not a terminal error.
func TestDispatch_CapabilityGateMidApply_503(t *testing.T) {
	tests := []struct {
		name     string
		registry *connectorcapabilities.Registry
	}{
		{name: "VerifierCapability", registry: buildRegistry(testNamespace, false, &fakeVerifier{}, &fakeParser{})},
		{name: "ParserCapability", registry: buildRegistryPerKind(testNamespace, true, false, &fakeVerifier{}, &fakeParser{})},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			midApplyCalls := 0
			enqueueCalls := 0
			d := &Dispatcher{
				resolvePack: stubPackActive(),
				packTransient: func(_ context.Context, _ string) (bool, error) {
					midApplyCalls++
					return true, nil // pack mid-apply: hash drift / blank install-id
				},
				resolveSecret: stubSecret("hmac-sha256", []byte("s")),
				registry:      tt.registry,
				enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
					enqueueCalls++
					_ids := make([]string, len(events))
					for _bi := range events {
						_ids[_bi] = ""
					}
					return _ids, make([]bool, len(events)), nil
				},
				routes:       noopRoutes,
				maxBodyBytes: maxInboundWebhookBodyBytes,
			}

			w := postRequest(d, testNamespace, testSecretName, `{}`)

			if w.Code != http.StatusServiceUnavailable {
				t.Fatalf("status = %d, want 503 (mid-apply capability miss must be retriable, not terminal 401)", w.Code)
			}
			if midApplyCalls != 1 {
				t.Fatalf("midApplyCalls = %d, want 1", midApplyCalls)
			}
			if enqueueCalls != 0 {
				t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
			}
		})
	}
}

// TestDispatch_CapabilityGateFalseNotMidApply_401: resolvePack ok=true, verifier returns
// ErrCapabilityNotAvailable, AND the mid-apply seam returns false (not mid-apply:
// capability genuinely not declared). Must stay terminal 401.
func TestDispatch_CapabilityGateFalseNotMidApply_401(t *testing.T) {
	reg := buildRegistry(testNamespace, false, &fakeVerifier{}, &fakeParser{})
	midApplyCalls := 0
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack: stubPackActive(),
		packTransient: func(_ context.Context, _ string) (bool, error) {
			midApplyCalls++
			return false, nil // not mid-apply: genuine capability miss
		},
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      reg,
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusUnauthorized {
		t.Fatalf("status = %d, want 401 (genuine capability miss, not mid-apply, must stay terminal)", w.Code)
	}
	if midApplyCalls != 1 {
		t.Fatalf("midApplyCalls = %d, want 1", midApplyCalls)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

// buildRegistryPerKind constructs a registry whose gate returns allowVerifier for
// SignatureVerifier capabilities and allowParser for EnvelopeParser capabilities.
// This allows tests to exercise Step 6 parser misses without triggering a Step 4
// verifier miss first.
func buildRegistryPerKind(ns string, allowVerifier, allowParser bool, verifier connectorcapabilities.SignatureVerifierHandler, parser connectorcapabilities.EnvelopeParserHandler) *connectorcapabilities.Registry {
	gate := func(_ context.Context, _ string, kind connectorcapabilities.CapabilityKind, _ string) (bool, error) {
		switch kind {
		case connectorcapabilities.SignatureVerifier:
			return allowVerifier, nil
		case connectorcapabilities.EnvelopeParser:
			return allowParser, nil
		default:
			return false, nil
		}
	}
	reg := connectorcapabilities.NewRegistry(gate)
	if verifier != nil {
		_ = reg.UpsertCapabilityForPackInstall(ns, connectorcapabilities.SignatureVerifier, testPackInstallID, verifier)
	}
	if parser != nil {
		_ = reg.UpsertCapabilityForPackInstall(ns, connectorcapabilities.EnvelopeParser, testPackInstallID, parser)
	}
	return reg
}

// TestDispatch_ParserNotDeclaredSteadyState_Returns503: resolvePack ok=true,
// ResolveEnvelopeParser returns ErrCapabilityNotAvailable (parser gate false),
// AND packTransient returns false (steady state — pack genuinely declares no
// envelope parser). Signature is already verified → pack-config gap
// (CONFIG-DRIFT). Return 503 so the provider retries and the admin can register
// the parser capability; dropping a legitimately-signed delivery would be data
// loss. NOT 401 (unlike Step 4's missing verifier) and NOT 500 (not a
// programming invariant — it's an admin-fixable config gap).
func TestDispatch_ParserNotDeclaredSteadyState_Returns503(t *testing.T) {
	// Verifier gate=true so Step 4 passes; parser gate=false so Step 6 returns
	// ErrCapabilityNotAvailable.
	reg := buildRegistryPerKind(testNamespace, true, false, &fakeVerifier{}, &fakeParser{})
	midApplyCalls := 0
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack: stubPackActive(),
		packTransient: func(_ context.Context, _ string) (bool, error) {
			midApplyCalls++
			return false, nil // not mid-apply: genuine pack-config gap
		},
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      reg,
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (steady-state missing parser is CONFIG-DRIFT, not 500 or 401)", w.Code)
	}
	if midApplyCalls != 1 {
		t.Fatalf("midApplyCalls = %d, want 1", midApplyCalls)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

// TestDispatch_SecretResolveMidApply_503: resolvePack ok=true, resolveSecret returns
// ErrWebhookSecretNotAvailable, AND packTransient returns true (mid-apply) → 503.
// Verifies that the same transient→503 logic applied at Steps 2 and 4 also covers
// Step 3 (the secret-resolve path).
func TestDispatch_SecretResolveMidApply_503(t *testing.T) {
	midApplyCalls := 0
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack: stubPackActive(),
		packTransient: func(_ context.Context, _ string) (bool, error) {
			midApplyCalls++
			return true, nil // pack mid-apply: triggers 503 instead of 401
		},
		resolveSecret: stubSecretMissing(), // ErrWebhookSecretNotAvailable
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, &fakeParser{}),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (secret miss during mid-apply must be retriable, not terminal 401)", w.Code)
	}
	if midApplyCalls != 1 {
		t.Fatalf("midApplyCalls = %d, want 1", midApplyCalls)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0", enqueueCalls)
	}
}

// TestDispatch_MixedCaseNamespace_LowercasedForRegistry: POST to
// /webhooks/MyConnector/default — namespace must be lowercased to "myconnector"
// before registry lookup. The registry is keyed under "myconnector" (lowercase);
// a missed canonicalization would produce 401 (handler not registered).
func TestDispatch_MixedCaseNamespace_LowercasedForRegistry(t *testing.T) {
	const mixedNS = "MyConnector"
	const canonNS = "myconnector"

	var gotNS string
	verifier := &fakeVerifier{}
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "e", Namespace: canonNS, ObjectKind: "contact", SourceID: "src-mixed-1", Payload: []byte(`{}`), IdempotencyKey: "k1"},
		},
	}

	// Pack + secret stubs capture the namespace they receive.
	resolvePack := func(_ context.Context, ns string) (*connector_packstate.ConnectorPackRow, bool, error) {
		gotNS = ns
		return &connector_packstate.ConnectorPackRow{Namespace: ns}, true, nil
	}
	resolveSecret := func(_ context.Context, ns, _, _ string) (*webhooks.ResolvedWebhookSecret, error) {
		return &webhooks.ResolvedWebhookSecret{Algorithm: "hmac-sha256", Secret: []byte("s")}, nil
	}
	enqueueCalls := 0

	// Registry keyed under lowercase "myconnector".
	gate := func(_ context.Context, _ string, _ connectorcapabilities.CapabilityKind, _ string) (bool, error) {
		return true, nil
	}
	reg := connectorcapabilities.NewRegistry(gate)
	_ = reg.UpsertCapabilityForPackInstall(canonNS, connectorcapabilities.SignatureVerifier, testPackInstallID, verifier)
	_ = reg.UpsertCapabilityForPackInstall(canonNS, connectorcapabilities.EnvelopeParser, testPackInstallID, parser)

	d := &Dispatcher{
		resolvePack:   resolvePack,
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: resolveSecret,
		registry:      reg,
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "id"
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, mixedNS, testSecretName, `{}`)

	if w.Code != http.StatusAccepted {
		t.Fatalf("status = %d, want 202 (mixed-case namespace must be lowercased for registry lookup)", w.Code)
	}
	if gotNS != canonNS {
		t.Fatalf("resolvePack received namespace = %q, want %q (must be canonicalized before any downstream call)", gotNS, canonNS)
	}
	if verifier.calls != 1 {
		t.Fatalf("verifier.calls = %d, want 1", verifier.calls)
	}
	if enqueueCalls != 1 {
		t.Fatalf("enqueueCalls = %d, want 1", enqueueCalls)
	}
}

// TestDispatch_IdempotencyKeyScopedByEndpoint pins endpoint-scoped idempotency.
// A connector pack may declare multiple named webhook secrets (endpoints), and
// all endpoints under a namespace share one envelope parser, so two endpoints
// that emit the same raw delivery ID must NOT collapse onto one
// (InstalledPackId, IdempotencyKey) dedup row. The same endpoint + same raw key
// must remain deterministic so genuine replays still dedup.
func TestDispatch_IdempotencyKeyScopedByEndpoint(t *testing.T) {
	const rawKey = "evt_1"
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-ep-1", Payload: []byte(`{}`), IdempotencyKey: rawKey},
		},
	}
	var enqueuedKeys []string
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			// Each dispatch has 1 event; capture the scoped idempotency key.
			for _, e := range events {
				enqueuedKeys = append(enqueuedKeys, e.IdempotencyKey)
			}
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "id"
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	// Two DIFFERENT endpoints (paths) under the same namespace, same raw key,
	// then the first endpoint again to check determinism.
	for _, ep := range []string{"endpoint-a", "endpoint-b", "endpoint-a"} {
		if w := postRequest(d, testNamespace, ep, `{}`); w.Code != http.StatusAccepted {
			t.Fatalf("endpoint %q status = %d, want 202", ep, w.Code)
		}
	}
	if len(enqueuedKeys) != 3 {
		t.Fatalf("enqueuedKeys len = %d, want 3", len(enqueuedKeys))
	}
	keyA, keyB, keyARepeat := enqueuedKeys[0], enqueuedKeys[1], enqueuedKeys[2]

	if keyA == keyB {
		t.Fatalf("distinct endpoints produced the same idempotency key %q; deliveries from different endpoints would collide", keyA)
	}
	if keyA != keyARepeat {
		t.Fatalf("same endpoint+raw key produced different keys %q vs %q; replay dedup would break", keyA, keyARepeat)
	}
	for _, k := range enqueuedKeys {
		if !strings.HasPrefix(k, "ep:") {
			t.Errorf("scoped key %q missing ep: prefix", k)
		}
		if k == rawKey {
			t.Errorf("scoped key %q equals the raw parser key; endpoint scoping not applied", k)
		}
		if len(k) > 255 {
			t.Errorf("scoped key length %d exceeds VARCHAR(255)", len(k))
		}
	}
}

// TestDispatch_MissingParser_503_NoEnqueue: ResolveEnvelopeParser returns an
// error because no handler is registered (server pack-config gap — verifier is
// registered but parser is not). Dispatcher step 6 maps this to CONFIG-DRIFT
// 503 and must not enqueue. 503 ensures the provider retries; 500 would
// eventually cause most providers to stop retrying and drop the delivery.
func TestDispatch_MissingParser_503_NoEnqueue(t *testing.T) {
	enqueueCalls := 0
	// Build a registry with a gate that always allows, a verifier registered,
	// but NO parser registered. ResolveEnvelopeParser will return an error
	// because no handler is present for the namespace.
	gate := func(_ context.Context, _ string, _ connectorcapabilities.CapabilityKind, _ string) (bool, error) {
		return true, nil
	}
	reg := connectorcapabilities.NewRegistry(gate)
	_ = reg.UpsertCapabilityForPackInstall(testNamespace, connectorcapabilities.SignatureVerifier, testPackInstallID, &fakeVerifier{})
	// Intentionally omit: reg.RegisterEnvelopeParser(testNamespace, ...)

	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      reg,
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (missing parser is CONFIG-DRIFT, must be retriable not 500)", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0 (must not enqueue when parser resolution fails)", enqueueCalls)
	}
}

// TestDispatch_EnqueueFault_503_NoExtraEnqueue: parser returns one valid event
// but the injected enqueue func returns a non-nil, non-ErrPackIncarnationChanged
// error (DB/infra write fault). Dispatcher step 8 must return 503 so the
// provider retries the delivery rather than treating it as successfully accepted.
// 503 (not 500) ensures providers keep retrying — a transient DB blip must not
// permanently drop a valid delivery. This is the highest data-loss risk site.
func TestDispatch_EnqueueFault_503_NoExtraEnqueue(t *testing.T) {
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-fault-1", Payload: []byte(`{}`), IdempotencyKey: "evt_enqueue_fault"},
		},
	}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			return nil, nil, errors.New("queue unavailable")
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}
	logs := captureStandardLogs(t)

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code == http.StatusInternalServerError {
		t.Fatalf("status = 500: enqueue DB fault must NOT return 500 (provider would stop retrying, delivery lost)")
	}
	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (enqueue DB/infra fault must be retriable)", w.Code)
	}
	if enqueueCalls != 1 {
		t.Fatalf("enqueueCalls = %d, want 1 (enqueue was attempted once before failing)", enqueueCalls)
	}
	for _, want := range []string{
		"ingress_503_during_enqueue",
		`stage="enqueue"`,
		`reason="enqueue_failed"`,
		"enqueue_attempted=true",
		"expected_event_absent=false",
	} {
		if !strings.Contains(logs.String(), want) {
			t.Fatalf("logs missing %q: %s", want, logs.String())
		}
	}
	if strings.Contains(logs.String(), "ingress_503_before_queue") {
		t.Fatalf("enqueue failure was misclassified as before queue: %s", logs.String())
	}
}

// TestDispatch_IdempotencyKeyCanonicalizesSecretPath pins that case/whitespace
// aliases of one webhook-secret path produce the SAME endpoint-scoped
// idempotency key. Webhook-secret names are canonicalized (lower-case + trim)
// at authoring and lookup, so "/webhooks/ns/Events" and "/webhooks/ns/events"
// resolve the same secret and verify the same body; the dispatcher must scope
// idempotency off the canonical name too, or a path alias would bypass replay
// protection and enqueue the same delivery twice.
func TestDispatch_IdempotencyKeyCanonicalizesSecretPath(t *testing.T) {
	const rawKey = "evt_1"
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-canon-1", Payload: []byte(`{}`), IdempotencyKey: rawKey},
		},
	}
	var enqueuedKeys []string
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			for _, e := range events {
				enqueuedKeys = append(enqueuedKeys, e.IdempotencyKey)
			}
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "id"
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	// Same secret path in different case → must produce the same scoped key.
	for _, ep := range []string{"Events", "events"} {
		if w := postRequest(d, testNamespace, ep, `{}`); w.Code != http.StatusAccepted {
			t.Fatalf("path %q status = %d, want 202", ep, w.Code)
		}
	}
	if len(enqueuedKeys) != 2 {
		t.Fatalf("enqueuedKeys len = %d, want 2", len(enqueuedKeys))
	}
	if enqueuedKeys[0] != enqueuedKeys[1] {
		t.Fatalf("case-aliased secret paths produced different idempotency keys %q vs %q; replay protection is bypassable via path case", enqueuedKeys[0], enqueuedKeys[1])
	}
}

// TestDispatcher_PackNotActive_ReturnsTransient: when resolvePack (Step 2) returns
// a row with an empty InstallID (corrupt / mid-apply DB state), the dispatcher must
// not enqueue — Enqueue's packInstallID guard returns an error → 500 (fail closed).
// This replaces the old "resolveActivePack returns ErrInactivePack" test: with the
// R7 single-resolve fix, Dispatch no longer calls resolveActivePack at Step 8; the
// Step-2 packRow.InstallID is the authoritative packInstallID for the entire request.
func TestDispatcher_PackNotActive_ReturnsTransient(t *testing.T) {
	enqueueCalls := 0
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-corrupt-1", Payload: []byte(`{}`), IdempotencyKey: "k1"},
		},
	}
	// resolvePack returns a row but with InstallID="" — simulates a pack row that
	// passed the active check but carries no install identity (e.g. mid-apply).
	corruptResolvePack := func(_ context.Context, ns string) (*connector_packstate.ConnectorPackRow, bool, error) {
		return &connector_packstate.ConnectorPackRow{Namespace: ns, InstallID: ""}, true, nil
	}
	d := &Dispatcher{
		resolvePack:   corruptResolvePack,
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, parser),
		// Mirror the real connector_events.Service.EnqueueBatch guard: reject empty packInstallID.
		enqueueBatch: func(_ context.Context, _, packID, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			if strings.TrimSpace(packID) == "" {
				return nil, nil, errors.New("connector-events: Enqueue requires a non-empty packInstallID")
			}
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "id"
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	// Enqueue rejects the empty packInstallID → dispatcher returns 503
	// (DB/infra fault path — enqueue returns a non-nil, non-incarnation error).
	// This is fail-closed: a corrupt pack row must never silently accept events,
	// and 503 keeps it retriable so the delivery is not permanently dropped.
	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (empty packInstallID enqueue error is DB/infra fault → 503)", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0 (must not complete enqueue with empty packInstallID)", enqueueCalls)
	}
}

// TestDispatcher_RouteResolverCalledWithSamePackInstallIDAsEnqueue is the explicit
// invariant test for the R6 fix: ResolveFlow and Enqueue must receive the same
// packInstallID within a single dispatch, eliminating the impossible-tuple race.
func TestDispatcher_RouteResolverCalledWithSamePackInstallIDAsEnqueue(t *testing.T) {
	const resolvedPackInstallID = "install-specific-abc"

	var routeResolvePackID string
	var enqueuePackID string

	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-route-1", Payload: []byte(`{}`), IdempotencyKey: "k1"},
		},
	}

	// stubRouteResolver captures the packInstallID passed to ResolveFlow.
	capturingRoutes := &capturingRouteResolver{
		version:       1,
		capturePackID: func(id string) { routeResolvePackID = id },
	}

	d := &Dispatcher{
		// R7 fix: the packInstallID originates from resolvePack (Step 2), NOT from
		// a separate resolveActivePack call. Use stubPackActiveWithInstallID so that
		// the Step-2 packRow carries the expected install ID. resolveActivePack is
		// still injected (it may be called by tests that reach that path) but Dispatch
		// does not call it — only resolvePack's row.InstallID is used.
		resolvePack:   stubPackActiveWithInstallID(resolvedPackInstallID),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistryForPackInstallID(testNamespace, resolvedPackInstallID, true, &fakeVerifier{}, parser),
		enqueueBatch: func(_ context.Context, _, packID, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueuePackID = packID
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "id"
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       capturingRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusAccepted {
		t.Fatalf("status = %d, want 202", w.Code)
	}
	if routeResolvePackID != resolvedPackInstallID {
		t.Fatalf("ResolveFlow received packInstallID=%q, want %q", routeResolvePackID, resolvedPackInstallID)
	}
	if enqueuePackID != resolvedPackInstallID {
		t.Fatalf("Enqueue received packInstallID=%q, want %q", enqueuePackID, resolvedPackInstallID)
	}
	if routeResolvePackID != enqueuePackID {
		t.Fatalf("ResolveFlow packInstallID=%q != Enqueue packInstallID=%q — impossible-tuple race invariant violated", routeResolvePackID, enqueuePackID)
	}
}

// TestDispatcher_Step2Step8_UsesSamePackInstallID is the security-boundary invariant
// test for the R7 fix: the packInstallID used at Step 2 (authorization gate —
// signature verification context) must be the SAME value used at Step 8 (route
// resolution + enqueue).  Two separate DB reads 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).
//
// The test simulates this race by wiring:
//   - resolvePack (Step 2) → returns a row with InstallID "install-pack-a"
//   - resolveActivePack (Step 8 — old path, no longer called) → returns "install-pack-b"
//
// Under the R7 fix, Dispatch must use "install-pack-a" for BOTH route resolution
// and enqueue.  If resolveActivePack were still called at Step 8, it would return
// "install-pack-b", and the test would catch the mismatch.
func TestDispatcher_Step2Step8_UsesSamePackInstallID(t *testing.T) {
	const step2PackInstallID = "install-pack-a"
	const step8WouldBePackInstallID = "install-pack-b" // returned by resolveActivePack — must NOT be used

	var routeResolvePackID string
	var enqueuePackID string

	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-step2-1", Payload: []byte(`{}`), IdempotencyKey: "key-1"},
		},
	}

	capturingRoutes := &capturingRouteResolver{
		version:       1,
		capturePackID: func(id string) { routeResolvePackID = id },
	}

	d := &Dispatcher{
		// Step 2: resolvePack returns pack-A's row (this is the "authorization gate" result).
		resolvePack: stubPackActiveWithInstallID(step2PackInstallID),
		// resolveActivePack is wired to return pack-B — if Dispatch calls this at Step 8,
		// the test will catch the mismatch as a security boundary violation.
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistryForPackInstallID(testNamespace, step2PackInstallID, true, &fakeVerifier{}, parser),
		enqueueBatch: func(_ context.Context, _, packID, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueuePackID = packID
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = "id"
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       capturingRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusAccepted {
		t.Fatalf("status = %d, want 202", w.Code)
	}
	// Both route resolution and enqueue must receive the Step-2-resolved pack ID.
	if routeResolvePackID != step2PackInstallID {
		t.Fatalf("ResolveFlow packInstallID=%q, want Step-2 value %q (must not use resolveActivePack result %q)",
			routeResolvePackID, step2PackInstallID, step8WouldBePackInstallID)
	}
	if enqueuePackID != step2PackInstallID {
		t.Fatalf("Enqueue packInstallID=%q, want Step-2 value %q (must not use resolveActivePack result %q)",
			enqueuePackID, step2PackInstallID, step8WouldBePackInstallID)
	}
	if routeResolvePackID != enqueuePackID {
		t.Fatalf("ResolveFlow and Enqueue used different pack IDs (%q vs %q) — security boundary violated",
			routeResolvePackID, enqueuePackID)
	}
}

// capturingRouteResolver is a test double that records the packInstallID passed
// to ResolveFlow without interfering with the happy path.
type capturingRouteResolver struct {
	version       int64
	flow          string
	capturePackID func(string)
	gotNamespace  string
	gotEventKind  string
}

func (r *capturingRouteResolver) ResolveFlow(_ context.Context, namespace, packInstallID, eventKind string) (int64, string, error) {
	if r.capturePackID != nil {
		r.capturePackID(packInstallID)
	}
	r.gotNamespace = namespace
	r.gotEventKind = eventKind
	flow := r.flow
	if flow == "" {
		flow = "CapturedHandler"
	}
	return r.version, flow, nil
}

func (r *capturingRouteResolver) ResolveFlowByVersion(_ context.Context, _, _ string, _ int64) (string, error) {
	if r.flow != "" {
		return r.flow, nil
	}
	return "CapturedHandler", nil
}

func (r *capturingRouteResolver) ResolveExecutionSpecByVersion(ctx context.Context, installedPackID, eventKind string, version int64) (connector_inbound_routing.RouteExecutionSpec, error) {
	flow, err := r.ResolveFlowByVersion(ctx, installedPackID, eventKind, version)
	if err != nil {
		return connector_inbound_routing.RouteExecutionSpec{}, err
	}
	return connector_inbound_routing.NormalizeRouteExecutionSpec(connector_inbound_routing.RouteExecutionSpec{FlowDeveloperName: flow})
}

// ─────────────────────────────────────────────────────────────────────────────
// TestDispatcher_PackIncarnationChangedDuringDispatch_Returns503:
// when Enqueue returns ErrPackIncarnationChanged (pack rotated between Step 2
// and Step 8), dispatcher must return 503 packTransient so the provider retries.
// On retry, Step 2 re-resolves the new active pack incarnation.
// ─────────────────────────────────────────────────────────────────────────────

func TestDispatcher_PackIncarnationChangedDuringDispatch_Returns503(t *testing.T) {
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-race-1", Payload: []byte(`{}`), IdempotencyKey: "evt-race"},
		},
	}
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActive(),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      buildRegistry(testNamespace, true, &fakeVerifier{}, parser),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			// Simulate pack incarnation changed between Step 2 and INSERT.
			return nil, nil, connector_events.ErrPackIncarnationChanged
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{"event":"contact.created"}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (pack incarnation change must be retriable)", w.Code)
	}
	if enqueueCalls != 1 {
		t.Fatalf("enqueueCalls = %d, want 1", enqueueCalls)
	}
	var resp map[string]interface{}
	if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
		t.Fatalf("decode response: %v", err)
	}
	if resp["error"] != "service unavailable" {
		t.Fatalf("response error = %q, want %q", resp["error"], "service unavailable")
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// R13: Steps 3+4 packInstallID incarnation-change tests
//
// These two tests cover the race window introduced by NCRM-INT-11 R12 and
// closed in R13: an admin reinstall of the connector pack between Step 2
// (resolvePack) and Step 3 (resolveSecret) / Step 4 (ResolveSignatureVerifier)
// must produce a retriable 503 instead of a terminal 401.
// ─────────────────────────────────────────────────────────────────────────────

// TestDispatcher_PackReinstallBetweenStep2AndStep3_Returns503NotTerminal401
// simulates a pack reinstall in the Step-2→Step-3 window. resolveSecret is
// stubbed to return ErrWebhookSecretIncarnationChanged — the sentinel that
// ResolveWebhookSecretForVerification emits when the active pack's
// PackInstallId diverges from the dispatcher's cached Step-2 value.
// The dispatcher must return 503, not 401.
func TestDispatcher_PackReinstallBetweenStep2AndStep3_Returns503NotTerminal401(t *testing.T) {
	enqueueCalls := 0
	d := &Dispatcher{
		resolvePack:   stubPackActiveWithInstallID("install-pack-a"),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		// Step 3 returns incarnation-changed sentinel: pack reinstalled as B between Step 2 and 3.
		resolveSecret: func(_ context.Context, _, _, _ string) (*webhooks.ResolvedWebhookSecret, error) {
			return nil, webhooks.ErrWebhookSecretIncarnationChanged
		},
		registry: buildRegistry(testNamespace, true, &fakeVerifier{}, &fakeParser{}),
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (Step-2→Step-3 incarnation change must be retriable, not terminal 401)", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0 (must not enqueue after incarnation-change error)", enqueueCalls)
	}
}

// TestDispatcher_PackReinstallBetweenStep2AndStep4_Returns503NotTerminal401
// simulates a pack reinstall in the Step-2→Step-4 window. The capability gate
// returns ErrCapabilityIncarnationChanged — the sentinel that the production
// CapabilityGate emits when the active pack's PackInstallId diverges from the
// dispatcher's cached Step-2 value. The dispatcher must return 503, not 401.
func TestDispatcher_PackReinstallBetweenStep2AndStep4_Returns503NotTerminal401(t *testing.T) {
	enqueueCalls := 0
	// Gate returns ErrCapabilityIncarnationChanged for the verifier kind, simulating
	// a pack reinstall discovered when the gate re-reads PackInstallId from DB.
	gate := func(_ context.Context, _ string, kind connectorcapabilities.CapabilityKind, _ string) (bool, error) {
		if kind == connectorcapabilities.SignatureVerifier {
			return false, connectorcapabilities.ErrCapabilityIncarnationChanged
		}
		return true, nil
	}
	reg := connectorcapabilities.NewRegistry(gate)
	_ = reg.UpsertCapabilityForPackInstall(testNamespace, connectorcapabilities.SignatureVerifier, "install-pack-a", &fakeVerifier{})
	_ = reg.UpsertCapabilityForPackInstall(testNamespace, connectorcapabilities.EnvelopeParser, "install-pack-a", &fakeParser{})

	d := &Dispatcher{
		resolvePack:   stubPackActiveWithInstallID("install-pack-a"),
		packTransient: func(context.Context, string) (bool, error) { return false, nil },
		resolveSecret: stubSecret("hmac-sha256", []byte("s")),
		registry:      reg,
		enqueueBatch: func(_ context.Context, _, _, _ string, events []connector_events.BatchEvent) ([]string, []bool, error) {
			enqueueCalls++
			_ids := make([]string, len(events))
			for _bi := range events {
				_ids[_bi] = ""
			}
			return _ids, make([]bool, len(events)), nil
		},
		routes:       noopRoutes,
		maxBodyBytes: maxInboundWebhookBodyBytes,
	}

	w := postRequest(d, testNamespace, testSecretName, `{}`)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (Step-2→Step-4 incarnation change must be retriable, not terminal 401)", w.Code)
	}
	if enqueueCalls != 0 {
		t.Fatalf("enqueueCalls = %d, want 0 (must not enqueue after incarnation-change error)", enqueueCalls)
	}
}