package packs

import (
	"context"
	"fmt"
	"regexp"
	"strings"
	"testing"

	sqlmock "github.com/DATA-DOG/go-sqlmock"

	kernelconnector "github.com/nexuscrm/kernel/pkg/connector"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
)

// ---------------------------------------------------------------------------
// parseConnectorFieldMappingGroupsPayload tests
// ---------------------------------------------------------------------------

func TestParseConnectorFieldMappingGroupsPayload_Valid(t *testing.T) {
	raw := []byte(`[
		{
			"sourceObject": "Contact",
			"targetObject": "Contact",
			"actionType":   "syncContact",
			"direction":    "inbound",
			"mappings": [
				{"sourceField": "email", "targetField": "Email"}
			]
		}
	]`)
	groups, err := parseConnectorFieldMappingGroupsPayload(raw)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(groups) != 1 {
		t.Fatalf("expected 1 group, got %d", len(groups))
	}
	if groups[0].Direction != "inbound" {
		t.Errorf("Direction = %q, want inbound", groups[0].Direction)
	}
	if len(groups[0].Mappings) != 1 {
		t.Fatalf("expected 1 mapping, got %d", len(groups[0].Mappings))
	}
	if groups[0].Mappings[0].SourceField != "email" {
		t.Errorf("SourceField = %q, want email", groups[0].Mappings[0].SourceField)
	}
}

func TestParseConnectorFieldMappingInputs_DuplicateBusinessKeyCaseInsensitive(t *testing.T) {
	raw := []byte(`[
		{
			"sourceObject": "Contact",
			"targetObject": "Contact",
			"actionType":   "syncContact",
			"direction":    "inbound",
			"mappings": [
				{"sourceField": "Email", "targetField": "Email"},
				{"sourceField": "email", "targetField": "OtherEmail"}
			]
		}
	]`)

	if _, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload(raw, "HubSpot"); err == nil {
		t.Fatal("case-only duplicate connector field mapping business key should be rejected")
	}
}

func TestParseConnectorFieldMappingGroupsPayload_RejectsUnknownKey(t *testing.T) {
	// "unknown_key" is not in the struct; DisallowUnknownFields must reject it.
	raw := []byte(`[
		{
			"sourceObject": "Contact",
			"targetObject": "Contact",
			"actionType":   "syncContact",
			"direction":    "inbound",
			"unknown_key":  "oops",
			"mappings": [{"sourceField": "email", "targetField": "Email"}]
		}
	]`)
	_, err := parseConnectorFieldMappingGroupsPayload(raw)
	if err == nil {
		t.Fatal("expected error for unknown key, got nil")
	}
}

func TestParseConnectorFieldMappingGroupsPayload_RejectsTrailingGarbage(t *testing.T) {
	// Valid JSON followed by trailing content — EOF guard must catch it.
	raw := []byte(`[{"sourceObject":"A","targetObject":"B","actionType":"x","direction":"inbound","mappings":[]}] {"extra":"garbage"}`)
	_, err := parseConnectorFieldMappingGroupsPayload(raw)
	if err == nil {
		t.Fatal("expected error for trailing JSON token, got nil")
	}
}

func TestParseConnectorFieldMappingGroupsPayload_RejectsInvalidTrailingGarbage(t *testing.T) {
	// Valid JSON followed by invalid trailing bytes must fail closed too; the
	// decoder returns a syntax error rather than a token in this path.
	raw := []byte(`[{"sourceObject":"A","targetObject":"B","actionType":"x","direction":"inbound","mappings":[]}] @`)
	_, err := parseConnectorFieldMappingGroupsPayload(raw)
	if err == nil {
		t.Fatal("expected error for invalid trailing JSON, got nil")
	}
}

func TestParseConnectorFieldMappingGroupsPayload_EmptyArrayAllowed(t *testing.T) {
	raw := []byte(`[]`)
	groups, err := parseConnectorFieldMappingGroupsPayload(raw)
	if err != nil {
		t.Fatalf("empty array should be valid, got error: %v", err)
	}
	if len(groups) != 0 {
		t.Errorf("expected 0 groups, got %d", len(groups))
	}
}

// ---------------------------------------------------------------------------
// InitializeConnectorFieldMappingsFromPayloadForOwner tests (sqlmock)
// Now upsert-only: no UPDATE sweep inside Initialize.
// ---------------------------------------------------------------------------

// validPackPayload is a minimal valid connector_field_mapping payload for tests.
const validPackPayloadJSON = `[
	{
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"mappings": [
			{"sourceField": "email", "targetField": "Email"},
			{"sourceField": "phone", "targetField": "Phone"}
		]
	}
]`

const valueMapPackPayloadJSON = `[
	{
		"sourceObject": "Lead",
		"targetObject": "Lead",
		"actionType":   "syncLead",
		"direction":    "inbound",
		"mappings": [
			{
				"sourceField": "status",
				"targetField": "Status__c",
				"valueMap": {"lead": "Open", "customer": "Won"}
			}
		]
	}
]`

func TestParseConnectorFieldMappingInputsFromPayload_ValueMapCanonicalized(t *testing.T) {
	rows, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(valueMapPackPayloadJSON), "test-pack")
	if err != nil {
		t.Fatalf("kernelconnector.ParseConnectorFieldMappingInputsFromPayload: %v", err)
	}
	if len(rows) != 1 {
		t.Fatalf("expected 1 row, got %d", len(rows))
	}
	if rows[0].ValueMap != `{"customer":"Won","lead":"Open"}` {
		t.Fatalf("ValueMap = %q, want canonical JSON", rows[0].ValueMap)
	}
}

type connectorFieldMappingTargetRow struct {
	objectAPIName string
	fieldAPIName  string
	updateable    bool
}

func connectorFieldMappingTarget(objectAPIName, fieldAPIName string) connectorFieldMappingTargetRow {
	return connectorFieldMappingTargetRow{objectAPIName: objectAPIName, fieldAPIName: fieldAPIName, updateable: true}
}

func connectorFieldMappingReadOnlyTarget(objectAPIName, fieldAPIName string) connectorFieldMappingTargetRow {
	return connectorFieldMappingTargetRow{objectAPIName: objectAPIName, fieldAPIName: fieldAPIName, updateable: false}
}

func expectConnectorFieldMappingTargetSchema(mock sqlmock.Sqlmock, targets ...connectorFieldMappingTargetRow) {
	rows := sqlmock.NewRows([]string{"ObjectAPIName", "FieldAPIName", "Updateable"})
	for _, target := range targets {
		rows.AddRow(target.objectAPIName, target.fieldAPIName, target.updateable)
	}
	mock.ExpectQuery("FROM `_System_Field`").
		WillReturnRows(rows)
}

func TestInitializeConnectorFieldMappings_WritesRowsInsideTx(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	// Target schema validation, then upsert-only: BEGIN + 2 INSERTs + COMMIT.
	// No sweep UPDATE.
	expectConnectorFieldMappingTargetSchema(mock,
		connectorFieldMappingTarget("Contact", "Email"),
		connectorFieldMappingTarget("Contact", "Phone"),
	)
	mock.ExpectBegin()
	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ObjectFieldMapping")).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ObjectFieldMapping")).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectCommit()

	rows, err := InitializeConnectorFieldMappingsFromPayloadForOwner(db, []byte(validPackPayloadJSON), "test-pack")
	if err != nil {
		t.Fatalf("InitializeConnectorFieldMappingsFromPayloadForOwner: %v", err)
	}
	if len(rows) != 2 {
		t.Errorf("expected 2 declared rows returned, got %d", len(rows))
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestInitializeConnectorFieldMappings_EmptyPayloadNoSQL(t *testing.T) {
	// Empty payload: no SQL issued at all (no tx, no upserts, no sweep).
	// The sweeper handles stale-row deactivation separately.
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	rows, err := InitializeConnectorFieldMappingsFromPayloadForOwner(db, []byte(`[]`), "test-pack")
	if err != nil {
		t.Fatalf("empty payload: %v", err)
	}
	if len(rows) != 0 {
		t.Errorf("expected 0 rows for empty payload, got %d", len(rows))
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unexpected sql call: %v", err)
	}
}

func TestInitializeConnectorFieldMappings_OneInvalidRowNoWrite(t *testing.T) {
	// A payload containing one row with blank Direction must fail before any DB call.
	raw := []byte(`[
		{
			"sourceObject": "Contact",
			"targetObject": "Contact",
			"actionType":   "syncContact",
			"direction":    "",
			"mappings": [{"sourceField": "email", "targetField": "Email"}]
		}
	]`)

	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	// No SQL at all should be issued.
	_, err = InitializeConnectorFieldMappingsFromPayloadForOwner(db, raw, "test-pack")
	if err == nil {
		t.Fatal("expected error for blank direction row, got nil")
	}

	// Confirm no SQL was issued (all expectations would have been unmatched otherwise).
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unexpected sql call: %v", err)
	}
}

func TestInitializeConnectorFieldMappings_TargetFieldMissingNoWrite(t *testing.T) {
	raw := []byte(`[
		{
			"sourceObject": "ExternalContact",
			"targetObject": "Contact",
			"actionType":   "syncContact",
			"direction":    "inbound",
			"mappings": [{"sourceField": "email_address", "targetField": "MissingField"}]
		}
	]`)

	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Contact", "Email"))

	_, err = InitializeConnectorFieldMappingsFromPayloadForOwner(db, raw, "test-pack")
	if err == nil {
		t.Fatal("expected target schema validation error, got nil")
	}
	if !strings.Contains(err.Error(), "TargetField") {
		t.Errorf("error should mention TargetField, got: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestInitializeConnectorFieldMappings_TargetFieldNotUpdateableNoWrite(t *testing.T) {
	raw := []byte(`[
		{
			"sourceObject": "ExternalContact",
			"targetObject": "Contact",
			"actionType":   "syncContact",
			"direction":    "inbound",
			"mappings": [{"sourceField": "email_address", "targetField": "CalculatedEmail"}]
		}
	]`)

	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingReadOnlyTarget("Contact", "CalculatedEmail"))

	_, err = InitializeConnectorFieldMappingsFromPayloadForOwner(db, raw, "test-pack")
	if err == nil {
		t.Fatal("expected non-updateable target validation error, got nil")
	}
	if !strings.Contains(err.Error(), "not updateable") {
		t.Errorf("error should mention non-updateable TargetField, got: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestInitializeConnectorFieldMappings_RollbackOnMappingWriteFailure(t *testing.T) {
	// First upsert succeeds, second fails — the transaction must be rolled back.
	raw := []byte(`[
		{
			"sourceObject": "Contact",
			"targetObject": "Contact",
			"actionType":   "syncContact",
			"direction":    "inbound",
			"mappings": [
				{"sourceField": "email", "targetField": "Email"},
				{"sourceField": "phone", "targetField": "Phone"}
			]
		}
	]`)

	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectConnectorFieldMappingTargetSchema(mock,
		connectorFieldMappingTarget("Contact", "Email"),
		connectorFieldMappingTarget("Contact", "Phone"),
	)
	mock.ExpectBegin()
	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ObjectFieldMapping")).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ObjectFieldMapping")).
		WillReturnError(fmt.Errorf("db error"))
	mock.ExpectRollback()

	_, err = InitializeConnectorFieldMappingsFromPayloadForOwner(db, raw, "test-pack")
	if err == nil {
		t.Fatal("expected error from second upsert, got nil")
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestInitializeConnectorFieldMappings_IdempotentReapply(t *testing.T) {
	// A second call with the same payload issues the same SQL (upsert semantics).
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	for range 2 {
		expectConnectorFieldMappingTargetSchema(mock,
			connectorFieldMappingTarget("Contact", "Email"),
			connectorFieldMappingTarget("Contact", "Phone"),
		)
		mock.ExpectBegin()
		mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ObjectFieldMapping")).
			WillReturnResult(sqlmock.NewResult(1, 1))
		mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ObjectFieldMapping")).
			WillReturnResult(sqlmock.NewResult(1, 1))
		mock.ExpectCommit()
	}

	for i := range 2 {
		_, err := InitializeConnectorFieldMappingsFromPayloadForOwner(db, []byte(validPackPayloadJSON), "test-pack")
		if err != nil {
			t.Fatalf("apply %d: %v", i+1, err)
		}
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestInitializeConnectorFieldMappings_EmptyOwnerRejected(t *testing.T) {
	db, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	_, err = InitializeConnectorFieldMappingsFromPayloadForOwner(db, []byte(`[]`), "")
	if err == nil {
		t.Fatal("expected error for blank owner, got nil")
	}
}

// ---------------------------------------------------------------------------
// SweepConnectorFieldMappingsForOwner tests (pack-level owner-wide sweep)
// ---------------------------------------------------------------------------

// payloadBJSON drops the "phone" row compared to validPackPayloadJSON.
const payloadBJSON = `[
	{
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"mappings": [
			{"sourceField": "email", "targetField": "Email"}
		]
	}
]`

// TestSweep_PrunesDroppedRow asserts that SweepConnectorFieldMappingsForOwner
// issues an owner-wide UPDATE Id NOT IN (...) when given a non-empty declared set.
func TestSweep_PrunesDroppedRow(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	declared, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(payloadBJSON), "test-pack")
	if err != nil {
		t.Fatalf("parse: %v", err)
	}

	mock.ExpectExec(`(?s)UPDATE _System_ObjectFieldMapping.*Id NOT IN`).
		WillReturnResult(sqlmock.NewResult(0, 1)) // stale phone row deactivated

	if err := SweepConnectorFieldMappingsForOwner(context.Background(), db, "test-pack", declared); err != nil {
		t.Fatalf("SweepConnectorFieldMappingsForOwner: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// TestSweep_EmptyDeclaredDeactivatesAll asserts that an empty declared slice
// (component fully dropped) issues an owner-wide deactivate without NOT IN.
func TestSweep_EmptyDeclaredDeactivatesAll(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	mock.ExpectExec(`(?s)UPDATE _System_ObjectFieldMapping.*WHERE Source = \?.*Direction <> '` + legacyObjectFieldMappingOwner + `'`).
		WillReturnResult(sqlmock.NewResult(2, 2))

	if err := SweepConnectorFieldMappingsForOwner(context.Background(), db, "test-pack", nil); err != nil {
		t.Fatalf("SweepConnectorFieldMappingsForOwner with empty declared: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// twoTuplePayloadJSON has two distinct (SourceObject, TargetObject, ActionType, Direction) tuples.
const twoTuplePayloadJSON = `[
	{
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"mappings": [
			{"sourceField": "email", "targetField": "Email"}
		]
	},
	{
		"sourceObject": "Lead",
		"targetObject": "Lead",
		"actionType":   "syncLead",
		"direction":    "inbound",
		"mappings": [
			{"sourceField": "company", "targetField": "Company"}
		]
	}
]`

// oneTuplePayloadJSON drops the syncLead tuple entirely compared to twoTuplePayloadJSON.
const oneTuplePayloadJSON = `[
	{
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"mappings": [
			{"sourceField": "email", "targetField": "Email"}
		]
	}
]`

// TestSweep_DroppedTuple asserts that SweepConnectorFieldMappingsForOwner catches
// dropped tuples via the owner-wide NOT IN sweep.
func TestSweep_DroppedTuple(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	declared, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(oneTuplePayloadJSON), "test-pack")
	if err != nil {
		t.Fatalf("parse: %v", err)
	}

	// Owner-wide sweep: catches the dropped syncLead.company row.
	mock.ExpectExec(`(?s)UPDATE _System_ObjectFieldMapping.*Id NOT IN`).
		WillReturnResult(sqlmock.NewResult(1, 1))

	if err := SweepConnectorFieldMappingsForOwner(context.Background(), db, "test-pack", declared); err != nil {
		t.Fatalf("SweepConnectorFieldMappingsForOwner: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// TestSweep_MultiComponentSafety proves that running a union sweep with rows from
// both components keeps both components' rows. Component A contributes syncContact/email;
// component B contributes syncLead/company. The union sweep must NOT delete either.
func TestSweep_MultiComponentSafety(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	// Parse both components' declared rows into the union.
	declaredA, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(oneTuplePayloadJSON), "test-pack")
	if err != nil {
		t.Fatalf("parse A: %v", err)
	}
	const compBPayload = `[{"sourceObject":"Lead","targetObject":"Lead","actionType":"syncLead","direction":"inbound","mappings":[{"sourceField":"company","targetField":"Company"}]}]`
	declaredB, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(compBPayload), "test-pack")
	if err != nil {
		t.Fatalf("parse B: %v", err)
	}
	union := append(declaredA, declaredB...)

	// The sweep with the full union fires an UPDATE Id NOT IN (both ids).
	// No rows should be deactivated (RowsAffected=0 means nothing stale).
	mock.ExpectExec(`(?s)UPDATE _System_ObjectFieldMapping.*Id NOT IN`).
		WillReturnResult(sqlmock.NewResult(0, 0))

	if err := SweepConnectorFieldMappingsForOwner(context.Background(), db, "test-pack", union); err != nil {
		t.Fatalf("SweepConnectorFieldMappingsForOwner: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// ---------------------------------------------------------------------------
// VerifyConnectorFieldMappingsFromPayloadForOwner tests (sqlmock)
// Per-component verify: checks own rows are live; no stale-row check.
// ---------------------------------------------------------------------------

// resolverColsV is the same set as resolverCols in connector_field_mapping_resolver_test.go.
var resolverColsV = []string{
	"Source", "SourceObject", "TargetObject", "ActionType",
	"SourceField", "TargetField", "DefaultValue", "SortOrder",
	"Direction", "TransformExpression", "ConflictPolicy", "ValueMap",
	"OnUnmappedValue",
}

func expectExecutableProjectionState(mock sqlmock.Sqlmock, owner, sourceObject, targetObject, actionType, sourceField, direction string) {
	mock.ExpectQuery(`(?s)SELECT ProjectionOnly.*FROM _System_ObjectFieldMapping`).
		WithArgs(owner, sourceObject, targetObject, actionType, sourceField, direction).
		WillReturnRows(sqlmock.NewRows([]string{"ProjectionOnly"}).AddRow(false))
}

func TestVerifyConnectorFieldMappings_PassesAfterApply(t *testing.T) {
	// Simulate: after apply the resolver returns both expected rows with the
	// declaration-order SortOrder values (email=0, phone=1).
	// Per-component verify does NOT run a stale-row check.
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	rows := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Contact", "Contact", "syncContact",
			"email", "Email", nil, 0, connector.DirectionInbound, nil, nil, nil, nil).
		AddRow("test-pack", "Contact", "Contact", "syncContact",
			"phone", "Phone", nil, 1, connector.DirectionInbound, nil, nil, nil, nil)

	expectConnectorFieldMappingTargetSchema(mock,
		connectorFieldMappingTarget("Contact", "Email"),
		connectorFieldMappingTarget("Contact", "Phone"),
	)
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Contact", "Contact", "syncContact", connector.DirectionInbound).
		WillReturnRows(rows)
	expectExecutableProjectionState(mock, "test-pack", "Contact", "Contact", "syncContact", "email", connector.DirectionInbound)
	expectExecutableProjectionState(mock, "test-pack", "Contact", "Contact", "syncContact", "phone", connector.DirectionInbound)

	if err := VerifyConnectorFieldMappingsFromPayloadForOwner(
		context.Background(), db, []byte(validPackPayloadJSON), "test-pack",
	); err != nil {
		t.Fatalf("verify should pass after apply, got: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorFieldMappings_FailsOnDrift(t *testing.T) {
	// Drift: phone→Phone is missing from the live result.
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	rows := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Contact", "Contact", "syncContact",
			"email", "Email", nil, 0, connector.DirectionInbound, nil, nil, nil, nil)
	// phone row is absent — drift.

	expectConnectorFieldMappingTargetSchema(mock,
		connectorFieldMappingTarget("Contact", "Email"),
		connectorFieldMappingTarget("Contact", "Phone"),
	)
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Contact", "Contact", "syncContact", connector.DirectionInbound).
		WillReturnRows(rows)
	expectExecutableProjectionState(mock, "test-pack", "Contact", "Contact", "syncContact", "email", connector.DirectionInbound)

	err = VerifyConnectorFieldMappingsFromPayloadForOwner(
		context.Background(), db, []byte(validPackPayloadJSON), "test-pack",
	)
	if err == nil {
		t.Fatal("expected drift error for missing phone row, got nil")
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorFieldMappings_TargetObjectMissingBeforeResolve(t *testing.T) {
	raw := []byte(`[
		{
			"sourceObject": "ExternalLead",
			"targetObject": "Lead",
			"actionType":   "syncLead",
			"direction":    "inbound",
			"mappings": [{"sourceField": "email_address", "targetField": "Email"}]
		}
	]`)

	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Contact", "Email"))

	err = VerifyConnectorFieldMappingsFromPayloadForOwner(context.Background(), db, raw, "test-pack")
	if err == nil {
		t.Fatal("expected target schema validation error, got nil")
	}
	if !strings.Contains(err.Error(), "TargetObject") {
		t.Errorf("error should mention TargetObject, got: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorFieldMappings_TargetFieldNotUpdateableBeforeResolve(t *testing.T) {
	raw := []byte(`[
		{
			"sourceObject": "ExternalLead",
			"targetObject": "Lead",
			"actionType":   "syncLead",
			"direction":    "inbound",
			"mappings": [{"sourceField": "score", "targetField": "FormulaScore"}]
		}
	]`)

	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingReadOnlyTarget("Lead", "FormulaScore"))

	err = VerifyConnectorFieldMappingsFromPayloadForOwner(context.Background(), db, raw, "test-pack")
	if err == nil {
		t.Fatal("expected non-updateable target validation error, got nil")
	}
	if !strings.Contains(err.Error(), "not updateable") {
		t.Errorf("error should mention non-updateable TargetField, got: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestValidateConnectorFieldMappingTargets_RejectsNonUpdateableTargetField(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingReadOnlyTarget("Contact", "FormulaEmail"))

	rows := []connectorFieldMappingInput{{
		Source:       "test-pack",
		SourceObject: "ExternalContactPayload",
		TargetObject: "Contact",
		ActionType:   "syncContact",
		SourceField:  "email_address",
		TargetField:  "FormulaEmail",
		Direction:    connector.DirectionInbound,
	}}
	err = validateConnectorFieldMappingTargets(context.Background(), db, rows)
	if err == nil {
		t.Fatal("expected non-updateable target validation error, got nil")
	}
	if !strings.Contains(err.Error(), "TargetField") || !strings.Contains(err.Error(), "not updateable") {
		t.Errorf("error should mention non-updateable TargetField, got: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestValidateConnectorFieldMappingTargets_DoesNotValidateSourceSideNames(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Contact", "Email"))

	rows := []connectorFieldMappingInput{{
		Source:       "test-pack",
		SourceObject: "ExternalContactPayload",
		TargetObject: "Contact",
		ActionType:   "syncContact",
		SourceField:  "email_address",
		TargetField:  "Email",
		Direction:    connector.DirectionInbound,
	}}
	if err := validateConnectorFieldMappingTargets(context.Background(), db, rows); err != nil {
		t.Fatalf("source-side names should remain external/opaque, got: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestValidateConnectorFieldMappingTargetsProjectionOnlySkipsLocalSchema(t *testing.T) {
	rows := []connectorFieldMappingInput{{
		Source:         "test-pack",
		SourceObject:   "contacts",
		TargetObject:   "Lead",
		ActionType:     "syncContact",
		SourceField:    "legacy_source_id",
		Direction:      connector.DirectionInbound,
		ProjectionOnly: true,
	}}
	if err := validateConnectorFieldMappingTargets(context.Background(), nil, rows); err != nil {
		t.Fatalf("projection-only row required a local target schema: %v", err)
	}
}

func TestUpsertConnectorFieldMappingPersistsProjectionOnly(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	mock.ExpectExec(`(?s)INSERT INTO _System_ObjectFieldMapping.*SourceLookupField\s*=\s*CASE WHEN VALUES\(ProjectionOnly\) = 1 THEN NULL ELSE SourceLookupField END.*IsCustomized\s*=\s*CASE WHEN VALUES\(ProjectionOnly\) = 1 THEN 0 ELSE IsCustomized END.*ProjectionOnly\s*=\s*VALUES\(ProjectionOnly\)`).
		WithArgs(
			sqlmock.AnyArg(), "contacts", "Lead", "syncContact", "legacy_source_id", "",
			nil, nil, 0, "test-pack", connector.DirectionInbound, nil, nil, nil, true,
			sqlmock.AnyArg(), sqlmock.AnyArg(),
		).
		WillReturnResult(sqlmock.NewResult(1, 1))

	if err := upsertConnectorFieldMapping(context.Background(), db, connectorFieldMappingInput{
		Source:         "test-pack",
		SourceObject:   "contacts",
		TargetObject:   "Lead",
		ActionType:     "syncContact",
		SourceField:    "legacy_source_id",
		Direction:      connector.DirectionInbound,
		ProjectionOnly: true,
	}); err != nil {
		t.Fatalf("upsert projection-only row: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func TestVerifyConnectorFieldMappingsProjectionOnlyReadsPhysicalRow(t *testing.T) {
	raw := []byte(`[{
		"sourceObject":"contacts",
		"targetObject":"Lead",
		"actionType":"syncContact",
		"direction":"inbound",
		"mappings":[{"sourceField":"legacy_source_id","projectionOnly":true}]
	}]`)
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	mock.ExpectQuery(`(?s)SELECT TargetField, SourceLookupField, Direction, TransformExpression, ConflictPolicy, DefaultValue, ValueMap, OnUnmappedValue, SortOrder, ProjectionOnly, IsCustomized, IsActive, IsDeleted.*FROM _System_ObjectFieldMapping`).
		WithArgs("test-pack", "contacts", "Lead", "syncContact", "legacy_source_id", connector.DirectionInbound).
		WillReturnRows(sqlmock.NewRows([]string{
			"TargetField", "SourceLookupField", "Direction", "TransformExpression", "ConflictPolicy", "DefaultValue", "ValueMap", "OnUnmappedValue", "SortOrder", "ProjectionOnly", "IsCustomized", "IsActive", "IsDeleted",
		}).AddRow("", nil, connector.DirectionInbound, nil, nil, nil, nil, nil, 0, true, false, true, false))

	if err := VerifyConnectorFieldMappingsFromPayloadForOwner(context.Background(), db, raw, "test-pack"); err != nil {
		t.Fatalf("verify projection-only row: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func TestVerifyConnectorFieldMappingsProjectionOnlyRejectsPhysicalDrift(t *testing.T) {
	raw := []byte(`[{
		"sourceObject":"contacts",
		"targetObject":"Lead",
		"actionType":"syncContact",
		"direction":"inbound",
		"mappings":[{"sourceField":"legacy_source_id","projectionOnly":true}]
	}]`)
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	mock.ExpectQuery(`(?s)SELECT TargetField, SourceLookupField, Direction, TransformExpression, ConflictPolicy, DefaultValue, ValueMap, OnUnmappedValue, SortOrder, ProjectionOnly, IsCustomized, IsActive, IsDeleted.*FROM _System_ObjectFieldMapping`).
		WithArgs("test-pack", "contacts", "Lead", "syncContact", "legacy_source_id", connector.DirectionInbound).
		WillReturnRows(sqlmock.NewRows([]string{
			"TargetField", "SourceLookupField", "Direction", "TransformExpression", "ConflictPolicy", "DefaultValue", "ValueMap", "OnUnmappedValue", "SortOrder", "ProjectionOnly", "IsCustomized", "IsActive", "IsDeleted",
		}).AddRow("", "LegacyTargetLookup", connector.DirectionInbound, nil, nil, nil, nil, nil, 0, true, true, true, false))

	err = VerifyConnectorFieldMappingsFromPayloadForOwner(context.Background(), db, raw, "test-pack")
	if err == nil || !strings.Contains(err.Error(), "physical row drifted") {
		t.Fatalf("drift error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func TestVerifyConnectorFieldMappingsExecutableRejectsProjectionOnlyPhysicalDrift(t *testing.T) {
	raw := []byte(`[{
		"sourceObject":"Contact",
		"targetObject":"Contact",
		"actionType":"syncContact",
		"direction":"inbound",
		"mappings":[{"sourceField":"email","targetField":"Email"}]
	}]`)
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Contact", "Email"))
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Contact", "Contact", "syncContact", connector.DirectionInbound).
		WillReturnRows(sqlmock.NewRows(resolverColsV).AddRow(
			"test-pack", "Contact", "Contact", "syncContact",
			"email", "Email", nil, 0, connector.DirectionInbound, nil, nil, nil, nil,
		))
	mock.ExpectQuery(`(?s)SELECT ProjectionOnly.*FROM _System_ObjectFieldMapping`).
		WithArgs("test-pack", "Contact", "Contact", "syncContact", "email", connector.DirectionInbound).
		WillReturnRows(sqlmock.NewRows([]string{"ProjectionOnly"}).AddRow(true))

	err = VerifyConnectorFieldMappingsFromPayloadForOwner(context.Background(), db, raw, "test-pack")
	if err == nil || !strings.Contains(err.Error(), "ProjectionOnly") {
		t.Fatalf("projection drift error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func TestValidateConnectorFieldMappingTargets_DirectionAwareLocalSide(t *testing.T) {
	cases := []struct {
		name      string
		row       connectorFieldMappingInput
		wantError string
	}{
		{
			name: "outbound target stays remote",
			row:  connectorFieldMappingInput{Source: "crm-flow-pack", SourceObject: "Contact", TargetObject: "hubspotContact", ActionType: "PushToConnector", SourceField: "Email", TargetField: "email", Direction: connector.DirectionOutbound},
		},
		{
			name:      "outbound source is local",
			row:       connectorFieldMappingInput{Source: "crm-flow-pack", SourceObject: "Contact", TargetObject: "hubspotContact", ActionType: "PushToConnector", SourceField: "MissingLocalField", TargetField: "email", Direction: connector.DirectionOutbound},
			wantError: "SourceField",
		},
		{
			name:      "bidirectional target is local",
			row:       connectorFieldMappingInput{Source: "test-pack", SourceObject: "Contact", TargetObject: "RemoteContact", ActionType: "syncContact", SourceField: "Email", TargetField: "email", Direction: connector.DirectionBidirectional},
			wantError: "TargetObject",
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock.New: %v", err)
			}
			defer db.Close()
			expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Contact", "Email"))

			err = validateConnectorFieldMappingTargets(context.Background(), db, []connectorFieldMappingInput{tc.row})
			if tc.wantError == "" && err != nil {
				t.Fatalf("validateConnectorFieldMappingTargets: %v", err)
			}
			if tc.wantError != "" && (err == nil || !strings.Contains(err.Error(), tc.wantError)) {
				t.Fatalf("error = %v, want substring %q", err, tc.wantError)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sql expectations unmet: %v", err)
			}
		})
	}
}

// ---------------------------------------------------------------------------
// VerifyNoStaleConnectorFieldMappingsForOwner tests (pack-level stale-row check)
// ---------------------------------------------------------------------------

// TestVerifySweep_NoStaleRowsPass asserts the stale-row verify passes when all
// live owner rows are in the declared set.
func TestVerifySweep_NoStaleRowsPass(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	declared, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(validPackPayloadJSON), "test-pack")
	if err != nil {
		t.Fatalf("parse: %v", err)
	}

	staleCols := []string{"Id", "SourceObject", "TargetObject", "ActionType", "SourceField", "Direction"}
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack").
		WillReturnRows(sqlmock.NewRows(staleCols)) // no stale rows

	if err := VerifyNoStaleConnectorFieldMappingsForOwner(context.Background(), db, "test-pack", declared); err != nil {
		t.Fatalf("expected no stale rows, got: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// TestVerifySweep_ReportsStaleRows asserts the stale-row verify fails when the
// DB contains an active connector row not in the declared set.
func TestVerifySweep_ReportsStaleRows(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	declared, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(payloadBJSON), "test-pack")
	if err != nil {
		t.Fatalf("parse: %v", err)
	}

	staleCols := []string{"Id", "SourceObject", "TargetObject", "ActionType", "SourceField", "Direction"}
	stalePhoneID := connectorFieldMappingID("test-pack", "Contact", "Contact", "syncContact", "phone", connector.DirectionInbound)
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack").
		WillReturnRows(sqlmock.NewRows(staleCols).
			AddRow(stalePhoneID, "Contact", "Contact", "syncContact", "phone", connector.DirectionInbound))

	err = VerifyNoStaleConnectorFieldMappingsForOwner(context.Background(), db, "test-pack", declared)
	if err == nil {
		t.Fatal("expected stale-row error, got nil")
	}
	if !strings.Contains(err.Error(), "stale connector row") {
		t.Errorf("error should mention stale connector row, got: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// TestVerifySweep_DroppedTuple asserts that a dropped (SourceObject, TargetObject,
// ActionType, Direction) tuple is caught as stale drift by the pack-level verify.
func TestVerifySweep_DroppedTuple(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	// Payload B (oneTuplePayloadJSON): only syncContact/email is declared.
	declared, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(oneTuplePayloadJSON), "test-pack")
	if err != nil {
		t.Fatalf("parse: %v", err)
	}

	staleCols := []string{"Id", "SourceObject", "TargetObject", "ActionType", "SourceField", "Direction"}
	staleLeadID := connectorFieldMappingID("test-pack", "Lead", "Lead", "syncLead", "company", connector.DirectionInbound)
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack").
		WillReturnRows(sqlmock.NewRows(staleCols).
			AddRow(staleLeadID, "Lead", "Lead", "syncLead", "company", connector.DirectionInbound))

	err = VerifyNoStaleConnectorFieldMappingsForOwner(context.Background(), db, "test-pack", declared)
	if err == nil {
		t.Fatal("expected stale-row drift error for dropped tuple, got nil")
	}
	if !strings.Contains(err.Error(), "stale connector row") {
		t.Errorf("error should mention stale connector row, got: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// TestVerifySweep_EmptyDeclaredReportsAllRowsStale asserts that when declared
// is empty, any live owner row is reported as stale (dropped-component case).
func TestVerifySweep_EmptyDeclaredReportsAllRowsStale(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	staleCols := []string{"Id", "SourceObject", "TargetObject", "ActionType", "SourceField", "Direction"}
	emailID := connectorFieldMappingID("test-pack", "Contact", "Contact", "syncContact", "email", connector.DirectionInbound)
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack").
		WillReturnRows(sqlmock.NewRows(staleCols).
			AddRow(emailID, "Contact", "Contact", "syncContact", "email", connector.DirectionInbound))

	err = VerifyNoStaleConnectorFieldMappingsForOwner(context.Background(), db, "test-pack", nil)
	if err == nil {
		t.Fatal("expected stale-row error for empty declared, got nil")
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// ---------------------------------------------------------------------------
// Part A: SortOrder assignment tests
// ---------------------------------------------------------------------------

// threeRowPayloadJSON has one group with three mappings in declaration order.
const threeRowPayloadJSON = `[
	{
		"sourceObject": "Account",
		"targetObject": "Account",
		"actionType":   "syncAccount",
		"direction":    "outbound",
		"mappings": [
			{"sourceField": "name",    "targetField": "Name"},
			{"sourceField": "phone",   "targetField": "Phone"},
			{"sourceField": "website", "targetField": "Website"}
		]
	}
]`

// TestParseConnectorFieldMappingInputs_AssignsSortOrderByDeclarationOrder verifies
// that a multi-row payload yields rows with SortOrder 0, 1, 2, ... in declaration order.
func TestParseConnectorFieldMappingInputs_AssignsSortOrderByDeclarationOrder(t *testing.T) {
	rows, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(threeRowPayloadJSON), "test-pack")
	if err != nil {
		t.Fatalf("kernelconnector.ParseConnectorFieldMappingInputsFromPayload: %v", err)
	}
	if len(rows) != 3 {
		t.Fatalf("expected 3 rows, got %d", len(rows))
	}
	for i, row := range rows {
		if row.SortOrder != i {
			t.Errorf("rows[%d].SortOrder = %d, want %d (source=%s)", i, row.SortOrder, i, row.SourceField)
		}
	}
}

// TestParseConnectorFieldMappingInputs_MultiGroupSortOrderContinues verifies that
// SortOrder is a global counter across groups, not reset per group.
func TestParseConnectorFieldMappingInputs_MultiGroupSortOrderContinues(t *testing.T) {
	rows, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(twoTuplePayloadJSON), "test-pack")
	if err != nil {
		t.Fatalf("kernelconnector.ParseConnectorFieldMappingInputsFromPayload: %v", err)
	}
	// twoTuplePayloadJSON: group1 has 1 mapping (email→Email), group2 has 1 mapping (company→Company).
	// Declaration order: email (SortOrder=0), company (SortOrder=1).
	if len(rows) != 2 {
		t.Fatalf("expected 2 rows, got %d", len(rows))
	}
	for i, row := range rows {
		if row.SortOrder != i {
			t.Errorf("rows[%d] (%s).SortOrder = %d, want %d", i, row.SourceField, row.SortOrder, i)
		}
	}
}

// TestInitializeConnectorFieldMappings_PersistsSortOrder verifies that after
// Initialize, the returned rows carry declaration-order SortOrder values.
// This asserts at the row-building level (the sqlmock DB does not inspect bind
// values, but the returned []connectorFieldMappingInput must carry the right SortOrder).
func TestInitializeConnectorFieldMappings_PersistsSortOrder(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	// validPackPayloadJSON has two rows: email (SortOrder=0), phone (SortOrder=1).
	expectConnectorFieldMappingTargetSchema(mock,
		connectorFieldMappingTarget("Contact", "Email"),
		connectorFieldMappingTarget("Contact", "Phone"),
	)
	mock.ExpectBegin()
	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ObjectFieldMapping")).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ObjectFieldMapping")).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectCommit()

	rows, err := InitializeConnectorFieldMappingsFromPayloadForOwner(db, []byte(validPackPayloadJSON), "test-pack")
	if err != nil {
		t.Fatalf("InitializeConnectorFieldMappingsFromPayloadForOwner: %v", err)
	}
	if len(rows) != 2 {
		t.Fatalf("expected 2 rows, got %d", len(rows))
	}
	for i, row := range rows {
		if row.SortOrder != i {
			t.Errorf("rows[%d] (%s).SortOrder = %d, want %d", i, row.SourceField, row.SortOrder, i)
		}
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// ---------------------------------------------------------------------------
// Part B: Verify completeness tests (DefaultValue, ConflictPolicy,
//         TransformExpression, SortOrder drift detection)
// ---------------------------------------------------------------------------

// verifyDriftPayloadJSON declares one row with all optional fields set.
const verifyDriftPayloadJSON = `[
	{
		"sourceObject": "Lead",
		"targetObject": "Lead",
		"actionType":   "syncLead",
		"direction":    "inbound",
		"mappings": [
			{
				"sourceField":         "email",
				"targetField":         "Email",
				"defaultValue":        "unknown@example.com",
				"transformExpression": "LOWER(email)",
				"conflictPolicy":      "skip_if_set"
			}
		]
	}
]`

func TestVerifyConnectorFieldMappings_ReportsDefaultValueDrift(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	// Live row has a different DefaultValue than payload declares.
	liveRows := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Lead", "Lead", "syncLead",
			"email", "Email", "WRONG_DEFAULT", 0, connector.DirectionInbound, "LOWER(email)", "skip_if_set", nil, nil)

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Lead", "Email"))
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Lead", "Lead", "syncLead", connector.DirectionInbound).
		WillReturnRows(liveRows)
	expectExecutableProjectionState(mock, "test-pack", "Lead", "Lead", "syncLead", "email", connector.DirectionInbound)

	err = VerifyConnectorFieldMappingsFromPayloadForOwner(
		context.Background(), db, []byte(verifyDriftPayloadJSON), "test-pack",
	)
	if err == nil {
		t.Fatal("expected drift error for DefaultValue mismatch, got nil")
	}
	if !strings.Contains(err.Error(), "DefaultValue") {
		t.Errorf("error should mention DefaultValue, got: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorFieldMappings_ReportsConflictPolicyDrift(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	// Live row has wrong ConflictPolicy.
	liveRows := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Lead", "Lead", "syncLead",
			"email", "Email", "unknown@example.com", 0, connector.DirectionInbound, "LOWER(email)", "overwrite", nil, nil)

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Lead", "Email"))
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Lead", "Lead", "syncLead", connector.DirectionInbound).
		WillReturnRows(liveRows)
	expectExecutableProjectionState(mock, "test-pack", "Lead", "Lead", "syncLead", "email", connector.DirectionInbound)

	err = VerifyConnectorFieldMappingsFromPayloadForOwner(
		context.Background(), db, []byte(verifyDriftPayloadJSON), "test-pack",
	)
	if err == nil {
		t.Fatal("expected drift error for ConflictPolicy mismatch, got nil")
	}
	if !strings.Contains(err.Error(), "ConflictPolicy") {
		t.Errorf("error should mention ConflictPolicy, got: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorFieldMappings_ReportsTransformExpressionDrift(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	// Live row has wrong TransformExpression.
	liveRows := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Lead", "Lead", "syncLead",
			"email", "Email", "unknown@example.com", 0, connector.DirectionInbound, "UPPER(email)", "skip_if_set", nil, nil)

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Lead", "Email"))
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Lead", "Lead", "syncLead", connector.DirectionInbound).
		WillReturnRows(liveRows)
	expectExecutableProjectionState(mock, "test-pack", "Lead", "Lead", "syncLead", "email", connector.DirectionInbound)

	err = VerifyConnectorFieldMappingsFromPayloadForOwner(
		context.Background(), db, []byte(verifyDriftPayloadJSON), "test-pack",
	)
	if err == nil {
		t.Fatal("expected drift error for TransformExpression mismatch, got nil")
	}
	if !strings.Contains(err.Error(), "TransformExpression") {
		t.Errorf("error should mention TransformExpression, got: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorFieldMappings_ReportsValueMapDrift(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	liveRows := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Lead", "Lead", "syncLead",
			"status", "Status__c", nil, 0, connector.DirectionInbound, nil, nil, `{"customer":"Closed","lead":"Open"}`, nil)

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Lead", "Status__c"))
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Lead", "Lead", "syncLead", connector.DirectionInbound).
		WillReturnRows(liveRows)
	expectExecutableProjectionState(mock, "test-pack", "Lead", "Lead", "syncLead", "status", connector.DirectionInbound)

	err = VerifyConnectorFieldMappingsFromPayloadForOwner(
		context.Background(), db, []byte(valueMapPackPayloadJSON), "test-pack",
	)
	if err == nil {
		t.Fatal("expected drift error for ValueMap mismatch, got nil")
	}
	if !strings.Contains(err.Error(), "ValueMap") {
		t.Errorf("error should mention ValueMap, got: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorFieldMappings_ReportsSortOrderDrift(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	// Live row has SortOrder=99 but payload declares SortOrder=0 (first row).
	liveRows := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Lead", "Lead", "syncLead",
			"email", "Email", "unknown@example.com", 99, connector.DirectionInbound, "LOWER(email)", "skip_if_set", nil, nil)

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Lead", "Email"))
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Lead", "Lead", "syncLead", connector.DirectionInbound).
		WillReturnRows(liveRows)
	expectExecutableProjectionState(mock, "test-pack", "Lead", "Lead", "syncLead", "email", connector.DirectionInbound)

	err = VerifyConnectorFieldMappingsFromPayloadForOwner(
		context.Background(), db, []byte(verifyDriftPayloadJSON), "test-pack",
	)
	if err == nil {
		t.Fatal("expected drift error for SortOrder mismatch, got nil")
	}
	if !strings.Contains(err.Error(), "SortOrder") {
		t.Errorf("error should mention SortOrder, got: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// ---------------------------------------------------------------------------
// Same SourceField in different tuples — no false drift (FIX 1)
// ---------------------------------------------------------------------------

// sharedSourceFieldTuplePayloadJSON declares two groups that share the same
// SourceField ("email") but map it to DIFFERENT TargetFields. This is the bug
// scenario: before the fix, the second group's expected row overwrote the first
// in the single-key map, causing a false drift report against the first group.
const sharedSourceFieldTuplePayloadJSON = `[
	{
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"mappings": [
			{"sourceField": "email", "targetField": "Email"}
		]
	},
	{
		"sourceObject": "Lead",
		"targetObject": "Lead",
		"actionType":   "syncLead",
		"direction":    "inbound",
		"mappings": [
			{"sourceField": "email", "targetField": "LeadEmail"}
		]
	}
]`

// TestVerifyConnectorFieldMappings_SameSourceFieldDifferentTuples_NoFalseDrift
// confirms that the same SourceField ("email") in two different groups does not
// produce false drift when each group's live rows are correct.
func TestVerifyConnectorFieldMappings_SameSourceFieldDifferentTuples_NoFalseDrift(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	// First group: Contact syncContact — email→Email (SortOrder=0).
	rowsContactGroup := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Contact", "Contact", "syncContact",
			"email", "Email", nil, 0, connector.DirectionInbound, nil, nil, nil, nil)
	// Second group: Lead syncLead — email→LeadEmail (SortOrder=1).
	rowsLeadGroup := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Lead", "Lead", "syncLead",
			"email", "LeadEmail", nil, 1, connector.DirectionInbound, nil, nil, nil, nil)

	expectConnectorFieldMappingTargetSchema(mock,
		connectorFieldMappingTarget("Contact", "Email"),
		connectorFieldMappingTarget("Lead", "LeadEmail"),
	)
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Contact", "Contact", "syncContact", connector.DirectionInbound).
		WillReturnRows(rowsContactGroup)
	expectExecutableProjectionState(mock, "test-pack", "Contact", "Contact", "syncContact", "email", connector.DirectionInbound)
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Lead", "Lead", "syncLead", connector.DirectionInbound).
		WillReturnRows(rowsLeadGroup)
	expectExecutableProjectionState(mock, "test-pack", "Lead", "Lead", "syncLead", "email", connector.DirectionInbound)

	if err := VerifyConnectorFieldMappingsFromPayloadForOwner(
		context.Background(), db, []byte(sharedSourceFieldTuplePayloadJSON), "test-pack",
	); err != nil {
		t.Fatalf("verify should pass when both groups are correct, got: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// TestVerifyConnectorFieldMappings_SameSourceFieldDifferentTuples_DriftReportedCorrectly
// confirms that when one group has a drifted TargetField, only that group's row
// is reported — the other group's correct row is NOT falsely reported.
func TestVerifyConnectorFieldMappings_SameSourceFieldDifferentTuples_DriftReportedCorrectly(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	// First group: Contact syncContact — email→Email (correct).
	rowsContactGroup := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Contact", "Contact", "syncContact",
			"email", "Email", nil, 0, connector.DirectionInbound, nil, nil, nil, nil)
	// Second group: Lead syncLead — email→WrongField (drifted; payload declares LeadEmail).
	rowsLeadGroup := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Lead", "Lead", "syncLead",
			"email", "WrongField", nil, 1, connector.DirectionInbound, nil, nil, nil, nil)

	expectConnectorFieldMappingTargetSchema(mock,
		connectorFieldMappingTarget("Contact", "Email"),
		connectorFieldMappingTarget("Lead", "LeadEmail"),
	)
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Contact", "Contact", "syncContact", connector.DirectionInbound).
		WillReturnRows(rowsContactGroup)
	expectExecutableProjectionState(mock, "test-pack", "Contact", "Contact", "syncContact", "email", connector.DirectionInbound)
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Lead", "Lead", "syncLead", connector.DirectionInbound).
		WillReturnRows(rowsLeadGroup)
	expectExecutableProjectionState(mock, "test-pack", "Lead", "Lead", "syncLead", "email", connector.DirectionInbound)

	err = VerifyConnectorFieldMappingsFromPayloadForOwner(
		context.Background(), db, []byte(sharedSourceFieldTuplePayloadJSON), "test-pack",
	)
	if err == nil {
		t.Fatal("expected drift error for drifted Lead group, got nil")
	}
	// The error must mention the Lead tuple, not the Contact tuple.
	if !strings.Contains(err.Error(), "Lead") {
		t.Errorf("drift error should mention Lead tuple, got: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// TestVerifyConnectorFieldMappings_AllColumnsMatchPassesClean asserts that a
// correctly-applied multi-row payload (all columns including SortOrder match)
// verifies clean, confirming Initialize and Parse produce identical SortOrder values.
func TestVerifyConnectorFieldMappings_AllColumnsMatchPassesClean(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	// verifyDriftPayloadJSON: one row, SortOrder=0 (first and only row in payload).
	liveRows := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Lead", "Lead", "syncLead",
			"email", "Email", "unknown@example.com", 0, connector.DirectionInbound, "LOWER(email)", "skip_if_set", nil, nil)

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Lead", "Email"))
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Lead", "Lead", "syncLead", connector.DirectionInbound).
		WillReturnRows(liveRows)
	expectExecutableProjectionState(mock, "test-pack", "Lead", "Lead", "syncLead", "email", connector.DirectionInbound)

	if err := VerifyConnectorFieldMappingsFromPayloadForOwner(
		context.Background(), db, []byte(verifyDriftPayloadJSON), "test-pack",
	); err != nil {
		t.Fatalf("verify should pass when all columns match, got: %v", err)
	}

	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestVerifyConnectorFieldMappings_CanonicalizesLiveSourceFieldLookup(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	payload := `[{
		"sourceObject":"Lead","targetObject":"Lead",
		"actionType":"syncLead","direction":"inbound",
		"mappings":[{"sourceField":"email","targetField":"Email"}]
	}]`
	liveRows := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Lead", "Lead", "syncLead",
			"Email", "Email", nil, 0, connector.DirectionInbound, nil, nil, nil, nil)

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Lead", "Email"))
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Lead", "Lead", "syncLead", connector.DirectionInbound).
		WillReturnRows(liveRows)
	expectExecutableProjectionState(mock, "test-pack", "Lead", "Lead", "syncLead", "email", connector.DirectionInbound)

	if err := VerifyConnectorFieldMappingsFromPayloadForOwner(
		context.Background(), db, []byte(payload), "test-pack",
	); err != nil {
		t.Fatalf("verify should case-fold live SourceField lookup, got: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

// ---------------------------------------------------------------------------
// Duplicate business key rejection tests
// ---------------------------------------------------------------------------

// duplicateKeyPayloadJSON has two rows in the same group with the same SourceField
// (email), which produces an identical business key.
const duplicateKeyPayloadJSON = `[
	{
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"mappings": [
			{"sourceField": "email", "targetField": "Email"},
			{"sourceField": "email", "targetField": "WorkEmail"}
		]
	}
]`

// TestParseConnectorFieldMappingInputs_DuplicateBusinessKeyRejected asserts that
// two rows sharing the same business key (same Source, SourceObject, TargetObject,
// ActionType, SourceField, Direction) are rejected with a "duplicate business key" em error.
func TestParseConnectorFieldMappingInputs_DuplicateBusinessKeyRejected(t *testing.T) {
	_, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(duplicateKeyPayloadJSON), "test-pack")
	if err == nil {
		t.Fatal("expected error for duplicate business key, got nil")
	}
	if !strings.Contains(err.Error(), "duplicate business key") {
		t.Errorf("error should mention 'duplicate business key', got: %v", err)
	}
}

const duplicateTargetFieldPayloadJSON = `[
	{
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"mappings": [
			{"sourceField": "email", "targetField": "Email"},
			{"sourceField": "alternateEmail", "targetField": "Email"}
		]
	}
]`

func TestParseConnectorFieldMappingInputs_DuplicateTargetFieldRejected(t *testing.T) {
	_, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(duplicateTargetFieldPayloadJSON), "test-pack")
	if err == nil {
		t.Fatal("expected error for duplicate target field, got nil")
	}
	if !strings.Contains(err.Error(), "duplicate effective target field") {
		t.Errorf("error should mention duplicate effective target field, got: %v", err)
	}
}

const duplicateTargetFieldExactBidirectionalPayloadJSON = `[
	{
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"mappings": [
			{"sourceField": "email", "targetField": "Email"}
		]
	},
	{
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "bidirectional",
		"mappings": [
			{"sourceField": "alternateEmail", "targetField": "Email"}
		]
	}
]`

func TestParseConnectorFieldMappingInputs_DuplicateTargetFieldAcrossBidirectionalFanOutRejected(t *testing.T) {
	_, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(duplicateTargetFieldExactBidirectionalPayloadJSON), "test-pack")
	if err == nil {
		t.Fatal("expected error for duplicate effective target field across bidirectional fan-out, got nil")
	}
	if !strings.Contains(err.Error(), "effective direction inbound") {
		t.Errorf("error should mention inbound effective direction, got: %v", err)
	}
}

// TestParseConnectorFieldMappingInputs_SameSourceFieldDifferentDirection_Allowed asserts that
// two rows with the same SourceField but different Direction values are NOT duplicates
// (Direction is part of the business key).
const sameFieldDifferentDirectionPayloadJSON = `[
	{
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"mappings": [
			{"sourceField": "email", "targetField": "Email"}
		]
	},
	{
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "outbound",
		"mappings": [
			{"sourceField": "email", "targetField": "Email"}
		]
	}
]`

func TestVerifyConnectorFieldMappings_ExactPlusBidirectionalNoFalseDrift(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	payload := `[{
		"sourceObject":"Contact","targetObject":"Contact",
		"actionType":"syncContact","direction":"inbound",
		"mappings":[{"sourceField":"email","targetField":"Email"}]
	}]`

	// bootstrap.ResolveFieldMapping returns both the exact-direction row (inbound)
	// AND a bidirectional fallback for the same SourceField. The exact
	// row must take precedence — verify should NOT report false drift
	// on Direction caused by the bidirectional row overwriting it.
	live := sqlmock.NewRows(resolverColsV).
		AddRow("test-pack", "Contact", "Contact", "syncContact",
			"email", "Email", nil, 0, connector.DirectionInbound, nil, nil, nil, nil).
		AddRow("test-pack", "Contact", "Contact", "syncContact",
			"email", "Email", nil, 0, connector.DirectionBidirectional, nil, nil, nil, nil)

	expectConnectorFieldMappingTargetSchema(mock, connectorFieldMappingTarget("Contact", "Email"))
	mock.ExpectQuery("FROM _System_ObjectFieldMapping").
		WithArgs("test-pack", "Contact", "Contact", "syncContact", connector.DirectionInbound).
		WillReturnRows(live)
	expectExecutableProjectionState(mock, "test-pack", "Contact", "Contact", "syncContact", "email", connector.DirectionInbound)

	if err := VerifyConnectorFieldMappingsFromPayloadForOwner(
		context.Background(), db, []byte(payload), "test-pack",
	); err != nil {
		t.Fatalf("verify should pass (exact direction takes precedence over bidirectional fallback), got: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations unmet: %v", err)
	}
}

func TestParseConnectorFieldMappingInputs_SameSourceFieldDifferentDirection_Allowed(t *testing.T) {
	rows, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload([]byte(sameFieldDifferentDirectionPayloadJSON), "test-pack")
	if err != nil {
		t.Fatalf("different Direction with same SourceField should be allowed, got: %v", err)
	}
	if len(rows) != 2 {
		t.Errorf("expected 2 rows, got %d", len(rows))
	}
}