package records

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/nexuscrm/kernel/pkg/recordlocks"
	domainSchema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/fieldprovenance"
	"github.com/nexuscrm/shared/pkg/metadatarel"
	"github.com/nexuscrm/shared/pkg/models"
)

type rollupMutationOperation string

const (
	rollupMutationOperationInsert rollupMutationOperation = "insert"
	rollupMutationOperationDelete rollupMutationOperation = "delete"
)

func (ps *PersistenceService) rollupUpdatePessimisticBoundaryDecision(ctx context.Context, objectName string, updates models.SObject) (bool, string) {
	if ps == nil || ps.metadata == nil {
		return false, "metadata_unavailable"
	}
	schema := ps.metadata.GetSchema(ctx, objectName)
	if ps.recordLifecyclePhaseRequiresPessimisticTransaction(ctx, schema, recordLifecycleAfterUpdateInTx) {
		return true, "record_lifecycle_after_update"
	}
	if schemaHasExclusiveSiblingFields(schema) {
		return true, "exclusive_sibling_fields"
	}
	if schemaHasLockableLookupWrite(schema, updates) {
		return true, "lockable_lookup_write"
	}
	if schemaHasLockableLookupFields(schema) && ps.hasActiveRecordTriggerSubscription(ctx, objectName, "RecordBeforeSave", "Update") {
		return true, "lockable_lookup_before_save_update"
	}
	if ps.metadata.HasRollupChildDependencies(ctx, objectName) {
		return true, "rollup_child_dependencies"
	}
	return false, "none"
}

func (ps *PersistenceService) preparedUpdateBoundaryData(schema *models.ObjectMetadata, prepared *models.PreparedRecordUpdate, extraUpdates models.SObject) (models.SObject, error) {
	data := models.SObject{}
	if prepared != nil {
		for key, value := range prepared.EffectiveUpdates {
			data[key] = value
		}
		internalUpdates, err := ps.normalizePreparedInternalUpdates(schema, prepared.InternalUpdates)
		if err != nil {
			return nil, err
		}
		for key, value := range internalUpdates {
			data[key] = value
		}
	}
	for key, value := range extraUpdates {
		data[key] = value
	}
	return data, nil
}

func (ps *PersistenceService) collectRollupParentKeysForRecords(ctx context.Context, objectName string, records ...models.SObject) ([]recordlocks.ParentKey, error) {
	return ps.collectRollupParentKeysForRecordsWithMode(ctx, objectName, false, records...)
}

func (ps *PersistenceService) collectRollupInsertParentKeysForRecords(ctx context.Context, objectName string, records ...models.SObject) ([]recordlocks.ParentKey, error) {
	return ps.collectRollupParentKeysForRecordsWithMode(ctx, objectName, true, records...)
}

func (ps *PersistenceService) collectRollupParentKeysForRecordsWithMode(ctx context.Context, objectName string, insertMode bool, records ...models.SObject) ([]recordlocks.ParentKey, error) {
	if ps == nil || ps.rollup == nil || !ps.IsTransactionActive(ctx) || recordlocks.ScopeFromContext(ctx) == nil {
		return nil, nil
	}
	keysByID := map[string]recordlocks.ParentKey{}
	visited := map[string]struct{}{}
	for _, record := range records {
		if len(record) == 0 {
			continue
		}
		if err := ps.collectRollupParentKeysRecursive(ctx, objectName, record, insertMode, keysByID, visited); err != nil {
			return nil, err
		}
	}
	keys := make([]recordlocks.ParentKey, 0, len(keysByID))
	for _, key := range keysByID {
		keys = append(keys, key)
	}
	return recordlocks.OrderedParentKeys(keys), nil
}

func (ps *PersistenceService) collectRollupParentKeysRecursive(ctx context.Context, objectName string, record models.SObject, insertMode bool, keysByID map[string]recordlocks.ParentKey, visited map[string]struct{}) error {
	var affected []recordlocks.ParentKey
	var err error
	if insertMode {
		affected, err = ps.rollup.AffectedParentRowsForInsert(ctx, objectName, record)
	} else {
		affected, err = ps.rollup.AffectedParentRows(ctx, objectName, record)
	}
	if err != nil {
		return wrapRollupHookError("failed to discover affected rollup parents", err)
	}
	for _, key := range affected {
		normalized, ok := recordlocks.NormalizeParentKey(key)
		if !ok {
			continue
		}
		keyID := recordlocks.KeyID(normalized)
		keysByID[keyID] = normalized
		if _, seen := visited[keyID]; seen {
			continue
		}
		visited[keyID] = struct{}{}

		parentRecord, err := ps.cachedPhysicalFindOne(
			constants.WithSecurityContext(ctx, constants.SystemContext, "records rollup parent prelock discovery"),
			normalized.ObjectName,
			normalized.RecordID,
		)
		if err != nil {
			return wrapRollupHookError(
				fmt.Sprintf("failed to load rollup parent %s/%s for nested parent-lock discovery", normalized.ObjectName, normalized.RecordID),
				err,
			)
		}
		if len(parentRecord) == 0 {
			continue
		}
		if err := ps.collectDerivedRollupUpdateLookupParentKeys(ctx, normalized.ObjectName, parentRecord, keysByID); err != nil {
			return err
		}
		if err := ps.collectRollupParentKeysRecursive(ctx, normalized.ObjectName, parentRecord, false, keysByID, visited); err != nil {
			return err
		}
	}
	return nil
}

func (ps *PersistenceService) collectRollupUpdateParentKeysForRecords(ctx context.Context, objectName string, finalRecord, oldRecord models.SObject) ([]recordlocks.ParentKey, error) {
	if ps == nil || ps.rollup == nil || !ps.IsTransactionActive(ctx) || recordlocks.ScopeFromContext(ctx) == nil {
		return nil, nil
	}
	keysByID := map[string]recordlocks.ParentKey{}
	visited := map[string]struct{}{}
	if err := ps.collectRollupUpdateParentKeysRecursive(ctx, objectName, finalRecord, oldRecord, keysByID, visited); err != nil {
		return nil, err
	}
	keys := make([]recordlocks.ParentKey, 0, len(keysByID))
	for _, key := range keysByID {
		keys = append(keys, key)
	}
	return recordlocks.OrderedParentKeys(keys), nil
}

func (ps *PersistenceService) collectRollupUpdateParentKeysRecursive(ctx context.Context, objectName string, finalRecord, oldRecord models.SObject, keysByID map[string]recordlocks.ParentKey, visited map[string]struct{}) error {
	if len(finalRecord) == 0 && len(oldRecord) == 0 {
		return nil
	}
	affected, err := ps.rollup.AffectedParentRowsForUpdate(ctx, objectName, finalRecord, oldRecord)
	if err != nil {
		return wrapRollupHookError("failed to discover affected rollup update parents", err)
	}
	for _, key := range affected {
		normalized, ok := recordlocks.NormalizeParentKey(key)
		if !ok {
			continue
		}
		keyID := recordlocks.KeyID(normalized)
		keysByID[keyID] = normalized
		if _, seen := visited[keyID]; seen {
			continue
		}
		visited[keyID] = struct{}{}

		parentRecord, err := ps.cachedPhysicalFindOne(
			constants.WithSecurityContext(ctx, constants.SystemContext, "records rollup update parent prelock discovery"),
			normalized.ObjectName,
			normalized.RecordID,
		)
		if err != nil {
			return wrapRollupHookError(
				fmt.Sprintf("failed to load rollup parent %s/%s for nested parent-lock discovery", normalized.ObjectName, normalized.RecordID),
				err,
			)
		}
		if len(parentRecord) == 0 {
			continue
		}
		if err := ps.collectDerivedRollupUpdateLookupParentKeys(ctx, normalized.ObjectName, parentRecord, keysByID); err != nil {
			return err
		}
		if err := ps.collectRollupParentKeysRecursive(ctx, normalized.ObjectName, parentRecord, false, keysByID, visited); err != nil {
			return err
		}
	}
	return nil
}

// collectDerivedRollupUpdateLookupParentKeys completes the parent-lock plan for
// a record that Analytics will update as a rollup parent. Discovering the
// rollup parent itself is not sufficient: the derived update has the same
// lock-order obligations as any other update, including unchanged
// master-detail, controlled-by-parent, and other lock-required lookup parents.
//
// Planning these keys before the child row becomes transaction-owned avoids a
// late lower-sorting parent discovery and a full transaction replay. Plain
// optional lookups remain excluded by orderedParentKeysForLookupTargets.
func (ps *PersistenceService) collectDerivedRollupUpdateLookupParentKeys(ctx context.Context, objectName string, parentRecord models.SObject, keysByID map[string]recordlocks.ParentKey) error {
	if ps == nil || ps.metadata == nil || len(parentRecord) == 0 {
		return nil
	}
	schema := ps.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return nil
	}
	targets, err := ps.collectUpdateNonPolymorphicLookupParentTargets(ctx, schema, parentRecord, parentRecord)
	if err != nil {
		return wrapRollupHookError(
			fmt.Sprintf("failed to discover lookup parents required by derived rollup update of %s", objectName),
			err,
		)
	}
	keys, _ := orderedParentKeysForLookupTargets(targets)
	for _, key := range keys {
		normalized, ok := recordlocks.NormalizeParentKey(key)
		if !ok {
			continue
		}
		keysByID[recordlocks.KeyID(normalized)] = normalized
	}
	return nil
}

func (ps *PersistenceService) ensureRollupParentsPrelocked(ctx context.Context, objectName string, parentPrelocks map[string]struct{}, records ...models.SObject) error {
	if ps == nil || ps.rollup == nil || !ps.IsTransactionActive(ctx) || recordlocks.ScopeFromContext(ctx) == nil {
		return nil
	}
	var missing []recordlocks.ParentKey
	scope := recordlocks.ScopeFromContext(ctx)
	collectMissing := func(keys []recordlocks.ParentKey) {
		for _, key := range keys {
			if key.ObjectName == "" || key.RecordID == "" {
				continue
			}
			lockID := recordlocks.KeyID(key)
			if _, ok := parentPrelocks[lockID]; ok {
				continue
			}
			if scope != nil && scope.HasAcquired(key) {
				continue
			}
			missing = append(missing, key)
		}
	}
	if len(records) >= 2 {
		keys, err := ps.rollup.AffectedParentRowsForUpdate(ctx, objectName, records[1], records[0])
		if err != nil {
			return wrapRollupHookError("failed to verify affected rollup parent locks", err)
		}
		collectMissing(keys)
	} else {
		for _, record := range records {
			if len(record) == 0 {
				continue
			}
			keys, err := ps.rollup.AffectedParentRows(ctx, objectName, record)
			if err != nil {
				return wrapRollupHookError("failed to verify affected rollup parent locks", err)
			}
			collectMissing(keys)
		}
	}
	if len(missing) > 0 {
		return &parentLockPrelockRetryError{keys: missing, reason: "rollup parent"}
	}
	return nil
}

func (ps *PersistenceService) collectRollupUpdateParentKeys(ctx context.Context, objectName string, schema *models.ObjectMetadata, id string, updates models.SObject, currentUser *models.UserSession) ([]recordlocks.ParentKey, error) {
	if ps == nil || ps.rollup == nil || !ps.IsTransactionActive(ctx) || recordlocks.ScopeFromContext(ctx) == nil {
		return nil, nil
	}
	oldRecord, err := ps.loadAndValidateAccessUnlocked(ctx, objectName, id, schema, currentUser)
	if err != nil {
		return nil, err
	}
	effectiveUpdates, hasChanges, err := ps.filterEditableFields(ctx, objectName, schema, updates, oldRecord, currentUser)
	if err != nil {
		return nil, err
	}
	finalRecord := oldRecord
	if hasChanges {
		finalRecord = ps.mergeRecords(oldRecord, effectiveUpdates)
	}
	return ps.collectRollupUpdateParentKeysForRecords(ctx, objectName, finalRecord, oldRecord)
}

func (ps *PersistenceService) collectRollupPreparedUpdateParentKeys(ctx context.Context, objectName string, id string, schema *models.ObjectMetadata, prepared *models.PreparedRecordUpdate, extraUpdates models.SObject, currentUser *models.UserSession) ([]recordlocks.ParentKey, error) {
	if ps == nil || ps.rollup == nil || !ps.IsTransactionActive(ctx) || recordlocks.ScopeFromContext(ctx) == nil {
		return nil, nil
	}
	oldRecord, err := ps.loadAndValidateAccessUnlocked(ctx, objectName, id, schema, currentUser)
	if err != nil {
		return nil, err
	}

	effectiveUpdates := cloneSObject(prepared.EffectiveUpdates)
	if effectiveUpdates == nil {
		effectiveUpdates = models.SObject{}
	}
	recordToValidate := cloneSObject(prepared.Record)
	if recordToValidate == nil {
		recordToValidate = ps.mergeRecords(oldRecord, effectiveUpdates)
	}

	filteredExtra, err := ps.preparePreparedFinalExtraUpdates(ctx, objectName, id, schema, extraUpdates, oldRecord, currentUser)
	if err != nil {
		return nil, err
	}
	for k, v := range filteredExtra {
		effectiveUpdates[k] = v
		recordToValidate[k] = v
	}

	internalUpdates, err := ps.normalizePreparedInternalUpdates(schema, prepared.InternalUpdates)
	if err != nil {
		return nil, err
	}
	for k, v := range internalUpdates {
		effectiveUpdates[k] = v
		recordToValidate[k] = v
	}
	return ps.collectRollupUpdateParentKeysForRecords(ctx, objectName, recordToValidate, oldRecord)
}

func rollupMutationRequiresPessimisticBoundaryForSchemas(allSchemas []*models.ObjectMetadata, objectName string, op rollupMutationOperation) bool {
	if op == "" {
		return false
	}
	if domainSchema.HasRollupChildDependencies(allSchemas, objectName) {
		return true
	}
	if schemaHasExclusiveSiblingFields(schemaByAPIName(allSchemas, objectName)) {
		return true
	}
	if op != rollupMutationOperationInsert {
		return false
	}

	schemaKey := strings.ToLower(strings.TrimSpace(objectName))
	if schemaKey == "" {
		return false
	}
	for _, schema := range allSchemas {
		if schema == nil || strings.ToLower(strings.TrimSpace(schema.APIName)) != schemaKey {
			continue
		}
		if schemaHasSiblingMaxDefaults(schema) || schemaHasExclusiveSiblingFields(schema) {
			return true
		}
		for _, row := range schema.DefaultChildRows {
			childSchema := schemaByAPIName(allSchemas, row.TargetObject)
			if domainSchema.HasRollupChildDependencies(allSchemas, row.TargetObject) || schemaHasSiblingMaxDefaults(childSchema) || schemaHasExclusiveSiblingFields(childSchema) {
				return true
			}
		}
		return false
	}
	return false
}

func (ps *PersistenceService) rollupInsertRequiresPessimisticBoundary(ctx context.Context, objectName string, data models.SObject) bool {
	requires, _ := ps.rollupInsertPessimisticBoundaryDecision(ctx, objectName, data)
	return requires
}

func (ps *PersistenceService) rollupInsertPessimisticBoundaryDecision(ctx context.Context, objectName string, data models.SObject) (bool, string) {
	if ps == nil || ps.metadata == nil {
		return false, "metadata_unavailable"
	}
	return ps.rollupInsertPessimisticBoundaryDecisionRecursive(ctx, objectName, data, map[string]struct{}{})
}

func (ps *PersistenceService) rollupInsertRequiresPessimisticBoundaryRecursive(
	ctx context.Context,
	objectName string,
	data models.SObject,
	visited map[string]struct{},
) bool {
	requires, _ := ps.rollupInsertPessimisticBoundaryDecisionRecursive(ctx, objectName, data, visited)
	return requires
}

func (ps *PersistenceService) rollupInsertPessimisticBoundaryDecisionRecursive(
	ctx context.Context,
	objectName string,
	data models.SObject,
	visited map[string]struct{},
) (bool, string) {
	objectKey := strings.ToLower(strings.TrimSpace(objectName))
	if objectKey == "" {
		return false, "object_missing"
	}
	if ps.metadata.HasRollupChildDependencies(ctx, objectName) {
		return true, "rollup_child_dependencies"
	}
	schema := ps.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return false, "schema_missing"
	}
	if ps.recordLifecyclePhaseRequiresPessimisticTransaction(ctx, schema, recordLifecycleAfterCreateInTx) {
		return true, "record_lifecycle_after_create"
	}
	if schemaHasExclusiveSiblingFields(schema) {
		return true, "exclusive_sibling_fields"
	}
	if schemaHasLockableLookupWrite(schema, data) {
		return true, "lockable_lookup_write"
	}
	if schemaHasLockableLookupFields(schema) && ps.hasActiveRecordTriggerSubscription(ctx, objectName, "RecordBeforeSave", "Create") {
		return true, "lockable_lookup_before_save_create"
	}
	if schemaHasSiblingMaxDefaults(schema) {
		return true, "sibling_max_defaults"
	}
	if _, seen := visited[objectKey]; seen {
		return false, "visited"
	}
	visited[objectKey] = struct{}{}
	defer delete(visited, objectKey)

	if len(schema.DefaultChildRows) > 0 {
		contracts, err := ResolveDefaultChildRowsContractsContext(ctx, ps.metadata, schema, schema.DefaultChildRows, ps.engineOwnedFieldSet)
		if err != nil {
			return true, "default_child_rows_contract_error"
		}
		for _, contract := range contracts {
			child := make(models.SObject, len(contract.InsertValues)+1)
			for key, value := range contract.InsertValues {
				child[key] = value
			}
			child[contract.TargetLookupField] = "__boundary_probe_parent__"
			if requires, reason := ps.rollupInsertPessimisticBoundaryDecisionRecursive(ctx, contract.TargetObject, child, visited); requires {
				return true, "default_child_row_" + reason
			}
		}
	}

	if ps.rollupInsertHasBeforeCreateSiblingMaxMutationRisk(ctx, schema) {
		return true, "before_create_sibling_max_mutation_risk"
	}
	if len(schema.ChildCloneRules) > 0 && ps.rollupInsertHasBeforeCreateCloneMutationRisk(ctx, schema, visited) {
		return true, "before_create_clone_mutation_risk"
	}
	if len(schema.ChildCloneRules) > 0 && ps.rollupInsertHasMetadataOwnedCloneMutationRisk(ctx, schema, data, visited) {
		return true, "metadata_owned_clone_mutation_risk"
	}
	for _, rule := range schema.ChildCloneRules {
		if ps.childCloneRuleWillInsertRollupChild(ctx, rule, data, visited) {
			return true, "child_clone_will_insert_rollup_child"
		}
	}

	return false, "none"
}

func schemaHasLockableLookupWrite(schema *models.ObjectMetadata, data models.SObject) bool {
	if schema == nil || len(data) == 0 || !schemaHasLockableLookupFields(schema) {
		return false
	}
	for i := range schema.Fields {
		field := &schema.Fields[i]
		if !isLockableLookupField(field) {
			continue
		}
		rawVal, present := lookupRecordValueCaseInsensitive(data, field.APIName)
		if !present || rawVal == nil {
			continue
		}
		if str, ok := rawVal.(string); ok && strings.TrimSpace(str) == "" {
			continue
		}
		return true
	}
	return false
}

func schemaHasLockableLookupFields(schema *models.ObjectMetadata) bool {
	if schema == nil || isSystemAutomationSchema(schema) {
		return false
	}
	for i := range schema.Fields {
		if isLockableLookupField(&schema.Fields[i]) {
			return true
		}
	}
	return false
}

func isLockableLookupField(field *models.FieldMetadata) bool {
	if !isNonPolymorphicReference(field) {
		return false
	}
	return !isValidatorSkippedField(field.APIName)
}

func isSystemAutomationSchema(schema *models.ObjectMetadata) bool {
	return schema != nil && strings.EqualFold(string(schema.TableType), "system_automation")
}

func schemaByAPIName(allSchemas []*models.ObjectMetadata, objectName string) *models.ObjectMetadata {
	objectKey := strings.ToLower(strings.TrimSpace(objectName))
	if objectKey == "" {
		return nil
	}
	for _, schema := range allSchemas {
		if schema != nil && strings.ToLower(strings.TrimSpace(schema.APIName)) == objectKey {
			return schema
		}
	}
	return nil
}

func (ps *PersistenceService) childCloneRuleWillInsertRollupChild(
	ctx context.Context,
	rule models.ChildCloneRule,
	data models.SObject,
	visited map[string]struct{},
) bool {
	if rule.ParentLookupField == "" || rule.SourceObject == "" || rule.SourceLookupField == "" || rule.TargetObject == "" {
		return false
	}
	sourceParentValue, ok := lookupRecordValueCaseInsensitive(data, rule.ParentLookupField)
	if !ok {
		return false
	}
	sourceParentID := inheritLookupString(sourceParentValue)
	if sourceParentID == "" {
		return false
	}
	targetSchema := ps.metadata.GetSchema(ctx, rule.TargetObject)
	child := buildBoundaryProbeChildCloneTarget(rule, targetSchema)
	return ps.rollupInsertRequiresPessimisticBoundaryRecursive(ctx, rule.TargetObject, child, visited)
}

func (ps *PersistenceService) rollupInsertHasBeforeCreateCloneMutationRisk(
	ctx context.Context,
	schema *models.ObjectMetadata,
	visited map[string]struct{},
) bool {
	if schema == nil || len(schema.ChildCloneRules) == 0 || !ps.hasActiveRecordTriggerSubscription(ctx, schema.APIName, "RecordBeforeSave", "Create") {
		return false
	}
	for _, rule := range schema.ChildCloneRules {
		if ps.rollupInsertRequiresPessimisticBoundaryRecursive(ctx, rule.TargetObject, nil, visited) {
			return true
		}
	}
	return false
}

func (ps *PersistenceService) rollupInsertHasBeforeCreateSiblingMaxMutationRisk(ctx context.Context, schema *models.ObjectMetadata) bool {
	if schema == nil || !ps.metadata.HasAnySiblingMaxDefaults(ctx) {
		return false
	}
	return ps.hasActiveRecordTriggerSubscription(ctx, schema.APIName, "RecordBeforeSave", "Create")
}

func (ps *PersistenceService) rollupInsertHasMetadataOwnedCloneMutationRisk(
	ctx context.Context,
	schema *models.ObjectMetadata,
	data models.SObject,
	visited map[string]struct{},
) bool {
	if schema == nil || len(schema.ChildCloneRules) == 0 {
		return false
	}
	for _, rule := range schema.ChildCloneRules {
		if !ps.childCloneRuleTargetCanReachRollupChild(ctx, rule.TargetObject, visited) {
			continue
		}
		if ps.childCloneParentLookupHasMetadataOwnedMutationRisk(ctx, schema, rule.ParentLookupField, data) {
			return true
		}
	}
	return false
}

func (ps *PersistenceService) childCloneParentLookupHasMetadataOwnedMutationRisk(
	ctx context.Context,
	schema *models.ObjectMetadata,
	fieldAPIName string,
	data models.SObject,
) bool {
	if schema == nil || strings.TrimSpace(fieldAPIName) == "" {
		return false
	}
	if currentValue, ok := lookupRecordValueCaseInsensitive(data, fieldAPIName); ok && !isBlankValue(currentValue) {
		return false
	}
	if ps.childCloneParentLookupHasFieldDefault(schema, fieldAPIName) {
		return true
	}
	if ps.childCloneParentLookupHasInsertInheritance(ctx, schema.APIName, fieldAPIName) {
		return true
	}
	if ps.childCloneParentLookupIsEngineOwned(ctx, schema.APIName, fieldAPIName) {
		return true
	}
	if childCloneParentLookupIsCompoundDerived(schema, fieldAPIName) {
		return true
	}
	return false
}

func (ps *PersistenceService) childCloneRuleTargetCanReachRollupChild(
	ctx context.Context,
	targetObject string,
	visited map[string]struct{},
) bool {
	return ps.rollupInsertRequiresPessimisticBoundaryRecursive(ctx, targetObject, nil, visited)
}

func (ps *PersistenceService) childCloneParentLookupHasFieldDefault(schema *models.ObjectMetadata, fieldAPIName string) bool {
	if schema == nil || fieldAPIName == "" {
		return false
	}
	for i := range schema.Fields {
		if !strings.EqualFold(schema.Fields[i].APIName, fieldAPIName) {
			continue
		}
		return schema.Fields[i].DefaultValue != nil && strings.TrimSpace(*schema.Fields[i].DefaultValue) != ""
	}
	return false
}

func (ps *PersistenceService) childCloneParentLookupHasInsertInheritance(ctx context.Context, objectAPIName, fieldAPIName string) bool {
	if ps == nil || objectAPIName == "" || fieldAPIName == "" {
		return false
	}
	rules, err := ps.cachedInsertInheritanceRules(ctx, objectAPIName)
	if err != nil {
		// This probe selects the safer transaction boundary. Unknown metadata
		// must never downgrade a mutation to the plain transaction path.
		return true
	}
	for _, rule := range rules {
		if strings.EqualFold(rule.TargetField, fieldAPIName) {
			return true
		}
	}
	return false
}

func buildBoundaryProbeChildCloneTarget(rule models.ChildCloneRule, targetSchema *models.ObjectMetadata) models.SObject {
	if rule.TargetLookupField == "" {
		return nil
	}
	child := models.SObject{
		rule.TargetLookupField: "__boundary_probe_parent__",
	}
	targetFields := indexSchemaFields(targetSchema)
	if rule.SourceIdField != "" && canCloneTargetField(rule.SourceIdField, rule.TargetLookupField, targetFields, nil) {
		child[rule.SourceIdField] = "__boundary_probe_source__"
	}
	for targetField := range rule.FieldMappings {
		if !canCloneTargetField(targetField, rule.TargetLookupField, targetFields, nil) {
			continue
		}
		child[targetField] = "__boundary_probe_source__"
	}
	return child
}

func (ps *PersistenceService) childCloneParentLookupIsEngineOwned(ctx context.Context, objectAPIName, fieldAPIName string) bool {
	engineOwned, err := ps.engineOwnedFieldSet(ctx, objectAPIName)
	if err != nil {
		// This probe decides whether the operation needs the safer rollup-aware
		// transaction. Unknown metadata must select the conservative path.
		return true
	}
	if len(engineOwned) == 0 {
		return false
	}
	return engineOwned[strings.ToLower(strings.TrimSpace(fieldAPIName))]
}

func childCloneParentLookupIsCompoundDerived(schema *models.ObjectMetadata, fieldAPIName string) bool {
	if schema == nil || strings.TrimSpace(fieldAPIName) == "" {
		return false
	}
	if field := models.FindField(schema, fieldAPIName); field != nil && field.IsCompound && len(field.ComposedFrom) > 0 {
		return true
	}
	isCompound, composedFrom := fieldprovenance.ClassifyCompound(schema.APIName, fieldAPIName)
	return isCompound && len(composedFrom) > 0
}

func lookupRecordValueCaseInsensitive(data recordValues, fieldAPIName string) (interface{}, bool) {
	if len(data) == 0 || strings.TrimSpace(fieldAPIName) == "" {
		return nil, false
	}
	if value, ok := data[fieldAPIName]; ok {
		return value, true
	}
	for key, value := range data {
		if strings.EqualFold(key, fieldAPIName) {
			return value, true
		}
	}
	return nil, false
}

func (ps *PersistenceService) deleteTransactionObjectNames(ctx context.Context, objectName string) []string {
	names := []string{}
	seen := map[string]struct{}{}
	addName := func(name string) {
		key := strings.ToLower(strings.TrimSpace(name))
		if key == "" {
			return
		}
		if _, ok := seen[key]; ok {
			return
		}
		seen[key] = struct{}{}
		names = append(names, name)
	}

	addName(objectName)
	if ps == nil || ps.metadata == nil {
		return names
	}

	walked := map[string]struct{}{}
	var walk func(parentObjectName string)
	walk = func(parentObjectName string) {
		parentKey := strings.ToLower(strings.TrimSpace(parentObjectName))
		if parentKey == "" {
			return
		}
		if _, ok := walked[parentKey]; ok {
			return
		}
		walked[parentKey] = struct{}{}

		for _, childSchema := range ps.metadata.GetChildRelationships(ctx, parentObjectName) {
			if childSchema == nil {
				continue
			}
			relationshipsByField := indexRelationshipsByParentAndField(ps.metadata.GetRelationships(ctx, childSchema.APIName))
			for _, field := range childSchema.Fields {
				if !metadatarel.FieldReferencesParent(field, parentObjectName) {
					continue
				}
				deleteSemantics := metadatarel.ResolveDeleteSemantics(field, relationshipsByField[parentKey+"|"+strings.ToLower(field.APIName)])
				if deleteSemantics.Cascade {
					addName(childSchema.APIName)
					walk(childSchema.APIName)
					continue
				}
				if strings.EqualFold(string(deleteSemantics.Rule), string(constants.DeleteRuleSetNull)) {
					addName(childSchema.APIName)
				}
			}
		}
	}
	walk(objectName)
	return names
}

func (ps *PersistenceService) hasActiveRecordTriggerSubscription(ctx context.Context, objectName, eventType, recordTriggerType string) bool {
	if ps == nil || ps.metadata == nil {
		return false
	}
	schema := ps.metadata.GetSchema(ctx, objectName)
	if schema == nil || schema.KeyPrefix == nil || strings.TrimSpace(*schema.KeyPrefix) == "" {
		return false
	}
	rows, err := ps.metadata.ListActiveTriggerSubscriptions(ctx, *schema.KeyPrefix, eventType, recordTriggerType)
	if err != nil {
		return true
	}
	return len(rows) > 0
}

// runRollupAwareTransaction selects the records-layer mutation transaction
// boundary for a direct CRUD entrypoint.
//
// D2b transition contract:
//   - root direct writes that can trigger parent rollup recompute enter the
//     pessimistic path immediately
//   - inherited plain transactions are no longer treated as a generic safe
//     rollback boundary for rollup-child writes; those callers must migrate to
//     a pessimistic outer transaction before D2c lands the runtime lockParent
//     fail-close wiring
func (ps *PersistenceService) runRollupAwareTransaction(ctx context.Context, objectName string, data models.SObject, fn func(context.Context) error) error {
	withCoordinator := func(txCtx context.Context) error {
		stageStart := time.Now()
		err := ps.runParentLockCoordinatorIfActive(txCtx, fn)
		logRecordStageTiming(txCtx, "Insert", "runRollupAwareTransaction.coordinator", objectName, "", stageStart, err, len(data))
		return err
	}
	decisionStart := time.Now()
	requiresPessimistic, reason := ps.rollupInsertPessimisticBoundaryDecision(ctx, objectName, data)
	if ps.permissionStateRequiresPessimisticTransaction(objectName) {
		requiresPessimistic = true
		reason = "permission_state_authority"
	}
	if !requiresPessimistic {
		logRecordTxBoundaryDecision(ctx, "Insert", objectName, "plain", reason, decisionStart, len(data))
		logRecordStageTiming(ctx, "Insert", "runRollupAwareTransaction.decide.plain", objectName, "", decisionStart, nil, len(data))
		stageStart := time.Now()
		err := ps.runGroupMembershipAwareTransaction(ctx, objectName, withCoordinator)
		logRecordStageTiming(ctx, "Insert", "runRollupAwareTransaction.groupMembership.plain", objectName, "", stageStart, err, len(data))
		return err
	}
	logRecordTxBoundaryDecision(ctx, "Insert", objectName, "pessimistic", reason, decisionStart, len(data))
	logRecordStageTiming(ctx, "Insert", "runRollupAwareTransaction.decide.pessimistic", objectName, "", decisionStart, nil, len(data))
	stageStart := time.Now()
	err := ps.runGroupMembershipAwarePessimisticTransaction(ctx, objectName, withCoordinator)
	logRecordStageTiming(ctx, "Insert", "runRollupAwareTransaction.groupMembership.pessimistic", objectName, "", stageStart, err, len(data))
	return err
}

func (ps *PersistenceService) runRollupAwareUpdateTransaction(ctx context.Context, objectName string, updates models.SObject, fn func(context.Context) error) error {
	withCoordinator := func(txCtx context.Context) error {
		stageStart := time.Now()
		err := ps.runParentLockCoordinatorIfActive(txCtx, fn)
		logRecordStageTiming(txCtx, "Update", "runRollupAwareUpdateTransaction.coordinator", objectName, "", stageStart, err, len(updates))
		return err
	}
	decisionStart := time.Now()
	requiresPessimistic, reason := ps.rollupUpdatePessimisticBoundaryDecision(ctx, objectName, updates)
	if ps.permissionStateRequiresPessimisticTransaction(objectName) {
		requiresPessimistic = true
		reason = "permission_state_authority"
	}
	if !requiresPessimistic {
		logRecordTxBoundaryDecision(ctx, "Update", objectName, "plain", reason, decisionStart, len(updates))
		logRecordStageTiming(ctx, "Update", "runRollupAwareUpdateTransaction.decide.plain", objectName, "", decisionStart, nil, len(updates))
		stageStart := time.Now()
		err := ps.runGroupMembershipAwareTransaction(ctx, objectName, withCoordinator)
		logRecordStageTiming(ctx, "Update", "runRollupAwareUpdateTransaction.groupMembership.plain", objectName, "", stageStart, err, len(updates))
		return err
	}
	logRecordTxBoundaryDecision(ctx, "Update", objectName, "pessimistic", reason, decisionStart, len(updates))
	logRecordStageTiming(ctx, "Update", "runRollupAwareUpdateTransaction.decide.pessimistic", objectName, "", decisionStart, nil, len(updates))
	stageStart := time.Now()
	err := ps.runGroupMembershipAwarePessimisticTransaction(ctx, objectName, withCoordinator)
	logRecordStageTiming(ctx, "Update", "runRollupAwareUpdateTransaction.groupMembership.pessimistic", objectName, "", stageStart, err, len(updates))
	return err
}

func (ps *PersistenceService) runRollupAwareDeleteTransaction(ctx context.Context, objectName, recordID string, fn func(context.Context) error) error {
	objectNames := ps.deleteTransactionObjectNames(ctx, objectName)
	withCoordinator := func(txCtx context.Context) error {
		stageStart := time.Now()
		err := ps.runParentLockCoordinatorIfActive(txCtx, fn)
		logRecordStageTiming(txCtx, "Delete", "runRollupAwareDeleteTransaction.coordinator", objectName, recordID, stageStart, err, 0)
		return err
	}
	logRecordStageTiming(ctx, "Delete", "runRollupAwareDeleteTransaction.decide.pessimistic", objectName, recordID, time.Now(), nil, 0)
	stageStart := time.Now()
	err := ps.runGroupMembershipAwarePessimisticTransactionForObjects(ctx, objectNames, withCoordinator)
	logRecordStageTiming(ctx, "Delete", "runRollupAwareDeleteTransaction.groupMembership.pessimistic", objectName, recordID, stageStart, err, 0)
	return err
}

func (ps *PersistenceService) runParentLockCoordinatorIfActive(ctx context.Context, fn func(context.Context) error) error {
	if ps == nil || ps.txManager == nil || !ps.txManager.IsTransactionActive(ctx) || !ps.txManager.IsPessimisticTx(ctx) {
		return fn(ctx)
	}
	return ps.RunWithParentLockCoordinator(ctx, fn)
}

// runRollupAwareTransactionIfNeeded is the narrower wrapper for entrypoints
// like UpsertByExternalID that historically did not own an outer transaction at
// all. Non-rollup objects keep the existing direct execution path; rollup-
// reachable objects gain the authoritative D2a boundary.
func (ps *PersistenceService) runRollupAwareTransactionIfNeeded(ctx context.Context, objectName string, data models.SObject, fn func(context.Context) error) error {
	if !ps.rollupInsertRequiresPessimisticBoundary(ctx, objectName, data) {
		return fn(ctx)
	}
	return ps.runRollupAwareTransaction(ctx, objectName, data, fn)
}