package records

import (
	"context"
	stderrors "errors"
	"strings"
	"testing"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/kernel/pkg/validation"
	metadataFeature "github.com/nexuscrm/metadata"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/shared/pkg/constants"
	sharedEvents "github.com/nexuscrm/shared/pkg/events"
	"github.com/nexuscrm/shared/pkg/fieldprovenance"
	"github.com/nexuscrm/shared/pkg/models"
)

type siblingMaxCaptureRepo struct {
	captureRecordRepo
	maxValue          float64
	calls             int
	events            []string
	requireParentLock bool
}

func (r *siblingMaxCaptureRepo) MaxNumericFieldForParent(ctx context.Context, schema *models.ObjectMetadata, targetField, parentField string, parentValue interface{}) (float64, error) {
	r.calls++
	r.events = append(r.events, "max")
	if r.requireParentLock {
		key, ok, err := siblingMaxParentLockIntent(schema, parentField, parentValue)
		if err != nil {
			return 0, err
		}
		if ok {
			scope := recordlocks.ScopeFromContext(ctx)
			if scope == nil || !scope.HasAcquired(key) {
				return 0, stderrors.New("missing pre-admitted sibling-max parent lock")
			}
		}
	}
	return r.maxValue, nil
}

func (r *siblingMaxCaptureRepo) GetLock(ctx context.Context, tableName string, id string) (models.SObject, error) {
	r.events = append(r.events, "lock:"+tableName+":"+id)
	return r.captureRecordRepo.GetLock(ctx, tableName, id)
}

func siblingMaxTestSchema(t *testing.T) *models.ObjectMetadata {
	t.Helper()
	previous := fieldprovenance.Default()
	t.Cleanup(func() { fieldprovenance.SetDefault(previous) })
	fieldprovenance.SetDefault(fieldprovenance.NewClassifier(nil, nil, nil, nil))

	parent := "ParentId"
	schema := &models.ObjectMetadata{
		APIName:      "LineItem__c",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString, Createable: false, Updateable: false},
			{APIName: "ParentId", Type: constants.FieldTypeReference, ReferenceTo: []string{"Parent__c"}, Createable: true, Updateable: true},
			{APIName: "SortOrder__c", Type: models.FieldType("integer"), Createable: true, Updateable: true, SiblingMaxParentField: &parent},
			{APIName: constants.FieldIsDeleted, Type: constants.FieldTypeBoolean, Createable: false, Updateable: false},
		},
	}
	makeFieldsMap(schema)
	return schema
}

func TestApplySiblingMaxDefaultsAssignsMaxPlusOne(t *testing.T) {
	schema := siblingMaxTestSchema(t)
	repo := &siblingMaxCaptureRepo{maxValue: 6}
	ps := &PersistenceService{repo: repo}
	record := models.SObject{"ParentId": "parent-1"}

	defaultedFields, err := ps.applySiblingMaxDefaults(context.Background(), schema.APIName, record, schema)
	if err != nil {
		t.Fatalf("applySiblingMaxDefaults: %v", err)
	}
	if len(defaultedFields) != 1 || defaultedFields[0] != "SortOrder__c" {
		t.Fatalf("defaultedFields = %#v, want [SortOrder__c]", defaultedFields)
	}

	if got := record["SortOrder__c"]; got != int64(7) {
		t.Fatalf("SortOrder__c = %#v, want int64(7)", got)
	}
	if repo.calls != 1 {
		t.Fatalf("MaxNumericFieldForParent calls = %d, want 1", repo.calls)
	}
}

func TestApplySiblingMaxDefaultsUsesProjectedCapabilitiesWithoutL1Classifier(t *testing.T) {
	previous := fieldprovenance.Default()
	t.Cleanup(func() { fieldprovenance.SetDefault(previous) })
	fieldprovenance.SetDefault(fieldprovenance.NewClassifier(nil, nil, nil, nil))

	parent := "CampaignId"
	schema := &models.ObjectMetadata{
		APIName:      "CampaignMemberStatus",
		SharingModel: constants.SharingModelControlledByParent,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString},
			{APIName: parent, Type: constants.FieldTypeReference, ReferenceTo: []string{"Campaign"}, Createable: true, Updateable: true},
			{APIName: constants.FieldSortOrder, Type: models.FieldType("integer"), Createable: true, Updateable: true, SiblingMaxParentField: &parent},
		},
	}
	makeFieldsMap(schema)
	repo := &siblingMaxCaptureRepo{maxValue: 1}
	ps := &PersistenceService{repo: repo}
	record := models.SObject{parent: "campaign-1"}

	defaultedFields, err := ps.applySiblingMaxDefaults(context.Background(), schema.APIName, record, schema)
	if err != nil {
		t.Fatal(err)
	}
	if len(defaultedFields) != 1 || defaultedFields[0] != constants.FieldSortOrder || record[constants.FieldSortOrder] != int64(2) {
		t.Fatalf("defaulted fields = %#v record = %#v", defaultedFields, record)
	}
}

func TestPreviewBeforeInsertOmittedSiblingMaxUsesAdmittedParentScope(t *testing.T) {
	schema := siblingMaxTestSchema(t)
	repo := &siblingMaxCaptureRepo{
		captureRecordRepo: captureRecordRepo{
			old:    models.SObject{constants.FieldID: "parent-1"},
			exists: map[string]map[string]bool{"Parent__c": {"parent-1": true}},
		},
		maxValue:          6,
		requireParentLock: true,
	}
	ps := newSiblingMaxBulkTestService(schema, repo)
	controller := &fakeParentLaneAdmissionController{}
	ps.parentLaneAdmission = controller

	preview, err := ps.PreviewBeforeInsert(context.Background(), schema.APIName, models.SObject{"ParentId": "parent-1"}, recordEntrypointUser())
	if err != nil {
		t.Fatalf("PreviewBeforeInsert: %v", err)
	}
	if got := preview["SortOrder__c"]; got != int64(7) {
		t.Fatalf("preview SortOrder__c = %#v, want int64(7)", got)
	}
	wantEvents := []string{"lock:Parent__c:parent-1", "max"}
	if len(repo.events) != len(wantEvents) {
		t.Fatalf("events = %#v, want %#v", repo.events, wantEvents)
	}
	for i := range wantEvents {
		if repo.events[i] != wantEvents[i] {
			t.Fatalf("events = %#v, want %#v", repo.events, wantEvents)
		}
	}
	if len(controller.acquired) != 1 || controller.acquired[0].ObjectName != "Parent__c" || controller.acquired[0].RecordID != "parent-1" {
		t.Fatalf("parent-lane admissions = %#v, want Parent__c/parent-1", controller.acquired)
	}
	if len(repo.bulkInserted) != 0 || repo.inserted != nil {
		t.Fatalf("PreviewBeforeInsert wrote records: inserted=%#v bulk=%#v", repo.inserted, repo.bulkInserted)
	}
}

func TestPreviewBeforeInsertBeforeCreateSiblingMaxParentMutationUsesAdmittedScope(t *testing.T) {
	schema := siblingMaxTestSchema(t)
	repo := &siblingMaxCaptureRepo{
		captureRecordRepo: captureRecordRepo{
			old:    models.SObject{constants.FieldID: "parent-from-hook"},
			exists: map[string]map[string]bool{"Parent__c": {"parent-from-hook": true}},
		},
		maxValue:          8,
		requireParentLock: true,
	}
	ps := newSiblingMaxBulkTestService(schema, repo)
	controller := &fakeParentLaneAdmissionController{}
	ps.parentLaneAdmission = controller
	ps.eventBus = mutatingRecordEventBus{mutate: func(eventType sharedEvents.EventType, record models.SObject) {
		if eventType != sharedEvents.RecordBeforeCreate {
			return
		}
		record["ParentId"] = "parent-from-hook"
		delete(record, "SortOrder__c")
	}}

	preview, err := ps.PreviewBeforeInsert(context.Background(), schema.APIName, models.SObject{}, recordEntrypointUser())
	if err != nil {
		t.Fatalf("PreviewBeforeInsert: %v", err)
	}
	if got := preview["ParentId"]; got != "parent-from-hook" {
		t.Fatalf("preview ParentId = %#v, want parent-from-hook", got)
	}
	if got := preview["SortOrder__c"]; got != int64(9) {
		t.Fatalf("preview SortOrder__c = %#v, want int64(9)", got)
	}
	wantEvents := []string{"lock:Parent__c:parent-from-hook", "max"}
	if len(repo.events) != len(wantEvents) {
		t.Fatalf("events = %#v, want %#v", repo.events, wantEvents)
	}
	for i := range wantEvents {
		if repo.events[i] != wantEvents[i] {
			t.Fatalf("events = %#v, want %#v", repo.events, wantEvents)
		}
	}
	if len(controller.acquired) != 1 || controller.acquired[0].ObjectName != "Parent__c" || controller.acquired[0].RecordID != "parent-from-hook" {
		t.Fatalf("parent-lane admissions = %#v, want Parent__c/parent-from-hook", controller.acquired)
	}
}

func TestApplySiblingMaxDefaultsPrelocksParentBeforeMax(t *testing.T) {
	schema := siblingMaxTestSchema(t)
	repo := &siblingMaxCaptureRepo{
		captureRecordRepo: captureRecordRepo{old: models.SObject{constants.FieldID: "parent-001"}},
		maxValue:          4,
	}
	tx := &struct{ id string }{id: "tx"}
	ps := &PersistenceService{repo: repo, txManager: parentLockCoordinatorTxManager{active: true, tx: tx}, parentLaneAdmission: &fakeParentLaneAdmissionController{}}
	ctx := recordlocks.WithScopeForTransaction(parentLaneAdmissionTestContext(), tx)

	record := models.SObject{"ParentId": "parent-001"}
	defaultedFields, err := ps.applySiblingMaxDefaults(ctx, schema.APIName, record, schema)
	if err != nil {
		t.Fatalf("applySiblingMaxDefaults: %v", err)
	}

	if len(defaultedFields) != 1 || defaultedFields[0] != "SortOrder__c" {
		t.Fatalf("defaultedFields = %#v, want SortOrder__c", defaultedFields)
	}
	if record["SortOrder__c"] != int64(5) {
		t.Fatalf("SortOrder__c = %#v, want int64(5)", record["SortOrder__c"])
	}
	wantEvents := []string{"lock:Parent__c:parent-001", "max"}
	if len(repo.events) != len(wantEvents) {
		t.Fatalf("events = %#v, want %#v", repo.events, wantEvents)
	}
	for i := range wantEvents {
		if repo.events[i] != wantEvents[i] {
			t.Fatalf("events = %#v, want %#v", repo.events, wantEvents)
		}
	}
	scope := recordlocks.ScopeFromContext(ctx)
	if scope == nil || !scope.HasAcquired(recordlocks.ParentKey{ObjectName: "Parent__c", RecordID: "parent-001"}) {
		t.Fatalf("coordinator scope did not hold sibling-max parent lock")
	}
}

func TestApplySiblingMaxDefaultsPreviewOverlaysPriorSyntheticCreates(t *testing.T) {
	schema := siblingMaxTestSchema(t)
	repo := &siblingMaxCaptureRepo{maxValue: 4}
	ps := &PersistenceService{repo: repo}
	ctx := constants.WithPreviewSyntheticRecordRegistry(constants.WithInsertPreviewPreparation(context.Background()))
	constants.RegisterPreviewSyntheticRecord(ctx, "preview-line-1", schema.APIName, map[string]interface{}{
		constants.FieldID: "preview-line-1",
		"ParentId":        "parent-1",
		"SortOrder__c":    int64(5),
	})
	constants.RegisterPreviewSyntheticRecord(ctx, "preview-line-2", schema.APIName, map[string]interface{}{
		constants.FieldID: "preview-line-2",
		"ParentId":        "other-parent",
		"SortOrder__c":    int64(99),
	})
	record := models.SObject{"ParentId": "parent-1"}

	defaultedFields, err := ps.applySiblingMaxDefaults(ctx, schema.APIName, record, schema)
	if err != nil {
		t.Fatalf("applySiblingMaxDefaults: %v", err)
	}
	if len(defaultedFields) != 1 || defaultedFields[0] != "SortOrder__c" {
		t.Fatalf("defaultedFields = %#v, want [SortOrder__c]", defaultedFields)
	}

	if got := record["SortOrder__c"]; got != int64(6) {
		t.Fatalf("SortOrder__c = %#v, want int64(6)", got)
	}
	if repo.calls != 1 {
		t.Fatalf("MaxNumericFieldForParent calls = %d, want 1", repo.calls)
	}
}

func TestApplySiblingMaxDefaultsPreviewRejectsFractionalSyntheticMaximum(t *testing.T) {
	schema := siblingMaxTestSchema(t)
	repo := &siblingMaxCaptureRepo{maxValue: 4}
	ps := &PersistenceService{repo: repo}
	ctx := constants.WithPreviewSyntheticRecordRegistry(constants.WithInsertPreviewPreparation(context.Background()))
	constants.RegisterPreviewSyntheticRecord(ctx, "preview-line-1", schema.APIName, map[string]interface{}{
		constants.FieldID: "preview-line-1",
		"ParentId":        "parent-1",
		"SortOrder__c":    5.5,
	})
	record := models.SObject{"ParentId": "parent-1"}

	_, err := ps.applySiblingMaxDefaults(ctx, schema.APIName, record, schema)
	if err == nil {
		t.Fatal("applySiblingMaxDefaults error = nil, want field integrity error")
	}
	if !strings.Contains(err.Error(), "whole finite numbers") {
		t.Fatalf("error = %v, want whole finite numbers diagnostic", err)
	}
	if repo.calls != 1 {
		t.Fatalf("MaxNumericFieldForParent calls = %d, want 1", repo.calls)
	}
}

func TestApplySiblingMaxDefaultsPreservesExplicitValue(t *testing.T) {
	schema := siblingMaxTestSchema(t)
	repo := &siblingMaxCaptureRepo{maxValue: 6}
	ps := &PersistenceService{repo: repo}
	record := models.SObject{"ParentId": "parent-1", "SortOrder__c": 42}

	defaultedFields, err := ps.applySiblingMaxDefaults(context.Background(), schema.APIName, record, schema)
	if err != nil {
		t.Fatalf("applySiblingMaxDefaults: %v", err)
	}
	if len(defaultedFields) != 0 {
		t.Fatalf("defaultedFields = %#v, want empty for explicit value", defaultedFields)
	}

	if got := record["SortOrder__c"]; got != 42 {
		t.Fatalf("SortOrder__c = %#v, want explicit 42", got)
	}
	if repo.calls != 0 {
		t.Fatalf("MaxNumericFieldForParent calls = %d, want 0", repo.calls)
	}
}

func TestApplySiblingMaxDefaultsRejectsDriftedStaticDefaultEvenWithExplicitValue(t *testing.T) {
	schema := siblingMaxTestSchema(t)
	staticDefault := "1"
	models.FindField(schema, "SortOrder__c").DefaultValue = &staticDefault
	repo := &siblingMaxCaptureRepo{maxValue: 6}
	ps := &PersistenceService{repo: repo}
	record := models.SObject{"ParentId": "parent-1", "SortOrder__c": 42}

	_, err := ps.applySiblingMaxDefaults(context.Background(), schema.APIName, record, schema)
	if err == nil {
		t.Fatal("applySiblingMaxDefaults error = nil, want invalid metadata rejection")
	}
	if !strings.Contains(err.Error(), "defaultValue") {
		t.Fatalf("error = %v, want defaultValue diagnostic", err)
	}
	if repo.calls != 0 {
		t.Fatalf("MaxNumericFieldForParent calls = %d, want 0 for invalid metadata", repo.calls)
	}
}

func TestApplySiblingMaxDefaultsRejectsDriftedBlankDefault(t *testing.T) {
	schema := siblingMaxTestSchema(t)
	blankDefault := ""
	models.FindField(schema, "SortOrder__c").DefaultValue = &blankDefault
	repo := &siblingMaxCaptureRepo{maxValue: 6}
	ps := &PersistenceService{repo: repo}
	record := models.SObject{"ParentId": "parent-1"}

	_, err := ps.applySiblingMaxDefaults(context.Background(), schema.APIName, record, schema)
	if err == nil {
		t.Fatal("applySiblingMaxDefaults error = nil, want invalid blank defaultValue rejection")
	}
	if !strings.Contains(err.Error(), "defaultValue") {
		t.Fatalf("error = %v, want defaultValue diagnostic", err)
	}
	if repo.calls != 0 {
		t.Fatalf("MaxNumericFieldForParent calls = %d, want 0 after invalid blank defaultValue", repo.calls)
	}
}

func TestApplySiblingMaxDefaultsRejectsDriftedNonWriteableTarget(t *testing.T) {
	cases := []struct {
		name   string
		mutate func(*models.FieldMetadata)
	}{
		{
			name: "not_createable",
			mutate: func(field *models.FieldMetadata) {
				field.Createable = false
			},
		},
		{
			name: "not_updateable",
			mutate: func(field *models.FieldMetadata) {
				field.Updateable = false
			},
		},
		{
			name: "intrinsic_write_protection",
			mutate: func(field *models.FieldMetadata) {
				field.WriteProtectionPolicy = constants.FieldWriteProtectionTrustedOnly
			},
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			schema := siblingMaxTestSchema(t)
			tc.mutate(models.FindField(schema, "SortOrder__c"))
			repo := &siblingMaxCaptureRepo{maxValue: 6}
			ps := &PersistenceService{repo: repo}
			record := models.SObject{"ParentId": "parent-1"}

			_, err := ps.applySiblingMaxDefaults(context.Background(), schema.APIName, record, schema)
			if err == nil {
				t.Fatal("applySiblingMaxDefaults error = nil, want invalid non-writeable metadata rejection")
			}
			if !strings.Contains(err.Error(), "createable and updateable") {
				t.Fatalf("error = %v, want createable/updateable diagnostic", err)
			}
			if repo.calls != 0 {
				t.Fatalf("MaxNumericFieldForParent calls = %d, want 0 after invalid target capabilities", repo.calls)
			}
		})
	}
}

func TestApplySiblingMaxDefaultsRejectsDriftedDerivedParent(t *testing.T) {
	formula := "OwnerId"
	cases := []struct {
		name   string
		mutate func(*models.FieldMetadata)
	}{
		{
			name: "formula_reference",
			mutate: func(field *models.FieldMetadata) {
				field.Formula = &formula
			},
		},
		{
			name: "projection_reference",
			mutate: func(field *models.FieldMetadata) {
				field.ProjectionConfig = &models.FieldProjectionConfig{Kind: "lookup"}
			},
		},
		{
			name: "virtual_reference",
			mutate: func(field *models.FieldMetadata) {
				field.IsVirtual = true
				field.VirtualKind = constants.VirtualFieldCompound
			},
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			schema := siblingMaxTestSchema(t)
			tc.mutate(models.FindField(schema, "ParentId"))
			repo := &siblingMaxCaptureRepo{maxValue: 6}
			ps := &PersistenceService{repo: repo}
			record := models.SObject{"ParentId": "parent-1"}

			_, err := ps.applySiblingMaxDefaults(context.Background(), schema.APIName, record, schema)
			if err == nil {
				t.Fatal("applySiblingMaxDefaults error = nil, want invalid derived parent metadata rejection")
			}
			if !strings.Contains(err.Error(), "persisted non-polymorphic reference") {
				t.Fatalf("error = %v, want persisted non-polymorphic reference diagnostic", err)
			}
			if repo.calls != 0 {
				t.Fatalf("MaxNumericFieldForParent calls = %d, want 0 after invalid parent shape", repo.calls)
			}
		})
	}
}

func TestApplySiblingMaxDefaultsRejectsDriftedDecimalTarget(t *testing.T) {
	schema := siblingMaxTestSchema(t)
	models.FindField(schema, "SortOrder__c").Type = constants.FieldTypeDouble
	repo := &siblingMaxCaptureRepo{maxValue: 6}
	ps := &PersistenceService{repo: repo}
	record := models.SObject{"ParentId": "parent-1"}

	_, err := ps.applySiblingMaxDefaults(context.Background(), schema.APIName, record, schema)
	if err == nil {
		t.Fatal("applySiblingMaxDefaults error = nil, want invalid decimal target rejection")
	}
	if !strings.Contains(err.Error(), "integer-compatible numeric") {
		t.Fatalf("error = %v, want integer-compatible numeric diagnostic", err)
	}
	if repo.calls != 0 {
		t.Fatalf("MaxNumericFieldForParent calls = %d, want 0 after invalid target type", repo.calls)
	}
}

func TestApplySiblingMaxDefaultsRejectsFractionalExistingMaximum(t *testing.T) {
	schema := siblingMaxTestSchema(t)
	repo := &siblingMaxCaptureRepo{maxValue: 1.5}
	ps := &PersistenceService{repo: repo}
	record := models.SObject{"ParentId": "parent-1"}

	_, err := ps.applySiblingMaxDefaults(context.Background(), schema.APIName, record, schema)
	if err == nil {
		t.Fatal("applySiblingMaxDefaults error = nil, want fractional data rejection")
	}
	if !strings.Contains(err.Error(), "whole finite numbers") {
		t.Fatalf("error = %v, want whole finite diagnostic", err)
	}
	if repo.calls != 1 {
		t.Fatalf("MaxNumericFieldForParent calls = %d, want 1", repo.calls)
	}
	if _, ok := record["SortOrder__c"]; ok {
		t.Fatalf("SortOrder__c was set despite fractional existing maximum: %#v", record["SortOrder__c"])
	}
}

func TestApplySiblingMaxDefaultsRequiresParentWhenDefaulting(t *testing.T) {
	schema := siblingMaxTestSchema(t)
	ps := &PersistenceService{repo: &siblingMaxCaptureRepo{maxValue: 6}}
	record := models.SObject{}

	_, err := ps.applySiblingMaxDefaults(context.Background(), schema.APIName, record, schema)
	if err == nil {
		t.Fatal("applySiblingMaxDefaults error = nil, want field integrity error")
	}
	var wire *appErrors.DomainWireError
	if !stderrors.As(err, &wire) || wire.Kind != appErrors.WireFieldIntegrity {
		t.Fatalf("error = %T %[1]v, want field integrity", err)
	}
	if !strings.Contains(err.Error(), "ParentId") {
		t.Fatalf("error = %v, want parent field in message", err)
	}
}

func TestValidateSiblingMaxDefaultParentForCreateMasterDetailUsesLookupGate(t *testing.T) {
	childSchema := siblingMaxTestSchema(t)
	parentField := models.FindField(childSchema, "ParentId")
	parentField.Type = constants.FieldTypeMasterDetail
	parentField.ReferenceTo = []string{"Parent__c"}
	parentSchema := &models.ObjectMetadata{
		APIName:      "Parent__c",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString},
			{APIName: constants.FieldName, Type: constants.FieldTypeString},
		},
	}
	makeFieldsMap(parentSchema)
	metaSvc := metadataFeature.NewMetadataService(&stubUpsertMetaRepo{
		schemas: []*models.ObjectMetadata{childSchema, parentSchema},
	}, nil, nil)
	permSvc := permissions.NewPermissionService(&stubUpsertPermRepo{grantCreate: true}, metaSvc, nil)
	ps := &PersistenceService{
		repo:        &captureRecordRepo{},
		metadata:    metaSvc,
		permissions: permSvc,
	}

	err := ps.ValidateSiblingMaxDefaultParentForCreate(context.Background(), childSchema.APIName, "ParentId", "missing-parent", &models.UserSession{
		ID:        "user-standard",
		ProfileID: constants.ProfileStandardUser,
	})
	if err == nil {
		t.Fatal("ValidateSiblingMaxDefaultParentForCreate accepted missing master-detail parent")
	}
	if !strings.Contains(err.Error(), "not accessible or does not exist") {
		t.Fatalf("error = %v, want lookup access/existence diagnostic", err)
	}
}

func TestRejectOmittedSiblingMaxDefaultsForBulk(t *testing.T) {
	schema := siblingMaxTestSchema(t)

	err := rejectOmittedSiblingMaxDefaultsForBulk(models.SObject{"ParentId": "parent-1"}, schema)
	if err == nil {
		t.Fatal("rejectOmittedSiblingMaxDefaultsForBulk error = nil, want field integrity error")
	}
	var wire *appErrors.DomainWireError
	if !stderrors.As(err, &wire) || wire.Kind != appErrors.WireFieldIntegrity {
		t.Fatalf("error = %T %[1]v, want field integrity", err)
	}

	if err := rejectOmittedSiblingMaxDefaultsForBulk(models.SObject{"ParentId": "parent-1", "SortOrder__c": 1}, schema); err != nil {
		t.Fatalf("explicit sibling value should pass bulk preflight: %v", err)
	}
}

func newSiblingMaxBulkTestService(schema *models.ObjectMetadata, repo *siblingMaxCaptureRepo) *PersistenceService {
	metaSvc := metadataFeature.NewMetadataService(&stubUpsertMetaRepo{schema: schema}, nil, nil)
	permSvc := permissions.NewPermissionService(&stubUpsertPermRepo{grantCreate: true, grantEdit: true}, metaSvc, nil)
	ps := &PersistenceService{
		repo:                repo,
		rollup:              noopRollupProcessor{},
		metadata:            metaSvc,
		permissions:         permSvc,
		eventBus:            mutatingRecordEventBus{},
		validator:           metadataFeature.NewValidationService(formula.NewEngine()),
		schemaValidator:     validation.NewSchemaValidator(),
		txManager:           inlineTxManager{},
		cacheInvalidations:  newRecordCacheInvalidations(),
		writeCanonicalizers: newRecordWriteCanonicalizers(),
	}
	ps.registerBuiltInSecurityCacheInvalidations()
	ps.registerBuiltInSecurityWriteCanonicalizers()
	ps.registerBuiltInSecurityWriteValidators()
	return ps
}

func TestBulkInsertSiblingMaxFinalImageRejectsBeforeSaveInvalidTarget(t *testing.T) {
	cases := []struct {
		name             string
		mutate           func(models.SObject)
		wantErrSubstring string
	}{
		{
			name: "cleared",
			mutate: func(record models.SObject) {
				delete(record, "SortOrder__c")
			},
			wantErrSubstring: "final image omitted sibling-max default field SortOrder__c",
		},
		{
			name: "non_numeric",
			mutate: func(record models.SObject) {
				record["SortOrder__c"] = "not-a-number"
			},
			wantErrSubstring: "final image has invalid sibling-max default field SortOrder__c",
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			schema := siblingMaxTestSchema(t)
			repo := &siblingMaxCaptureRepo{
				captureRecordRepo: captureRecordRepo{
					findOneFn: func(_ context.Context, tableName string, id string) (models.SObject, error) {
						if tableName == "Parent__c" && id == "parent-1" {
							return models.SObject{constants.FieldID: id}, nil
						}
						return nil, nil
					},
				},
				maxValue: 6,
			}
			ps := newSiblingMaxBulkTestService(schema, repo)

			var beforeCreateCount int
			ps.eventBus = mutatingRecordEventBus{mutate: func(eventType sharedEvents.EventType, record models.SObject) {
				if eventType != sharedEvents.RecordBeforeCreate {
					return
				}
				beforeCreateCount++
				tc.mutate(record)
			}}

			result, err := ps.BulkInsert(context.Background(), schema.APIName, []models.SObject{{
				"ParentId":     "parent-1",
				"SortOrder__c": 10,
			}}, recordEntrypointUser(), BulkInsertOptions{})
			if err == nil {
				t.Fatal("BulkInsert err = nil; want final-image sibling-max rejection")
			}
			if result.SuccessCount != 0 || result.FailedCount != 1 {
				t.Fatalf("BulkInsert result = %+v; want one final-image sibling-max failure", result)
			}
			if beforeCreateCount != 1 {
				t.Fatalf("RecordBeforeCreate count = %d; want 1 (err=%v result=%+v)", beforeCreateCount, err, result)
			}
			if len(result.Errors) != 1 || !strings.Contains(result.Errors[0], tc.wantErrSubstring) {
				t.Fatalf("result.Errors = %#v; want %q", result.Errors, tc.wantErrSubstring)
			}
			if len(repo.bulkInserted) != 0 {
				t.Fatalf("bulkInserted = %#v; want no SQL after final-image sibling-max rejection", repo.bulkInserted)
			}
		})
	}
}