package packs

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"strings"
	"time"

	"github.com/nexuscrm/shared/pkg/constants"

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

// This file hosts the connector field-mapping WRITE/INIT/VERIFY surface,
// relocated verbatim (behavior-identical) from internal/bootstrap as part of
// the Wave-5 control-plane decoupling. The READ surface
// (ResolveFieldMapping/ResolvePackConnectorFieldMapping/GetFieldMappings) is
// SHARED with the automation feature via the contract
// runtimeports.ConnectorFieldMappingReader port and STAYS in internal/bootstrap;
// the relocated verify path reaches it through the db-passing seam in
// connector_field_mapping_reader_seam.go.

// --- Connector field-mapping constants/helpers needed by the relocated write
// side. These mirror the canonical bootstrap definitions byte-for-byte; the
// bootstrap copies remain in place because the SHARED reader surface (resolver,
// validator) still depends on them. Duplicating the small, pure literals here
// keeps the relocated write side free of an internal/bootstrap call-site import.

const (
	tableObjectFieldMapping       = "_System_ObjectFieldMapping"
	legacyObjectFieldMappingOwner = "legacy"
)

// Direction constants for connector field mapping rows (alias to the contract
// enum the bootstrap validator also aliases).
const (
	directionInbound       = connector.DirectionInbound
	directionOutbound      = connector.DirectionOutbound
	directionBidirectional = connector.DirectionBidirectional
)

var validConnectorDirections = map[string]struct{}{
	directionInbound:       {},
	directionOutbound:      {},
	directionBidirectional: {},
}

// connectorFieldMappingInput is the input struct for pack-managed connector
// rows (alias to the contract DTO; the durable JSON projection is byte-identical).
type connectorFieldMappingInput = connector.ConnectorFieldMappingInput

func connectorFieldMappingKeySegment(value string) string {
	return strings.ToLower(strings.TrimSpace(value))
}

// connectorFieldMappingBusinessKey returns a stable, canonical string derived
// from the six fields that form the unique business key for connector rows.
func connectorFieldMappingBusinessKey(source, sourceObject, targetObject, actionType, sourceField, direction string) string {
	return strings.Join([]string{
		connectorFieldMappingKeySegment(source),
		connectorFieldMappingKeySegment(sourceObject),
		connectorFieldMappingKeySegment(targetObject),
		connectorFieldMappingKeySegment(actionType),
		connectorFieldMappingKeySegment(sourceField),
		connectorFieldMappingKeySegment(direction),
	}, "\x00")
}

// connectorFieldMappingID derives a deterministic, stable Id from the canonical
// business key. The id is prefixed "ofmc_" (connector variant).
func connectorFieldMappingID(source, sourceObject, targetObject, actionType, sourceField, direction string) string {
	key := connectorFieldMappingBusinessKey(source, sourceObject, targetObject, actionType, sourceField, direction)
	sum := sha256.Sum256([]byte(key))
	return fmt.Sprintf("ofmc_%x", sum)
}

// sqlExecer is the narrow interface satisfied by both *sql.DB and *sql.Tx so
// upsertConnectorFieldMapping can participate in a caller-managed transaction.
type sqlExecer interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
}

// connectorFieldMappingIDForInput computes the deterministic Id for a given
// connectorFieldMappingInput after canonicalisation. Relocated alongside the
// connector field-mapping write surface for package-internal test use.
func connectorFieldMappingIDForInput(row connectorFieldMappingInput) string {
	return connectorFieldMappingID(
		row.Source,
		row.SourceObject,
		row.TargetObject,
		row.ActionType,
		row.SourceField,
		row.Direction,
	)
}

// upsertConnectorFieldMapping inserts or updates a single connector field
// mapping row in _System_ObjectFieldMapping (relocated verbatim from
// bootstrap.UpsertConnectorFieldMapping).
func upsertConnectorFieldMapping(ctx context.Context, db sqlExecer, row connectorFieldMappingInput) error {
	if err := kernelconnector.ValidateConnectorFieldMappingRow(row); err != nil {
		return err
	}

	source := strings.TrimSpace(row.Source)
	sourceObject := strings.TrimSpace(row.SourceObject)
	targetObject := strings.TrimSpace(row.TargetObject)
	actionType := strings.TrimSpace(row.ActionType)
	sourceField := strings.TrimSpace(row.SourceField)
	targetField := strings.TrimSpace(row.TargetField)
	direction := strings.TrimSpace(row.Direction)
	conflictPolicy := strings.TrimSpace(row.ConflictPolicy)
	transformExpression := strings.TrimSpace(row.TransformExpression)
	defaultValue := strings.TrimSpace(row.DefaultValue)
	valueMap := strings.TrimSpace(row.ValueMap)
	onUnmappedValue := strings.TrimSpace(row.OnUnmappedValue)

	id := connectorFieldMappingID(source, sourceObject, targetObject, actionType, sourceField, direction)

	now := time.Now().Format("2006-01-02 15:04:05")

	var conflictPolicyVal interface{} = nil
	if conflictPolicy != "" {
		conflictPolicyVal = conflictPolicy
	}
	var transformExpressionVal interface{} = nil
	if transformExpression != "" {
		transformExpressionVal = transformExpression
	}
	var defaultValueVal interface{} = nil
	if defaultValue != "" {
		defaultValueVal = defaultValue
	}
	var valueMapVal interface{} = nil
	if valueMap != "" {
		valueMapVal = valueMap
	}
	var onUnmappedValueVal interface{} = nil
	if onUnmappedValue != "" {
		onUnmappedValueVal = onUnmappedValue
	}

	const insertSQL = `INSERT INTO _System_ObjectFieldMapping
		(Id, SourceObject, TargetObject, ActionType, SourceField, TargetField,
		 SourceLookupField, DefaultValue, ValueMap, SortOrder, IsCustomized, Source, IsActive,
		 IsDeleted, Direction, TransformExpression, ConflictPolicy, OnUnmappedValue, ProjectionOnly,
		 CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, ?,
		        NULL, ?, ?, ?, 0, ?, 1,
		        0, ?, ?, ?, ?, ?,
		        ?, ?)
		ON DUPLICATE KEY UPDATE
		  Id                   = VALUES(Id),
		  SourceObject         = VALUES(SourceObject),
		  TargetObject         = VALUES(TargetObject),
		  ActionType           = VALUES(ActionType),
		  SourceField          = VALUES(SourceField),
		  TargetField          = VALUES(TargetField),
		  SourceLookupField    = CASE WHEN VALUES(ProjectionOnly) = 1 THEN NULL ELSE SourceLookupField END,
		  DefaultValue         = VALUES(DefaultValue),
		  ValueMap             = VALUES(ValueMap),
		  SortOrder            = VALUES(SortOrder),
		  IsCustomized         = CASE WHEN VALUES(ProjectionOnly) = 1 THEN 0 ELSE IsCustomized END,
		  Source               = VALUES(Source),
		  Direction            = VALUES(Direction),
		  TransformExpression  = VALUES(TransformExpression),
		  ConflictPolicy       = VALUES(ConflictPolicy),
		  OnUnmappedValue      = VALUES(OnUnmappedValue),
		  ProjectionOnly       = VALUES(ProjectionOnly),
		  IsActive             = 1,
		  IsDeleted            = 0,
		  LastModifiedDate     = VALUES(LastModifiedDate)`

	_, err := db.ExecContext(ctx, insertSQL,
		id,
		sourceObject,
		targetObject,
		actionType,
		sourceField,
		targetField,
		defaultValueVal,
		valueMapVal,
		row.SortOrder,
		source,
		direction,
		transformExpressionVal,
		conflictPolicyVal,
		onUnmappedValueVal,
		row.ProjectionOnly,
		now,
		now,
	)
	if err != nil {
		return fmt.Errorf("upsert connector field mapping %s/%s→%s(%s).%s[%s]: %w",
			source, sourceObject, targetObject, actionType, sourceField, direction, err)
	}
	return nil
}

// --- Local-schema target validation (relocated verbatim; consumed only by the
// relocated init/verify path).

type connectorFieldMappingTargetField struct {
	updateable bool
}

func validateConnectorFieldMappingTargets(ctx context.Context, rawDB *sql.DB, rows []connectorFieldMappingInput) error {
	if len(rows) == 0 {
		return nil
	}
	requiresLocalSchema := false
	for _, row := range rows {
		if row.ProjectionOnly {
			continue
		}
		direction := strings.TrimSpace(row.Direction)
		if strings.EqualFold(direction, directionOutbound) || strings.EqualFold(direction, directionInbound) || strings.EqualFold(direction, directionBidirectional) {
			requiresLocalSchema = true
			break
		}
	}
	if !requiresLocalSchema {
		return nil
	}
	if rawDB == nil {
		return fmt.Errorf("connector_field_mapping local schema validation: db is required")
	}
	registry, err := loadConnectorFieldMappingTargetRegistry(ctx, rawDB)
	if err != nil {
		return fmt.Errorf("connector_field_mapping local schema validation: load metadata: %w", err)
	}

	var errs []string
	for i, row := range rows {
		if row.ProjectionOnly {
			continue
		}
		direction := strings.TrimSpace(row.Direction)
		if strings.EqualFold(direction, directionOutbound) || strings.EqualFold(direction, directionBidirectional) {
			sourceObject := strings.TrimSpace(row.SourceObject)
			sourceField := strings.TrimSpace(row.SourceField)
			fields, ok := registry[strings.ToLower(sourceObject)]
			if !ok {
				errs = append(errs, fmt.Sprintf("row %d: SourceObject %q not found in live metadata", i, sourceObject))
			} else if _, ok := fields[strings.ToLower(sourceField)]; !ok {
				errs = append(errs, fmt.Sprintf("row %d: SourceField %q does not exist on SourceObject %q", i, sourceField, sourceObject))
			}
		}
		if strings.EqualFold(direction, directionInbound) || strings.EqualFold(direction, directionBidirectional) {
			targetObject := strings.TrimSpace(row.TargetObject)
			targetField := strings.TrimSpace(row.TargetField)
			fields, ok := registry[strings.ToLower(targetObject)]
			if !ok {
				errs = append(errs, fmt.Sprintf("row %d: TargetObject %q not found in live metadata", i, targetObject))
				continue
			}
			field, ok := fields[strings.ToLower(targetField)]
			if !ok {
				errs = append(errs, fmt.Sprintf("row %d: TargetField %q does not exist on TargetObject %q", i, targetField, targetObject))
				continue
			}
			if !field.updateable {
				errs = append(errs, fmt.Sprintf("row %d: TargetField %q on TargetObject %q is not updateable", i, targetField, targetObject))
			}
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("connector_field_mapping local schema validation failed:\n  - %s", strings.Join(errs, "\n  - "))
	}
	return nil
}

func loadConnectorFieldMappingTargetRegistry(ctx context.Context, rawDB *sql.DB) (map[string]map[string]connectorFieldMappingTargetField, error) {
	rows, err := rawDB.QueryContext(ctx, fmt.Sprintf(
		"SELECT o.`%s`, f.`%s`, f.`%s` FROM `%s` f JOIN `%s` o ON o.`%s` = f.`%s` WHERE COALESCE(f.`%s`, 0) = 0 AND COALESCE(o.`%s`, 0) = 0",
		constants.FieldSysObject_APIName,
		constants.FieldSysField_APIName,
		constants.FieldSysField_Updateable,
		constants.TableField,
		constants.TableObject,
		constants.FieldSysObject_ID,
		constants.FieldSysField_ObjectID,
		constants.FieldSysField_IsDeleted,
		constants.FieldSysObject_IsDeleted,
	))
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	registry := map[string]map[string]connectorFieldMappingTargetField{}
	for rows.Next() {
		var objectAPIName, fieldAPIName string
		var updateable bool
		if err := rows.Scan(&objectAPIName, &fieldAPIName, &updateable); err != nil {
			return nil, err
		}
		objectKey := strings.ToLower(strings.TrimSpace(objectAPIName))
		fieldKey := strings.ToLower(strings.TrimSpace(fieldAPIName))
		if objectKey == "" || fieldKey == "" {
			continue
		}
		if registry[objectKey] == nil {
			registry[objectKey] = map[string]connectorFieldMappingTargetField{}
		}
		registry[objectKey][fieldKey] = connectorFieldMappingTargetField{updateable: updateable}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return registry, nil
}

// --- Group payload parsing (relocated verbatim; consumed only by the verify path).

// ConnectorFieldMappingGroup represents a group of connector field mappings
// between two objects for a given action type and direction.
type ConnectorFieldMappingGroup struct {
	SourceObject string                        `json:"sourceObject"`
	TargetObject string                        `json:"targetObject"`
	ActionType   string                        `json:"actionType"`
	Direction    string                        `json:"direction"`
	Mappings     []ConnectorFieldMappingRowDef `json:"mappings"`
}

// ConnectorFieldMappingRowDef is a single source→target row within a group.
type ConnectorFieldMappingRowDef struct {
	SourceField         string          `json:"sourceField"`
	TargetField         string          `json:"targetField"`
	TransformExpression string          `json:"transformExpression,omitempty"`
	ConflictPolicy      string          `json:"conflictPolicy,omitempty"`
	DefaultValue        string          `json:"defaultValue,omitempty"`
	ValueMap            json.RawMessage `json:"valueMap,omitempty"`
	OnUnmappedValue     string          `json:"onUnmappedValue,omitempty"`
	ProjectionOnly      bool            `json:"projectionOnly,omitempty"`
}

func parseConnectorFieldMappingGroupsPayload(raw []byte) ([]ConnectorFieldMappingGroup, error) {
	dec := json.NewDecoder(bytes.NewReader(raw))
	dec.DisallowUnknownFields()

	var groups []ConnectorFieldMappingGroup
	if err := dec.Decode(&groups); err != nil {
		return nil, fmt.Errorf("connector_field_mapping: failed to parse payload: %w", err)
	}

	if tok, err := dec.Token(); err == nil {
		return nil, fmt.Errorf("connector_field_mapping: unexpected trailing token %v in payload", tok)
	} else if err != io.EOF {
		return nil, fmt.Errorf("connector_field_mapping: invalid trailing JSON in payload: %w", err)
	}

	return groups, nil
}

// InitializeConnectorFieldMappingsFromPayloadForOwner parses the JSON payload
// and writes all declared rows atomically (upsert-only). Relocated verbatim
// from internal/bootstrap.
func InitializeConnectorFieldMappingsFromPayloadForOwner(rawDB *sql.DB, raw []byte, owner string) ([]connectorFieldMappingInput, error) {
	rows, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload(raw, owner)
	if err != nil {
		return nil, err
	}
	if err := validateConnectorFieldMappingTargets(context.Background(), rawDB, rows); err != nil {
		return nil, err
	}

	if len(rows) == 0 {
		return rows, nil
	}

	tx, err := rawDB.BeginTx(context.Background(), nil)
	if err != nil {
		return nil, fmt.Errorf("connector_field_mapping: begin tx: %w", err)
	}

	for _, row := range rows {
		if err := upsertConnectorFieldMapping(context.Background(), tx, row); err != nil {
			_ = tx.Rollback()
			return nil, fmt.Errorf("connector_field_mapping: upsert in tx: %w", err)
		}
	}

	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf("connector_field_mapping: commit tx: %w", err)
	}
	return rows, nil
}

// SweepConnectorFieldMappingsForOwner performs the pack-level owner-wide
// reconcile. Relocated verbatim from internal/bootstrap.
func SweepConnectorFieldMappingsForOwner(ctx context.Context, rawDB *sql.DB, owner string, declared []connectorFieldMappingInput) error {
	owner = strings.TrimSpace(owner)
	if owner == "" {
		return fmt.Errorf("connector_field_mapping sweep: owner must not be blank")
	}

	payloadIDs := connectorFieldMappingDeclaredIDSet(declared)
	ids := make([]string, 0, len(payloadIDs))
	for id := range payloadIDs {
		ids = append(ids, id)
	}

	now := time.Now().Format("2006-01-02 15:04:05")
	return deactivateStaleConnectorFieldMappingRowsForOwner(ctx, rawDB, owner, declared, ids, now)
}

// VerifyNoStaleConnectorFieldMappingsForOwner checks that no active connector
// rows exist for the owner that are not in the declared union. Relocated
// verbatim from internal/bootstrap.
func VerifyNoStaleConnectorFieldMappingsForOwner(ctx context.Context, rawDB *sql.DB, owner string, declared []connectorFieldMappingInput) error {
	owner = strings.TrimSpace(owner)
	if owner == "" {
		return fmt.Errorf("connector_field_mapping verify sweep: owner must not be blank")
	}

	payloadIDSet := connectorFieldMappingDeclaredIDSet(declared)
	staleErrs, err := findStaleConnectorRowsForOwner(ctx, rawDB, owner, nil, payloadIDSet)
	if err != nil {
		return fmt.Errorf("connector_field_mapping verify sweep: stale-row query: %w", err)
	}
	if len(staleErrs) > 0 {
		return fmt.Errorf("connector field mapping stale rows for owner %s:\n  - %s", owner, strings.Join(staleErrs, "\n  - "))
	}
	return nil
}

func connectorFieldMappingDeclaredIDSet(declared []connectorFieldMappingInput) map[string]struct{} {
	set := make(map[string]struct{}, len(declared))
	for _, row := range declared {
		id := connectorFieldMappingID(
			row.Source,
			row.SourceObject,
			row.TargetObject,
			row.ActionType,
			row.SourceField,
			row.Direction,
		)
		set[id] = struct{}{}
	}
	return set
}

func deactivateStaleConnectorFieldMappingRowsForOwner(ctx context.Context, db sqlExecer, owner string, _ []connectorFieldMappingInput, payloadIDs []string, now string) error {
	if len(payloadIDs) > 0 {
		placeholders := strings.Repeat("?,", len(payloadIDs))
		placeholders = placeholders[:len(placeholders)-1]
		query := fmt.Sprintf(
			`UPDATE %s SET IsDeleted = 1, LastModifiedDate = ?
			 WHERE Source = ? AND Direction <> '%s' AND IsDeleted = 0
			   AND Id NOT IN (%s)`,
			tableObjectFieldMapping,
			legacyObjectFieldMappingOwner,
			placeholders,
		)
		args := make([]interface{}, 0, 2+len(payloadIDs))
		args = append(args, now, owner)
		for _, id := range payloadIDs {
			args = append(args, id)
		}
		_, err := db.ExecContext(ctx, query, args...)
		return err
	}
	query := fmt.Sprintf(
		`UPDATE %s SET IsDeleted = 1, LastModifiedDate = ?
		 WHERE Source = ? AND Direction <> '%s' AND IsDeleted = 0`,
		tableObjectFieldMapping,
		legacyObjectFieldMappingOwner,
	)
	_, err := db.ExecContext(ctx, query, now, owner)
	return err
}

// VerifyConnectorFieldMappingsFromPayloadForOwner confirms that every row
// declared in the payload is live in the database with the correct direction.
// Relocated verbatim from internal/bootstrap; the live read routes through the
// db-passing reader seam (resolveFieldMapping) instead of importing bootstrap.
func VerifyConnectorFieldMappingsFromPayloadForOwner(ctx context.Context, rawDB *sql.DB, raw []byte, owner string) error {
	expectedRows, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload(raw, owner)
	if err != nil {
		return fmt.Errorf("connector_field_mapping verify: %w", err)
	}
	if err := validateConnectorFieldMappingTargets(ctx, rawDB, expectedRows); err != nil {
		return fmt.Errorf("connector_field_mapping verify: %w", err)
	}

	groups, err := parseConnectorFieldMappingGroupsPayload(raw)
	if err != nil {
		return err
	}

	connectorFieldMappingTupleKey := func(sourceObject, targetObject, actionType, direction, sourceField string) string {
		return strings.Join([]string{
			connectorFieldMappingKeySegment(sourceObject),
			connectorFieldMappingKeySegment(targetObject),
			connectorFieldMappingKeySegment(actionType),
			connectorFieldMappingKeySegment(direction),
			connectorFieldMappingKeySegment(sourceField),
		}, "\x00")
	}
	expectedByTuple := make(map[string]connectorFieldMappingInput, len(expectedRows))
	for _, r := range expectedRows {
		k := connectorFieldMappingTupleKey(r.SourceObject, r.TargetObject, r.ActionType, r.Direction, r.SourceField)
		expectedByTuple[k] = r
	}

	var errs []string

	for _, group := range groups {
		if len(group.Mappings) == 0 {
			continue
		}

		dir := strings.TrimSpace(group.Direction)
		if _, ok := validConnectorDirections[dir]; !ok {
			errs = append(errs, fmt.Sprintf(
				"group %s→%s(%s): direction %q is invalid (must be inbound/outbound/bidirectional)",
				group.SourceObject, group.TargetObject, group.ActionType, group.Direction,
			))
			continue
		}

		hasExecutableMappings := false
		for _, mapping := range group.Mappings {
			if !mapping.ProjectionOnly {
				hasExecutableMappings = true
				break
			}
		}

		liveBySource := map[string]connectorFieldMappingInput{}
		if hasExecutableMappings {
			live, err := resolveFieldMapping(ctx, rawDB, ResolveQuery{
				Source:       owner,
				SourceObject: group.SourceObject,
				TargetObject: group.TargetObject,
				ActionType:   group.ActionType,
				Direction:    dir,
			})
			if err != nil {
				return fmt.Errorf("connector_field_mapping verify: resolve %s→%s(%s)[%s]: %w",
					group.SourceObject, group.TargetObject, group.ActionType, dir, err)
			}
			for _, row := range live {
				sf := connectorFieldMappingKeySegment(row.SourceField)
				if _, exists := liveBySource[sf]; !exists {
					liveBySource[sf] = row
				}
			}
		}

		for _, m := range group.Mappings {
			sf := strings.TrimSpace(m.SourceField)
			want, hasExpected := expectedByTuple[connectorFieldMappingTupleKey(group.SourceObject, group.TargetObject, group.ActionType, dir, sf)]
			if m.ProjectionOnly {
				if !hasExpected {
					continue
				}
				if err := verifyProjectionOnlyConnectorFieldMapping(ctx, rawDB, owner, want); err != nil {
					errs = append(errs, err.Error())
				}
				continue
			}
			liveRow, found := liveBySource[connectorFieldMappingKeySegment(sf)]
			if !found {
				errs = append(errs, fmt.Sprintf(
					"connector field mapping %s→%s(%s)[%s].%s is missing",
					group.SourceObject, group.TargetObject, group.ActionType, dir, sf,
				))
				continue
			}
			if !hasExpected {
				continue
			}
			if err := verifyExecutableConnectorFieldMappingProjectionState(ctx, rawDB, owner, want); err != nil {
				errs = append(errs, err.Error())
				continue
			}

			wantTarget := strings.TrimSpace(want.TargetField)
			if strings.TrimSpace(liveRow.TargetField) != wantTarget {
				errs = append(errs, fmt.Sprintf(
					"connector field mapping %s→%s(%s)[%s].%s TargetField = %q, want %q",
					group.SourceObject, group.TargetObject, group.ActionType, dir, sf,
					liveRow.TargetField, wantTarget,
				))
			}
			if strings.TrimSpace(liveRow.Direction) != dir {
				errs = append(errs, fmt.Sprintf(
					"connector field mapping %s→%s(%s)[%s].%s Direction = %q, want %q",
					group.SourceObject, group.TargetObject, group.ActionType, dir, sf,
					liveRow.Direction, dir,
				))
			}
			wantDefault := strings.TrimSpace(want.DefaultValue)
			if strings.TrimSpace(liveRow.DefaultValue) != wantDefault {
				errs = append(errs, fmt.Sprintf(
					"connector field mapping %s→%s(%s)[%s].%s DefaultValue = %q, want %q",
					group.SourceObject, group.TargetObject, group.ActionType, dir, sf,
					liveRow.DefaultValue, wantDefault,
				))
			}
			wantTransform := strings.TrimSpace(want.TransformExpression)
			if strings.TrimSpace(liveRow.TransformExpression) != wantTransform {
				errs = append(errs, fmt.Sprintf(
					"connector field mapping %s→%s(%s)[%s].%s TransformExpression = %q, want %q",
					group.SourceObject, group.TargetObject, group.ActionType, dir, sf,
					liveRow.TransformExpression, wantTransform,
				))
			}
			wantConflict := strings.TrimSpace(want.ConflictPolicy)
			if strings.TrimSpace(liveRow.ConflictPolicy) != wantConflict {
				errs = append(errs, fmt.Sprintf(
					"connector field mapping %s→%s(%s)[%s].%s ConflictPolicy = %q, want %q",
					group.SourceObject, group.TargetObject, group.ActionType, dir, sf,
					liveRow.ConflictPolicy, wantConflict,
				))
			}
			wantValueMap := strings.TrimSpace(want.ValueMap)
			if strings.TrimSpace(liveRow.ValueMap) != wantValueMap {
				errs = append(errs, fmt.Sprintf(
					"connector field mapping %s→%s(%s)[%s].%s ValueMap = %q, want %q",
					group.SourceObject, group.TargetObject, group.ActionType, dir, sf,
					liveRow.ValueMap, wantValueMap,
				))
			}
			wantOnUnmapped := strings.TrimSpace(want.OnUnmappedValue)
			if strings.TrimSpace(liveRow.OnUnmappedValue) != wantOnUnmapped {
				errs = append(errs, fmt.Sprintf(
					"connector field mapping %s→%s(%s)[%s].%s OnUnmappedValue = %q, want %q",
					group.SourceObject, group.TargetObject, group.ActionType, dir, sf,
					liveRow.OnUnmappedValue, wantOnUnmapped,
				))
			}
			if liveRow.SortOrder != want.SortOrder {
				errs = append(errs, fmt.Sprintf(
					"connector field mapping %s→%s(%s)[%s].%s SortOrder = %d, want %d",
					group.SourceObject, group.TargetObject, group.ActionType, dir, sf,
					liveRow.SortOrder, want.SortOrder,
				))
			}
		}
	}

	if len(errs) > 0 {
		return fmt.Errorf("connector field mapping drift:\n  - %s", strings.Join(errs, "\n  - "))
	}
	return nil
}

func verifyExecutableConnectorFieldMappingProjectionState(ctx context.Context, rawDB *sql.DB, owner string, want connectorFieldMappingInput) error {
	var projectionOnly bool
	err := rawDB.QueryRowContext(ctx, fmt.Sprintf(`
SELECT ProjectionOnly
  FROM %s
 WHERE Source = ?
   AND SourceObject = ?
   AND TargetObject = ?
   AND ActionType = ?
   AND SourceField = ?
   AND Direction = ?`, tableObjectFieldMapping),
		strings.TrimSpace(owner),
		strings.TrimSpace(want.SourceObject),
		strings.TrimSpace(want.TargetObject),
		strings.TrimSpace(want.ActionType),
		strings.TrimSpace(want.SourceField),
		strings.TrimSpace(want.Direction),
	).Scan(&projectionOnly)
	rowLabel := fmt.Sprintf("connector field mapping %s→%s(%s)[%s].%s",
		want.SourceObject, want.TargetObject, want.ActionType, want.Direction, want.SourceField)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf("%s is missing", rowLabel)
	}
	if err != nil {
		return fmt.Errorf("verify %s executable physical row: %w", rowLabel, err)
	}
	if projectionOnly {
		return fmt.Errorf("%s executable physical row ProjectionOnly = true, want false", rowLabel)
	}
	return nil
}

func verifyProjectionOnlyConnectorFieldMapping(ctx context.Context, rawDB *sql.DB, owner string, want connectorFieldMappingInput) error {
	var targetField, direction string
	var sourceLookupField, transformExpression, conflictPolicy, defaultValue, valueMap, onUnmappedValue sql.NullString
	var sortOrder int
	var projectionOnly, isCustomized, isActive, isDeleted bool
	err := rawDB.QueryRowContext(ctx, fmt.Sprintf(`
SELECT TargetField, SourceLookupField, Direction, TransformExpression, ConflictPolicy, DefaultValue, ValueMap, OnUnmappedValue,
       SortOrder, ProjectionOnly, IsCustomized, IsActive, IsDeleted
  FROM %s
 WHERE Source = ?
   AND SourceObject = ?
   AND TargetObject = ?
   AND ActionType = ?
   AND SourceField = ?
   AND Direction = ?`, tableObjectFieldMapping),
		strings.TrimSpace(owner),
		strings.TrimSpace(want.SourceObject),
		strings.TrimSpace(want.TargetObject),
		strings.TrimSpace(want.ActionType),
		strings.TrimSpace(want.SourceField),
		strings.TrimSpace(want.Direction),
	).Scan(
		&targetField,
		&sourceLookupField,
		&direction,
		&transformExpression,
		&conflictPolicy,
		&defaultValue,
		&valueMap,
		&onUnmappedValue,
		&sortOrder,
		&projectionOnly,
		&isCustomized,
		&isActive,
		&isDeleted,
	)
	rowLabel := fmt.Sprintf("connector field mapping %s→%s(%s)[%s].%s",
		want.SourceObject, want.TargetObject, want.ActionType, want.Direction, want.SourceField)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf("%s is missing", rowLabel)
	}
	if err != nil {
		return fmt.Errorf("verify %s physical projection row: %w", rowLabel, err)
	}
	if strings.TrimSpace(targetField) != "" ||
		strings.TrimSpace(sourceLookupField.String) != "" ||
		strings.TrimSpace(direction) != strings.TrimSpace(want.Direction) ||
		strings.TrimSpace(transformExpression.String) != "" ||
		strings.TrimSpace(conflictPolicy.String) != "" ||
		strings.TrimSpace(defaultValue.String) != "" ||
		strings.TrimSpace(valueMap.String) != "" ||
		strings.TrimSpace(onUnmappedValue.String) != "" ||
		sortOrder != want.SortOrder ||
		!projectionOnly || isCustomized || !isActive || isDeleted {
		return fmt.Errorf("%s projection-only physical row drifted", rowLabel)
	}
	return nil
}

func findStaleConnectorRowsForOwner(ctx context.Context, db *sql.DB, owner string, _ map[string]struct{}, payloadIDSet map[string]struct{}) ([]string, error) {
	query := fmt.Sprintf(
		`SELECT Id, SourceObject, TargetObject, ActionType, SourceField, Direction
		 FROM %s
		 WHERE Source = ? AND Direction <> '%s'
		   AND IsActive = 1 AND IsDeleted = 0`,
		tableObjectFieldMapping,
		legacyObjectFieldMappingOwner,
	)
	rows, err := db.QueryContext(ctx, query, owner)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var staleErrs []string
	for rows.Next() {
		var id, sourceObject, targetObject, actionType, sourceField, direction string
		if err := rows.Scan(&id, &sourceObject, &targetObject, &actionType, &sourceField, &direction); err != nil {
			return nil, err
		}
		if _, inPayload := payloadIDSet[id]; !inPayload {
			staleErrs = append(staleErrs, fmt.Sprintf(
				"stale connector row for owner %s: %s→%s(%s)[%s].%s (id=%s) is active but not in payload",
				owner, sourceObject, targetObject, actionType, direction, sourceField, id,
			))
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return staleErrs, nil
}