// Package connectorcapabilities is the process-local handler projection for
// connector-pack source-declared inbound capabilities (signature verification,
// envelope parsing).
//
// This is Surface 1 of the two-surface design documented at
// https://github.com/pingcap-inc/NexusCRM/issues/1445#issuecomment-4563730401.
// Surface 2 (source-declaration tracking) lives in
// backend/internal/controlplane/packs (_System_ConnectorCapabilityDeclaration,
// IsCapabilityDeclaredByConnectorPack). The two surfaces are not redundant:
// Surface 1 owns the refreshable (namespace, kind) → handler projection; Surface
// 2 owns "is this binding authorized by installed pack data right now".
//
// Capability bindings are source-declared facts on the connector pack's
// manifest (ComponentKindConnectorCapability), tracked through apply/hash/verify.
// Handlers may be registered by legacy compiled test modules or built from
// source-pack handler config rows. Handler registration alone does NOT make a
// binding dispatchable: Resolve requires both a registered handler projection
// and active, hash-stable pack data in the bound CapabilityGate.
//
// v1 scope:
//
//	This package resolves (namespace, kind) → handler for the two capability
//	kinds declared in v1 connector packs. Capability kinds are a CLOSED v1
//	enum (signature_verifier, envelope_parser). Adding a third kind is the
//	explicit RFC trigger for the Connector SDK, NOT a code change here.
//
// Resolution invariant:
//
//	Resolve* returns a handler iff:
//	  (a) a handler projection is registered for (namespace, kind), AND
//	  (b) the connector pack declares the same binding, AND
//	  (c) the connector pack has a live handler-config row for the same kind OR
//	      it is an installed legacy connector pack with a compiled handler projection
//	      registered for the same (namespace, kind), AND
//	  (d) the pack is active, Category == ConnectorPack, Applied == Desired == currentHash
//	Any failing condition collapses to ErrCapabilityNotAvailable.
//
// SSoT discipline: this registry is only an in-memory projection. It does NOT
// own pack active-state, capability-declaration, handler-config, or legacy
// installed-row compatibility facts; every Resolve call delegates those durable
// facts to the bound CapabilityGate.
package connectorcapabilities

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"sync"
)

// isNilHandler returns true when h is either an interface-nil value OR a
// typed-nil pointer/interface/etc wrapped in a non-nil interface. The bare
// `h == nil` check catches only the first case; without reflect, a caller
// that passes `var v *MyVerifier = nil; RegisterCapability(ns, kind, v)`
// would store a non-nil interface holding a nil pointer, which the registry
// then returns from Resolve* and the caller dereferences → runtime panic.
// Per Codex Review BLOCK on PR #1521, registration MUST fail closed with
// ErrNilHandler in both directions.
func isNilHandler(h any) bool {
	if h == nil {
		return true
	}
	v := reflect.ValueOf(h)
	switch v.Kind() {
	case reflect.Ptr, reflect.Interface, reflect.Map, reflect.Slice, reflect.Chan, reflect.Func:
		return v.IsNil()
	}
	return false
}

// CapabilityKind is the closed v1 enum of capability kinds. Adding a third
// kind triggers the Connector SDK RFC (#1428 §10 rung 3), NOT a code change here.
type CapabilityKind string

const (
	SignatureVerifier CapabilityKind = "signature_verifier"
	EnvelopeParser    CapabilityKind = "envelope_parser"
)

// IsKnown reports whether the value is one of the v1 closed-set kinds.
func (k CapabilityKind) IsKnown() bool {
	switch k {
	case SignatureVerifier, EnvelopeParser:
		return true
	}
	return false
}

// VerifierInput carries the inbound request shape passed to a signature
// verifier. Extended by the generic inbound webhook dispatcher (#338) to carry
// the resolved secret bytes and algorithm so verifier implementations receive
// everything needed for HMAC/signature validation in one struct.
//
// Method, RequestURI, and RequestURL are server-observed values populated by
// the generic webhook dispatcher (#1527). Method comes from Request.Method;
// RequestURI preserves the raw Request.RequestURI/path+query target for v1
// signing-template compatibility; RequestURL is the opt-in full URL built from
// a trusted base plus the raw request target. They are used for signing-string
// construction instead of any caller-controlled header. Verifiers MUST NOT read
// Method, RequestURI, or RequestURL from input.Headers (caller-controlled); use
// these fields instead.
type VerifierInput struct {
	Headers   map[string]string
	RawBody   []byte
	Secret    []byte // populated by dispatcher (#338); nil if not applicable
	Algorithm string // populated by dispatcher (#338); empty if not applicable
	// SecretName is the canonical (lower-cased, trimmed) webhook-secret name /
	// endpoint the delivery arrived on. A pack may declare multiple named webhook
	// secrets; this field disambiguates which endpoint so verifiers can perform
	// per-endpoint signature canonicalization if needed.
	SecretName string
	// Method is the inbound HTTP method (e.g. "POST", "GET") as observed by the
	// dispatcher from Request.Method. Populated by the generic webhook dispatcher
	// (#1527); use this for signing-string construction instead of any
	// caller-controlled header. Empty if the verifier is not in a request context
	// (e.g. test fixtures).
	Method string
	// RequestURI is the raw inbound request target used for legacy v1 signing
	// templates (e.g. "/webhooks/ns/default?x=1"). The dispatcher reads
	// Request.RequestURI, falling back to URL.RequestURI(), and never includes
	// scheme or host. Use this for existing {RequestURI} signing-string
	// construction instead of any caller-controlled header. Empty if not in a
	// request context.
	RequestURI string
	// RequestURL is the full inbound request URL for opt-in signing templates
	// (e.g. "https://public.example.test/webhooks/ns/default?x=1"). The
	// dispatcher constructs it from WEBHOOK_BASE_URL, then API_BASE_URL, then a
	// direct TLS+Host fallback only when no Forwarded / X-Forwarded-* headers
	// are present. Empty if not in a request context.
	RequestURL string
	// DeliveryIdentity is an optional stable provider delivery identity populated
	// by a signature verifier only after the identity has been covered by signed
	// request material. Dispatchers use it for receipt-level idempotency.
	DeliveryIdentity string
	// ReceiptHeaderAllowlist is the optional pack-owned set of inbound headers
	// that may be encrypted onto a verified ingress receipt. Empty means store no
	// headers; L0 does not carry provider/header-specific admission policy.
	ReceiptHeaderAllowlist []string
	// EnableVerifiedIngressReceipt is the runtime switch for receipt-only verifier
	// outputs. Pack metadata may be staged before the global receipt mode is enabled;
	// verifiers must leave DeliveryIdentity and ReceiptHeaderAllowlist inert until
	// the dispatcher sets this flag for a receipt-enabled request.
	EnableVerifiedIngressReceipt bool
}

// ParserInput carries the inbound request shape passed to an envelope parser.
// It mirrors VerifierInput and is extended by the generic inbound webhook
// dispatcher (#338) so parsers receive the verified body together with the
// endpoint identity.
type ParserInput struct {
	Payload []byte // the raw verified request body
	// SecretName is the canonical (lower-cased, trimmed) webhook-secret name /
	// endpoint the delivery arrived on (see VerifierInput.SecretName). A pack
	// may declare multiple named webhook secrets; this field disambiguates which
	// endpoint so parsers can perform per-endpoint envelope routing if needed.
	SecretName string
}

// ConnectorEvent is the parser output. One inbound envelope may produce zero,
// one, or many events. Extended by the generic inbound webhook dispatcher (#338)
// with IdempotencyKey so parsers can surface provider event-IDs for replay protection.
//
// ObjectKind (added in NCRM-INT-11) is the connector object kind (e.g. "contact",
// "deal") this event applies to. Parsers MUST set ObjectKind to a non-empty value;
// [connector_inbound.EnvelopeParser] enforces this at construction. The dispatcher
// rejects events with an empty ObjectKind as a parser-bug class error (503).
//
// SourceID (added in NCRM-INT-11 round 22) is the connector-side object identifier
// (e.g. the provider's recordID). Parsers MUST set SourceID to a non-empty value;
// the dispatcher rejects events with an empty SourceID as a parser-bug class error (503).
//
// Cross-process queue carriage: the dispatcher encodes ObjectKind, SourceID, and
// the user Payload bytes into a connector_inbound.QueueEnvelope before calling
// Enqueue. The worker decodes the envelope to recover the typed metadata and passes
// the original user payload bytes to BatchRead. Payload bytes are not modified by
// the dispatcher (no reserved-key injection).
type ConnectorEvent struct {
	EventType string
	Namespace string
	// ObjectKind is the connector object kind (e.g. "contact", "deal") the event
	// refers to. EnvelopeParserHandler implementations MUST populate it; the
	// dispatcher rejects events with an empty ObjectKind as a parser-bug class error (503).
	// It is carried to the worker via the queue envelope (see connector_inbound/queue_envelope.go).
	ObjectKind string
	// SourceID is the connector-side object identifier (e.g. provider's recordID)
	// used by the worker to BatchRead the full record. EnvelopeParserHandler
	// implementations MUST populate it; the dispatcher rejects events with an empty
	// SourceID as a parser-bug class error (503). Carried via the queue envelope.
	SourceID string
	// SourceLaneObjectKind and SourceLaneID optionally override the queue claim
	// lane used to serialize this event against sibling events. They do not
	// change ObjectKind/SourceID carried to the worker for BatchRead. Use this
	// only when the provider event mutates a shared parent/resource that is
	// different from the source object being read.
	SourceLaneObjectKind string
	SourceLaneID         string
	// Payload is the user payload bytes (no reserved-key injection). The dispatcher
	// wraps these bytes with typed metadata (ObjectKind, SourceID, IdempotencyKey,
	// EventType) into a queue envelope before Enqueue; the worker unwraps and routes.
	Payload []byte
	// IdempotencyKey is populated by the connector's envelope parser. It MUST be
	// a stable, unique key per event (e.g. the provider's delivery/event ID) so
	// replay protection works. Dedup is scoped to (InstalledPackId, IdempotencyKey);
	// an empty key collides with every other keyless event from the same pack and
	// folds them into one row, so parsers must not leave it empty for distinct events.
	IdempotencyKey string // populated by parser; required for dedup (see #1527)
}

// SignatureVerifierHandler verifies an inbound request's signature/HMAC.
// Connector Go modules implement this interface and register via RegisterCapability.
//
// Error contract: Verify MUST return ErrInvalidSignature (optionally wrapped
// with detail via fmt.Errorf("...: %w", ErrInvalidSignature)) when the signature
// is definitively invalid — the dispatcher maps that, and ONLY that, to a
// terminal 401. Any other error is treated as a server-side fault and mapped to
// a retriable 5xx, so a handler bug or a transient dependency failure is never
// reported to the provider as bad credentials (which would make it stop retrying
// a legitimate delivery).
type SignatureVerifierHandler interface {
	Verify(ctx context.Context, input *VerifierInput) error
}

// EnvelopeParserHandler parses a verified inbound payload into connector events.
// Connector Go modules implement this interface and register via RegisterCapability.
//
// Error contract: Parse MUST return ErrMalformedPayload (optionally wrapped via
// fmt.Errorf("...: %w", ErrMalformedPayload)) when the body is definitively
// malformed for this connector — the dispatcher maps that, and ONLY that, to a
// terminal 400. Any other error is treated as a server-side fault and mapped to
// a retriable 5xx, so a parser defect or a transient failure does not
// permanently drop a valid delivery by misclassifying it as a client error.
//
// input.SecretName carries the canonical endpoint identity so parsers can route
// per-endpoint when a pack declares multiple named webhook secrets.
type EnvelopeParserHandler interface {
	Parse(ctx context.Context, input *ParserInput) ([]ConnectorEvent, error)
}

// CapabilityGate decides whether a connector pack currently has a given
// (namespace, kind) capability declared in its source manifest and is active.
// Production wires this to ProductionGate(db); tests inject stubs.
//
// packInstallID is the per-incarnation identity resolved at dispatcher Step 2.
// The gate implementation SHOULD verify that the currently-active pack still
// carries this PackInstallId so that an admin reinstall between Step 2 and this
// call returns (false, ErrCapabilityIncarnationChanged) — a distinct sentinel
// that the Resolve path maps to a retriable 503 instead of a terminal 401.
// An empty packInstallID skips the incarnation check (used by admin/test paths
// that have no prior Step-2 identity).
//
// Returning (false, nil) means: pack missing / inactive / soft-deleted / wrong
// Category / hash drift / capability not declared in source — all collapse to
// ErrCapabilityNotAvailable in the Resolve path.
// Returning (false, ErrCapabilityIncarnationChanged) means: pack active and
// capability declared, but the pack was reinstalled since Step 2 — maps to 503.
type CapabilityGate func(ctx context.Context, namespace string, kind CapabilityKind, packInstallID string) (bool, error)

type signatureVerifierEntry struct {
	handler        SignatureVerifierHandler
	packInstallID  string
	legacyCompiled bool
}

type envelopeParserEntry struct {
	handler        EnvelopeParserHandler
	packInstallID  string
	legacyCompiled bool
}

// Sentinel errors. Resolution failures use ErrCapabilityNotAvailable as the
// outermost sentinel; registration failures use distinct sentinels.
var (
	ErrCapabilityAlreadyRegistered = errors.New("connectorcapabilities: (namespace, kind) already registered")
	ErrUnknownCapabilityKind       = errors.New("connectorcapabilities: unknown capability kind")
	ErrCapabilityNotAvailable      = errors.New("connectorcapabilities: capability not available")
	ErrEmptyNamespace              = errors.New("connectorcapabilities: empty namespace")
	ErrNilHandler                  = errors.New("connectorcapabilities: nil handler")
	ErrGateNotBound                = errors.New("connectorcapabilities: default capability gate not bound")

	// ErrCapabilityIncarnationChanged is returned by the capability gate when the
	// active connector pack's PackInstallId no longer matches the packInstallID
	// supplied by the dispatcher (resolved at Step 2). An admin reinstall between
	// Step 2 and the capability gate check triggered this transition. The Resolve
	// path maps this to a retriable 503 rather than a terminal 401 so the provider
	// does not permanently drop a delivery that was valid for the previous incarnation.
	ErrCapabilityIncarnationChanged = errors.New("connectorcapabilities: pack incarnation changed since route resolution")

	// Handler-contract sentinels — returned BY capability handlers (not by the
	// registry) so a consumer (the inbound webhook dispatcher, NCRM-INT-04 #338)
	// can classify a handler failure as a terminal CLIENT error vs a retriable
	// SERVER fault. A handler returning any non-sentinel error is treated as a
	// server-side fault. See SignatureVerifierHandler.Verify (ErrInvalidSignature
	// → 401) and EnvelopeParserHandler.Parse (ErrMalformedPayload → 400).
	ErrInvalidSignature = errors.New("connectorcapabilities: invalid signature")
	ErrMalformedPayload = errors.New("connectorcapabilities: malformed payload")

	// ErrParserConfigDrift is returned by EnvelopeParserHandler.Parse when a
	// matched rule's typed-field extraction (SourceIDPath, IdempotencyKeyPath,
	// EventTypeFromPath) resolves to absent/null/empty — no scalar value was
	// found at the configured path. This is most likely a pack mis-configured
	// path (the provider + body may be valid).
	//
	// Distinct from ErrMalformedPayload (composite value at a scalar path →
	// client sent wrong shape → terminal 400): a missing value means the rule's
	// path probably does not exist in the delivered item, which is an admin-
	// addressable config gap, not a permanently-malformed provider payload.
	//
	// The dispatcher maps ErrParserConfigDrift to a retriable 503 and emits a
	// structured config-drift log so an admin can identify and fix the path.
	// The delivery is NOT dropped: durability during the drift window depends on
	// the provider's own retry budget (no config-drift DLQ exists yet).
	ErrParserConfigDrift = errors.New("connectorcapabilities: parser config drift (typed-field path resolved to no scalar value)")
)

// Registry is the process-local (namespace, kind) → handler lookup. It may be
// populated by legacy compiled test modules via RegisterCapability, by pack-data
// handler config via UpsertCapabilityForPackInstall, or by tests via NewRegistry.
//
// Handler registration alone does NOT make a binding dispatchable. Resolve*
// requires both: a handler projection registered here AND source-declared,
// active, hash-locked durable facts accepted by the CapabilityGate.
type Registry struct {
	mu        sync.RWMutex
	verifiers map[string]signatureVerifierEntry
	parsers   map[string]envelopeParserEntry
	gate      CapabilityGate
}

// NewRegistry constructs a registry with an explicit capability gate. Tests
// inject a stub gate; production code reaches the singleton via Default().
func NewRegistry(gate CapabilityGate) *Registry {
	return &Registry{
		verifiers: make(map[string]signatureVerifierEntry),
		parsers:   make(map[string]envelopeParserEntry),
		gate:      gate,
	}
}

// RegisterSignatureVerifier installs handler at namespace. Fails closed on
// empty namespace, nil handler, or duplicate (namespace, signature_verifier).
//
// Registration alone does NOT make the handler dispatchable — Resolve requires
// a source-declared binding in the installed connector pack's manifest.
func (r *Registry) RegisterSignatureVerifier(namespace string, handler SignatureVerifierHandler) error {
	if namespace == "" {
		return ErrEmptyNamespace
	}
	if isNilHandler(handler) {
		return ErrNilHandler
	}
	ns := strings.ToLower(strings.TrimSpace(namespace))
	r.mu.Lock()
	defer r.mu.Unlock()
	if _, exists := r.verifiers[ns]; exists {
		return fmt.Errorf("%w: namespace=%s kind=%s", ErrCapabilityAlreadyRegistered, namespace, SignatureVerifier)
	}
	r.verifiers[ns] = signatureVerifierEntry{handler: handler, legacyCompiled: true}
	return nil
}

// RegisterEnvelopeParser installs handler at namespace. Same fail-closed rules.
//
// Registration alone does NOT make the handler dispatchable — Resolve requires
// a source-declared binding in the installed connector pack's manifest.
func (r *Registry) RegisterEnvelopeParser(namespace string, handler EnvelopeParserHandler) error {
	if namespace == "" {
		return ErrEmptyNamespace
	}
	if isNilHandler(handler) {
		return ErrNilHandler
	}
	ns := strings.ToLower(strings.TrimSpace(namespace))
	r.mu.Lock()
	defer r.mu.Unlock()
	if _, exists := r.parsers[ns]; exists {
		return fmt.Errorf("%w: namespace=%s kind=%s", ErrCapabilityAlreadyRegistered, namespace, EnvelopeParser)
	}
	r.parsers[ns] = envelopeParserEntry{handler: handler, legacyCompiled: true}
	return nil
}

// ResolveSignatureVerifier returns the handler for namespace iff:
//   - the capability gate confirms the binding is source-declared in the
//     installed connector pack (active + hash-consistent + capability declared), AND
//   - a handler projection is registered for (namespace, SignatureVerifier).
//
// packInstallID is the per-incarnation identity from dispatcher Step 2. The gate
// validates that the currently-active pack still carries this ID; a mismatch
// (admin reinstall between Step 2 and this call) returns ErrCapabilityIncarnationChanged
// so the dispatcher can return a retriable 503 instead of a terminal 401.
// Pass an empty string to skip the incarnation check (admin/test paths with no
// prior Step-2 identity).
//
// Returns ErrCapabilityNotAvailable when the pack is missing/soft-deleted/
// inactive/wrong-Category/hash-drift OR no binding/config declared OR no handler registered.
// Returns ErrCapabilityIncarnationChanged when the pack was reinstalled since Step 2.
func (r *Registry) ResolveSignatureVerifier(ctx context.Context, namespace, packInstallID string) (SignatureVerifierHandler, error) {
	if namespace == "" {
		return nil, ErrEmptyNamespace
	}
	if err := r.assertCapabilityAvailable(ctx, namespace, SignatureVerifier, packInstallID); err != nil {
		return nil, err
	}
	ns := strings.ToLower(strings.TrimSpace(namespace))
	r.mu.RLock()
	entry, ok := r.verifiers[ns]
	r.mu.RUnlock()
	if !ok {
		return nil, fmt.Errorf("%w: namespace=%s kind=%s (handler not registered)", ErrCapabilityNotAvailable, namespace, SignatureVerifier)
	}
	if err := assertProjectedPackInstallID(namespace, SignatureVerifier, entry.packInstallID, packInstallID); err != nil {
		return nil, err
	}
	return entry.handler, nil
}

// ResolveEnvelopeParser returns the EnvelopeParserHandler for namespace iff:
//   - the capability gate confirms the binding is source-declared in the
//     installed connector pack (active + hash-consistent + capability declared), AND
//   - a handler projection is registered for (namespace, EnvelopeParser).
//
// packInstallID is the per-incarnation identity from dispatcher Step 2. The gate
// validates that the currently-active pack still carries this ID; a mismatch
// (admin reinstall between Step 2 and this call) returns ErrCapabilityIncarnationChanged
// so the dispatcher can return a retriable 503 instead of misrouting the delivery
// under a stale incarnation's envelope mapping.
// Pass an empty string to skip the incarnation check (admin/test paths with no
// prior Step-2 identity).
//
// Returns ErrCapabilityNotAvailable when the pack is missing/soft-deleted/
// inactive/wrong-Category/hash-drift OR no binding/config declared OR no handler registered.
// Returns ErrCapabilityIncarnationChanged when the pack was reinstalled since Step 2.
func (r *Registry) ResolveEnvelopeParser(ctx context.Context, namespace, packInstallID string) (EnvelopeParserHandler, error) {
	if namespace == "" {
		return nil, ErrEmptyNamespace
	}
	if err := r.assertCapabilityAvailable(ctx, namespace, EnvelopeParser, packInstallID); err != nil {
		return nil, err
	}
	ns := strings.ToLower(strings.TrimSpace(namespace))
	r.mu.RLock()
	entry, ok := r.parsers[ns]
	r.mu.RUnlock()
	if !ok {
		return nil, fmt.Errorf("%w: namespace=%s kind=%s (handler not registered)", ErrCapabilityNotAvailable, namespace, EnvelopeParser)
	}
	if err := assertProjectedPackInstallID(namespace, EnvelopeParser, entry.packInstallID, packInstallID); err != nil {
		return nil, err
	}
	return entry.handler, nil
}

func assertProjectedPackInstallID(namespace string, kind CapabilityKind, projectedPackInstallID string, requestedPackInstallID string) error {
	projectedPackInstallID = strings.TrimSpace(projectedPackInstallID)
	requestedPackInstallID = strings.TrimSpace(requestedPackInstallID)
	if requestedPackInstallID == "" || projectedPackInstallID == requestedPackInstallID {
		return nil
	}
	if projectedPackInstallID == "" {
		return fmt.Errorf("%w: namespace=%q kind=%s requestedPackInstallId=%q projectedPackInstallId=<empty>",
			ErrCapabilityIncarnationChanged, namespace, kind, requestedPackInstallID)
	}
	return fmt.Errorf("%w: namespace=%q kind=%s requestedPackInstallId=%q projectedPackInstallId=%q",
		ErrCapabilityIncarnationChanged, namespace, kind, requestedPackInstallID, projectedPackInstallID)
}

// assertCapabilityAvailable consults the capability gate. A nil gate fails
// closed (ErrGateNotBound). The gate field is snapshotted under r.mu.RLock so
// a concurrent BindDefaultCapabilityGate cannot race the read.
//
// packInstallID is forwarded to the gate for incarnation-change detection (see
// CapabilityGate). An empty string skips the check.
//
// Returning (false, nil) from the gate collapses to ErrCapabilityNotAvailable.
// Returning (false, ErrCapabilityIncarnationChanged) propagates that sentinel
// directly so callers can map it to a retriable 503.
func (r *Registry) assertCapabilityAvailable(ctx context.Context, namespace string, kind CapabilityKind, packInstallID string) error {
	r.mu.RLock()
	gate := r.gate
	r.mu.RUnlock()
	if gate == nil {
		return fmt.Errorf("%w: namespace=%s kind=%s", ErrGateNotBound, namespace, kind)
	}
	available, err := gate(ctx, namespace, kind, packInstallID)
	if err != nil {
		// Propagate ErrCapabilityIncarnationChanged as-is so the dispatcher can
		// distinguish it from a generic gate fault and return 503 instead of 500.
		if errors.Is(err, ErrCapabilityIncarnationChanged) {
			return err
		}
		return fmt.Errorf("connectorcapabilities: capability gate failed for namespace=%s kind=%s: %w", namespace, kind, err)
	}
	if !available {
		return fmt.Errorf("%w: namespace=%s kind=%s (pack missing, inactive, wrong Category, hash drift, or capability not source-declared)", ErrCapabilityNotAvailable, namespace, kind)
	}
	return nil
}

// RegisterCapability is the polymorphic convenience helper called by connector
// modules from their init(). handler must satisfy SignatureVerifierHandler when
// kind == SignatureVerifier, or EnvelopeParserHandler when kind == EnvelopeParser.
// Unknown kind or wrong-type handler fails closed.
//
// Registration alone does NOT make the handler dispatchable — only source-declared
// bindings in the installed connector pack's manifest dispatch. See package doc.
func (r *Registry) RegisterCapability(namespace string, kind CapabilityKind, handler any) error {
	if !kind.IsKnown() {
		return fmt.Errorf("%w: kind=%q", ErrUnknownCapabilityKind, kind)
	}
	switch kind {
	case SignatureVerifier:
		h, ok := handler.(SignatureVerifierHandler)
		if !ok {
			return fmt.Errorf("connectorcapabilities: handler does not implement SignatureVerifierHandler for namespace=%s", namespace)
		}
		return r.RegisterSignatureVerifier(namespace, h)
	case EnvelopeParser:
		h, ok := handler.(EnvelopeParserHandler)
		if !ok {
			return fmt.Errorf("connectorcapabilities: handler does not implement EnvelopeParserHandler for namespace=%s", namespace)
		}
		return r.RegisterEnvelopeParser(namespace, h)
	}
	// Defensive — unreachable given IsKnown check above.
	return fmt.Errorf("%w: kind=%q", ErrUnknownCapabilityKind, kind)
}

// UpsertCapability installs or replaces a handler projection without binding it
// to a specific pack incarnation. It is intended for tests and legacy callers;
// pack-data handler config should use UpsertCapabilityForPackInstall so Resolve
// can fail closed if a stale in-memory projection is observed for a newer
// PackInstallId.
func (r *Registry) UpsertCapability(namespace string, kind CapabilityKind, handler any) error {
	return r.upsertCapability(namespace, kind, "", handler)
}

// UpsertCapabilityForPackInstall installs or replaces a handler projection
// built from _System_ConnectorInboundHandlerConfig for a specific pack
// incarnation.
func (r *Registry) UpsertCapabilityForPackInstall(namespace string, kind CapabilityKind, packInstallID string, handler any) error {
	packInstallID = strings.TrimSpace(packInstallID)
	if packInstallID == "" {
		return fmt.Errorf("connectorcapabilities: empty packInstallId for namespace=%s kind=%s", namespace, kind)
	}
	return r.upsertCapability(namespace, kind, packInstallID, handler)
}

func (r *Registry) upsertCapability(namespace string, kind CapabilityKind, packInstallID string, handler any) error {
	if !kind.IsKnown() {
		return fmt.Errorf("%w: kind=%q", ErrUnknownCapabilityKind, kind)
	}
	if strings.TrimSpace(namespace) == "" {
		return ErrEmptyNamespace
	}
	if isNilHandler(handler) {
		return ErrNilHandler
	}
	ns := strings.ToLower(strings.TrimSpace(namespace))
	switch kind {
	case SignatureVerifier:
		h, ok := handler.(SignatureVerifierHandler)
		if !ok {
			return fmt.Errorf("connectorcapabilities: handler does not implement SignatureVerifierHandler for namespace=%s", namespace)
		}
		r.mu.Lock()
		packInstallID = strings.TrimSpace(packInstallID)
		r.verifiers[ns] = signatureVerifierEntry{handler: h, packInstallID: packInstallID, legacyCompiled: packInstallID == ""}
		r.mu.Unlock()
		return nil
	case EnvelopeParser:
		h, ok := handler.(EnvelopeParserHandler)
		if !ok {
			return fmt.Errorf("connectorcapabilities: handler does not implement EnvelopeParserHandler for namespace=%s", namespace)
		}
		r.mu.Lock()
		packInstallID = strings.TrimSpace(packInstallID)
		r.parsers[ns] = envelopeParserEntry{handler: h, packInstallID: packInstallID, legacyCompiled: packInstallID == ""}
		r.mu.Unlock()
		return nil
	}
	return fmt.Errorf("%w: kind=%q", ErrUnknownCapabilityKind, kind)
}

// UnregisterCapability removes a process-local handler projection. It returns
// true only when a matching projection was present.
func (r *Registry) UnregisterCapability(namespace string, kind CapabilityKind) bool {
	ns := strings.ToLower(strings.TrimSpace(namespace))
	if ns == "" {
		return false
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	switch kind {
	case SignatureVerifier:
		if _, ok := r.verifiers[ns]; !ok {
			return false
		}
		delete(r.verifiers, ns)
		return true
	case EnvelopeParser:
		if _, ok := r.parsers[ns]; !ok {
			return false
		}
		delete(r.parsers, ns)
		return true
	default:
		return false
	}
}

var (
	defaultRegistry   = NewRegistry(nil)
	defaultGateBindMu sync.Mutex
	defaultGateBound  bool
)

// Default returns the process-global registry. At server bootstrap, call
// BindDefaultCapabilityGate to wire the production gate. Tests should construct
// their own *Registry via NewRegistry(gate) — do not use Default in tests unless
// specifically testing the default-gate binding path.
func Default() *Registry {
	return defaultRegistry
}

// BindDefaultCapabilityGate wires the production capability gate into the
// process-global default registry. One-shot: calling twice with a different
// gate is a fatal misconfiguration and returns an error rather than silently
// replacing — this preserves the "one fact, one SSoT" invariant.
func BindDefaultCapabilityGate(gate CapabilityGate) error {
	if gate == nil {
		return ErrGateNotBound
	}
	defaultGateBindMu.Lock()
	defer defaultGateBindMu.Unlock()
	if defaultGateBound {
		return errors.New("connectorcapabilities: default capability gate already bound")
	}
	defaultRegistry.mu.Lock()
	defaultRegistry.gate = gate
	defaultRegistry.mu.Unlock()
	defaultGateBound = true
	return nil
}

// ProductionGate builds a CapabilityGate from a *sql.DB. It enforces live
// handler-config rows without installed-row compatibility. Production bootstrap
// uses ProductionGateWithRegistry so legacy installed connector packs that still
// rely on compiled handlers can be grandfathered until they are re-applied with
// connector_inbound_handler_config components.
func ProductionGate(db *sql.DB) CapabilityGate {
	return ProductionGateWithRegistry(db, nil)
}

// ProductionGateWithRegistry builds a CapabilityGate from a *sql.DB plus an
// optional registry used only for installed-row compatibility. When a connector
// capability is declared on an active installed pack but lacks a handler-config
// row, a legacy compiled handler projection in legacyFallback permits dispatch
// for that already-installed row. Pack-data projections remain incarnation-bound:
// a stale pack-bound projection still fails in assertProjectedPackInstallID.
//
// Per #1489 namespace-reservation: namespace IS the installed-pack Name for
// connector packs, so packName == namespace. The production gate enforces in
// a single atomic JOIN:
//   - ConnectorPack category
//   - active status (not soft-deleted, not installing/removing/failed)
//   - AppliedPayloadHash == DesiredPayloadHash (no in-flight install)
//   - capability (namespace, kind) declared in the source manifest with IsDeleted=0
//   - handler config for the same (PackInstallId, namespace, kind) with IsDeleted=0,
//     unless installed-row compatibility is satisfied by a legacy compiled
//     handler projection registered for the same (namespace, kind)
//
// Any failing condition returns (false, nil); DB errors return (false, err).
//
// Incarnation check: if packInstallID is non-empty, the gate issues ONE atomic
// SQL that adds a PackInstallId = packInstallID predicate to the existing JOIN,
// replacing the old two-step pre-check (ResolveConnectorPackByNamespace) + query
// pattern. Collapsing both into one statement closes the race where an admin
// reinstall occurring between the pre-check SELECT and the capability JOIN SELECT
// would evaluate capability state against the new incarnation.
//
// B24: 0 rows from the atomic JOIN are disambiguated by a follow-up pack-active
// query. Pack still active under this PackInstallId → capability undeclared →
// returns (false, ErrCapabilityNotAvailable) (terminal 401). Pack no longer
// active → incarnation changed → returns (false, ErrCapabilityIncarnationChanged)
// (retriable 503).
//
// When packInstallID is "" (admin/test callers), the same active/hash/capability/
// config-or-legacy predicates apply without the incarnation predicate.
func ProductionGateWithRegistry(db *sql.DB, legacyFallback *Registry) CapabilityGate {
	return func(ctx context.Context, namespace string, kind CapabilityKind, packInstallID string) (bool, error) {
		if packInstallID != "" {
			// Atomic path: collapse incarnation check + capability declaration lookup
			// into one SQL. The PackInstallId predicate on _System_InstalledPack
			// atomically enforces that we are evaluating the same incarnation that
			// was resolved at dispatcher Step 2. Any admin reinstall between Step 2
			// and this call causes the JOIN to return 0 rows.
			//
			// The SQL mirrors IsCapabilityDeclaredByConnectorPack's JOIN but adds
			// p.PackInstallId = packInstallID as an additional WHERE condition.
			// If the helper's predicates change (new column in active-row contract),
			// this inline SQL must be updated in lock-step — see the lock-step note
			// in installed_pack_helpers.go:IsCapabilityDeclaredByConnectorPack.
			if namespace == "" || string(kind) == "" {
				return false, nil
			}
			var exists int
			queryErr := db.QueryRowContext(ctx, `
SELECT 1
  FROM _System_ConnectorCapabilityDeclaration d
  JOIN _System_InstalledPack p ON p.Name = d.InstalledPackName
  JOIN _System_ConnectorInboundHandlerConfig h
    ON h.InstalledPackId = p.PackInstallId
   AND h.Namespace = d.Namespace
   AND h.CapabilityKind = d.Kind
 WHERE p.Name = ?
   AND p.PackInstallId = ?
   AND p.Category = 'connector-pack'
   AND p.Status = 'active'
   AND COALESCE(p.IsDeleted, 0) = 0
   AND COALESCE(p.PackInstallId, '') <> ''
   AND p.AppliedPayloadHash IS NOT NULL
   AND p.AppliedPayloadHash <> ''
   AND p.DesiredPayloadHash IS NOT NULL
   AND p.DesiredPayloadHash <> ''
   AND p.AppliedPayloadHash = p.DesiredPayloadHash
   AND d.Namespace = ?
   AND d.Kind = ?
   AND COALESCE(d.IsDeleted, 0) = 0
   AND COALESCE(h.IsDeleted, 0) = 0
 LIMIT 1`,
				namespace, packInstallID,
				namespace, string(kind),
			).Scan(&exists)
			if queryErr == sql.ErrNoRows {
				// B24: 0 rows — could be (a) pack incarnation changed (PackInstallId stale)
				// or (b) pack still active but capability not declared. Distinguish with a
				// follow-up query: active pack → capability undeclared → terminal 401.
				// Inactive pack → incarnation changed → retriable 503.
				var packActive bool
				disambiguateErr := db.QueryRowContext(ctx, `
SELECT EXISTS(
  SELECT 1 FROM _System_InstalledPack
   WHERE Name = ?
     AND PackInstallId = ?
     AND Category = 'connector-pack'
     AND Status = 'active'
     AND COALESCE(IsDeleted, 0) = 0
)`, namespace, packInstallID).Scan(&packActive)
				if disambiguateErr != nil {
					// B24 → R19: disambiguation-query DB fault must NOT downgrade to
					// terminal 401. Symmetric with webhook_secret_verifier (R18 B23):
					// a transient DB fault during the follow-up EXISTS query returns
					// ErrCapabilityIncarnationChanged so the dispatcher emits 503 and
					// the provider retries — exactly the install-transition window this
					// PR exists to protect.
					return false, fmt.Errorf("%w: namespace=%q packInstallID=%q kind=%s (disambiguate-query fault: %v)",
						ErrCapabilityIncarnationChanged, namespace, packInstallID, kind, disambiguateErr)
				}
				if !packActive {
					// Pack no longer active under this PackInstallId — retriable 503.
					return false, fmt.Errorf("%w: namespace=%q packInstallID=%q kind=%s",
						ErrCapabilityIncarnationChanged, namespace, packInstallID, kind)
				}
				if legacyFallbackHasCompiledHandler(legacyFallback, namespace, kind) {
					ok, legacyErr := legacyDeclaredCapabilityAvailable(ctx, db, namespace, kind, packInstallID)
					if legacyErr != nil {
						return false, legacyErr
					}
					if ok {
						if legacyFallback.BindLegacyHandlerToPackInstall(namespace, kind, packInstallID) {
							return true, nil
						}
						return false, fmt.Errorf("%w: namespace=%q packInstallID=%q kind=%s (legacy compiled projection bound to another install)",
							ErrCapabilityIncarnationChanged, namespace, packInstallID, kind)
					}
				}
				// Pack still active but capability genuinely undeclared or has no
				// handler-config/legacy-compiled compatibility — terminal 401.
				return false, fmt.Errorf("%w: namespace=%q packInstallID=%q kind=%s (capability or handler config not declared in manifest)",
					ErrCapabilityNotAvailable, namespace, packInstallID, kind)
			}
			if queryErr != nil {
				return false, fmt.Errorf("connectorcapabilities: production gate atomic capability check %q/%s: %w", namespace, kind, queryErr)
			}
			return true, nil
		}
		// Legacy/admin path: no packInstallID supplied, so incarnation check is
		// skipped but the active/hash/capability/config predicates still apply.
		if namespace == "" || string(kind) == "" {
			return false, nil
		}
		var exists int
		queryErr := db.QueryRowContext(ctx, `
SELECT 1
  FROM _System_ConnectorCapabilityDeclaration d
  JOIN _System_InstalledPack p ON p.Name = d.InstalledPackName
  JOIN _System_ConnectorInboundHandlerConfig h
    ON h.InstalledPackId = p.PackInstallId
   AND h.Namespace = d.Namespace
   AND h.CapabilityKind = d.Kind
 WHERE p.Name = ?
   AND p.Category = 'connector-pack'
   AND p.Status = 'active'
   AND COALESCE(p.IsDeleted, 0) = 0
   AND COALESCE(p.PackInstallId, '') <> ''
   AND p.AppliedPayloadHash IS NOT NULL
   AND p.AppliedPayloadHash <> ''
   AND p.DesiredPayloadHash IS NOT NULL
   AND p.DesiredPayloadHash <> ''
   AND p.AppliedPayloadHash = p.DesiredPayloadHash
   AND d.Namespace = ?
   AND d.Kind = ?
   AND COALESCE(d.IsDeleted, 0) = 0
   AND COALESCE(h.IsDeleted, 0) = 0
 LIMIT 1`,
			namespace,
			namespace, string(kind),
		).Scan(&exists)
		if queryErr == sql.ErrNoRows {
			if legacyFallbackHasCompiledHandler(legacyFallback, namespace, kind) {
				ok, legacyErr := legacyDeclaredCapabilityAvailable(ctx, db, namespace, kind, "")
				if legacyErr != nil {
					return false, legacyErr
				}
				return ok, nil
			}
			return false, nil
		}
		if queryErr != nil {
			return false, fmt.Errorf("connectorcapabilities: production gate capability/config check %q/%s: %w", namespace, kind, queryErr)
		}
		return true, nil
	}
}

func legacyFallbackHasCompiledHandler(registry *Registry, namespace string, kind CapabilityKind) bool {
	return registry != nil && registry.HasLegacyCompiledHandler(namespace, kind)
}

func legacyDeclaredCapabilityAvailable(ctx context.Context, db *sql.DB, namespace string, kind CapabilityKind, packInstallID string) (bool, error) {
	if namespace == "" || string(kind) == "" {
		return false, nil
	}
	packInstallPredicate := ""
	args := []any{namespace}
	if strings.TrimSpace(packInstallID) != "" {
		packInstallPredicate = "   AND p.PackInstallId = ?\n"
		args = append(args, strings.TrimSpace(packInstallID))
	}
	args = append(args, namespace, string(kind))
	var exists int
	queryErr := db.QueryRowContext(ctx, `
SELECT 1
  FROM _System_ConnectorCapabilityDeclaration d
  JOIN _System_InstalledPack p ON p.Name = d.InstalledPackName
 WHERE p.Name = ?
`+packInstallPredicate+`   AND p.Category = 'connector-pack'
   AND p.Status = 'active'
   AND COALESCE(p.IsDeleted, 0) = 0
   AND COALESCE(p.PackInstallId, '') <> ''
   AND p.AppliedPayloadHash IS NOT NULL
   AND p.AppliedPayloadHash <> ''
   AND p.DesiredPayloadHash IS NOT NULL
   AND p.DesiredPayloadHash <> ''
   AND p.AppliedPayloadHash = p.DesiredPayloadHash
   AND d.Namespace = ?
   AND d.Kind = ?
   AND COALESCE(d.IsDeleted, 0) = 0
 LIMIT 1`, args...).Scan(&exists)
	if queryErr == sql.ErrNoRows {
		return false, nil
	}
	if queryErr != nil {
		return false, fmt.Errorf("connectorcapabilities: production gate legacy compiled capability check %q/%s: %w", namespace, kind, queryErr)
	}
	return true, nil
}

// RegisterCapability is the package-level convenience for connector-module
// init() functions. Equivalent to Default().RegisterCapability(...).
//
// Registration alone does NOT make the binding dispatchable. Only connector
// packs whose source manifest declares the (namespace, kind) via
// ComponentKindConnectorCapability will dispatch via Resolve*.
func RegisterCapability(namespace string, kind CapabilityKind, handler any) error {
	return defaultRegistry.RegisterCapability(namespace, kind, handler)
}

// HasHandler reports whether a process-local handler projection is registered
// for (namespace, kind) without consulting the capability gate. It is a
// registry introspection helper, not an install/verify authorization check.
func (r *Registry) HasHandler(namespace string, kind CapabilityKind) bool {
	ns := strings.ToLower(strings.TrimSpace(namespace))
	r.mu.RLock()
	defer r.mu.RUnlock()
	switch kind {
	case SignatureVerifier:
		_, ok := r.verifiers[ns]
		return ok
	case EnvelopeParser:
		_, ok := r.parsers[ns]
		return ok
	default:
		return false
	}
}

// HasUnboundHandler reports whether a process-local legacy/compiled projection
// is registered for (namespace, kind). It intentionally excludes pack-data
// projections built with UpsertCapabilityForPackInstall, so installed-row
// compatibility cannot be satisfied by stale config-derived state.
func (r *Registry) HasUnboundHandler(namespace string, kind CapabilityKind) bool {
	ns := strings.ToLower(strings.TrimSpace(namespace))
	r.mu.RLock()
	defer r.mu.RUnlock()
	switch kind {
	case SignatureVerifier:
		entry, ok := r.verifiers[ns]
		return ok && strings.TrimSpace(entry.packInstallID) == ""
	case EnvelopeParser:
		entry, ok := r.parsers[ns]
		return ok && strings.TrimSpace(entry.packInstallID) == ""
	default:
		return false
	}
}

// HasLegacyCompiledHandler reports whether a process-local legacy/compiled
// projection is registered for (namespace, kind). The projection may have been
// bound to the current installed-pack incarnation by installed-row compatibility,
// but it remains distinct from pack-data projections built from handler-config
// rows.
func (r *Registry) HasLegacyCompiledHandler(namespace string, kind CapabilityKind) bool {
	ns := strings.ToLower(strings.TrimSpace(namespace))
	r.mu.RLock()
	defer r.mu.RUnlock()
	switch kind {
	case SignatureVerifier:
		entry, ok := r.verifiers[ns]
		return ok && entry.legacyCompiled
	case EnvelopeParser:
		entry, ok := r.parsers[ns]
		return ok && entry.legacyCompiled
	default:
		return false
	}
}

// BindLegacyHandlerToPackInstall pins a legacy/compiled projection to the
// current installed-pack incarnation. It never binds pack-data projections and
// refuses to move an already-bound legacy projection to a different incarnation.
func (r *Registry) BindLegacyHandlerToPackInstall(namespace string, kind CapabilityKind, packInstallID string) bool {
	ns := strings.ToLower(strings.TrimSpace(namespace))
	packInstallID = strings.TrimSpace(packInstallID)
	if ns == "" || packInstallID == "" {
		return false
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	switch kind {
	case SignatureVerifier:
		entry, ok := r.verifiers[ns]
		if !ok || !entry.legacyCompiled {
			return false
		}
		if strings.TrimSpace(entry.packInstallID) != "" && strings.TrimSpace(entry.packInstallID) != packInstallID {
			return false
		}
		entry.packInstallID = packInstallID
		r.verifiers[ns] = entry
		return true
	case EnvelopeParser:
		entry, ok := r.parsers[ns]
		if !ok || !entry.legacyCompiled {
			return false
		}
		if strings.TrimSpace(entry.packInstallID) != "" && strings.TrimSpace(entry.packInstallID) != packInstallID {
			return false
		}
		entry.packInstallID = packInstallID
		r.parsers[ns] = entry
		return true
	default:
		return false
	}
}

// HasHandler is the package-level convenience for the default registry.
// Equivalent to Default().HasHandler(...).
func HasHandler(namespace string, kind CapabilityKind) bool {
	return defaultRegistry.HasHandler(namespace, kind)
}

// HasUnboundHandler is the package-level convenience for the default registry.
// Equivalent to Default().HasUnboundHandler(...).
func HasUnboundHandler(namespace string, kind CapabilityKind) bool {
	return defaultRegistry.HasUnboundHandler(namespace, kind)
}

// HasLegacyCompiledHandler is the package-level convenience for the default
// registry. Equivalent to Default().HasLegacyCompiledHandler(...).
func HasLegacyCompiledHandler(namespace string, kind CapabilityKind) bool {
	return defaultRegistry.HasLegacyCompiledHandler(namespace, kind)
}

// UpsertCapability is the package-level convenience for the default registry.
func UpsertCapability(namespace string, kind CapabilityKind, handler any) error {
	return defaultRegistry.UpsertCapability(namespace, kind, handler)
}

// UpsertCapabilityForPackInstall is the package-level convenience for handler
// projections built from connector_inbound_handler_config rows.
func UpsertCapabilityForPackInstall(namespace string, kind CapabilityKind, packInstallID string, handler any) error {
	return defaultRegistry.UpsertCapabilityForPackInstall(namespace, kind, packInstallID, handler)
}

// UnregisterCapability is the package-level convenience for the default registry.
func UnregisterCapability(namespace string, kind CapabilityKind) bool {
	return defaultRegistry.UnregisterCapability(namespace, kind)
}