package records

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	"github.com/nexuscrm/audit"
	eventFeature "github.com/nexuscrm/events"
	"github.com/nexuscrm/kernel/pkg/apierr"
	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	infraPersistence "github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/events"
	"github.com/nexuscrm/shared/pkg/metadatarel"
	"github.com/nexuscrm/shared/pkg/models"
)

type recycleBinMutationTombstone struct {
	id                      string
	recordID                string
	objectName              string
	deletedByID             string
	deletedDate             string
	deleteOperationID       string
	deleteRootObjectName    string
	deleteRootRecordID      string
	cascadeParentObjectName string
	cascadeParentRecordID   string
	cascadeFieldAPIName     string
	cascadeDepth            int
}

type recycleBinRestoreParentTarget struct {
	fieldAPIName string
	key          recordlocks.ParentKey
}

type recycleBinRestoreChange struct {
	objectName string
	before     models.SObject
	after      models.SObject
}

type recycleBinRestoreFinalizingKey struct{}

func withRecycleBinRestoreFinalizing(ctx context.Context) context.Context {
	return context.WithValue(ctx, recycleBinRestoreFinalizingKey{}, true)
}

func recycleBinRestoreFinalizing(ctx context.Context) bool {
	restoring, _ := ctx.Value(recycleBinRestoreFinalizingKey{}).(bool)
	return restoring
}

// RestoreRecord restores the latest recycle-bin tombstone for recordID.
//
// Authority is deliberately split:
//   - `_System_RecycleBin` owns delete-event visibility and object identity.
//   - PermissionService owns object/per-record Edit authority.
//   - Object/field relationship metadata owns parent-lifecycle validation.
//
// The tombstone is the restore-specific visibility gate. After the deleted row
// is locked, Edit authority is revalidated through loadAndValidateAccess, the
// same record-aware write gate used by normal updates.
func (ps *PersistenceService) RestoreRecord(ctx context.Context, recordID string, currentUser *models.UserSession) (err error) {
	recordID = strings.TrimSpace(recordID)
	if recordID == "" {
		return apierr.Invalid("record id is required")
	}
	if currentUser == nil {
		return apierr.Invalid("user session required")
	}
	if ps == nil || ps.repo == nil || ps.metadata == nil || ps.permissions == nil {
		return apierr.Internal(fmt.Errorf("recycle-bin restore: persistence service is not fully wired"))
	}

	tombstone, err := ps.loadRecycleBinTombstone(ctx, recordID)
	if err != nil {
		return err
	}

	// Rejected-save audit (#4261). Registered only once the tombstone has
	// resolved the object name — before that there is no object to attribute a
	// rejection to, and the failures above are malformed-request/wiring
	// errors, not validation rejections (classifyRejectedSave would drop them
	// anyway). Restore re-enters the reactivation constraints and the same
	// uniqueness check as a normal save, so it rejects for the same reasons.
	//
	// A restore has no caller payload — the user supplied a record id, not
	// fields — so requestedFields is nil rather than a synthesized set.
	ctx = withRejectedSaveClaim(ctx)
	defer func() {
		if err == nil {
			return
		}
		ps.emitRejectedSaveAudit(ctx, tombstone.objectName, recordID,
			audit.ActionUndelete, nil, err, currentUser)
	}()
	schema := ps.metadata.GetSchema(ctx, tombstone.objectName)
	if !isRecycleBinManagedObject(schema) {
		return recycleBinNotFound()
	}
	if !ps.currentUserCanSeeRecycleBinTombstone(ctx, tombstone, currentUser) {
		return recycleBinNotFound()
	}
	if err := ps.checkOperationPermission(ctx, schema, schema.APIName, constants.PermEdit, currentUser); err != nil {
		return err
	}
	if !constants.GetSecurityContext(ctx).BypassesCRUDandFLS() {
		if err := rejectToolingOnlyTableWrites(tombstone.objectName); err != nil {
			return err
		}
	}

	restoreTombstones, err := ps.loadRecycleBinCascadeRestoreTombstones(ctx, tombstone)
	if err != nil {
		return err
	}
	transactionObjectNames := recycleBinTombstoneObjectNames(restoreTombstones)
	for _, objectName := range transactionObjectNames {
		if err := ps.validateRecordCacheInvalidationConfiguration(ctx, objectName); err != nil {
			return err
		}
	}

	var restoreChanges []recycleBinRestoreChange
	forcedParentPrelocks := parentLockForcedPrelocksFromContext(ctx)
	if lineageKeys := recycleBinCascadeLineageParentLockKeys(restoreTombstones); len(lineageKeys) > 0 {
		forcedParentPrelocks, _ = mergeParentLockKeys(forcedParentPrelocks, lineageKeys)
	}
	maxAttempts := len(restoreTombstones) + 1
	if maxAttempts < 3 {
		maxAttempts = 3
	}
	for attempt := 0; attempt < maxAttempts; attempt++ {
		err = ps.runGroupMembershipAwarePessimisticTransactionForObjects(ctx, transactionObjectNames, func(txCtx context.Context) error {
			txCtx = withParentLockForcedPrelocks(txCtx, forcedParentPrelocks)
			changes, err := ps.restoreRecycleBinTombstonesInTransaction(txCtx, ctx, recordID, tombstone, currentUser)
			if err != nil {
				return err
			}
			restoreChanges = changes
			return nil
		})
		if err == nil {
			break
		}
		retryKeys, shouldRetry := parentLockPrelockRetryKeys(err)
		if !shouldRetry || !ps.canRetryParentLockPrelock(ctx) {
			break
		}
		var changed bool
		forcedParentPrelocks, changed = mergeParentLockKeys(forcedParentPrelocks, retryKeys)
		if !changed {
			break
		}
	}
	if err != nil {
		return err
	}

	if !permissionAuthoringAggregateOwnsRecordsConvergence(ctx) {
		for _, change := range restoreChanges {
			ps.invalidateSecurityCaches(ctx, change.objectName, change.before, change.after)
		}
	}
	return nil
}

func (ps *PersistenceService) restoreRecycleBinTombstonesInTransaction(
	txCtx context.Context,
	rootCtx context.Context,
	requestedRecordID string,
	initialRoot recycleBinMutationTombstone,
	currentUser *models.UserSession,
) (changes []recycleBinRestoreChange, err error) {
	// Rejected-save audit (#4261): attribute a cascade failure to the record
	// that actually failed.
	//
	// A restore is one user action but many record lifecycles: the loop below
	// runs each cascade tombstone's update in this same transaction, and any
	// child's validation or uniqueness failure returns straight out of here.
	// RestoreRecord's hook only knows the ROOT tombstone, so without this the
	// child's rule and field identity would be recorded against the parent —
	// the same misattribution the nested-write claim prevents for Insert, but
	// reached a different way, because no child restore entry point exists to
	// claim it.
	//
	// Claiming here means RestoreRecord's outer hook sees the error already
	// recorded and skips it, so the row count stays one per rejected restore.
	var failing *recycleBinMutationTombstone
	defer func() {
		if err == nil || failing == nil {
			return
		}
		ps.emitRejectedSaveAudit(txCtx, failing.objectName, failing.recordID,
			audit.ActionUndelete, nil, err, currentUser)
	}()
	currentRoot, err := ps.loadRecycleBinTombstoneForMutation(txCtx, requestedRecordID, initialRoot.objectName, currentUser, "restore")
	if err != nil {
		return nil, err
	}
	restoreTombstones, err := ps.loadRecycleBinCascadeRestoreTombstones(txCtx, currentRoot)
	if err != nil {
		return nil, err
	}
	lockedTombstones, err := ps.lockRecycleBinTombstonesForMutation(txCtx, restoreTombstones, currentUser, "restore")
	if err != nil {
		return nil, err
	}
	restoreTombstones, err = recycleBinLockedTombstonesInOrder(restoreTombstones, lockedTombstones, "restore")
	if err != nil {
		return nil, err
	}
	parentPrelocks, err := ps.prelockRecycleBinMutationParentRows(txCtx, restoreTombstones)
	if err != nil {
		return nil, err
	}
	if err := ps.validateRecycleBinRestoreCascadeRecords(txCtx, restoreTombstones); err != nil {
		return nil, err
	}
	if err := ps.validateRecycleBinRestoreCascadeCompleteness(txCtx, restoreTombstones); err != nil {
		return nil, err
	}

	changes = make([]recycleBinRestoreChange, 0, len(restoreTombstones))
	for i := range restoreTombstones {
		lockedAuthority := restoreTombstones[i]
		// Everything from here on is attributable to THIS tombstone.
		failing = &restoreTombstones[i]
		schema := ps.metadata.GetSchema(txCtx, lockedAuthority.objectName)
		if !isRecycleBinManagedObject(schema) {
			return nil, recycleBinNotFound()
		}
		if err := ps.checkOperationPermission(txCtx, schema, schema.APIName, constants.PermEdit, currentUser); err != nil {
			return nil, err
		}
		if !constants.GetSecurityContext(txCtx).BypassesCRUDandFLS() {
			if err := rejectToolingOnlyTableWrites(lockedAuthority.objectName); err != nil {
				return nil, err
			}
		}

		lockedRecord, err := ps.loadAndValidateAccess(txCtx, lockedAuthority.objectName, lockedAuthority.recordID, schema, currentUser)
		if err != nil {
			return nil, err
		}
		if !isGuardFieldTruthy(lockedRecord[constants.FieldIsDeleted]) {
			return nil, apierr.Conflict("record is not deleted")
		}
		if err := validateRecycleBinCascadeRecordStillMatchesLineage(lockedAuthority, lockedRecord, "restore"); err != nil {
			return nil, err
		}
		if err := ps.validateRecycleBinRestoreMutation(txCtx, lockedAuthority.objectName, lockedRecord, currentUser); err != nil {
			return nil, err
		}

		restoreRootCtx := withRecycleBinRestoreFinalizing(withExclusiveSiblingRestoreNoReset(rootCtx))
		finalRecord, err := ps.completeRecycleBinRestoreUpdateLifecycle(txCtx, restoreRootCtx, lockedAuthority.objectName, lockedAuthority.recordID, schema, lockedRecord, currentUser, parentPrelocks)
		if err != nil {
			return nil, err
		}
		if preparation := permissionAuthoringRecordPreparationFromContext(rootCtx); preparation != nil {
			if preparation.update == nil || !preparation.matches(lockedAuthority.objectName, lockedAuthority.recordID) {
				return nil, apierr.Internal(fmt.Errorf("permission authoring restore did not produce its prepared Records update"))
			}
			preparation.update.restore = true
			preparation.update.recycleBinEntryID = lockedAuthority.id
			changes = append(changes, recycleBinRestoreChange{
				objectName: lockedAuthority.objectName,
				before:     lockedRecord,
				after:      finalRecord,
			})
			continue
		}
		if err := ps.repo.PhysicalDelete(txCtx, constants.TableRecycleBin, lockedAuthority.id); err != nil {
			return nil, apierr.Internal(fmt.Errorf("failed to remove recycle-bin entry after restore: %w", err))
		}
		if err := ps.emitUndeleteAuditV2(txCtx, lockedAuthority.objectName, lockedAuthority.recordID, currentUser); err != nil {
			return nil, err
		}
		changes = append(changes, recycleBinRestoreChange{
			objectName: lockedAuthority.objectName,
			before:     lockedRecord,
			after:      finalRecord,
		})
	}
	return changes, nil
}

func (ps *PersistenceService) completeRecycleBinRestoreUpdateLifecycle(
	txCtx context.Context,
	rootCtx context.Context,
	objectName, id string,
	schema *models.ObjectMetadata,
	oldRecord models.SObject,
	currentUser *models.UserSession,
	parentPrelocks map[string]struct{},
) (models.SObject, error) {
	restoreUpdates := models.SObject{constants.FieldIsDeleted: false}
	if preparation := permissionAuthoringRecordPreparationFromContext(rootCtx); preparation != nil &&
		preparation.matches(objectName, id) {
		for fieldName, value := range preparation.restoreUpdates {
			restoreUpdates[fieldName] = value
		}
	}
	if _, ok := filterToLogicalFields(schema, restoreUpdates)[constants.FieldIsDeleted]; !ok {
		return nil, apierr.Internal(fmt.Errorf("recycle-bin restore: %s does not expose a persistable %s field", objectName, constants.FieldIsDeleted))
	}

	recordToValidate := cloneSObject(oldRecord)
	effectiveUpdates := cloneSObject(restoreUpdates)
	for k, v := range restoreUpdates {
		recordToValidate[k] = v
	}

	if err := ps.publishRecordEvent(txCtx, events.RecordBeforeUpdate, objectName, recordToValidate, &oldRecord, currentUser); err != nil {
		return nil, err
	}

	recordToValidate = schemautil.NormalizeSObject(schema, recordToValidate)
	restorePolymorphicDiscriminatorsFromOldRecord(schema, recordToValidate, effectiveUpdates, oldRecord)
	if err := ps.runRecordWriteCanonicalizers(txCtx, recordWriteValidationBeforeRestore, recordWriteCanonicalizationEvent{
		ObjectName:       objectName,
		RecordID:         id,
		Record:           recordToValidate,
		OldRecord:        oldRecord,
		EffectiveUpdates: effectiveUpdates,
		CurrentUser:      currentUser,
	}); err != nil {
		return nil, err
	}
	ps.recalculateEffectiveUpdates(schema, recordToValidate, oldRecord, effectiveUpdates, true)

	// IsDeleted is the restore transition's internal finalization stamp.
	// Before-save metadata observes the restored image, but it must not erase
	// the platform-owned restore mutation before the shared update finalizer.
	recordToValidate[constants.FieldIsDeleted] = false
	effectiveUpdates[constants.FieldIsDeleted] = false

	if err := ps.runRecordWriteValidators(txCtx, recordWriteValidationBeforeRestore, recordWriteValidationEvent{
		ObjectName:  objectName,
		RecordID:    id,
		Record:      recordToValidate,
		OldRecord:   oldRecord,
		CurrentUser: currentUser,
	}); err != nil {
		return nil, err
	}

	if err := ps.validateRecycleBinRestoreReactivationImage(txCtx, objectName, id, schema, recordToValidate, oldRecord, effectiveUpdates, currentUser, parentPrelocks); err != nil {
		return nil, err
	}

	txCtxForUpdate, err := ps.stageRecycleBinRestoreOldActiveProjection(txCtx, schema, id, currentUser)
	if err != nil {
		return nil, err
	}
	return ps.completeUpdateAfterBeforeSave(txCtxForUpdate, rootCtx, objectName, id, schema, recordToValidate, effectiveUpdates, oldRecord, currentUser, nil, nil, parentPrelocks)
}

func (ps *PersistenceService) stageRecycleBinRestoreOldActiveProjection(
	ctx context.Context,
	schema *models.ObjectMetadata,
	id string,
	currentUser *models.UserSession,
) (context.Context, error) {
	if ps.activeFormulaProjector == nil {
		return ctx, nil
	}
	ctx = eventFeature.ClearActiveProjections(ctx)
	ctx = eventFeature.MarkActiveProjectionAttempted(ctx)
	projCtx := constants.WithSecurityContext(
		ctx, constants.SystemContext,
		"active-formula write-side projection: pre-restore deleted old record",
	)
	oldSnap, projErr := ps.activeFormulaProjector.projectIncludingDeleted(projCtx, currentUser, schema, id)
	if projErr != nil {
		return nil, appErrors.NewInternalError("active-formula projection (restore, old deleted record) failed", projErr)
	}
	updatedCtx, encErr := eventFeature.WithActiveProjections(ctx, nil, oldSnap)
	if encErr != nil {
		return nil, appErrors.NewInternalError("encode active-projection restore old-record snapshot failed", encErr)
	}
	return updatedCtx, nil
}

func (ps *PersistenceService) validateRecycleBinRestoreReactivationImage(
	txCtx context.Context,
	objectName, id string,
	schema *models.ObjectMetadata,
	recordToValidate models.SObject,
	storedRecord models.SObject,
	effectiveUpdates models.SObject,
	currentUser *models.UserSession,
	parentPrelocks map[string]struct{},
) error {
	storedImage := filterToLogicalFields(schema, storedRecord)
	storedImage[constants.FieldIsDeleted] = false
	finalImage := filterToLogicalFields(schema, recordToValidate)
	finalImage[constants.FieldIsDeleted] = false
	if err := ps.validator.ValidateRecordReactivationConstraintsWithStoredRecord(finalImage, storedImage, schema); err != nil {
		return err
	}
	if err := ps.validateRecordTypeId(txCtx, objectName, finalImage, currentUser); err != nil {
		return err
	}
	// A system-owned record's OwnerId=system is the platform's own rowless owner,
	// unchanged from creation — the restoring admin is not setting it. Validate it
	// as a platform write: a nil actor AND a narrowly SystemContext-elevated ctx,
	// so validateQueueAssignment's combined actor+SystemContext gate accepts the
	// unchanged owner without a `system` User row. A real-user-owned record keeps
	// the restoring admin's actor and context and validates its owner normally.
	ownerValidationActor := currentUser
	ownerValidationCtx := txCtx
	if ownerID, _ := finalImage[constants.FieldOwnerID].(string); ownerID == constants.SystemUserID {
		ownerValidationActor = nil
		ownerValidationCtx = constants.WithSecurityContext(txCtx, constants.SystemContext, "restore of system-owned record: revalidate unchanged platform owner")
	}
	if err := ps.validateQueueAssignment(ownerValidationCtx, objectName, finalImage, ownerValidationActor); err != nil {
		return err
	}
	// Required restore parents are prelocked on the post-before-save image so
	// metadata can repair a deleted pre-image before the undelete is judged.
	if err := ps.ensureRecycleBinRestoreParentRowsAlreadyLocked(txCtx, schema, finalImage, parentPrelocks); err != nil {
		return err
	}
	if err := ps.ensureNonPolymorphicLookupParentsPrelocked(txCtx, finalImage, schema, parentPrelocks); err != nil {
		return err
	}
	if err := ps.ensureExclusiveSiblingRestoreParentsAlreadyLocked(txCtx, schema, finalImage, parentPrelocks); err != nil {
		return err
	}
	if err := ps.validateExclusiveSiblingRestoreConflicts(txCtx, schema, id, finalImage); err != nil {
		return err
	}
	if err := ps.validateNonPolymorphicLookups(txCtx, finalImage, schema, currentUser); err != nil {
		return err
	}
	if _, err := ps.validatePolymorphicLookups(txCtx, finalImage, schema); err != nil {
		return err
	}
	if err := ps.validateStaticDependentPicklists(schema, finalImage, effectiveUpdates); err != nil {
		return err
	}
	if err := ps.validateRuntimeScopedPicklists(txCtx, schema, finalImage, nil); err != nil {
		return err
	}
	if err := ps.validateRuntimeScopedLookups(txCtx, schema, finalImage, nil); err != nil {
		return err
	}
	return ps.checkUniqueness(txCtx, objectName, finalImage, schema, id)
}

func (ps *PersistenceService) loadRecycleBinTombstone(ctx context.Context, recordID string) (recycleBinMutationTombstone, error) {
	row, err := ps.repo.FindRecycleBinEntry(ctx, recordID)
	if err != nil {
		return recycleBinMutationTombstone{}, apierr.Internal(fmt.Errorf("failed to load recycle-bin entry: %w", err))
	}
	return recycleBinMutationTombstoneFromRow(row)
}

func (ps *PersistenceService) loadRecycleBinTombstoneForMutation(ctx context.Context, recordID, expectedObjectName string, currentUser *models.UserSession, surface string) (recycleBinMutationTombstone, error) {
	currentTombstone, err := ps.loadRecycleBinTombstone(ctx, recordID)
	if err != nil {
		if isRecycleBinNotFoundError(err) {
			return recycleBinMutationTombstone{}, apierr.Conflict(fmt.Sprintf("recycle bin entry changed during %s", surface))
		}
		return recycleBinMutationTombstone{}, err
	}
	if !strings.EqualFold(currentTombstone.objectName, expectedObjectName) {
		return recycleBinMutationTombstone{}, apierr.Conflict(fmt.Sprintf("recycle bin entry changed during %s", surface))
	}
	if !ps.currentUserCanSeeRecycleBinTombstone(ctx, currentTombstone, currentUser) {
		return recycleBinMutationTombstone{}, recycleBinNotFound()
	}
	return currentTombstone, nil
}

func (ps *PersistenceService) lockRecycleBinTombstonesForMutation(ctx context.Context, tombstones []recycleBinMutationTombstone, currentUser *models.UserSession, surface string) (map[string]recycleBinMutationTombstone, error) {
	ordered := append([]recycleBinMutationTombstone(nil), tombstones...)
	sortRecycleBinRestoreTombstones(ordered)
	locked := make(map[string]recycleBinMutationTombstone, len(ordered))
	for _, expected := range ordered {
		key := recycleBinTombstoneTreeKey(expected.objectName, expected.recordID)
		if key == "\x00" {
			return nil, apierr.Conflict(fmt.Sprintf("cannot %s recycle-bin cascade tree because a candidate is missing object or record identity", surface))
		}
		currentTombstone, err := ps.loadRecycleBinTombstoneForMutation(ctx, expected.recordID, expected.objectName, currentUser, surface)
		if err != nil {
			return nil, err
		}
		lockedRow, err := ps.repo.GetLock(ctx, constants.TableRecycleBin, currentTombstone.id)
		if err != nil {
			return nil, apierr.Internal(fmt.Errorf("failed to lock recycle-bin entry for %s: %w", surface, err))
		}
		if lockedRow == nil {
			return nil, apierr.Conflict(fmt.Sprintf("recycle bin entry changed during %s", surface))
		}
		lockedAuthority, err := recycleBinMutationTombstoneFromRow(lockedRow)
		if err != nil {
			return nil, err
		}
		if !sameRecycleBinTombstoneLockCandidate(expected, lockedAuthority) {
			return nil, apierr.Conflict(fmt.Sprintf("recycle bin entry changed during %s", surface))
		}
		lockedAuthority = recycleBinTombstoneAuthorityForMutation(expected, lockedAuthority)
		locked[key] = lockedAuthority
	}
	return locked, nil
}

func sameRecycleBinTombstoneLockCandidate(expected, actual recycleBinMutationTombstone) bool {
	if !strings.EqualFold(expected.id, actual.id) ||
		!strings.EqualFold(expected.recordID, actual.recordID) ||
		!strings.EqualFold(expected.objectName, actual.objectName) {
		return false
	}
	if strings.TrimSpace(expected.deleteOperationID) == "" {
		return strings.TrimSpace(actual.deleteOperationID) == "" &&
			strings.TrimSpace(actual.deleteRootObjectName) == "" &&
			strings.TrimSpace(actual.deleteRootRecordID) == "" &&
			strings.TrimSpace(actual.cascadeParentObjectName) == "" &&
			strings.TrimSpace(actual.cascadeParentRecordID) == "" &&
			strings.TrimSpace(actual.cascadeFieldAPIName) == "" &&
			actual.cascadeDepth == 0
	}
	return sameRecycleBinRestoreCandidate(expected, actual)
}

func recycleBinTombstoneAuthorityForMutation(expected, actual recycleBinMutationTombstone) recycleBinMutationTombstone {
	if strings.TrimSpace(expected.deleteOperationID) == "" {
		actual.deleteOperationID = ""
		actual.deleteRootObjectName = ""
		actual.deleteRootRecordID = ""
		actual.cascadeParentObjectName = expected.cascadeParentObjectName
		actual.cascadeParentRecordID = expected.cascadeParentRecordID
		actual.cascadeFieldAPIName = expected.cascadeFieldAPIName
		actual.cascadeDepth = expected.cascadeDepth
	}
	return actual
}

func recycleBinLockedTombstonesInOrder(expected []recycleBinMutationTombstone, locked map[string]recycleBinMutationTombstone, surface string) ([]recycleBinMutationTombstone, error) {
	ordered := append([]recycleBinMutationTombstone(nil), expected...)
	sortRecycleBinRestoreTombstones(ordered)
	for i, tombstone := range ordered {
		key := recycleBinTombstoneTreeKey(tombstone.objectName, tombstone.recordID)
		lockedTombstone, ok := locked[key]
		if !ok {
			return nil, apierr.Conflict(fmt.Sprintf("recycle bin entry changed during %s", surface))
		}
		ordered[i] = lockedTombstone
	}
	return ordered, nil
}

func recycleBinMutationTombstoneFromRow(row models.SObject) (recycleBinMutationTombstone, error) {
	if row == nil {
		return recycleBinMutationTombstone{}, recycleBinNotFound()
	}
	t := recycleBinMutationTombstone{
		id:                      strings.TrimSpace(stringValue(row[constants.FieldSysRecycleBin_ID])),
		recordID:                strings.TrimSpace(stringValue(row[constants.FieldRecordID])),
		objectName:              strings.TrimSpace(stringValue(row[constants.FieldObjectAPIName])),
		deletedByID:             strings.TrimSpace(stringValue(row[constants.FieldSysRecycleBin_DeletedByID])),
		deletedDate:             strings.TrimSpace(stringValue(row[constants.FieldSysRecycleBin_DeletedDate])),
		deleteOperationID:       strings.TrimSpace(stringValue(row[constants.FieldSysRecycleBin_DeleteOperationID])),
		deleteRootObjectName:    strings.TrimSpace(stringValue(row[constants.FieldSysRecycleBin_DeleteRootObjectAPIName])),
		deleteRootRecordID:      strings.TrimSpace(stringValue(row[constants.FieldSysRecycleBin_DeleteRootRecordID])),
		cascadeParentObjectName: strings.TrimSpace(stringValue(row[constants.FieldSysRecycleBin_CascadeParentObjectAPIName])),
		cascadeParentRecordID:   strings.TrimSpace(stringValue(row[constants.FieldSysRecycleBin_CascadeParentRecordID])),
		cascadeFieldAPIName:     strings.TrimSpace(stringValue(row[constants.FieldSysRecycleBin_CascadeFieldAPIName])),
		cascadeDepth:            intValue(row[constants.FieldSysRecycleBin_CascadeDepth]),
	}
	if t.id == "" || t.recordID == "" || t.objectName == "" {
		return recycleBinMutationTombstone{}, recycleBinNotFound()
	}
	return t, nil
}

func intValue(v interface{}) int {
	switch x := v.(type) {
	case int:
		return x
	case int8:
		return int(x)
	case int16:
		return int(x)
	case int32:
		return int(x)
	case int64:
		return int(x)
	case uint:
		return int(x)
	case uint8:
		return int(x)
	case uint16:
		return int(x)
	case uint32:
		return int(x)
	case uint64:
		return int(x)
	case float32:
		return int(x)
	case float64:
		return int(x)
	case []byte:
		n, _ := strconv.Atoi(string(x))
		return n
	case string:
		n, _ := strconv.Atoi(strings.TrimSpace(x))
		return n
	default:
		return 0
	}
}

func recycleBinNotFound() *apierr.Error {
	return apierr.NotFound("recycle bin entry not found")
}

func (ps *PersistenceService) currentUserCanSeeRecycleBinTombstone(ctx context.Context, tombstone recycleBinMutationTombstone, user *models.UserSession) bool {
	if user == nil {
		return false
	}
	if tombstone.deletedByID != "" && strings.EqualFold(tombstone.deletedByID, user.ID) {
		return true
	}
	return ps.callerCanViewAllRecycleBin(ctx, user)
}

func (ps *PersistenceService) validateRecycleBinRestoreMutation(ctx context.Context, objectName string, oldRecord models.SObject, currentUser *models.UserSession) error {
	if constants.GetSecurityContext(ctx) < constants.SystemContext &&
		strings.EqualFold(objectName, constants.TableRecordShare) {
		if err := ps.checkRecordShareParentNotLocked(ctx, oldRecord, currentUser); err != nil {
			return err
		}
	}
	if constants.GetSecurityContext(ctx) < constants.SystemContext {
		if spec, ok := lookupProtectedSpec(objectName); ok && spec.GuardField != "" && isGuardFieldTruthy(oldRecord[spec.GuardField]) {
			return apierr.Forbidden(fmt.Sprintf("%s: %s", spec.ErrorCode, spec.Message))
		}
	}
	return nil
}

func (ps *PersistenceService) prelockRecycleBinMutationParentRows(
	ctx context.Context,
	tombstones []recycleBinMutationTombstone,
) (map[string]struct{}, error) {
	if recordlocks.ScopeFromContext(ctx) == nil {
		return map[string]struct{}{}, nil
	}
	keys := parentLockForcedPrelocksFromContext(ctx)
	readCtx := infraPersistence.WithPhysicalReadOnly(authoritativePermissionLockReadContext(ctx))
	for _, tombstone := range tombstones {
		relation, found, err := managedPermissionSetDefinitionRelation(tombstone.objectName)
		if err != nil {
			return nil, appErrors.NewInternalError("managed Permission Set definition inventory is unavailable", err)
		}
		if !found || relation.Role == schemacatalog.PermissionSetDefinitionRoleRoot {
			continue
		}
		record, err := ps.repo.FindAny(readCtx, tombstone.objectName, tombstone.recordID)
		if err != nil {
			return nil, appErrors.NewInternalError(
				fmt.Sprintf("failed to plan restore parent locks for %s/%s", tombstone.objectName, tombstone.recordID),
				err,
			)
		}
		if record == nil {
			continue
		}
		managedKeys, err := ps.managedPermissionSetDefinitionParentKeys(
			ctx,
			tombstone.objectName,
			contract.Record(record),
		)
		if err != nil {
			return nil, err
		}
		keys = append(keys, managedKeys...)
	}
	if len(keys) == 0 {
		return map[string]struct{}{}, nil
	}
	locked, err := ps.LockParentRows(ctx, keys)
	if err != nil {
		if appErrors.IsNotFound(err) {
			return nil, apierr.Conflict("cannot restore record because a required parent record is missing")
		}
		return nil, err
	}
	out := map[string]struct{}{}
	for _, key := range locked {
		out[recordlocks.KeyID(key)] = struct{}{}
	}
	return out, nil
}

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

func (ps *PersistenceService) collectRecycleBinRestoreParentTargets(ctx context.Context, schema *models.ObjectMetadata, record models.SObject) ([]recycleBinRestoreParentTarget, error) {
	if schema == nil || record == nil || ps == nil || ps.metadata == nil {
		return nil, nil
	}
	relationshipsByField := indexRelationshipsByParentAndField(ps.metadata.GetRelationships(ctx, schema.APIName))
	targets := make([]recycleBinRestoreParentTarget, 0)
	for i := range schema.Fields {
		field := &schema.Fields[i]
		if !isNonPolymorphicReference(field) || isValidatorSkippedField(field.APIName) {
			continue
		}
		parentObject := strings.TrimSpace(field.ReferenceTo[0])
		if parentObject == "" {
			continue
		}
		deleteSemantics := metadatarel.ResolveDeleteSemantics(*field, relationshipsByField[strings.ToLower(parentObject)+"|"+strings.ToLower(field.APIName)])
		required := field.Required || field.IsMasterDetail || deleteSemantics.Cascade
		if !required {
			continue
		}
		parentID := strings.TrimSpace(stringValue(record[field.APIName]))
		if parentID == "" {
			if field.Required || field.IsMasterDetail {
				return nil, apierr.Conflict(fmt.Sprintf("cannot restore record because required parent field %s is empty", field.APIName))
			}
			continue
		}
		targets = append(targets, recycleBinRestoreParentTarget{
			fieldAPIName: field.APIName,
			key:          recordlocks.ParentKey{ObjectName: parentObject, RecordID: parentID},
		})
	}
	return targets, nil
}