package connector_inbound

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"testing"

	"github.com/prometheus/client_golang/prometheus/testutil"

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

// ── __test_noop__ 4-rule fixture config ──────────────────────────────────────
//
// This is the canonical __test_noop__ fixture mirroring the spike.md DSL.
// It exercises all 4 event types: created / updated / deleted / merge.
//
// The fixture is also used by noop_fixture_test.go for the end-to-end path.

const noopParserNamespace = "__test_noop__"

// noopParserConfig returns the 4-rule EnvelopeMappingConfig used by the
// __test_noop__ fixture. Exported as a package-internal helper for
// noop_fixture_test.go to share without duplication.
func noopParserConfig() EnvelopeMappingConfig {
	return EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchPath:          ".eventType",
				MatchEquals:        "noop.created",
				EventTypeLiteral:   "noop.created",
				ObjectKind:         "noop_object",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".deliveryId",
				NamespaceFromCtx:   true,
				PayloadProjection: []PayloadField{
					{Name: "objectId", Path: ".objectId"},
					{Name: "occurredAt", Path: ".occurredAt"},
				},
			},
			{
				MatchPath:          ".eventType",
				MatchEquals:        "noop.updated",
				EventTypeLiteral:   "noop.updated",
				ObjectKind:         "noop_object",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".deliveryId",
				NamespaceFromCtx:   true,
				PayloadProjection: []PayloadField{
					{Name: "objectId", Path: ".objectId"},
					{Name: "changedField", Path: ".changedField"},
				},
			},
			{
				MatchPath:          ".eventType",
				MatchEquals:        "noop.deleted",
				EventTypeLiteral:   "noop.deleted",
				ObjectKind:         "noop_object",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".deliveryId",
				NamespaceFromCtx:   true,
				PayloadProjection: []PayloadField{
					{Name: "objectId", Path: ".objectId"},
				},
			},
			{
				MatchPath:          ".eventType",
				MatchEquals:        "noop.merge",
				EventTypeLiteral:   "noop.merge",
				ObjectKind:         "noop_object",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".deliveryId",
				NamespaceFromCtx:   true,
				PayloadProjection: []PayloadField{
					{Name: "objectId", Path: ".objectId"},
					{Name: "mergedIds", Path: ".mergedIds"},
				},
			},
		},
	}
}

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

func mustNewParser(t *testing.T, ns string, cfg EnvelopeMappingConfig) *EnvelopeParser {
	t.Helper()
	p, err := NewEnvelopeParser(ns, cfg)
	if err != nil {
		t.Fatalf("NewEnvelopeParser: %v", err)
	}
	return p
}

func TestEnvelopeMappingRule_AllowsPreTargetLaneUnkeyedLiteralShape(t *testing.T) {
	rule := EnvelopeMappingRule{
		"",
		"",
		nil,
		"",
		"thing.updated",
		true,
		"thing",
		".objectId",
		"",
		"",
		".eventId",
		nil,
	}
	if rule.EventTypeLiteral != "thing.updated" || rule.IdempotencyKeyPath != ".eventId" {
		t.Fatalf("unkeyed EnvelopeMappingRule layout shifted: %+v", rule)
	}
}

// payloadDoesNotContainReservedKeys asserts that the payload does not inject
// __objectKind or __sourceID reserved keys (no longer written after R22).
func payloadDoesNotContainReservedKeys(t *testing.T, payload []byte) {
	t.Helper()
	if len(payload) == 0 {
		return // empty payload is fine
	}
	var m map[string]json.RawMessage
	if err := json.Unmarshal(payload, &m); err != nil {
		return // not a JSON object — fine, no reserved keys possible
	}
	if _, ok := m["__objectKind"]; ok {
		t.Error("payload must not contain reserved __objectKind key after R22 envelope-wrapper refactor")
	}
	if _, ok := m["__sourceID"]; ok {
		t.Error("payload must not contain reserved __sourceID key after R22 envelope-wrapper refactor")
	}
}

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

// TestEnvelopeParser_NoopFixture4Rules verifies the 4-rule __test_noop__ fixture
// parses a multi-event envelope into the correct ConnectorEvent slice.
func TestEnvelopeParser_NoopFixture4Rules(t *testing.T) {
	p := mustNewParser(t, noopParserNamespace, noopParserConfig())

	envelope := []byte(`[
		{"eventType":"noop.created","objectId":1,"occurredAt":1000000,"deliveryId":"dlv-001"},
		{"eventType":"noop.updated","objectId":2,"changedField":"name","deliveryId":"dlv-002"},
		{"eventType":"noop.deleted","objectId":3,"deliveryId":"dlv-003"},
		{"eventType":"noop.merge","objectId":4,"mergedIds":[5,6],"deliveryId":"dlv-004"}
	]`)

	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: envelope})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	if len(events) != 4 {
		t.Fatalf("want 4 events, got %d", len(events))
	}

	for i, ev := range events {
		if ev.ObjectKind != "noop_object" {
			t.Errorf("event[%d].ObjectKind = %q, want %q", i, ev.ObjectKind, "noop_object")
		}
		if ev.Namespace != noopParserNamespace {
			t.Errorf("event[%d].Namespace = %q, want %q", i, ev.Namespace, noopParserNamespace)
		}
		// SourceID must be non-empty (extracted from SourceIDPath = ".objectId").
		if ev.SourceID == "" {
			t.Errorf("event[%d].SourceID is empty; parser must set it from SourceIDPath", i)
		}
		// Payload must NOT contain reserved keys after R22 (no reserved-key injection).
		payloadDoesNotContainReservedKeys(t, ev.Payload)
		// IdempotencyKey must be non-empty (from rule.IdempotencyKeyPath = ".deliveryId").
		if ev.IdempotencyKey == "" {
			t.Errorf("event[%d].IdempotencyKey is empty; parser must set it from IdempotencyKeyPath", i)
		}
	}

	// Check event types in order.
	want := []string{"noop.created", "noop.updated", "noop.deleted", "noop.merge"}
	for i, ev := range events {
		if ev.EventType != want[i] {
			t.Errorf("event[%d].EventType = %q, want %q", i, ev.EventType, want[i])
		}
	}
}

// TestNewEnvelopeParser_RejectsEmptyObjectKind verifies construction rejects a
// rule with an empty ObjectKind.
func TestNewEnvelopeParser_RejectsEmptyObjectKind(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "noop.created",
				ObjectKind:         "", // EMPTY — must be rejected
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".deliveryId",
			},
		},
	}
	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig, got %v", err)
	}
}

// TestNewEnvelopeParser_WhitespaceObjectKind_Rejected verifies that construction
// rejects an ObjectKind that is non-empty but consists only of whitespace.
// A whitespace-only ObjectKind would pass the old `== ""` check and then fail
// every delivery at dispatcher Step 8 with 503; failing at construction surfaces
// the pack-config error to the author immediately.
func TestNewEnvelopeParser_WhitespaceObjectKind_Rejected(t *testing.T) {
	for _, ws := range []string{" ", "   ", "\t", " \t "} {
		ws := ws
		t.Run(fmt.Sprintf("%q", ws), func(t *testing.T) {
			cfg := EnvelopeMappingConfig{
				EventArrayPath: ".",
				Rules: []EnvelopeMappingRule{
					{
						EventTypeLiteral:   "noop.created",
						ObjectKind:         ws, // whitespace-only — must be rejected
						SourceIDPath:       ".objectId",
						IdempotencyKeyPath: ".deliveryId",
					},
				},
			}
			_, err := NewEnvelopeParser("ns", cfg)
			if !errors.Is(err, ErrInvalidParserConfig) {
				t.Fatalf("ObjectKind=%q: want ErrInvalidParserConfig, got %v", ws, err)
			}
		})
	}
}

// TestNewEnvelopeParser_ObjectKindWithSurroundingSpaces_Canonicalized verifies
// that a valid ObjectKind with surrounding whitespace is accepted and stored
// with the whitespace trimmed (consistent with EventTypeLiteral canonicalization).
func TestNewEnvelopeParser_ObjectKindWithSurroundingSpaces_Canonicalized(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "contact.created",
				ObjectKind:         "  contact  ", // valid value with surrounding spaces
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".deliveryId",
			},
		},
	}
	p, err := NewEnvelopeParser("ns", cfg)
	if err != nil {
		t.Fatalf("unexpected construction error: %v", err)
	}
	// Verify the stored rule has the trimmed value by parsing a matching event
	// and checking the emitted ObjectKind.
	events, parseErr := p.Parse(context.Background(), &connectorcapabilities.ParserInput{
		Payload: []byte(`[{"objectId":"src-1","deliveryId":"key-1"}]`),
	})
	if parseErr != nil {
		t.Fatalf("Parse error: %v", parseErr)
	}
	if len(events) != 1 {
		t.Fatalf("len(events) = %d, want 1", len(events))
	}
	if events[0].ObjectKind != "contact" {
		t.Errorf("ObjectKind = %q, want %q (should be trimmed)", events[0].ObjectKind, "contact")
	}
}

func TestEnvelopeParser_SourceLaneOverrideDoesNotReplaceSourceID(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventObjectPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:     "workflow.action.execute",
				NamespaceFromCtx:     true,
				ObjectKind:           "workflow-actions",
				SourceIDPath:         ".object.objectId",
				SourceLaneObjectKind: " shared-parent ",
				SourceLaneIDPath:     ".inputFields.parent_id",
				IdempotencyKeyPath:   ".callbackId",
				PayloadProjection: []PayloadField{
					{Name: "parent_id", Path: ".inputFields.parent_id"},
				},
			},
		},
	}
	p := mustNewParser(t, "hubspot-crm-connector", cfg)

	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{
		Payload: []byte(`{
			"callbackId":"callback-1",
			"object":{"objectId":"contact-123"},
			"inputFields":{"parent_id":" parent-001 "}
		}`),
	})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	if len(events) != 1 {
		t.Fatalf("len(events) = %d, want 1", len(events))
	}
	ev := events[0]
	if ev.SourceID != "contact-123" {
		t.Fatalf("SourceID = %q, want contact object id", ev.SourceID)
	}
	if ev.SourceLaneObjectKind != "shared-parent" || ev.SourceLaneID != " parent-001 " {
		t.Fatalf("source lane = %q/%q, want shared-parent/ parent-001 ", ev.SourceLaneObjectKind, ev.SourceLaneID)
	}
}

func TestNewEnvelopeParser_SourceLaneOverrideRequiresKindAndPath(t *testing.T) {
	tests := []struct {
		name   string
		kind   string
		idPath string
	}{
		{name: "kind without id path", kind: "shared-parent"},
		{name: "id path without kind", idPath: ".inputFields.parent_id"},
		{name: "whitespace kind with id path", kind: "   ", idPath: ".inputFields.parent_id"},
	}
	for _, tt := range tests {
		tt := tt
		t.Run(tt.name, func(t *testing.T) {
			_, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
				EventObjectPath: ".",
				Rules: []EnvelopeMappingRule{
					{
						EventTypeLiteral:     "workflow.action.execute",
						ObjectKind:           "workflow-actions",
						SourceIDPath:         ".object.objectId",
						SourceLaneObjectKind: tt.kind,
						SourceLaneIDPath:     tt.idPath,
						IdempotencyKeyPath:   ".callbackId",
					},
				},
			})
			if !errors.Is(err, ErrInvalidParserConfig) {
				t.Fatalf("want ErrInvalidParserConfig, got %v", err)
			}
		})
	}
}

func TestEnvelopeParser_SourceLaneIDPathMissingIsConfigDrift(t *testing.T) {
	p := mustNewParser(t, "hubspot-crm-connector", EnvelopeMappingConfig{
		EventObjectPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:     "workflow.action.execute",
				ObjectKind:           "workflow-actions",
				SourceIDPath:         ".object.objectId",
				SourceLaneObjectKind: "shared-parent",
				SourceLaneIDPath:     ".inputFields.parent_id",
				IdempotencyKeyPath:   ".callbackId",
			},
		},
	})

	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{
		Payload: []byte(`{"callbackId":"callback-1","object":{"objectId":"contact-123"},"inputFields":{}}`),
	})
	if !errors.Is(err, connectorcapabilities.ErrParserConfigDrift) {
		t.Fatalf("want ErrParserConfigDrift, got %v", err)
	}
}

func TestNewEnvelopeParser_PreservesOmittedInboundProcessingMode(t *testing.T) {
	p, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "contact.created",
				ObjectKind:         "contact",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".deliveryId",
			},
		},
	})
	if err != nil {
		t.Fatalf("unexpected construction error: %v", err)
	}
	if p.cfg.InboundProcessingMode != "" {
		t.Fatalf("InboundProcessingMode = %q, want omitted empty mode", p.cfg.InboundProcessingMode)
	}
}

func TestNewEnvelopeParser_RejectsUnknownInboundProcessingMode(t *testing.T) {
	_, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
		EventArrayPath:        ".",
		InboundProcessingMode: "surprise",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "contact.created",
				ObjectKind:         "contact",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".deliveryId",
			},
		},
	})
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig, got %v", err)
	}
}

// TestNewEnvelopeParser_RejectsEmptySourceIDPath verifies construction rejects a
// rule with an empty SourceIDPath.
func TestNewEnvelopeParser_RejectsEmptySourceIDPath(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "noop.created",
				ObjectKind:         "noop_object",
				SourceIDPath:       "", // EMPTY — must be rejected
				IdempotencyKeyPath: ".deliveryId",
			},
		},
	}
	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig for empty SourceIDPath, got %v", err)
	}
}

// TestNewEnvelopeParser_RejectsEmptyIdempotencyKeyPath verifies construction rejects a
// rule with an empty IdempotencyKeyPath.
func TestNewEnvelopeParser_RejectsEmptyIdempotencyKeyPath(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "noop.created",
				ObjectKind:         "noop_object",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: "", // EMPTY — must be rejected
			},
		},
	}
	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig for empty IdempotencyKeyPath, got %v", err)
	}
}

// TestNewEnvelopeParser_AllowsFormerlyReservedFieldNamesInProjection verifies
// that "__objectKind" and "__sourceID" are no longer reserved PayloadProjection
// field names after R22. The dispatcher no longer injects these keys into Payload;
// typed fields on ConnectorEvent carry the metadata instead.
func TestNewEnvelopeParser_AllowsFormerlyReservedFieldNamesInProjection(t *testing.T) {
	for _, name := range []string{"__objectKind", "__sourceID"} {
		name := name
		t.Run(name, func(t *testing.T) {
			cfg := EnvelopeMappingConfig{
				EventArrayPath: ".",
				Rules: []EnvelopeMappingRule{
					{
						EventTypeLiteral:   "noop.created",
						ObjectKind:         "contact",
						SourceIDPath:       ".id",
						IdempotencyKeyPath: ".deliveryId",
						PayloadProjection: []PayloadField{
							{Name: name, Path: ".someField"},
						},
					},
				},
			}
			_, err := NewEnvelopeParser("ns", cfg)
			if err != nil {
				t.Errorf("NewEnvelopeParser with field name %q should now succeed (no longer reserved), got %v", name, err)
			}
		})
	}
}

// TestEnvelopeParser_EmittedObjectKindMatchesRule verifies ConnectorEvent.ObjectKind
// equals the rule's ObjectKind.
func TestEnvelopeParser_EmittedObjectKindMatchesRule(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "deal.created",
				ObjectKind:         "deal",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".deliveryId",
				NamespaceFromCtx:   true,
			},
		},
	}
	p := mustNewParser(t, "crm", cfg)
	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(`[{"id":"other-abc","deliveryId":"dlv-001"}]`)})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	if len(events) != 1 {
		t.Fatalf("want 1 event, got %d", len(events))
	}
	if events[0].ObjectKind != "deal" {
		t.Errorf("ObjectKind = %q, want %q", events[0].ObjectKind, "deal")
	}
}

// TestEnvelopeParser_TypedSourceIDField verifies that after R22 the parser populates
// ConnectorEvent.SourceID (typed field) from SourceIDPath, and that the Payload bytes
// contain only the PayloadProjection fields (no reserved-key injection).
func TestEnvelopeParser_TypedSourceIDField(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "contact.created",
				ObjectKind:         "contact",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".deliveryId",
				NamespaceFromCtx:   true,
				PayloadProjection: []PayloadField{
					{Name: "name", Path: ".name"},
				},
			},
		},
	}
	p := mustNewParser(t, "ns", cfg)
	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(`[{"id":"contact-42","name":"Alice","deliveryId":"dlv-xyz"}]`)})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	if len(events) != 1 {
		t.Fatalf("want 1 event")
	}

	// SourceID must be the typed field value, not a reserved payload key.
	if events[0].SourceID != "contact-42" {
		t.Errorf("ConnectorEvent.SourceID = %q, want %q", events[0].SourceID, "contact-42")
	}

	var m map[string]json.RawMessage
	if err := json.Unmarshal(events[0].Payload, &m); err != nil {
		t.Fatalf("unmarshal payload: %v", err)
	}

	// Payload must NOT contain reserved keys after R22.
	payloadDoesNotContainReservedKeys(t, events[0].Payload)

	// Only the projection field must be present.
	if _, ok := m["name"]; !ok {
		t.Error("payload missing 'name' projection field")
	}
}

func TestEnvelopeParser_PayloadProjectionModes_CoerceScalarAndArrayIDsToStrings(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "contact.merge",
				ObjectKind:         "contact",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
				PayloadProjection: []PayloadField{
					{Name: "objectId", Path: ".objectId"},
					{Name: "primaryObjectId", Path: ".primaryObjectId", Mode: PayloadFieldModeScalarString},
					{Name: "mergedObjectIds", Path: ".mergedObjectIds", Mode: PayloadFieldModeScalarStringArray},
				},
			},
		},
	}
	p := mustNewParser(t, "ns", cfg)
	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{
		Payload: []byte(`[{"eventId":"evt-1","objectId":227944685622,"primaryObjectId":227943645990,"mergedObjectIds":[227943586330,227943645990]}]`),
	})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	if len(events) != 1 {
		t.Fatalf("want 1 event, got %d", len(events))
	}

	var got map[string]any
	if err := json.Unmarshal(events[0].Payload, &got); err != nil {
		t.Fatalf("unmarshal payload: %v", err)
	}
	if _, ok := got["objectId"].(float64); !ok {
		t.Fatalf("objectId type = %T, want float64 raw passthrough", got["objectId"])
	}
	if got["primaryObjectId"] != "227943645990" {
		t.Fatalf("primaryObjectId = %#v, want stringified exact ID", got["primaryObjectId"])
	}
	merged, ok := got["mergedObjectIds"].([]any)
	if !ok {
		t.Fatalf("mergedObjectIds type = %T, want []any", got["mergedObjectIds"])
	}
	if len(merged) != 2 || merged[0] != "227943586330" || merged[1] != "227943645990" {
		t.Fatalf("mergedObjectIds = %#v, want exact string array", merged)
	}
}

func TestNewEnvelopeParser_RejectsUnknownPayloadFieldMode(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "contact.merge",
				ObjectKind:         "contact",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
				PayloadProjection: []PayloadField{
					{Name: "primaryObjectId", Path: ".primaryObjectId", Mode: PayloadFieldMode("surprise")},
				},
			},
		},
	}
	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig, got %v", err)
	}
}

// TestEnvelopeParser_MissingSourceIDPath_ErrParserConfigDrift verifies that a
// SourceIDPath that resolves to absent/empty returns ErrParserConfigDrift (503,
// config-drift class), NOT ErrMalformedPayload (400). A missing field most likely
// means the path is mis-configured in the pack — the delivery should not be
// permanently dropped as a client error.
func TestEnvelopeParser_MissingSourceIDPath_ErrParserConfigDrift(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "deal.created",
				ObjectKind:         "deal",
				SourceIDPath:       ".missingField", // path absent in item
				IdempotencyKeyPath: ".deliveryId",
				NamespaceFromCtx:   true,
			},
		},
	}
	p := mustNewParser(t, "ns", cfg)
	// The item does not have "missingField" → sourceID resolves to absent → ErrParserConfigDrift.
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(`[{"deliveryId":"dlv-001","otherField":"value"}]`)})
	if !errors.Is(err, connectorcapabilities.ErrParserConfigDrift) {
		t.Fatalf("want ErrParserConfigDrift for absent SourceIDPath, got %v", err)
	}
	if errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Fatalf("must NOT be ErrMalformedPayload (would terminal-400-drop a valid delivery): %v", err)
	}
}

// TestEnvelopeParser_EmptyEventArray returns empty slice without error.
func TestEnvelopeParser_EmptyEventArray(t *testing.T) {
	p := mustNewParser(t, "ns", noopParserConfig())
	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(`[]`)})
	if err != nil {
		t.Fatalf("Parse empty array: %v", err)
	}
	if len(events) != 0 {
		t.Errorf("want 0 events, got %d", len(events))
	}
}

// TestEnvelopeParser_UnknownEventTypeSkipped verifies that items not matching
// any rule are skipped (observable via metric and log; issue #2127).
func TestEnvelopeParser_UnknownEventTypeSkipped(t *testing.T) {
	p := mustNewParser(t, noopParserNamespace, noopParserConfig())

	envelope := []byte(`[
		{"eventType":"noop.created","objectId":1,"deliveryId":"dlv-001"},
		{"eventType":"unknown.event","objectId":99,"deliveryId":"dlv-unknown"},
		{"eventType":"noop.deleted","objectId":3,"deliveryId":"dlv-003"}
	]`)
	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: envelope})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	// unknown.event is skipped; only created + deleted survive.
	if len(events) != 2 {
		t.Errorf("want 2 events, got %d", len(events))
	}
	if events[0].EventType != "noop.created" || events[1].EventType != "noop.deleted" {
		t.Errorf("unexpected event types: %q, %q", events[0].EventType, events[1].EventType)
	}
}

// TestEnvelopeParser_SkippedItemIncrementsMetric verifies that each unmatched
// envelope item increments the skipped-items counter (issue #2127).
func TestEnvelopeParser_SkippedItemIncrementsMetric(t *testing.T) {
	p := mustNewParser(t, noopParserNamespace, noopParserConfig())

	before := testutil.ToFloat64(connectorInboundItemsSkippedTotal.WithLabelValues(noopParserNamespace))

	envelope := []byte(`[
		{"eventType":"noop.created","objectId":1,"deliveryId":"dlv-001"},
		{"eventType":"unknown.event","objectId":99,"deliveryId":"dlv-unknown"},
		{"eventType":"another.unknown","objectId":100,"deliveryId":"dlv-unknown-2"},
		{"eventType":"noop.deleted","objectId":3,"deliveryId":"dlv-003"}
	]`)
	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: envelope})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	if len(events) != 2 {
		t.Fatalf("want 2 events, got %d", len(events))
	}

	after := testutil.ToFloat64(connectorInboundItemsSkippedTotal.WithLabelValues(noopParserNamespace))
	if delta := after - before; delta != 2 {
		t.Errorf("want skipped counter delta 2, got %v", delta)
	}
}

// TestEnvelopeParser_NoSkippedItemsNoMetricIncrement verifies that a fully
// matched envelope does not increment the skipped-items counter (issue #2127).
func TestEnvelopeParser_NoSkippedItemsNoMetricIncrement(t *testing.T) {
	p := mustNewParser(t, noopParserNamespace, noopParserConfig())

	before := testutil.ToFloat64(connectorInboundItemsSkippedTotal.WithLabelValues(noopParserNamespace))

	envelope := []byte(`[
		{"eventType":"noop.created","objectId":1,"deliveryId":"dlv-001"},
		{"eventType":"noop.deleted","objectId":3,"deliveryId":"dlv-003"}
	]`)
	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: envelope})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	if len(events) != 2 {
		t.Fatalf("want 2 events, got %d", len(events))
	}

	after := testutil.ToFloat64(connectorInboundItemsSkippedTotal.WithLabelValues(noopParserNamespace))
	if delta := after - before; delta != 0 {
		t.Errorf("want skipped counter delta 0, got %v", delta)
	}
}

// TestEnvelopeParser_MalformedJSON returns ErrMalformedPayload.
func TestEnvelopeParser_MalformedJSON(t *testing.T) {
	p := mustNewParser(t, "ns", noopParserConfig())
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(`not json at all`)})
	if !errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Fatalf("want ErrMalformedPayload, got %v", err)
	}
}

// TestEnvelopeParser_MissingEventArrayPath returns ErrMalformedPayload when
// the configured EventArrayPath resolves to null.
func TestEnvelopeParser_MissingEventArrayPath(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".events",
		Rules: []EnvelopeMappingRule{
			{EventTypeLiteral: "x", ObjectKind: "obj", SourceIDPath: ".id", IdempotencyKeyPath: ".deliveryId"},
		},
	}
	p := mustNewParser(t, "ns", cfg)
	// Payload is a JSON object but "events" key is absent → array path resolves to nil.
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(`{"other":"field"}`)})
	if !errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Fatalf("want ErrMalformedPayload for missing array path, got %v", err)
	}
}

func TestEnvelopeParser_EventObjectPathParsesSingleObject(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventObjectPath: ".",
		Rules: []EnvelopeMappingRule{{
			EventTypeLiteral:   "singleton.object.received",
			NamespaceFromCtx:   true,
			ObjectKind:         "singleton-events",
			SourceIDPath:       ".object.id",
			IdempotencyKeyPath: ".deliveryId",
			PayloadProjection: []PayloadField{
				{Name: "external_id", Path: ".input.external_id"},
				{Name: "status", Path: ".input.status"},
			},
		}},
	}
	p := mustNewParser(t, "generic-connector", cfg)
	payload := []byte(`{
  "deliveryId":"delivery-001",
  "object":{"id":"object-001","type":"example"},
  "input":{"external_id":"external-001","status":"received"}
}`)

	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	if len(events) != 1 {
		t.Fatalf("len(events) = %d, want 1", len(events))
	}
	ev := events[0]
	if ev.Namespace != "generic-connector" || ev.EventType != "singleton.object.received" || ev.ObjectKind != "singleton-events" {
		t.Fatalf("unexpected event identity: %+v", ev)
	}
	if ev.SourceID != "object-001" || ev.IdempotencyKey != "delivery-001" {
		t.Fatalf("unexpected event keys: SourceID=%q IdempotencyKey=%q", ev.SourceID, ev.IdempotencyKey)
	}
	var projected map[string]string
	if err := json.Unmarshal(ev.Payload, &projected); err != nil {
		t.Fatalf("unmarshal payload: %v", err)
	}
	if projected["external_id"] != "external-001" || projected["status"] != "received" {
		t.Fatalf("unexpected payload projection: %v", projected)
	}
}

func TestEnvelopeParser_EventObjectPathRejectsArrayPayload(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventObjectPath: ".",
		Rules: []EnvelopeMappingRule{{
			EventTypeLiteral:   "singleton.object.received",
			ObjectKind:         "singleton-events",
			SourceIDPath:       ".objectId",
			IdempotencyKeyPath: ".deliveryId",
		}},
	}
	p := mustNewParser(t, "generic-connector", cfg)

	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(`[{"objectId":"1","deliveryId":"delivery-1"}]`)})
	if !errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Fatalf("want ErrMalformedPayload for array at EventObjectPath, got %v", err)
	}
}

func TestEnvelopeParser_EventObjectPathRejectsNonObjectPayloads(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventObjectPath: ".event",
		Rules: []EnvelopeMappingRule{{
			EventTypeLiteral:   "singleton.object.received",
			ObjectKind:         "singleton-events",
			SourceIDPath:       ".objectId",
			IdempotencyKeyPath: ".deliveryId",
		}},
	}
	p := mustNewParser(t, "generic-connector", cfg)

	cases := []struct {
		name    string
		payload string
	}{
		{name: "scalar", payload: `{"event":"not-an-object"}`},
		{name: "null", payload: `{"event":null}`},
		{name: "missing", payload: `{}`},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(tc.payload)})
			if !errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
				t.Fatalf("want ErrMalformedPayload for %s EventObjectPath payload, got %v", tc.name, err)
			}
		})
	}
}

func TestNewEnvelopeParser_RejectsBothEventArrayAndObjectPath(t *testing.T) {
	_, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
		EventArrayPath:  ".events",
		EventObjectPath: ".",
		Rules: []EnvelopeMappingRule{{
			EventTypeLiteral:   "x",
			ObjectKind:         "obj",
			SourceIDPath:       ".id",
			IdempotencyKeyPath: ".eventId",
		}},
	})
	if err == nil {
		t.Fatal("NewEnvelopeParser = nil, want mutually exclusive envelope source error")
	}
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig, got %v", err)
	}
}

func TestNewEnvelopeParser_RejectsMissingEventArrayAndObjectPath(t *testing.T) {
	_, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
		Rules: []EnvelopeMappingRule{{
			EventTypeLiteral:   "x",
			ObjectKind:         "obj",
			SourceIDPath:       ".id",
			IdempotencyKeyPath: ".eventId",
		}},
	})
	if err == nil {
		t.Fatal("NewEnvelopeParser = nil, want missing envelope source error")
	}
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig, got %v", err)
	}
}

// TestParse_MissingIdempotencyKey_ConfigDrift verifies that when IdempotencyKeyPath
// resolves to an absent field, Parse returns ErrParserConfigDrift (503, config-drift
// class), NOT ErrMalformedPayload. An absent field at IdempotencyKeyPath most
// likely means the path is mis-configured — the delivery must not be permanently dropped.
func TestParse_MissingIdempotencyKey_ConfigDrift(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "contact.created",
				ObjectKind:         "contact",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".missingDeliveryId", // absent in item
				NamespaceFromCtx:   true,
			},
		},
	}
	p := mustNewParser(t, "ns", cfg)
	// Item has no "missingDeliveryId" field → IdempotencyKey resolves to absent → ErrParserConfigDrift.
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(`[{"id":"contact-42"}]`)})
	if !errors.Is(err, connectorcapabilities.ErrParserConfigDrift) {
		t.Fatalf("want ErrParserConfigDrift for absent IdempotencyKeyPath, got %v", err)
	}
	if errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Fatalf("must NOT be ErrMalformedPayload (would terminal-400-drop a valid delivery): %v", err)
	}
}

// TestParse_EmptyIdempotencyKey_ConfigDrift verifies that when IdempotencyKeyPath
// resolves to an empty string, Parse returns ErrParserConfigDrift (503, config-drift
// class), NOT ErrMalformedPayload. An empty string at IdempotencyKeyPath is the same
// absent/empty class — likely a mis-configured path.
func TestParse_EmptyIdempotencyKey_ConfigDrift(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "contact.created",
				ObjectKind:         "contact",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".deliveryId",
				NamespaceFromCtx:   true,
			},
		},
	}
	p := mustNewParser(t, "ns", cfg)
	// Item has deliveryId = "" → IdempotencyKey resolves to empty string → ErrParserConfigDrift.
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(`[{"id":"contact-42","deliveryId":""}]`)})
	if !errors.Is(err, connectorcapabilities.ErrParserConfigDrift) {
		t.Fatalf("want ErrParserConfigDrift for empty IdempotencyKey value, got %v", err)
	}
	if errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Fatalf("must NOT be ErrMalformedPayload: %v", err)
	}
}

// TestParse_TwoUpdatesToSameSourceIDProduceDistinctIdempotencyKeys verifies that
// two events for the same object (same SourceID) produce distinct IdempotencyKey
// values when they carry different delivery IDs — ensuring the queue does not
// fold them into a single row.
func TestParse_TwoUpdatesToSameSourceIDProduceDistinctIdempotencyKeys(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "contact.updated",
				ObjectKind:         "contact",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".deliveryId",
				NamespaceFromCtx:   true,
			},
		},
	}
	p := mustNewParser(t, "ns", cfg)
	// Two updates to the same object (objectId="contact-42") with different delivery IDs.
	envelope := []byte(`[
		{"objectId":"contact-42","deliveryId":"dlv-abc"},
		{"objectId":"contact-42","deliveryId":"dlv-xyz"}
	]`)
	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: envelope})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	if len(events) != 2 {
		t.Fatalf("want 2 events, got %d", len(events))
	}
	if events[0].IdempotencyKey == events[1].IdempotencyKey {
		t.Errorf("two updates to same object produced identical IdempotencyKey %q; expected distinct keys",
			events[0].IdempotencyKey)
	}
	// Both events share the same SourceID (carried via the typed ConnectorEvent.SourceID field).
	if events[0].SourceID != "contact-42" {
		t.Errorf("event[0].SourceID = %q, want %q", events[0].SourceID, "contact-42")
	}
	if events[1].SourceID != "contact-42" {
		t.Errorf("event[1].SourceID = %q, want %q", events[1].SourceID, "contact-42")
	}
}

// TestNewEnvelopeParser_RejectsZeroRules verifies that construction rejects a
// config with nil Rules and a config with an empty (non-nil) Rules slice.
// A parser with zero rules would silently drop every delivery, turning a valid
// signed HTTP 200 into a permanent silent data loss (HTTP 202 response with no
// events queued). Fail closed at construction is safer than fail-open at runtime.
func TestNewEnvelopeParser_RejectsZeroRules(t *testing.T) {
	t.Run("nil rules", func(t *testing.T) {
		_, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
			EventArrayPath: ".events",
			Rules:          nil,
		})
		if err == nil {
			t.Fatal("expected error for nil rules, got nil")
		}
		if !errors.Is(err, ErrInvalidParserConfig) {
			t.Fatalf("expected ErrInvalidParserConfig, got %v", err)
		}
	})
	t.Run("empty slice rules", func(t *testing.T) {
		_, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
			EventArrayPath: ".events",
			Rules:          []EnvelopeMappingRule{},
		})
		if err == nil {
			t.Fatal("expected error for empty rules slice, got nil")
		}
		if !errors.Is(err, ErrInvalidParserConfig) {
			t.Fatalf("expected ErrInvalidParserConfig, got %v", err)
		}
	})
}

// TestNewEnvelopeParser_RejectsIdempotencyKeyPathEqualSourceIDPath verifies that
// construction rejects a rule where IdempotencyKeyPath equals SourceIDPath (trimmed).
// Using the object's source ID as the idempotency key collapses all updates for
// the same object into one queue row (recreates R7-B2 silent-drop via misconfiguration).
func TestNewEnvelopeParser_RejectsIdempotencyKeyPathEqualSourceIDPath(t *testing.T) {
	cases := []struct {
		name     string
		sourceID string
		idemKey  string
	}{
		{name: "exact_match", sourceID: ".objectId", idemKey: ".objectId"},
		{name: "whitespace_trimmed_equal", sourceID: ".objectId", idemKey: "  .objectId  "},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			_, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
				EventArrayPath: ".events",
				Rules: []EnvelopeMappingRule{
					{
						ObjectKind:         "contact",
						SourceIDPath:       tc.sourceID,
						IdempotencyKeyPath: tc.idemKey,
						EventTypeLiteral:   "contact.created",
					},
				},
			})
			if err == nil {
				t.Fatal("expected error for IdempotencyKeyPath == SourceIDPath, got nil")
			}
			if !errors.Is(err, ErrInvalidParserConfig) {
				t.Fatalf("expected ErrInvalidParserConfig, got %v", err)
			}
		})
	}
}

// ── Fix 3 — PayloadField.Name empty ──────────────────────────────────────────

// TestNewEnvelopeParser_RejectsEmptyPayloadFieldName verifies that construction
// rejects a PayloadProjection field whose Name is empty or whitespace-only.
// An empty Name silently emits a "" JSON key in every payload, corrupting the
// cross-process queue row.
func TestNewEnvelopeParser_RejectsEmptyPayloadFieldName(t *testing.T) {
	cases := []struct {
		name      string
		fieldName string
	}{
		{name: "empty_string", fieldName: ""},
		{name: "whitespace_only", fieldName: "   "},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			cfg := EnvelopeMappingConfig{
				EventArrayPath: ".",
				Rules: []EnvelopeMappingRule{
					{
						EventTypeLiteral:   "contact.created",
						ObjectKind:         "contact",
						SourceIDPath:       ".id",
						IdempotencyKeyPath: ".deliveryId",
						PayloadProjection: []PayloadField{
							{Name: tc.fieldName, Path: ".someField"},
						},
					},
				},
			}
			_, err := NewEnvelopeParser("ns", cfg)
			if !errors.Is(err, ErrInvalidParserConfig) {
				t.Fatalf("want ErrInvalidParserConfig for empty PayloadField.Name %q, got %v", tc.fieldName, err)
			}
		})
	}
}

// ── Fix 4 — Duplicate PayloadField.Name within one rule ───────────────────────

// TestNewEnvelopeParser_RejectsDuplicatePayloadFieldName verifies that
// construction rejects a rule where two PayloadProjection fields share the same
// Name. JSON last-write-wins would silently overwrite the first value, causing
// data loss on every delivery.
func TestNewEnvelopeParser_RejectsDuplicatePayloadFieldName(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				EventTypeLiteral:   "contact.created",
				ObjectKind:         "contact",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".deliveryId",
				PayloadProjection: []PayloadField{
					{Name: "email", Path: ".email"},
					{Name: "email", Path: ".primaryEmail"}, // duplicate Name
				},
			},
		},
	}
	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig for duplicate PayloadField.Name, got %v", err)
	}
}

// ── Fix 5 — Catch-all rule must be last ───────────────────────────────────────

// TestNewEnvelopeParser_RejectsCatchAllNotLast verifies that construction
// rejects a catch-all rule (MatchPath == "") that is not the final rule. A
// catch-all before other rules makes subsequent rules unreachable dead code:
// every event matches the catch-all first and routes to its ObjectKind/EventType.
func TestNewEnvelopeParser_RejectsCatchAllNotLast(t *testing.T) {
	t.Run("catch_all_at_index_0_with_another_rule_after", func(t *testing.T) {
		cfg := EnvelopeMappingConfig{
			EventArrayPath: ".",
			Rules: []EnvelopeMappingRule{
				{
					// catch-all: MatchPath empty
					EventTypeLiteral:   "any.event",
					ObjectKind:         "generic",
					SourceIDPath:       ".id",
					IdempotencyKeyPath: ".deliveryId",
				},
				{
					MatchPath:          ".eventType",
					MatchEquals:        "contact.created",
					EventTypeLiteral:   "contact.created",
					ObjectKind:         "contact",
					SourceIDPath:       ".id",
					IdempotencyKeyPath: ".deliveryId",
				},
			},
		}
		_, err := NewEnvelopeParser("ns", cfg)
		if !errors.Is(err, ErrInvalidParserConfig) {
			t.Fatalf("want ErrInvalidParserConfig for catch-all not last, got %v", err)
		}
	})

	t.Run("two_catch_all_rules", func(t *testing.T) {
		cfg := EnvelopeMappingConfig{
			EventArrayPath: ".",
			Rules: []EnvelopeMappingRule{
				{
					EventTypeLiteral:   "any.event",
					ObjectKind:         "generic",
					SourceIDPath:       ".id",
					IdempotencyKeyPath: ".deliveryId",
				},
				{
					EventTypeLiteral:   "other.event",
					ObjectKind:         "other",
					SourceIDPath:       ".id",
					IdempotencyKeyPath: ".deliveryId",
				},
			},
		}
		_, err := NewEnvelopeParser("ns", cfg)
		if !errors.Is(err, ErrInvalidParserConfig) {
			t.Fatalf("want ErrInvalidParserConfig for two catch-all rules, got %v", err)
		}
	})

	t.Run("single_catch_all_rule_is_ok", func(t *testing.T) {
		cfg := EnvelopeMappingConfig{
			EventArrayPath: ".",
			Rules: []EnvelopeMappingRule{
				{
					// single catch-all as the only rule — allowed
					EventTypeLiteral:   "any.event",
					ObjectKind:         "generic",
					SourceIDPath:       ".id",
					IdempotencyKeyPath: ".deliveryId",
				},
			},
		}
		_, err := NewEnvelopeParser("ns", cfg)
		if err != nil {
			t.Fatalf("unexpected error for single catch-all rule: %v", err)
		}
	})

	t.Run("catch_all_at_last_position_is_ok", func(t *testing.T) {
		cfg := EnvelopeMappingConfig{
			EventArrayPath: ".",
			Rules: []EnvelopeMappingRule{
				{
					MatchPath:          ".eventType",
					MatchEquals:        "contact.created",
					EventTypeLiteral:   "contact.created",
					ObjectKind:         "contact",
					SourceIDPath:       ".id",
					IdempotencyKeyPath: ".deliveryId",
				},
				{
					// catch-all at last position — allowed
					EventTypeLiteral:   "any.event",
					ObjectKind:         "generic",
					SourceIDPath:       ".id",
					IdempotencyKeyPath: ".deliveryId",
				},
			},
		}
		_, err := NewEnvelopeParser("ns", cfg)
		if err != nil {
			t.Fatalf("unexpected error for catch-all in last position: %v", err)
		}
	})
}

// ── Fix 6 — Duplicate (MatchPath, MatchEquals) selector across rules ──────────

// TestNewEnvelopeParser_RejectsDuplicateRuleSelector verifies that construction
// rejects two non-catch-all rules with the same (MatchPath, MatchEquals) pair.
// First-match-wins makes the second rule unreachable: events that match the
// selector always route to the first rule's ObjectKind/EventType, silently
// ignoring the second rule.
func TestNewEnvelopeParser_RejectsDuplicateRuleSelector(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchPath:          ".eventType",
				MatchEquals:        "contact.created",
				EventTypeLiteral:   "contact.created",
				ObjectKind:         "contact",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".deliveryId",
			},
			{
				// Same selector as above — unreachable
				MatchPath:          ".eventType",
				MatchEquals:        "contact.created",
				EventTypeLiteral:   "contact.updated",
				ObjectKind:         "contact",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".deliveryId",
			},
		},
	}
	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig for duplicate rule selector, got %v", err)
	}
}

func TestEnvelopeParser_MatchAllRequiresEveryCondition(t *testing.T) {
	p := mustNewParser(t, "multi-condition", EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchAll: []MatchCondition{
					{Path: ".eventCategory", Equals: "entity.propertyChange"},
					{Path: ".entityTypeId", Equals: "0-2"},
				},
				EventTypeLiteral:   "entity.propertychange",
				ObjectKind:         "entities",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
				NamespaceFromCtx:   true,
			},
		},
	})

	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(`[
		{"eventCategory":"entity.propertyChange","entityTypeId":"0-2","objectId":"entity-1","eventId":"evt-1"},
		{"eventCategory":"object.creation","entityTypeId":"0-2","objectId":"entity-2","eventId":"evt-2"},
		{"eventCategory":"entity.propertyChange","entityTypeId":"0-3","objectId":"other-1","eventId":"evt-3"}
	]`)})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	if len(events) != 1 {
		t.Fatalf("len(events) = %d, want 1", len(events))
	}
	if got, want := events[0].EventType, "entity.propertychange"; got != want {
		t.Fatalf("EventType = %q, want %q", got, want)
	}
	if got, want := events[0].SourceID, "entity-1"; got != want {
		t.Fatalf("SourceID = %q, want %q", got, want)
	}
}

func TestNewEnvelopeParser_RejectsEmptyMatchAllSelector(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchAll:           []MatchCondition{},
				EventTypeLiteral:   "thing.updated",
				ObjectKind:         "thing",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
		},
	}

	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig for empty matchAll selector, got %v", err)
	}
}

func TestNewEnvelopeParser_RejectsDuplicateMatchAllSelector(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchAll: []MatchCondition{
					{Path: ".eventCategory", Equals: "entity.propertyChange"},
					{Path: ".entityTypeId", Equals: "0-2"},
				},
				EventTypeLiteral:   "entity.propertychange",
				ObjectKind:         "entities",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
			{
				MatchAll: []MatchCondition{
					{Path: ".entityTypeId", Equals: "0-2"},
					{Path: ".eventCategory", Equals: "entity.propertyChange"},
				},
				EventTypeLiteral:   "entity.propertychange.v2",
				ObjectKind:         "entities",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
		},
	}

	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig for duplicate matchAll selector, got %v", err)
	}
}

func TestNewEnvelopeParser_RejectsContradictoryMatchAllConditionsOnSamePath(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchAll: []MatchCondition{
					{Path: ".eventType", Equals: "create"},
					{Path: ".eventType", Equals: "update"},
				},
				EventTypeLiteral:   "thing.updated",
				ObjectKind:         "thing",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
		},
	}

	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig for contradictory matchAll conditions, got %v", err)
	}
}

func TestNewEnvelopeParser_RejectsRedundantMatchAllConditionsOnSamePath(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchAll: []MatchCondition{
					{Path: ".eventType"},
					{Path: ".eventType", Equals: "create"},
				},
				EventTypeLiteral:   "thing.created",
				ObjectKind:         "thing",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
		},
	}

	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig for redundant matchAll conditions, got %v", err)
	}
}

func TestNewEnvelopeParser_RejectsEquivalentPathSpellingsInMatchAllConditions(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchAll: []MatchCondition{
					{Path: ".items[1]", Equals: "create"},
					{Path: ".items[01]", Equals: "update"},
				},
				EventTypeLiteral:   "thing.updated",
				ObjectKind:         "thing",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
		},
	}

	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig for equivalent matchAll paths, got %v", err)
	}
}

func TestNewEnvelopeParser_RejectsLegacySelectorShadowingLaterMatchAllSelector(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchPath:          ".eventCategory",
				MatchEquals:        "entity.propertyChange",
				EventTypeLiteral:   "generic.propertychange",
				ObjectKind:         "objects",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
			{
				MatchAll: []MatchCondition{
					{Path: ".eventCategory", Equals: "entity.propertyChange"},
					{Path: ".entityTypeId", Equals: "0-2"},
				},
				EventTypeLiteral:   "entity.propertychange",
				ObjectKind:         "entities",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
		},
	}

	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig for shadowed matchAll selector, got %v", err)
	}
}

func TestNewEnvelopeParser_RejectsEquivalentLegacySelectorShadowingLaterMatchAllSelector(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchPath:          ".items[1]",
				MatchEquals:        "entity.propertyChange",
				EventTypeLiteral:   "generic.propertychange",
				ObjectKind:         "objects",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
			{
				MatchAll: []MatchCondition{
					{Path: ".items[01]", Equals: "entity.propertyChange"},
					{Path: ".entityTypeId", Equals: "0-2"},
				},
				EventTypeLiteral:   "entity.propertychange",
				ObjectKind:         "entities",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
		},
	}

	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig for equivalent-path shadowed matchAll selector, got %v", err)
	}
}

func TestNewEnvelopeParser_RejectsMatchAllSelectorShadowingLaterNarrowerMatchAllSelector(t *testing.T) {
	cfg := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchAll: []MatchCondition{
					{Path: ".eventCategory", Equals: "entity.propertyChange"},
				},
				EventTypeLiteral:   "generic.propertychange",
				ObjectKind:         "objects",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
			{
				MatchAll: []MatchCondition{
					{Path: ".eventCategory", Equals: "entity.propertyChange"},
					{Path: ".entityTypeId", Equals: "0-2"},
				},
				EventTypeLiteral:   "entity.propertychange",
				ObjectKind:         "entities",
				SourceIDPath:       ".objectId",
				IdempotencyKeyPath: ".eventId",
			},
		},
	}

	_, err := NewEnvelopeParser("ns", cfg)
	if !errors.Is(err, ErrInvalidParserConfig) {
		t.Fatalf("want ErrInvalidParserConfig for shadowed matchAll selector, got %v", err)
	}
}

// ── Fix 7 — IdempotencyKeyPath == EventTypeFromPath ───────────────────────────

// TestNewEnvelopeParser_RejectsIdempotencyKeyPathEqualEventTypeFromPath verifies
// that construction rejects a rule where IdempotencyKeyPath equals
// EventTypeFromPath (trimmed). Using the same path for event dedup and event
// type collapses all events of the same type to a single dedup row — subsequent
// deliveries of the same event type are silently dropped.
func TestNewEnvelopeParser_RejectsIdempotencyKeyPathEqualEventTypeFromPath(t *testing.T) {
	cases := []struct {
		name               string
		idempotencyKeyPath string
		eventTypeFromPath  string
	}{
		{name: "exact_match", idempotencyKeyPath: ".eventType", eventTypeFromPath: ".eventType"},
		{name: "whitespace_trimmed_equal", idempotencyKeyPath: "  .eventType  ", eventTypeFromPath: ".eventType"},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			cfg := EnvelopeMappingConfig{
				EventArrayPath: ".",
				Rules: []EnvelopeMappingRule{
					{
						ObjectKind:         "contact",
						SourceIDPath:       ".id",
						IdempotencyKeyPath: tc.idempotencyKeyPath,
						EventTypeFromPath:  tc.eventTypeFromPath,
					},
				},
			}
			_, err := NewEnvelopeParser("ns", cfg)
			if !errors.Is(err, ErrInvalidParserConfig) {
				t.Fatalf("want ErrInvalidParserConfig for IdempotencyKeyPath == EventTypeFromPath, got %v", err)
			}
		})
	}
}

// ── B1 — blank EventType rejection ────────────────────────────────────────────

// TestNewEnvelopeParser_RejectsWhitespaceOnlyEventTypeLiteral verifies that
// construction rejects a rule whose EventTypeLiteral is non-empty but contains
// only whitespace characters. Such a literal would route every matched event to
// a blank eventKind, which the dispatcher cannot look up and the queue cannot route.
func TestNewEnvelopeParser_RejectsWhitespaceOnlyEventTypeLiteral(t *testing.T) {
	cases := []struct{ name, lit string }{
		{"single_space", " "},
		{"tabs", "\t\t"},
		{"newlines", "\n\n"},
		{"mixed", " \t\n "},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			_, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
				EventArrayPath: ".events",
				Rules: []EnvelopeMappingRule{{
					ObjectKind:         "contact",
					SourceIDPath:       ".id",
					IdempotencyKeyPath: ".deliveryId",
					EventTypeLiteral:   tc.lit,
				}},
			})
			if err == nil {
				t.Fatal("expected error for whitespace-only EventTypeLiteral, got nil")
			}
			if !errors.Is(err, ErrInvalidParserConfig) {
				t.Fatalf("expected ErrInvalidParserConfig, got %v", err)
			}
		})
	}
}

// TestParse_EmptyResolvedEventType_ConfigDrift verifies that Parse returns
// ErrParserConfigDrift (503, config-drift class) — NOT ErrMalformedPayload —
// when EventTypeFromPath resolves to an empty string or a whitespace-only string.
// An empty/whitespace value at EventTypeFromPath means the path likely resolved to
// absent/null/empty in the item, which is a pack mis-configured path, not a
// client-malformed body.
func TestParse_EmptyResolvedEventType_ConfigDrift(t *testing.T) {
	// Construct a parser that uses EventTypeFromPath (so construction succeeds —
	// the blank check is parse-time for EventTypeFromPath).
	p, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
		EventArrayPath: ".events",
		Rules: []EnvelopeMappingRule{{
			ObjectKind:         "contact",
			SourceIDPath:       ".id",
			IdempotencyKeyPath: ".deliveryId",
			EventTypeFromPath:  ".type",
		}},
	})
	if err != nil {
		t.Fatal(err)
	}

	cases := []struct {
		name    string
		payload []byte
	}{
		{
			name:    "empty_string_type",
			payload: []byte(`{"events": [{"id": "evt1", "deliveryId": "d1", "type": ""}]}`),
		},
		{
			name:    "whitespace_only_type",
			payload: []byte(`{"events": [{"id": "evt2", "deliveryId": "d2", "type": "  "}]}`),
		},
		{
			name:    "tab_only_type",
			payload: []byte(`{"events": [{"id": "evt3", "deliveryId": "d3", "type": "\t"}]}`),
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			_, parseErr := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: tc.payload})
			if parseErr == nil {
				t.Fatal("expected ErrParserConfigDrift for empty/whitespace resolved EventType, got nil")
			}
			if !errors.Is(parseErr, connectorcapabilities.ErrParserConfigDrift) {
				t.Fatalf("expected ErrParserConfigDrift (absent/empty typed field → config-drift, not 400), got %v", parseErr)
			}
			if errors.Is(parseErr, connectorcapabilities.ErrMalformedPayload) {
				t.Fatalf("must NOT be ErrMalformedPayload (would terminal-400-drop a valid delivery): %v", parseErr)
			}
		})
	}
}

// TestEnvelopeParser_MultiEventEnvelopeOrder verifies that multiple events in
// one envelope are returned in document order.
func TestEnvelopeParser_MultiEventEnvelopeOrder(t *testing.T) {
	p := mustNewParser(t, noopParserNamespace, noopParserConfig())

	envelope := []byte(`[
		{"eventType":"noop.deleted","objectId":10,"deliveryId":"dlv-010"},
		{"eventType":"noop.created","objectId":20,"deliveryId":"dlv-020"},
		{"eventType":"noop.updated","objectId":30,"deliveryId":"dlv-030"}
	]`)
	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: envelope})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	if len(events) != 3 {
		t.Fatalf("want 3, got %d", len(events))
	}
	order := []string{"noop.deleted", "noop.created", "noop.updated"}
	for i, ev := range events {
		if ev.EventType != order[i] {
			t.Errorf("event[%d].EventType = %q, want %q", i, ev.EventType, order[i])
		}
	}
}

// ---------------------------------------------------------------------------
// extractScalarStrict regression tests (composite JSON at typed fields).
// These guard the fix for: extractScalarString silently coercing object/array
// values at EventType/SourceID/IdempotencyKey into bogus non-empty strings.
// ---------------------------------------------------------------------------

// makeStrictParser is a helper that builds a parser using EventTypeFromPath so
// all three typed-field extractors can be exercised.
func makeStrictParser(t *testing.T) *EnvelopeParser {
	t.Helper()
	p, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{{
			ObjectKind:         "contact",
			EventTypeFromPath:  ".eventType",
			SourceIDPath:       ".id",
			IdempotencyKeyPath: ".deliveryId",
		}},
	})
	if err != nil {
		t.Fatalf("NewEnvelopeParser: %v", err)
	}
	return p
}

// TestEmitEvent_ObjectAtSourceIDPath_RejectedMalformed verifies that a JSON
// object at SourceIDPath is rejected with ErrMalformedPayload.
func TestEmitEvent_ObjectAtSourceIDPath_RejectedMalformed(t *testing.T) {
	p := makeStrictParser(t)
	payload := []byte(`[{"eventType":"contact.created","id":{"nested":"value"},"deliveryId":"dlv-001"}]`)
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if err == nil {
		t.Fatal("want ErrMalformedPayload for object at SourceIDPath, got nil")
	}
	if !errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Fatalf("want ErrMalformedPayload, got %v", err)
	}
}

// TestEmitEvent_ArrayAtIdempotencyKeyPath_RejectedMalformed verifies that a
// JSON array at IdempotencyKeyPath is rejected with ErrMalformedPayload.
func TestEmitEvent_ArrayAtIdempotencyKeyPath_RejectedMalformed(t *testing.T) {
	p := makeStrictParser(t)
	payload := []byte(`[{"eventType":"contact.created","id":"contact-1","deliveryId":["a","b"]}]`)
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if err == nil {
		t.Fatal("want ErrMalformedPayload for array at IdempotencyKeyPath, got nil")
	}
	if !errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Fatalf("want ErrMalformedPayload, got %v", err)
	}
}

// TestEmitEvent_ObjectAtEventTypeFromPath_RejectedMalformed verifies that a
// JSON object at EventTypeFromPath is rejected with ErrMalformedPayload.
func TestEmitEvent_ObjectAtEventTypeFromPath_RejectedMalformed(t *testing.T) {
	p := makeStrictParser(t)
	payload := []byte(`[{"eventType":{"type":"contact.created"},"id":"contact-1","deliveryId":"dlv-001"}]`)
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if err == nil {
		t.Fatal("want ErrMalformedPayload for object at EventTypeFromPath, got nil")
	}
	if !errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Fatalf("want ErrMalformedPayload, got %v", err)
	}
}

// TestEmitEvent_NumericSourceID_StillCoerced verifies that a JSON number at
// SourceIDPath is accepted and coerced to its string representation. This
// preserves the intentional numeric-ID behavior.
func TestEmitEvent_NumericSourceID_StillCoerced(t *testing.T) {
	p := makeStrictParser(t)
	payload := []byte(`[{"eventType":"contact.created","id":42,"deliveryId":"dlv-001"}]`)
	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if err != nil {
		t.Fatalf("want success for numeric SourceIDPath, got %v", err)
	}
	if len(events) != 1 {
		t.Fatalf("want 1 event, got %d", len(events))
	}
	if events[0].SourceID != "42" {
		t.Errorf("want SourceID=%q, got %q", "42", events[0].SourceID)
	}
}

// TestEmitEvent_BoolSourceID_StillCoerced verifies that a JSON bool at
// SourceIDPath is accepted and coerced to its string representation.
func TestEmitEvent_BoolSourceID_StillCoerced(t *testing.T) {
	p := makeStrictParser(t)
	payload := []byte(`[{"eventType":"contact.created","id":true,"deliveryId":"dlv-001"}]`)
	events, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if err != nil {
		t.Fatalf("want success for bool SourceIDPath, got %v", err)
	}
	if len(events) != 1 {
		t.Fatalf("want 1 event, got %d", len(events))
	}
	if events[0].SourceID != "true" {
		t.Errorf("want SourceID=%q, got %q", "true", events[0].SourceID)
	}
}

// TestMatchRule_ObjectAtMatchPath_IsNonMatch verifies that the lenient
// match-path extractor treats a JSON object as a non-match (no error), not a
// malformed-payload error. This ensures the strict extractor change does not
// regress MatchPath semantics.
func TestMatchRule_ObjectAtMatchPath_IsNonMatch(t *testing.T) {
	// Parser with two rules: first uses MatchPath that will resolve to an object,
	// second is a catch-all. The object item must fall through to the catch-all.
	p, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchPath:          ".meta",
				MatchEquals:        "simple",
				EventTypeLiteral:   "should.not.match",
				ObjectKind:         "contact",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".deliveryId",
			},
			{
				// catch-all
				EventTypeLiteral:   "catch.all",
				ObjectKind:         "contact",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".deliveryId",
			},
		},
	})
	if err != nil {
		t.Fatalf("NewEnvelopeParser: %v", err)
	}
	// .meta is an object, not the string "simple" — first rule should not match.
	payload := []byte(`[{"meta":{"key":"val"},"id":"c-1","deliveryId":"dlv-1"}]`)
	events, parseErr := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if parseErr != nil {
		t.Fatalf("want no error for object at MatchPath (lenient), got %v", parseErr)
	}
	if len(events) != 1 {
		t.Fatalf("want 1 event (catch-all), got %d", len(events))
	}
	if events[0].EventType != "catch.all" {
		t.Errorf("want EventType=%q, got %q", "catch.all", events[0].EventType)
	}
}

// ── Classification-contract tests (NCRM-INT-11) ──────────────────────────────
//
// These tests verify the three-way classification:
//   composite at typed field  → ErrMalformedPayload (400, terminal)
//   absent/null/empty at typed field → ErrParserConfigDrift (503, retriable)
//   structural (bad JSON / non-array) → ErrMalformedPayload (400, terminal)

// makeClassificationParser returns a parser whose rule has SourceIDPath=".id",
// IdempotencyKeyPath=".deliveryId", and uses EventTypeLiteral so EventType is
// not a variable under test for SourceID/IdempotencyKey tests.
func makeClassificationParser(t *testing.T) *EnvelopeParser {
	t.Helper()
	p, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{{
			EventTypeLiteral:   "contact.created",
			ObjectKind:         "contact",
			SourceIDPath:       ".id",
			IdempotencyKeyPath: ".deliveryId",
		}},
	})
	if err != nil {
		t.Fatalf("makeClassificationParser: %v", err)
	}
	return p
}

// TestParse_AbsentSourceID_ConfigDrift_Not400 verifies that a SourceIDPath that
// resolves to absent (field missing in JSON) returns ErrParserConfigDrift — NOT
// ErrMalformedPayload. The delivery must not be terminal-400-dropped.
func TestParse_AbsentSourceID_ConfigDrift_Not400(t *testing.T) {
	p := makeClassificationParser(t)
	// ".id" is absent — only ".deliveryId" present.
	payload := []byte(`[{"deliveryId":"dlv-001"}]`)
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if err == nil {
		t.Fatal("want error for absent SourceIDPath, got nil")
	}
	if !errors.Is(err, connectorcapabilities.ErrParserConfigDrift) {
		t.Errorf("want ErrParserConfigDrift, got %v", err)
	}
	if errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Errorf("must NOT be ErrMalformedPayload (would terminal-400-drop a valid delivery): %v", err)
	}
}

// TestParse_NullSourceID_ConfigDrift_Not400 verifies that a SourceIDPath
// resolving to JSON null returns ErrParserConfigDrift (not ErrMalformedPayload).
func TestParse_NullSourceID_ConfigDrift_Not400(t *testing.T) {
	p := makeClassificationParser(t)
	payload := []byte(`[{"id":null,"deliveryId":"dlv-001"}]`)
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if err == nil {
		t.Fatal("want error for null SourceIDPath, got nil")
	}
	if !errors.Is(err, connectorcapabilities.ErrParserConfigDrift) {
		t.Errorf("want ErrParserConfigDrift, got %v", err)
	}
	if errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Errorf("must NOT be ErrMalformedPayload: %v", err)
	}
}

// TestParse_AbsentIdempotencyKey_ConfigDrift_Not400 verifies the same
// ErrParserConfigDrift classification for IdempotencyKeyPath.
func TestParse_AbsentIdempotencyKey_ConfigDrift_Not400(t *testing.T) {
	p := makeClassificationParser(t)
	// ".deliveryId" is absent.
	payload := []byte(`[{"id":"c-42"}]`)
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if err == nil {
		t.Fatal("want error for absent IdempotencyKeyPath, got nil")
	}
	if !errors.Is(err, connectorcapabilities.ErrParserConfigDrift) {
		t.Errorf("want ErrParserConfigDrift, got %v", err)
	}
	if errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Errorf("must NOT be ErrMalformedPayload: %v", err)
	}
}

// TestParse_CompositeSourceID_StillMalformed400 verifies that a composite JSON
// value (object/array) at SourceIDPath returns ErrMalformedPayload — the client
// sent the wrong shape at a contracted scalar path → terminal 400 (R9).
func TestParse_CompositeSourceID_StillMalformed400(t *testing.T) {
	p := makeClassificationParser(t)
	// ".id" is an object — composite at a scalar typed field → terminal 400.
	payload := []byte(`[{"id":{"nested":"val"},"deliveryId":"dlv-001"}]`)
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if err == nil {
		t.Fatal("want error for composite SourceIDPath, got nil")
	}
	if !errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Errorf("want ErrMalformedPayload for composite typed field, got %v", err)
	}
	if errors.Is(err, connectorcapabilities.ErrParserConfigDrift) {
		t.Errorf("must NOT be ErrParserConfigDrift for composite (client sent wrong shape): %v", err)
	}
}

// TestParse_StructuralBadJSON_Malformed400 verifies that an invalid JSON body
// returns ErrMalformedPayload (structural → terminal 400).
func TestParse_StructuralBadJSON_Malformed400(t *testing.T) {
	p := makeClassificationParser(t)
	payload := []byte(`not valid json`)
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if err == nil {
		t.Fatal("want error for invalid JSON, got nil")
	}
	if !errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Errorf("want ErrMalformedPayload for invalid JSON, got %v", err)
	}
}

// TestParse_StructuralNonArrayEventPath_Malformed400 verifies that a valid JSON
// body where EventArrayPath resolves to a non-array value returns ErrMalformedPayload.
func TestParse_StructuralNonArrayEventPath_Malformed400(t *testing.T) {
	// Parser whose EventArrayPath="." expects a root array, but we send an object.
	p := makeClassificationParser(t)
	payload := []byte(`{"id":"c-1","deliveryId":"dlv-1"}`) // object, not array
	_, err := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if err == nil {
		t.Fatal("want error for non-array event path, got nil")
	}
	if !errors.Is(err, connectorcapabilities.ErrMalformedPayload) {
		t.Errorf("want ErrMalformedPayload for non-array event path, got %v", err)
	}
}

// TestMatchRule_CompositeAtMatchPath_NonMatch verifies that a MatchPath-only rule
// (MatchEquals=="") where the path resolves to a composite JSON value (object/array)
// is treated as a non-match — the composite is NOT stringified and the rule is
// NOT selected. The item falls through to the next rule.
func TestMatchRule_CompositeAtMatchPath_NonMatch(t *testing.T) {
	// Two rules: first is MatchPath-only (no MatchEquals) which would incorrectly
	// select any non-empty value if composite were stringified; second is catch-all.
	p, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				// MatchPath-only: any non-empty, non-null value at .meta selects this rule.
				// If composite were stringified, {"key":"val"} would match (non-empty string).
				MatchPath:          ".meta",
				MatchEquals:        "",
				EventTypeLiteral:   "should.not.match",
				ObjectKind:         "contact",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".deliveryId",
			},
			{
				// catch-all — should match when the composite at .meta is a non-match.
				EventTypeLiteral:   "catch.all",
				ObjectKind:         "contact",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".deliveryId",
			},
		},
	})
	if err != nil {
		t.Fatalf("NewEnvelopeParser: %v", err)
	}
	// .meta is an object — composite at MatchPath-only rule must be a non-match.
	payload := []byte(`[{"meta":{"key":"val"},"id":"c-1","deliveryId":"dlv-1"}]`)
	events, parseErr := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if parseErr != nil {
		t.Fatalf("want no error, got %v", parseErr)
	}
	if len(events) != 1 {
		t.Fatalf("want 1 event (catch-all), got %d", len(events))
	}
	if events[0].EventType != "catch.all" {
		t.Errorf("composite at MatchPath-only rule must fall through to catch-all; got EventType=%q", events[0].EventType)
	}
}

// TestMatchRule_ArrayAtMatchPath_NonMatch verifies the same composite→non-match
// behavior when the value at MatchPath is a JSON array (not just an object).
func TestMatchRule_ArrayAtMatchPath_NonMatch(t *testing.T) {
	p, err := NewEnvelopeParser("ns", EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{
			{
				MatchPath:          ".tags",
				MatchEquals:        "",
				EventTypeLiteral:   "should.not.match",
				ObjectKind:         "contact",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".deliveryId",
			},
			{
				EventTypeLiteral:   "catch.all",
				ObjectKind:         "contact",
				SourceIDPath:       ".id",
				IdempotencyKeyPath: ".deliveryId",
			},
		},
	})
	if err != nil {
		t.Fatalf("NewEnvelopeParser: %v", err)
	}
	// .tags is an array — composite at MatchPath-only must be a non-match.
	payload := []byte(`[{"tags":["a","b"],"id":"c-2","deliveryId":"dlv-2"}]`)
	events, parseErr := p.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: payload})
	if parseErr != nil {
		t.Fatalf("want no error, got %v", parseErr)
	}
	if len(events) != 1 {
		t.Fatalf("want 1 event (catch-all), got %d", len(events))
	}
	if events[0].EventType != "catch.all" {
		t.Errorf("array at MatchPath-only rule must fall through to catch-all; got EventType=%q", events[0].EventType)
	}
}

type compositeParserTestConfig struct {
	config     EnvelopeMappingConfig
	composites []*IdempotencyKeyComposite
}

func compositeParserConfig(parts ...string) compositeParserTestConfig {
	compositeParts := make([]IdempotencyKeyPart, 0, len(parts))
	for _, path := range parts {
		compositeParts = append(compositeParts, IdempotencyKeyPart{Path: path})
	}
	return compositeParserTestConfig{
		config: EnvelopeMappingConfig{
			EventArrayPath: ".",
			Rules: []EnvelopeMappingRule{{
				EventTypeLiteral: "thing.updated",
				ObjectKind:       "thing",
				SourceIDPath:     ".objectId",
			}},
		},
		composites: []*IdempotencyKeyComposite{{
			Version:  "v1",
			Encoding: "sha256",
			Parts:    compositeParts,
		}},
	}
}

func parseCompositeKey(t *testing.T, cfg compositeParserTestConfig, payload string) (string, error) {
	t.Helper()
	parser, err := NewEnvelopeParserWithIdempotencyComposites("test-connector", cfg.config, cfg.composites)
	if err != nil {
		return "", err
	}
	events, err := parser.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(payload)})
	if err != nil {
		return "", err
	}
	if len(events) != 1 {
		t.Fatalf("events = %d, want 1", len(events))
	}
	return events[0].IdempotencyKey, nil
}

func TestCompositeIdempotency_LegacyCompatibilityAndSidecarShape(t *testing.T) {
	legacy := EnvelopeMappingConfig{
		EventArrayPath: ".",
		Rules: []EnvelopeMappingRule{{
			EventTypeLiteral:   "thing.updated",
			ObjectKind:         "thing",
			SourceIDPath:       ".objectId",
			IdempotencyKeyPath: ".eventId",
		}},
	}
	key, err := parseCompositeKey(t, compositeParserTestConfig{config: legacy}, `[{"objectId":"1","eventId":"legacy-event"}]`)
	if err != nil {
		t.Fatalf("legacy parse: %v", err)
	}
	if key != "legacy-event" {
		t.Fatalf("legacy IdempotencyKey = %q, want exact legacy value", key)
	}

	key, err = parseCompositeKey(t, compositeParserTestConfig{config: legacy, composites: []*IdempotencyKeyComposite{nil}}, `[{"objectId":"1","eventId":"legacy-event"}]`)
	if err != nil || key != "legacy-event" {
		t.Fatalf("full-length nil sidecar = %q, %v; want legacy-event", key, err)
	}

	if _, err := NewEnvelopeParserWithIdempotencyComposites("test-connector", legacy, []*IdempotencyKeyComposite{nil, nil}); err == nil || !strings.Contains(err.Error(), "length") {
		t.Fatalf("sidecar length mismatch error = %v", err)
	}
}

func TestCompositeIdempotency_RejectsOnePartObjectOrEventTypeIdentity(t *testing.T) {
	tests := []struct {
		name      string
		part      string
		configure func(*EnvelopeMappingRule)
		wantError string
	}{
		{
			name:      "source id",
			part:      ".objectId",
			configure: func(*EnvelopeMappingRule) {},
			wantError: "SourceIDPath",
		},
		{
			name: "event type",
			part: ".eventType",
			configure: func(rule *EnvelopeMappingRule) {
				rule.EventTypeLiteral = ""
				rule.EventTypeFromPath = ".eventType"
			},
			wantError: "EventTypeFromPath",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cfg := compositeParserConfig(tt.part)
			tt.configure(&cfg.config.Rules[0])
			_, err := NewEnvelopeParserWithIdempotencyComposites("test-connector", cfg.config, cfg.composites)
			if err == nil || !strings.Contains(err.Error(), tt.wantError) {
				t.Fatalf("error = %v, want substring %q", err, tt.wantError)
			}
		})
	}

	cfg := compositeParserConfig(".objectId", ".eventId")
	if _, err := NewEnvelopeParserWithIdempotencyComposites("test-connector", cfg.config, cfg.composites); err != nil {
		t.Fatalf("multi-part composite containing SourceIDPath must remain valid: %v", err)
	}
}

func TestCompositeIdempotency_MixedRulesKeepIndexAlignment(t *testing.T) {
	cfg := compositeParserTestConfig{
		config: EnvelopeMappingConfig{
			EventArrayPath: ".",
			Rules: []EnvelopeMappingRule{
				{
					MatchPath:          ".kind",
					MatchEquals:        "legacy",
					EventTypeLiteral:   "thing.legacy",
					ObjectKind:         "thing",
					SourceIDPath:       ".objectId",
					IdempotencyKeyPath: ".eventId",
				},
				{
					EventTypeLiteral: "thing.composite",
					ObjectKind:       "thing",
					SourceIDPath:     ".objectId",
				},
			},
		},
		composites: []*IdempotencyKeyComposite{
			nil,
			{Version: "v1", Encoding: "sha256", Parts: []IdempotencyKeyPart{{Path: ".eventId"}, {Path: ".occurredAt"}}},
		},
	}
	parser, err := NewEnvelopeParserWithIdempotencyComposites("test-connector", cfg.config, cfg.composites)
	if err != nil {
		t.Fatalf("NewEnvelopeParser: %v", err)
	}
	events, err := parser.Parse(context.Background(), &connectorcapabilities.ParserInput{Payload: []byte(`[
  {"kind":"legacy","objectId":"1","eventId":"legacy-1"},
  {"kind":"new","objectId":"2","eventId":"event-2","occurredAt":2}
]`)})
	if err != nil {
		t.Fatalf("Parse: %v", err)
	}
	if len(events) != 2 || events[0].IdempotencyKey != "legacy-1" || !strings.HasPrefix(events[1].IdempotencyKey, "cmp:v1:") {
		t.Fatalf("events = %#v, want index-aligned legacy then composite keys", events)
	}
}

func TestCompositeIdempotency_RejectsAmbiguousAndInvalidConfig(t *testing.T) {
	tests := []struct {
		name      string
		mutate    func(*compositeParserTestConfig)
		wantError string
	}{
		{
			name: "legacy and composite",
			mutate: func(cfg *compositeParserTestConfig) {
				cfg.config.Rules[0].IdempotencyKeyPath = ".eventId"
			},
			wantError: "exactly one",
		},
		{
			name: "unknown version",
			mutate: func(cfg *compositeParserTestConfig) {
				cfg.composites[0].Version = "v2"
			},
			wantError: "Version",
		},
		{
			name: "unknown encoding",
			mutate: func(cfg *compositeParserTestConfig) {
				cfg.composites[0].Encoding = "plain"
			},
			wantError: "Encoding",
		},
		{
			name: "empty parts",
			mutate: func(cfg *compositeParserTestConfig) {
				cfg.composites[0].Parts = nil
			},
			wantError: "Parts is empty",
		},
		{
			name: "canonical duplicate path",
			mutate: func(cfg *compositeParserTestConfig) {
				cfg.composites[0].Parts = []IdempotencyKeyPart{{Path: ".items[1]"}, {Path: ".items[01]"}}
			},
			wantError: "duplicates",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cfg := compositeParserConfig(".eventId", ".occurredAt")
			tt.mutate(&cfg)
			_, err := NewEnvelopeParserWithIdempotencyComposites("test-connector", cfg.config, cfg.composites)
			if err == nil || !strings.Contains(err.Error(), tt.wantError) {
				t.Fatalf("error = %v, want substring %q", err, tt.wantError)
			}
		})
	}
}

func TestCompositeIdempotency_DeterministicFramingOrderAndUnicode(t *testing.T) {
	payload := `[{"objectId":"1","left":"a|b\u0000雪","right":"c:d"}]`
	key, err := parseCompositeKey(t, compositeParserConfig(".left", ".right"), payload)
	if err != nil {
		t.Fatalf("parse composite: %v", err)
	}
	keyAgain, err := parseCompositeKey(t, compositeParserConfig(".left", ".right"), payload)
	if err != nil || keyAgain != key {
		t.Fatalf("repeat key = %q, %v; want %q", keyAgain, err, key)
	}
	if len(key) != 84 || !strings.HasPrefix(key, "cmp:v1:") {
		t.Fatalf("key = %q (len=%d), want cmp:v1 fixed length 84", key, len(key))
	}

	reordered, err := parseCompositeKey(t, compositeParserConfig(".right", ".left"), payload)
	if err != nil {
		t.Fatalf("parse reordered: %v", err)
	}
	if reordered == key {
		t.Fatal("component order must affect identity")
	}

	differentStructure, err := parseCompositeKey(t, compositeParserConfig(".first", ".second"), `[{"objectId":"1","first":"a","second":"|b\u0000雪c:d"}]`)
	if err != nil {
		t.Fatalf("parse different structure: %v", err)
	}
	if differentStructure == key {
		t.Fatal("length-framed values and canonical paths must prevent structural collision")
	}
	stringNumber, err := parseCompositeKey(t, compositeParserConfig(".left", ".right"), `[{"objectId":"1","left":"1","right":"x"}]`)
	if err != nil {
		t.Fatalf("parse string number: %v", err)
	}
	numericNumber, err := parseCompositeKey(t, compositeParserConfig(".left", ".right"), `[{"objectId":"1","left":1,"right":"x"}]`)
	if err != nil {
		t.Fatalf("parse numeric number: %v", err)
	}
	if stringNumber == numericNumber {
		t.Fatal("different scalar types must not collide")
	}
}

func TestCompositeIdempotency_NumbersUseExactCanonicalRationals(t *testing.T) {
	keyFor := func(number string) string {
		t.Helper()
		payload := fmt.Sprintf(`[{"objectId":"1","number":%s,"scope":"same"}]`, number)
		key, err := parseCompositeKey(t, compositeParserConfig(".number", ".scope"), payload)
		if err != nil {
			t.Fatalf("number %s: %v", number, err)
		}
		return key
	}

	if one, decimal, exponent := keyFor("1"), keyFor("1.0"), keyFor("1e0"); one != decimal || one != exponent {
		t.Fatalf("equivalent one encodings differ: 1=%q 1.0=%q 1e0=%q", one, decimal, exponent)
	}
	if zero, negativeZero := keyFor("0"), keyFor("-0"); zero != negativeZero {
		t.Fatalf("0 and -0 differ: %q vs %q", zero, negativeZero)
	}
	if integer, decimalInteger := keyFor("123456789012345678901234567890"), keyFor("123456789012345678901234567890.0"); integer != decimalInteger {
		t.Fatalf("exact large integer encodings differ: %q vs %q", integer, decimalInteger)
	}
	if exact, alternate := keyFor("123456789012345678901234567890.123400"), keyFor("1234567890123456789012345678901234e-4"); exact != alternate {
		t.Fatalf("exact large decimal encodings differ: %q vs %q", exact, alternate)
	}
	if keyFor("1.0000000000000000000000000000000000001") == keyFor("1") {
		t.Fatal("distinct exact numeric values collided")
	}
}

func TestCompositeIdempotency_ComponentFailuresAreClosedAndRedacted(t *testing.T) {
	tests := []struct {
		name       string
		payload    string
		want       error
		secretText string
	}{
		{name: "missing", payload: `[{"objectId":"1","scope":"s"}]`, want: connectorcapabilities.ErrParserConfigDrift},
		{name: "null", payload: `[{"objectId":"1","eventId":null,"scope":"s"}]`, want: connectorcapabilities.ErrParserConfigDrift},
		{name: "empty", payload: `[{"objectId":"1","eventId":"   ","scope":"s"}]`, want: connectorcapabilities.ErrParserConfigDrift, secretText: "   "},
		{name: "object", payload: `[{"objectId":"1","eventId":{"secret":"do-not-log"},"scope":"s"}]`, want: connectorcapabilities.ErrMalformedPayload, secretText: "do-not-log"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			_, err := parseCompositeKey(t, compositeParserConfig(".eventId", ".scope"), tt.payload)
			if !errors.Is(err, tt.want) {
				t.Fatalf("error = %v, want %v", err, tt.want)
			}
			if tt.secretText != "" && strings.Contains(err.Error(), tt.secretText) {
				t.Fatalf("error leaked component value: %v", err)
			}
		})
	}
}