package records

import (
	"context"
	"errors"
	"net/http"
	"strings"
	"testing"

	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/kernel/pkg/ports"
	metadataFeature "github.com/nexuscrm/metadata"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
)

type recycleBinPurgeRepoStub struct {
	ports.RecordRepository

	records    map[string]models.SObject
	tombstones map[string]models.SObject

	findRecycleBinEntryCalls       int
	tombstoneMissingAfterFirstFind bool
	getDeletedChildrenErr          error
	missingChildFieldColumns       map[string]struct{}

	physicalDeletes []string
	updates         []string
	lockLog         []string
}

func (r *recycleBinPurgeRepoStub) FindRecycleBinEntry(_ context.Context, recordID string) (models.SObject, error) {
	r.findRecycleBinEntryCalls++
	if r.tombstoneMissingAfterFirstFind && r.findRecycleBinEntryCalls > 1 {
		return nil, nil
	}
	if r.tombstones == nil {
		return nil, nil
	}
	return cloneSObject(r.tombstones[strings.ToLower(recordID)]), nil
}

func (r *recycleBinPurgeRepoStub) FindRecycleBinEntriesByDeleteOperation(_ context.Context, operationID string) ([]models.SObject, error) {
	var out []models.SObject
	for _, tombstone := range r.tombstones {
		if strings.EqualFold(stringValue(tombstone[constants.FieldSysRecycleBin_DeleteOperationID]), operationID) {
			out = append(out, cloneSObject(tombstone))
		}
	}
	return out, nil
}

func (r *recycleBinPurgeRepoStub) GetLock(_ context.Context, tableName string, id string) (models.SObject, error) {
	r.lockLog = append(r.lockLog, "lock:"+tableName+":"+id)
	if strings.EqualFold(tableName, constants.TableRecycleBin) {
		for _, tombstone := range r.tombstones {
			if strings.EqualFold(stringValue(tombstone[constants.FieldSysRecycleBin_ID]), id) {
				return cloneSObject(tombstone), nil
			}
		}
		return nil, nil
	}
	return cloneSObject(r.records[purgeRepoRecordKey(tableName, id)]), nil
}

func (r *recycleBinPurgeRepoStub) FindOne(_ context.Context, tableName string, id string) (models.SObject, error) {
	return cloneSObject(r.records[purgeRepoRecordKey(tableName, id)]), nil
}

func (r *recycleBinPurgeRepoStub) FindAny(ctx context.Context, tableName string, id string) (models.SObject, error) {
	return r.FindOne(ctx, tableName, id)
}

func (r *recycleBinPurgeRepoStub) GetDeletedChildren(_ context.Context, childTable string, foreignKey string, parentID string) ([]models.SObject, error) {
	if err := r.childFieldSchemaDriftError(childTable, foreignKey); err != nil {
		return nil, err
	}
	if r.getDeletedChildrenErr != nil {
		return nil, r.getDeletedChildrenErr
	}
	out := []models.SObject{}
	prefix := strings.ToLower(childTable) + "|"
	for key, record := range r.records {
		if !strings.HasPrefix(key, prefix) {
			continue
		}
		if !isGuardFieldTruthy(record[constants.FieldIsDeleted]) {
			continue
		}
		if !strings.EqualFold(strings.TrimSpace(stringValue(record[foreignKey])), parentID) {
			continue
		}
		out = append(out, cloneSObject(record))
	}
	return out, nil
}

func (r *recycleBinPurgeRepoStub) GetChildren(_ context.Context, childTable string, foreignKey string, parentID string) ([]models.SObject, error) {
	if err := r.childFieldSchemaDriftError(childTable, foreignKey); err != nil {
		return nil, err
	}
	out := []models.SObject{}
	prefix := strings.ToLower(childTable) + "|"
	for key, record := range r.records {
		if !strings.HasPrefix(key, prefix) {
			continue
		}
		if isGuardFieldTruthy(record[constants.FieldIsDeleted]) {
			continue
		}
		if !strings.EqualFold(strings.TrimSpace(stringValue(record[foreignKey])), parentID) {
			continue
		}
		out = append(out, cloneSObject(record))
	}
	return out, nil
}

func (r *recycleBinPurgeRepoStub) markChildFieldSchemaDrift(childTable, foreignKey string) {
	if r.missingChildFieldColumns == nil {
		r.missingChildFieldColumns = map[string]struct{}{}
	}
	r.missingChildFieldColumns[purgeChildFieldKey(childTable, foreignKey)] = struct{}{}
}

func (r *recycleBinPurgeRepoStub) childFieldSchemaDriftError(childTable, foreignKey string) error {
	if r == nil || len(r.missingChildFieldColumns) == 0 {
		return nil
	}
	if _, ok := r.missingChildFieldColumns[purgeChildFieldKey(childTable, foreignKey)]; !ok {
		return nil
	}
	return errors.New("Unknown column '" + foreignKey + "' in 'where clause'")
}

func purgeChildFieldKey(childTable, foreignKey string) string {
	return strings.ToLower(strings.TrimSpace(childTable)) + "|" + strings.ToLower(strings.TrimSpace(foreignKey))
}

func (r *recycleBinPurgeRepoStub) PhysicalDelete(_ context.Context, tableName string, id string) error {
	r.physicalDeletes = append(r.physicalDeletes, tableName+"|"+id)
	if strings.EqualFold(tableName, constants.TableRecycleBin) {
		for recordID, tombstone := range r.tombstones {
			if strings.EqualFold(stringValue(tombstone[constants.FieldSysRecycleBin_ID]), id) {
				delete(r.tombstones, recordID)
				break
			}
		}
	} else {
		delete(r.records, purgeRepoRecordKey(tableName, id))
	}
	return nil
}

func (r *recycleBinPurgeRepoStub) Update(_ context.Context, tableName string, id string, _ models.SObject) error {
	r.updates = append(r.updates, tableName+"|"+id)
	return nil
}

func purgeRepoRecordKey(tableName, id string) string {
	return strings.ToLower(tableName) + "|" + strings.ToLower(id)
}

func TestPurgeRecordPurgesLeafWithoutLifecycleUpdate(t *testing.T) {
	user := recordEntrypointUser()
	schema := restoreObjectSchema("Account")
	repo := newRecycleBinPurgeRepo("Account", "rec-1", user.ID)
	ps := newRecycleBinPurgeService(t, repo, true, schema)

	if err := ps.PurgeRecord(context.Background(), "rec-1", user); err != nil {
		t.Fatalf("PurgeRecord: %v", err)
	}
	if got, want := strings.Join(repo.physicalDeletes, ","), "Account|rec-1,"+constants.TableRecycleBin+"|rb-rec-1"; got != want {
		t.Fatalf("physical deletes = %s, want %s", got, want)
	}
	if len(repo.updates) != 0 {
		t.Fatalf("PurgeRecord called Update/lifecycle path: %#v", repo.updates)
	}
}

func TestPurgeRecordPermissionSetAssignmentAllowsRetiredTarget(t *testing.T) {
	user := recordEntrypointUser()
	assignmentSchema := restoreObjectSchema(constants.TablePermissionSetAssignment)
	assignmentSchema.Fields = append(
		assignmentSchema.Fields,
		models.FieldMetadata{
			APIName:    constants.FieldPermissionSetAssignment_AssigneeID,
			Type:       constants.FieldTypeString,
			Createable: true,
			Updateable: true,
		},
		models.FieldMetadata{
			APIName:    constants.FieldPermissionSetAssignment_PermissionSetID,
			Type:       constants.FieldTypeString,
			Createable: true,
			Updateable: true,
		},
	)
	assignmentSchema.AccessPolicy = permissionSetAssignmentWriteValidationAccessPolicy()
	makeFieldsMap(assignmentSchema)
	permissionSetSchema := restoreObjectSchema(constants.TablePermissionSet)

	repo := newRecycleBinPurgeRepo(constants.TablePermissionSetAssignment, "psa-retired", user.ID)
	repo.records[purgeRepoRecordKey(constants.TablePermissionSetAssignment, "psa-retired")][constants.FieldPermissionSetAssignment_AssigneeID] = "assignee-retired"
	repo.records[purgeRepoRecordKey(constants.TablePermissionSetAssignment, "psa-retired")][constants.FieldPermissionSetAssignment_PermissionSetID] = "ps-retired"
	repo.records[purgeRepoRecordKey(constants.TablePermissionSet, "ps-retired")] = models.SObject{
		constants.FieldID: "ps-retired",
		constants.FieldPermissionSet_IsOwnedByProfile: false,
		constants.FieldPermissionSet_IsActive:         false,
		constants.FieldIsDeleted:                      true,
	}
	ps := newRecycleBinPurgeService(t, repo, true, permissionSetSchema, assignmentSchema)
	ps.parentLaneAdmission = &fakeParentLaneAdmissionController{}
	ps.registerBuiltInSecurityWriteValidators()
	installNoOpPermissionStateAuthority(ps)

	if err := ps.PurgeRecord(context.Background(), "psa-retired", user); err != nil {
		t.Fatalf("PurgeRecord PermissionSetAssignment with retired target: %v", err)
	}
	if got, want := strings.Join(repo.physicalDeletes, ","), constants.TablePermissionSetAssignment+"|psa-retired,"+constants.TableRecycleBin+"|rb-psa-retired"; got != want {
		t.Fatalf("physical deletes = %s, want %s", got, want)
	}
	assertLockBefore(
		t,
		repo.lockLog,
		"lock:"+constants.TablePermissionSet+":ps-retired",
		"lock:"+constants.TablePermissionSetAssignment+":psa-retired",
	)
}

func TestPurgeRecordAllowsExclusiveSiblingLeafWithoutLifecycleUpdate(t *testing.T) {
	user := recordEntrypointUser()
	schema := restoreObjectSchema("LineItem__c")
	schema.Fields = append(schema.Fields,
		models.FieldMetadata{APIName: "ParentId", Type: constants.FieldTypeReference, ReferenceTo: []string{"Parent__c"}, Createable: true, Updateable: true},
		models.FieldMetadata{APIName: "IsPrimary__c", Type: constants.FieldTypeBoolean, Createable: true, Updateable: true},
	)
	schema.ExclusiveSiblingFields = []models.ExclusiveSiblingField{{
		FieldAPIName:          "IsPrimary__c",
		PartitionFieldAPIName: "ParentId",
	}}
	makeFieldsMap(schema)
	repo := newRecycleBinPurgeRepo("LineItem__c", "line-1", user.ID)
	repo.records[purgeRepoRecordKey("LineItem__c", "line-1")]["ParentId"] = "parent-1"
	repo.records[purgeRepoRecordKey("LineItem__c", "line-1")]["IsPrimary__c"] = true
	ps := newRecycleBinPurgeService(t, repo, true, schema)

	if err := ps.PurgeRecord(context.Background(), "line-1", user); err != nil {
		t.Fatalf("PurgeRecord exclusive sibling leaf: %v", err)
	}
	if got, want := strings.Join(repo.physicalDeletes, ","), "LineItem__c|line-1,"+constants.TableRecycleBin+"|rb-line-1"; got != want {
		t.Fatalf("physical deletes = %s, want %s", got, want)
	}
	if len(repo.updates) != 0 {
		t.Fatalf("PurgeRecord called Update/lifecycle path: %#v", repo.updates)
	}
}

func TestPurgeRecordRunsBeforePurgeWriteValidatorBeforePhysicalDelete(t *testing.T) {
	user := recordEntrypointUser()
	schema := restoreObjectSchema("PurgeProbe__c")
	schema.AccessPolicy = &models.ObjectAccessPolicy{
		WriteValidation: &models.RecordWriteValidationAccessPolicy{
			Validators: []models.RecordWriteValidationPolicy{{
				Phase:      string(contract.RecordWriteValidationBeforePurge),
				Capability: "test.purge.validate",
			}},
		},
	}
	repo := newRecycleBinPurgeRepo("PurgeProbe__c", "rec-1", user.ID)
	ps := newRecycleBinPurgeService(t, repo, true, schema)
	sentinel := errors.New("blocked by purge validator")
	var called bool

	if err := ps.RegisterRecordWriteValidator(contract.RecordWriteValidatorRegistration{
		Capability: "test.purge.validate",
		Phase:      contract.RecordWriteValidationBeforePurge,
		Validator: func(_ context.Context, event contract.RecordWriteValidationEvent) error {
			called = true
			if len(repo.physicalDeletes) != 0 {
				t.Fatalf("before_purge validator ran after physical delete: %#v", repo.physicalDeletes)
			}
			if event.ObjectName != "PurgeProbe__c" {
				t.Fatalf("ObjectName = %q, want PurgeProbe__c", event.ObjectName)
			}
			if event.RecordID != "rec-1" {
				t.Fatalf("RecordID = %q, want rec-1", event.RecordID)
			}
			if event.Record[constants.FieldIsDeleted] != true {
				t.Fatalf("Record IsDeleted = %#v, want true", event.Record[constants.FieldIsDeleted])
			}
			if event.OldRecord[constants.FieldIsDeleted] != true {
				t.Fatalf("OldRecord IsDeleted = %#v, want true", event.OldRecord[constants.FieldIsDeleted])
			}
			return sentinel
		},
	}); err != nil {
		t.Fatalf("RegisterRecordWriteValidator returned error: %v", err)
	}

	err := ps.PurgeRecord(context.Background(), "rec-1", user)
	if !errors.Is(err, sentinel) {
		t.Fatalf("PurgeRecord error = %v, want sentinel validator error", err)
	}
	if !called {
		t.Fatal("before_purge validator was not called")
	}
	if len(repo.physicalDeletes) != 0 {
		t.Fatalf("PurgeRecord physically deleted rows after failing validator: %#v", repo.physicalDeletes)
	}
}

func TestPurgeRecordCascadePurgesDeletedChildrenLeafUp(t *testing.T) {
	user := recordEntrypointUser()
	parentSchema := restoreObjectSchema("Parent__c")
	childSchema := restoreObjectSchema("Child__c")
	cascade := constants.DeleteRuleCascade
	childSchema.Fields = append(childSchema.Fields, models.FieldMetadata{
		APIName:     "Parent__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Parent__c"},
		DeleteRule:  &cascade,
		Createable:  true,
		Updateable:  true,
	})
	makeFieldsMap(childSchema)
	repo := newRecycleBinPurgeRepo("Parent__c", "parent-1", user.ID)
	repo.addDeletedRecord("Child__c", "child-1", user.ID, models.SObject{"Parent__c": "parent-1"})
	repo.addDeleteOperationLineage("parent-1", "delete-op-1", "Parent__c", "parent-1", "", "", "", 0)
	repo.addDeleteOperationLineage("child-1", "delete-op-1", "Parent__c", "parent-1", "Parent__c", "parent-1", "Parent__c", 1)
	ps := newRecycleBinPurgeService(t, repo, true, parentSchema, childSchema)

	if err := ps.PurgeRecord(context.Background(), "parent-1", user); err != nil {
		t.Fatalf("PurgeRecord: %v", err)
	}
	wantDeletes := "Child__c|child-1," + constants.TableRecycleBin + "|rb-child-1,Parent__c|parent-1," + constants.TableRecycleBin + "|rb-parent-1"
	if got := strings.Join(repo.physicalDeletes, ","); got != wantDeletes {
		t.Fatalf("physical deletes = %s, want %s", got, wantDeletes)
	}
	assertLockBefore(t, repo.lockLog, "lock:"+constants.TableRecycleBin+":rb-parent-1", "lock:Parent__c:parent-1")
	assertLockBefore(t, repo.lockLog, "lock:"+constants.TableRecycleBin+":rb-child-1", "lock:Child__c:child-1")
	assertLockBefore(t, repo.lockLog, "lock:"+constants.TableRecycleBin+":rb-child-1", "lock:Parent__c:parent-1")
}

func TestPurgeRecordCascadeTrustsLineageWhenRelationshipFieldRemoved(t *testing.T) {
	user := recordEntrypointUser()
	parentSchema := restoreObjectSchema("Parent__c")
	childSchema := restoreObjectSchema("Child__c")
	repo := newRecycleBinPurgeRepo("Parent__c", "parent-1", user.ID)
	repo.addDeletedRecord("Child__c", "child-1", user.ID, nil)
	repo.addDeleteOperationLineage("parent-1", "delete-op-1", "Parent__c", "parent-1", "", "", "", 0)
	repo.addDeleteOperationLineage("child-1", "delete-op-1", "Parent__c", "parent-1", "Parent__c", "parent-1", "Parent__c", 1)
	repo.markChildFieldSchemaDrift("Child__c", "Parent__c")
	ps := newRecycleBinPurgeService(t, repo, true, parentSchema, childSchema)

	if err := ps.PurgeRecord(context.Background(), "parent-1", user); err != nil {
		t.Fatalf("PurgeRecord after relationship field removal: %v", err)
	}
	wantDeletes := "Child__c|child-1," + constants.TableRecycleBin + "|rb-child-1,Parent__c|parent-1," + constants.TableRecycleBin + "|rb-parent-1"
	if got := strings.Join(repo.physicalDeletes, ","); got != wantDeletes {
		t.Fatalf("physical deletes = %s, want %s", got, wantDeletes)
	}
}

func TestPurgeRecordLiveCascadeEdgeTreatsSchemaDriftAsInternalError(t *testing.T) {
	user := recordEntrypointUser()
	parentSchema := restoreObjectSchema("Parent__c")
	childSchema := restoreObjectSchema("Child__c")
	cascade := constants.DeleteRuleCascade
	childSchema.Fields = append(childSchema.Fields, models.FieldMetadata{
		APIName:     "Parent__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Parent__c"},
		DeleteRule:  &cascade,
		Createable:  true,
		Updateable:  true,
	})
	makeFieldsMap(childSchema)
	repo := newRecycleBinPurgeRepo("Parent__c", "parent-1", user.ID)
	repo.markChildFieldSchemaDrift("Child__c", "Parent__c")
	ps := newRecycleBinPurgeService(t, repo, true, parentSchema, childSchema)

	err := ps.PurgeRecord(context.Background(), "parent-1", user)
	if statusOfAPIError(t, err) != http.StatusInternalServerError {
		t.Fatalf("PurgeRecord error = %v, want 500", err)
	}
	if len(repo.physicalDeletes) != 0 {
		t.Fatalf("PurgeRecord deleted rows despite live-metadata schema drift: %#v", repo.physicalDeletes)
	}
}

func TestPurgeRecordReauthorizesCascadeChildren(t *testing.T) {
	user := recordEntrypointUser()
	parentSchema := restoreObjectSchema("Parent__c")
	childSchema := restoreObjectSchema("Child__c")
	cascade := constants.DeleteRuleCascade
	childSchema.AccessPolicy = &models.ObjectAccessPolicy{
		Mode: "derived",
		ObjectDelete: &models.AccessPolicyExpr{
			Kind:  "recordFieldEqualsCurrentUser",
			Field: constants.FieldOwnerID,
		},
	}
	childSchema.Fields = append(childSchema.Fields, models.FieldMetadata{
		APIName:     "Parent__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Parent__c"},
		DeleteRule:  &cascade,
		Createable:  true,
		Updateable:  true,
	})
	makeFieldsMap(childSchema)
	repo := newRecycleBinPurgeRepo("Parent__c", "parent-1", user.ID)
	repo.addDeletedRecord("Child__c", "child-1", user.ID, models.SObject{
		constants.FieldOwnerID: "005-other-owner",
		"Parent__c":            "parent-1",
	})
	repo.addDeleteOperationLineage("parent-1", "delete-op-1", "Parent__c", "parent-1", "", "", "", 0)
	repo.addDeleteOperationLineage("child-1", "delete-op-1", "Parent__c", "parent-1", "Parent__c", "parent-1", "Parent__c", 1)
	ps := newRecycleBinPurgeService(t, repo, true, parentSchema, childSchema)

	err := ps.PurgeRecord(context.Background(), "parent-1", user)
	if statusOfAPIError(t, err) != http.StatusForbidden {
		t.Fatalf("PurgeRecord error = %v, want 403", err)
	}
	if len(repo.physicalDeletes) != 0 {
		t.Fatalf("PurgeRecord deleted rows despite child Delete denial: %#v", repo.physicalDeletes)
	}
}

func TestPurgeRecordDoesNotScanRelationshipThatNoLongerCascades(t *testing.T) {
	user := recordEntrypointUser()
	parentSchema := restoreObjectSchema("Parent__c")
	childSchema := restoreObjectSchema("Child__c")
	setNull := constants.DeleteRuleSetNull
	childSchema.Fields = append(childSchema.Fields, models.FieldMetadata{
		APIName:     "Parent__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Parent__c"},
		DeleteRule:  &setNull,
		Createable:  true,
		Updateable:  true,
	})
	makeFieldsMap(childSchema)
	repo := newRecycleBinPurgeRepo("Parent__c", "parent-1", user.ID)
	repo.getDeletedChildrenErr = errors.New("schema drift")
	ps := newRecycleBinPurgeService(t, repo, true, parentSchema, childSchema)

	if err := ps.PurgeRecord(context.Background(), "parent-1", user); err != nil {
		t.Fatalf("PurgeRecord: %v", err)
	}
	if got, want := strings.Join(repo.physicalDeletes, ","), "Parent__c|parent-1,"+constants.TableRecycleBin+"|rb-parent-1"; got != want {
		t.Fatalf("physical deletes = %s, want %s", got, want)
	}
}

func TestPurgeRecordRejectsLineageWhenCurrentCascadeChildRemainsActive(t *testing.T) {
	user := recordEntrypointUser()
	parentSchema := restoreObjectSchema("Parent__c")
	childSchema := restoreObjectSchema("Child__c")
	cascade := constants.DeleteRuleCascade
	childSchema.Fields = append(childSchema.Fields, models.FieldMetadata{
		APIName:     "Parent__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Parent__c"},
		DeleteRule:  &cascade,
		Createable:  true,
		Updateable:  true,
	})
	makeFieldsMap(childSchema)
	repo := newRecycleBinPurgeRepo("Parent__c", "parent-1", user.ID)
	repo.records[purgeRepoRecordKey("Child__c", "child-1")] = models.SObject{
		constants.FieldID:        "child-1",
		constants.FieldName:      "Active child",
		constants.FieldOwnerID:   user.ID,
		constants.FieldIsDeleted: false,
		"Parent__c":              "parent-1",
	}
	ps := newRecycleBinPurgeService(t, repo, true, parentSchema, childSchema)

	err := ps.PurgeRecord(context.Background(), "parent-1", user)
	if statusOfAPIError(t, err) != http.StatusConflict {
		t.Fatalf("PurgeRecord error = %v, want 409", err)
	}
	if len(repo.physicalDeletes) != 0 {
		t.Fatalf("PurgeRecord deleted parent despite active child: %#v", repo.physicalDeletes)
	}
	if _, ok := repo.records[purgeRepoRecordKey("Child__c", "child-1")]; !ok {
		t.Fatal("PurgeRecord deleted active child outside recorded lineage")
	}
}

func TestPurgeRecordRejectsCascadeTreeWithoutRecoverableLineage(t *testing.T) {
	tests := []struct {
		name             string
		configureRepo    func(repo *recycleBinPurgeRepoStub)
		noDeleteMessage  string
		childKeptMessage string // empty skips the child-survival assertion
	}{
		{
			name: "LineageWhenCurrentDeletedChildIsOutsideOperation",
			configureRepo: func(repo *recycleBinPurgeRepoStub) {
				repo.addDeleteOperationLineage("parent-1", "delete-op-1", "Parent__c", "parent-1", "", "", "", 0)
			},
			noDeleteMessage:  "PurgeRecord deleted parent despite deleted child outside lineage",
			childKeptMessage: "PurgeRecord deleted child outside recorded lineage",
		},
		{
			name: "LegacyCascadeTreeWithoutDeleteOperationId",
			configureRepo: func(repo *recycleBinPurgeRepoStub) {
				removeRecycleBinLineage(repo.tombstones["parent-1"])
				removeRecycleBinLineage(repo.tombstones["child-1"])
			},
			noDeleteMessage: "PurgeRecord purged unmigrated legacy cascade tree",
		},
		{
			name: "RootOnlyLineageWithDeletedNonLineageCascadeChild",
			configureRepo: func(repo *recycleBinPurgeRepoStub) {
				repo.addDeleteOperationLineage("parent-1", "legacy-root-parent-1", "Parent__c", "parent-1", "", "", "", 0)
				repo.addDeleteOperationLineage("child-1", "legacy-root-child-1", "Child__c", "child-1", "", "", "", 0)
			},
			noDeleteMessage:  "PurgeRecord deleted parent despite deleted child outside lineage",
			childKeptMessage: "PurgeRecord deleted separate child operation",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			user := recordEntrypointUser()
			parentSchema := restoreObjectSchema("Parent__c")
			childSchema := restoreObjectSchema("Child__c")
			cascade := constants.DeleteRuleCascade
			childSchema.Fields = append(childSchema.Fields, models.FieldMetadata{
				APIName:     "Parent__c",
				Type:        constants.FieldTypeReference,
				ReferenceTo: []string{"Parent__c"},
				DeleteRule:  &cascade,
				Createable:  true,
				Updateable:  true,
			})
			makeFieldsMap(childSchema)
			repo := newRecycleBinPurgeRepo("Parent__c", "parent-1", user.ID)
			repo.addDeletedRecord("Child__c", "child-1", user.ID, models.SObject{"Parent__c": "parent-1"})
			tt.configureRepo(repo)
			ps := newRecycleBinPurgeService(t, repo, true, parentSchema, childSchema)

			err := ps.PurgeRecord(context.Background(), "parent-1", user)
			if statusOfAPIError(t, err) != http.StatusConflict {
				t.Fatalf("PurgeRecord error = %v, want 409", err)
			}
			if len(repo.physicalDeletes) != 0 {
				t.Fatalf("%s: %#v", tt.noDeleteMessage, repo.physicalDeletes)
			}
			if tt.childKeptMessage != "" {
				if _, ok := repo.records[purgeRepoRecordKey("Child__c", "child-1")]; !ok {
					t.Fatal(tt.childKeptMessage)
				}
			}
		})
	}
}

func TestPurgeRecordAllowsLegacyLeafWithoutDeleteOperationId(t *testing.T) {
	user := recordEntrypointUser()
	schema := restoreObjectSchema("Account")
	repo := newRecycleBinPurgeRepo("Account", "rec-1", user.ID)
	removeRecycleBinLineage(repo.tombstones["rec-1"])
	ps := newRecycleBinPurgeService(t, repo, true, schema)

	if err := ps.PurgeRecord(context.Background(), "rec-1", user); err != nil {
		t.Fatalf("PurgeRecord legacy leaf: %v", err)
	}
	if got, want := strings.Join(repo.physicalDeletes, ","), "Account|rec-1,"+constants.TableRecycleBin+"|rb-rec-1"; got != want {
		t.Fatalf("physical deletes = %s, want %s", got, want)
	}
}

func TestPurgeRecordRejectsCascadeLineageWhenDeletedChildFKDrifted(t *testing.T) {
	user := recordEntrypointUser()
	parentSchema := restoreObjectSchema("Parent__c")
	childSchema := restoreObjectSchema("Child__c")
	cascade := constants.DeleteRuleCascade
	childSchema.Fields = append(childSchema.Fields, models.FieldMetadata{
		APIName:     "Parent__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Parent__c"},
		DeleteRule:  &cascade,
		Createable:  true,
		Updateable:  true,
	})
	makeFieldsMap(childSchema)
	repo := newRecycleBinPurgeRepo("Parent__c", "parent-1", user.ID)
	repo.addDeletedRecord("Child__c", "child-1", user.ID, models.SObject{"Parent__c": "other-parent"})
	repo.addDeleteOperationLineage("parent-1", "delete-op-1", "Parent__c", "parent-1", "", "", "", 0)
	repo.addDeleteOperationLineage("child-1", "delete-op-1", "Parent__c", "parent-1", "Parent__c", "parent-1", "Parent__c", 1)
	ps := newRecycleBinPurgeService(t, repo, true, parentSchema, childSchema)

	err := ps.PurgeRecord(context.Background(), "parent-1", user)
	if statusOfAPIError(t, err) != http.StatusConflict {
		t.Fatalf("PurgeRecord error = %v, want 409", err)
	}
	if len(repo.physicalDeletes) != 0 {
		t.Fatalf("PurgeRecord deleted rows despite drifted child edge: %#v", repo.physicalDeletes)
	}
}

func TestPurgeRecordNonRootPurgesLineageBackedDescendantSubtree(t *testing.T) {
	user := recordEntrypointUser()
	parentSchema := restoreObjectSchema("Parent__c")
	childSchema := restoreObjectSchema("Child__c")
	grandchildSchema := restoreObjectSchema("Grandchild__c")
	cascade := constants.DeleteRuleCascade
	childSchema.Fields = append(childSchema.Fields, models.FieldMetadata{
		APIName:     "Parent__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Parent__c"},
		DeleteRule:  &cascade,
		Createable:  true,
		Updateable:  true,
	})
	grandchildSchema.Fields = append(grandchildSchema.Fields, models.FieldMetadata{
		APIName:     "Child__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Child__c"},
		DeleteRule:  &cascade,
		Createable:  true,
		Updateable:  true,
	})
	makeFieldsMap(childSchema)
	makeFieldsMap(grandchildSchema)
	repo := newRecycleBinPurgeRepo("Parent__c", "parent-1", user.ID)
	repo.addDeletedRecord("Child__c", "child-1", user.ID, models.SObject{"Parent__c": "parent-1"})
	repo.addDeletedRecord("Grandchild__c", "grandchild-1", user.ID, models.SObject{"Child__c": "child-1"})
	repo.addDeleteOperationLineage("parent-1", "delete-op-1", "Parent__c", "parent-1", "", "", "", 0)
	repo.addDeleteOperationLineage("child-1", "delete-op-1", "Parent__c", "parent-1", "Parent__c", "parent-1", "Parent__c", 1)
	repo.addDeleteOperationLineage("grandchild-1", "delete-op-1", "Parent__c", "parent-1", "Child__c", "child-1", "Child__c", 2)
	ps := newRecycleBinPurgeService(t, repo, true, parentSchema, childSchema, grandchildSchema)

	if err := ps.PurgeRecord(context.Background(), "child-1", user); err != nil {
		t.Fatalf("PurgeRecord non-root subtree: %v", err)
	}
	wantDeletes := "Grandchild__c|grandchild-1," + constants.TableRecycleBin + "|rb-grandchild-1,Child__c|child-1," + constants.TableRecycleBin + "|rb-child-1"
	if got := strings.Join(repo.physicalDeletes, ","); got != wantDeletes {
		t.Fatalf("physical deletes = %s, want %s", got, wantDeletes)
	}
	if _, ok := repo.records[purgeRepoRecordKey("Parent__c", "parent-1")]; !ok {
		t.Fatal("PurgeRecord non-root subtree deleted ancestor parent")
	}
}

func TestPurgeRecordNonRootSubtreeTrustsLineageWhenParentRelationshipFieldRemoved(t *testing.T) {
	user := recordEntrypointUser()
	parentSchema := restoreObjectSchema("Parent__c")
	childSchema := restoreObjectSchema("Child__c")
	grandchildSchema := restoreObjectSchema("Grandchild__c")
	cascade := constants.DeleteRuleCascade
	grandchildSchema.Fields = append(grandchildSchema.Fields, models.FieldMetadata{
		APIName:     "Child__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Child__c"},
		DeleteRule:  &cascade,
		Createable:  true,
		Updateable:  true,
	})
	makeFieldsMap(grandchildSchema)
	repo := newRecycleBinPurgeRepo("Parent__c", "parent-1", user.ID)
	repo.addDeletedRecord("Child__c", "child-1", user.ID, nil)
	repo.addDeletedRecord("Grandchild__c", "grandchild-1", user.ID, models.SObject{"Child__c": "child-1"})
	repo.addDeleteOperationLineage("parent-1", "delete-op-1", "Parent__c", "parent-1", "", "", "", 0)
	repo.addDeleteOperationLineage("child-1", "delete-op-1", "Parent__c", "parent-1", "Parent__c", "parent-1", "Parent__c", 1)
	repo.addDeleteOperationLineage("grandchild-1", "delete-op-1", "Parent__c", "parent-1", "Child__c", "child-1", "Child__c", 2)
	repo.markChildFieldSchemaDrift("Child__c", "Parent__c")
	ps := newRecycleBinPurgeService(t, repo, true, parentSchema, childSchema, grandchildSchema)

	if err := ps.PurgeRecord(context.Background(), "child-1", user); err != nil {
		t.Fatalf("PurgeRecord non-root subtree after parent relationship field removal: %v", err)
	}
	wantDeletes := "Grandchild__c|grandchild-1," + constants.TableRecycleBin + "|rb-grandchild-1,Child__c|child-1," + constants.TableRecycleBin + "|rb-child-1"
	if got := strings.Join(repo.physicalDeletes, ","); got != wantDeletes {
		t.Fatalf("physical deletes = %s, want %s", got, wantDeletes)
	}
	if _, ok := repo.records[purgeRepoRecordKey("Parent__c", "parent-1")]; !ok {
		t.Fatal("PurgeRecord non-root subtree deleted ancestor parent")
	}
}

func TestPurgeRecordNonRootSubtreeTrustsLineageWhenRelationshipNoLongerCascade(t *testing.T) {
	user := recordEntrypointUser()
	childSchema := restoreObjectSchema("Child__c")
	grandchildSchema := restoreObjectSchema("Grandchild__c")
	setNull := constants.DeleteRuleSetNull
	grandchildSchema.Fields = append(grandchildSchema.Fields, models.FieldMetadata{
		APIName:     "Child__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Child__c"},
		DeleteRule:  &setNull,
		Createable:  true,
		Updateable:  true,
	})
	makeFieldsMap(grandchildSchema)
	repo := newRecycleBinPurgeRepo("Child__c", "child-1", user.ID)
	repo.records[purgeRepoRecordKey("Child__c", "child-1")]["Parent__c"] = "parent-1"
	repo.addDeletedRecord("Grandchild__c", "grandchild-1", user.ID, models.SObject{"Child__c": "child-1"})
	repo.addDeleteOperationLineage("child-1", "delete-op-1", "Parent__c", "parent-1", "Parent__c", "parent-1", "Parent__c", 1)
	repo.addDeleteOperationLineage("grandchild-1", "delete-op-1", "Parent__c", "parent-1", "Child__c", "child-1", "Child__c", 2)
	ps := newRecycleBinPurgeService(t, repo, true, childSchema, grandchildSchema)

	if err := ps.PurgeRecord(context.Background(), "child-1", user); err != nil {
		t.Fatalf("PurgeRecord non-root subtree after relationship changed to SetNull: %v", err)
	}
	wantDeletes := "Grandchild__c|grandchild-1," + constants.TableRecycleBin + "|rb-grandchild-1,Child__c|child-1," + constants.TableRecycleBin + "|rb-child-1"
	if got := strings.Join(repo.physicalDeletes, ","); got != wantDeletes {
		t.Fatalf("physical deletes = %s, want %s", got, wantDeletes)
	}
}

func TestPurgeRecordNonRootSubtreeRejectsNonLineageDeletedGrandchildWhenRelationshipNoLongerCascade(t *testing.T) {
	user := recordEntrypointUser()
	childSchema := restoreObjectSchema("Child__c")
	grandchildSchema := restoreObjectSchema("Grandchild__c")
	setNull := constants.DeleteRuleSetNull
	grandchildSchema.Fields = append(grandchildSchema.Fields, models.FieldMetadata{
		APIName:     "Child__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Child__c"},
		DeleteRule:  &setNull,
		Createable:  true,
		Updateable:  true,
	})
	makeFieldsMap(grandchildSchema)
	repo := newRecycleBinPurgeRepo("Child__c", "child-1", user.ID)
	repo.records[purgeRepoRecordKey("Child__c", "child-1")]["Parent__c"] = "parent-1"
	repo.addDeletedRecord("Grandchild__c", "grandchild-1", user.ID, models.SObject{"Child__c": "child-1"})
	repo.addDeletedRecord("Grandchild__c", "grandchild-2", user.ID, models.SObject{"Child__c": "child-1"})
	repo.addDeleteOperationLineage("child-1", "delete-op-1", "Parent__c", "parent-1", "Parent__c", "parent-1", "Parent__c", 1)
	repo.addDeleteOperationLineage("grandchild-1", "delete-op-1", "Parent__c", "parent-1", "Child__c", "child-1", "Child__c", 2)
	ps := newRecycleBinPurgeService(t, repo, true, childSchema, grandchildSchema)

	err := ps.PurgeRecord(context.Background(), "child-1", user)
	if statusOfAPIError(t, err) != http.StatusConflict {
		t.Fatalf("PurgeRecord error = %v, want 409", err)
	}
	if len(repo.physicalDeletes) != 0 {
		t.Fatalf("PurgeRecord deleted selected subtree despite deleted child outside lineage: %#v", repo.physicalDeletes)
	}
	if _, ok := repo.records[purgeRepoRecordKey("Grandchild__c", "grandchild-2")]; !ok {
		t.Fatal("PurgeRecord deleted grandchild outside selected lineage")
	}
}

func TestPurgeRecordHidesInvisibleTombstone(t *testing.T) {
	user := recordEntrypointUser()
	schema := restoreObjectSchema("Account")
	repo := newRecycleBinPurgeRepo("Account", "rec-1", "005-other-deleter")
	ps := newRecycleBinPurgeService(t, repo, true, schema)

	err := ps.PurgeRecord(context.Background(), "rec-1", user)
	if statusOfAPIError(t, err) != http.StatusNotFound {
		t.Fatalf("PurgeRecord error = %v, want 404", err)
	}
	if len(repo.physicalDeletes) != 0 {
		t.Fatalf("PurgeRecord deleted hidden tombstone: %#v", repo.physicalDeletes)
	}
}

func TestPurgeRecordRequiresDeletePermission(t *testing.T) {
	user := recordEntrypointUser()
	schema := restoreObjectSchema("Account")
	repo := newRecycleBinPurgeRepo("Account", "rec-1", user.ID)
	ps := newRecycleBinPurgeService(t, repo, false, schema)

	err := ps.PurgeRecord(context.Background(), "rec-1", user)
	if statusOfAPIError(t, err) != http.StatusForbidden {
		t.Fatalf("PurgeRecord error = %v, want 403", err)
	}
	if len(repo.physicalDeletes) != 0 {
		t.Fatalf("PurgeRecord deleted without Delete permission: %#v", repo.physicalDeletes)
	}
}

func TestPurgeRecordReturnsConflictWhenRecordWasRestoredDuringPurge(t *testing.T) {
	user := recordEntrypointUser()
	schema := restoreObjectSchema("Account")
	repo := newRecycleBinPurgeRepo("Account", "rec-1", user.ID)
	repo.records[purgeRepoRecordKey("Account", "rec-1")][constants.FieldIsDeleted] = false
	ps := newRecycleBinPurgeService(t, repo, true, schema)

	err := ps.PurgeRecord(context.Background(), "rec-1", user)
	if statusOfAPIError(t, err) != http.StatusConflict {
		t.Fatalf("PurgeRecord error = %v, want 409", err)
	}
	if len(repo.physicalDeletes) != 0 {
		t.Fatalf("PurgeRecord deleted restored row: %#v", repo.physicalDeletes)
	}
}

func TestPurgeRecordReturnsConflictWhenTombstoneWasClearedDuringPurge(t *testing.T) {
	user := recordEntrypointUser()
	schema := restoreObjectSchema("Account")
	repo := newRecycleBinPurgeRepo("Account", "rec-1", user.ID)
	repo.tombstoneMissingAfterFirstFind = true
	ps := newRecycleBinPurgeService(t, repo, true, schema)

	err := ps.PurgeRecord(context.Background(), "rec-1", user)
	if statusOfAPIError(t, err) != http.StatusConflict {
		t.Fatalf("PurgeRecord error = %v, want 409", err)
	}
	if len(repo.physicalDeletes) != 0 {
		t.Fatalf("PurgeRecord deleted row after tombstone drift: %#v", repo.physicalDeletes)
	}
}

func TestPurgeRecordDoesNotCascadeSetNullDeletedChild(t *testing.T) {
	user := recordEntrypointUser()
	parentSchema := restoreObjectSchema("Parent__c")
	childSchema := restoreObjectSchema("Child__c")
	setNull := constants.DeleteRuleSetNull
	childSchema.Fields = append(childSchema.Fields, models.FieldMetadata{
		APIName:     "Parent__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Parent__c"},
		DeleteRule:  &setNull,
		Createable:  true,
		Updateable:  true,
	})
	makeFieldsMap(childSchema)
	repo := newRecycleBinPurgeRepo("Parent__c", "parent-1", user.ID)
	repo.addDeletedRecord("Child__c", "child-1", user.ID, models.SObject{"Parent__c": "parent-1"})
	ps := newRecycleBinPurgeService(t, repo, true, parentSchema, childSchema)

	if err := ps.PurgeRecord(context.Background(), "parent-1", user); err != nil {
		t.Fatalf("PurgeRecord: %v", err)
	}
	if got, want := strings.Join(repo.physicalDeletes, ","), "Parent__c|parent-1,"+constants.TableRecycleBin+"|rb-parent-1"; got != want {
		t.Fatalf("physical deletes = %s, want %s", got, want)
	}
	if _, ok := repo.records[purgeRepoRecordKey("Child__c", "child-1")]; !ok {
		t.Fatal("PurgeRecord physically deleted SetNull child")
	}
}

func TestPhysicalCascadeChildEdgesUsesEffectiveRelationshipSemantics(t *testing.T) {
	parentSchema := restoreObjectSchema("Parent__c")
	overlayChildSchema := restoreObjectSchema("OverlayChild__c")
	overlayChildSchema.Fields = append(overlayChildSchema.Fields, models.FieldMetadata{
		APIName:     "Parent__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Parent__c"},
		Createable:  true,
		Updateable:  true,
	})
	setNull := constants.DeleteRuleSetNull
	setNullChildSchema := restoreObjectSchema("SetNullChild__c")
	setNullChildSchema.Fields = append(setNullChildSchema.Fields, models.FieldMetadata{
		APIName:     "Parent__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Parent__c"},
		DeleteRule:  &setNull,
		Createable:  true,
		Updateable:  true,
	})
	makeFieldsMap(overlayChildSchema)
	makeFieldsMap(setNullChildSchema)

	metaSvc := metadataFeature.NewMetadataService(&fanoutMetadataRepo{
		schemas: []*models.ObjectMetadata{parentSchema, overlayChildSchema, setNullChildSchema},
		relationships: []*models.Relationship{{
			ChildObjectAPIName:  "OverlayChild__c",
			ParentObjectAPIName: "Parent__c",
			FieldAPIName:        "Parent__c",
			CascadeDelete:       true,
		}},
	}, nil, nil)
	ps := &PersistenceService{metadata: metaSvc}

	edges := ps.physicalCascadeChildEdges(context.Background(), "Parent__c")
	if len(edges) != 1 {
		t.Fatalf("physicalCascadeChildEdges len = %d (%#v), want 1 overlay cascade edge", len(edges), edges)
	}
	if edge := edges[0]; edge.childObjectName != "OverlayChild__c" || edge.fieldAPIName != "Parent__c" {
		t.Fatalf("physicalCascadeChildEdges[0] = %#v, want OverlayChild__c.Parent__c", edge)
	}
}

func newRecycleBinPurgeRepo(objectName, recordID, deletedByID string) *recycleBinPurgeRepoStub {
	repo := &recycleBinPurgeRepoStub{
		records:    map[string]models.SObject{},
		tombstones: map[string]models.SObject{},
	}
	repo.addDeletedRecord(objectName, recordID, deletedByID, nil)
	return repo
}

func (r *recycleBinPurgeRepoStub) addDeletedRecord(objectName, recordID, deletedByID string, extra models.SObject) {
	row := models.SObject{
		constants.FieldID:        recordID,
		constants.FieldName:      "Deleted row",
		constants.FieldOwnerID:   deletedByID,
		constants.FieldIsDeleted: true,
	}
	for k, v := range extra {
		row[k] = v
	}
	r.records[purgeRepoRecordKey(objectName, recordID)] = row
	r.tombstones[strings.ToLower(recordID)] = models.SObject{
		constants.FieldSysRecycleBin_ID:                      "rb-" + recordID,
		constants.FieldRecordID:                              recordID,
		constants.FieldObjectAPIName:                         objectName,
		constants.FieldRecordName:                            "Deleted row",
		constants.FieldSysRecycleBin_DeletedByID:             deletedByID,
		constants.FieldSysRecycleBin_DeleteOperationID:       "root-op-" + recordID,
		constants.FieldSysRecycleBin_DeleteRootObjectAPIName: objectName,
		constants.FieldSysRecycleBin_DeleteRootRecordID:      recordID,
		constants.FieldSysRecycleBin_CascadeDepth:            0,
	}
}

func (r *recycleBinPurgeRepoStub) addDeleteOperationLineage(recordID, operationID, rootObjectName, rootRecordID, parentObjectName, parentRecordID, fieldAPIName string, depth int) {
	tombstone := r.tombstones[strings.ToLower(recordID)]
	if tombstone == nil {
		return
	}
	tombstone[constants.FieldSysRecycleBin_DeleteOperationID] = operationID
	tombstone[constants.FieldSysRecycleBin_DeleteRootObjectAPIName] = rootObjectName
	tombstone[constants.FieldSysRecycleBin_DeleteRootRecordID] = rootRecordID
	tombstone[constants.FieldSysRecycleBin_CascadeDepth] = depth
	if parentObjectName != "" {
		tombstone[constants.FieldSysRecycleBin_CascadeParentObjectAPIName] = parentObjectName
	}
	if parentRecordID != "" {
		tombstone[constants.FieldSysRecycleBin_CascadeParentRecordID] = parentRecordID
	}
	if fieldAPIName != "" {
		tombstone[constants.FieldSysRecycleBin_CascadeFieldAPIName] = fieldAPIName
	}
}

func newRecycleBinPurgeService(t *testing.T, repo *recycleBinPurgeRepoStub, grantDelete bool, schemas ...*models.ObjectMetadata) *PersistenceService {
	t.Helper()
	metaSvc := metadataFeature.NewMetadataService(&fanoutMetadataRepo{schemas: schemas}, nil, nil)
	permSvc := permissions.NewPermissionService(&stubUpsertPermRepo{grantDelete: grantDelete}, metaSvc, nil)
	return &PersistenceService{
		repo:                      repo,
		metadata:                  metaSvc,
		permissions:               permSvc,
		eventBus:                  nil,
		validator:                 metadataFeature.NewValidationService(formula.NewEngine()),
		rollup:                    noopRollupProcessor{},
		txManager:                 inlineTxManager{},
		logger:                    logging.New(context.Background()),
		assignmentPolicyAuthority: &activeClaimAuthorityTestGuard{},
	}
}