package records

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"testing"
	"time"

	eventFeature "github.com/nexuscrm/events"
	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	metadataFeature "github.com/nexuscrm/metadata"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	sharedEvents "github.com/nexuscrm/shared/pkg/events"
	"github.com/nexuscrm/shared/pkg/models"
)

type exclusiveSiblingRuntimeRepo struct {
	stubUpsertRecordRepo

	rows              map[string]map[string]models.SObject
	events            []string
	resets            []exclusiveSiblingResetCall
	landingOverride   []bool
	onGetLock         func(tableName, id string, row models.SObject)
	onResetLock       func()
	findOneErrByTable map[string]error
	hideUserFindOne   map[string]bool
}

type exclusiveSiblingResetCall struct {
	objectName     string
	fieldName      string
	partitionField string
	partitionValue string
	winnerID       string
}

type exclusiveSiblingOutboxRepo struct {
	payloads []interface{}
}

func (r *exclusiveSiblingOutboxRepo) Enqueue(_ context.Context, _ string, payload interface{}) (string, error) {
	r.payloads = append(r.payloads, payload)
	return fmt.Sprintf("evt-%d", len(r.payloads)), nil
}
func (r *exclusiveSiblingOutboxRepo) GetPendingEvents(_ context.Context, _ int) ([]models.OutboxEvent, error) {
	return nil, nil
}
func (r *exclusiveSiblingOutboxRepo) GetPendingEventsWithRecordFIFO(_ context.Context, _ int) ([]models.OutboxEvent, error) {
	return nil, nil
}
func (r *exclusiveSiblingOutboxRepo) ClaimEvent(_ context.Context, _ string) (models.OutboxClaim, error) {
	return models.OutboxClaim{}, nil
}
func (r *exclusiveSiblingOutboxRepo) CreateOutboxAttemptSavepoint(_ context.Context) error {
	return nil
}
func (r *exclusiveSiblingOutboxRepo) RollbackToOutboxAttemptSavepoint(_ context.Context) error {
	return nil
}
func (r *exclusiveSiblingOutboxRepo) ReleaseOutboxAttemptSavepoint(_ context.Context) error {
	return nil
}
func (r *exclusiveSiblingOutboxRepo) RefreshOutboxProcessingClaim(_ context.Context, _ models.OutboxClaim) error {
	return nil
}
func (r *exclusiveSiblingOutboxRepo) UpdateStatus(_ context.Context, _ models.OutboxClaim, _, _ string) error {
	return nil
}
func (r *exclusiveSiblingOutboxRepo) MarkFailed(_ context.Context, _ models.OutboxClaim, _ string) error {
	return nil
}
func (r *exclusiveSiblingOutboxRepo) IncrementRetry(_ context.Context, _ models.OutboxClaim, _ string) error {
	return nil
}
func (r *exclusiveSiblingOutboxRepo) CleanupProcessed(_ context.Context, _ time.Time) (int64, error) {
	return 0, nil
}

func newExclusiveSiblingRuntimeRepo(rows map[string]map[string]models.SObject) *exclusiveSiblingRuntimeRepo {
	cloned := make(map[string]map[string]models.SObject, len(rows))
	for objectName, byID := range rows {
		cloned[objectName] = make(map[string]models.SObject, len(byID))
		for id, row := range byID {
			cloned[objectName][id] = cloneSObject(row)
		}
	}
	return &exclusiveSiblingRuntimeRepo{rows: cloned}
}

func (r *exclusiveSiblingRuntimeRepo) FindOne(ctx context.Context, tableName string, id string) (models.SObject, error) {
	if err := r.findOneErrByTable[tableName]; err != nil {
		return nil, err
	}
	if r.hideUserFindOne[tableName] && constants.GetSecurityContext(ctx) < constants.SystemContext {
		return nil, nil
	}
	if row := r.row(tableName, id); row != nil {
		return cloneSObject(row), nil
	}
	return nil, nil
}

func (r *exclusiveSiblingRuntimeRepo) FindMany(ctx context.Context, tableName string, ids []string) (map[string]models.SObject, error) {
	out := make(map[string]models.SObject, len(ids))
	for _, id := range ids {
		row, err := r.FindOne(ctx, tableName, id)
		if err != nil {
			return nil, err
		}
		if row != nil {
			out[id] = row
		}
	}
	return out, nil
}

func (r *exclusiveSiblingRuntimeRepo) FindOneByField(_ context.Context, tableName string, fieldName string, value interface{}) (models.SObject, error) {
	for _, row := range r.rows[tableName] {
		if fmt.Sprint(row[fieldName]) == fmt.Sprint(value) {
			return cloneSObject(row), nil
		}
	}
	return nil, nil
}

func (r *exclusiveSiblingRuntimeRepo) GetLock(_ context.Context, tableName string, id string) (models.SObject, error) {
	r.events = append(r.events, fmt.Sprintf("lock:%s:%s", tableName, id))
	if r.onGetLock != nil {
		r.onGetLock(tableName, id, r.row(tableName, id))
	}
	if row := r.row(tableName, id); row != nil {
		return cloneSObject(row), nil
	}
	return nil, nil
}

func (r *exclusiveSiblingRuntimeRepo) Exists(_ context.Context, tableName string, id string) (bool, error) {
	return r.row(tableName, id) != nil, nil
}

func (r *exclusiveSiblingRuntimeRepo) CheckUniqueness(_ context.Context, schema *models.ObjectMetadata, fieldName string, value interface{}, excludeID string) (bool, error) {
	if schema == nil {
		return false, nil
	}
	for id, row := range r.rows[schema.APIName] {
		if id == excludeID || exclusiveSiblingTruthy(row[constants.FieldIsDeleted]) {
			continue
		}
		if fmt.Sprint(row[fieldName]) == fmt.Sprint(value) {
			return true, nil
		}
	}
	return false, nil
}

func (r *exclusiveSiblingRuntimeRepo) Insert(_ context.Context, tableName string, record models.SObject) error {
	id := exclusiveSiblingRecordID(record[constants.FieldID])
	r.events = append(r.events, fmt.Sprintf("insert:%s:%s", tableName, id))
	if r.rows[tableName] == nil {
		r.rows[tableName] = map[string]models.SObject{}
	}
	r.rows[tableName][id] = cloneSObject(record)
	return nil
}

func (r *exclusiveSiblingRuntimeRepo) BulkInsert(_ context.Context, tableName string, records []models.SObject, _ []string, opts interface{}) error {
	_, err := r.BulkInsertWithLanding(context.Background(), tableName, records, nil, opts)
	return err
}

func (r *exclusiveSiblingRuntimeRepo) BulkInsertWithLanding(_ context.Context, tableName string, records []models.SObject, _ []string, _ interface{}) ([]bool, error) {
	ids := make([]string, len(records))
	for i, record := range records {
		ids[i] = exclusiveSiblingRecordID(record[constants.FieldID])
	}
	r.events = append(r.events, fmt.Sprintf("bulk_insert:%s:%s", tableName, strings.Join(ids, ",")))
	landed := make([]bool, len(records))
	for i := range landed {
		landed[i] = true
	}
	if r.landingOverride != nil {
		copy(landed, r.landingOverride)
	}
	if r.rows[tableName] == nil {
		r.rows[tableName] = map[string]models.SObject{}
	}
	for i, record := range records {
		if i >= len(landed) || !landed[i] {
			continue
		}
		id := exclusiveSiblingRecordID(record[constants.FieldID])
		r.rows[tableName][id] = cloneSObject(record)
	}
	return landed, nil
}

func (r *exclusiveSiblingRuntimeRepo) Update(_ context.Context, tableName string, id string, updates models.SObject) error {
	r.events = append(r.events, fmt.Sprintf("update:%s:%s", tableName, id))
	if r.rows[tableName] == nil {
		r.rows[tableName] = map[string]models.SObject{}
	}
	row := cloneSObject(r.rows[tableName][id])
	if row == nil {
		row = models.SObject{constants.FieldID: id}
	}
	for key, value := range updates {
		row[key] = value
	}
	r.rows[tableName][id] = row
	return nil
}

func (r *exclusiveSiblingRuntimeRepo) LockExclusiveSiblingResetRows(_ context.Context, schema *models.ObjectMetadata, fieldAPIName, partitionFieldAPIName string, partitionValue interface{}, winnerID string) ([]models.SObject, error) {
	partition := exclusiveSiblingRecordID(partitionValue)
	r.events = append(r.events, fmt.Sprintf("reset:%s:%s:%s:%s", schema.APIName, fieldAPIName, partition, winnerID))
	if r.onResetLock != nil {
		r.onResetLock()
	}
	r.resets = append(r.resets, exclusiveSiblingResetCall{
		objectName:     schema.APIName,
		fieldName:      fieldAPIName,
		partitionField: partitionFieldAPIName,
		partitionValue: partition,
		winnerID:       winnerID,
	})
	var oldRows []models.SObject
	for id, row := range r.rows[schema.APIName] {
		if id == winnerID || exclusiveSiblingRecordID(row[partitionFieldAPIName]) != partition || !exclusiveSiblingTruthy(row[fieldAPIName]) {
			continue
		}
		if exclusiveSiblingTruthy(row[constants.FieldIsDeleted]) {
			continue
		}
		oldRows = append(oldRows, cloneSObject(row))
	}
	return oldRows, nil
}

func (r *exclusiveSiblingRuntimeRepo) row(tableName string, id string) models.SObject {
	if r.rows == nil || r.rows[tableName] == nil {
		return nil
	}
	return r.rows[tableName][id]
}

func exclusiveSiblingParentSchema() *models.ObjectMetadata {
	schema := &models.ObjectMetadata{
		APIName:      "Parent__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},
			{APIName: constants.FieldIsDeleted, Type: constants.FieldTypeBoolean, Createable: false, Updateable: false},
		},
	}
	makeFieldsMap(schema)
	return schema
}

func exclusiveSiblingChildSchema() *models.ObjectMetadata {
	schema := &models.ObjectMetadata{
		APIName:      "LineItem__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},
			{
				APIName:     "ParentId",
				Type:        constants.FieldTypeReference,
				ReferenceTo: []string{"Parent__c"},
				Createable:  true,
				Updateable:  true,
			},
			{APIName: "IsPrimary__c", Type: constants.FieldTypeBoolean, Createable: true, Updateable: true},
			{APIName: "ExternalId__c", Type: constants.FieldTypeString, Createable: true, Updateable: true},
			{APIName: constants.FieldLastModifiedByID, Type: constants.FieldTypeReference, ReferenceTo: []string{constants.TableUser}, Createable: false, Updateable: false},
			{APIName: constants.FieldLastModifiedDate, Type: constants.FieldTypeDatetime, Createable: false, Updateable: false},
			{APIName: constants.FieldIsDeleted, Type: constants.FieldTypeBoolean, Createable: false, Updateable: false},
		},
		ExclusiveSiblingFields: []models.ExclusiveSiblingField{
			{FieldAPIName: "IsPrimary__c", PartitionFieldAPIName: "ParentId"},
		},
	}
	makeFieldsMap(schema)
	return schema
}

func exclusiveSiblingChildSchemaWithFrozenPrimarySnapshot() *models.ObjectMetadata {
	schema := exclusiveSiblingChildSchema()
	formula := "IF(IsPrimary__c, 'primary', 'secondary')"
	formulaMode := "frozen"
	returnType := models.FieldType(constants.FieldTypeString)
	schema.Fields = append(schema.Fields, models.FieldMetadata{
		ID:          "field-primary-snapshot",
		APIName:     "PrimarySnapshot__c",
		Type:        constants.FieldTypeFormula,
		Formula:     &formula,
		FormulaMode: &formulaMode,
		ReturnType:  &returnType,
		Createable:  false,
		Updateable:  false,
	})
	makeFieldsMap(schema)
	return schema
}

func exclusiveSiblingChildSchemaWithUniqueExternalID() *models.ObjectMetadata {
	schema := exclusiveSiblingChildSchema()
	for i := range schema.Fields {
		if schema.Fields[i].APIName == "ExternalId__c" {
			schema.Fields[i].IsUnique = true
			break
		}
	}
	makeFieldsMap(schema)
	return schema
}

func exclusiveSiblingChildSchemaWithUniquePrimarySnapshot() *models.ObjectMetadata {
	schema := exclusiveSiblingChildSchemaWithFrozenPrimarySnapshot()
	for i := range schema.Fields {
		if schema.Fields[i].APIName == "PrimarySnapshot__c" {
			schema.Fields[i].IsUnique = true
			break
		}
	}
	makeFieldsMap(schema)
	return schema
}

func exclusiveSiblingChildSchemaWithUniqueChainedPrimarySnapshot() *models.ObjectMetadata {
	schema := exclusiveSiblingChildSchemaWithFrozenPrimarySnapshot()
	formula := "PrimarySnapshot__c"
	formulaMode := "frozen"
	returnType := models.FieldType(constants.FieldTypeString)
	schema.Fields = append(schema.Fields, models.FieldMetadata{
		ID:          "field-primary-code",
		APIName:     "PrimaryCode__c",
		Type:        constants.FieldTypeFormula,
		Formula:     &formula,
		FormulaMode: &formulaMode,
		ReturnType:  &returnType,
		Createable:  false,
		Updateable:  false,
		IsUnique:    true,
	})
	makeFieldsMap(schema)
	return schema
}

func exclusiveSiblingChildSchemaWithUniquePrimaryOrExternal() *models.ObjectMetadata {
	schema := exclusiveSiblingChildSchema()
	formula := "IF(IsPrimary__c, 'primary', ExternalId__c)"
	formulaMode := "frozen"
	returnType := models.FieldType(constants.FieldTypeString)
	schema.Fields = append(schema.Fields, models.FieldMetadata{
		ID:          "field-primary-or-external",
		APIName:     "PrimaryOrExternal__c",
		Type:        constants.FieldTypeFormula,
		Formula:     &formula,
		FormulaMode: &formulaMode,
		ReturnType:  &returnType,
		Createable:  false,
		Updateable:  false,
		IsUnique:    true,
	})
	makeFieldsMap(schema)
	return schema
}

func exclusiveSiblingChildSchemaWithFrozenParentNameSnapshot() *models.ObjectMetadata {
	schema := exclusiveSiblingChildSchema()
	formula := "Parent.Name"
	formulaMode := "frozen"
	returnType := models.FieldType(constants.FieldTypeString)
	schema.Fields = append(schema.Fields, models.FieldMetadata{
		ID:          "field-parent-name-snapshot",
		APIName:     "ParentNameSnapshot__c",
		Type:        constants.FieldTypeFormula,
		Formula:     &formula,
		FormulaMode: &formulaMode,
		ReturnType:  &returnType,
		Createable:  false,
		Updateable:  false,
	})
	makeFieldsMap(schema)
	return schema
}

func newExclusiveSiblingRuntimeService(repo *exclusiveSiblingRuntimeRepo) *PersistenceService {
	ps := newRollupBoundaryTestService(repo, exclusiveSiblingParentSchema(), exclusiveSiblingChildSchema())
	ps.txManager = &recordingBulkTxManager{}
	return ps
}

func exclusiveSiblingRuntimeRows() map[string]map[string]models.SObject {
	return map[string]map[string]models.SObject{
		"Parent__c": {
			"parent-1": {constants.FieldID: "parent-1", constants.FieldName: "Parent 1"},
			"parent-2": {constants.FieldID: "parent-2", constants.FieldName: "Parent 2"},
		},
		"LineItem__c": {
			"line-1": {
				constants.FieldID:        "line-1",
				constants.FieldName:      "Line 1",
				"ParentId":               "parent-1",
				"IsPrimary__c":           true,
				"ExternalId__c":          "ext-1",
				constants.FieldIsDeleted: false,
			},
			"line-2": {
				constants.FieldID:        "line-2",
				constants.FieldName:      "Line 2",
				"ParentId":               "parent-1",
				"IsPrimary__c":           false,
				"ExternalId__c":          "ext-2",
				constants.FieldIsDeleted: false,
			},
		},
	}
}

func exclusiveSiblingSystemContext() context.Context {
	return constants.WithSecurityContext(context.Background(), constants.SystemContext, "exclusive sibling runtime test")
}

func TestExclusiveSiblingInsertTrueClearsPriorSibling(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)

	inserted, err := ps.Insert(exclusiveSiblingSystemContext(), "LineItem__c", models.SObject{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
	}, recordEntrypointUser())
	if err != nil {
		t.Fatalf("Insert: %v", err)
	}
	id := exclusiveSiblingRecordID(inserted[constants.FieldID])
	if id != "line-3" {
		t.Fatalf("insert id = %q, want line-3", id)
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-1"][constants.FieldLastModifiedByID] != constants.SystemUserID {
		t.Fatalf("line-1 LastModifiedById = %v, want %s", repo.rows["LineItem__c"]["line-1"][constants.FieldLastModifiedByID], constants.SystemUserID)
	}
	if repo.rows["LineItem__c"]["line-3"]["IsPrimary__c"] != true {
		t.Fatalf("line-3 IsPrimary__c = %v, want true", repo.rows["LineItem__c"]["line-3"]["IsPrimary__c"])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"insert:LineItem__c:line-3",
		"reset:LineItem__c:IsPrimary__c:parent-1:line-3",
	)
}

func TestExclusiveSiblingInsertFalseAndOmittedAreNoops(t *testing.T) {
	for _, tc := range []struct {
		name   string
		record models.SObject
	}{
		{
			name: "false",
			record: models.SObject{
				constants.FieldID:   "line-3",
				constants.FieldName: "Line 3",
				"ParentId":          "parent-1",
				"IsPrimary__c":      false,
			},
		},
		{
			name: "omitted",
			record: models.SObject{
				constants.FieldID:   "line-4",
				constants.FieldName: "Line 4",
				"ParentId":          "parent-1",
			},
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
			ps := newExclusiveSiblingRuntimeService(repo)

			if _, err := ps.Insert(exclusiveSiblingSystemContext(), "LineItem__c", tc.record, recordEntrypointUser()); err != nil {
				t.Fatalf("Insert: %v", err)
			}
			if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != true {
				t.Fatalf("line-1 IsPrimary__c = %v, want true", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
			}
			if len(repo.resets) != 0 {
				t.Fatalf("unexpected resets: %#v", repo.resets)
			}
			recordID, _ := tc.record[constants.FieldID].(string)
			assertEventOrder(t, repo.events,
				"lock:Parent__c:parent-1",
				"insert:LineItem__c:"+recordID,
			)
			assertNoEventPrefix(t, repo.events, "reset:LineItem__c:")
		})
	}
}

func TestExclusiveSiblingInsertMissingParentFailsBeforeInsert(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)

	_, err := ps.Insert(exclusiveSiblingSystemContext(), "LineItem__c", models.SObject{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "missing-parent",
		"IsPrimary__c":      true,
	}, recordEntrypointUser())
	if err == nil {
		t.Fatal("Insert returned nil error, want missing parent failure")
	}
	assertNoEventPrefix(t, repo.events, "insert:LineItem__c:")
	assertNoEventPrefix(t, repo.events, "reset:LineItem__c:")
}

func TestExclusiveSiblingUpdateTrueClearsPriorSiblingAndPrelocksParent(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)

	if err := ps.Update(exclusiveSiblingSystemContext(), "LineItem__c", "line-2", models.SObject{
		"IsPrimary__c": true,
	}, recordEntrypointUser()); err != nil {
		t.Fatalf("Update: %v", err)
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-1"][constants.FieldLastModifiedByID] != constants.SystemUserID {
		t.Fatalf("line-1 LastModifiedById = %v, want %s", repo.rows["LineItem__c"]["line-1"][constants.FieldLastModifiedByID], constants.SystemUserID)
	}
	if repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"] != true {
		t.Fatalf("line-2 IsPrimary__c = %v, want true", repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-2",
		"update:LineItem__c:line-2",
		"reset:LineItem__c:IsPrimary__c:parent-1:line-2",
	)
}

func TestExclusiveSiblingUpdatePartitionMoveClearsFinalPartitionOnly(t *testing.T) {
	rows := exclusiveSiblingRuntimeRows()
	rows["LineItem__c"]["line-1"]["ParentId"] = "parent-2"
	rows["LineItem__c"]["line-1"]["IsPrimary__c"] = true
	rows["LineItem__c"]["line-2"]["ParentId"] = "parent-1"
	rows["LineItem__c"]["line-2"]["IsPrimary__c"] = true
	repo := newExclusiveSiblingRuntimeRepo(rows)
	ps := newExclusiveSiblingRuntimeService(repo)

	if err := ps.Update(exclusiveSiblingSystemContext(), "LineItem__c", "line-2", models.SObject{
		"ParentId": "parent-2",
	}, recordEntrypointUser()); err != nil {
		t.Fatalf("Update: %v", err)
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"] != true {
		t.Fatalf("line-2 IsPrimary__c = %v, want true", repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"])
	}
	if len(repo.resets) != 1 || repo.resets[0].partitionValue != "parent-2" {
		t.Fatalf("resets = %#v, want exactly final parent-2 reset", repo.resets)
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-2",
		"lock:LineItem__c:line-2",
		"reset:LineItem__c:IsPrimary__c:parent-2:line-2",
	)
}

func TestExclusiveSiblingUpdateUnrelatedFieldOnWinnerPrelocksParentWithoutReset(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)

	if err := ps.Update(exclusiveSiblingSystemContext(), "LineItem__c", "line-1", models.SObject{
		constants.FieldName: "Line 1 renamed",
	}, recordEntrypointUser()); err != nil {
		t.Fatalf("Update: %v", err)
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != true {
		t.Fatalf("line-1 IsPrimary__c = %v, want true", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"] != false {
		t.Fatalf("line-2 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"])
	}
	if len(repo.resets) != 0 {
		t.Fatalf("unexpected resets: %#v", repo.resets)
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-1",
		"update:LineItem__c:line-1",
	)
}

func TestExclusiveSiblingUpdateClearingWinnerPrelocksParentWithoutReset(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)

	if err := ps.Update(exclusiveSiblingSystemContext(), "LineItem__c", "line-1", models.SObject{
		"IsPrimary__c": false,
	}, recordEntrypointUser()); err != nil {
		t.Fatalf("Update: %v", err)
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"] != false {
		t.Fatalf("line-2 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"])
	}
	if len(repo.resets) != 0 {
		t.Fatalf("unexpected resets: %#v", repo.resets)
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-1",
		"update:LineItem__c:line-1",
	)
}

func TestExclusiveSiblingUpsertByExternalIDUsesUpdatePath(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)

	_, created, err := ps.UpsertByExternalID(exclusiveSiblingSystemContext(), "LineItem__c", "ExternalId__c", "ext-2", models.SObject{
		"IsPrimary__c": true,
	}, recordEntrypointUser())
	if err != nil {
		t.Fatalf("UpsertByExternalID: %v", err)
	}
	if created {
		t.Fatal("UpsertByExternalID created = true, want update path")
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"] != true {
		t.Fatalf("line-2 IsPrimary__c = %v, want true", repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-2",
		"reset:LineItem__c:IsPrimary__c:parent-1:line-2",
	)
}

func TestExclusiveSiblingUpsertByExternalIDUsesInsertPath(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)

	inserted, created, err := ps.UpsertByExternalID(exclusiveSiblingSystemContext(), "LineItem__c", "ExternalId__c", "ext-3", models.SObject{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
	}, recordEntrypointUser())
	if err != nil {
		t.Fatalf("UpsertByExternalID: %v", err)
	}
	if !created {
		t.Fatal("UpsertByExternalID created = false, want insert path")
	}
	if exclusiveSiblingRecordID(inserted[constants.FieldID]) != "line-3" {
		t.Fatalf("inserted id = %v, want line-3", inserted[constants.FieldID])
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-3"]["IsPrimary__c"] != true {
		t.Fatalf("line-3 IsPrimary__c = %v, want true", repo.rows["LineItem__c"]["line-3"]["IsPrimary__c"])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"insert:LineItem__c:line-3",
		"reset:LineItem__c:IsPrimary__c:parent-1:line-3",
	)
}

func TestExclusiveSiblingDirectWritesUsePessimisticTransactions(t *testing.T) {
	tests := []struct {
		name string
		run  func(*PersistenceService) error
	}{
		{
			name: "insert",
			run: func(ps *PersistenceService) error {
				_, err := ps.Insert(exclusiveSiblingSystemContext(), "LineItem__c", models.SObject{
					constants.FieldID:   "line-3",
					constants.FieldName: "Line 3",
					"ParentId":          "parent-1",
					"IsPrimary__c":      true,
				}, recordEntrypointUser())
				return err
			},
		},
		{
			name: "update",
			run: func(ps *PersistenceService) error {
				return ps.Update(exclusiveSiblingSystemContext(), "LineItem__c", "line-2", models.SObject{
					"IsPrimary__c": true,
				}, recordEntrypointUser())
			},
		},
		{
			name: "upsert",
			run: func(ps *PersistenceService) error {
				_, _, err := ps.UpsertByExternalID(exclusiveSiblingSystemContext(), "LineItem__c", "ExternalId__c", "ext-2", models.SObject{
					"IsPrimary__c": true,
				}, recordEntrypointUser())
				return err
			},
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
			ps := newExclusiveSiblingRuntimeService(repo)
			txm := &recordingBulkTxManager{}
			ps.txManager = txm

			if err := tc.run(ps); err != nil {
				t.Fatalf("%s: %v", tc.name, err)
			}
			assertBulkPessimisticOnly(t, txm)
		})
	}
}

func TestExclusiveSiblingDefaultChildRowsUsePessimisticRoot(t *testing.T) {
	parentSchema := exclusiveSiblingParentSchema()
	parentSchema.DefaultChildRows = []models.DefaultChildRow{{
		TargetObject:      "LineItem__c",
		TargetLookupField: "ParentId",
		FieldValues: models.SObject{
			constants.FieldName: "Default Line",
			"IsPrimary__c":      true,
		},
	}}
	makeFieldsMap(parentSchema)
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newRollupBoundaryTestService(repo, parentSchema, exclusiveSiblingChildSchema())
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	if _, err := ps.Insert(exclusiveSiblingSystemContext(), "Parent__c", models.SObject{
		constants.FieldID:   "parent-3",
		constants.FieldName: "Parent 3",
	}, recordEntrypointUser()); err != nil {
		t.Fatalf("Insert parent with exclusive-sibling default child rows: %v", err)
	}
	assertBulkPessimisticOnly(t, txm)
}

func TestExclusiveSiblingChildCloneTargetRequiresPessimisticRoot(t *testing.T) {
	rootSchema := childCloneParentSchemaForBoundaryTest()
	rootSchema.ChildCloneRules[0].TargetObject = "LineItem__c"
	rootSchema.ChildCloneRules[0].TargetLookupField = "ParentId"
	ps := newRollupBoundaryTestService(&captureRecordRepo{}, rootSchema, templateTaskSchemaForBoundaryTest(), exclusiveSiblingChildSchema())

	if !ps.rollupInsertRequiresPessimisticBoundary(exclusiveSiblingSystemContext(), "Quote__c", models.SObject{
		"SourceQuoteId__c": "source-quote-001",
	}) {
		t.Fatal("rollupInsertRequiresPessimisticBoundary(child clone target with ExclusiveSiblingFields) = false, want true")
	}
}

func TestExclusiveSiblingCommitPreparedUpdateUsesParentLockResetPath(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)

	err := ps.CommitPreparedUpdate(exclusiveSiblingSystemContext(), "LineItem__c", "line-2", &models.PreparedRecordUpdate{
		Record: models.SObject{
			constants.FieldID: "line-2",
			"ParentId":        "parent-1",
			"IsPrimary__c":    true,
		},
		EffectiveUpdates: models.SObject{"IsPrimary__c": true},
	}, nil, recordEntrypointUser())
	if err != nil {
		t.Fatalf("CommitPreparedUpdate: %v", err)
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"] != true {
		t.Fatalf("line-2 IsPrimary__c = %v, want true", repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-2",
		"update:LineItem__c:line-2",
		"reset:LineItem__c:IsPrimary__c:parent-1:line-2",
		"update:LineItem__c:line-1",
	)
}

func TestExclusiveSiblingCommitPreparedUpdateUnrelatedFieldOnWinnerPrelocksParentWithoutReset(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)

	err := ps.CommitPreparedUpdate(exclusiveSiblingSystemContext(), "LineItem__c", "line-1", &models.PreparedRecordUpdate{
		Record: models.SObject{
			constants.FieldID:   "line-1",
			constants.FieldName: "Line 1 renamed",
			"ParentId":          "parent-1",
			"IsPrimary__c":      true,
		},
		EffectiveUpdates: models.SObject{constants.FieldName: "Line 1 renamed"},
	}, nil, recordEntrypointUser())
	if err != nil {
		t.Fatalf("CommitPreparedUpdate: %v", err)
	}
	if repo.rows["LineItem__c"]["line-1"][constants.FieldName] != "Line 1 renamed" {
		t.Fatalf("line-1 Name = %v, want renamed", repo.rows["LineItem__c"]["line-1"][constants.FieldName])
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != true {
		t.Fatalf("line-1 IsPrimary__c = %v, want true", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if len(repo.resets) != 0 {
		t.Fatalf("unexpected resets: %#v", repo.resets)
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-1",
		"update:LineItem__c:line-1",
	)
}

func TestExclusiveSiblingCommitPreparedUpdateInsidePessimisticOuterTransactionFailsClosed(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	err := ps.RunInPessimisticTransaction(exclusiveSiblingSystemContext(), func(txCtx context.Context) error {
		return ps.CommitPreparedUpdate(txCtx, "LineItem__c", "line-2", &models.PreparedRecordUpdate{
			Record: models.SObject{
				constants.FieldID: "line-2",
				"ParentId":        "parent-1",
				"IsPrimary__c":    true,
			},
			EffectiveUpdates: models.SObject{"IsPrimary__c": true},
		}, nil, recordEntrypointUser())
	})
	if err == nil {
		t.Fatal("CommitPreparedUpdate inside pessimistic outer transaction returned nil, want fail-closed validation error")
	}
	if !appErrors.IsValidation(err) {
		t.Fatalf("CommitPreparedUpdate inside pessimistic outer transaction error = %T %v, want validation error", err, err)
	}
	if txm.pessimisticCalls != 1 {
		t.Fatalf("pessimistic transaction calls = %d, want only the caller-owned transaction", txm.pessimisticCalls)
	}
	if len(repo.events) != 0 {
		t.Fatalf("repo events = %s, want no mutation before fail-closed guard", strings.Join(repo.events, ", "))
	}
}

func TestExclusiveSiblingCommitPreparedUpdatePrelocksLookupParentWhenLockedOldWinnerDiffersFromPreload(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	lockCount := 0
	repo.onGetLock = func(tableName, id string, row models.SObject) {
		if tableName != "LineItem__c" || id != "line-2" || row == nil {
			return
		}
		lockCount++
		if lockCount == 1 {
			row["ParentId"] = "parent-1"
			row["IsPrimary__c"] = true
		}
	}
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	err := ps.CommitPreparedUpdate(exclusiveSiblingSystemContext(), "LineItem__c", "line-2", &models.PreparedRecordUpdate{
		Record: models.SObject{
			constants.FieldID:   "line-2",
			constants.FieldName: "Line 2 renamed",
			"ParentId":          "parent-1",
			"IsPrimary__c":      true,
		},
		EffectiveUpdates: models.SObject{constants.FieldName: "Line 2 renamed"},
	}, nil, recordEntrypointUser())
	if err != nil {
		t.Fatalf("CommitPreparedUpdate after stale old-partition preload: %v", err)
	}
	if txm.pessimisticCalls != 1 {
		t.Fatalf("pessimistic transaction calls = %d, want single transaction after lookup parent prelock", txm.pessimisticCalls)
	}
	if repo.rows["LineItem__c"]["line-2"][constants.FieldName] != "Line 2 renamed" {
		t.Fatalf("line-2 Name = %v, want renamed", repo.rows["LineItem__c"]["line-2"][constants.FieldName])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-2",
		"update:LineItem__c:line-2",
	)
}

func TestLookupParentCommitPreparedUpdateRetriesWhenLockedOldParentDiffersFromPreload(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	lockCount := 0
	repo.onGetLock = func(tableName, id string, row models.SObject) {
		if tableName != "LineItem__c" || id != "line-2" || row == nil {
			return
		}
		lockCount++
		if lockCount == 1 {
			row["ParentId"] = "parent-2"
		}
	}
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	err := ps.CommitPreparedUpdate(exclusiveSiblingSystemContext(), "LineItem__c", "line-2", &models.PreparedRecordUpdate{
		EffectiveUpdates: models.SObject{constants.FieldName: "Line 2 renamed"},
	}, nil, recordEntrypointUser())
	if err != nil {
		t.Fatalf("CommitPreparedUpdate after stale lookup parent preload: %v", err)
	}
	if txm.pessimisticCalls != 2 {
		t.Fatalf("pessimistic transaction calls = %d, want retry after locked old lookup parent changed", txm.pessimisticCalls)
	}
	if repo.rows["LineItem__c"]["line-2"][constants.FieldName] != "Line 2 renamed" {
		t.Fatalf("line-2 Name = %v, want renamed", repo.rows["LineItem__c"]["line-2"][constants.FieldName])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-2",
		"lock:Parent__c:parent-2",
		"lock:LineItem__c:line-2",
		"update:LineItem__c:line-2",
	)
}

func TestExclusiveSiblingUpdateInsidePessimisticOuterTransactionUsesCallerBoundary(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	err := ps.RunInPessimisticTransaction(exclusiveSiblingSystemContext(), func(txCtx context.Context) error {
		return ps.Update(txCtx, "LineItem__c", "line-2", models.SObject{
			"IsPrimary__c": true,
		}, recordEntrypointUser())
	})
	if err != nil {
		t.Fatalf("Update inside pessimistic outer transaction: %v", err)
	}
	if txm.pessimisticCalls != 1 {
		t.Fatalf("pessimistic transaction calls = %d, want only the caller-owned transaction", txm.pessimisticCalls)
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"] != true {
		t.Fatalf("line-2 IsPrimary__c = %v, want true", repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-2",
		"update:LineItem__c:line-2",
		"reset:LineItem__c:IsPrimary__c:parent-1:line-2",
		"update:LineItem__c:line-1",
	)
}

func TestExclusiveSiblingUpdatePrelocksLookupParentWhenLockedOldWinnerDiffersFromPreload(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	lockCount := 0
	repo.onGetLock = func(tableName, id string, row models.SObject) {
		if tableName != "LineItem__c" || id != "line-2" || row == nil {
			return
		}
		lockCount++
		if lockCount == 1 {
			row["ParentId"] = "parent-1"
			row["IsPrimary__c"] = true
		}
	}
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	err := ps.Update(exclusiveSiblingSystemContext(), "LineItem__c", "line-2", models.SObject{
		constants.FieldName: "Line 2 renamed",
	}, recordEntrypointUser())
	if err != nil {
		t.Fatalf("Update after stale old-partition preload: %v", err)
	}
	if txm.pessimisticCalls != 1 {
		t.Fatalf("pessimistic transaction calls = %d, want single transaction after lookup parent prelock", txm.pessimisticCalls)
	}
	if repo.rows["LineItem__c"]["line-2"][constants.FieldName] != "Line 2 renamed" {
		t.Fatalf("line-2 Name = %v, want renamed", repo.rows["LineItem__c"]["line-2"][constants.FieldName])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-2",
		"update:LineItem__c:line-2",
	)
}

func TestLookupParentUpdateRetriesWhenLockedOldParentDiffersFromPreload(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	lockCount := 0
	repo.onGetLock = func(tableName, id string, row models.SObject) {
		if tableName != "LineItem__c" || id != "line-2" || row == nil {
			return
		}
		lockCount++
		if lockCount == 1 {
			row["ParentId"] = "parent-2"
		}
	}
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	err := ps.Update(exclusiveSiblingSystemContext(), "LineItem__c", "line-2", models.SObject{
		constants.FieldName: "Line 2 renamed",
	}, recordEntrypointUser())
	if err != nil {
		t.Fatalf("Update after stale lookup parent preload: %v", err)
	}
	if txm.pessimisticCalls != 2 {
		t.Fatalf("pessimistic transaction calls = %d, want retry after locked old lookup parent changed", txm.pessimisticCalls)
	}
	if repo.rows["LineItem__c"]["line-2"][constants.FieldName] != "Line 2 renamed" {
		t.Fatalf("line-2 Name = %v, want renamed", repo.rows["LineItem__c"]["line-2"][constants.FieldName])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-2",
		"lock:Parent__c:parent-2",
		"lock:LineItem__c:line-2",
		"update:LineItem__c:line-2",
	)
}

func TestLookupParentUpdatePrelockDoesNotValidateUnchangedLookup(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	repo.hideUserFindOne = map[string]bool{"Parent__c": true}
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	err := ps.runGroupMembershipAwarePessimisticTransactionForObjects(context.Background(), []string{"LineItem__c"}, func(txCtx context.Context) error {
		_, err := ps.prelockNonPolymorphicLookupUpdateParents(txCtx, "LineItem__c", exclusiveSiblingChildSchema(), "line-2", models.SObject{
			constants.FieldName: "Line 2 renamed",
		}, recordEntrypointUser(), nil)
		return err
	})
	if err != nil {
		t.Fatalf("prelock unchanged lookup update: %v", err)
	}
	assertEventOrder(t, repo.events)
}

func TestLookupParentUpdatePrelockLocksChangedLookupParents(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	err := ps.runGroupMembershipAwarePessimisticTransactionForObjects(context.Background(), []string{"LineItem__c"}, func(txCtx context.Context) error {
		_, err := ps.prelockNonPolymorphicLookupUpdateParents(txCtx, "LineItem__c", exclusiveSiblingChildSchema(), "line-2", models.SObject{
			"ParentId": "parent-2",
		}, recordEntrypointUser(), nil)
		return err
	})
	if err != nil {
		t.Fatalf("prelock changed lookup update: %v", err)
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:Parent__c:parent-2",
	)
}

func TestLookupParentPreparedUpdatePrelockDoesNotValidateUnchangedLookup(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	repo.hideUserFindOne = map[string]bool{"Parent__c": true}
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	err := ps.runGroupMembershipAwarePessimisticTransactionForObjects(context.Background(), []string{"LineItem__c"}, func(txCtx context.Context) error {
		_, err := ps.prelockNonPolymorphicLookupPreparedUpdateParents(txCtx, "LineItem__c", "line-2", exclusiveSiblingChildSchema(), &models.PreparedRecordUpdate{
			EffectiveUpdates: models.SObject{constants.FieldName: "Line 2 renamed"},
		}, nil, recordEntrypointUser(), nil)
		return err
	})
	if err != nil {
		t.Fatalf("prelock unchanged lookup prepared update: %v", err)
	}
	assertEventOrder(t, repo.events)
}

func TestLookupParentPreparedUpdatePrelockLocksChangedLookupParents(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	err := ps.runGroupMembershipAwarePessimisticTransactionForObjects(context.Background(), []string{"LineItem__c"}, func(txCtx context.Context) error {
		_, err := ps.prelockNonPolymorphicLookupPreparedUpdateParents(txCtx, "LineItem__c", "line-2", exclusiveSiblingChildSchema(), &models.PreparedRecordUpdate{
			EffectiveUpdates: models.SObject{"ParentId": "parent-2"},
		}, nil, recordEntrypointUser(), nil)
		return err
	})
	if err != nil {
		t.Fatalf("prelock changed lookup prepared update: %v", err)
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:Parent__c:parent-2",
	)
}

func TestExclusiveSiblingUpdateInsidePessimisticOuterTransactionPrelocksBeforeSaveMutationParent(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm
	beforeUpdateCount := 0
	ps.eventBus = mutatingRecordEventBus{mutate: func(eventType sharedEvents.EventType, record models.SObject) {
		if eventType != sharedEvents.RecordBeforeUpdate || exclusiveSiblingRecordID(record[constants.FieldID]) != "line-2" {
			return
		}
		beforeUpdateCount++
		record["ParentId"] = "parent-1"
		record["IsPrimary__c"] = true
	}}

	err := ps.RunInPessimisticTransaction(exclusiveSiblingSystemContext(), func(txCtx context.Context) error {
		return ps.Update(txCtx, "LineItem__c", "line-2", models.SObject{
			constants.FieldName: "Line 2 renamed",
		}, recordEntrypointUser())
	})
	if err != nil {
		t.Fatalf("Update inside pessimistic outer transaction after before-save ES mutation: %v", err)
	}
	if beforeUpdateCount != 1 {
		t.Fatalf("RecordBeforeUpdate calls = %d, want single update body after lookup parent prelock", beforeUpdateCount)
	}
	if txm.pessimisticCalls != 1 {
		t.Fatalf("pessimistic transaction calls = %d, want caller-owned transaction reused without retry", txm.pessimisticCalls)
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"] != true {
		t.Fatalf("line-2 IsPrimary__c = %v, want true", repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-2",
		"update:LineItem__c:line-2",
		"reset:LineItem__c:IsPrimary__c:parent-1:line-2",
		"update:LineItem__c:line-1",
	)
}

func TestExclusiveSiblingUpdateInsidePlainOuterTransactionFailsClosed(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	err := ps.txManager.WithTransaction(exclusiveSiblingSystemContext(), func(txCtx context.Context) error {
		return ps.Update(txCtx, "LineItem__c", "line-2", models.SObject{
			"IsPrimary__c": true,
		}, recordEntrypointUser())
	})
	if err == nil {
		t.Fatal("Update inside plain outer transaction returned nil, want fail-closed validation error")
	}
	if !appErrors.IsValidation(err) {
		t.Fatalf("Update inside plain outer transaction error = %T %v, want validation error", err, err)
	}
	if len(repo.events) != 0 {
		t.Fatalf("repo events = %s, want no mutation before fail-closed guard", strings.Join(repo.events, ", "))
	}
}

func TestExclusiveSiblingUpsertInsidePessimisticOuterTransactionUsesCallerBoundary(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)
	txm := &recordingBulkTxManager{}
	ps.txManager = txm

	err := ps.RunInPessimisticTransaction(exclusiveSiblingSystemContext(), func(txCtx context.Context) error {
		_, created, err := ps.UpsertByExternalID(txCtx, "LineItem__c", "ExternalId__c", "ext-2", models.SObject{
			"IsPrimary__c": true,
		}, recordEntrypointUser())
		if err != nil {
			return err
		}
		if created {
			t.Fatal("UpsertByExternalID created row, want update branch")
		}
		return nil
	})
	if err != nil {
		t.Fatalf("UpsertByExternalID inside pessimistic outer transaction: %v", err)
	}
	if txm.pessimisticCalls != 1 {
		t.Fatalf("pessimistic transaction calls = %d, want only the caller-owned transaction", txm.pessimisticCalls)
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"] != true {
		t.Fatalf("line-2 IsPrimary__c = %v, want true", repo.rows["LineItem__c"]["line-2"]["IsPrimary__c"])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"lock:LineItem__c:line-2",
		"update:LineItem__c:line-2",
		"reset:LineItem__c:IsPrimary__c:parent-1:line-2",
		"update:LineItem__c:line-1",
	)
}

func exclusiveSiblingParentWithDefaultChildRowsSchema() *models.ObjectMetadata {
	parentSchema := exclusiveSiblingParentSchema()
	parentSchema.DefaultChildRows = []models.DefaultChildRow{{
		TargetObject:      "LineItem__c",
		TargetLookupField: "ParentId",
		FieldValues: models.SObject{
			constants.FieldName: "Default Line",
			"IsPrimary__c":      true,
		},
	}}
	makeFieldsMap(parentSchema)
	return parentSchema
}

func newGeneratedExclusiveSiblingRuntimeService(repo *exclusiveSiblingRuntimeRepo) *PersistenceService {
	return newRollupBoundaryTestService(repo, exclusiveSiblingParentWithDefaultChildRowsSchema(), exclusiveSiblingChildSchema())
}

func TestExclusiveSiblingCollectionSurfacesFailClosed(t *testing.T) {
	tests := []struct {
		name string
		run  func(*PersistenceService) error
	}{
		{
			name: "create records all or none",
			run: func(ps *PersistenceService) error {
				_, err := ps.CreateRecords(exclusiveSiblingSystemContext(), []models.SObject{{
					"attributes":        map[string]interface{}{"type": "LineItem__c"},
					constants.FieldName: "Line 3",
					"ParentId":          "parent-1",
					"IsPrimary__c":      true,
				}}, true, recordEntrypointUser())
				return err
			},
		},
		{
			name: "create records partial success",
			run: func(ps *PersistenceService) error {
				_, err := ps.CreateRecords(exclusiveSiblingSystemContext(), []models.SObject{{
					"attributes":        map[string]interface{}{"type": "LineItem__c"},
					constants.FieldName: "Line 3",
					"ParentId":          "parent-1",
					"IsPrimary__c":      true,
				}}, false, recordEntrypointUser())
				return err
			},
		},
		{
			name: "update records all or none",
			run: func(ps *PersistenceService) error {
				_, err := ps.UpdateRecords(exclusiveSiblingSystemContext(), []models.SObject{{
					"attributes":      map[string]interface{}{"type": "LineItem__c"},
					constants.FieldID: "line-2",
					"IsPrimary__c":    true,
				}}, true, recordEntrypointUser())
				return err
			},
		},
		{
			name: "update records partial success",
			run: func(ps *PersistenceService) error {
				_, err := ps.UpdateRecords(exclusiveSiblingSystemContext(), []models.SObject{{
					"attributes":      map[string]interface{}{"type": "LineItem__c"},
					constants.FieldID: "line-2",
					"IsPrimary__c":    true,
				}}, false, recordEntrypointUser())
				return err
			},
		},
		{
			name: "upsert records all or none",
			run: func(ps *PersistenceService) error {
				_, err := ps.UpsertRecords(exclusiveSiblingSystemContext(), "LineItem__c", "ExternalId__c", []models.SObject{{
					"ExternalId__c": "ext-2",
					"IsPrimary__c":  true,
				}}, true, recordEntrypointUser())
				return err
			},
		},
		{
			name: "upsert records partial success",
			run: func(ps *PersistenceService) error {
				_, err := ps.UpsertRecords(exclusiveSiblingSystemContext(), "LineItem__c", "ExternalId__c", []models.SObject{{
					"ExternalId__c": "ext-2",
					"IsPrimary__c":  true,
				}}, false, recordEntrypointUser())
				return err
			},
		},
		{
			name: "create tree nodes",
			run: func(ps *PersistenceService) error {
				_, err := ps.CreateTreeNodes(exclusiveSiblingSystemContext(), []contract.TreeNode{{
					ObjectType:  "LineItem__c",
					ReferenceID: "lineRef",
					Fields: contract.Record{
						constants.FieldName: "Line 3",
						"ParentId":          "parent-1",
						"IsPrimary__c":      true,
					},
				}}, recordEntrypointUser())
				return err
			},
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
			ps := newExclusiveSiblingRuntimeService(repo)

			err := tc.run(ps)
			if err == nil {
				t.Fatal("collection surface returned nil error, want fail-closed validation error")
			}
			if !appErrors.IsValidation(err) {
				t.Fatalf("collection surface error = %T %v, want validation error", err, err)
			}
			if len(repo.events) != 0 {
				t.Fatalf("repo events = %s, want no mutation before fail-closed guard", strings.Join(repo.events, ", "))
			}
		})
	}
}

func TestExclusiveSiblingGeneratedChildSharedWriteSurfacesFailClosed(t *testing.T) {
	tests := []struct {
		name string
		run  func(*PersistenceService) error
	}{
		{
			name: "create records all or none default child rows",
			run: func(ps *PersistenceService) error {
				_, err := ps.CreateRecords(exclusiveSiblingSystemContext(), []models.SObject{{
					"attributes":        map[string]interface{}{"type": "Parent__c"},
					constants.FieldName: "Parent 3",
				}}, true, recordEntrypointUser())
				return err
			},
		},
		{
			name: "upsert records all or none default child rows",
			run: func(ps *PersistenceService) error {
				_, err := ps.UpsertRecords(exclusiveSiblingSystemContext(), "Parent__c", "ExternalId__c", []models.SObject{{
					"ExternalId__c":     "parent-ext-3",
					constants.FieldName: "Parent 3",
				}}, true, recordEntrypointUser())
				return err
			},
		},
		{
			name: "create tree nodes default child rows",
			run: func(ps *PersistenceService) error {
				_, err := ps.CreateTreeNodes(exclusiveSiblingSystemContext(), []contract.TreeNode{{
					ObjectType:  "Parent__c",
					ReferenceID: "parentRef",
					Fields:      contract.Record{constants.FieldName: "Parent 3"},
				}}, recordEntrypointUser())
				return err
			},
		},
		{
			name: "bulk insert default child rows",
			run: func(ps *PersistenceService) error {
				_, err := ps.BulkInsert(exclusiveSiblingSystemContext(), "Parent__c", []models.SObject{{
					constants.FieldID:   "parent-3",
					constants.FieldName: "Parent 3",
				}}, recordEntrypointUser(), BulkInsertOptions{})
				return err
			},
		},
		{
			name: "create records all or none child clone target",
			run: func(ps *PersistenceService) error {
				rootSchema := childCloneParentSchemaForBoundaryTest()
				rootSchema.ChildCloneRules[0].TargetObject = "LineItem__c"
				rootSchema.ChildCloneRules[0].TargetLookupField = "ParentId"
				ps.metadata = metadataFeature.NewMetadataService(&stubUpsertMetaRepo{
					schemas: []*models.ObjectMetadata{rootSchema, templateTaskSchemaForBoundaryTest(), exclusiveSiblingChildSchema()},
				}, nil, nil)
				_, err := ps.CreateRecords(exclusiveSiblingSystemContext(), []models.SObject{{
					"attributes":        map[string]interface{}{"type": "Quote__c"},
					constants.FieldName: "Quote 3",
				}}, true, recordEntrypointUser())
				return err
			},
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
			ps := newGeneratedExclusiveSiblingRuntimeService(repo)

			err := tc.run(ps)
			if err == nil {
				t.Fatal("generated-child shared write surface returned nil error, want fail-closed validation error")
			}
			if !appErrors.IsValidation(err) {
				t.Fatalf("generated-child shared write surface error = %T %v, want validation error", err, err)
			}
			if len(repo.events) != 0 {
				t.Fatalf("repo events = %s, want no mutation before fail-closed guard", strings.Join(repo.events, ", "))
			}
		})
	}
}

func TestExclusiveSiblingResetSkipsLosingSiblingRecordAutomation(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)
	beforeUpdateSeen := false
	ps.eventBus = mutatingRecordEventBus{mutate: func(eventType sharedEvents.EventType, record models.SObject) {
		if eventType != sharedEvents.RecordBeforeUpdate || exclusiveSiblingRecordID(record[constants.FieldID]) != "line-1" {
			return
		}
		beforeUpdateSeen = true
		record[constants.FieldName] = "Line 1 lifecycle"
	}}

	_, err := ps.Insert(exclusiveSiblingSystemContext(), "LineItem__c", models.SObject{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
	}, recordEntrypointUser())
	if err != nil {
		t.Fatalf("Insert: %v", err)
	}
	if beforeUpdateSeen {
		t.Fatal("auto-cleared sibling ran RecordBeforeUpdate lifecycle")
	}
	if repo.rows["LineItem__c"]["line-1"][constants.FieldName] != "Line 1" {
		t.Fatalf("line-1 Name = %v, want reset to skip before-update mutation", repo.rows["LineItem__c"]["line-1"][constants.FieldName])
	}
	if repo.rows["LineItem__c"]["line-1"][constants.FieldLastModifiedByID] != constants.SystemUserID {
		t.Fatalf("line-1 LastModifiedById = %v, want %s", repo.rows["LineItem__c"]["line-1"][constants.FieldLastModifiedByID], constants.SystemUserID)
	}
}

func TestExclusiveSiblingBulkInsertLastLandedWinnerClearsExistingAndSameBatch(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)

	result, err := ps.BulkInsert(exclusiveSiblingSystemContext(), "LineItem__c", []models.SObject{{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
	}, {
		constants.FieldID:   "line-4",
		constants.FieldName: "Line 4",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
	}}, recordEntrypointUser(), BulkInsertOptions{})
	if err != nil {
		t.Fatalf("BulkInsert: %v", err)
	}
	if result.SuccessCount != 2 || result.FailedCount != 0 {
		t.Fatalf("BulkInsert result = %+v, want 2 successes", result)
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-3"]["IsPrimary__c"] != false {
		t.Fatalf("line-3 IsPrimary__c = %v, want same-batch loser false", repo.rows["LineItem__c"]["line-3"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-4"]["IsPrimary__c"] != true {
		t.Fatalf("line-4 IsPrimary__c = %v, want last landed winner true", repo.rows["LineItem__c"]["line-4"]["IsPrimary__c"])
	}
	if len(repo.resets) != 1 || repo.resets[0].winnerID != "line-4" {
		t.Fatalf("resets = %#v, want exactly one reset plan with line-4 winner", repo.resets)
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:parent-1",
		"bulk_insert:LineItem__c:line-3,line-4",
		"update:LineItem__c:line-3",
		"reset:LineItem__c:IsPrimary__c:parent-1:line-4",
		"update:LineItem__c:line-1",
	)
}

func TestExclusiveSiblingBulkInsertIgnoredDuplicateDoesNotClearExistingWinner(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	repo.landingOverride = []bool{false}
	ps := newExclusiveSiblingRuntimeService(repo)

	result, err := ps.BulkInsert(exclusiveSiblingSystemContext(), "LineItem__c", []models.SObject{{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
	}}, recordEntrypointUser(), BulkInsertOptions{IgnoreDuplicates: true})
	if err != nil {
		t.Fatalf("BulkInsert ignored duplicate: %v", err)
	}
	if result.SuccessCount != 1 || result.FailedCount != 0 {
		t.Fatalf("BulkInsert result = %+v, want legacy ignored duplicate success accounting", result)
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != true {
		t.Fatalf("line-1 IsPrimary__c = %v, want existing winner unchanged", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if _, ok := repo.rows["LineItem__c"]["line-3"]; ok {
		t.Fatal("line-3 landed despite landingOverride=false")
	}
	if len(repo.resets) != 0 {
		t.Fatalf("unexpected reset plans for non-landed duplicate: %#v", repo.resets)
	}
}

func TestExclusiveSiblingBulkInsertMissingParentDropsOnlyThatRowBeforeLanding(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	ps := newExclusiveSiblingRuntimeService(repo)

	result, err := ps.BulkInsert(exclusiveSiblingSystemContext(), "LineItem__c", []models.SObject{{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "missing-parent",
		"IsPrimary__c":      true,
	}, {
		constants.FieldID:   "line-4",
		constants.FieldName: "Line 4",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
	}}, recordEntrypointUser(), BulkInsertOptions{SkipValidation: true})
	if err != nil {
		t.Fatalf("BulkInsert with one missing parent: %v", err)
	}
	if result.SuccessCount != 1 || result.FailedCount != 1 {
		t.Fatalf("BulkInsert result = %+v, want one success and one failed parent lock", result)
	}
	if _, ok := repo.rows["LineItem__c"]["line-3"]; ok {
		t.Fatal("line-3 landed despite missing exclusive sibling parent")
	}
	if repo.rows["LineItem__c"]["line-4"]["IsPrimary__c"] != true {
		t.Fatalf("line-4 IsPrimary__c = %v, want surviving winner true", repo.rows["LineItem__c"]["line-4"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want cleared by surviving winner", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	assertEventOrder(t, repo.events,
		"lock:Parent__c:missing-parent",
		"lock:Parent__c:parent-1",
		"bulk_insert:LineItem__c:line-4",
		"reset:LineItem__c:IsPrimary__c:parent-1:line-4",
	)
}

func TestExclusiveSiblingBulkInsertBatchUniquenessIgnoresParentPrelockDroppedRows(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	repo.rows["Parent__c"]["parent-3"] = models.SObject{constants.FieldID: "parent-3", constants.FieldName: "Parent 3"}
	repo.onGetLock = func(tableName, id string, _ models.SObject) {
		if tableName == "Parent__c" && id == "parent-3" {
			delete(repo.rows["Parent__c"], "parent-3")
		}
	}
	childSchema := exclusiveSiblingChildSchemaWithUniqueExternalID()
	ps := newRollupBoundaryTestService(repo, exclusiveSiblingParentSchema(), childSchema)
	ps.txManager = &recordingBulkTxManager{}

	result, err := ps.BulkInsert(exclusiveSiblingSystemContext(), "LineItem__c", []models.SObject{{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "parent-3",
		"IsPrimary__c":      true,
		"ExternalId__c":     "duplicate-ext",
	}, {
		constants.FieldID:   "line-4",
		constants.FieldName: "Line 4",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
		"ExternalId__c":     "duplicate-ext",
	}}, recordEntrypointUser(), BulkInsertOptions{})
	if err != nil {
		t.Fatalf("BulkInsert with prelock-dropped duplicate: %v result=%+v", err, result)
	}
	if result.SuccessCount != 1 || result.FailedCount != 1 {
		t.Fatalf("BulkInsert result = %+v, want one success and one failed parent lock", result)
	}
	if _, ok := repo.rows["LineItem__c"]["line-3"]; ok {
		t.Fatal("line-3 landed despite parent disappearing before ES lock")
	}
	if repo.rows["LineItem__c"]["line-4"]["ExternalId__c"] != "duplicate-ext" {
		t.Fatalf("line-4 ExternalId__c = %v, want duplicate-ext", repo.rows["LineItem__c"]["line-4"]["ExternalId__c"])
	}
}

func TestExclusiveSiblingBulkInsertBatchUniquenessDropsDuplicateSurvivorBeforeInsert(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	childSchema := exclusiveSiblingChildSchemaWithUniqueExternalID()
	ps := newRollupBoundaryTestService(repo, exclusiveSiblingParentSchema(), childSchema)
	ps.txManager = &recordingBulkTxManager{}

	result, err := ps.BulkInsert(exclusiveSiblingSystemContext(), "LineItem__c", []models.SObject{{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "parent-1",
		"IsPrimary__c":      false,
		"ExternalId__c":     "duplicate-ext",
	}, {
		constants.FieldID:   "line-4",
		constants.FieldName: "Line 4",
		"ParentId":          "parent-1",
		"IsPrimary__c":      false,
		"ExternalId__c":     "duplicate-ext",
	}}, recordEntrypointUser(), BulkInsertOptions{})
	if err != nil {
		t.Fatalf("BulkInsert with duplicate survivor: %v result=%+v", err, result)
	}
	if result.SuccessCount != 1 || result.FailedCount != 1 {
		t.Fatalf("BulkInsert result = %+v, want one success and one duplicate failure", result)
	}
	if _, ok := repo.rows["LineItem__c"]["line-3"]; !ok {
		t.Fatal("line-3 did not land")
	}
	if _, ok := repo.rows["LineItem__c"]["line-4"]; ok {
		t.Fatal("line-4 landed despite duplicate batch unique value")
	}
	assertEventOrder(t, repo.events,
		"bulk_insert:LineItem__c:line-3",
	)
}

func TestExclusiveSiblingBulkInsertBatchUniquenessRecomputesWinnerAfterDroppingDuplicate(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	childSchema := exclusiveSiblingChildSchemaWithUniqueExternalID()
	ps := newRollupBoundaryTestService(repo, exclusiveSiblingParentSchema(), childSchema)
	ps.txManager = &recordingBulkTxManager{}

	result, err := ps.BulkInsert(exclusiveSiblingSystemContext(), "LineItem__c", []models.SObject{{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
		"ExternalId__c":     "duplicate-ext",
	}, {
		constants.FieldID:   "line-4",
		constants.FieldName: "Line 4",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
		"ExternalId__c":     "duplicate-ext",
	}}, recordEntrypointUser(), BulkInsertOptions{})
	if err != nil {
		t.Fatalf("BulkInsert with dropped duplicate winner: %v result=%+v", err, result)
	}
	if result.SuccessCount != 1 || result.FailedCount != 1 {
		t.Fatalf("BulkInsert result = %+v, want one success and one duplicate failure", result)
	}
	if repo.rows["LineItem__c"]["line-3"]["IsPrimary__c"] != true {
		t.Fatalf("line-3 IsPrimary__c = %v, want recomputed survivor winner true", repo.rows["LineItem__c"]["line-3"]["IsPrimary__c"])
	}
	if _, ok := repo.rows["LineItem__c"]["line-4"]; ok {
		t.Fatal("line-4 landed despite duplicate batch unique value")
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want cleared by recomputed survivor winner", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
}

func TestExclusiveSiblingBulkInsertSameBatchLoserRecomputesWriteMaterializedFormula(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	childSchema := exclusiveSiblingChildSchemaWithFrozenPrimarySnapshot()
	ps := newRollupBoundaryTestService(repo, exclusiveSiblingParentSchema(), childSchema)
	ps.txManager = &recordingBulkTxManager{}
	outboxRepo := &exclusiveSiblingOutboxRepo{}
	ps.outbox = eventFeature.NewOutboxService(outboxRepo, &eventFeature.NoOpEventBus{}, inlineTxManager{})

	result, err := ps.BulkInsert(exclusiveSiblingSystemContext(), "LineItem__c", []models.SObject{{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
	}, {
		constants.FieldID:   "line-4",
		constants.FieldName: "Line 4",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
	}}, recordEntrypointUser(), BulkInsertOptions{})
	if err != nil {
		t.Fatalf("BulkInsert with frozen formula: %v", err)
	}
	if result.SuccessCount != 2 || result.FailedCount != 0 {
		t.Fatalf("BulkInsert result = %+v, want 2 successes", result)
	}
	if repo.rows["LineItem__c"]["line-3"]["IsPrimary__c"] != false {
		t.Fatalf("line-3 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-3"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-3"]["PrimarySnapshot__c"] != "secondary" {
		t.Fatalf("line-3 PrimarySnapshot__c = %v, want secondary", repo.rows["LineItem__c"]["line-3"]["PrimarySnapshot__c"])
	}
	if repo.rows["LineItem__c"]["line-4"]["PrimarySnapshot__c"] != "primary" {
		t.Fatalf("line-4 PrimarySnapshot__c = %v, want primary", repo.rows["LineItem__c"]["line-4"]["PrimarySnapshot__c"])
	}

	loserEvent := exclusiveSiblingCreatedEventForRecord(t, outboxRepo.payloads, "line-3")
	if loserEvent.Record["IsPrimary__c"] != false {
		t.Fatalf("line-3 create payload IsPrimary__c = %v, want false", loserEvent.Record["IsPrimary__c"])
	}
	if loserEvent.Record["PrimarySnapshot__c"] != "secondary" {
		t.Fatalf("line-3 create payload PrimarySnapshot__c = %v, want secondary", loserEvent.Record["PrimarySnapshot__c"])
	}
}

func TestExclusiveSiblingBulkInsertBatchUniquenessUsesFinalFormulaImages(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	childSchema := exclusiveSiblingChildSchemaWithUniquePrimarySnapshot()
	ps := newRollupBoundaryTestService(repo, exclusiveSiblingParentSchema(), childSchema)
	ps.txManager = &recordingBulkTxManager{}

	result, err := ps.BulkInsert(exclusiveSiblingSystemContext(), "LineItem__c", []models.SObject{{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
	}, {
		constants.FieldID:   "line-4",
		constants.FieldName: "Line 4",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
	}}, recordEntrypointUser(), BulkInsertOptions{})
	if err != nil {
		t.Fatalf("BulkInsert with unique formula final images: %v result=%+v", err, result)
	}
	if result.SuccessCount != 2 || result.FailedCount != 0 {
		t.Fatalf("BulkInsert result = %+v, want 2 successes", result)
	}
	if repo.rows["LineItem__c"]["line-3"]["PrimarySnapshot__c"] != "secondary" {
		t.Fatalf("line-3 PrimarySnapshot__c = %v, want secondary", repo.rows["LineItem__c"]["line-3"]["PrimarySnapshot__c"])
	}
	if repo.rows["LineItem__c"]["line-4"]["PrimarySnapshot__c"] != "primary" {
		t.Fatalf("line-4 PrimarySnapshot__c = %v, want primary", repo.rows["LineItem__c"]["line-4"]["PrimarySnapshot__c"])
	}
}

func TestExclusiveSiblingBulkInsertUniqueFormulaAllowsExistingWinnerReplacement(t *testing.T) {
	tests := []struct {
		name        string
		childSchema func() *models.ObjectMetadata
		seedExtra   func(repo *exclusiveSiblingRuntimeRepo)
		assertExtra func(t *testing.T, repo *exclusiveSiblingRuntimeRepo)
		description string
	}{
		{
			name:        "Formula",
			childSchema: exclusiveSiblingChildSchemaWithUniquePrimarySnapshot,
			description: "unique formula winner",
		},
		{
			name:        "ChainedFormula",
			childSchema: exclusiveSiblingChildSchemaWithUniqueChainedPrimarySnapshot,
			seedExtra: func(repo *exclusiveSiblingRuntimeRepo) {
				repo.rows["LineItem__c"]["line-1"]["PrimaryCode__c"] = "primary"
			},
			assertExtra: func(t *testing.T, repo *exclusiveSiblingRuntimeRepo) {
				if repo.rows["LineItem__c"]["line-1"]["PrimaryCode__c"] != "secondary" {
					t.Fatalf("line-1 PrimaryCode__c = %v, want secondary", repo.rows["LineItem__c"]["line-1"]["PrimaryCode__c"])
				}
				if repo.rows["LineItem__c"]["line-3"]["PrimaryCode__c"] != "primary" {
					t.Fatalf("line-3 PrimaryCode__c = %v, want primary", repo.rows["LineItem__c"]["line-3"]["PrimaryCode__c"])
				}
			},
			description: "unique chained formula winner",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
			repo.rows["LineItem__c"]["line-1"]["PrimarySnapshot__c"] = "primary"
			if tt.seedExtra != nil {
				tt.seedExtra(repo)
			}
			ps := newRollupBoundaryTestService(repo, exclusiveSiblingParentSchema(), tt.childSchema())
			ps.txManager = &recordingBulkTxManager{}

			result, err := ps.BulkInsert(exclusiveSiblingSystemContext(), "LineItem__c", []models.SObject{{
				constants.FieldID:   "line-3",
				constants.FieldName: "Line 3",
				"ParentId":          "parent-1",
				"IsPrimary__c":      true,
			}}, recordEntrypointUser(), BulkInsertOptions{})
			if err != nil {
				t.Fatalf("BulkInsert replacing existing %s: %v result=%+v", tt.description, err, result)
			}
			if result.SuccessCount != 1 || result.FailedCount != 0 {
				t.Fatalf("BulkInsert result = %+v, want 1 success", result)
			}
			if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
				t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
			}
			if repo.rows["LineItem__c"]["line-1"]["PrimarySnapshot__c"] != "secondary" {
				t.Fatalf("line-1 PrimarySnapshot__c = %v, want secondary", repo.rows["LineItem__c"]["line-1"]["PrimarySnapshot__c"])
			}
			if repo.rows["LineItem__c"]["line-3"]["PrimarySnapshot__c"] != "primary" {
				t.Fatalf("line-3 PrimarySnapshot__c = %v, want primary", repo.rows["LineItem__c"]["line-3"]["PrimarySnapshot__c"])
			}
			if tt.assertExtra != nil {
				tt.assertExtra(t, repo)
			}
		})
	}
}

func TestExclusiveSiblingBulkInsertDeferredUniqueFormulaFiltersUnresolvedFalseBranchConflict(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	repo.rows["LineItem__c"]["line-1"]["PrimaryOrExternal__c"] = "primary"
	repo.rows["LineItem__c"]["line-2"]["PrimaryOrExternal__c"] = "ext-2"
	childSchema := exclusiveSiblingChildSchemaWithUniquePrimaryOrExternal()
	ps := newRollupBoundaryTestService(repo, exclusiveSiblingParentSchema(), childSchema)
	ps.txManager = &recordingBulkTxManager{}

	result, err := ps.BulkInsert(exclusiveSiblingSystemContext(), "LineItem__c", []models.SObject{{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "parent-2",
		"IsPrimary__c":      false,
		"ExternalId__c":     "ext-2",
	}, {
		constants.FieldID:   "line-4",
		constants.FieldName: "Line 4",
		"ParentId":          "parent-2",
		"IsPrimary__c":      false,
		"ExternalId__c":     "ext-4",
	}}, recordEntrypointUser(), BulkInsertOptions{})
	if err != nil {
		t.Fatalf("BulkInsert with unresolved deferred unique conflict: %v result=%+v", err, result)
	}
	if result.SuccessCount != 1 || result.FailedCount != 1 {
		t.Fatalf("BulkInsert result = %+v, want 1 success and 1 failure", result)
	}
	if _, ok := repo.rows["LineItem__c"]["line-3"]; ok {
		t.Fatalf("line-3 landed despite unresolved deferred unique conflict")
	}
	if repo.rows["LineItem__c"]["line-4"]["PrimaryOrExternal__c"] != "ext-4" {
		t.Fatalf("line-4 PrimaryOrExternal__c = %v, want ext-4", repo.rows["LineItem__c"]["line-4"]["PrimaryOrExternal__c"])
	}
	if len(result.Errors) != 1 || !strings.Contains(result.Errors[0], "PrimaryOrExternal__c") {
		t.Fatalf("BulkInsert errors = %+v, want PrimaryOrExternal__c conflict", result.Errors)
	}
}

func TestExclusiveSiblingBulkInsertResetPreservesFormulaOnRelationFetchError(t *testing.T) {
	repo := newExclusiveSiblingRuntimeRepo(exclusiveSiblingRuntimeRows())
	repo.rows["LineItem__c"]["line-1"]["ParentNameSnapshot__c"] = "stored-parent"
	repo.onResetLock = func() {
		repo.findOneErrByTable = map[string]error{
			"Parent__c": errors.New("transient parent fetch"),
		}
	}
	childSchema := exclusiveSiblingChildSchemaWithFrozenParentNameSnapshot()
	ps := newRollupBoundaryTestService(repo, exclusiveSiblingParentSchema(), childSchema)
	ps.txManager = &recordingBulkTxManager{}

	result, err := ps.BulkInsert(exclusiveSiblingSystemContext(), "LineItem__c", []models.SObject{{
		constants.FieldID:   "line-3",
		constants.FieldName: "Line 3",
		"ParentId":          "parent-1",
		"IsPrimary__c":      true,
	}}, recordEntrypointUser(), BulkInsertOptions{})
	if err != nil {
		t.Fatalf("BulkInsert with relation-fetch formula reset: %v result=%+v", err, result)
	}
	if result.SuccessCount != 1 || result.FailedCount != 0 {
		t.Fatalf("BulkInsert result = %+v, want 1 success", result)
	}
	if repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"] != false {
		t.Fatalf("line-1 IsPrimary__c = %v, want false", repo.rows["LineItem__c"]["line-1"]["IsPrimary__c"])
	}
	if repo.rows["LineItem__c"]["line-1"]["ParentNameSnapshot__c"] != "stored-parent" {
		t.Fatalf("line-1 ParentNameSnapshot__c = %v, want stored-parent", repo.rows["LineItem__c"]["line-1"]["ParentNameSnapshot__c"])
	}
}

func exclusiveSiblingCreatedEventForRecord(t *testing.T, payloads []interface{}, id string) eventFeature.RecordEventPayload {
	t.Helper()
	for _, payload := range payloads {
		envelope, ok := payload.(eventFeature.InternalRecordEventEnvelope)
		if !ok {
			t.Fatalf("outbox payload type = %T, want InternalRecordEventEnvelope", payload)
		}
		if exclusiveSiblingRecordID(envelope.PublicPayload.Record[constants.FieldID]) == id {
			return envelope.PublicPayload
		}
	}
	t.Fatalf("missing RecordCreated payload for %s; payloads=%#v", id, payloads)
	return eventFeature.RecordEventPayload{}
}

func assertEventOrder(t *testing.T, events []string, expected ...string) {
	t.Helper()
	next := 0
	for _, event := range events {
		if next < len(expected) && event == expected[next] {
			next++
		}
	}
	if next != len(expected) {
		t.Fatalf("events = %s, want order %s", strings.Join(events, ", "), strings.Join(expected, ", "))
	}
}

func assertNoEventPrefix(t *testing.T, events []string, prefix string) {
	t.Helper()
	for _, event := range events {
		if strings.HasPrefix(event, prefix) {
			t.Fatalf("events = %s, did not expect prefix %q", strings.Join(events, ", "), prefix)
		}
	}
}