package records

import (
	"context"
	stderrors "errors"
	"fmt"
	"testing"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// recordingLookupRepo is a test record repository that records GetLock events
// and handles FindOne from a pre-populated map. Used by T2.
type recordingLookupRepo struct {
	stubUpsertRecordRepo
	rows       map[string]map[string]models.SObject // objectName → id → record
	lockEvents []string                             // "ObjectName/id" in call order
}

func (r *recordingLookupRepo) GetLock(ctx context.Context, objectName string, id string) (models.SObject, error) {
	r.lockEvents = append(r.lockEvents, fmt.Sprintf("%s/%s", objectName, id))
	if r.rows != nil {
		if rec, ok := r.rows[objectName][id]; ok {
			return cloneSObject(rec), nil
		}
	}
	return nil, nil
}

func (r *recordingLookupRepo) FindOne(ctx context.Context, objectName string, id string) (models.SObject, error) {
	if r.rows != nil {
		if rec, ok := r.rows[objectName][id]; ok {
			return cloneSObject(rec), nil
		}
	}
	return nil, nil
}

// lookupTestRepo is a test record repository that handles GetLock and FindOne
// by looking up records in a pre-populated map. Used by T5/T6 which need both
// the lock AND the post-lock existence check to succeed.
type lookupTestRepo struct {
	stubUpsertRecordRepo
	// records maps "ObjectName/id" → SObject; also checked as "ObjectName" key.
	records   map[string]models.SObject
	lockCalls int
	findCalls int
}

func (r *lookupTestRepo) GetLock(ctx context.Context, tableName string, id string) (models.SObject, error) {
	r.lockCalls++
	if r.records != nil {
		key := tableName + "/" + id
		if rec, ok := r.records[key]; ok {
			return cloneSObject(rec), nil
		}
	}
	return nil, nil
}

func (r *lookupTestRepo) FindOne(ctx context.Context, tableName string, id string) (models.SObject, error) {
	r.findCalls++
	if r.records != nil {
		key := tableName + "/" + id
		if rec, ok := r.records[key]; ok {
			return cloneSObject(rec), nil
		}
	}
	return nil, nil
}

// toyLookupSchema builds a minimal schema with three Reference fields pointing
// at three toy objects: Alpha__c, Beta__c, Gamma__c. None of them sort above
// Zenith__c alphabetically, so any held Zenith__c key creates the out-of-order
// condition for all three.
func toyLookupThreeFieldSchema() *models.ObjectMetadata {
	schema := &models.ObjectMetadata{
		APIName:      "Leaf__c",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString, Createable: false, Updateable: false},
			{APIName: constants.FieldName, Type: constants.FieldTypeString, Createable: true, Updateable: true},
			lockRequiredLookupField("AlphaId__c", "Alpha__c"),
			lockRequiredLookupField("BetaId__c", "Beta__c"),
			lockRequiredLookupField("GammaId__c", "Gamma__c"),
		},
	}
	makeFieldsMap(schema)
	return schema
}

func lockRequiredLookupField(apiName string, targetObject string) models.FieldMetadata {
	return models.FieldMetadata{
		APIName:        apiName,
		Type:           constants.FieldTypeReference,
		ReferenceTo:    []string{targetObject},
		Createable:     true,
		Updateable:     true,
		IsMasterDetail: true,
	}
}

func toyAlphaSchema() *models.ObjectMetadata {
	s := &models.ObjectMetadata{
		APIName:      "Alpha__c",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString, Createable: false, Updateable: false},
			{APIName: constants.FieldName, Type: constants.FieldTypeString, Createable: true, Updateable: true},
		},
	}
	makeFieldsMap(s)
	return s
}

func toyBetaSchema() *models.ObjectMetadata {
	s := &models.ObjectMetadata{
		APIName:      "Beta__c",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString, Createable: false, Updateable: false},
			{APIName: constants.FieldName, Type: constants.FieldTypeString, Createable: true, Updateable: true},
		},
	}
	makeFieldsMap(s)
	return s
}

func toyGammaSchema() *models.ObjectMetadata {
	s := &models.ObjectMetadata{
		APIName:      "Gamma__c",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString, Createable: false, Updateable: false},
			{APIName: constants.FieldName, Type: constants.FieldTypeString, Createable: true, Updateable: true},
		},
	}
	makeFieldsMap(s)
	return s
}

func toyZenithSchema() *models.ObjectMetadata {
	s := &models.ObjectMetadata{
		APIName:      "Zenith__c",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString, Createable: false, Updateable: false},
			{APIName: constants.FieldName, Type: constants.FieldTypeString, Createable: true, Updateable: true},
		},
	}
	makeFieldsMap(s)
	return s
}

// TestLockLookupParentsCarriesCompleteRetryKeySet is the T1 lead reproducer.
//
// Setup: a pessimistic-tx coordinator already holds Zenith__c/z-1, then
// validateNonPolymorphicLookups is called with three FK fields pointing at
// Alpha__c/a-1, Beta__c/b-1, Gamma__c/g-1. All three sort below Zenith__c, so
// the per-key loop triggers an out-of-order error on the FIRST key and returns
// a RetryablePrelockError carrying only one key.
//
// Expected (post-fix): the single batched LockParentRows call returns an error
// carrying ALL THREE keys.
//
// Pre-fix: the per-key loop returns after the first key → only one key in the
// retry error → the test fails.
func TestLockLookupParentsCarriesCompleteRetryKeySet(t *testing.T) {
	// Arrange
	allParents := map[string]models.SObject{
		"Alpha__c/a-1":  {constants.FieldID: "a-1"},
		"Beta__c/b-1":   {constants.FieldID: "b-1"},
		"Gamma__c/g-1":  {constants.FieldID: "g-1"},
		"Zenith__c/z-1": {constants.FieldID: "z-1"},
	}

	var lockCalls []string
	repo := &stubUpsertRecordRepo{
		getLockFn: func(ctx context.Context, objectName string, id string) (models.SObject, error) {
			key := objectName + "/" + id
			lockCalls = append(lockCalls, key)
			if row, ok := allParents[key]; ok {
				return row, nil
			}
			return nil, nil
		},
	}

	// Build a PersistenceService with Zenith__c + Alpha__c + Beta__c + Gamma__c schemas
	ps := newRollupBoundaryTestService(repo,
		toyLookupThreeFieldSchema(), toyAlphaSchema(), toyBetaSchema(), toyGammaSchema(), toyZenithSchema(),
	)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	// Use system context so Retrieve-based access checks are bypassed
	ctx := constants.WithSecurityContext(context.Background(), constants.SystemContext, "T1 test")

	var gotErr error
	runErr := ps.runGroupMembershipAwarePessimisticTransactionForObjects(ctx, []string{"Leaf__c"}, func(txCtx context.Context) error {
		return ps.runParentLockCoordinatorIfActive(txCtx, func(coordCtx context.Context) error {
			// First: acquire Zenith__c/z-1 so it becomes the high-water mark.
			if _, err := ps.LockParentRows(coordCtx, []recordlocks.ParentKey{
				{ObjectName: "Zenith__c", RecordID: "z-1"},
			}); err != nil {
				return err
			}

			// Now call validateNonPolymorphicLookups with all three FK fields pointing
			// at Alpha, Beta, Gamma — all sorting below Zenith. The lock attempt
			// should discover out-of-order and return a retryable error carrying ALL
			// three keys. Pre-fix: only one key returned.
			schema := ps.metadata.GetSchema(coordCtx, "Leaf__c")
			data := models.SObject{
				"AlphaId__c": "a-1",
				"BetaId__c":  "b-1",
				"GammaId__c": "g-1",
			}
			gotErr = ps.validateNonPolymorphicLookups(coordCtx, data, schema, &models.UserSession{ID: "u1"})
			return nil // don't propagate; we inspect gotErr below
		})
	})
	_ = runErr

	// Assertions
	if gotErr == nil {
		t.Fatal("validateNonPolymorphicLookups returned nil, want retryable prelock error")
	}

	// Must NOT be a ValidationError — an out-of-order error is internal/retryable.
	if appErrors.IsValidation(gotErr) {
		t.Fatalf("validateNonPolymorphicLookups returned validation error %v, want retryable prelock error (IsInternal)", gotErr)
	}

	// Must satisfy RetryablePrelockError with ALL 3 keys.
	var retryErr recordlocks.RetryablePrelockError
	if !stderrors.As(gotErr, &retryErr) {
		t.Fatalf("validateNonPolymorphicLookups error does not satisfy RetryablePrelockError: %T %v", gotErr, gotErr)
	}
	keys := retryErr.ParentPrelockKeys()
	if len(keys) != 3 {
		t.Fatalf("ParentPrelockKeys() = %d keys, want 3; keys=%#v; err=%v", len(keys), keys, gotErr)
	}

	// IsInternal must be true (out-of-order abort is an engine-internal signal)
	if !appErrors.IsInternal(gotErr) {
		t.Fatalf("validateNonPolymorphicLookups error IsInternal=false, want true; err=%v", gotErr)
	}
}

// TestLockLookupParentsLocksParentInsideCoordinatorMultiTarget extends the
// existing T6 single-target test (TestValidateNonPolymorphicLookupsLocksParentInsideCoordinator
// at nonpoly_lookup_validator_test.go:64) for the multi-target batching shape:
// two FK fields pointing at Alpha__c and Beta__c should succeed with both parents
// accessible.
func TestLockLookupParentsLocksParentInsideCoordinatorMultiTarget(t *testing.T) {
	alpha := toyAlphaSchema()
	beta := toyBetaSchema()
	child := &models.ObjectMetadata{
		APIName:      "Leaf__c",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString, Createable: false, Updateable: false},
			{APIName: constants.FieldName, Type: constants.FieldTypeString, Createable: true, Updateable: true},
			lockRequiredLookupField("AlphaId__c", "Alpha__c"),
			lockRequiredLookupField("BetaId__c", "Beta__c"),
		},
	}
	makeFieldsMap(child)

	// lookupTestRepo provides both GetLock (for the lock step) and FindOne (for
	// the post-lock existence check in validateNonPolymorphicLookups).
	repo := &lookupTestRepo{
		records: map[string]models.SObject{
			"Alpha__c/a-1": {constants.FieldID: "a-1"},
			"Beta__c/b-1":  {constants.FieldID: "b-1"},
		},
	}

	ps := newRollupBoundaryTestService(repo, child, alpha, beta)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm
	ctx := constants.WithSecurityContext(context.Background(), constants.SystemContext, "T6 multi-target test")

	err := ps.runGroupMembershipAwarePessimisticTransactionForObjects(ctx, []string{"Leaf__c"}, func(txCtx context.Context) error {
		return ps.runParentLockCoordinatorIfActive(txCtx, func(coordCtx context.Context) error {
			return ps.validateNonPolymorphicLookups(coordCtx, models.SObject{
				"AlphaId__c": "a-1",
				"BetaId__c":  "b-1",
			}, child, &models.UserSession{ID: "u1"})
		})
	})
	if err != nil {
		t.Fatalf("validateNonPolymorphicLookups two targets: %v", err)
	}
	// Both parents were locked (one GetLock call per parent).
	if repo.lockCalls != 2 {
		t.Fatalf("GetLock calls = %d, want 2 (one per unique parent)", repo.lockCalls)
	}
}

// TestLockLookupParentsMissingParentReturnsValidation (T3) pins that a missing
// parent row returns a byte-identical 400 validation error, not a retryable error.
func TestLockLookupParentsMissingParentReturnsValidation(t *testing.T) {
	// lookupTestRepo with no records: both GetLock and FindOne return nil.
	repo := &lookupTestRepo{records: map[string]models.SObject{}}
	alpha := toyAlphaSchema()
	child := &models.ObjectMetadata{
		APIName:      "Leaf__c",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString, Createable: false, Updateable: false},
			{
				APIName:     "AlphaId__c",
				Type:        constants.FieldTypeReference,
				ReferenceTo: []string{"Alpha__c"},
				Createable:  true,
				Updateable:  true,
			},
		},
	}
	makeFieldsMap(child)

	ps := newRollupBoundaryTestService(repo, child, alpha)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm
	ctx := constants.WithSecurityContext(context.Background(), constants.SystemContext, "T3 test")

	var gotErr error
	_ = ps.runGroupMembershipAwarePessimisticTransactionForObjects(ctx, []string{"Leaf__c"}, func(txCtx context.Context) error {
		return ps.runParentLockCoordinatorIfActive(txCtx, func(coordCtx context.Context) error {
			gotErr = ps.validateNonPolymorphicLookups(coordCtx, models.SObject{
				"AlphaId__c": "absent-001",
			}, child, &models.UserSession{ID: "u1"})
			return nil
		})
	})

	if gotErr == nil {
		t.Fatal("missing parent returned nil, want validation error")
	}
	if !appErrors.IsValidation(gotErr) {
		t.Fatalf("missing parent error = %T %v, want validation error", gotErr, gotErr)
	}
	var retryErr recordlocks.RetryablePrelockError
	if stderrors.As(gotErr, &retryErr) {
		t.Fatalf("missing parent returned retryable error %v, want plain validation error", gotErr)
	}
}

// TestLockLookupParentsTransactionOwnedSkipsLock (T5) pins that a key marked
// MarkTransactionOwned below the current high-water is not locked (no extra GetLock
// call) and that validateNonPolymorphicLookups succeeds (pins NX#2336/#2204).
func TestLockLookupParentsTransactionOwnedSkipsLock(t *testing.T) {
	// lookupTestRepo handles both GetLock (for Zenith) and FindOne (for Alpha's
	// post-lock existence check).
	repo := &lookupTestRepo{
		records: map[string]models.SObject{
			"Alpha__c/a-1":  {constants.FieldID: "a-1"},
			"Zenith__c/z-1": {constants.FieldID: "z-1"},
		},
	}
	alpha := toyAlphaSchema()
	zenith := toyZenithSchema()
	child := &models.ObjectMetadata{
		APIName:      "Leaf__c",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString, Createable: false, Updateable: false},
			{
				APIName:     "AlphaId__c",
				Type:        constants.FieldTypeReference,
				ReferenceTo: []string{"Alpha__c"},
				Createable:  true,
				Updateable:  true,
			},
		},
	}
	makeFieldsMap(child)

	ps := newRollupBoundaryTestService(repo, child, alpha, zenith)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm
	ctx := constants.WithSecurityContext(context.Background(), constants.SystemContext, "T5 test")

	var gotErr error
	_ = ps.runGroupMembershipAwarePessimisticTransactionForObjects(ctx, []string{"Leaf__c"}, func(txCtx context.Context) error {
		return ps.runParentLockCoordinatorIfActive(txCtx, func(coordCtx context.Context) error {
			// First acquire Zenith__c/z-1 to set high-water.
			if _, err := ps.LockParentRows(coordCtx, []recordlocks.ParentKey{
				{ObjectName: "Zenith__c", RecordID: "z-1"},
			}); err != nil {
				return err
			}
			// Mark Alpha__c/a-1 as transaction-owned (simulating it was inserted
			// in this tx) — it sorts below Zenith but should not produce an error.
			scope := recordlocks.ScopeFromContext(coordCtx)
			scope.MarkTransactionOwned(recordlocks.ParentKey{ObjectName: "Alpha__c", RecordID: "a-1"})

			gotErr = ps.validateNonPolymorphicLookups(coordCtx, models.SObject{
				"AlphaId__c": "a-1",
			}, child, &models.UserSession{ID: "u1"})
			return nil
		})
	})

	if gotErr != nil {
		t.Fatalf("MarkTransactionOwned key below high-water returned error: %v", gotErr)
	}
	// Zenith lock call happened; Alpha was skipped by the coordinator.
	if repo.lockCalls != 1 {
		t.Fatalf("GetLock calls = %d, want 1 (only Zenith; Alpha was transaction-owned)", repo.lockCalls)
	}
}

// TestPreviewBeforeUpdateConvergesWithForcedPrelocks is the T2 records-layer
// boundary contract test.
//
// Setup: RunInPessimisticTransaction with a coordinator; on each attempt the fn
// first acquires any forced prelocks from context, then Zenith__c/z-1 (high-water),
// then calls PreviewBeforeUpdate(Leaf__c, leaf-001, {AlphaId__c:a-1, BetaId__c:b-1,
// GammaId__c:g-1}).
//
// Attempt 1: no forced prelocks → Zenith acquired first → validateNonPolymorphicLookups
// tries to lock Alpha/Beta/Gamma all below Zenith → outOfOrderParentLockRetryError{keys:
// [Alpha,Beta,Gamma]} → RunInPessimisticTransaction catches it, sets forcedPrelocks =
// [Alpha,Beta,Gamma] and retries.
//
// Attempt 2: forced prelocks = [Alpha,Beta,Gamma] → fn locks Alpha,Beta,Gamma first
// (in order), then Zenith → validateNonPolymorphicLookups calls LockParentRows([Alpha,
// Beta,Gamma]) → all already held → nil → existence check passes → success.
//
// Pre-fix: per-key loop returns after first out-of-order key → budget exhausted after
// 3 single-key retries without converging → FAILS.
// Post-fix: batched call returns all 3 keys → single retry → PASS.
func TestPreviewBeforeUpdateConvergesWithForcedPrelocks(t *testing.T) {
	repo := &recordingLookupRepo{
		rows: map[string]map[string]models.SObject{
			// Leaf__c record has NO FK values set → updates will be effective changes.
			"Leaf__c": {
				"leaf-001": {constants.FieldID: "leaf-001", constants.FieldName: "leaf-001"},
			},
			"Alpha__c":  {"a-1": {constants.FieldID: "a-1"}},
			"Beta__c":   {"b-1": {constants.FieldID: "b-1"}},
			"Gamma__c":  {"g-1": {constants.FieldID: "g-1"}},
			"Zenith__c": {"z-1": {constants.FieldID: "z-1"}},
		},
	}

	ps := newRollupBoundaryTestService(repo,
		toyLookupThreeFieldSchema(), toyAlphaSchema(), toyBetaSchema(), toyGammaSchema(), toyZenithSchema(),
	)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	user := &models.UserSession{ID: "u1"}
	ctx := constants.WithSecurityContext(context.Background(), constants.SystemContext, "T2 test")

	// Capture lock events per pessimistic attempt so we can verify attempt-2 order.
	var attempt1Locks, attempt2Locks []string

	err := ps.RunInPessimisticTransaction(ctx, func(txCtx context.Context) error {
		return ps.runParentLockCoordinatorIfActive(txCtx, func(coordCtx context.Context) error {
			// Step 1: acquire any forced prelocks carried from a prior retry so they
			// sort before Zenith in the global order.
			forcedKeys := recordlocks.ForcedParentPrelocksFromContext(coordCtx)
			if len(forcedKeys) > 0 {
				if _, err := ps.LockParentRows(coordCtx, forcedKeys); err != nil {
					return err
				}
			}

			// Step 2: acquire Zenith__c/z-1 to establish the high-water mark.
			if _, err := ps.LockParentRows(coordCtx, []recordlocks.ParentKey{
				{ObjectName: "Zenith__c", RecordID: "z-1"},
			}); err != nil {
				return err
			}

			// Snapshot lock events for the current attempt.
			snapshot := make([]string, len(repo.lockEvents))
			copy(snapshot, repo.lockEvents)
			if txm.pessimisticCalls == 1 {
				attempt1Locks = snapshot
			} else {
				attempt2Locks = snapshot
			}
			// Reset for the next attempt.
			repo.lockEvents = repo.lockEvents[:0]

			// Step 3: preview the update — this triggers validateNonPolymorphicLookups.
			_, err := ps.PreviewBeforeUpdate(coordCtx, "Leaf__c", "leaf-001",
				models.SObject{
					"AlphaId__c": "a-1",
					"BetaId__c":  "b-1",
					"GammaId__c": "g-1",
				},
				nil, // internalStamps — nil per §2.2 (real approval path)
				user,
			)
			return err
		})
	})
	if err != nil {
		t.Fatalf("RunInPessimisticTransaction returned error: %v", err)
	}

	// Must have converged in exactly 2 pessimistic attempts.
	if txm.pessimisticCalls != 2 {
		t.Fatalf("pessimistic transaction calls = %d, want 2 (one retry with forced prelocks)", txm.pessimisticCalls)
	}

	// Attempt 1: Zenith is locked, then PreviewBeforeUpdate returns out-of-order.
	_ = attempt1Locks

	// Attempt 2: parent lock order must be Alpha < Beta < Gamma < Zenith.
	// (Leaf__c/leaf-001 may appear first as the child-row lock from prepareUpdateBeforeSave;
	// we assert only the relative ordering of the four parent locks.)
	wantParentOrder := []string{"Alpha__c/a-1", "Beta__c/b-1", "Gamma__c/g-1", "Zenith__c/z-1"}
	wantIdx := 0
	for _, lock := range attempt2Locks {
		if wantIdx < len(wantParentOrder) && lock == wantParentOrder[wantIdx] {
			wantIdx++
		}
	}
	if wantIdx != len(wantParentOrder) {
		t.Fatalf("attempt-2 parent lock order: missing %q after matching %v; full lock sequence: %v",
			wantParentOrder[wantIdx], wantParentOrder[:wantIdx], attempt2Locks)
	}
}

// TestLockLookupParentsMixedBatchOutOfOrderAndMissing is the T4 test.
//
// Setup: coordinator holds Zenith__c/z-1 (high-water); call
// validateNonPolymorphicLookups with two FK fields: AlphaId__c→Alpha__c/a-1
// (exists) and BetaId__c→Beta__c/b-1 (absent from repo).
//
// Step 1 (first call): lockLookupParentsIfCoordinated([Alpha,Beta]) →
// LockParentRows([Alpha,Beta]) → scope.KeysToAcquire sees both sort below Zenith
// → ErrOutOfOrderParentLock → outOfOrderParentLockRetryError{keys:[Alpha,Beta]}.
// The out-of-order check fires at PLANNING time (scope.KeysToAcquire), before
// GetLock is attempted, so Beta's absence is invisible at this stage.
// Assert: retryable error with BOTH Alpha+Beta keys.
//
// Step 2 (simulate forced-prelock retry): manually mark Alpha and Beta as
// acquired (mimicking what the retry boundary would do after prelocking them),
// then call validateNonPolymorphicLookups again.
// lockLookupParentsIfCoordinated([Alpha,Beta]) → LockParentRows([Alpha,Beta]) →
// scope.KeysToAcquire → both already held → toAcquire=[] → nil.
// Existence loop: Alpha found → pass; Beta absent → ValidationError.
// Assert: ValidationError for Beta.
//
// Pre-fix: step 1 returns only ONE key (Alpha) → retry loop would converge on
// Alpha but never discover Beta → FAILS.
func TestLockLookupParentsMixedBatchOutOfOrderAndMissing(t *testing.T) {
	// Alpha exists; Beta absent.
	repo := &lookupTestRepo{
		records: map[string]models.SObject{
			"Alpha__c/a-1":  {constants.FieldID: "a-1"},
			"Zenith__c/z-1": {constants.FieldID: "z-1"},
			// Beta__c/b-1 intentionally absent
		},
	}

	alpha := toyAlphaSchema()
	beta := toyBetaSchema()
	zenith := toyZenithSchema()
	child := &models.ObjectMetadata{
		APIName:      "Leaf__c",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString, Createable: false, Updateable: false},
			lockRequiredLookupField("AlphaId__c", "Alpha__c"),
			lockRequiredLookupField("BetaId__c", "Beta__c"),
		},
	}
	makeFieldsMap(child)

	ps := newRollupBoundaryTestService(repo, child, alpha, beta, zenith)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm
	ctx := constants.WithSecurityContext(context.Background(), constants.SystemContext, "T4 test")

	var step1Err error
	_ = ps.runGroupMembershipAwarePessimisticTransactionForObjects(ctx, []string{"Leaf__c"}, func(txCtx context.Context) error {
		return ps.runParentLockCoordinatorIfActive(txCtx, func(coordCtx context.Context) error {
			// Establish Zenith as the high-water mark.
			if _, err := ps.LockParentRows(coordCtx, []recordlocks.ParentKey{
				{ObjectName: "Zenith__c", RecordID: "z-1"},
			}); err != nil {
				return err
			}

			// ── Step 1: first call ─────────────────────────────────────────────────
			// Expect retryable out-of-order with BOTH Alpha+Beta (batched planning
			// fires before GetLock, so Beta's absence is invisible here).
			step1Err = ps.validateNonPolymorphicLookups(coordCtx, models.SObject{
				"AlphaId__c": "a-1",
				"BetaId__c":  "b-1",
			}, child, &models.UserSession{ID: "u1"})
			if step1Err == nil {
				t.Error("step 1: expected retryable error, got nil")
				return nil
			}
			var retryErr recordlocks.RetryablePrelockError
			if !stderrors.As(step1Err, &retryErr) {
				t.Errorf("step 1: error does not satisfy RetryablePrelockError: %T %v", step1Err, step1Err)
				return nil
			}
			keys := retryErr.ParentPrelockKeys()
			if len(keys) != 2 {
				t.Errorf("step 1: ParentPrelockKeys() = %d keys, want 2; keys=%#v; err=%v", len(keys), keys, step1Err)
				return nil
			}
			// Both keys must be present: Alpha__c/a-1 and Beta__c/b-1.
			foundAlpha, foundBeta := false, false
			for _, k := range keys {
				switch {
				case k.ObjectName == "Alpha__c" && k.RecordID == "a-1":
					foundAlpha = true
				case k.ObjectName == "Beta__c" && k.RecordID == "b-1":
					foundBeta = true
				default:
					t.Errorf("step 1: unexpected key %s/%s in ParentPrelockKeys()", k.ObjectName, k.RecordID)
				}
			}
			if !foundAlpha {
				t.Errorf("step 1: Alpha__c/a-1 missing from ParentPrelockKeys()")
			}
			if !foundBeta {
				t.Errorf("step 1: Beta__c/b-1 missing from ParentPrelockKeys()")
			}

			// ── Step 2: simulate forced-prelock retry ──────────────────────────────
			// The retry boundary would have pre-acquired Alpha and Beta before
			// re-running the fn. Simulate by marking them as acquired in scope.
			scope := recordlocks.ScopeFromContext(coordCtx)
			scope.MarkAcquired(recordlocks.ParentKey{ObjectName: "Alpha__c", RecordID: "a-1"})
			scope.MarkAcquired(recordlocks.ParentKey{ObjectName: "Beta__c", RecordID: "b-1"})

			// Now lockLookupParentsIfCoordinated sees both already held → nil.
			// The post-lock existence check discovers Beta absent → ValidationError.
			step2Err := ps.validateNonPolymorphicLookups(coordCtx, models.SObject{
				"AlphaId__c": "a-1",
				"BetaId__c":  "b-1",
			}, child, &models.UserSession{ID: "u1"})
			if step2Err == nil {
				t.Error("step 2: expected ValidationError for Beta, got nil")
				return nil
			}
			if !appErrors.IsValidation(step2Err) {
				t.Errorf("step 2: error = %T %v, want ValidationError", step2Err, step2Err)
			}
			var retryErr2 recordlocks.RetryablePrelockError
			if stderrors.As(step2Err, &retryErr2) {
				t.Errorf("step 2: got retryable error %v, want plain ValidationError", step2Err)
			}
			return nil // do not propagate; we inspected the errors above
		})
	})
}

// TestPreviewRetryableInsideActiveTxPassesThrough is the T7 preview-in-tx
// raw-exit regression pin.
//
// Design §5 T7: "retryable error inside active tx | exits PreviewBeforeUpdate
// UNCHANGED via RunInTransaction direct-execute :1247-1250 (no inner retry)".
//
// This test pins the invariant that RunInTransaction does NOT retry when
// IsTransactionActive is true.  It is a regression guard against the risk that
// a mis-wired RunInTransaction path could swallow or re-wrap a retryable prelock
// error before it reaches the outer pessimistic boundary loop.
//
// Setup:
//   - Put ctx in a plain (non-pessimistic) transaction via
//     recordingBulkTxManager.WithRetry (sets txActiveKey=true, txPessimisticKey=false).
//   - Inside that tx, call ps.RunInTransaction with a fn that returns an
//     outOfOrderParentLockRetryError directly (simulating what validateNonPolymorphicLookups
//     returns when a lookup lock is out of order).
//
// Assertions:
//   - RunInTransaction returns the error unchanged (no inner retry, no wrap).
//   - errors.As(&RetryablePrelockError) still holds on the returned error.
//   - The fn is called exactly once (no inner retry loop).
func TestPreviewRetryableInsideActiveTxPassesThrough(t *testing.T) {
	txm := &recordingBulkTxManager{}
	ps := &PersistenceService{txManager: txm}

	// Build the retryable error the inner fn will return.
	retryKeys := []recordlocks.ParentKey{
		{ObjectName: "Alpha__c", RecordID: "a-1"},
		{ObjectName: "Beta__c", RecordID: "b-1"},
	}
	syntheticErr := &outOfOrderParentLockRetryError{keys: retryKeys}

	fnCallCount := 0
	var gotErr error

	// WithRetry puts us in a plain (non-pessimistic) tx; inside, call
	// RunInTransaction which must detect IsTransactionActive==true and
	// execute the fn directly without opening a retry loop.
	_ = txm.WithRetry(context.Background(), func(txCtx context.Context) error {
		fnCallCount++ // count this outer WithRetry call

		if !txm.IsTransactionActive(txCtx) {
			t.Fatal("WithRetry did not mark transaction as active")
		}
		if txm.IsPessimisticTx(txCtx) {
			t.Fatal("WithRetry should not mark transaction as pessimistic")
		}

		innerCallCount := 0
		gotErr = ps.RunInTransaction(txCtx, func(innerCtx context.Context) error {
			innerCallCount++
			return syntheticErr
		})

		// The inner fn must have been called exactly once — no retry inside
		// RunInTransaction when IsTransactionActive is true.
		if innerCallCount != 1 {
			t.Errorf("inner fn called %d times inside RunInTransaction with active tx, want 1", innerCallCount)
		}

		return nil // do not propagate; we check gotErr below
	}, 3)

	if gotErr == nil {
		t.Fatal("RunInTransaction returned nil, want the synthetic outOfOrderParentLockRetryError")
	}

	// The error must satisfy RetryablePrelockError so the outer pessimistic
	// boundary can extract the keys.
	var retryErr recordlocks.RetryablePrelockError
	if !stderrors.As(gotErr, &retryErr) {
		t.Fatalf("RunInTransaction error does not satisfy RetryablePrelockError: %T %v", gotErr, gotErr)
	}
	gotKeys := retryErr.ParentPrelockKeys()
	if len(gotKeys) != len(retryKeys) {
		t.Fatalf("ParentPrelockKeys() = %d keys, want %d; keys=%#v", len(gotKeys), len(retryKeys), gotKeys)
	}
}