package records

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

	eventFeature "github.com/nexuscrm/events"
	"github.com/nexuscrm/kernel/pkg/apierr"
	"github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/kernel/pkg/utils"
	"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"
	zlog "github.com/rs/zerolog/log"
)

func indexRelationshipsByParentAndField(relationships []*models.Relationship) map[string]*models.Relationship {
	byField := make(map[string]*models.Relationship, len(relationships))
	for _, rel := range relationships {
		if rel == nil || rel.ParentObjectAPIName == "" || rel.FieldAPIName == "" {
			continue
		}
		key := strings.ToLower(rel.ParentObjectAPIName) + "|" + strings.ToLower(rel.FieldAPIName)
		byField[key] = rel
	}
	return byField
}

type recordHardDeleteTargetKey struct{}

type recordHardDeleteTarget struct {
	objectName string
	recordID   string
}

// This private storage disposition is scoped to one exact row. It does not
// bypass authorization, validation, parent locking, or record lifecycle hooks.
func withRecordHardDeleteTarget(ctx context.Context, objectName, recordID string) context.Context {
	return context.WithValue(ctx, recordHardDeleteTargetKey{}, recordHardDeleteTarget{
		objectName: strings.TrimSpace(objectName),
		recordID:   strings.TrimSpace(recordID),
	})
}

func recordHardDeleteAllowed(ctx context.Context, objectName, recordID string) bool {
	target, ok := ctx.Value(recordHardDeleteTargetKey{}).(recordHardDeleteTarget)
	return ok && strings.EqualFold(target.objectName, strings.TrimSpace(objectName)) && target.recordID == strings.TrimSpace(recordID)
}

// Delete applies the owning lifecycle's delete disposition with ACID guarantees.
func (ps *PersistenceService) Delete(
	ctx context.Context,
	objectName string,
	id string,
	currentUser *models.UserSession,
) error {
	denied := func() error {
		return apierr.PermissionDenied(fmt.Sprintf("permission denied: cannot delete %s/%s", objectName, id))
	}
	return ps.deleteWithOutcomes(ctx, objectName, id, currentUser, nil, denied, denied)
}

// deleteMetadataDeployRecord applies the normal locked-row delete pipeline but
// permits a hard delete only for the exact target. The cascade engine may derive
// another exact target only while traversing a metadata-declared Cascade edge;
// Restrict edges and every child's own delete policy still fail closed.
func (ps *PersistenceService) deleteMetadataDeployRecord(
	ctx context.Context,
	objectName string,
	id string,
	requiredMatch map[string]string,
	currentUser *models.UserSession,
) error {
	if ps == nil || ps.metadata == nil {
		return errors.NewInternalError("metadata deploy delete metadata unavailable", nil)
	}
	schema := ps.metadata.GetSchema(ctx, objectName)
	if !metadataDeployMatchIncludesUniqueField(schema, requiredMatch) {
		return apierr.Invalid(fmt.Sprintf("metadata deploy delete for %s requires a unique match field", objectName))
	}

	missing := func() error { return errors.NewNotFoundError(objectName, id) }
	denied := func() error {
		return apierr.Forbidden(fmt.Sprintf("permission denied: cannot delete %s/%s", objectName, id))
	}
	ctx = withRecordHardDeleteTarget(ctx, objectName, id)
	return ps.deleteWithOutcomesAuthorized(ctx, objectName, id, currentUser, missing, denied, denied, func(_ context.Context, locked models.SObject) error {
		if !metadataDeployLockedRecordIsLive(schema, locked) {
			return missing()
		}
		if !metadataDeployLockedRecordMatches(locked, requiredMatch) {
			return apierr.Conflict(fmt.Sprintf("%s %s no longer matches metadata deploy delete", objectName, id))
		}
		return nil
	})
}

// deleteWithOutcomes keeps one delete implementation and one authorization
// path while allowing the owning use-case adapter to project missing, object
// CRUD denial, and record-visibility denial into its own neutral outcomes.
// Nil missingRecord preserves the legacy idempotent-delete behavior.
func (ps *PersistenceService) deleteWithOutcomes(
	ctx context.Context,
	objectName string,
	id string,
	currentUser *models.UserSession,
	missingRecord func() error,
	objectDenied func() error,
	recordDenied func() error,
) error {
	return ps.deleteWithOutcomesAuthorized(ctx, objectName, id, currentUser, missingRecord, objectDenied, recordDenied, nil)
}

// deleteWithOutcomesAuthorized evaluates every root-row-dependent delete
// decision after the normal delete pipeline has acquired the record lock. The
// initial unlocked read exists for parent-first lock planning and established
// generic-CRUD fail-fast gates; it is never Runtime policy evidence. authorize
// is the use-case-owned extension for additional metadata policy and receives
// the same locked row image used by validators, events, cascades, and DML.
func (ps *PersistenceService) deleteWithOutcomesAuthorized(
	ctx context.Context,
	objectName string,
	id string,
	currentUser *models.UserSession,
	missingRecord func() error,
	objectDenied func() error,
	recordDenied func() error,
	authorize func(context.Context, models.SObject) error,
) error {
	reqID := requestIDFromCtx(ctx)
	zlog.Info().Str("request_id", reqID).Str("object", objectName).Str("record_id", id).Str("user", getUserID(currentUser)).Msg("PersistenceService.Delete")

	schema, err := ps.prepareOperation(ctx, objectName, constants.PermDelete, currentUser)
	if err != nil {
		return err
	}
	ctx, err = permissionSetRootDeleteContext(ctx, objectName, id)
	if err != nil {
		return err
	}

	if !constants.GetSecurityContext(ctx).BypassesCRUDandFLS() {
		if err := rejectToolingOnlyTableWrites(objectName); err != nil {
			return err
		}
	}

	// This read is planning evidence only. Delete must acquire relationship
	// parents before the child row to preserve the global lock order, then verify
	// the planned parent set against the current row after locking it.
	record, err := ps.repo.FindOne(ctx, objectName, id)
	if err != nil {
		return err
	}

	if record == nil {
		if missingRecord != nil {
			return missingRecord()
		}
		return nil // Already deleted or not found on the legacy seam.
	}
	// OwnerPack is an unbypassable generic-CRUD immutable fact. Keep this
	// ownership-surface rejection before transaction setup so malformed generic
	// metadata writes continue to fail fast; mutable authorization and policy
	// state is re-evaluated from the locked row below.
	if err := rejectManagedEmailTemplateGenericMutation(objectName, record, "deleted"); err != nil {
		return err
	}

	// _System_RecordShare lock gate: a user may not remove a share entry from a
	// record that is currently locked. This established cross-row gate remains a
	// preflight until its parent-lock protocol is owned explicitly; do not treat
	// the locked share row below as proof of the parent record's state.
	if constants.GetSecurityContext(ctx) < constants.SystemContext &&
		strings.EqualFold(objectName, constants.TableRecordShare) {
		if err := ps.checkRecordShareParentNotLocked(ctx, record, currentUser); err != nil {
			return err
		}
	}
	lineageObjectName := objectName
	if schema != nil && strings.TrimSpace(schema.APIName) != "" {
		lineageObjectName = schema.APIName
	}
	ctx = withRecycleBinRootDeleteLineage(ctx, lineageObjectName, id)
	if err := ps.validateRecordCacheInvalidationConfiguration(ctx, objectName); err != nil {
		return err
	}

	// Execute Transactional Work
	err = ps.runRollupAwareDeleteTransaction(ctx, objectName, id, func(txCtx context.Context) error {
		if err := requireInitialHistoryBusinessRowPrelocked(txCtx, objectName, id); err != nil {
			return err
		}
		if err := ps.prelockDeleteParentRows(txCtx, objectName, id, record); err != nil {
			return err
		}
		locked, err := ps.repo.GetLock(txCtx, objectName, id)
		if err != nil {
			return errors.NewInternalError("failed to lock record for delete", err)
		}
		if locked == nil {
			if missingRecord != nil {
				return missingRecord()
			}
			return nil
		}
		record = locked
		if err := ps.ensureDeleteParentRowsAlreadyLocked(txCtx, objectName, id, record); err != nil {
			return err
		}
		if err := ps.authorizeLockedDelete(txCtx, schema, objectName, record, currentUser, objectDenied, recordDenied); err != nil {
			return err
		}
		if authorize != nil {
			if err := authorize(txCtx, cloneSObject(record)); err != nil {
				return err
			}
		}
		if recordHardDeleteAllowed(txCtx, objectName, id) && !metadataDeployLockedRecordIsLive(schema, record) {
			return apierr.Conflict(fmt.Sprintf("%s %s is no longer live for metadata deploy delete", objectName, id))
		}

		if err := ps.checkRestrictChildRelationships(txCtx, schema, objectName, id); err != nil {
			return err
		}

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

		// Publish beforeDelete event
		if err := ps.publishRecordEvent(txCtx, events.RecordBeforeDelete, objectName, record, nil, currentUser); err != nil {
			return err
		}

		// Cascade Delete Children (Application-Level)
		if err := ps.cascadeDeleteChildren(txCtx, currentUser, objectName, id); err != nil {
			return errors.NewInternalError("failed to cascade delete children", err)
		}

		// Apply SetNull lookup semantics for soft deletes. Hard deletes may also
		// have physical FK actions, but clearing first keeps runtime behavior
		// aligned with the metadata source of truth.
		if err := ps.setNullRelationshipChildren(txCtx, objectName, id, currentUser); err != nil {
			return errors.NewInternalError("failed to clear set-null child relationships", err)
		}
		if err := ps.cascadePermissionSetOwnedReferences(txCtx, currentUser, objectName, id); err != nil {
			return err
		}

		var permissionStateAfter models.SObject
		if schema != nil && (recordHardDeleteAllowed(txCtx, objectName, id) || strings.EqualFold(strings.TrimSpace(schema.DeleteMode), constants.DeleteModeHard)) {
			// Hard delete: permanently remove the record, no recycle bin entry
			if err := ps.repo.PhysicalDelete(txCtx, objectName, id); err != nil {
				return wrapRepositoryMutationError("hard delete failed", err)
			}
			InvalidateRecordReadCache(txCtx, objectName, id)
		} else if schemaFieldExists(schema, constants.FieldIsDeleted) {
			// Soft delete within transaction.
			softDeleteUpdate := models.SObject{constants.FieldIsDeleted: 1}
			if err := ps.repo.Update(txCtx, objectName, id, softDeleteUpdate); err != nil {
				return wrapRepositoryMutationError("delete failed", err)
			}
			InvalidateRecordReadCache(txCtx, objectName, id)

			// Add to recycle bin within transaction
			recordName := ps.getRecordName(record, schema)
			// DeletedBy is the delete-time display-name snapshot (rendered on the
			// wire); DeletedById is the stable deleter identity that scope=mine
			// keys on (pingcap-inc/NexusCRM#1750). Both come from the same actor:
			// the current user, or the system actor for context-less deletes.
			deletedBy := constants.SystemUserName
			deletedByID := constants.SystemUserID
			if currentUser != nil {
				deletedBy = currentUser.Name
				deletedByID = currentUser.ID
			}

			binID := fmt.Sprintf("%s-%d", id, time.Now().UnixNano())
			binRecord := models.SObject{
				constants.FieldID:                        binID,
				constants.FieldRecordID:                  id,
				constants.FieldObjectAPIName:             objectName,
				constants.FieldRecordName:                recordName,
				constants.FieldDeletedBy:                 deletedBy,
				constants.FieldSysRecycleBin_DeletedByID: deletedByID,
				constants.FieldDeletedDate:               utils.NowTimestamp(),
				constants.FieldCreatedDate:               utils.NowTimestamp(),
				constants.FieldLastModifiedDate:          utils.NowTimestamp(),
			}
			for fieldName, value := range recycleBinLineageFieldsFromContext(txCtx) {
				binRecord[fieldName] = value
			}

			// Use Repo for Recycle Bin Insert
			if err := ps.repo.Insert(txCtx, constants.TableRecycleBin, binRecord); err != nil {
				return errors.NewInternalError("failed to add to recycle bin", err)
			}
			permissionStateAfter = cloneSObject(record)
			permissionStateAfter[constants.FieldIsDeleted] = 1
		} else {
			return apierr.Invalid(fmt.Sprintf("object %s does not support deletion", objectName))
		}
		if err := ps.requestPermissionStateAdvanceForDelete(txCtx, objectName, record, permissionStateAfter); err != nil {
			return errors.NewInternalError("permission state advance request failed", err)
		}
		if err := ps.runRecordLifecycleHooks(txCtx, recordLifecycleAfterDeleteInTx, recordLifecycleEvent{
			ObjectName:  objectName,
			RecordID:    id,
			OldRecord:   cloneSObject(record),
			CurrentUser: currentUser,
		}); err != nil {
			return err
		}
		if err := ps.markGroupMembershipClosureDirtyAfterMutation(txCtx, objectName, groupMembershipClosureMutationDelete, nil); err != nil {
			return errors.NewInternalError("group membership closure invalidation failed", err)
		}

		// When deleting a ConnectedApp, revoke all sessions AND refresh tokens for
		// that app so nothing issued under it survives the deletion.
		if strings.EqualFold(objectName, constants.TableConnectedApp) {
			if ps.sessionRepo != nil {
				if _, revokeErr := ps.sessionRepo.RevokeAllSessionsForConnectedApp(txCtx, id); revokeErr != nil {
					return errors.NewInternalError("failed to revoke sessions for deleted ConnectedApp", revokeErr)
				}
			}
			if err := ps.revokeConnectedAppRefreshTokens(txCtx, id); err != nil {
				return err
			}
			ps.invalidateConnectedAppCache(txCtx, id)
		}

		// When deleting a ConnectedAppAccess grant, revoke sessions for the parent app.
		if strings.EqualFold(objectName, constants.TableConnectedAppAccess) && ps.sessionRepo != nil {
			if appID, ok := record[constants.ColSysConnectedAppAccess_ConnectedAppID].(string); ok && appID != "" {
				if _, revokeErr := ps.sessionRepo.RevokeAllSessionsForConnectedApp(txCtx, appID); revokeErr != nil {
					return errors.NewInternalError("failed to revoke sessions after ConnectedAppAccess deletion", revokeErr)
				}
				if err := ps.revokeConnectedAppRefreshTokens(txCtx, appID); err != nil {
					return err
				}
			}
		}

		// Hook: Rollup Summary (inside transaction for ACID compliance)
		if err := ps.rollup.ProcessRollups(txCtx, objectName, record); err != nil {
			log.Printf("⚠️ Failed to process rollups for deleted record %s/%s: %v", objectName, id, err)
			return wrapRollupHookError("failed to process rollups", err)
		}

		// C11: emit DELETE AuditEvent via V2 pipeline. Gated by
		// the complete audit v2 fan-out path is ready. No-op outside a tx or when id is missing.
		// The outbox worker expands this head row into a single
		// `__deleted__` sentinel row so `<Object>History` surfaces a
		// DELETE entry (§3.6, §5.1). Emitted inside the same business
		// tx as the physical/soft delete for §3.3 CH-3 atomicity.
		if err := ps.emitDeleteAuditV2(txCtx, objectName, id, currentUser); err != nil {
			return err
		}

		// Enqueue afterDelete event to outbox (inside transaction for guaranteed delivery)
		if ps.outbox != nil {
			if err := ps.enqueueRecordEvent(txCtx, events.RecordDeleted, eventFeature.RecordEventPayload{
				ObjectAPIName: objectName,
				Record:        record,
				CurrentUser:   currentUser,
			}); err != nil {
				return errors.NewInternalError("failed to enqueue record deleted event", err)
			}
		}

		return ps.verifyPermissionSetDeleteClosure(txCtx, objectName, id)
	})

	if err != nil {
		return err
	}

	ps.invalidateSecurityCaches(ctx, objectName, record, nil)
	log.Printf("🗑️ Deleted record %s in %s (User: %s)", id, objectName, getUserID(currentUser))

	return nil
}

func (ps *PersistenceService) authorizeLockedDelete(
	ctx context.Context,
	schema *models.ObjectMetadata,
	objectName string,
	record models.SObject,
	currentUser *models.UserSession,
	objectDenied func() error,
	recordDenied func() error,
) error {
	if err := persistence.RejectGenericFieldMetadataRemoval(objectName, schema); err != nil {
		return err
	}

	// Retain the early generic-CRUD rejection, but re-evaluate it here so even a
	// privileged concurrent ownership transition cannot turn an unmanaged
	// preflight image into a managed row that generic delete removes.
	if err := rejectManagedEmailTemplateGenericMutation(objectName, record, "deleted"); err != nil {
		return err
	}

	// protectedRecords is a row-state guard, so only the locked image may decide
	// it. SystemContext remains the explicit repair/bootstrap bypass.
	if constants.GetSecurityContext(ctx) < constants.SystemContext {
		canonicalName := objectName
		if schema != nil && schema.APIName != "" {
			canonicalName = schema.APIName
		}
		if spec, ok := lookupProtectedSpec(canonicalName); ok && isGuardFieldTruthy(record[spec.GuardField]) {
			return apierr.Forbidden(fmt.Sprintf("%s: %s", spec.ErrorCode, spec.Message))
		}
	}

	// Record-aware object permission and RLS both depend on row state. Running
	// them before GetLock would let a concurrent OwnerId or policy-field update
	// invalidate the decision before delete commits.
	if schema != nil && constants.GetSecurityContext(ctx) < constants.SystemContext &&
		!constants.IsScopedInternalAccess(ctx, objectName, constants.PermDelete) {
		if !ps.permissions.CheckObjectPermissionOnRecord(ctx, schema, record, constants.PermDelete, currentUser) {
			return objectDenied()
		}
		if !ps.permissions.CheckRecordAccess(ctx, schema, record, constants.PermDelete, currentUser) {
			return recordDenied()
		}
	}
	return nil
}

func metadataDeployLockedRecordMatches(record models.SObject, requiredMatch map[string]string) bool {
	if requiredMatch == nil {
		return true
	}
	for fieldName, expected := range requiredMatch {
		actual, found := metadataDeployLockedRecordField(record, fieldName)
		if !found || actual == nil || fmt.Sprint(actual) != expected {
			return false
		}
	}
	return true
}

func metadataDeployLockedRecordIsLive(schema *models.ObjectMetadata, record models.SObject) bool {
	if !schemaFieldExists(schema, constants.FieldIsDeleted) {
		return true
	}
	isDeleted, found := metadataDeployLockedRecordField(record, constants.FieldIsDeleted)
	return found && !isGuardFieldTruthy(isDeleted)
}

func metadataDeployLockedRecordField(record models.SObject, fieldName string) (interface{}, bool) {
	for candidate, value := range record {
		if strings.EqualFold(strings.TrimSpace(candidate), strings.TrimSpace(fieldName)) {
			return value, true
		}
	}
	return nil, false
}

func metadataDeployMatchIncludesUniqueField(schema *models.ObjectMetadata, matchFields map[string]string) bool {
	if schema == nil || len(matchFields) == 0 {
		return false
	}
	for fieldName := range matchFields {
		if strings.EqualFold(strings.TrimSpace(fieldName), constants.FieldID) {
			return true
		}
		field := models.FindField(schema, fieldName)
		if field != nil && field.IsUnique {
			return true
		}
	}
	return false
}

func (ps *PersistenceService) prelockDeleteParentRows(ctx context.Context, objectName, id string, record models.SObject) error {
	if recordlocks.ScopeFromContext(ctx) == nil {
		return ps.prelockDeleteRollupParents(ctx, objectName, record)
	}
	keys, err := ps.collectDeleteParentLockKeys(ctx, objectName, id, record, map[string]struct{}{})
	if err != nil {
		return err
	}
	keys = append(keys, parentLockForcedPrelocksFromContext(ctx)...)
	if len(keys) == 0 {
		return nil
	}
	_, err = ps.lockInitialParentRows(ctx, keys)
	return err
}

func (ps *PersistenceService) collectDeleteParentLockKeys(ctx context.Context, objectName, id string, record models.SObject, visited map[string]struct{}) ([]recordlocks.ParentKey, error) {
	if ps == nil {
		return nil, nil
	}
	rowKey := recordlocks.ParentKey{ObjectName: objectName, RecordID: id}
	keyID := recordlocks.KeyID(rowKey)
	if keyID == "\x00" {
		return nil, nil
	}
	if _, ok := visited[keyID]; ok {
		return nil, nil
	}
	visited[keyID] = struct{}{}

	var keys []recordlocks.ParentKey
	managedDefinitionKeys, err := ps.managedPermissionSetDefinitionParentKeys(ctx, objectName, contract.Record(record))
	if err != nil {
		return nil, err
	}
	keys = append(keys, managedDefinitionKeys...)
	if ps.rollup != nil && len(record) > 0 {
		rollupKeys, err := ps.collectRollupParentKeysForRecords(ctx, objectName, record)
		if err != nil {
			return nil, err
		}
		keys = append(keys, rollupKeys...)
	}
	if ps.metadata == nil {
		return recordlocks.OrderedParentKeys(keys), nil
	}
	schema := ps.metadata.GetSchema(ctx, objectName)
	lookupTargets, err := ps.collectNonPolymorphicLookupLockTargetsForRecords(ctx, schema, record)
	if err != nil {
		return nil, err
	}
	lookupKeys, _ := orderedParentKeysForLookupTargetsFiltered(lookupTargets, true)
	keys = append(keys, lookupKeys...)
	exclusiveKeys, err := ps.collectExclusiveSiblingDeleteParentKeys(schema, record)
	if err != nil {
		return nil, err
	}
	keys = append(keys, exclusiveKeys...)

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

			children, err := ps.repo.GetChildren(ctx, childSchema.APIName, field.APIName, id)
			if err != nil {
				if strings.Contains(err.Error(), "doesn't exist") {
					continue
				}
				return nil, errors.NewInternalError(fmt.Sprintf("failed to query delete lock-plan children of %s via %s", childSchema.APIName, field.APIName), err)
			}
			for _, child := range children {
				childRecord, childID, ok, err := ps.loadDeleteLockPlanChild(ctx, childSchema.APIName, child)
				if err != nil {
					return nil, err
				}
				if !ok {
					continue
				}
				keys = append(keys, rowKey)
				if deleteSemantics.Cascade {
					childKeys, err := ps.collectDeleteParentLockKeys(ctx, childSchema.APIName, childID, childRecord, visited)
					if err != nil {
						return nil, err
					}
					keys = append(keys, childKeys...)
					continue
				}

				setNullKeys, err := ps.collectSetNullChildParentLockKeys(ctx, childSchema, childRecord, field.APIName)
				if err != nil {
					return nil, err
				}
				keys = append(keys, setNullKeys...)
			}
		}
	}
	return recordlocks.OrderedParentKeys(keys), nil
}

func (ps *PersistenceService) loadDeleteLockPlanChild(ctx context.Context, childObjectName string, child models.SObject) (models.SObject, string, bool, error) {
	childID, ok := child[constants.FieldID].(string)
	childID = strings.TrimSpace(childID)
	if !ok || childID == "" {
		return nil, "", false, nil
	}
	loaded, err := ps.repo.FindOne(ctx, childObjectName, childID)
	if err != nil {
		return nil, "", false, errors.NewInternalError(fmt.Sprintf("failed to load delete lock-plan child %s/%s", childObjectName, childID), err)
	}
	if loaded != nil {
		return loaded, childID, true, nil
	}
	if len(child) > 1 {
		return child, childID, true, nil
	}
	return nil, "", false, nil
}

func (ps *PersistenceService) collectSetNullChildParentLockKeys(ctx context.Context, childSchema *models.ObjectMetadata, child models.SObject, lookupFieldName string) ([]recordlocks.ParentKey, error) {
	finalChild := cloneSObject(child)
	if finalChild == nil {
		finalChild = models.SObject{}
	}
	finalChild[lookupFieldName] = nil
	if polymorphicTypeField := lookupFieldName + "_type"; schemaFieldExists(childSchema, polymorphicTypeField) {
		finalChild[polymorphicTypeField] = nil
	}

	targets, err := ps.collectNonPolymorphicLookupLockTargetsForRecords(ctx, childSchema, child, finalChild)
	if err != nil {
		return nil, err
	}
	keys, _ := orderedParentKeysForLookupTargets(targets)
	return keys, nil
}

func (ps *PersistenceService) ensureDeleteParentRowsAlreadyLocked(ctx context.Context, objectName, id string, record models.SObject) error {
	scope := recordlocks.ScopeFromContext(ctx)
	if scope == nil {
		return nil
	}
	keys, err := ps.collectDeleteParentLockKeys(ctx, objectName, id, record, map[string]struct{}{})
	if err != nil {
		return err
	}
	var missing []recordlocks.ParentKey
	for _, key := range keys {
		if !scope.HasAcquired(key) {
			missing = append(missing, key)
		}
	}
	if len(missing) > 0 {
		return &parentLockPrelockRetryError{keys: missing, reason: "delete parent"}
	}
	return nil
}

func (ps *PersistenceService) prelockDeleteRollupParents(ctx context.Context, objectName string, record models.SObject) error {
	if ps == nil || ps.rollup == nil || len(record) == 0 || recordlocks.ScopeFromContext(ctx) == nil {
		return nil
	}
	keys, err := ps.rollup.AffectedParentRows(ctx, objectName, record)
	if err != nil {
		return err
	}
	if len(keys) == 0 {
		return nil
	}
	_, err = ps.LockParentRows(ctx, keys)
	return err
}

func (ps *PersistenceService) checkRestrictChildRelationships(ctx context.Context, schema *models.ObjectMetadata, objectName, id string) error {
	return ps.checkLiveChildReferences(ctx, schema, objectName, id, true, false)
}

func (ps *PersistenceService) checkAnyLiveChildReference(ctx context.Context, schema *models.ObjectMetadata, objectName, id string) error {
	return ps.checkLiveChildReferences(ctx, schema, objectName, id, false, true)
}

func (ps *PersistenceService) checkLiveChildReferences(ctx context.Context, schema *models.ObjectMetadata, objectName, id string, restrictOnly, coveredEdgesOnly bool) error {
	if schema == nil {
		return nil
	}

	children := ps.metadata.GetChildRelationships(ctx, objectName)
	for _, childSchema := range children {
		if coveredEdgesOnly && !liveParentGuardSchemaHasCoveredWriter(childSchema) {
			continue
		}
		relationshipsByField := indexRelationshipsByParentAndField(ps.metadata.GetRelationships(ctx, childSchema.APIName))
		for _, field := range childSchema.Fields {
			if coveredEdgesOnly && !isNonPolymorphicReference(&field) {
				continue
			}
			if !metadatarel.FieldReferencesParent(field, objectName) {
				continue
			}
			relationship := relationshipsByField[strings.ToLower(objectName)+"|"+strings.ToLower(field.APIName)]
			if coveredEdgesOnly && !ps.lookupTargetRequiresParentLockWithRelationship(ctx, childSchema, &field, relationship) {
				continue
			}

			if restrictOnly {
				deleteSemantics := metadatarel.ResolveDeleteSemantics(field, relationship)
				if !strings.EqualFold(string(deleteSemantics.Rule), string(constants.DeleteRuleRestrict)) {
					continue
				}
			}

			var exists bool
			var err error
			if coveredEdgesOnly {
				exists, err = ps.liveChildReferenceExists(ctx, childSchema, field.APIName, objectName, id)
			} else {
				exists, err = ps.repo.ExistsByField(ctx, childSchema.APIName, field.APIName, id)
			}
			if err != nil {
				if persistence.IsSchemaDrift(err) {
					continue
				}
				return errors.NewInternalError(
					fmt.Sprintf("failed to check child records for %s.%s", childSchema.APIName, field.APIName),
					err,
				)
			}
			if exists {
				return apierr.Conflict(fmt.Sprintf("%s %s referenced by %s", objectName, id, childSchema.PluralLabel))
			}
		}
	}
	return nil
}

func liveParentGuardSchemaHasShape(schema *models.ObjectMetadata) bool {
	return models.FindField(schema, constants.FieldID) != nil &&
		models.FindField(schema, constants.FieldIsDeleted) != nil
}

func liveParentGuardSchemaHasCoveredWriter(schema *models.ObjectMetadata) bool {
	if schema == nil || !liveParentGuardSchemaHasShape(schema) {
		return false
	}
	if schema.ReadSource != nil {
		return false
	}
	return !constants.IsManagedSystemTable(schema.APIName)
}

func (ps *PersistenceService) liveChildReferenceExists(ctx context.Context, childSchema *models.ObjectMetadata, fieldName, parentObjectName, parentID string) (bool, error) {
	if childSchema != nil && strings.EqualFold(childSchema.APIName, parentObjectName) {
		children, err := ps.repo.GetChildren(ctx, childSchema.APIName, fieldName, parentID)
		if err != nil {
			return false, err
		}
		for _, child := range children {
			if !strings.EqualFold(strings.TrimSpace(stringValue(child[constants.FieldID])), strings.TrimSpace(parentID)) {
				return true, nil
			}
		}
		return false, nil
	}
	return ps.repo.ExistsByField(ctx, childSchema.APIName, fieldName, parentID)
}

// cascadeDeleteChildren finds and deletes child records that have DeleteRuleCascade
// DESIGN ASSUMPTION: DeleteRule values may be stored in mixed case (e.g., "CASCADE", "Cascade").
// Always use strings.EqualFold for DeleteRule comparisons to ensure case-insensitive matching.
// DESIGN ASSUMPTION: Object API names are case-insensitive (see ContainsStringIgnoreCase).
func (ps *PersistenceService) cascadeDeleteChildren(ctx context.Context, user *models.UserSession, parentObjName, parentID string) error {
	schemas := ps.metadata.GetSchemas(ctx)

	for _, schema := range schemas {
		relationshipsByField := indexRelationshipsByParentAndField(ps.metadata.GetRelationships(ctx, schema.APIName))
		for _, field := range schema.Fields {
			// Check if field references our parent object with Cascade delete
			// IMPORTANT: Use case-insensitive checks for both object names and delete rules
			isReference := metadatarel.FieldReferencesParent(field, parentObjName)

			deleteSemantics := metadatarel.ResolveDeleteSemantics(field, relationshipsByField[strings.ToLower(parentObjName)+"|"+strings.ToLower(field.APIName)])
			if isReference && deleteSemantics.Cascade {

				childObjName := schema.APIName
				lookupFieldName := field.APIName

				// Find IDs of children (excluding already deleted) via Repo
				// ACID: Use existing transaction if available
				children, err := ps.repo.GetChildren(ctx, childObjName, lookupFieldName, parentID)
				if err != nil {
					// Ignore table not found errors (race conditions or init issues)
					if strings.Contains(err.Error(), "doesn't exist") {
						continue
					}
					return errors.NewInternalError(fmt.Sprintf("failed to query children of %s via %s", childObjName, lookupFieldName), err)
				}

				var childIDs []string
				for _, child := range children {
					if id, ok := child[constants.FieldID].(string); ok {
						childIDs = append(childIDs, id)
					}
				}

				// Recursively delete children
				for _, childID := range childIDs {
					// Log for safety/debugging
					log.Printf("Cascade deleting child %s/%s because parent %s/%s was deleted", childObjName, childID, parentObjName, parentID)
					childCtx := withRecycleBinCascadeDeleteLineage(ctx, childObjName, childID, parentObjName, parentID, lookupFieldName)
					// A metadata deploy hard delete propagates only across this
					// authoritative Cascade edge. The child still runs the complete
					// locked-row delete pipeline, including its own Restrict checks,
					// but is physically removed without creating a recycle-bin row
					// that the parent's physical FK cascade would orphan.
					if recordHardDeleteAllowed(ctx, parentObjName, parentID) {
						childCtx = withRecordHardDeleteTarget(childCtx, childObjName, childID)
					}
					if err := ps.Delete(childCtx, childObjName, childID, user); err != nil {
						// Don't fail if already deleted (idempotency)
						if !errors.IsNotFound(err) {
							return errors.NewInternalError(fmt.Sprintf("failed to cascade delete child %s (%s)", childObjName, childID), err)
						}
					}
				}
			}
		}
	}
	return nil
}

func (ps *PersistenceService) setNullRelationshipChildren(ctx context.Context, parentObjName, parentID string, currentUser *models.UserSession) error {
	schemas := ps.metadata.GetSchemas(ctx)
	updateCtx := constants.WithSecurityContext(ctx, constants.SystemContext, "relationship_delete_set_null")

	for _, schema := range schemas {
		relationshipsByField := indexRelationshipsByParentAndField(ps.metadata.GetRelationships(ctx, schema.APIName))
		for _, field := range schema.Fields {
			if !metadatarel.FieldReferencesParent(field, parentObjName) {
				continue
			}

			deleteSemantics := metadatarel.ResolveDeleteSemantics(field, relationshipsByField[strings.ToLower(parentObjName)+"|"+strings.ToLower(field.APIName)])
			if !strings.EqualFold(string(deleteSemantics.Rule), string(constants.DeleteRuleSetNull)) {
				continue
			}

			childObjName := schema.APIName
			lookupFieldName := field.APIName
			children, err := ps.repo.GetChildren(ctx, childObjName, lookupFieldName, parentID)
			if err != nil {
				if strings.Contains(err.Error(), "doesn't exist") {
					continue
				}
				return errors.NewInternalError(fmt.Sprintf("failed to query set-null children of %s via %s", childObjName, lookupFieldName), err)
			}

			for _, child := range children {
				childID, ok := child[constants.FieldID].(string)
				if !ok || strings.TrimSpace(childID) == "" {
					continue
				}
				updates := models.SObject{lookupFieldName: nil}
				if err := ps.Update(updateCtx, childObjName, childID, updates, currentUser); err != nil {
					return errors.NewInternalError(fmt.Sprintf("failed to clear set-null relationship %s.%s for child %s", childObjName, lookupFieldName, childID), err)
				}
			}
		}
	}
	return nil
}

// lookupProtectedSpec returns the protectedRecords spec for a table name.
// The map in constants.ProtectedRecords is keyed by the canonical TableName
// from system_tables.json; this helper first tries an exact match (fast path,
// hit on every well-formed caller) and falls back to a case-insensitive scan
// so the guard is not bypassed by URL-case variance on the REST path.
func lookupProtectedSpec(tableName string) (constants.ProtectedRecordSpec, bool) {
	if spec, ok := constants.ProtectedRecords[tableName]; ok {
		return spec, true
	}
	for k, spec := range constants.ProtectedRecords {
		if strings.EqualFold(k, tableName) {
			return spec, true
		}
	}
	return constants.ProtectedRecordSpec{}, false
}

// isGuardFieldTruthy reports whether a protectedRecords guard-field value on a
// loaded record should be treated as "protected". Handles the mix of types the
// persistence layer returns for boolean-like columns (int, int64, bool, string).
func isGuardFieldTruthy(v interface{}) bool {
	switch x := v.(type) {
	case nil:
		return false
	case bool:
		return x
	case int:
		return x != 0
	case int32:
		return x != 0
	case int64:
		return x != 0
	case uint8:
		return x != 0
	case []byte:
		return len(x) == 1 && x[0] != 0
	case string:
		return x == "1" || strings.EqualFold(x, "true")
	default:
		return false
	}
}

// schemaFieldExists returns true if the schema has a field with the given API name (case-insensitive).
func schemaFieldExists(schema *models.ObjectMetadata, fieldName string) bool {
	if schema == nil {
		return false
	}
	for _, f := range schema.Fields {
		if strings.EqualFold(f.APIName, fieldName) {
			return true
		}
	}
	return false
}