package records

import (
	"context"
	stderrors "errors"
	"fmt"
	"reflect"
	"strconv"
	"strings"

	"github.com/nexuscrm/kernel/pkg/apierr"
	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/formulastorage"
	"github.com/nexuscrm/shared/pkg/models"
	zlog "github.com/rs/zerolog/log"
)

type exclusiveSiblingContract struct {
	FieldAPIName          string
	PartitionFieldAPIName string
	ParentObjectAPIName   string
}

type exclusiveSiblingResetPlan struct {
	contract          exclusiveSiblingContract
	partitionValue    string
	winnerID          string
	skippedSiblingIDs map[string]struct{}
}

type exclusiveSiblingResetRepository interface {
	LockExclusiveSiblingResetRows(ctx context.Context, schema *models.ObjectMetadata, fieldAPIName, partitionFieldAPIName string, partitionValue interface{}, winnerID string) ([]models.SObject, error)
}

type parentLockPrelockRetryError struct {
	keys   []recordlocks.ParentKey
	reason string
}

func (e *parentLockPrelockRetryError) Error() string {
	reason := "parent"
	if e != nil && strings.TrimSpace(e.reason) != "" {
		reason = e.reason
	}
	if e == nil || len(e.keys) == 0 {
		return reason + " was discovered after child row lock; retry update"
	}
	parts := make([]string, 0, len(e.keys))
	for _, key := range recordlocks.OrderedParentKeys(e.keys) {
		parts = append(parts, fmt.Sprintf("%s:%s", key.ObjectName, key.RecordID))
	}
	return fmt.Sprintf("%s %s was discovered after child row lock; retry update", reason, strings.Join(parts, ", "))
}

func (e *parentLockPrelockRetryError) ParentPrelockKeys() []recordlocks.ParentKey {
	if e == nil || len(e.keys) == 0 {
		return nil
	}
	return append([]recordlocks.ParentKey(nil), e.keys...)
}

func parentLockPrelockRetryKeys(err error) ([]recordlocks.ParentKey, bool) {
	var retryErr recordlocks.RetryablePrelockError
	if !stderrors.As(err, &retryErr) || retryErr == nil {
		return nil, false
	}
	keys := retryErr.ParentPrelockKeys()
	if len(keys) == 0 {
		return nil, false
	}
	return append([]recordlocks.ParentKey(nil), keys...), true
}

func parentLockForcedPrelocksFromContext(ctx context.Context) []recordlocks.ParentKey {
	return recordlocks.ForcedParentPrelocksFromContext(ctx)
}

func withParentLockForcedPrelocks(ctx context.Context, keys []recordlocks.ParentKey) context.Context {
	return recordlocks.WithForcedParentPrelocks(ctx, keys)
}

func schemaHasExclusiveSiblingFields(schema *models.ObjectMetadata) bool {
	return schema != nil && len(schema.ExclusiveSiblingFields) > 0
}

func exclusiveSiblingDeferredUniqueFields(schema *models.ObjectMetadata) map[string]struct{} {
	if schema == nil || len(schema.ExclusiveSiblingFields) == 0 {
		return nil
	}
	exclusiveFields := map[string]struct{}{}
	for _, rule := range schema.ExclusiveSiblingFields {
		fieldName := strings.TrimSpace(rule.FieldAPIName)
		if fieldName != "" {
			exclusiveFields[strings.ToLower(fieldName)] = struct{}{}
		}
	}
	if len(exclusiveFields) == 0 {
		return nil
	}
	esSensitiveFields := make(map[string]struct{}, len(exclusiveFields))
	for fieldName := range exclusiveFields {
		esSensitiveFields[fieldName] = struct{}{}
	}
	for changed := true; changed; {
		changed = false
		for i := range schema.Fields {
			field := &schema.Fields[i]
			fieldName := strings.ToLower(strings.TrimSpace(field.APIName))
			if fieldName == "" {
				continue
			}
			if _, ok := esSensitiveFields[fieldName]; ok {
				continue
			}
			if field.Formula == nil || strings.TrimSpace(*field.Formula) == "" {
				continue
			}
			storage, err := formulastorage.Resolve(formulastorage.SpecFromField(field))
			if err != nil {
				zlog.Warn().
					Str("object", schema.APIName).
					Str("field", field.APIName).
					Str("error", err.Error()).
					Msg("records.exclusiveSibling: invalid formula storage metadata; using canonical fallback")
			}
			if !storage.IsWriteMaterialized() {
				continue
			}
			if formulaReferencesAnyField(*field.Formula, esSensitiveFields) {
				esSensitiveFields[fieldName] = struct{}{}
				changed = true
			}
		}
	}

	deferred := map[string]struct{}{}
	for i := range schema.Fields {
		field := &schema.Fields[i]
		if !field.IsUnique {
			continue
		}
		fieldName := strings.ToLower(strings.TrimSpace(field.APIName))
		if fieldName == "" {
			continue
		}
		if _, ok := esSensitiveFields[fieldName]; ok {
			deferred[fieldName] = struct{}{}
		}
	}
	if len(deferred) == 0 {
		return nil
	}
	return deferred
}

func formulaReferencesAnyField(formulaText string, fieldNames map[string]struct{}) bool {
	formulaText = strings.ToLower(formulaText)
	for fieldName := range fieldNames {
		if formulaReferencesField(formulaText, fieldName) {
			return true
		}
	}
	return false
}

func formulaReferencesField(formulaText, fieldName string) bool {
	formulaText = strings.ToLower(formulaText)
	fieldName = strings.ToLower(strings.TrimSpace(fieldName))
	if formulaText == "" || fieldName == "" {
		return false
	}
	searchFrom := 0
	for {
		idx := strings.Index(formulaText[searchFrom:], fieldName)
		if idx < 0 {
			return false
		}
		idx += searchFrom
		beforeOK := idx == 0 || !isFormulaIdentifierRune(rune(formulaText[idx-1]))
		afterIdx := idx + len(fieldName)
		afterOK := afterIdx >= len(formulaText) || !isFormulaIdentifierRune(rune(formulaText[afterIdx]))
		if beforeOK && afterOK {
			return true
		}
		searchFrom = idx + len(fieldName)
	}
}

func isFormulaIdentifierRune(r rune) bool {
	return r == '_' ||
		(r >= 'a' && r <= 'z') ||
		(r >= 'A' && r <= 'Z') ||
		(r >= '0' && r <= '9')
}

func (ps *PersistenceService) anyObjectHasExclusiveSiblingFields(ctx context.Context, objectNames []string) bool {
	if ps == nil || ps.metadata == nil {
		return false
	}
	seen := make(map[string]struct{}, len(objectNames))
	for _, objectName := range objectNames {
		if ps.objectCanReachExclusiveSiblingFields(ctx, objectName, seen) {
			return true
		}
	}
	return false
}

// objectCanReachExclusiveSiblingFields is the PR-C shared-surface capability
// gate: direct ES writes are implemented, but generated shared-surface writes
// must fail before the first physical write until their coordinator support
// lands.
func (ps *PersistenceService) objectCanReachExclusiveSiblingFields(ctx context.Context, objectName string, seen map[string]struct{}) bool {
	if ps == nil || ps.metadata == nil {
		return false
	}
	objectName = strings.TrimSpace(objectName)
	if objectName == "" {
		return false
	}
	if seen == nil {
		seen = map[string]struct{}{}
	}
	key := strings.ToLower(objectName)
	if _, ok := seen[key]; ok {
		return false
	}
	seen[key] = struct{}{}

	schema := ps.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return false
	}
	if schemaHasExclusiveSiblingFields(schema) {
		return true
	}
	if len(schema.DefaultChildRows) > 0 {
		contracts, err := ResolveDefaultChildRowsContractsContext(ctx, ps.metadata, schema, schema.DefaultChildRows, ps.engineOwnedFieldSet)
		if err != nil {
			return true
		}
		for _, contract := range contracts {
			if ps.objectCanReachExclusiveSiblingFields(ctx, contract.TargetObject, seen) {
				return true
			}
		}
	}
	for _, rule := range schema.ChildCloneRules {
		if ps.objectCanReachExclusiveSiblingFields(ctx, rule.TargetObject, seen) {
			return true
		}
	}
	return false
}

func (ps *PersistenceService) objectGeneratedWritesCanReachExclusiveSiblingFields(ctx context.Context, objectName string, seen map[string]struct{}) bool {
	if ps == nil || ps.metadata == nil {
		return false
	}
	objectName = strings.TrimSpace(objectName)
	if objectName == "" {
		return false
	}
	if seen == nil {
		seen = map[string]struct{}{}
	}
	key := strings.ToLower(objectName)
	if _, ok := seen[key]; ok {
		return false
	}
	seen[key] = struct{}{}

	schema := ps.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return false
	}
	if len(schema.DefaultChildRows) > 0 {
		contracts, err := ResolveDefaultChildRowsContractsContext(ctx, ps.metadata, schema, schema.DefaultChildRows, ps.engineOwnedFieldSet)
		if err != nil {
			return true
		}
		for _, contract := range contracts {
			if ps.objectCanReachExclusiveSiblingFields(ctx, contract.TargetObject, nil) {
				return true
			}
		}
	}
	for _, rule := range schema.ChildCloneRules {
		if ps.objectCanReachExclusiveSiblingFields(ctx, rule.TargetObject, nil) {
			return true
		}
	}
	return false
}

func (ps *PersistenceService) rejectExclusiveSiblingSharedWriteSurface(ctx context.Context, surface string, objectNames []string) error {
	if !ps.anyObjectHasExclusiveSiblingFields(ctx, objectNames) {
		return nil
	}
	return appErrors.NewValidationError("ExclusiveSiblingFields", fmt.Sprintf("%s does not support ExclusiveSiblingFields yet; use single-record Insert, Update, or UpsertByExternalID", surface))
}

func (ps *PersistenceService) rejectExclusiveSiblingDeleteRecordsAllOrNone(ctx context.Context, objectNames []string) error {
	if !ps.anyObjectHasExclusiveSiblingFields(ctx, objectNames) {
		return nil
	}
	return appErrors.NewValidationError("ExclusiveSiblingFields", "DeleteRecords(allOrNone=true) does not support ExclusiveSiblingFields yet; use single-record Delete or DeleteRecords(allOrNone=false)")
}

func mergeParentLockKeys(base, extra []recordlocks.ParentKey) ([]recordlocks.ParentKey, bool) {
	if len(extra) == 0 {
		return base, false
	}
	seen := make(map[string]struct{}, len(base)+len(extra))
	merged := make([]recordlocks.ParentKey, 0, len(base)+len(extra))
	for _, key := range base {
		normalized, ok := recordlocks.NormalizeParentKey(key)
		if !ok {
			continue
		}
		keyID := recordlocks.KeyID(normalized)
		if _, ok := seen[keyID]; ok {
			continue
		}
		seen[keyID] = struct{}{}
		merged = append(merged, normalized)
	}
	changed := false
	for _, key := range extra {
		normalized, ok := recordlocks.NormalizeParentKey(key)
		if !ok {
			continue
		}
		keyID := recordlocks.KeyID(normalized)
		if _, ok := seen[keyID]; ok {
			continue
		}
		seen[keyID] = struct{}{}
		merged = append(merged, normalized)
		changed = true
	}
	return merged, changed
}

func (ps *PersistenceService) canRetryParentLockPrelock(ctx context.Context) bool {
	return ps == nil || ps.txManager == nil || !ps.txManager.IsTransactionActive(ctx)
}

func (ps *PersistenceService) rejectExclusiveSiblingPlainOuterTransaction(ctx context.Context, surface string) error {
	if ps == nil || ps.txManager == nil || !ps.txManager.IsTransactionActive(ctx) || ps.txManager.IsPessimisticTx(ctx) {
		return nil
	}
	return appErrors.NewValidationError("ExclusiveSiblingFields", fmt.Sprintf("%s does not support ExclusiveSiblingFields inside a plain outer transaction; use a pessimistic outer transaction or call the single-record path outside an existing transaction", surface))
}

func (ps *PersistenceService) exclusiveSiblingContracts(schema *models.ObjectMetadata) ([]exclusiveSiblingContract, error) {
	if schema == nil || len(schema.ExclusiveSiblingFields) == 0 {
		return nil, nil
	}
	contracts := make([]exclusiveSiblingContract, 0, len(schema.ExclusiveSiblingFields))
	seen := map[string]struct{}{}
	for i, rule := range schema.ExclusiveSiblingFields {
		fieldName := strings.TrimSpace(rule.FieldAPIName)
		partitionName := strings.TrimSpace(rule.PartitionFieldAPIName)
		prefix := fmt.Sprintf("%s.ExclusiveSiblingFields[%d]", schema.APIName, i)
		if fieldName == "" || partitionName == "" {
			return nil, appErrors.NewInternalError(fmt.Sprintf("%s is missing fieldApiName or partitionFieldApiName", prefix), nil)
		}
		if strings.EqualFold(fieldName, partitionName) {
			return nil, appErrors.NewInternalError(fmt.Sprintf("%s fieldApiName and partitionFieldApiName must differ", prefix), nil)
		}
		field := models.FindField(schema, fieldName)
		if field == nil || field.IsVirtual || field.Formula != nil || field.ProjectionConfig != nil || constants.NormalizeFieldType(string(field.Type)) != constants.FieldTypeBoolean {
			return nil, appErrors.NewInternalError(fmt.Sprintf("%s fieldApiName %q must be a physical boolean field", prefix, fieldName), nil)
		}
		partitionField := models.FindField(schema, partitionName)
		if partitionField == nil || partitionField.IsVirtual || partitionField.Formula != nil || partitionField.ProjectionConfig != nil {
			return nil, appErrors.NewInternalError(fmt.Sprintf("%s partitionFieldApiName %q must be a physical reference field", prefix, partitionName), nil)
		}
		sem := models.ResolveFieldRelationshipSemantics(partitionField, models.LifecyclePersistedRead)
		if !sem.IsRelationship || sem.IsPolymorphic || len(sem.ReferenceTargets) != 1 || strings.TrimSpace(sem.ReferenceTargets[0]) == "" {
			return nil, appErrors.NewInternalError(fmt.Sprintf("%s partitionFieldApiName %q must reference exactly one parent object", prefix, partitionName), nil)
		}
		key := strings.ToLower(field.APIName) + "\x00" + strings.ToLower(partitionField.APIName)
		if _, ok := seen[key]; ok {
			return nil, appErrors.NewInternalError(fmt.Sprintf("%s duplicates another exclusive sibling field declaration", prefix), nil)
		}
		seen[key] = struct{}{}
		contracts = append(contracts, exclusiveSiblingContract{
			FieldAPIName:          field.APIName,
			PartitionFieldAPIName: partitionField.APIName,
			ParentObjectAPIName:   strings.TrimSpace(sem.ReferenceTargets[0]),
		})
	}
	return contracts, nil
}

func (ps *PersistenceService) prepareExclusiveSiblingInsert(ctx context.Context, schema *models.ObjectMetadata, finalRecord models.SObject) ([]exclusiveSiblingResetPlan, error) {
	contracts, err := ps.exclusiveSiblingContracts(schema)
	if err != nil || len(contracts) == 0 {
		return nil, err
	}
	winnerID := exclusiveSiblingRecordID(finalRecord[constants.FieldID])
	var plans []exclusiveSiblingResetPlan
	for _, contract := range contracts {
		if !exclusiveSiblingTruthy(finalRecord[contract.FieldAPIName]) {
			continue
		}
		partition := exclusiveSiblingRecordID(finalRecord[contract.PartitionFieldAPIName])
		if partition == "" {
			continue
		}
		if winnerID != "" {
			plans = append(plans, exclusiveSiblingResetPlan{contract: contract, partitionValue: partition, winnerID: winnerID})
		}
	}
	keys, err := ps.collectExclusiveSiblingInsertParentKeys(schema, finalRecord)
	if err != nil {
		return nil, err
	}
	if _, err := ps.lockInitialParentRows(ctx, keys); err != nil {
		return nil, err
	}
	return plans, nil
}

func (ps *PersistenceService) collectExclusiveSiblingInsertParentKeys(schema *models.ObjectMetadata, finalRecord models.SObject) ([]recordlocks.ParentKey, error) {
	contracts, err := ps.exclusiveSiblingContracts(schema)
	if err != nil || len(contracts) == 0 {
		return nil, err
	}
	var keys []recordlocks.ParentKey
	for _, contract := range contracts {
		if !exclusiveSiblingTruthy(finalRecord[contract.FieldAPIName]) {
			continue
		}
		partition := exclusiveSiblingRecordID(finalRecord[contract.PartitionFieldAPIName])
		if partition == "" {
			continue
		}
		keys = append(keys, recordlocks.ParentKey{ObjectName: contract.ParentObjectAPIName, RecordID: partition})
	}
	return keys, nil
}

func (ps *PersistenceService) collectExclusiveSiblingDeleteParentKeys(schema *models.ObjectMetadata, oldRecord models.SObject) ([]recordlocks.ParentKey, error) {
	contracts, err := ps.exclusiveSiblingContracts(schema)
	if err != nil || len(contracts) == 0 {
		return nil, err
	}
	keys := make([]recordlocks.ParentKey, 0, len(contracts))
	for _, contract := range contracts {
		if !exclusiveSiblingTruthy(oldRecord[contract.FieldAPIName]) {
			continue
		}
		partition := exclusiveSiblingRecordID(oldRecord[contract.PartitionFieldAPIName])
		if partition == "" {
			continue
		}
		keys = append(keys, recordlocks.ParentKey{ObjectName: contract.ParentObjectAPIName, RecordID: partition})
	}
	return keys, nil
}

func (ps *PersistenceService) collectExclusiveSiblingUpdateParentKeys(ctx context.Context, objectName string, schema *models.ObjectMetadata, id string, updates models.SObject, currentUser *models.UserSession) ([]recordlocks.ParentKey, error) {
	contracts, err := ps.exclusiveSiblingContracts(schema)
	if err != nil || len(contracts) == 0 {
		return nil, err
	}
	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
	}
	if !hasChanges {
		return nil, nil
	}
	candidate := ps.mergeRecords(oldRecord, effectiveUpdates)
	keys := make([]recordlocks.ParentKey, 0, len(contracts))
	for _, contract := range contracts {
		if oldPartition := exclusiveSiblingRecordID(oldRecord[contract.PartitionFieldAPIName]); oldPartition != "" {
			keys = append(keys, recordlocks.ParentKey{ObjectName: contract.ParentObjectAPIName, RecordID: oldPartition})
		}

		_, fieldChanged := effectiveUpdates[contract.FieldAPIName]
		_, partitionChanged := effectiveUpdates[contract.PartitionFieldAPIName]
		if !fieldChanged && !partitionChanged {
			continue
		}
		if !exclusiveSiblingTruthy(candidate[contract.FieldAPIName]) {
			continue
		}
		partition := exclusiveSiblingRecordID(candidate[contract.PartitionFieldAPIName])
		if partition == "" {
			continue
		}
		keys = append(keys, recordlocks.ParentKey{ObjectName: contract.ParentObjectAPIName, RecordID: partition})
	}
	return keys, nil
}

func (ps *PersistenceService) collectExclusiveSiblingPreparedUpdateParentKeys(ctx context.Context, objectName, id string, schema *models.ObjectMetadata, prepared *models.PreparedRecordUpdate, extraUpdates models.SObject, currentUser *models.UserSession) ([]recordlocks.ParentKey, error) {
	contracts, err := ps.exclusiveSiblingContracts(schema)
	if err != nil || len(contracts) == 0 || prepared == nil {
		return nil, err
	}
	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{}
	}
	finalRecord := cloneSObject(prepared.Record)
	if finalRecord == nil {
		finalRecord = models.SObject{}
	}
	if len(extraUpdates) > 0 {
		for key, value := range schemautil.NormalizeSObject(schema, extraUpdates) {
			effectiveUpdates[key] = value
			finalRecord[key] = value
		}
	}
	internalUpdates, err := ps.normalizePreparedInternalUpdates(schema, prepared.InternalUpdates)
	if err != nil {
		return nil, err
	}
	for key, value := range internalUpdates {
		effectiveUpdates[key] = value
		finalRecord[key] = value
	}
	if len(effectiveUpdates) == 0 {
		return nil, nil
	}

	keys := make([]recordlocks.ParentKey, 0, len(contracts))
	for _, contract := range contracts {
		if oldPartition := exclusiveSiblingRecordID(oldRecord[contract.PartitionFieldAPIName]); oldPartition != "" {
			keys = append(keys, recordlocks.ParentKey{ObjectName: contract.ParentObjectAPIName, RecordID: oldPartition})
		}

		_, fieldChanged := effectiveUpdates[contract.FieldAPIName]
		_, partitionChanged := effectiveUpdates[contract.PartitionFieldAPIName]
		if !fieldChanged && !partitionChanged {
			continue
		}
		if !exclusiveSiblingTruthy(finalRecord[contract.FieldAPIName]) {
			continue
		}
		partition := exclusiveSiblingRecordID(finalRecord[contract.PartitionFieldAPIName])
		if partition == "" {
			continue
		}
		keys = append(keys, recordlocks.ParentKey{ObjectName: contract.ParentObjectAPIName, RecordID: partition})
	}
	return keys, nil
}

func (ps *PersistenceService) ensureExclusiveSiblingLockedOldParentsPrelocked(schema *models.ObjectMetadata, oldRecord models.SObject, prelocked map[string]struct{}) error {
	contracts, err := ps.exclusiveSiblingContracts(schema)
	if err != nil || len(contracts) == 0 {
		return err
	}
	var missing []recordlocks.ParentKey
	for _, contract := range contracts {
		if !exclusiveSiblingTruthy(oldRecord[contract.FieldAPIName]) {
			continue
		}
		partition := exclusiveSiblingRecordID(oldRecord[contract.PartitionFieldAPIName])
		if partition == "" {
			continue
		}
		key := recordlocks.ParentKey{ObjectName: contract.ParentObjectAPIName, RecordID: partition}
		if _, ok := prelocked[recordlocks.KeyID(key)]; !ok {
			missing = append(missing, key)
		}
	}
	if len(missing) > 0 {
		return &parentLockPrelockRetryError{keys: missing, reason: "exclusive sibling parent"}
	}
	return nil
}

func (ps *PersistenceService) prepareExclusiveSiblingUpdate(ctx context.Context, schema *models.ObjectMetadata, id string, finalRecord, effectiveUpdates, oldRecord models.SObject, prelocked map[string]struct{}) ([]exclusiveSiblingResetPlan, error) {
	if prelocked == nil {
		return nil, nil
	}
	contracts, err := ps.exclusiveSiblingContracts(schema)
	if err != nil || len(contracts) == 0 {
		return nil, err
	}
	var plans []exclusiveSiblingResetPlan
	var missing []recordlocks.ParentKey
	for _, contract := range contracts {
		if !exclusiveSiblingTruthy(finalRecord[contract.FieldAPIName]) {
			continue
		}
		partition := exclusiveSiblingRecordID(finalRecord[contract.PartitionFieldAPIName])
		if partition == "" {
			continue
		}

		oldTrue := exclusiveSiblingTruthy(oldRecord[contract.FieldAPIName])
		oldPartition := exclusiveSiblingRecordID(oldRecord[contract.PartitionFieldAPIName])
		_, fieldChanged := effectiveUpdates[contract.FieldAPIName]
		_, partitionChanged := effectiveUpdates[contract.PartitionFieldAPIName]
		needsReset := (!oldTrue && fieldChanged) || (oldTrue && partitionChanged && oldPartition != partition)
		if !needsReset {
			continue
		}

		key := recordlocks.ParentKey{ObjectName: contract.ParentObjectAPIName, RecordID: partition}
		if _, ok := prelocked[recordlocks.KeyID(key)]; !ok {
			missing = append(missing, key)
			continue
		}
		plans = append(plans, exclusiveSiblingResetPlan{contract: contract, partitionValue: partition, winnerID: id})
	}
	if len(missing) > 0 {
		return nil, &parentLockPrelockRetryError{keys: missing, reason: "exclusive sibling parent"}
	}
	return plans, nil
}

type exclusiveSiblingRestoreNoResetKey struct{}

func withExclusiveSiblingRestoreNoReset(ctx context.Context) context.Context {
	return context.WithValue(ctx, exclusiveSiblingRestoreNoResetKey{}, true)
}

func exclusiveSiblingRestoreNoReset(ctx context.Context) bool {
	disabled, _ := ctx.Value(exclusiveSiblingRestoreNoResetKey{}).(bool)
	return disabled
}

func (ps *PersistenceService) ensureExclusiveSiblingRestoreParentsAlreadyLocked(ctx context.Context, schema *models.ObjectMetadata, finalRecord models.SObject, prelocked map[string]struct{}) error {
	scope := recordlocks.ScopeFromContext(ctx)
	if scope == nil {
		return nil
	}
	keys, err := ps.collectExclusiveSiblingInsertParentKeys(schema, finalRecord)
	if err != nil {
		return err
	}
	var missing []recordlocks.ParentKey
	for _, key := range keys {
		if _, ok := prelocked[recordlocks.KeyID(key)]; ok {
			continue
		}
		if !scope.HasAcquired(key) {
			missing = append(missing, key)
		}
	}
	if len(missing) > 0 {
		return &parentLockPrelockRetryError{keys: missing, reason: "exclusive sibling restore parent"}
	}
	return nil
}

func (ps *PersistenceService) validateExclusiveSiblingRestoreConflicts(ctx context.Context, schema *models.ObjectMetadata, id string, finalRecord models.SObject) error {
	contracts, err := ps.exclusiveSiblingContracts(schema)
	if err != nil || len(contracts) == 0 {
		return err
	}
	repo, ok := ps.repo.(exclusiveSiblingResetRepository)
	if !ok {
		return appErrors.NewInternalError("exclusive sibling restore requires repository support", nil)
	}
	for _, contract := range contracts {
		if !exclusiveSiblingTruthy(finalRecord[contract.FieldAPIName]) {
			continue
		}
		partition := exclusiveSiblingRecordID(finalRecord[contract.PartitionFieldAPIName])
		if partition == "" {
			continue
		}
		siblings, err := repo.LockExclusiveSiblingResetRows(ctx, schema, contract.FieldAPIName, contract.PartitionFieldAPIName, partition, id)
		if err != nil {
			return appErrors.NewInternalError("exclusive sibling restore conflict check failed", err)
		}
		if len(siblings) > 0 {
			return apierr.Conflict(
				fmt.Sprintf("cannot restore %s/%s because an active sibling already owns %s for %s=%s", schema.APIName, id, contract.FieldAPIName, contract.PartitionFieldAPIName, partition),
			)
		}
	}
	return nil
}

func (ps *PersistenceService) exclusiveSiblingResetFinalUpdates(ctx context.Context, schema *models.ObjectMetadata, oldRecord models.SObject, updates models.SObject) (models.SObject, models.SObject, error) {
	if schema == nil {
		return nil, nil, appErrors.NewInternalError("exclusive sibling reset formula recompute requires schema", nil)
	}
	finalRecord := ps.mergeRecords(oldRecord, updates)
	fetchErrFields := computeFormulaFieldsWithPrior(ctx, schema, finalRecord, oldRecord, ps.repo, ps.metadata, schema.APIName)
	preserveFetchErroredFormulaFields(oldRecord, finalRecord, fetchErrFields)
	return finalRecord, exclusiveSiblingChangedLogicalUpdates(schema, oldRecord, finalRecord), nil
}

func preserveFetchErroredFormulaFields(oldRecord, finalRecord models.SObject, fetchErrFields map[string]struct{}) {
	for fieldName := range fetchErrFields {
		if oldRecord != nil {
			if oldValue, ok := oldRecord[fieldName]; ok {
				finalRecord[fieldName] = oldValue
				continue
			}
		}
		delete(finalRecord, fieldName)
	}
}

func exclusiveSiblingChangedLogicalUpdates(schema *models.ObjectMetadata, oldRecord, finalRecord models.SObject) models.SObject {
	oldLogical := filterToLogicalFields(schema, oldRecord)
	finalLogical := filterToLogicalFields(schema, finalRecord)
	updates := models.SObject{}
	for fieldName, value := range finalLogical {
		if reflect.DeepEqual(oldLogical[fieldName], value) {
			continue
		}
		updates[fieldName] = value
	}
	return updates
}

func (ps *PersistenceService) applyExclusiveSiblingResets(ctx context.Context, schema *models.ObjectMetadata, plans []exclusiveSiblingResetPlan) error {
	if len(plans) == 0 {
		return nil
	}
	repo, ok := ps.repo.(exclusiveSiblingResetRepository)
	if !ok {
		return appErrors.NewInternalError("exclusive sibling reset requires repository support", nil)
	}
	for _, plan := range plans {
		if schema == nil || plan.partitionValue == "" || plan.winnerID == "" {
			continue
		}
		clearedRows, err := repo.LockExclusiveSiblingResetRows(ctx, schema, plan.contract.FieldAPIName, plan.contract.PartitionFieldAPIName, plan.partitionValue, plan.winnerID)
		if err != nil {
			return appErrors.NewInternalError("exclusive sibling reset failed", err)
		}
		for _, oldRecord := range clearedRows {
			id := exclusiveSiblingRecordID(oldRecord[constants.FieldID])
			if id == "" {
				continue
			}
			if _, skip := plan.skippedSiblingIDs[id]; skip {
				continue
			}
			if err := ps.applyExclusiveSiblingResetRow(ctx, schema, oldRecord, plan.contract.FieldAPIName); err != nil {
				return err
			}
		}
	}
	return nil
}

func (ps *PersistenceService) applyExclusiveSiblingResetRow(ctx context.Context, schema *models.ObjectMetadata, oldRecord models.SObject, fieldAPIName string) error {
	if schema == nil {
		return nil
	}
	id := exclusiveSiblingRecordID(oldRecord[constants.FieldID])
	if id == "" {
		return nil
	}

	resetCtx, resetUser, finalRecord, auditUpdates, err := ps.exclusiveSiblingResetFinalRecord(ctx, schema, oldRecord, fieldAPIName)
	if err != nil {
		return err
	}
	if err := ps.emitUpdateAudits(resetCtx, schema, schema.APIName, id, auditUpdates, oldRecord, resetUser); err != nil {
		return err
	}

	if len(auditUpdates) > 0 {
		resetCtxForUpdate, err := ps.stageOldRecordActiveProjection(resetCtx, schema, id, resetUser)
		if err != nil {
			return err
		}
		if err := ps.repo.Update(resetCtxForUpdate, schema.APIName, id, auditUpdates); err != nil {
			return appErrors.NewInternalError("exclusive sibling reset update failed", err)
		}
		InvalidateRecordReadCache(resetCtxForUpdate, schema.APIName, id)
		if err := ps.markGroupMembershipClosureDirtyAfterMutation(resetCtxForUpdate, schema.APIName, groupMembershipClosureMutationUpdate, auditUpdates); err != nil {
			return appErrors.NewInternalError("group membership closure invalidation failed", err)
		}
		if err := ps.processPostUpdateHooks(resetCtxForUpdate, schema.APIName, finalRecord, oldRecord, resetUser); err != nil {
			return err
		}
		ps.invalidateSecurityCaches(resetCtxForUpdate, schema.APIName, oldRecord, finalRecord)
	}
	return nil
}

func (ps *PersistenceService) exclusiveSiblingResetFinalRecord(ctx context.Context, schema *models.ObjectMetadata, oldRecord models.SObject, fieldAPIName string) (context.Context, *models.UserSession, models.SObject, models.SObject, error) {
	id := exclusiveSiblingRecordID(oldRecord[constants.FieldID])
	resetCtx := constants.WithSecurityContext(ctx, constants.SystemContext, "exclusive sibling reset")
	resetCtx = context.WithValue(resetCtx, constants.ContextKeySuppressRecordTriggeredAutomation, schema.APIName+"/"+id)
	resetUser, err := ps.LoadSystemUserSession(resetCtx)
	if err != nil {
		return resetCtx, nil, nil, nil, appErrors.NewInternalError("exclusive sibling reset system user failed", err)
	}

	effectiveUpdates := ps.generateSystemFields(resetCtx, schema.APIName, models.SObject{fieldAPIName: false}, resetUser, false)
	finalRecord, auditUpdates, err := ps.exclusiveSiblingResetFinalUpdates(resetCtx, schema, oldRecord, effectiveUpdates)
	if err != nil {
		return resetCtx, resetUser, nil, nil, err
	}
	return resetCtx, resetUser, finalRecord, auditUpdates, nil
}

func (ps *PersistenceService) loadAndValidateAccessUnlocked(ctx context.Context, objectName, id string, schema *models.ObjectMetadata, currentUser *models.UserSession) (models.SObject, error) {
	// Parent-lock planning can ask the lookup, exclusive-sibling, and rollup
	// planners for the same unlocked logical row in one mutation attempt. Reuse
	// the attempt-owned read rather than reopening the derived record view for
	// every planner. Locked reads still use GetLock, and every successful write
	// invalidates this key before later planning can consume the new state.
	oldRecord, err := ps.cachedFindOne(ctx, objectName, id)
	if err != nil {
		return nil, appErrors.NewInternalError("failed to load record", err)
	}
	if oldRecord == nil {
		return nil, appErrors.NewNotFoundError("Record", "")
	}
	if constants.GetSecurityContext(ctx) < constants.SystemContext &&
		!constants.IsScopedInternalAccess(ctx, objectName, constants.PermEdit) {
		if !ps.permissions.CheckObjectPermissionOnRecord(ctx, schema, oldRecord, constants.PermEdit, currentUser) {
			return nil, appErrors.NewNotFoundError("Record", "")
		}
		if !ps.permissions.CheckRecordAccess(ctx, schema, oldRecord, constants.PermEdit, currentUser) {
			return nil, appErrors.NewNotFoundError("Record", "")
		}
	}
	return oldRecord, nil
}

func exclusiveSiblingTruthy(v interface{}) bool {
	switch t := v.(type) {
	case bool:
		return t
	case int:
		return t != 0
	case int64:
		return t != 0
	case float64:
		return t != 0
	case []byte:
		return exclusiveSiblingTruthy(string(t))
	case string:
		s := strings.TrimSpace(t)
		if s == "" {
			return false
		}
		if b, err := strconv.ParseBool(strings.ToLower(s)); err == nil {
			return b
		}
		return s == "1"
	default:
		return false
	}
}

func exclusiveSiblingRecordID(v interface{}) string {
	switch t := v.(type) {
	case string:
		return strings.TrimSpace(t)
	case []byte:
		return strings.TrimSpace(string(t))
	default:
		if t == nil {
			return ""
		}
		return strings.TrimSpace(fmt.Sprintf("%v", t))
	}
}