package automation

import (
	"context"
	"encoding/json"
	"fmt"
	"math"
	"strconv"
	"strings"

	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
)

// executeApplyConnectorFieldMappings implements the "applyConnectorFieldMappings"
// generic invocable action.
//
// Expected inputParameters (any may be omitted if the flow has not yet
// resolved a value):
//
//   - source       string — optional assertion matching the executing Flow owner pack
//   - sourceObject string — API name of the source SObject (required)
//   - targetObject string — API name of the target SObject (required)
//   - actionType   string — mapping action type scope (required)
//   - direction    string — "inbound", "outbound", or "bidirectional" (required)
//   - sourceRecord map    — elementReference to the source payload record
//   - transformSourceRecord map — optional supplemental inputs for sparse
//     transforms. Only observed source fields select mappings; this record can
//     fill missing expression dependencies but never replace observed values.
//   - targetId     string — Id of the target record to update
//   - targetRecord map    — optional legacy caller snapshot; ConflictPolicy
//     enforcement re-reads the authoritative target row by targetId.
//   - extraUpdates map or []{field,value} — optional additional target fields
//     to merge into the same target update after connector mappings.
//   - applyExtraUpdates boolean — optional guard for extraUpdates; defaults true.
//   - filterMappingsBySourcePresence boolean — optional opt-in sparse-payload
//     mode. When true, only mappings whose source inputs are present are applied.
//     This mode cannot be combined with applyExtraUpdates.
//   - coalescedPropertyChanges boolean — optional opt-in reducer for a settled
//     property-change projection. When true, coalescedPayloads is reduced by
//     property name and newest occurredAt before sparse mapping selection.
//   - coalescedPayloads collection — settled property-change envelopes whose
//     payloads contain propertyName, propertyValue, and occurredAt.
//   - coalescedSourceDerivations collection — optional declarations that
//     materialize an occurredAt value into a sparse source field when the
//     selected source field has a specified value.
//   - coalescedSourceFieldRegexValidations collection — optional metadata
//     constraints applied to selected coalesced source values before mapping.
//     Each entry requires sourceField and pattern; allowBlank defaults false.
//
// Writes the mapped fields to the target via persistence.Update using the same
// seam that every other record-update action uses.
func (e *FlowExecutor) executeApplyConnectorFieldMappings(ctx context.Context, interview *models.FlowInterview, action *models.FlowActionCall, user *models.UserSession) (string, error) {
	fault := func(err error) (string, error) {
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, err)
		}
		return "", err
	}

	coalescedSourceFieldRegexValidationsAuthored := false
	transformSourceRecordAuthored := false
	for _, p := range action.InputParameters {
		if p.Name == "transformSourceRecord" {
			if transformSourceRecordAuthored {
				return fault(fmt.Errorf("applyConnectorFieldMappings: duplicate transformSourceRecord input"))
			}
			transformSourceRecordAuthored = true
		}
		if p.Name != "coalescedSourceFieldRegexValidations" {
			continue
		}
		if coalescedSourceFieldRegexValidationsAuthored {
			return fault(fmt.Errorf("applyConnectorFieldMappings: duplicate coalescedSourceFieldRegexValidations input"))
		}
		coalescedSourceFieldRegexValidationsAuthored = true
	}

	// Resolve all input parameters from the interview context.
	params := make(map[string]interface{}, len(action.InputParameters))
	for _, p := range action.InputParameters {
		if p.Name == "coalescedSourceFieldRegexValidations" {
			resolved, err := e.resolveConnectorCoalescedSourceFieldRegexValidationInput(ctx, interview, user, p.Value)
			if err != nil {
				return fault(fmt.Errorf("applyConnectorFieldMappings: coalescedSourceFieldRegexValidations: %w", err))
			}
			params[p.Name] = resolved
			continue
		}
		resolved, err := e.deepResolveFlowResource(ctx, interview, user, p.Value)
		if err != nil {
			return fault(err)
		}
		params[p.Name] = resolved
	}

	sourceAssertion, _ := params["source"].(string)
	sourceAssertion = strings.TrimSpace(sourceAssertion)
	sourceObject, _ := params["sourceObject"].(string)
	sourceObject = strings.TrimSpace(sourceObject)
	targetObject, _ := params["targetObject"].(string)
	targetObject = strings.TrimSpace(targetObject)
	actionType, _ := params["actionType"].(string)
	actionType = strings.TrimSpace(actionType)
	direction, _ := params["direction"].(string)
	direction = strings.TrimSpace(direction)
	targetID, _ := params["targetId"].(string)
	targetID = strings.TrimSpace(targetID)

	flowIdentity, ok := flowRuntimeContextFromCtx(ctx).CurrentFlow()
	source := strings.TrimSpace(flowIdentity.OwnerPack)
	if !ok || source == "" {
		return fault(fmt.Errorf("applyConnectorFieldMappings: executing Flow owner pack is required"))
	}
	if !flowIdentityIsPackOwnedStandard(flowIdentity) {
		return fault(fmt.Errorf("applyConnectorFieldMappings: Flow %q must be an uncustomized standard FlowVersion to resolve pack-owned connector mappings", flowIdentity.APIName))
	}
	if sourceAssertion != "" && sourceAssertion != source {
		return fault(fmt.Errorf("applyConnectorFieldMappings: source assertion %q does not match executing Flow owner pack %q", sourceAssertion, source))
	}
	if sourceObject == "" {
		return fault(fmt.Errorf("applyConnectorFieldMappings: sourceObject must not be blank"))
	}
	if targetObject == "" {
		return fault(fmt.Errorf("applyConnectorFieldMappings: targetObject must not be blank"))
	}
	if actionType == "" {
		return fault(fmt.Errorf("applyConnectorFieldMappings: actionType must not be blank"))
	}
	if direction == "" {
		return fault(fmt.Errorf("applyConnectorFieldMappings: direction must not be blank"))
	}
	if targetID == "" {
		return fault(fmt.Errorf("applyConnectorFieldMappings: targetId must not be blank"))
	}

	sourceRecord := coerceFlowRecord(params["sourceRecord"])
	if sourceRecord == nil {
		sourceRecord = models.SObject{}
	}
	mappingSourceRecord := sourceRecord
	filterMappingsBySourcePresence := connectorFieldMappingFilterMappingsBySourcePresence(params["filterMappingsBySourcePresence"])
	applyExtraUpdates := connectorFieldMappingApplyExtraUpdates(params["applyExtraUpdates"])
	skipBlankTargetFields := connectorFieldMappingSkipBlankTargetFields(params["skipBlankTargetFields"])
	coalescedPropertyChanges := connectorFieldMappingCoalescedPropertyChanges(params["coalescedPropertyChanges"])
	var transformSourceRecord models.SObject
	if transformSourceRecordAuthored {
		if !filterMappingsBySourcePresence && !coalescedPropertyChanges {
			return fault(fmt.Errorf("applyConnectorFieldMappings: transformSourceRecord requires sparse mapping selection"))
		}
		transformSourceRecord = coerceFlowRecord(params["transformSourceRecord"])
		if transformSourceRecord == nil {
			return fault(fmt.Errorf("applyConnectorFieldMappings: transformSourceRecord must resolve to a record"))
		}
	}
	rawCoalescedSourceFieldRegexValidations := params["coalescedSourceFieldRegexValidations"]
	if coalescedSourceFieldRegexValidationsAuthored && !coalescedPropertyChanges {
		return fault(fmt.Errorf("applyConnectorFieldMappings: coalescedSourceFieldRegexValidations requires coalescedPropertyChanges=true"))
	}
	if coalescedSourceFieldRegexValidationsAuthored && rawCoalescedSourceFieldRegexValidations == nil {
		return fault(fmt.Errorf("applyConnectorFieldMappings: coalescedSourceFieldRegexValidations must resolve to a collection"))
	}
	coalescedSourceFieldRegexValidations, err := connectorCoalescedSourceFieldRegexValidations(rawCoalescedSourceFieldRegexValidations)
	if err != nil {
		return fault(fmt.Errorf("applyConnectorFieldMappings: coalescedSourceFieldRegexValidations: %w", err))
	}
	if filterMappingsBySourcePresence && applyExtraUpdates {
		return fault(fmt.Errorf("applyConnectorFieldMappings: filterMappingsBySourcePresence cannot be used with applyExtraUpdates"))
	}
	if coalescedPropertyChanges && applyExtraUpdates {
		return fault(fmt.Errorf("applyConnectorFieldMappings: coalescedPropertyChanges cannot be used with applyExtraUpdates"))
	}
	// Resolve mappings via the test seam when set, otherwise via the
	// production ConnectorMappingResolver.
	resolverFn := e.connectorMappingResolverFn
	if resolverFn == nil {
		if e.connectorMappingResolver == nil {
			return fault(fmt.Errorf("applyConnectorFieldMappings: no connector mapping resolver wired (set connectorMappingResolver or connectorMappingResolverFn)"))
		}
		resolverFn = e.connectorMappingResolver.ResolveConnectorMappings
	}

	q := connector.ResolveQuery{
		Source:       source,
		SourceObject: sourceObject,
		TargetObject: targetObject,
		ActionType:   actionType,
		Direction:    direction,
	}
	mappings, err := resolverFn(ctx, q)
	if err != nil {
		return fault(fmt.Errorf("applyConnectorFieldMappings: resolve mappings: %w", err))
	}
	// Validate the resolver's complete configuration before sparse filtering. A
	// partial source payload must not hide a duplicate or invalid target mapping.
	if err := validateConnectorFieldMappingRuntimeTargetWrites(mappings); err != nil {
		return fault(err)
	}
	if err := e.validateConnectorFieldMappingTargetSchema(ctx, targetObject, mappings); err != nil {
		return fault(err)
	}
	if coalescedPropertyChanges {
		mappingSourceRecord, err = reduceConnectorCoalescedPropertyChanges(
			params["coalescedPayloads"],
			params["coalescedSourceDerivations"],
		)
		if err != nil {
			return fault(fmt.Errorf("applyConnectorFieldMappings: reduce coalesced property changes: %w", err))
		}
		if err := validateConnectorCoalescedSourceFieldRegexes(mappingSourceRecord, coalescedSourceFieldRegexValidations); err != nil {
			return fault(fmt.Errorf("applyConnectorFieldMappings: validate coalesced source fields: %w", err))
		}
		filterMappingsBySourcePresence = true
	}
	if filterMappingsBySourcePresence {
		if transformSourceRecordAuthored {
			mappingSourceRecord, mappings, err = e.supplementConnectorTransformInputs(mappingSourceRecord, transformSourceRecord, mappings)
			if err != nil {
				return fault(err)
			}
		}
		mappingSourceRecord, mappings, err = e.filterConnectorFieldMappingsBySourcePresence(mappingSourceRecord, mappings, coalescedPropertyChanges)
		if err != nil {
			return fault(err)
		}
	}
	targetFields := e.connectorFieldMappingTargetFields(ctx, targetObject)
	var extraUpdates models.SObject
	if applyExtraUpdates {
		var err error
		extraUpdates, err = e.connectorFieldMappingExtraUpdates(params["extraUpdates"], targetObject, targetFields)
		if err != nil {
			return fault(err)
		}
	}

	needsAuthoritativeTarget, err := connectorMappingsNeedAuthoritativeTargetRecord(mappings)
	if err != nil {
		return fault(err)
	}

	applyUpdates := func(writeCtx context.Context, targetRecord models.SObject) error {
		updates, err := e.buildConnectorFieldMappingUpdates(mappingSourceRecord, targetRecord, mappings, targetFields, true, skipBlankTargetFields)
		if err != nil {
			return err
		}
		if err := mergeConnectorFieldMappingExtraUpdates(updates, extraUpdates); err != nil {
			return err
		}
		if len(updates) == 0 {
			return nil
		}
		// Preview must not perform DML. Mirror executeRecordUpdate's preview guard.
		if isPreviewMode(ctx) {
			return errPreviewRecordUpdateAttempt
		}
		if e.persistence == nil {
			return fmt.Errorf("applyConnectorFieldMappings: no persistence provider wired for target update")
		}
		if err := e.persistence.Update(writeCtx, targetObject, targetID, updates, user); err != nil {
			return fmt.Errorf(
				"applyConnectorFieldMappings: update target %s/%s: %w",
				targetObject,
				targetID,
				flowRecordMutationValidation(err),
			)
		}
		// Publish the engine-confirmed upsert into the post-upsert registry so the
		// connector-inbound register action (issue #2080) resolves the source
		// identity from this connector-field-mapping write too — it is the 4th
		// FlowExecutor inbound source-record write path, alongside recordCreate,
		// applyRecordUpdatePersistence, and CommitPreparedUpdate. Without this,
		// an inbound flow that stamps remoteIDField via field mappings then calls
		// register would blank-no-op and strand the record without evidence.
		// Gate on writeCtx (the Update's ctx), not the outer ctx: the write
		// transaction lives on writeCtx (may be an inner pessimistic-tx ctx),
		// so IsTransactionActive must be checked there, not on the outer ctx.
		snap := make(models.SObject, len(updates)+1)
		for k, v := range updates {
			snap[k] = v
		}
		snap[constants.FieldID] = targetID
		e.publishPostUpsertRecord(writeCtx, targetObject, snap)
		return nil
	}

	if needsAuthoritativeTarget {
		if e.persistence == nil {
			return fault(fmt.Errorf("applyConnectorFieldMappings: ConflictPolicy requires authoritative target state but no persistence provider is wired"))
		}
		if isPreviewMode(ctx) {
			targetRecord, err := e.loadConnectorConflictPreviewTargetRecord(ctx, targetObject, targetID)
			if err != nil {
				return fault(err)
			}
			if err := applyUpdates(ctx, targetRecord); err != nil {
				return fault(err)
			}
		} else {
			err := e.persistence.RunInPessimisticTransaction(ctx, func(txCtx context.Context) error {
				targetRecord, err := e.persistence.GetLock(txCtx, targetObject, targetID)
				if err != nil {
					return fmt.Errorf("applyConnectorFieldMappings: lock authoritative target %s/%s for ConflictPolicy: %w", targetObject, targetID, err)
				}
				if targetRecord == nil {
					return fmt.Errorf("applyConnectorFieldMappings: target %s/%s not found for ConflictPolicy", targetObject, targetID)
				}
				return applyUpdates(txCtx, targetRecord)
			})
			if err != nil {
				return fault(err)
			}
		}
	} else if err := applyUpdates(ctx, models.SObject{}); err != nil {
		return fault(err)
	}

	if action.Connector != nil {
		return action.Connector.TargetReference, nil
	}
	return "", nil
}

// supplementConnectorTransformInputs fixes the eligible target set from observed
// fields before copying any supplemental values. TransformExpression, interpreted
// by the canonical formula engine, is the only dependency declaration. A copied
// dependency can therefore never make another mapping eligible for a write.
func (e *FlowExecutor) supplementConnectorTransformInputs(observed, supplemental models.SObject, mappings []connector.ConnectorFieldMappingInput) (models.SObject, []connector.ConnectorFieldMappingInput, error) {
	inputs := make(models.SObject, len(observed))
	for field, value := range observed {
		inputs[field] = value
	}
	selected := make([]connector.ConnectorFieldMappingInput, 0, len(mappings))
	for _, mapping := range mappings {
		if strings.TrimSpace(mapping.TransformExpression) == "" {
			if _, present := lookupRecordFieldValue(observed, mapping.SourceField); present {
				selected = append(selected, mapping)
			}
			continue
		}
		if e.formulaEngine == nil {
			return nil, nil, fmt.Errorf("applyConnectorFieldMappings: transform %q on field %s requires a formula engine but none is wired", mapping.TransformExpression, mapping.SourceField)
		}
		fields, err := e.formulaEngine.ConnectorTransformReferencedSourceFields(mapping.TransformExpression)
		if err != nil {
			return nil, nil, fmt.Errorf("applyConnectorFieldMappings: transform %q on field %s: %w", mapping.TransformExpression, mapping.SourceField, err)
		}
		affected := false
		for _, field := range fields {
			if _, present := lookupRecordFieldValue(observed, field); present {
				affected = true
				break
			}
		}
		if !affected {
			continue
		}
		selected = append(selected, mapping)
		for _, field := range fields {
			if value, present := lookupRecordFieldValue(observed, field); present {
				inputs[field] = value
				continue
			}
			if value, present := lookupRecordFieldValue(supplemental, field); present {
				inputs[field] = value
			}
		}
	}
	return inputs, selected, nil
}

// filterConnectorFieldMappingsBySourcePresence removes mappings which a sparse
// source record cannot execute. Coalesced payloads preserve all-blank transform
// inputs because transforms treat them as authoritative clears; legacy sparse
// mode keeps its existing non-blank selection rule.
func (e *FlowExecutor) filterConnectorFieldMappingsBySourcePresence(sourceRecord models.SObject, mappings []connector.ConnectorFieldMappingInput, coalescedPropertyChanges bool) (models.SObject, []connector.ConnectorFieldMappingInput, error) {
	filteredSourceRecord := sourceRecord
	copiedSourceRecord := false
	selected := make([]connector.ConnectorFieldMappingInput, 0, len(mappings))
	for _, mapping := range mappings {
		if strings.TrimSpace(mapping.TransformExpression) == "" {
			_, sourcePresent := sourceRecord[mapping.SourceField]
			if !sourcePresent && coalescedPropertyChanges {
				value, foldedSourcePresent := lookupRecordFieldValue(sourceRecord, mapping.SourceField)
				if foldedSourcePresent {
					if !copiedSourceRecord {
						filteredSourceRecord = make(models.SObject, len(sourceRecord)+1)
						for key, sourceValue := range sourceRecord {
							filteredSourceRecord[key] = sourceValue
						}
						copiedSourceRecord = true
					}
					filteredSourceRecord[mapping.SourceField] = value
					sourcePresent = true
				}
			}
			if !sourcePresent {
				continue
			}
			selected = append(selected, mapping)
			continue
		}

		if e.formulaEngine == nil {
			return nil, nil, fmt.Errorf("applyConnectorFieldMappings: transform %q on field %s requires a formula engine but none is wired", mapping.TransformExpression, mapping.SourceField)
		}
		referencedFields, err := e.formulaEngine.ConnectorTransformReferencedSourceFields(mapping.TransformExpression)
		if err != nil {
			return nil, nil, fmt.Errorf("applyConnectorFieldMappings: transform %q on field %s: %w", mapping.TransformExpression, mapping.SourceField, err)
		}
		if !connectorTransformReferencedSourceFieldsPresent(sourceRecord, referencedFields, coalescedPropertyChanges) {
			continue
		}

		selected = append(selected, mapping)
		for _, field := range referencedFields {
			field = strings.TrimSpace(field)
			if field == "" {
				continue
			}
			if _, present := filteredSourceRecord[field]; present {
				continue
			}
			value, present := lookupRecordFieldValue(sourceRecord, field)
			if !present {
				continue
			}
			if !copiedSourceRecord {
				filteredSourceRecord = make(models.SObject, len(sourceRecord)+len(referencedFields))
				for key, sourceValue := range sourceRecord {
					filteredSourceRecord[key] = sourceValue
				}
				copiedSourceRecord = true
			}
			filteredSourceRecord[field] = value
		}
	}
	return filteredSourceRecord, selected, nil
}

func connectorTransformReferencedSourceFieldsPresent(sourceRecord models.SObject, fields []string, allowAllBlank bool) bool {
	hasReferencedField := false
	hasNonBlankReferencedValue := false
	for _, field := range fields {
		field = strings.TrimSpace(field)
		if field == "" {
			continue
		}
		hasReferencedField = true
		value, present := lookupRecordFieldValue(sourceRecord, field)
		if !present {
			return false
		}
		if !isBlankValue(value) {
			hasNonBlankReferencedValue = true
		}
	}
	return hasReferencedField && (allowAllBlank || hasNonBlankReferencedValue)
}

func (e *FlowExecutor) buildConnectorFieldMappingUpdates(sourceRecord, targetRecord models.SObject, mappings []connector.ConnectorFieldMappingInput, targetFields map[string]models.FieldMetadata, allowOverwriteBlankClears bool, skipBlankTargetFields map[string]struct{}) (models.SObject, error) {
	return buildConnectorFieldMappingUpdatesWithEngine(
		e.formulaEngine,
		sourceRecord,
		targetRecord,
		mappings,
		targetFields,
		allowOverwriteBlankClears,
		skipBlankTargetFields,
	)
}

func buildConnectorFieldMappingUpdatesWithEngine(
	formulaEngine *formula.Engine,
	sourceRecord, targetRecord models.SObject,
	mappings []connector.ConnectorFieldMappingInput,
	targetFields map[string]models.FieldMetadata,
	allowOverwriteBlankClears bool,
	skipBlankTargetFields map[string]struct{},
) (models.SObject, error) {
	if targetRecord == nil {
		targetRecord = models.SObject{}
	}

	// Apply each mapping row: conflict-policy → transform → collect.
	// skip_if_set is checked BEFORE the transform so it can short-circuit
	// without evaluating a potentially-faulting transform. error is checked
	// only after we know this sparse payload has a value to write.
	updates := make(models.SObject, len(mappings))
	if err := validateConnectorFieldMappingRuntimeTargetWrites(mappings); err != nil {
		return nil, err
	}
	for _, m := range mappings {
		// Step 1: enforce ConflictPolicy early.
		// Empty/unset ConflictPolicy defaults to "overwrite" — the value is
		// always written to the target regardless of existing content.
		policy := connectorConflictPolicy(m.ConflictPolicy)

		switch policy {
		case connector.ConflictPolicyOverwrite:
			// Always write — fall through to value derivation.
		case connector.ConflictPolicySkipIfSet:
			if existing, ok := lookupRecordFieldValue(targetRecord, m.TargetField); ok && !isBlankValue(existing) {
				continue
			}
		case connector.ConflictPolicyError:
			// Defer the conflict check until after value derivation. An absent
			// source/default/transform output is a sparse-payload no-op, not a
			// conflicting write attempt.
		default:
			// Install-time validation restricts ConflictPolicy to the known set,
			// but a live row read from _System_ObjectFieldMapping could carry a
			// drifted value (manual edit, migration, future writer bug). Fail
			// closed rather than fall through to an implicit overwrite, which
			// would silently apply the wrong conflict semantics.
			return nil, fmt.Errorf("applyConnectorFieldMappings: ConflictPolicy %q on field %s is not a recognised value (expected overwrite, skip_if_set, error, or empty)", policy, m.TargetField)
		}

		// Step 2: derive the source value.
		var value interface{}
		hasValue := false
		valueDerivedByTransform := false
		if v, ok := sourceRecord[m.SourceField]; ok && v != nil {
			value = v
			hasValue = true
		} else if m.DefaultValue != "" {
			value = m.DefaultValue
			hasValue = true
		}

		// Step 3: apply TransformExpression when present. A non-empty
		// expression with no formula engine wired is a fail-closed error —
		// never silently write the raw, untransformed value.
		if m.TransformExpression != "" {
			if formulaEngine == nil {
				return nil, fmt.Errorf("applyConnectorFieldMappings: transform %q on field %s requires a formula engine but none is wired", m.TransformExpression, m.SourceField)
			}
			record := make(models.SObject, len(sourceRecord)+1)
			for k, v := range sourceRecord {
				record[k] = v
			}
			if v, present := record[m.SourceField]; (!present || v == nil) && m.DefaultValue != "" {
				record[m.SourceField] = m.DefaultValue
			}
			referencedFields, err := formulaEngine.ConnectorTransformReferencedSourceFields(m.TransformExpression)
			if err != nil {
				return nil, fmt.Errorf("applyConnectorFieldMappings: transform %q on field %s: %w", m.TransformExpression, m.SourceField, err)
			}
			if connectorTransformReferencesMissingSource(record, referencedFields) {
				continue
			}
			fctx := &formula.Context{
				Record: record,
			}
			evaluated, evalErr := formulaEngine.Evaluate(m.TransformExpression, fctx)
			if evalErr != nil {
				return nil, fmt.Errorf("applyConnectorFieldMappings: transform %q on field %s: %w", m.TransformExpression, m.SourceField, evalErr)
			}
			// A transform must yield a single scalar field value. A map/slice
			// result — a bare $Record reference, or a transform returning a nested
			// object/array from the source payload — must never be written straight
			// into one target field. Fail closed instead of persisting a structure.
			switch evaluated.(type) {
			case map[string]interface{}, []interface{}, models.SObject:
				return nil, fmt.Errorf("applyConnectorFieldMappings: transform %q on field %s produced a non-scalar %T; a transform must yield a single field value", m.TransformExpression, m.SourceField, evaluated)
			}
			value = evaluated
			hasValue = hasValue || evaluated != nil
			valueDerivedByTransform = true
		}

		// Sparse-payload safety: if the source field is absent, no default was
		// supplied, and no transform produced a value, leave the target field
		// unchanged. Never overwrite an existing value with NULL just because an
		// inbound payload was sparse (mirrors legacy ApplyFieldMappings).
		if !hasValue {
			continue
		}

		valueMapTranslated := false
		if strings.TrimSpace(m.ValueMap) != "" {
			translated, include, matched, err := connector.TranslateMappedValue(m.ValueMap, m.OnUnmappedValue, value)
			if err != nil {
				return nil, fmt.Errorf("applyConnectorFieldMappings: ValueMap on field %s: %w", m.SourceField, err)
			}
			if !include {
				continue
			}
			value = translated
			valueMapTranslated = matched
		}

		if connectorRequiredBlankClearShouldSkip(m, value, targetFields) {
			continue
		}
		if allowOverwriteBlankClears {
			value = normalizeConnectorOverwriteBlankClear(m, value, targetFields)
		}
		metadataAuthoredBlankClear := valueDerivedByTransform || valueMapTranslated
		_, skipBlankTarget := skipBlankTargetFields[strings.ToLower(strings.TrimSpace(m.TargetField))]
		if isBlankValue(value) && !metadataAuthoredBlankClear && policy == connector.ConflictPolicyOverwrite && (!allowOverwriteBlankClears || skipBlankTarget) {
			continue
		}

		if policy == connector.ConflictPolicyError {
			if existing, ok := lookupRecordFieldValue(targetRecord, m.TargetField); ok && !isBlankValue(existing) {
				return nil, fmt.Errorf("applyConnectorFieldMappings: ConflictPolicy=error: target field %s already has a value", m.TargetField)
			}
		}

		updates[m.TargetField] = value
	}

	return updates, nil
}

func connectorRequiredBlankClearShouldSkip(m connector.ConnectorFieldMappingInput, value interface{}, targetFields map[string]models.FieldMetadata) bool {
	if connectorConflictPolicy(m.ConflictPolicy) != connector.ConflictPolicyOverwrite || !isBlankValue(value) {
		return false
	}
	field, ok := targetFields[strings.ToLower(strings.TrimSpace(m.TargetField))]
	if !ok {
		return false
	}
	sem := models.ResolveFieldRelationshipSemantics(&field, models.LifecyclePersistedRead)
	return field.Required || sem.IsMasterDetail
}

func (e *FlowExecutor) connectorFieldMappingTargetFields(ctx context.Context, targetObject string) map[string]models.FieldMetadata {
	schema := e.lookupObjectSchema(ctx, targetObject)
	if schema == nil {
		return nil
	}
	fields := make(map[string]models.FieldMetadata, len(schema.Fields))
	for _, field := range schema.Fields {
		key := strings.ToLower(strings.TrimSpace(field.APIName))
		if key == "" {
			continue
		}
		fields[key] = field
	}
	return fields
}

func normalizeConnectorOverwriteBlankClear(m connector.ConnectorFieldMappingInput, value interface{}, targetFields map[string]models.FieldMetadata) interface{} {
	if connectorConflictPolicy(m.ConflictPolicy) != connector.ConflictPolicyOverwrite || !isBlankValue(value) {
		return value
	}
	field, ok := targetFields[strings.ToLower(strings.TrimSpace(m.TargetField))]
	if !ok || connectorBlankClearPreservesStringValue(field) {
		return value
	}
	return nil
}

func connectorBlankClearPreservesStringValue(field models.FieldMetadata) bool {
	switch constants.NormalizeFieldType(string(field.Type)) {
	case constants.FieldTypeString,
		constants.FieldTypeTextarea,
		constants.FieldTypeEmail,
		constants.FieldTypePhone,
		constants.FieldTypeUrl,
		constants.FieldTypePicklist,
		constants.FieldTypeMultiPicklist,
		constants.FieldTypeRichtextarea,
		constants.FieldTypeEncryptedString,
		constants.FieldTypePassword:
		return true
	default:
		return false
	}
}

func connectorTransformReferencesMissingSource(record models.SObject, fields []string) bool {
	if len(fields) == 0 {
		return false
	}
	for _, field := range fields {
		if _, ok := lookupRecordFieldValue(record, field); !ok {
			return true
		}
	}
	return false
}

func validateConnectorFieldMappingRuntimeTargetWrites(mappings []connector.ConnectorFieldMappingInput) error {
	type keyInfo struct {
		index       int
		sourceField string
	}
	seen := make(map[string]keyInfo, len(mappings))
	for i, m := range mappings {
		targetField := strings.TrimSpace(m.TargetField)
		targetFieldKey := strings.ToLower(targetField)
		if targetFieldKey == "" {
			return fmt.Errorf("applyConnectorFieldMappings: target field must not be blank")
		}
		sourceField := strings.TrimSpace(m.SourceField)
		if prev, dup := seen[targetFieldKey]; dup {
			if strings.EqualFold(prev.sourceField, sourceField) {
				continue
			}
			return fmt.Errorf("applyConnectorFieldMappings: duplicate target field %s from source field %s conflicts with row %d source field %s", targetField, sourceField, prev.index, prev.sourceField)
		}
		seen[targetFieldKey] = keyInfo{index: i, sourceField: sourceField}
	}
	return nil
}

func (e *FlowExecutor) validateConnectorFieldMappingTargetSchema(ctx context.Context, targetObject string, mappings []connector.ConnectorFieldMappingInput) error {
	return e.validateConnectorFieldMappingTargetSchemaForWrite(ctx, targetObject, mappings, false)
}

func (e *FlowExecutor) validateConnectorFieldMappingCreateTargetSchema(ctx context.Context, targetObject string, mappings []connector.ConnectorFieldMappingInput) error {
	return e.validateConnectorFieldMappingTargetSchemaForWrite(ctx, targetObject, mappings, true)
}

func (e *FlowExecutor) validateConnectorFieldMappingTargetSchemaForWrite(ctx context.Context, targetObject string, mappings []connector.ConnectorFieldMappingInput, create bool) error {
	if len(mappings) == 0 {
		return nil
	}
	// Unit tests can construct a FlowExecutor without metadata wiring. Production
	// executors have a metadata service; when that read path exists but cannot
	// resolve the target schema, fail closed below.
	if e == nil || (e.objectSchemaLookup == nil && e.metadata == nil) {
		return nil
	}
	schema := e.lookupObjectSchema(ctx, targetObject)
	if schema == nil {
		return fmt.Errorf("applyConnectorFieldMappings: target object %s schema is unavailable; cannot validate connector mapping target fields", targetObject)
	}
	fields := make(map[string]models.FieldMetadata, len(schema.Fields))
	for _, field := range schema.Fields {
		fields[strings.ToLower(strings.TrimSpace(field.APIName))] = field
	}
	for _, m := range mappings {
		targetField := strings.TrimSpace(m.TargetField)
		field, ok := fields[strings.ToLower(targetField)]
		if !ok {
			return fmt.Errorf("applyConnectorFieldMappings: target field %s.%s does not exist", targetObject, targetField)
		}
		if constants.NormalizeFieldType(string(field.Type)) == constants.FieldTypeAutonumber {
			return fmt.Errorf("applyConnectorFieldMappings: target field %s.%s is autonumber and cannot be written by connector mappings", targetObject, targetField)
		}
		if create && !field.Createable {
			return fmt.Errorf("applyConnectorFieldMappings: target field %s.%s is not createable", targetObject, targetField)
		}
		if !create && !field.Updateable {
			return fmt.Errorf("applyConnectorFieldMappings: target field %s.%s is not updateable", targetObject, targetField)
		}
	}
	return nil
}

// connectorFieldMappingSkipBlankTargetFields normalizes the optional Flow list
// of mapping targets whose raw blank overwrite values should be omitted.
func connectorFieldMappingSkipBlankTargetFields(raw interface{}) map[string]struct{} {
	fields := map[string]struct{}{}
	add := func(value interface{}) {
		field := strings.ToLower(strings.TrimSpace(fmt.Sprint(value)))
		if field != "" {
			fields[field] = struct{}{}
		}
	}
	switch values := raw.(type) {
	case []interface{}:
		for _, value := range values {
			add(value)
		}
	case []string:
		for _, value := range values {
			add(value)
		}
	}
	return fields
}

func (e *FlowExecutor) connectorFieldMappingExtraUpdates(raw interface{}, targetObject string, targetFields map[string]models.FieldMetadata) (models.SObject, error) {
	return e.connectorFieldMappingExtraUpdatesForWrite(raw, targetObject, targetFields, false)
}

func (e *FlowExecutor) connectorFieldMappingCreateExtraUpdates(raw interface{}, targetObject string, targetFields map[string]models.FieldMetadata) (models.SObject, error) {
	return e.connectorFieldMappingExtraUpdatesForWrite(raw, targetObject, targetFields, true)
}

func (e *FlowExecutor) connectorFieldMappingExtraUpdatesForWrite(raw interface{}, targetObject string, targetFields map[string]models.FieldMetadata, create bool) (models.SObject, error) {
	if raw == nil {
		return nil, nil
	}
	updates := models.SObject{}
	add := func(field string, value interface{}) error {
		field = strings.TrimSpace(field)
		if field == "" {
			return fmt.Errorf("applyConnectorFieldMappings: extraUpdates field must not be blank")
		}
		if _, dup := lookupRecordFieldValue(updates, field); dup {
			return fmt.Errorf("applyConnectorFieldMappings: duplicate extraUpdates field %s", field)
		}
		if len(targetFields) > 0 {
			meta, ok := targetFields[strings.ToLower(field)]
			if !ok {
				return fmt.Errorf("applyConnectorFieldMappings: target field %s.%s does not exist", targetObject, field)
			}
			if constants.NormalizeFieldType(string(meta.Type)) == constants.FieldTypeAutonumber {
				return fmt.Errorf("applyConnectorFieldMappings: target field %s.%s is autonumber and cannot be written by connector mappings", targetObject, field)
			}
			if create && !meta.Createable {
				return fmt.Errorf("applyConnectorFieldMappings: target field %s.%s is not createable", targetObject, field)
			}
			if !create && !meta.Updateable {
				return fmt.Errorf("applyConnectorFieldMappings: target field %s.%s is not updateable", targetObject, field)
			}
		}
		updates[field] = value
		return nil
	}

	switch v := raw.(type) {
	case models.SObject:
		for field, value := range v {
			if err := add(field, value); err != nil {
				return nil, err
			}
		}
	case map[string]interface{}:
		for field, value := range v {
			if err := add(field, value); err != nil {
				return nil, err
			}
		}
	case []interface{}:
		for _, item := range v {
			itemMap, ok := item.(map[string]interface{})
			if !ok {
				return nil, fmt.Errorf("applyConnectorFieldMappings: extraUpdates list entries must be objects with field and value")
			}
			field, _ := itemMap["field"].(string)
			if field == "" {
				if name, ok := itemMap["name"].(string); ok {
					field = name
				}
			}
			if err := add(field, itemMap["value"]); err != nil {
				return nil, err
			}
		}
	default:
		return nil, fmt.Errorf("applyConnectorFieldMappings: extraUpdates must be an object or list of {field,value} entries")
	}
	if len(updates) == 0 {
		return nil, nil
	}
	return updates, nil
}

func connectorFieldMappingApplyExtraUpdates(raw interface{}) bool {
	if raw == nil {
		return true
	}
	switch v := raw.(type) {
	case bool:
		return v
	case string:
		return !strings.EqualFold(strings.TrimSpace(v), "false")
	default:
		return true
	}
}

func connectorFieldMappingFilterMappingsBySourcePresence(raw interface{}) bool {
	switch v := raw.(type) {
	case bool:
		return v
	case string:
		parsed, err := strconv.ParseBool(strings.TrimSpace(v))
		return err == nil && parsed
	default:
		return false
	}
}

// connectorCoalescedPropertyChange is the newest settled source delta for one
// remote property. The collection order is only used as a deterministic tie
// breaker; occurredAt remains the primary ordering key.
type connectorCoalescedPropertyChange struct {
	value      interface{}
	occurredAt int64
}

type connectorCoalescedSourceDerivation struct {
	sourceField string
	fromField   string
	whenEquals  string
}

type connectorCoalescedSourceFieldRegexValidation struct {
	sourceField string
	pattern     string
	allowBlank  bool
}

// resolveConnectorCoalescedSourceFieldRegexValidationInput preserves the
// authored strings inside an inline validation declaration as metadata
// literals. The top level is a closed grammar: either a direct collection or
// one element reference that supplies the collection at runtime. Nested typed
// literals are unwrapped only when their wrapper has one correctly typed arm.
func (e *FlowExecutor) resolveConnectorCoalescedSourceFieldRegexValidationInput(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, raw interface{}) (interface{}, error) {
	switch value := raw.(type) {
	case nil:
		return nil, fmt.Errorf("authored input must not be nil")
	case map[string]interface{}:
		if len(value) != 1 {
			return nil, fmt.Errorf("top-level wrapper must contain exactly one elementReference arm")
		}
		rawReference, isReference := value["elementReference"]
		if !isReference {
			return nil, fmt.Errorf("top-level wrapper must contain exactly one elementReference arm")
		}
		reference, ok := rawReference.(string)
		if !ok || strings.TrimSpace(reference) == "" {
			return nil, fmt.Errorf("top-level elementReference must be a non-blank string")
		}
		resolved, _, err := e.ResolveFlowResource(ctx, interview, user, raw)
		return resolved, err
	case []interface{}, []models.SObject, []map[string]interface{}:
		return unwrapConnectorCoalescedSourceFieldRegexValidationLiterals(raw)
	default:
		return nil, fmt.Errorf("input must be a direct collection or an elementReference wrapper")
	}
}

func unwrapConnectorCoalescedSourceFieldRegexValidationLiterals(raw interface{}) (interface{}, error) {
	switch value := raw.(type) {
	case map[string]interface{}:
		literalKeys := []string{"stringValue", "numberValue", "booleanValue", "dateValue", "dateTimeValue"}
		literalKey := ""
		for _, key := range literalKeys {
			if _, present := value[key]; !present {
				continue
			}
			if literalKey != "" {
				return nil, fmt.Errorf("nested typed literal wrapper must contain exactly one literal arm")
			}
			literalKey = key
		}
		if literalKey != "" {
			if len(value) != 1 {
				return nil, fmt.Errorf("nested typed literal wrapper must contain exactly one literal arm")
			}
			literal := value[literalKey]
			if literal == nil || !connectorCoalescedValidationLiteralHasExpectedType(literalKey, literal) {
				return nil, fmt.Errorf("nested %s must contain a non-nil scalar of its declared type", literalKey)
			}
			return literal, nil
		}
		out := make(map[string]interface{}, len(value))
		for key, item := range value {
			unwrapped, err := unwrapConnectorCoalescedSourceFieldRegexValidationLiterals(item)
			if err != nil {
				return nil, err
			}
			out[key] = unwrapped
		}
		return out, nil
	case []interface{}:
		out := make([]interface{}, len(value))
		for index, item := range value {
			unwrapped, err := unwrapConnectorCoalescedSourceFieldRegexValidationLiterals(item)
			if err != nil {
				return nil, err
			}
			out[index] = unwrapped
		}
		return out, nil
	case []models.SObject:
		out := make([]interface{}, len(value))
		for index, item := range value {
			unwrapped, err := unwrapConnectorCoalescedSourceFieldRegexValidationLiterals(item)
			if err != nil {
				return nil, err
			}
			out[index] = unwrapped
		}
		return out, nil
	case []map[string]interface{}:
		out := make([]interface{}, len(value))
		for index, item := range value {
			unwrapped, err := unwrapConnectorCoalescedSourceFieldRegexValidationLiterals(item)
			if err != nil {
				return nil, err
			}
			out[index] = unwrapped
		}
		return out, nil
	default:
		return raw, nil
	}
}

func connectorCoalescedValidationLiteralHasExpectedType(key string, value interface{}) bool {
	switch key {
	case "stringValue", "dateValue", "dateTimeValue":
		_, ok := value.(string)
		return ok
	case "booleanValue":
		_, ok := value.(bool)
		return ok
	case "numberValue":
		switch value.(type) {
		case int, int8, int16, int32, int64,
			uint, uint8, uint16, uint32, uint64,
			float32, float64, json.Number:
			return true
		default:
			return false
		}
	default:
		return false
	}
}

func connectorFieldMappingCoalescedPropertyChanges(raw interface{}) bool {
	return connectorFieldMappingFilterMappingsBySourcePresence(raw)
}

func connectorCoalescedSourceFieldRegexValidations(raw interface{}) ([]connectorCoalescedSourceFieldRegexValidation, error) {
	if raw == nil {
		return nil, nil
	}
	records, err := connectorCoalescedRecordCollection(raw, "coalescedSourceFieldRegexValidations")
	if err != nil {
		return nil, err
	}
	validations := make([]connectorCoalescedSourceFieldRegexValidation, 0, len(records))
	seen := make(map[string]struct{}, len(records))
	for index, record := range records {
		rawSourceField, present := lookupRecordFieldValue(record, "sourceField")
		sourceField, ok := rawSourceField.(string)
		sourceField = strings.TrimSpace(sourceField)
		if !present || !ok || sourceField == "" {
			return nil, fmt.Errorf("entry %d sourceField must be a non-blank string", index)
		}
		key := strings.ToLower(sourceField)
		if _, duplicate := seen[key]; duplicate {
			return nil, fmt.Errorf("entry %d duplicates sourceField %q", index, sourceField)
		}

		rawPattern, present := lookupRecordFieldValue(record, "pattern")
		pattern, ok := rawPattern.(string)
		if !present || !ok || strings.TrimSpace(pattern) == "" {
			return nil, fmt.Errorf("entry %d pattern must be a non-blank RE2 string", index)
		}
		if _, valid := matchesFlowRegex("", pattern); !valid {
			return nil, fmt.Errorf("entry %d pattern must be a valid RE2 string", index)
		}

		allowBlank := false
		if rawAllowBlank, present := lookupRecordFieldValue(record, "allowBlank"); present {
			var ok bool
			allowBlank, ok = rawAllowBlank.(bool)
			if !ok {
				return nil, fmt.Errorf("entry %d allowBlank must be a boolean", index)
			}
		}
		seen[key] = struct{}{}
		validations = append(validations, connectorCoalescedSourceFieldRegexValidation{
			sourceField: sourceField,
			pattern:     pattern,
			allowBlank:  allowBlank,
		})
	}
	return validations, nil
}

func validateConnectorCoalescedSourceFieldRegexes(sourceRecord models.SObject, validations []connectorCoalescedSourceFieldRegexValidation) error {
	for _, validation := range validations {
		value, present := lookupRecordFieldValue(sourceRecord, validation.sourceField)
		if !present {
			continue
		}
		if isBlankValue(value) {
			if validation.allowBlank {
				continue
			}
			return fmt.Errorf("selected source field %q is blank but allowBlank is false", validation.sourceField)
		}
		matched, valid := matchesFlowRegex(value, validation.pattern)
		if !valid {
			return fmt.Errorf("selected source field %q has an invalid validation pattern", validation.sourceField)
		}
		if !matched {
			return fmt.Errorf("selected source field %q does not match its required pattern", validation.sourceField)
		}
	}
	return nil
}

// reduceConnectorCoalescedPropertyChanges turns a settled collection of
// property-change envelopes into one intentionally sparse mapping record.
// Source-field identity is case-insensitive, so case variants compete in the
// same newest-by-occurredAt selection. It never merges the caller's full
// sourceRecord: only properties actually observed in the coalesced window map.
func reduceConnectorCoalescedPropertyChanges(rawPayloads, rawDerivations interface{}) (models.SObject, error) {
	envelopes, err := connectorCoalescedRecordCollection(rawPayloads, "coalescedPayloads")
	if err != nil {
		return nil, err
	}

	selected := make(map[string]connectorCoalescedPropertyChange, len(envelopes))
	for index, envelope := range envelopes {
		payload := envelope
		if rawPayload, present := lookupRecordFieldValue(envelope, "payload"); present {
			payload = coerceFlowRecord(rawPayload)
			if payload == nil {
				return nil, fmt.Errorf("coalescedPayloads[%d].payload must be an object", index)
			}
		}

		rawPropertyName, present := lookupRecordFieldValue(payload, "propertyName")
		if !present {
			// Settled connector windows can contain non-property events. They do
			// not participate in this reducer.
			continue
		}
		propertyName, ok := rawPropertyName.(string)
		propertyName = strings.TrimSpace(propertyName)
		if !ok || propertyName == "" {
			return nil, fmt.Errorf("coalescedPayloads[%d].payload.propertyName must be a non-blank string", index)
		}
		propertyKey := strings.ToLower(propertyName)
		propertyValue, present := lookupRecordFieldValue(payload, "propertyValue")
		if !present {
			return nil, fmt.Errorf("coalescedPayloads[%d].payload.propertyValue is required for property %q", index, propertyName)
		}
		rawOccurredAt, present := lookupRecordFieldValue(payload, "occurredAt")
		if !present {
			return nil, fmt.Errorf("coalescedPayloads[%d].payload.occurredAt is required for property %q", index, propertyName)
		}
		occurredAt, err := connectorCoalescedOccurredAt(rawOccurredAt)
		if err != nil {
			return nil, fmt.Errorf("coalescedPayloads[%d].payload.occurredAt for property %q: %w", index, propertyName, err)
		}

		change := connectorCoalescedPropertyChange{value: propertyValue, occurredAt: occurredAt}
		if current, found := selected[propertyKey]; !found || change.occurredAt >= current.occurredAt {
			selected[propertyKey] = change
		}
	}

	derivations, err := connectorCoalescedDerivations(rawDerivations)
	if err != nil {
		return nil, err
	}
	reduced := make(models.SObject, len(selected)+len(derivations))
	for field, change := range selected {
		reduced[field] = change.value
	}
	for _, derivation := range derivations {
		fromFieldKey := strings.ToLower(derivation.fromField)
		change, found := selected[fromFieldKey]
		if !found || fmt.Sprint(change.value) != derivation.whenEquals {
			continue
		}
		sourceFieldKey := strings.ToLower(derivation.sourceField)
		if _, alreadySelected := selected[sourceFieldKey]; alreadySelected {
			return nil, fmt.Errorf("coalescedSourceDerivations sourceField %q conflicts with a coalesced payload property", derivation.sourceField)
		}
		reduced[sourceFieldKey] = change.occurredAt
	}
	return reduced, nil
}

func connectorCoalescedRecordCollection(raw interface{}, parameterName string) ([]models.SObject, error) {
	if raw == nil {
		return nil, fmt.Errorf("%s is required when coalescedPropertyChanges is true", parameterName)
	}

	var values []interface{}
	switch collection := raw.(type) {
	case []interface{}:
		values = collection
	case []models.SObject:
		values = make([]interface{}, len(collection))
		for index := range collection {
			values[index] = collection[index]
		}
	case []map[string]interface{}:
		values = make([]interface{}, len(collection))
		for index := range collection {
			values[index] = collection[index]
		}
	default:
		return nil, fmt.Errorf("%s must be a collection of objects", parameterName)
	}

	records := make([]models.SObject, 0, len(values))
	for index, value := range values {
		record := coerceFlowRecord(value)
		if record == nil {
			return nil, fmt.Errorf("%s[%d] must be an object", parameterName, index)
		}
		records = append(records, record)
	}
	return records, nil
}

func connectorCoalescedDerivations(raw interface{}) ([]connectorCoalescedSourceDerivation, error) {
	if raw == nil {
		return nil, nil
	}
	records, err := connectorCoalescedRecordCollection(raw, "coalescedSourceDerivations")
	if err != nil {
		return nil, err
	}

	derivations := make([]connectorCoalescedSourceDerivation, 0, len(records))
	for index, record := range records {
		sourceField, sourceFieldOK := lookupRecordString(record, "sourceField")
		fromField, fromFieldOK := lookupRecordString(record, "fromField")
		whenEquals, whenEqualsOK := lookupRecordString(record, "whenEquals")
		if !sourceFieldOK || !fromFieldOK || !whenEqualsOK {
			return nil, fmt.Errorf("coalescedSourceDerivations[%d] requires non-blank sourceField, fromField, and whenEquals strings", index)
		}
		if valueFrom, present := lookupRecordFieldValue(record, "valueFrom"); present {
			valueFromString, ok := valueFrom.(string)
			if !ok || strings.TrimSpace(valueFromString) != "occurredAt" {
				return nil, fmt.Errorf("coalescedSourceDerivations[%d].valueFrom must be occurredAt", index)
			}
		}
		derivations = append(derivations, connectorCoalescedSourceDerivation{
			sourceField: sourceField,
			fromField:   fromField,
			whenEquals:  whenEquals,
		})
	}
	return derivations, nil
}

func lookupRecordString(record models.SObject, field string) (string, bool) {
	raw, present := lookupRecordFieldValue(record, field)
	value, ok := raw.(string)
	value = strings.TrimSpace(value)
	return value, present && ok && value != ""
}

func connectorCoalescedOccurredAt(raw interface{}) (int64, error) {
	const maxExactlyRepresentableInteger = float64(1 << 53)
	switch value := raw.(type) {
	case int:
		return int64(value), nil
	case int32:
		return int64(value), nil
	case int64:
		return value, nil
	case uint:
		if uint64(value) > uint64(^uint64(0)>>1) {
			return 0, fmt.Errorf("must fit in int64")
		}
		return int64(value), nil
	case uint32:
		return int64(value), nil
	case uint64:
		if value > uint64(^uint64(0)>>1) {
			return 0, fmt.Errorf("must fit in int64")
		}
		return int64(value), nil
	case float32:
		return connectorCoalescedFloatOccurredAt(float64(value), maxExactlyRepresentableInteger)
	case float64:
		return connectorCoalescedFloatOccurredAt(value, maxExactlyRepresentableInteger)
	case json.Number:
		parsed, err := strconv.ParseInt(string(value), 10, 64)
		if err != nil {
			return 0, fmt.Errorf("must be an integer: %w", err)
		}
		return parsed, nil
	case string:
		parsed, err := strconv.ParseInt(strings.TrimSpace(value), 10, 64)
		if err != nil {
			return 0, fmt.Errorf("must be an integer: %w", err)
		}
		return parsed, nil
	default:
		return 0, fmt.Errorf("must be an integer timestamp, got %T", raw)
	}
}

func connectorCoalescedFloatOccurredAt(value, maxExactlyRepresentableInteger float64) (int64, error) {
	if math.IsNaN(value) || math.IsInf(value, 0) || math.Trunc(value) != value {
		return 0, fmt.Errorf("must be a finite integer")
	}
	if value < -maxExactlyRepresentableInteger || value > maxExactlyRepresentableInteger {
		return 0, fmt.Errorf("must be within the exact integer range")
	}
	return int64(value), nil
}

func mergeConnectorFieldMappingExtraUpdates(updates models.SObject, extraUpdates models.SObject) error {
	if len(extraUpdates) == 0 {
		return nil
	}
	for field, value := range extraUpdates {
		if _, exists := lookupRecordFieldValue(updates, field); exists {
			return fmt.Errorf("applyConnectorFieldMappings: extraUpdates field %s conflicts with a connector mapping target field", field)
		}
		updates[field] = value
	}
	return nil
}

func connectorConflictPolicy(raw string) string {
	policy := strings.TrimSpace(raw)
	if policy == "" {
		return connector.ConflictPolicyOverwrite
	}
	return policy
}

func connectorMappingsNeedAuthoritativeTargetRecord(mappings []connector.ConnectorFieldMappingInput) (bool, error) {
	needsTargetRecord := false
	for _, m := range mappings {
		switch connectorConflictPolicy(m.ConflictPolicy) {
		case connector.ConflictPolicyOverwrite:
		case connector.ConflictPolicySkipIfSet, connector.ConflictPolicyError:
			needsTargetRecord = true
		default:
			return false, fmt.Errorf("applyConnectorFieldMappings: ConflictPolicy %q on field %s is not a recognised value (expected overwrite, skip_if_set, error, or empty)", m.ConflictPolicy, m.TargetField)
		}
	}
	return needsTargetRecord, nil
}

func (e *FlowExecutor) loadConnectorConflictPreviewTargetRecord(ctx context.Context, targetObject, targetID string) (models.SObject, error) {
	if e.persistence == nil {
		return nil, fmt.Errorf("applyConnectorFieldMappings: ConflictPolicy requires authoritative target state but no persistence provider is wired")
	}
	record, err := e.persistence.GetRaw(ctx, targetObject, targetID)
	if err != nil {
		return nil, fmt.Errorf("applyConnectorFieldMappings: load authoritative target %s/%s for ConflictPolicy: %w", targetObject, targetID, err)
	}
	if record == nil {
		return models.SObject{}, nil
	}
	return record, nil
}

func lookupRecordFieldValue(data models.SObject, key string) (interface{}, bool) {
	if len(data) == 0 || key == "" {
		return nil, false
	}
	if v, ok := data[key]; ok {
		return v, true
	}
	for k, v := range data {
		if strings.EqualFold(k, key) {
			return v, true
		}
	}
	return nil, false
}