package connector_inbound

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"log"
	"math/big"
	"sort"
	"strconv"
	"strings"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"

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

// connectorInboundItemsSkippedTotal counts envelope items that matched no
// configured rule and were skipped by the parser. Labeled by namespace.
// The skip is intentional (multi-subscription webhooks only handle declared
// types), but the absence of observability meant a misconfigured or newly-added
// provider event type was acknowledged-and-lost with no trace (issue #2127).
var connectorInboundItemsSkippedTotal = promauto.NewCounterVec(
	prometheus.CounterOpts{
		Name: "nexuscrm_connector_inbound_envelope_items_skipped_total",
		Help: "Envelope items that matched no configured rule and were skipped by the parser. " +
			"A non-zero rate may indicate configuration drift: an event type that SHOULD be " +
			"handled is mis-declared or whose route was swept. Issue #2127.",
	},
	[]string{"namespace"},
)

// EnvelopeMappingConfig is the pack-supplied declarative config for an
// [EnvelopeParser]. It describes how to turn an inbound webhook envelope into
// a slice of [connectorcapabilities.ConnectorEvent]s using JSON-path
// extraction — no connector-specific Go code required.
type EnvelopeMappingConfig struct {
	// EventArrayPath is the JSON-path to the array of event items within the
	// envelope. Use "." to denote a bare root-array envelope. Mutually
	// exclusive with EventObjectPath.
	EventArrayPath string

	// EventObjectPath is the JSON-path to one event item object within the
	// envelope. Use "." to denote a bare root-object envelope. Mutually
	// exclusive with EventArrayPath.
	EventObjectPath string

	// InboundProcessingMode declares whether downstream runtime may accept
	// payload-only processing or requires record enrichment. Legacy configs may
	// omit this field; startup can then infer compatibility from exact
	// install-scoped metadata instead of treating omission as an explicit mode.
	InboundProcessingMode InboundProcessingMode

	// Rules is the ordered list of mapping rules. The first rule that matches
	// an item wins; unmatched items are skipped and counted via the
	// nexuscrm_connector_inbound_envelope_items_skipped_total metric and a
	// server-side log line (issue #2127).
	Rules []EnvelopeMappingRule
}

// IdempotencyKeyComposite defines one provider-neutral, versioned event identity.
// Parts are evaluated in declared order; retry-attempt semantics do not belong
// in this L0 contract.
type IdempotencyKeyComposite struct {
	Version  string
	Encoding string
	Parts    []IdempotencyKeyPart
}

// IdempotencyKeyPart selects one scalar component from an envelope item.
type IdempotencyKeyPart struct {
	Path string
}

type InboundProcessingMode string

const (
	InboundProcessingModeEnrichmentRequired InboundProcessingMode = "enrichment_required"
	InboundProcessingModePayloadOnlyAllowed InboundProcessingMode = "payload_only_allowed"
)

type PayloadFieldMode string

const (
	PayloadFieldModeRaw               PayloadFieldMode = ""
	PayloadFieldModeScalarString      PayloadFieldMode = "scalar_string"
	PayloadFieldModeScalarStringArray PayloadFieldMode = "scalar_string_array"
)

// EnvelopeMappingRule describes how one class of event items maps to a
// [connectorcapabilities.ConnectorEvent].
type EnvelopeMappingRule struct {
	// MatchPath is the JSON-path within the item whose non-null, non-empty
	// value selects this rule. If both MatchPath and MatchEquals are empty the
	// rule matches every item (catch-all; place last), unless MatchAll is set.
	MatchPath string

	// MatchEquals, when non-empty, further constrains the match: the value at
	// MatchPath must equal this string. When empty, any non-null value at
	// MatchPath is sufficient to match.
	MatchEquals string

	// MatchAll, when non-empty, selects this rule only when every condition
	// matches. It is mutually exclusive with MatchPath/MatchEquals. Each
	// condition uses the same scalar extraction semantics as MatchPath:
	// absent, empty, object, or array values are non-matches; an empty Equals
	// accepts any non-empty scalar.
	MatchAll []MatchCondition

	// EventTypeFromPath is the JSON-path within the item from which to read
	// the emitted EventType. Mutually exclusive with EventTypeLiteral; exactly
	// one must be non-empty (validated at construction).
	EventTypeFromPath string

	// EventTypeLiteral is a hardcoded EventType emitted for this rule.
	// Mutually exclusive with EventTypeFromPath.
	EventTypeLiteral string

	// NamespaceFromCtx, when true, sets ConnectorEvent.Namespace to the
	// parser's bound namespace. Always recommended; false emits an empty
	// Namespace.
	NamespaceFromCtx bool

	// ObjectKind is the connector object kind (e.g. "contact", "deal") that
	// every ConnectorEvent produced by this rule applies to. Required —
	// [NewEnvelopeParser] rejects any rule with an empty ObjectKind.
	//
	// The value is written to ConnectorEvent.ObjectKind. The dispatcher carries
	// it to the worker via the typed QueueEnvelope (no reserved-key injection
	// into Payload bytes).
	//
	// v1: literal value only; path-based ObjectKind derivation is deferred.
	ObjectKind string

	// SourceIDPath is the JSON-path within the item from which to extract the
	// external / source ID of the object being described by this event (e.g.
	// ".objectId", ".id", ".fromObjectId"). Required — [NewEnvelopeParser]
	// rejects any rule with an empty SourceIDPath.
	//
	// Unlike ObjectKind (which is a literal), SourceIDPath is always path-based
	// because the source ID lives in the envelope data. The extracted value is
	// written to ConnectorEvent.SourceID. The dispatcher carries it to the worker
	// via the typed QueueEnvelope so the worker can pass it to
	// [BatchReadClient.BatchRead] as the sourceID argument.
	SourceIDPath string

	// SourceLaneObjectKind and SourceLaneIDPath optionally override the queue
	// claim lane used to serialize event execution. They do not replace
	// ObjectKind/SourceID, which still describe the source object passed to the
	// worker. Configure both fields together for events that read one object but
	// mutate a shared parent/resource.
	SourceLaneObjectKind string
	SourceLaneIDPath     string

	// IdempotencyKeyPath is the JSON-path (e.g. ".id", ".deliveryId",
	// ".event.id") whose value identifies THIS SPECIFIC EVENT (not the object).
	// Required — [NewEnvelopeParser] rejects any rule with an empty
	// IdempotencyKeyPath. The extracted value must be non-empty at parse time;
	// an absent or empty value returns [connectorcapabilities.ErrMalformedPayload].
	//
	// Distinct from SourceIDPath: SourceIDPath identifies the OBJECT (e.g.
	// contact-42) used as the BatchRead lookup key. IdempotencyKeyPath identifies
	// the EVENT and is used by the queue for dedup
	// (InstalledPackId, IdempotencyKey UNIQUE). Two distinct updates to the
	// same object must produce two distinct IdempotencyKey values — using
	// SourceID for both would fold every update into one queue row.
	IdempotencyKeyPath string

	// PayloadProjection is the ordered list of (Name, Path) fields assembled
	// into the emitted ConnectorEvent.Payload JSON object. Missing paths emit
	// JSON null for that key. Field order in the output matches the slice order.
	PayloadProjection []PayloadField
}

// MatchCondition is one generic scalar selector predicate in an envelope parser rule.
type MatchCondition struct {
	Path   string
	Equals string
}

// PayloadField is a (name, JSON-path) pair used by [EnvelopeMappingRule.PayloadProjection].
type PayloadField struct {
	// Name is the key in the emitted Payload JSON object.
	Name string
	// Path is the JSON-path within the envelope item. Absent path → null.
	Path string
	// Mode controls how scalar/array values are serialized into the payload.
	// Omitted means raw JSON passthrough for backward compatibility.
	Mode PayloadFieldMode
}

// EnvelopeParser is a [connectorcapabilities.EnvelopeParserHandler] that uses
// a declarative [EnvelopeMappingConfig] to turn inbound webhook payloads into
// connector events. Construct via [NewEnvelopeParser].
type EnvelopeParser struct {
	namespace                        string
	cfg                              EnvelopeMappingConfig
	compiledRuleIdempotencyComposite []*compiledIdempotencyKeyComposite
}

type compiledIdempotencyKeyComposite struct {
	version        string
	encoding       string
	specHash       string
	extractPaths   []string
	canonicalPaths []string
}

const (
	idempotencyCompositeVersionV1       = "v1"
	idempotencyCompositeEncodingSHA256  = "sha256"
	idempotencyCompositeKeyPrefix       = "cmp:v1:"
	idempotencyCompositeSpecHashHexSize = 12
)

// NewEnvelopeParser constructs and validates an [EnvelopeParser] for namespace.
//
// Returns [ErrInvalidParserConfig] when cfg is invalid:
//   - Exactly one of EventArrayPath or EventObjectPath is set.
//   - Rules is empty (a parser with zero rules silently drops every delivery).
//   - Any rule has an empty ObjectKind (required for worker BatchRead + queue carriage).
//   - Any rule has an empty SourceIDPath (required for worker BatchRead source ID carriage).
//   - Any rule configures exactly one of SourceLaneObjectKind or SourceLaneIDPath.
//   - The optional composite sidecar is neither empty nor index-aligned with Rules.
//   - Any rule configures neither or both legacy and composite idempotency metadata.
//   - Any legacy IdempotencyKeyPath equals SourceIDPath or EventTypeFromPath.
//   - Any composite has an unknown version/encoding, empty/duplicate parts, or invalid paths.
//   - Any rule's PayloadProjection contains a field with an empty Name (silently emits "" JSON key).
//   - Any rule's PayloadProjection contains two fields with the same Name (last-write-wins silently overwrites).
//   - Any rule has both EventTypeFromPath and EventTypeLiteral set (ambiguous).
//   - Any rule has neither EventTypeFromPath nor EventTypeLiteral set (unresolvable).
//   - More than one catch-all rule (MatchPath empty) is present.
//   - A catch-all rule is not the last rule (earlier rules would be shadowed / later rules unreachable).
//   - Two non-catch-all rules share the same selector (second rule is unreachable).
//   - Any path in Rules or PayloadProjection fails the path syntax check.
func NewEnvelopeParser(namespace string, cfg EnvelopeMappingConfig) (*EnvelopeParser, error) {
	return newEnvelopeParser(namespace, cfg, nil)
}

// NewEnvelopeParserWithIdempotencyComposites builds an envelope parser with an
// index-aligned composite identity sidecar. A nil entry selects the legacy
// IdempotencyKeyPath for that rule; a non-nil entry selects the versioned
// composite and requires the rule's IdempotencyKeyPath to be empty.
func NewEnvelopeParserWithIdempotencyComposites(namespace string, cfg EnvelopeMappingConfig, ruleComposites []*IdempotencyKeyComposite) (*EnvelopeParser, error) {
	return newEnvelopeParser(namespace, cfg, ruleComposites)
}

func newEnvelopeParser(namespace string, cfg EnvelopeMappingConfig, ruleComposites []*IdempotencyKeyComposite) (*EnvelopeParser, error) {
	if namespace == "" {
		return nil, fmt.Errorf("%w: namespace is empty", ErrInvalidParserConfig)
	}
	mode, err := canonicalizeInboundProcessingMode(cfg.InboundProcessingMode)
	if err != nil {
		return nil, fmt.Errorf("%w: %v", ErrInvalidParserConfig, err)
	}
	cfg.InboundProcessingMode = mode
	hasArrayPath := strings.TrimSpace(cfg.EventArrayPath) != ""
	hasObjectPath := strings.TrimSpace(cfg.EventObjectPath) != ""
	if hasArrayPath == hasObjectPath {
		return nil, fmt.Errorf("%w: exactly one of EventArrayPath or EventObjectPath must be set", ErrInvalidParserConfig)
	}
	if hasArrayPath {
		if _, err := parsePath(cfg.EventArrayPath); err != nil {
			return nil, fmt.Errorf("%w: EventArrayPath %q: %v", ErrInvalidParserConfig, cfg.EventArrayPath, err)
		}
	}
	if hasObjectPath {
		if _, err := parsePath(cfg.EventObjectPath); err != nil {
			return nil, fmt.Errorf("%w: EventObjectPath %q: %v", ErrInvalidParserConfig, cfg.EventObjectPath, err)
		}
	}
	if len(cfg.Rules) == 0 {
		return nil, fmt.Errorf("%w: Rules is empty (a parser with zero rules silently drops every delivery; declare at least one rule)", ErrInvalidParserConfig)
	}
	if len(ruleComposites) != 0 && len(ruleComposites) != len(cfg.Rules) {
		return nil, fmt.Errorf("%w: rule idempotency composite sidecar length %d must be zero or equal Rules length %d", ErrInvalidParserConfig, len(ruleComposites), len(cfg.Rules))
	}
	ruleComposites = append([]*IdempotencyKeyComposite(nil), ruleComposites...)
	compiledComposites := make([]*compiledIdempotencyKeyComposite, len(cfg.Rules))
	for i, r := range cfg.Rules {
		if strings.TrimSpace(r.ObjectKind) == "" {
			return nil, fmt.Errorf("%w: rule %d: ObjectKind is empty or whitespace-only (required for worker BatchRead + queue carriage)", ErrInvalidParserConfig, i)
		}
		if r.SourceIDPath == "" {
			return nil, fmt.Errorf("%w: rule %d: SourceIDPath is empty (required for worker BatchRead source ID carriage)", ErrInvalidParserConfig, i)
		}
		if _, err := parsePath(r.SourceIDPath); err != nil {
			return nil, fmt.Errorf("%w: rule %d SourceIDPath %q: %v", ErrInvalidParserConfig, i, r.SourceIDPath, err)
		}
		hasSourceLaneObjectKind := strings.TrimSpace(r.SourceLaneObjectKind) != ""
		hasSourceLaneIDPath := strings.TrimSpace(r.SourceLaneIDPath) != ""
		if hasSourceLaneObjectKind != hasSourceLaneIDPath {
			return nil, fmt.Errorf("%w: rule %d: SourceLaneObjectKind and SourceLaneIDPath must be configured together", ErrInvalidParserConfig, i)
		}
		if hasSourceLaneIDPath {
			if _, err := parsePath(r.SourceLaneIDPath); err != nil {
				return nil, fmt.Errorf("%w: rule %d SourceLaneIDPath %q: %v", ErrInvalidParserConfig, i, r.SourceLaneIDPath, err)
			}
		}
		var composite *IdempotencyKeyComposite
		if len(ruleComposites) != 0 {
			composite = ruleComposites[i]
		}
		hasLegacyIdempotencyPath := strings.TrimSpace(r.IdempotencyKeyPath) != ""
		if hasLegacyIdempotencyPath == (composite != nil) {
			return nil, fmt.Errorf("%w: rule %d: configure exactly one of IdempotencyKeyPath or IdempotencyKeyComposite", ErrInvalidParserConfig, i)
		}
		if composite != nil {
			canonical, compiled, err := compileIdempotencyKeyComposite(*composite)
			if err != nil {
				return nil, fmt.Errorf("%w: rule %d IdempotencyKeyComposite: %v", ErrInvalidParserConfig, i, err)
			}
			if compositeDegeneratesToPath(compiled, r.SourceIDPath) {
				return nil, fmt.Errorf("%w: rule %d: one-part IdempotencyKeyComposite must differ from SourceIDPath (using the object's source ID for event dedup folds distinct updates for the same object)", ErrInvalidParserConfig, i)
			}
			if r.EventTypeFromPath != "" && compositeDegeneratesToPath(compiled, r.EventTypeFromPath) {
				return nil, fmt.Errorf("%w: rule %d: one-part IdempotencyKeyComposite must differ from EventTypeFromPath (folds all events of the same type to one dedup row → silent loss of subsequent deliveries)", ErrInvalidParserConfig, i)
			}
			ruleComposites[i] = &canonical
			compiledComposites[i] = compiled
		} else {
			if _, err := parsePath(r.IdempotencyKeyPath); err != nil {
				return nil, fmt.Errorf("%w: rule %d IdempotencyKeyPath %q: %v", ErrInvalidParserConfig, i, r.IdempotencyKeyPath, err)
			}
			if strings.TrimSpace(r.IdempotencyKeyPath) == strings.TrimSpace(r.SourceIDPath) {
				return nil, fmt.Errorf("%w: rule %d: IdempotencyKeyPath %q must differ from SourceIDPath (using the object's source ID for event dedup folds distinct updates for the same object)", ErrInvalidParserConfig, i, r.IdempotencyKeyPath)
			}
			// IdempotencyKeyPath must also differ from EventTypeFromPath. Using the same
			// path for both collapses all events of the same type to a single dedup row.
			if r.EventTypeFromPath != "" && strings.TrimSpace(r.IdempotencyKeyPath) == strings.TrimSpace(r.EventTypeFromPath) {
				return nil, fmt.Errorf("%w: rule %d: IdempotencyKeyPath %q must differ from EventTypeFromPath (folds all events of the same type to one dedup row → silent loss of subsequent deliveries)", ErrInvalidParserConfig, i, r.IdempotencyKeyPath)
			}
		}
		seenPayloadNames := make(map[string]struct{})
		for j, pf := range r.PayloadProjection {
			// Empty PayloadField.Name silently emits a "" JSON key — reject at
			// construction so the pack author sees the error immediately.
			name := strings.TrimSpace(pf.Name)
			if name == "" {
				return nil, fmt.Errorf("%w: rule %d field %d: PayloadField.Name is empty", ErrInvalidParserConfig, i, j)
			}
			// Duplicate PayloadField.Name within one rule: JSON last-write-wins
			// silently overwrites the first value → data loss on every delivery.
			if _, dup := seenPayloadNames[pf.Name]; dup {
				return nil, fmt.Errorf("%w: rule %d field %d: PayloadField.Name %q is duplicated within the same rule (JSON parse would silently keep last value)", ErrInvalidParserConfig, i, j, pf.Name)
			}
			seenPayloadNames[pf.Name] = struct{}{}
			mode, err := canonicalizePayloadFieldMode(pf.Mode)
			if err != nil {
				return nil, fmt.Errorf("%w: rule %d field %d mode %q: %v", ErrInvalidParserConfig, i, j, pf.Mode, err)
			}
			cfg.Rules[i].PayloadProjection[j].Mode = mode
		}
		if r.EventTypeFromPath != "" && r.EventTypeLiteral != "" {
			return nil, fmt.Errorf("%w: rule %d: EventTypeFromPath and EventTypeLiteral are mutually exclusive", ErrInvalidParserConfig, i)
		}
		if r.EventTypeFromPath == "" && r.EventTypeLiteral == "" {
			return nil, fmt.Errorf("%w: rule %d: one of EventTypeFromPath or EventTypeLiteral must be set", ErrInvalidParserConfig, i)
		}
		// B1 — whitespace-only EventTypeLiteral: would route every matched event to a
		// blank eventKind, which the dispatcher cannot look up and the queue cannot route.
		// Fail closed at construction so the pack author sees the error immediately.
		if r.EventTypeLiteral != "" && strings.TrimSpace(r.EventTypeLiteral) == "" {
			return nil, fmt.Errorf("%w: rule %d: EventTypeLiteral is whitespace-only (%q) — would route to blank eventKind", ErrInvalidParserConfig, i, r.EventTypeLiteral)
		}
		// Canonicalize: trim surrounding whitespace from ObjectKind and
		// EventTypeLiteral so downstream emit uses trimmed values (consistent with
		// header-name canonicalization in R13/R14).
		cfg.Rules[i].ObjectKind = strings.TrimSpace(r.ObjectKind)
		cfg.Rules[i].SourceLaneObjectKind = strings.TrimSpace(r.SourceLaneObjectKind)
		cfg.Rules[i].EventTypeLiteral = strings.TrimSpace(r.EventTypeLiteral)
		if r.MatchPath != "" {
			if _, err := parsePath(r.MatchPath); err != nil {
				return nil, fmt.Errorf("%w: rule %d MatchPath %q: %v", ErrInvalidParserConfig, i, r.MatchPath, err)
			}
		}
		if r.MatchAll != nil && len(r.MatchAll) == 0 {
			return nil, fmt.Errorf("%w: rule %d: MatchAll must contain at least one condition when present", ErrInvalidParserConfig, i)
		}
		if len(r.MatchAll) > 0 {
			if strings.TrimSpace(r.MatchPath) != "" || r.MatchEquals != "" {
				return nil, fmt.Errorf("%w: rule %d: MatchAll is mutually exclusive with MatchPath/MatchEquals", ErrInvalidParserConfig, i)
			}
			seenPaths := make(map[string]struct{}, len(r.MatchAll))
			for j, cond := range r.MatchAll {
				if cond.Path == "" {
					return nil, fmt.Errorf("%w: rule %d MatchAll[%d]: Path is empty", ErrInvalidParserConfig, i, j)
				}
				path, err := canonicalPathKey(cond.Path)
				if err != nil {
					return nil, fmt.Errorf("%w: rule %d MatchAll[%d] Path %q: %v", ErrInvalidParserConfig, i, j, cond.Path, err)
				}
				if _, dup := seenPaths[path]; dup {
					return nil, fmt.Errorf("%w: rule %d MatchAll[%d] repeats Path %q; each MatchAll Path may appear at most once", ErrInvalidParserConfig, i, j, strings.TrimSpace(cond.Path))
				}
				seenPaths[path] = struct{}{}
			}
		}
		if r.EventTypeFromPath != "" {
			if _, err := parsePath(r.EventTypeFromPath); err != nil {
				return nil, fmt.Errorf("%w: rule %d EventTypeFromPath %q: %v", ErrInvalidParserConfig, i, r.EventTypeFromPath, err)
			}
		}
		for j, pf := range r.PayloadProjection {
			if _, err := parsePath(pf.Path); err != nil {
				return nil, fmt.Errorf("%w: rule %d field %d path %q: %v", ErrInvalidParserConfig, i, j, pf.Path, err)
			}
		}
	}
	// Post-rule-loop checks: catch-all ordering and duplicate selectors.
	//
	// Fix 5 — catch-all rule (MatchPath == "") must be last.
	// A catch-all at index K with more rules at K+1..N-1 makes those rules
	// unreachable: every event matches the catch-all first-match-wins, routing
	// silently to the catch-all ObjectKind/EventType instead of the intended one.
	catchAllIndex := -1
	for i, r := range cfg.Rules {
		if isCatchAllRule(r) {
			if catchAllIndex >= 0 {
				return nil, fmt.Errorf("%w: rule %d: at most one catch-all rule (MatchPath empty) allowed; rule %d is also a catch-all", ErrInvalidParserConfig, i, catchAllIndex)
			}
			catchAllIndex = i
		}
	}
	if catchAllIndex >= 0 && catchAllIndex != len(cfg.Rules)-1 {
		return nil, fmt.Errorf("%w: catch-all rule (MatchPath empty) at index %d must be the LAST rule (rules at index %d..%d are unreachable dead code)", ErrInvalidParserConfig, catchAllIndex, catchAllIndex+1, len(cfg.Rules)-1)
	}

	// Fix 6 — duplicate (MatchPath, MatchEquals) selector across rules.
	// First-match-wins makes the second occurrence unreachable: events that
	// match the selector always route to the first rule's ObjectKind/EventType,
	// silently ignoring the second rule entirely.
	type seenRuleSelector struct {
		index      int
		predicates []ruleSelector
	}
	seenSelectors := make(map[string]int)
	seenPredicateSelectors := make([]seenRuleSelector, 0, len(cfg.Rules))
	for i, r := range cfg.Rules {
		sel := selectorKey(r)
		if sel == "" {
			// catch-all handled above; skip dedup check (there can be at most one)
			continue
		}
		if first, dup := seenSelectors[sel]; dup {
			return nil, fmt.Errorf("%w: rule %d has the same selector as rule %d — first-match-wins makes rule %d unreachable", ErrInvalidParserConfig, i, first, i)
		}
		predicates := selectorPredicates(r)
		for _, previous := range seenPredicateSelectors {
			if selectorCovers(previous.predicates, predicates) {
				return nil, fmt.Errorf("%w: rule %d is shadowed by broader selector in rule %d — first-match-wins makes rule %d unreachable", ErrInvalidParserConfig, i, previous.index, i)
			}
		}
		seenSelectors[sel] = i
		seenPredicateSelectors = append(seenPredicateSelectors, seenRuleSelector{index: i, predicates: predicates})
	}

	return &EnvelopeParser{
		namespace:                        namespace,
		cfg:                              cfg,
		compiledRuleIdempotencyComposite: compiledComposites,
	}, nil
}

func compileIdempotencyKeyComposite(raw IdempotencyKeyComposite) (IdempotencyKeyComposite, *compiledIdempotencyKeyComposite, error) {
	version := strings.ToLower(strings.TrimSpace(raw.Version))
	if version != idempotencyCompositeVersionV1 {
		return IdempotencyKeyComposite{}, nil, fmt.Errorf("Version %q is not supported", raw.Version)
	}
	encoding := strings.ToLower(strings.TrimSpace(raw.Encoding))
	if encoding != idempotencyCompositeEncodingSHA256 {
		return IdempotencyKeyComposite{}, nil, fmt.Errorf("Encoding %q is not supported", raw.Encoding)
	}
	if len(raw.Parts) == 0 {
		return IdempotencyKeyComposite{}, nil, fmt.Errorf("Parts is empty")
	}

	canonical := IdempotencyKeyComposite{
		Version:  version,
		Encoding: encoding,
		Parts:    make([]IdempotencyKeyPart, 0, len(raw.Parts)),
	}
	compiled := &compiledIdempotencyKeyComposite{
		version:        version,
		encoding:       encoding,
		extractPaths:   make([]string, 0, len(raw.Parts)),
		canonicalPaths: make([]string, 0, len(raw.Parts)),
	}
	seenPaths := make(map[string]struct{}, len(raw.Parts))
	for i, part := range raw.Parts {
		path := strings.TrimSpace(part.Path)
		if path == "" {
			return IdempotencyKeyComposite{}, nil, fmt.Errorf("Parts[%d].Path is empty", i)
		}
		canonicalPath, err := canonicalIdempotencyPath(path)
		if err != nil {
			return IdempotencyKeyComposite{}, nil, fmt.Errorf("Parts[%d].Path %q: %v", i, path, err)
		}
		if _, duplicate := seenPaths[canonicalPath]; duplicate {
			return IdempotencyKeyComposite{}, nil, fmt.Errorf("Parts[%d].Path %q duplicates an earlier canonical path", i, path)
		}
		seenPaths[canonicalPath] = struct{}{}
		canonical.Parts = append(canonical.Parts, IdempotencyKeyPart{Path: path})
		compiled.extractPaths = append(compiled.extractPaths, path)
		compiled.canonicalPaths = append(compiled.canonicalPaths, canonicalPath)
	}

	var spec bytes.Buffer
	writeIdempotencyFrame(&spec, []byte("nexuscrm.connector.idempotency.composite.spec"))
	writeIdempotencyFrame(&spec, []byte(compiled.version))
	writeIdempotencyFrame(&spec, []byte(compiled.encoding))
	writeIdempotencyCount(&spec, len(compiled.canonicalPaths))
	for _, path := range compiled.canonicalPaths {
		writeIdempotencyFrame(&spec, []byte(path))
	}
	sum := sha256.Sum256(spec.Bytes())
	compiled.specHash = hex.EncodeToString(sum[:])[:idempotencyCompositeSpecHashHexSize]
	return canonical, compiled, nil
}

func compositeDegeneratesToPath(composite *compiledIdempotencyKeyComposite, path string) bool {
	if composite == nil || len(composite.canonicalPaths) != 1 {
		return false
	}
	canonicalPath, err := canonicalIdempotencyPath(path)
	return err == nil && composite.canonicalPaths[0] == canonicalPath
}

func canonicalIdempotencyPath(path string) (string, error) {
	segments, err := parsePath(path)
	if err != nil {
		return "", err
	}
	var framed bytes.Buffer
	writeIdempotencyCount(&framed, len(segments))
	for _, segment := range segments {
		if segment.isIndex {
			writeIdempotencyFrame(&framed, []byte("index"))
			writeIdempotencyFrame(&framed, []byte(strconv.Itoa(segment.index)))
			continue
		}
		writeIdempotencyFrame(&framed, []byte("key"))
		writeIdempotencyFrame(&framed, []byte(segment.key))
	}
	return framed.String(), nil
}

type ruleSelector struct{ path, equals string }

func isCatchAllRule(r EnvelopeMappingRule) bool {
	return r.MatchAll == nil && strings.TrimSpace(r.MatchPath) == ""
}

func selectorKey(r EnvelopeMappingRule) string {
	if r.MatchAll == nil {
		path, _ := canonicalPathKey(r.MatchPath)
		if path == "" {
			return ""
		}
		return "single\x00" + path + "\x00" + r.MatchEquals
	}
	parts := make([]string, 0, len(r.MatchAll))
	for _, cond := range r.MatchAll {
		path, _ := canonicalPathKey(cond.Path)
		parts = append(parts, path+"\x00"+cond.Equals)
	}
	sort.Strings(parts)
	if len(parts) == 1 {
		return "single\x00" + parts[0]
	}
	return "all\x00" + strings.Join(parts, "\x1e")
}

func selectorPredicates(r EnvelopeMappingRule) []ruleSelector {
	if r.MatchAll == nil {
		path, _ := canonicalPathKey(r.MatchPath)
		if path == "" {
			return nil
		}
		return []ruleSelector{{path: path, equals: r.MatchEquals}}
	}
	predicates := make([]ruleSelector, 0, len(r.MatchAll))
	for _, cond := range r.MatchAll {
		path, _ := canonicalPathKey(cond.Path)
		predicates = append(predicates, ruleSelector{path: path, equals: cond.Equals})
	}
	return predicates
}

func selectorCovers(previous, current []ruleSelector) bool {
	if len(previous) == 0 || len(previous) > len(current) {
		return false
	}
	currentByPath := make(map[string]string, len(current))
	for _, predicate := range current {
		currentByPath[predicate.path] = predicate.equals
	}
	for _, predicate := range previous {
		currentEquals, exists := currentByPath[predicate.path]
		if !exists {
			return false
		}
		if predicate.equals != "" && predicate.equals != currentEquals {
			return false
		}
	}
	return true
}

func canonicalPathKey(path string) (string, error) {
	segs, err := parsePath(path)
	if err != nil {
		return "", err
	}
	if len(segs) == 0 {
		return ".", nil
	}
	parts := make([]string, 0, len(segs))
	for _, seg := range segs {
		if seg.isIndex {
			parts = append(parts, "i:"+strconv.Itoa(seg.index))
			continue
		}
		parts = append(parts, "k:"+seg.key)
	}
	return strings.Join(parts, "\x1f"), nil
}

func canonicalizeInboundProcessingMode(raw InboundProcessingMode) (InboundProcessingMode, error) {
	mode := InboundProcessingMode(strings.ToLower(strings.TrimSpace(string(raw))))
	if mode == "" {
		return "", nil
	}
	switch mode {
	case InboundProcessingModeEnrichmentRequired, InboundProcessingModePayloadOnlyAllowed:
		return mode, nil
	default:
		return "", fmt.Errorf("InboundProcessingMode %q is not one of {\"%s\", \"%s\"}", raw, InboundProcessingModeEnrichmentRequired, InboundProcessingModePayloadOnlyAllowed)
	}
}

// Parse implements [connectorcapabilities.EnvelopeParserHandler].
//
// Returns [connectorcapabilities.ErrMalformedPayload] when:
//   - input.Payload is not valid JSON.
//   - The configured event item source is absent or has the wrong JSON shape.
//   - Any matched item's IdempotencyKeyPath resolves to an absent or empty value.
//
// Unmatched items (no rule applies) are skipped — not an error. Each skip
// increments the nexuscrm_connector_inbound_envelope_items_skipped_total counter
// (labeled by namespace) and emits a server-side log line (issue #2127).
//
// Each emitted event gets an IdempotencyKey from exactly one configured
// identity source: the legacy scalar path or a versioned ordered composite.
// Missing/empty components are rejected to prevent silently folding distinct
// events into one queue row.
func (p *EnvelopeParser) Parse(_ context.Context, input *connectorcapabilities.ParserInput) ([]connectorcapabilities.ConnectorEvent, error) {
	payload := input.Payload
	if !json.Valid(payload) {
		return nil, fmt.Errorf("invalid JSON: %w", connectorcapabilities.ErrMalformedPayload)
	}

	items, err := p.extractEventItems(payload)
	if err != nil {
		return nil, err
	}

	var out []connectorcapabilities.ConnectorEvent
	for _, item := range items {
		ruleIndex, rule, ok := matchRule(p.cfg.Rules, item)
		if !ok {
			connectorInboundItemsSkippedTotal.WithLabelValues(p.namespace).Inc()
			log.Printf("connector_inbound: envelope item skipped namespace=%q (no matching rule)", p.namespace)
			continue
		}
		ev, err := emitEvent(rule, p.compiledRuleIdempotencyComposite[ruleIndex], item, p.namespace)
		if err != nil {
			// Preserve the sentinel from emitEvent:
			//   composite at typed field → ErrMalformedPayload (400, client-malformed)
			//   absent/null/empty at typed field → ErrParserConfigDrift (503, config-drift)
			// Do NOT blanket-wrap into ErrMalformedPayload — that would mis-classify
			// config-drift as a terminal client error and silently drop valid deliveries.
			return nil, fmt.Errorf("emit event for rule: %w", err)
		}
		out = append(out, ev)
	}
	return out, nil
}

func (p *EnvelopeParser) extractEventItems(payload json.RawMessage) ([]json.RawMessage, error) {
	if strings.TrimSpace(p.cfg.EventArrayPath) != "" {
		items, err := extractArray(p.cfg.EventArrayPath, payload)
		if err != nil {
			return nil, fmt.Errorf("extract event array at %q: %w: %v",
				p.cfg.EventArrayPath, connectorcapabilities.ErrMalformedPayload, err)
		}
		if items == nil {
			return nil, fmt.Errorf("event array missing or null at %q: %w",
				p.cfg.EventArrayPath, connectorcapabilities.ErrMalformedPayload)
		}
		return items, nil
	}

	item, err := extractObject(p.cfg.EventObjectPath, payload)
	if err != nil {
		return nil, fmt.Errorf("extract event object at %q: %w: %v",
			p.cfg.EventObjectPath, connectorcapabilities.ErrMalformedPayload, err)
	}
	if item == nil {
		return nil, fmt.Errorf("event object missing or null at %q: %w",
			p.cfg.EventObjectPath, connectorcapabilities.ErrMalformedPayload)
	}
	return []json.RawMessage{item}, nil
}

// matchRule returns the index and value of the first rule that matches item.
func matchRule(rules []EnvelopeMappingRule, item json.RawMessage) (int, EnvelopeMappingRule, bool) {
	for i, r := range rules {
		if len(r.MatchAll) > 0 {
			if matchAll(r.MatchAll, item) {
				return i, r, true
			}
			continue
		}
		if r.MatchAll == nil && r.MatchPath == "" {
			// Catch-all rule.
			return i, r, true
		}
		if matchCondition(MatchCondition{Path: r.MatchPath, Equals: r.MatchEquals}, item) {
			return i, r, true
		}
	}
	return -1, EnvelopeMappingRule{}, false
}

func matchAll(conditions []MatchCondition, item json.RawMessage) bool {
	for _, cond := range conditions {
		if !matchCondition(cond, item) {
			return false
		}
	}
	return true
}

func matchCondition(cond MatchCondition, item json.RawMessage) bool {
	// lenient: object/array at a match path is a non-match, not a
	// malformed-payload error — only typed fields (emitEvent) use the strict
	// extractor.
	val, err := extractScalarString(cond.Path, item)
	if err != nil || val == "" {
		return false
	}
	return cond.Equals == "" || val == cond.Equals
}

// emitEvent assembles a [connectorcapabilities.ConnectorEvent] from rule and item.
//
// The emitted ConnectorEvent carries ObjectKind and SourceID as typed fields.
// The Payload bytes contain only the PayloadProjection fields (no reserved-key
// injection). The dispatcher wraps ObjectKind, SourceID, and Payload into a
// connector_inbound.QueueEnvelope before enqueue; the worker decodes that envelope.
func emitEvent(rule EnvelopeMappingRule, composite *compiledIdempotencyKeyComposite, item json.RawMessage, namespace string) (connectorcapabilities.ConnectorEvent, error) {
	var eventType string
	if rule.EventTypeLiteral != "" {
		// EventTypeLiteral was already TrimSpace'd at construction time (B1 canonicalization).
		eventType = rule.EventTypeLiteral
	} else {
		et, err := extractScalarStrict(rule.EventTypeFromPath, item)
		if err != nil {
			// extractScalarStrict returns ErrMalformedPayload for composite values
			// (object/array at a scalar path — client sent wrong shape → 400). Propagate as-is.
			return connectorcapabilities.ConnectorEvent{}, fmt.Errorf("EventTypeFromPath %q: %w", rule.EventTypeFromPath, err)
		}
		eventType = et
	}
	// B1 parse-time guard: TrimSpace then reject empty. This catches:
	//   (a) EventTypeFromPath resolving to "" or whitespace-only strings (absent/null/empty path
	//       → CONFIG-DRIFT, not client malformed → ErrParserConfigDrift so dispatcher returns 503).
	//   (b) Defense-in-depth if a literal somehow slips through (e.g. future
	//       code paths that bypass the construction-time canonicalization).
	eventType = strings.TrimSpace(eventType)
	if eventType == "" {
		return connectorcapabilities.ConnectorEvent{}, fmt.Errorf("EventTypeFromPath %q resolved to absent/null/empty (likely pack mis-configured path): %w", rule.EventTypeFromPath, connectorcapabilities.ErrParserConfigDrift)
	}

	ns := ""
	if rule.NamespaceFromCtx {
		ns = namespace
	}

	sourceID, err := extractScalarStrict(rule.SourceIDPath, item)
	if err != nil {
		// Composite at SourceIDPath → ErrMalformedPayload (client sent object/array at scalar path → 400).
		return connectorcapabilities.ConnectorEvent{}, fmt.Errorf("SourceIDPath %q: %w", rule.SourceIDPath, err)
	}
	if sourceID == "" {
		// Absent/null/empty → CONFIG-DRIFT: path likely mis-configured → ErrParserConfigDrift → 503.
		return connectorcapabilities.ConnectorEvent{}, fmt.Errorf("SourceIDPath %q resolved to absent/null/empty (likely pack mis-configured path): %w", rule.SourceIDPath, connectorcapabilities.ErrParserConfigDrift)
	}

	sourceLaneObjectKind := rule.SourceLaneObjectKind
	sourceLaneIDPath := rule.SourceLaneIDPath
	sourceLaneID := ""
	if sourceLaneObjectKind != "" {
		laneID, err := extractScalarStrict(sourceLaneIDPath, item)
		if err != nil {
			return connectorcapabilities.ConnectorEvent{}, fmt.Errorf("SourceLaneIDPath %q: %w", sourceLaneIDPath, err)
		}
		sourceLaneID = laneID
		if strings.TrimSpace(sourceLaneID) == "" {
			return connectorcapabilities.ConnectorEvent{}, fmt.Errorf("SourceLaneIDPath %q resolved to absent/null/empty (likely pack mis-configured path): %w", sourceLaneIDPath, connectorcapabilities.ErrParserConfigDrift)
		}
	}

	idempotencyKey, err := resolveIdempotencyKey(rule.IdempotencyKeyPath, composite, item)
	if err != nil {
		return connectorcapabilities.ConnectorEvent{}, err
	}

	payload, err := buildPayload(rule.PayloadProjection, item)
	if err != nil {
		return connectorcapabilities.ConnectorEvent{}, fmt.Errorf("build payload: %w", err)
	}

	return connectorcapabilities.ConnectorEvent{
		EventType:            eventType,
		Namespace:            ns,
		ObjectKind:           rule.ObjectKind,
		SourceID:             sourceID,
		SourceLaneObjectKind: sourceLaneObjectKind,
		SourceLaneID:         sourceLaneID,
		Payload:              payload,
		IdempotencyKey:       idempotencyKey,
	}, nil
}

func resolveIdempotencyKey(legacyPath string, composite *compiledIdempotencyKeyComposite, item json.RawMessage) (string, error) {
	if composite == nil {
		key, err := extractScalarStrict(legacyPath, item)
		if err != nil {
			return "", fmt.Errorf("IdempotencyKeyPath %q: %w", legacyPath, err)
		}
		if key == "" {
			return "", fmt.Errorf("IdempotencyKeyPath %q resolved to absent/null/empty (likely pack mis-configured path): %w", legacyPath, connectorcapabilities.ErrParserConfigDrift)
		}
		return key, nil
	}

	var preimage bytes.Buffer
	writeIdempotencyFrame(&preimage, []byte("nexuscrm.connector.idempotency.composite.value"))
	writeIdempotencyFrame(&preimage, []byte(composite.version))
	writeIdempotencyFrame(&preimage, []byte(composite.encoding))
	writeIdempotencyCount(&preimage, len(composite.canonicalPaths))
	for i, path := range composite.extractPaths {
		typeTag, value, err := extractIdempotencyComponent(path, item)
		if err != nil {
			// Never include the component value in errors or logs.
			return "", fmt.Errorf("IdempotencyKeyComposite.Parts[%d] path %q: %w", i, path, err)
		}
		writeIdempotencyFrame(&preimage, []byte(composite.canonicalPaths[i]))
		writeIdempotencyFrame(&preimage, []byte(typeTag))
		writeIdempotencyFrame(&preimage, value)
	}
	sum := sha256.Sum256(preimage.Bytes())
	return idempotencyCompositeKeyPrefix + composite.specHash + ":" + hex.EncodeToString(sum[:]), nil
}

// extractIdempotencyComponent returns a typed, deterministic scalar frame.
// Strings are decoded to UTF-8, booleans use true/false, and numbers are reduced
// to sign + zero-free coefficient + decimal exponent. Thus 1/1.0/1e0 and -0/0
// share an identity without floating-point precision loss or exponent expansion.
func extractIdempotencyComponent(path string, item json.RawMessage) (string, []byte, error) {
	raw, err := extractRaw(path, item)
	if err != nil {
		return "", nil, err
	}
	trimmed := bytes.TrimSpace(raw)
	if len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) {
		return "", nil, fmt.Errorf("resolved to absent/null/empty: %w", connectorcapabilities.ErrParserConfigDrift)
	}
	if trimmed[0] == '{' || trimmed[0] == '[' {
		kind := map[byte]string{'{': "object", '[': "array"}[trimmed[0]]
		return "", nil, fmt.Errorf("%w: resolved to JSON %s, expected scalar", connectorcapabilities.ErrMalformedPayload, kind)
	}
	if trimmed[0] == '"' {
		var value string
		if err := json.Unmarshal(trimmed, &value); err != nil {
			return "", nil, fmt.Errorf("%w: invalid JSON string component", connectorcapabilities.ErrMalformedPayload)
		}
		if strings.TrimSpace(value) == "" {
			return "", nil, fmt.Errorf("resolved to empty string: %w", connectorcapabilities.ErrParserConfigDrift)
		}
		return "string", []byte(value), nil
	}
	if bytes.Equal(trimmed, []byte("true")) || bytes.Equal(trimmed, []byte("false")) {
		return "bool", append([]byte(nil), trimmed...), nil
	}
	var number json.Number
	if err := json.Unmarshal(trimmed, &number); err != nil {
		return "", nil, fmt.Errorf("%w: resolved to unsupported scalar type", connectorcapabilities.ErrMalformedPayload)
	}
	canonicalNumber, ok := canonicalizeIdempotencyNumber(number.String())
	if !ok {
		return "", nil, fmt.Errorf("%w: resolved to invalid JSON number", connectorcapabilities.ErrMalformedPayload)
	}
	return "number", []byte(canonicalNumber), nil
}

func canonicalizeIdempotencyNumber(token string) (string, bool) {
	negative := strings.HasPrefix(token, "-")
	if negative {
		token = token[1:]
	}

	exponent := new(big.Int)
	if index := strings.IndexAny(token, "eE"); index >= 0 {
		exponentToken := token[index+1:]
		if strings.HasPrefix(exponentToken, "+") {
			exponentToken = exponentToken[1:]
		}
		if _, ok := exponent.SetString(exponentToken, 10); !ok {
			return "", false
		}
		token = token[:index]
	}

	fractionDigits := 0
	if dot := strings.IndexByte(token, '.'); dot >= 0 {
		fractionDigits = len(token) - dot - 1
		token = token[:dot] + token[dot+1:]
	}
	digits := strings.TrimLeft(token, "0")
	if digits == "" {
		return "0", true
	}
	trailingZeros := len(digits) - len(strings.TrimRight(digits, "0"))
	digits = strings.TrimRight(digits, "0")
	exponent.Sub(exponent, big.NewInt(int64(fractionDigits)))
	exponent.Add(exponent, big.NewInt(int64(trailingZeros)))
	if negative {
		digits = "-" + digits
	}
	return digits + "e" + exponent.String(), true
}

func writeIdempotencyFrame(buf *bytes.Buffer, value []byte) {
	var size [4]byte
	binary.BigEndian.PutUint32(size[:], uint32(len(value)))
	buf.Write(size[:])
	buf.Write(value)
}

func writeIdempotencyCount(buf *bytes.Buffer, count int) {
	var encoded [4]byte
	binary.BigEndian.PutUint32(encoded[:], uint32(count))
	buf.Write(encoded[:])
}

// (The old errEmptyEventType, errMalformedSourceID, errMalformedIdempotencyKey
// sentinels have been removed. Absent/null/empty typed-field paths now wrap
// connectorcapabilities.ErrParserConfigDrift directly in emitEvent so the
// dispatcher maps them to a retriable 503+log instead of the old terminal 400.)

// buildPayload assembles the ordered payload JSON object from projection fields.
// The returned bytes contain only the PayloadProjection fields — no reserved-key
// injection. ObjectKind and SourceID are carried as typed fields on ConnectorEvent
// and wrapped into a QueueEnvelope by the dispatcher before Enqueue.
func buildPayload(projection []PayloadField, item json.RawMessage) ([]byte, error) {
	if len(projection) == 0 {
		return []byte("{}"), nil
	}
	var buf bytes.Buffer
	buf.WriteByte('{')
	for i, pf := range projection {
		if i > 0 {
			buf.WriteByte(',')
		}
		keyBytes, err := json.Marshal(pf.Name)
		if err != nil {
			return nil, fmt.Errorf("marshal key %q: %w", pf.Name, err)
		}
		buf.Write(keyBytes)
		buf.WriteByte(':')
		switch pf.Mode {
		case PayloadFieldModeRaw:
			val, err := extractRaw(pf.Path, item)
			if err != nil || val == nil {
				buf.WriteString("null")
			} else {
				buf.Write(val)
			}
		case PayloadFieldModeScalarString:
			val, present, err := extractScalarStringPayload(pf.Path, item)
			if err != nil {
				return nil, err
			}
			if !present {
				buf.WriteString("null")
			} else {
				encoded, err := json.Marshal(val)
				if err != nil {
					return nil, fmt.Errorf("marshal scalar_string payload %q: %w", pf.Name, err)
				}
				buf.Write(encoded)
			}
		case PayloadFieldModeScalarStringArray:
			vals, present, err := extractScalarStringArrayPayload(pf.Path, item)
			if err != nil {
				return nil, err
			}
			if !present {
				buf.WriteString("null")
			} else {
				encoded, err := json.Marshal(vals)
				if err != nil {
					return nil, fmt.Errorf("marshal scalar_string_array payload %q: %w", pf.Name, err)
				}
				buf.Write(encoded)
			}
		default:
			return nil, fmt.Errorf("payload field %q uses unsupported mode %q", pf.Name, pf.Mode)
		}
	}
	buf.WriteByte('}')
	return buf.Bytes(), nil
}

func canonicalizePayloadFieldMode(raw PayloadFieldMode) (PayloadFieldMode, error) {
	mode := PayloadFieldMode(strings.ToLower(strings.TrimSpace(string(raw))))
	switch mode {
	case PayloadFieldModeRaw, PayloadFieldModeScalarString, PayloadFieldModeScalarStringArray:
		return mode, nil
	default:
		return "", fmt.Errorf("must be one of {\"%s\", \"%s\"} when set", PayloadFieldModeScalarString, PayloadFieldModeScalarStringArray)
	}
}

func extractScalarStringPayload(path string, data json.RawMessage) (string, bool, error) {
	raw, err := extractRaw(path, data)
	if err != nil {
		return "", false, nil
	}
	return scalarStringFromRaw(path, raw, true)
}

func extractScalarStringArrayPayload(path string, data json.RawMessage) ([]string, bool, error) {
	raw, err := extractRaw(path, data)
	if err != nil || raw == nil || string(raw) == "null" {
		return nil, false, nil
	}
	trimmed := bytes.TrimSpace(raw)
	if len(trimmed) == 0 || trimmed[0] != '[' {
		return nil, true, fmt.Errorf("%w: value at %q is not a JSON array, expected an array of scalar values",
			connectorcapabilities.ErrMalformedPayload, path)
	}
	var arr []json.RawMessage
	if err := json.Unmarshal(raw, &arr); err != nil {
		return nil, true, fmt.Errorf("%w: value at %q is not a valid JSON array: %v",
			connectorcapabilities.ErrMalformedPayload, path, err)
	}
	out := make([]string, 0, len(arr))
	for i, elem := range arr {
		elemPath := fmt.Sprintf("%s[%d]", path, i)
		val, present, err := scalarStringFromRaw(elemPath, elem, false)
		if err != nil {
			return nil, true, err
		}
		if !present {
			return nil, true, fmt.Errorf("%w: value at %q is null, expected a scalar value",
				connectorcapabilities.ErrMalformedPayload, elemPath)
		}
		out = append(out, val)
	}
	return out, true, nil
}

// ============================================================================
// JSON-path resolver — minimal subset: ., .foo, .foo.bar, .foo[N]
// ============================================================================

// pathSegment is a parsed segment of a JSON-path.
type pathSegment struct {
	key     string // non-empty → field access
	index   int    // used when isIndex == true
	isIndex bool
}

// parsePath parses a JSON-path string into segments. Supports:
//   - "." — root (empty segments)
//   - ".foo" — single field
//   - ".foo.bar" — nested fields
//   - ".foo[N]" — array index
//
// Returns an error on syntax violations.
func parsePath(path string) ([]pathSegment, error) {
	if path == "" {
		return nil, fmt.Errorf("path is empty")
	}
	if path == "." {
		return nil, nil // root
	}
	if !strings.HasPrefix(path, ".") {
		return nil, fmt.Errorf("path must start with '.'")
	}
	rest := path[1:] // drop leading dot
	var segs []pathSegment
	for rest != "" {
		// Array index?
		if strings.HasPrefix(rest, "[") {
			end := strings.Index(rest, "]")
			if end < 0 {
				return nil, fmt.Errorf("unclosed '[' in path %q", path)
			}
			idxStr := rest[1:end]
			idx, err := strconv.Atoi(idxStr)
			if err != nil {
				return nil, fmt.Errorf("non-integer index %q in path %q", idxStr, path)
			}
			segs = append(segs, pathSegment{index: idx, isIndex: true})
			rest = rest[end+1:]
			continue
		}
		// Field name — read up to next '.' or '['.
		end := strings.IndexAny(rest, ".[")
		var field string
		if end < 0 {
			field = rest
			rest = ""
		} else {
			field = rest[:end]
			rest = rest[end:]
			if strings.HasPrefix(rest, ".") {
				rest = rest[1:] // consume separator dot
			}
		}
		if field == "" {
			return nil, fmt.Errorf("empty field name in path %q", path)
		}
		segs = append(segs, pathSegment{key: field})
	}
	return segs, nil
}

// extractRaw resolves path against data and returns the raw JSON value, or
// (nil, nil) when the path is absent. Returns (nil, error) on syntax error.
func extractRaw(path string, data json.RawMessage) (json.RawMessage, error) {
	segs, err := parsePath(path)
	if err != nil {
		return nil, err
	}
	return resolveSegments(segs, data)
}

// resolveSegments walks segments over current JSON value.
func resolveSegments(segs []pathSegment, cur json.RawMessage) (json.RawMessage, error) {
	if len(segs) == 0 {
		return cur, nil
	}
	seg := segs[0]
	rest := segs[1:]

	if seg.isIndex {
		// cur must be a JSON array.
		var arr []json.RawMessage
		if err := json.Unmarshal(cur, &arr); err != nil {
			return nil, nil // not an array — path absent
		}
		if seg.index < 0 || seg.index >= len(arr) {
			return nil, nil // out of bounds — path absent
		}
		return resolveSegments(rest, arr[seg.index])
	}

	// Field access — cur must be a JSON object.
	var obj map[string]json.RawMessage
	if err := json.Unmarshal(cur, &obj); err != nil {
		return nil, nil // not an object — path absent
	}
	val, ok := obj[seg.key]
	if !ok {
		return nil, nil // field absent
	}
	return resolveSegments(rest, val)
}

// extractArray resolves path against data and returns it as a slice of raw
// JSON items. Returns (nil, nil) when the path resolves to null or the value
// is absent. Returns an error when the resolved value is not a JSON array.
func extractArray(path string, data json.RawMessage) ([]json.RawMessage, error) {
	raw, err := extractRaw(path, data)
	if err != nil {
		return nil, err
	}
	if raw == nil || string(raw) == "null" {
		return nil, nil
	}
	var arr []json.RawMessage
	if err := json.Unmarshal(raw, &arr); err != nil {
		return nil, fmt.Errorf("value at %q is not a JSON array: %w", path, err)
	}
	return arr, nil
}

// extractObject resolves path against data and returns it as one raw JSON
// object. Returns nil when the path resolves to null or the value is absent.
// Returns an error when the resolved value is not a JSON object.
func extractObject(path string, data json.RawMessage) (json.RawMessage, error) {
	raw, err := extractRaw(path, data)
	if err != nil {
		return nil, err
	}
	if raw == nil || string(raw) == "null" {
		return nil, nil
	}
	trimmed := bytes.TrimSpace(raw)
	if len(trimmed) == 0 || trimmed[0] != '{' {
		return nil, fmt.Errorf("value at %q is not a JSON object", path)
	}
	var obj map[string]json.RawMessage
	if err := json.Unmarshal(trimmed, &obj); err != nil {
		return nil, fmt.Errorf("value at %q is not a JSON object: %w", path, err)
	}
	return json.RawMessage(trimmed), nil
}

// extractScalarString resolves path against data and returns the JSON string
// value as a Go string. Returns "" when the path is absent, null, or not a
// string. Non-string scalars (numbers, bools) are coerced to their raw string
// representation so MatchEquals comparisons against numeric event IDs work.
//
// Composite JSON values (object/array — leading byte '{' or '[') return ""
// instead of being stringified. A composite at a match-selector path means
// the item value is structurally mismatched for this rule — treating it as a
// non-match (return "") rather than stringifying it prevents phantom matches
// where a MatchPath-only rule (MatchEquals=="") incorrectly selects an item
// whose MatchPath field is an object/array rather than a scalar.
//
// This is the lenient extractor. Use it ONLY for match-selector paths (MatchPath)
// where a composite at the path is simply a non-match, not a malformed-payload
// error. For typed queue fields (EventType, SourceID, IdempotencyKey) use
// extractScalarStrict, which rejects composite JSON values with ErrMalformedPayload.
func extractScalarString(path string, data json.RawMessage) (string, error) {
	raw, err := extractRaw(path, data)
	if err != nil {
		return "", err
	}
	if raw == nil || string(raw) == "null" {
		return "", nil
	}
	// Composite JSON (object/array) at a match-selector path is a non-match, not
	// a value to stringify. Return "" so the rule is skipped cleanly.
	trimmed := bytes.TrimSpace(raw)
	if len(trimmed) > 0 && (trimmed[0] == '{' || trimmed[0] == '[') {
		return "", nil
	}
	var s string
	if err := json.Unmarshal(raw, &s); err != nil {
		// Value exists but is not a JSON string and not composite — number or bool.
		// Return its raw string representation so MatchEquals comparisons against
		// numeric event IDs still work.
		return strings.Trim(string(raw), `"`), nil
	}
	return s, nil
}

// extractScalarStrict resolves path and returns a scalar string, like
// extractScalarString, but REJECTS composite JSON (object/array) with
// ErrMalformedPayload instead of coercing it to a bogus string. Used for the
// typed fields that flow into the queue row (EventType, SourceID,
// IdempotencyKey) — a non-scalar there is a malformed delivery, not a value to
// stringify. JSON string/number/bool are all accepted (number/bool stringified,
// preserving the numeric-ID behavior). null/absent returns "".
func extractScalarStrict(path string, data json.RawMessage) (string, error) {
	raw, err := extractRaw(path, data)
	if err != nil {
		return "", err
	}
	s, _, err := scalarStringFromRaw(path, raw, true)
	return s, err
}

func scalarStringFromRaw(path string, raw json.RawMessage, allowNull bool) (string, bool, error) {
	if raw == nil || string(raw) == "null" {
		if allowNull {
			return "", false, nil
		}
		return "", false, fmt.Errorf("%w: value at %q is null, expected a scalar (string/number/bool)",
			connectorcapabilities.ErrMalformedPayload, path)
	}
	trimmed := bytes.TrimSpace(raw)
	if len(trimmed) > 0 && (trimmed[0] == '{' || trimmed[0] == '[') {
		kind := map[byte]string{'{': "object", '[': "array"}[trimmed[0]]
		return "", true, fmt.Errorf("%w: value at %q is a JSON %s, expected a scalar (string/number/bool)",
			connectorcapabilities.ErrMalformedPayload, path, kind)
	}
	var s string
	if err := json.Unmarshal(raw, &s); err != nil {
		return strings.Trim(string(trimmed), `"`), true, nil
	}
	return s, true, nil
}