package records

import (
	"context"
	"fmt"
	"strings"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	domainSchema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/metadatarel"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/referenceidentity"
)

type nonPolymorphicLookupTarget struct {
	field           models.FieldMetadata
	targetObject    string
	idVal           string
	accessOperation string
	// requiresLock distinguishes FK values that must be FOR UPDATE-locked on the
	// referenced parent (master-detail, ControlledByParent sharing-parent, or a
	// lookup that feeds a parent rollup) from plain optional lookups that only
	// need existence/FLS validation. Salesforce locks the former on child write
	// but never a plain lookup; locking every FK serialized unrelated writes on
	// shared low-cardinality parents (e.g. Pricebook2). Validation still covers
	// every target — only lock acquisition filters on this flag.
	requiresLock bool
}

type skipCoordinatedLookupParentLockKey struct{}
type nonPolymorphicLookupCreateIntentKey struct{}

type nonPolymorphicLookupValidationProofKey struct {
	targetObject    string
	idVal           string
	accessOperation string
	userID          string
	profileID       string
	roleID          string
	securityContext constants.SecurityContext
	scopedAccess    bool
	superUserAccess bool
	privileged      bool
}

func withoutCoordinatedLookupParentLock(ctx context.Context) context.Context {
	return context.WithValue(ctx, skipCoordinatedLookupParentLockKey{}, true)
}

func withNonPolymorphicLookupCreateIntent(ctx context.Context) context.Context {
	return context.WithValue(ctx, nonPolymorphicLookupCreateIntentKey{}, true)
}

func nonPolymorphicLookupValidationProofKeyFor(
	ctx context.Context,
	target nonPolymorphicLookupTarget,
	operation string,
	currentUser *models.UserSession,
	scopedAccess bool,
	superUserAccess bool,
	privileged bool,
) nonPolymorphicLookupValidationProofKey {
	if operation == "" {
		operation = constants.PermRead
	}
	userID, profileID, roleID := "", "", ""
	if currentUser != nil {
		userID = currentUser.ID
		profileID = currentUser.ProfileID
		if currentUser.RoleID != nil {
			roleID = *currentUser.RoleID
		}
	}
	return nonPolymorphicLookupValidationProofKey{
		targetObject:    strings.ToLower(strings.TrimSpace(target.targetObject)),
		idVal:           strings.TrimSpace(target.idVal),
		accessOperation: strings.ToLower(strings.TrimSpace(operation)),
		userID:          userID,
		profileID:       profileID,
		roleID:          roleID,
		securityContext: constants.GetSecurityContext(ctx),
		scopedAccess:    scopedAccess,
		superUserAccess: superUserAccess,
		privileged:      privileged,
	}
}

func nonPolymorphicLookupTargetValidationError(target nonPolymorphicLookupTarget) error {
	return appErrors.NewValidationError("", fmt.Sprintf(
		"%s: referenced %s record %q is not accessible or does not exist",
		target.field.APIName, target.targetObject, target.idVal,
	))
}

// validateNonPolymorphicLookups is the named owner for "does the caller
// supplying a non-polymorphic Reference FK value have the required access to
// the referenced record?" — round-4 reviewer requirement. The baseline
// requirement is Read. For the effective sharing-parent FK on a
// ControlledByParent child, concrete insert/re-parent/FK writes require parent
// Edit. Optional-edge owner/share fallback remains a record-reach behavior for
// no-parent/orphan records; it is not the authorization rule for changing the
// sharing-parent pointer.
//
// Pair to validatePolymorphicLookups (which covers polymorphic refs via
// system-level existence probes). This sibling uses user-level
// ps.Retrieve so BOTH "record missing" AND "user cannot see the record
// under RLS" reject with a neutral validation error. The Retrieve path
// intentionally conflates those two outcomes into a not-found-style error to
// prevent ID-enumeration (persistence_bulk_ops.go:640-641); this
// validator preserves that security posture by treating either as
// equally invalid.
//
// Runs BEFORE applyInsertInheritance. Once the validator has confirmed
// the parent is accessible, inheritance is a pure opportunistic copier
// — no need to double-check access at inheritance time. This is the
// Class-2 cutover-closure shape the reviewer repeatedly asked for:
// inheritance stops acting as a half-validator, a single named owner
// exists for non-polymorphic FK access, and downstream behaviour is
// deterministic.
//
// Scope:
//   - Iterates schema.Fields
//   - Considers Reference / MasterDetail fields that declare exactly one
//     referenceTo target, including transitional single-target polymorphic
//     fields. Multi-target polymorphic references are skipped; schema-level
//     validators own malformed shapes.
//   - Skips fields absent from data, nil, or carrying an empty string
//
// Error shape:
//   - not-found / forbidden / permission / any plain error from Retrieve
//     → ValidationError(…field name…) → the boundary adapter's operation
//     fallback code
//   - appErrors.IsInternal(err) → bubble unchanged (DB outages surface
//     as 500; the boundary projector keeps IsInternal first)
func (ps *PersistenceService) validateNonPolymorphicLookups(
	ctx context.Context,
	data models.SObject,
	schema *models.ObjectMetadata,
	currentUser *models.UserSession,
) error {
	if schema == nil {
		return nil
	}
	if err := ps.validateParentControlledCreateSharingParentPresence(ctx, data, schema); err != nil {
		return err
	}
	if len(data) == 0 {
		return nil
	}

	targets, err := ps.collectNonPolymorphicLookupTargets(ctx, data, schema)
	if err != nil {
		return err
	}
	idsByObject := make(map[string][]string)
	for _, target := range targets {
		idsByObject[target.targetObject] = append(idsByObject[target.targetObject], target.idVal)
	}
	identityDispositions, err := resolveReferenceIdentityDispositions(ctx, idsByObject, []string{constants.FieldID})
	if err != nil {
		return err
	}
	for _, target := range targets {
		if byID := identityDispositions[target.targetObject]; byID != nil {
			if byID[target.idVal] == referenceidentity.OwnedMissing {
				return nonPolymorphicLookupTargetValidationError(target)
			}
		}
	}
	if err := ps.lockLookupParentsIfCoordinated(ctx, targets); err != nil {
		return err
	}

	validated := make(map[nonPolymorphicLookupValidationProofKey]struct{}, len(targets))
	for _, target := range targets {
		if identityDispositions[target.targetObject][target.idVal] == referenceidentity.Immutable {
			continue
		}
		// Scheduled/system automation runs in SystemContext: CRUD/FLS/RLS are
		// bypassed, but blocked-object boundaries still apply. The generic
		// Retrieve path intentionally conflates "missing" and "not visible"
		// through record-access checks for ordinary user reads; using it here
		// would incorrectly reject a valid FK under privileged runtime contexts
		// such as scheduled FlowInterview creation. In the privileged branch we
		// therefore keep prepareOperation's hard security boundary, then probe
		// raw existence directly.
		operation := target.accessOperation
		if operation == "" {
			operation = constants.PermRead
		}
		scopedAccess := constants.IsScopedInternalAccess(ctx, target.targetObject, operation)
		superUserAccess := ps.permissions != nil && ps.permissions.HasSuperUserCapability(ctx, currentUser)
		privilegedAccess := constants.GetSecurityContext(ctx) >= constants.SystemContext || scopedAccess || superUserAccess
		proofKey := nonPolymorphicLookupValidationProofKeyFor(ctx, target, operation, currentUser, scopedAccess, superUserAccess, privilegedAccess)
		if _, ok := validated[proofKey]; ok {
			continue
		}
		if privilegedAccess {
			if _, err := ps.prepareOperation(ctx, target.targetObject, operation, currentUser); err != nil {
				if appErrors.IsInternal(err) {
					return err
				}
				return nonPolymorphicLookupTargetValidationError(target)
			}
			record, err := ps.cachedFindOne(ctx, target.targetObject, target.idVal)
			if err != nil {
				return appErrors.NewInternalError("failed to validate lookup target", err)
			}
			if record != nil {
				validated[proofKey] = struct{}{}
				continue
			}
			return nonPolymorphicLookupTargetValidationError(target)
		}

		// User-level read access + existence check in one call. Retrieve's
		// permission gate catches Object-level denial; its RLS gate at
		// persistence_bulk_ops.go:640 catches record-level denial; its
		// FindOne gate catches missing rows.
		_, err := ps.Retrieve(ctx, target.targetObject, target.idVal, currentUser)
		if err != nil {
			// Ordering load-bearing: IsInternal first. See the same rule in
			// the boundary error projector and the inheritance classifier in
			// insert_inheritance.go.
			if appErrors.IsInternal(err) {
				return err
			}

			// Everything else becomes a clean neutral 400. The v63 adapter maps it
			// to the operation fallback payload (INSERT_FAILED / UPDATE_FAILED).
			return nonPolymorphicLookupTargetValidationError(target)
		}
		if operation == constants.PermEdit {
			if ps.permissions == nil || ps.metadata == nil {
				return nonPolymorphicLookupTargetValidationError(target)
			}
			targetSchema := ps.metadata.GetSchema(ctx, target.targetObject)
			if targetSchema == nil {
				return nonPolymorphicLookupTargetValidationError(target)
			}
			record, err := ps.cachedFindOne(ctx, target.targetObject, target.idVal)
			if err != nil {
				return appErrors.NewInternalError("failed to validate lookup target", err)
			}
			if record == nil ||
				!ps.permissions.CheckObjectPermissionOnRecord(ctx, targetSchema, record, constants.PermEdit, currentUser) ||
				!ps.permissions.CheckRecordAccess(ctx, targetSchema, record, constants.PermEdit, currentUser) {
				return nonPolymorphicLookupTargetValidationError(target)
			}
		}
		validated[proofKey] = struct{}{}
	}

	return nil
}

func (ps *PersistenceService) collectNonPolymorphicLookupTargets(ctx context.Context, data models.SObject, schema *models.ObjectMetadata) ([]nonPolymorphicLookupTarget, error) {
	return ps.collectNonPolymorphicLookupTargetsWithWriteValidation(ctx, data, schema, true)
}

// collectNonPolymorphicLookupTargetsForLocking returns relationship targets for
// a lock plan only. A delete does not write the reference value, so it must not
// apply WritableReferenceTo's user-context write restriction to an existing
// record while it determines which parents require serialization.
func (ps *PersistenceService) collectNonPolymorphicLookupTargetsForLocking(ctx context.Context, data models.SObject, schema *models.ObjectMetadata) ([]nonPolymorphicLookupTarget, error) {
	return ps.collectNonPolymorphicLookupTargetsWithWriteValidation(ctx, data, schema, false)
}

func (ps *PersistenceService) collectNonPolymorphicLookupTargetsWithWriteValidation(ctx context.Context, data models.SObject, schema *models.ObjectMetadata, validateWritableReferenceTarget bool) ([]nonPolymorphicLookupTarget, error) {
	if schema == nil || len(data) == 0 {
		return nil, nil
	}
	var targets []nonPolymorphicLookupTarget
	for i := range schema.Fields {
		field := &schema.Fields[i]
		if !isNonPolymorphicReference(field) {
			continue
		}
		if isValidatorSkippedField(field.APIName) {
			continue
		}
		rawVal, present := data[field.APIName]
		if !present || rawVal == nil {
			continue
		}
		idVal, ok := rawVal.(string)
		if !ok {
			continue
		}
		idVal = strings.TrimSpace(idVal)
		if idVal == "" {
			continue
		}

		targetObject := field.ReferenceTo[0]
		if validateWritableReferenceTarget && !isWritableReferenceTarget(ctx, field, targetObject) {
			return nil, appErrors.NewValidationError("", fmt.Sprintf(
				"%s: referenced %s records are not writable through this field",
				field.APIName, targetObject,
			))
		}

		// Preview short-circuit: under a no-DML insert preview preparation,
		// accept synthetic preview IDs that the request-local registry knows
		// about for THIS target object. Non-preview contexts and unregistered
		// literals fall through to the normal access/existence gate.
		if constants.HasInsertPreviewPreparation(ctx) {
			if _, ok := constants.PreviewSyntheticRecordForObject(ctx, targetObject, idVal); ok {
				continue
			}
		}

		// Skip validation when the target object's schema isn't loaded. Test
		// harnesses often register only the object under test; production
		// metadata has the complete graph. EXCEPTION: a ControlledByParent
		// child's sharing-parent FK must fail closed when its parent object does
		// not resolve — a malformed/missing parent must never be silently skipped.
		if ps.metadata == nil {
			continue
		}
		if ps.metadata.GetSchema(ctx, targetObject) == nil {
			if ps.nonPolymorphicLookupTargetOperation(ctx, schema, field) == constants.PermEdit {
				return nil, appErrors.NewValidationError("", fmt.Sprintf(
					"%s: parent-controlled parent object %s is not configured",
					field.APIName, targetObject,
				))
			}
			continue
		}

		targets = append(targets, nonPolymorphicLookupTarget{
			field:           *field,
			targetObject:    targetObject,
			idVal:           idVal,
			accessOperation: ps.nonPolymorphicLookupTargetOperation(ctx, schema, field),
			requiresLock:    ps.lookupTargetRequiresParentLock(ctx, schema, field),
		})
	}
	return targets, nil
}

// lookupTargetRequiresParentLock reports whether a non-polymorphic lookup FK
// must FOR UPDATE-lock its referenced parent on child write. The records
// runtime locks master-detail, Cascade references, ControlledByParent
// sharing-parent fields, exclusive-sibling partitions, and lookups that feed a
// parent rollup-summary. A plain optional lookup is validated but never locked.
// The rollup-feeding parents are also locked independently by the rollup
// parent-lock path (collectRollupParentKeysForRecords), so this flag is a
// fail-safe alignment rather than the sole owner of that lock.
func (ps *PersistenceService) lookupTargetRequiresParentLock(ctx context.Context, schema *models.ObjectMetadata, field *models.FieldMetadata) bool {
	if isCreate, _ := ctx.Value(nonPolymorphicLookupCreateIntentKey{}).(bool); isCreate {
		return ps.lookupTargetRequiresParentLockForInsert(ctx, schema, field)
	}
	return ps.lookupTargetRequiresParentLockWithRelationship(ctx, schema, field, ps.lookupTargetRelationship(ctx, schema, field))
}

func (ps *PersistenceService) lookupTargetRequiresParentLockForInsert(ctx context.Context, schema *models.ObjectMetadata, field *models.FieldMetadata) bool {
	return ps.lookupTargetRequiresParentLockWithRelationshipForInsert(ctx, schema, field, ps.lookupTargetRelationship(ctx, schema, field))
}

func (ps *PersistenceService) lookupTargetRequiresParentLockWithRelationship(ctx context.Context, schema *models.ObjectMetadata, field *models.FieldMetadata, rel *models.Relationship) bool {
	return ps.lookupTargetRequiresParentLockWithRelationshipMode(ctx, schema, field, rel, false)
}

func (ps *PersistenceService) lookupTargetRequiresParentLockWithRelationshipForInsert(ctx context.Context, schema *models.ObjectMetadata, field *models.FieldMetadata, rel *models.Relationship) bool {
	return ps.lookupTargetRequiresParentLockWithRelationshipMode(ctx, schema, field, rel, true)
}

func (ps *PersistenceService) lookupTargetRequiresParentLockWithRelationshipMode(ctx context.Context, schema *models.ObjectMetadata, field *models.FieldMetadata, rel *models.Relationship, insertMode bool) bool {
	if field == nil {
		return false
	}
	if models.ResolveFieldRelationshipSemantics(field, models.LifecyclePersistedRead).IsMasterDetail {
		return true
	}
	if metadatarel.ResolveDeleteSemantics(*field, rel).Cascade {
		return true
	}
	if ps.nonPolymorphicLookupTargetOperation(ctx, schema, field) == constants.PermEdit {
		// Sole PermEdit case is the ControlledByParent sharing-parent field.
		return true
	}
	if isExclusiveSiblingPartitionField(schema, field) {
		// The partition lookup of an exclusive-sibling field must lock its
		// parent so all sibling writes sharing that partition serialize — the
		// "exactly one winner per partition" guarantee depends on it, orthogonal
		// to whether the field is otherwise a plain lookup.
		return true
	}
	return ps.lookupFieldFeedsRollup(ctx, schema, field, insertMode)
}

func (ps *PersistenceService) lookupTargetRelationship(ctx context.Context, schema *models.ObjectMetadata, field *models.FieldMetadata) *models.Relationship {
	if ps == nil || ps.metadata == nil || schema == nil || field == nil || len(field.ReferenceTo) != 1 {
		return nil
	}
	relationshipsByField := indexRelationshipsByParentAndField(ps.metadata.GetRelationships(ctx, schema.APIName))
	return relationshipsByField[strings.ToLower(field.ReferenceTo[0])+"|"+strings.ToLower(field.APIName)]
}

// isExclusiveSiblingPartitionField reports whether field is the partition
// (parent-pointing) field of any exclusive-sibling declaration on schema.
func isExclusiveSiblingPartitionField(schema *models.ObjectMetadata, field *models.FieldMetadata) bool {
	if schema == nil || field == nil {
		return false
	}
	for i := range schema.ExclusiveSiblingFields {
		if strings.EqualFold(
			strings.TrimSpace(schema.ExclusiveSiblingFields[i].PartitionFieldAPIName),
			strings.TrimSpace(field.APIName),
		) {
			return true
		}
	}
	return false
}

// lookupFieldFeedsRollup reports whether this lookup field participates in a
// parent rollup path. Safe explicit eventual COUNT rollups can skip only the
// root child-insert lookup lock; update/delete/shared guard paths remain
// fail-closed and must continue treating the lookup as rollup-fed.
func (ps *PersistenceService) lookupFieldFeedsRollup(ctx context.Context, schema *models.ObjectMetadata, field *models.FieldMetadata, insertMode bool) bool {
	if ps == nil || ps.metadata == nil || schema == nil || field == nil {
		return false
	}
	deps := ps.metadata.RollupChildDependencies(ctx, schema.APIName)
	for i := range deps {
		cfg := deps[i].RollupField.RollupConfig
		if cfg == nil {
			continue
		}
		if strings.EqualFold(strings.TrimSpace(cfg.RelationshipField), strings.TrimSpace(field.APIName)) {
			if insertMode && ps.rollupDependencyCanSkipInsertParentLock(ctx, deps[i]) {
				continue
			}
			return true
		}
	}
	return false
}

func (ps *PersistenceService) rollupDependencyCanSkipInsertParentLock(ctx context.Context, dep domainSchema.RollupChildDependency) bool {
	cfg := dep.RollupField.RollupConfig
	if ps == nil || cfg == nil {
		return false
	}
	if !strings.EqualFold(strings.TrimSpace(cfg.CalcType), "COUNT") {
		return false
	}
	if !strings.EqualFold(strings.TrimSpace(cfg.MaintenanceMode), "eventual") {
		return false
	}
	if ps.metadata == nil || ps.metadata.HasRollupChildDependencies(ctx, dep.ParentObjectAPIName) {
		return false
	}
	canApply, err := ps.CanApplyDerivedCountRollupDeltas(ctx, dep.ParentObjectAPIName, dep.RollupField.APIName)
	if err != nil || !canApply {
		return false
	}
	isEventual, err := ps.CountRollupFieldsDeclareEventual(ctx, dep.ParentObjectAPIName, dep.RollupField.APIName)
	if err != nil || !isEventual {
		return false
	}
	return true
}

func (ps *PersistenceService) nonPolymorphicLookupTargetOperation(ctx context.Context, schema *models.ObjectMetadata, field *models.FieldMetadata) string {
	if ps == nil || ps.metadata == nil || schema == nil || field == nil ||
		schema.SharingModel != constants.SharingModelControlledByParent {
		return constants.PermRead
	}
	rel := ps.metadata.GetSharingParentRelationship(ctx, schema.APIName)
	if rel == nil || !strings.EqualFold(rel.FieldAPIName, field.APIName) {
		return constants.PermRead
	}
	return constants.PermEdit
}

func (ps *PersistenceService) validateParentControlledCreateSharingParentPresence(ctx context.Context, data models.SObject, schema *models.ObjectMetadata) error {
	isCreate, _ := ctx.Value(nonPolymorphicLookupCreateIntentKey{}).(bool)
	if !isCreate || ps == nil || ps.metadata == nil || ps.permissions == nil || schema == nil ||
		schema.SharingModel != constants.SharingModelControlledByParent {
		return nil
	}
	rel := ps.metadata.GetSharingParentRelationship(ctx, schema.APIName)
	if rel == nil {
		return appErrors.NewValidationError("", fmt.Sprintf(
			"%s: parent-controlled object requires a sharing-parent relationship",
			schema.APIName,
		))
	}
	field := sharingParentFieldForLookupValidation(schema, rel.FieldAPIName)
	if field == nil {
		return appErrors.NewValidationError("", fmt.Sprintf(
			"%s: parent-controlled sharing-parent field %s is not configured",
			schema.APIName, rel.FieldAPIName,
		))
	}
	if permissions.ParentControlledRequiredWritePermissionIsCapability(schema, constants.PermCreate) {
		return nil
	}
	if lookupValuePresent(data, field.APIName) {
		return nil
	}
	parentSchema := ps.metadata.GetSchema(ctx, rel.ParentObjectAPIName)
	if parentSchema == nil {
		// Malformed ControlledByParent edge: the sharing parent points at an
		// object that does not resolve (partial/two-step apply, deleted parent).
		// Fail closed rather than silently admit the create.
		return appErrors.NewValidationError("", fmt.Sprintf(
			"%s: parent-controlled parent object %s is not configured",
			schema.APIName, rel.ParentObjectAPIName,
		))
	}
	if !ps.permissions.RequiresRecordAwareCRUDDeferral(ctx, parentSchema, constants.PermEdit) {
		return nil
	}
	return appErrors.NewValidationError("", fmt.Sprintf(
		"%s: referenced %s record is required when parent edit requires record context",
		field.APIName, rel.ParentObjectAPIName,
	))
}

func sharingParentFieldForLookupValidation(schema *models.ObjectMetadata, fieldAPIName string) *models.FieldMetadata {
	if schema == nil || strings.TrimSpace(fieldAPIName) == "" {
		return nil
	}
	for i := range schema.Fields {
		if strings.EqualFold(schema.Fields[i].APIName, fieldAPIName) {
			return &schema.Fields[i]
		}
	}
	return nil
}

func lookupValuePresent(data models.SObject, fieldAPIName string) bool {
	rawVal, present := data[fieldAPIName]
	if !present || rawVal == nil {
		return false
	}
	idVal, ok := rawVal.(string)
	if !ok {
		return true
	}
	return strings.TrimSpace(idVal) != ""
}

func (ps *PersistenceService) lockLookupParentsIfCoordinated(ctx context.Context, targets []nonPolymorphicLookupTarget) error {
	skipLock, _ := ctx.Value(skipCoordinatedLookupParentLockKey{}).(bool)
	if skipLock {
		return nil
	}
	if ps == nil || ps.txManager == nil || recordlocks.ScopeFromContext(ctx) == nil {
		return nil
	}
	if !ps.txManager.IsPessimisticTx(ctx) {
		return appErrors.NewInternalError("lookup parent lock requires pessimistic transaction", nil)
	}
	keys, targetsByKey := orderedParentKeysForLookupTargets(targets)
	// Batch all keys into a single LockParentRows call so that an out-of-order
	// abort returns an outOfOrderParentLockRetryError carrying the COMPLETE
	// ordered key set.  The per-key loop previously returned after the first
	// out-of-order hit, starving the retry loop of the remaining keys (NX#2175
	// Defect A).  lockInitialParentRowsWithLookupTargets translates only
	// *missingParentLockRowError into a validation error; out-of-order and all
	// other internal errors are propagated verbatim.
	_, err := ps.lockInitialParentRowsWithLookupTargets(ctx, keys, targetsByKey)
	return err
}

func (ps *PersistenceService) prelockNonPolymorphicLookupUpdateParents(ctx context.Context, objectName string, schema *models.ObjectMetadata, id string, updates models.SObject, currentUser *models.UserSession, forced []recordlocks.ParentKey) (map[string]struct{}, error) {
	targets, _, err := ps.collectNonPolymorphicLookupUpdateParentTargets(ctx, objectName, schema, id, updates, currentUser)
	if err != nil {
		return nil, err
	}
	keys, targetsByKey := orderedParentKeysForLookupTargets(targets)
	keys = append(keys, forced...)
	return ps.lockInitialParentRowsWithLookupTargets(ctx, keys, targetsByKey)
}

func (ps *PersistenceService) collectNonPolymorphicLookupUpdateParentTargets(ctx context.Context, objectName string, schema *models.ObjectMetadata, id string, updates models.SObject, currentUser *models.UserSession) ([]nonPolymorphicLookupTarget, models.SObject, error) {
	if !shouldPrelockNonPolymorphicLookupParents(ctx, ps, schema) {
		return nil, nil, nil
	}
	oldRecord, err := ps.loadAndValidateAccessUnlocked(ctx, objectName, id, schema, currentUser)
	if err != nil {
		return nil, nil, err
	}
	effectiveUpdates, hasChanges, err := ps.filterEditableFields(ctx, objectName, schema, updates, oldRecord, currentUser)
	if err != nil {
		return nil, nil, err
	}
	candidate := oldRecord
	if hasChanges {
		candidate = ps.mergeRecords(oldRecord, effectiveUpdates)
	}
	if hasChanges {
		if err := ps.validateNonPolymorphicLookupWrites(withoutCoordinatedLookupParentLock(ctx), effectiveUpdates, schema, currentUser); err != nil {
			return nil, nil, err
		}
	}
	targets, err := ps.collectUpdateNonPolymorphicLookupParentTargets(ctx, schema, oldRecord, candidate)
	return targets, oldRecord, err
}

func (ps *PersistenceService) validateNonPolymorphicLookupWrites(ctx context.Context, updates models.SObject, schema *models.ObjectMetadata, currentUser *models.UserSession) error {
	if len(updates) == 0 {
		return nil
	}
	return ps.validateNonPolymorphicLookups(ctx, updates, schema, currentUser)
}

func mergeSObjectsForLookupValidation(target models.SObject, updates models.SObject) models.SObject {
	if len(updates) == 0 {
		return target
	}
	if target == nil {
		target = models.SObject{}
	}
	for key, value := range updates {
		target[key] = value
	}
	return target
}

func (ps *PersistenceService) prelockNonPolymorphicLookupPreparedUpdateParents(ctx context.Context, objectName, id string, schema *models.ObjectMetadata, prepared *models.PreparedRecordUpdate, extraUpdates models.SObject, currentUser *models.UserSession, forced []recordlocks.ParentKey) (map[string]struct{}, error) {
	targets, _, err := ps.collectNonPolymorphicLookupPreparedUpdateParentTargets(ctx, objectName, id, schema, prepared, extraUpdates, currentUser)
	if err != nil {
		return nil, err
	}
	keys, targetsByKey := orderedParentKeysForLookupTargets(targets)
	keys = append(keys, forced...)
	return ps.lockInitialParentRowsWithLookupTargets(ctx, keys, targetsByKey)
}

func (ps *PersistenceService) collectNonPolymorphicLookupPreparedUpdateParentTargets(ctx context.Context, objectName, id string, schema *models.ObjectMetadata, prepared *models.PreparedRecordUpdate, extraUpdates models.SObject, currentUser *models.UserSession) ([]nonPolymorphicLookupTarget, models.SObject, error) {
	if !shouldPrelockNonPolymorphicLookupParents(ctx, ps, schema) {
		return nil, nil, nil
	}
	oldRecord, err := ps.loadAndValidateAccessUnlocked(ctx, objectName, id, schema, currentUser)
	if err != nil {
		return nil, nil, err
	}
	finalRecord := cloneSObject(oldRecord)
	if prepared != nil && prepared.Record != nil {
		finalRecord = cloneSObject(prepared.Record)
	}
	if finalRecord == nil {
		finalRecord = models.SObject{}
	}
	if prepared != nil {
		for key, value := range prepared.EffectiveUpdates {
			finalRecord[key] = value
		}
	}
	var lookupValidationWrites models.SObject
	if prepared != nil {
		lookupValidationWrites = mergeSObjectsForLookupValidation(lookupValidationWrites, prepared.EffectiveUpdates)
	}
	if len(extraUpdates) > 0 {
		normalizedExtraUpdates := schemautil.NormalizeSObject(schema, extraUpdates)
		for key, value := range normalizedExtraUpdates {
			finalRecord[key] = value
		}
		lookupValidationWrites = mergeSObjectsForLookupValidation(lookupValidationWrites, normalizedExtraUpdates)
	}
	if prepared != nil {
		internalUpdates, err := ps.normalizePreparedInternalUpdates(schema, prepared.InternalUpdates)
		if err != nil {
			return nil, nil, err
		}
		for key, value := range internalUpdates {
			finalRecord[key] = value
		}
		lookupValidationWrites = mergeSObjectsForLookupValidation(lookupValidationWrites, internalUpdates)
	}
	if err := ps.validateNonPolymorphicLookupWrites(withoutCoordinatedLookupParentLock(ctx), lookupValidationWrites, schema, currentUser); err != nil {
		return nil, nil, err
	}
	targets, err := ps.collectUpdateNonPolymorphicLookupParentTargets(ctx, schema, oldRecord, finalRecord)
	return targets, oldRecord, err
}

func shouldPrelockNonPolymorphicLookupParents(ctx context.Context, ps *PersistenceService, schema *models.ObjectMetadata) bool {
	return schemaHasLockableLookupFields(schema) &&
		ps != nil &&
		ps.txManager != nil &&
		recordlocks.ScopeFromContext(ctx) != nil &&
		ps.txManager.IsPessimisticTx(ctx)
}

func (ps *PersistenceService) ensureNonPolymorphicLookupParentsPrelocked(ctx context.Context, data models.SObject, schema *models.ObjectMetadata, prelocked map[string]struct{}) error {
	if len(data) == 0 || schema == nil {
		return nil
	}
	targets, err := ps.collectNonPolymorphicLookupTargets(ctx, data, schema)
	if err != nil {
		return err
	}
	return ps.ensureNonPolymorphicLookupTargetsPrelocked(ctx, targets, prelocked)
}

func (ps *PersistenceService) ensureNonPolymorphicLockedOldParentsPrelocked(ctx context.Context, schema *models.ObjectMetadata, preloadedOldRecord, lockedOldRecord models.SObject, prelocked map[string]struct{}) error {
	if !schemaHasLockableLookupFields(schema) || len(preloadedOldRecord) == 0 || len(lockedOldRecord) == 0 {
		return nil
	}
	// Update-specific lookup prelock verification is owned by
	// ensureUpdateNonPolymorphicLookupParentsPrelocked, which compares old and
	// final relationship values. This early check covers the concurrency window
	// where initial prelock used an unlocked old record, then the later child-row
	// lock observed a different old FK value. Unchanged lock-required old FKs are
	// still part of the authoritative update prelock set, so a missing entry here
	// is retryable instead of being allowed to drift into child-before-parent
	// ordering.
	targets, err := ps.collectUpdateNonPolymorphicLookupParentTargets(ctx, schema, preloadedOldRecord, lockedOldRecord)
	if err != nil {
		return err
	}
	return ps.ensureNonPolymorphicLookupTargetsPrelocked(ctx, targets, prelocked)
}

func (ps *PersistenceService) ensureUpdateNonPolymorphicLookupParentsPrelocked(ctx context.Context, schema *models.ObjectMetadata, oldRecord, finalRecord models.SObject, prelocked map[string]struct{}) error {
	if !schemaHasLockableLookupFields(schema) {
		return nil
	}
	targets, err := ps.collectUpdateNonPolymorphicLookupParentTargets(ctx, schema, oldRecord, finalRecord)
	if err != nil {
		return err
	}
	return ps.ensureNonPolymorphicLookupTargetsPrelocked(ctx, targets, prelocked)
}

func (ps *PersistenceService) ensureNonPolymorphicLookupTargetsPrelocked(ctx context.Context, targets []nonPolymorphicLookupTarget, prelocked map[string]struct{}) error {
	if len(targets) == 0 {
		return nil
	}
	scope := recordlocks.ScopeFromContext(ctx)
	var missing []recordlocks.ParentKey
	for _, target := range targets {
		// Mirror the lock-set filter in orderedParentKeysForLookupTargets: a
		// plain lookup is never locked, so it must not be demanded here either,
		// or this verifier would raise a spurious prelock-retry for a parent we
		// deliberately left unlocked.
		if !target.requiresLock {
			continue
		}
		key := recordlocks.ParentKey{ObjectName: target.targetObject, RecordID: target.idVal}
		keyID := recordlocks.KeyID(key)
		if keyID == "\x00" {
			continue
		}
		if _, ok := prelocked[keyID]; ok {
			continue
		}
		if scope != nil && scope.HasAcquired(key) {
			continue
		}
		missing = append(missing, key)
	}
	if len(missing) > 0 {
		return &parentLockPrelockRetryError{keys: missing, reason: "lookup parent"}
	}
	return nil
}

func (ps *PersistenceService) collectUpdateNonPolymorphicLookupParentTargets(ctx context.Context, schema *models.ObjectMetadata, oldRecord, finalRecord models.SObject) ([]nonPolymorphicLookupTarget, error) {
	if schema == nil || (len(oldRecord) == 0 && len(finalRecord) == 0) {
		return nil, nil
	}
	var targets []nonPolymorphicLookupTarget
	for i := range schema.Fields {
		field := &schema.Fields[i]
		if !isNonPolymorphicReference(field) || isValidatorSkippedField(field.APIName) {
			continue
		}
		oldVal, oldPresent := lookupRecordValueCaseInsensitive(oldRecord, field.APIName)
		newVal, newPresent := lookupRecordValueCaseInsensitive(finalRecord, field.APIName)
		if !oldPresent && !newPresent {
			continue
		}
		if ps.areFieldValuesEqual(fieldByKey(schema, field.APIName), oldVal, newVal) {
			fieldTargets, err := ps.collectNonPolymorphicLookupTargets(ctx, models.SObject{field.APIName: newVal}, schema)
			if err != nil {
				return nil, err
			}
			for _, target := range fieldTargets {
				if target.requiresLock {
					targets = append(targets, target)
				}
			}
			continue
		}
		for _, value := range []interface{}{oldVal, newVal} {
			fieldTargets, err := ps.collectNonPolymorphicLookupTargets(ctx, models.SObject{field.APIName: value}, schema)
			if err != nil {
				return nil, err
			}
			targets = append(targets, fieldTargets...)
		}
	}
	return targets, nil
}

func (ps *PersistenceService) collectNonPolymorphicLookupTargetsForRecords(ctx context.Context, schema *models.ObjectMetadata, records ...models.SObject) ([]nonPolymorphicLookupTarget, error) {
	var targets []nonPolymorphicLookupTarget
	for _, record := range records {
		recordTargets, err := ps.collectNonPolymorphicLookupTargets(ctx, record, schema)
		if err != nil {
			return nil, err
		}
		targets = append(targets, recordTargets...)
	}
	return targets, nil
}

func (ps *PersistenceService) collectNonPolymorphicLookupLockTargetsForRecords(ctx context.Context, schema *models.ObjectMetadata, records ...models.SObject) ([]nonPolymorphicLookupTarget, error) {
	var targets []nonPolymorphicLookupTarget
	for _, record := range records {
		recordTargets, err := ps.collectNonPolymorphicLookupTargetsForLocking(ctx, record, schema)
		if err != nil {
			return nil, err
		}
		targets = append(targets, recordTargets...)
	}
	return targets, nil
}

// orderedParentKeysForLookupTargets is the lock-acquisition authority: it
// returns only the targets that genuinely require an exclusive FOR UPDATE parent
// lock (master-detail, Cascade reference, ControlledByParent sharing-parent,
// exclusive-sibling partition, rollup-feeding). Plain optional lookups are
// intentionally excluded so unrelated writes do not serialize on shared
// low-cardinality parents.
//
// IMPORTANT: this set must NOT be used as a referential-integrity (missing
// parent) guard, because it drops plain lookups. The bulk-insert missing-parent
// check uses orderedExistenceParentKeysForLookupTargets, which keeps every
// lookup, so a dangling FK is still rejected even under SkipValidation.
func orderedParentKeysForLookupTargets(targets []nonPolymorphicLookupTarget) ([]recordlocks.ParentKey, map[string]nonPolymorphicLookupTarget) {
	return orderedParentKeysForLookupTargetsFiltered(targets, true)
}

// orderedExistenceParentKeysForLookupTargets is the existence-check authority:
// it returns every non-polymorphic lookup target regardless of requiresLock, so
// callers that rely on this set as a missing-parent guard (the SkipValidation
// bulk-insert prelock, which is the last referential-integrity check before rows
// land) continue to reject dangling foreign keys on plain optional lookups.
func orderedExistenceParentKeysForLookupTargets(targets []nonPolymorphicLookupTarget) ([]recordlocks.ParentKey, map[string]nonPolymorphicLookupTarget) {
	return orderedParentKeysForLookupTargetsFiltered(targets, false)
}

func orderedParentKeysForLookupTargetsFiltered(targets []nonPolymorphicLookupTarget, onlyLockable bool) ([]recordlocks.ParentKey, map[string]nonPolymorphicLookupTarget) {
	targetsByKey := make(map[string]nonPolymorphicLookupTarget, len(targets))
	keys := make([]recordlocks.ParentKey, 0, len(targets))
	for _, target := range targets {
		if onlyLockable && !target.requiresLock {
			continue
		}
		key := recordlocks.ParentKey{ObjectName: target.targetObject, RecordID: target.idVal}
		keyID := recordlocks.KeyID(key)
		if keyID == "\x00" {
			continue
		}
		if _, ok := targetsByKey[keyID]; ok {
			continue
		}
		targetsByKey[keyID] = target
		keys = append(keys, key)
	}
	return recordlocks.OrderedParentKeys(keys), targetsByKey
}

// isValidatorSkippedField returns true for non-polymorphic FK fields
// whose access/existence is NOT owned by this generic validator:
//
//   - OwnerId  — has its own owner-sharing model in SF semantics,
//     its own assignment-permission check via ChangeOwner /
//     validateQueueAssignment, and is often defaulted to the current
//     user. Treating it as a plain FK-accessibility check adds a
//     new enforcement the platform never did and breaks any caller
//     that sets OwnerId to a User the current request-user can't
//     Read via the regular sharing model. Canonical example: a
//     cron/system job creating records on behalf of a user whose
//     User row exists but isn't RLS-visible to the job's session
//     identity.
//   - RecordTypeId — has a dedicated owner, validateRecordTypeId,
//     which checks object membership, active state, and the master
//     record type special-case. Re-validating it here through
//     Retrieve("RecordType", ...) creates a second incompatible owner
//     for the same write fact.
//   - CreatedById / LastModifiedById — system-set audit fields, not
//     generic user-authored FKs. They flow through the audit-field
//     writer path, not through user-provided payload validation.
//
// Every other non-polymorphic Reference field IS validated — Quote
// .OpportunityId, Order.QuoteId, custom __c lookups, etc. The skip
// set is deliberately narrow: only fields whose access is owned by
// a separate, pre-existing mechanism. Adding to this list requires
// naming that alternate owner in the comment above; a bare skip
// without a named alternate owner would reopen Class-2 drift.
func isValidatorSkippedField(apiName string) bool {
	switch strings.ToLower(apiName) {
	case "ownerid", "recordtypeid", "createdbyid", "lastmodifiedbyid":
		return true
	}
	return false
}

// isNonPolymorphicReference is the fence for the user-level lookup access
// gate: Reference-type, exactly one referenceTo target, and not a multi-target
// polymorphic lookup. Transitional single-target polymorphic references still
// pass through this gate so marking a field polymorphic to generate
// <field>_type does not silently weaken the public write contract.
func isNonPolymorphicReference(field *models.FieldMetadata) bool {
	if field == nil {
		return false
	}
	if !constants.IsReferenceType(constants.NormalizeFieldType(string(field.Type))) {
		return false
	}
	if len(field.ReferenceTo) != 1 {
		return false
	}
	if strings.TrimSpace(field.ReferenceTo[0]) == "" {
		return false
	}
	return true
}