package records

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

	eventFeature "github.com/nexuscrm/events"
	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	domainSchema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/events"
	"github.com/nexuscrm/shared/pkg/formulastorage"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/recordsguard"
)

// BulkInsertOptions configures bulk insert behavior
type BulkInsertOptions struct {
	BatchSize         int  // Records per batch (default 100)
	SkipValidation    bool // Skip record validation (for trusted imports)
	SkipFlows         bool // Skip BeforeCreate flows (faster import)
	SkipAutoNumbers   bool // Skip auto-number generation (if pre-assigned)
	IgnoreDuplicates  bool // Use INSERT IGNORE to skip duplicate errors
	AllowSystemFields bool // Allow Id and audit fields in input (for data migration)
	NormalizeFromCSV  bool // Run NormalizeSObject to fix case and coerce string types (CSV bulk path)
	// AllowLegacyListViewBulkIngest is only for worker-side processing of
	// _System_ListView ingest jobs that were persisted before the public
	// create entrypoints started rejecting them. Do not set it on live request
	// paths or direct generic BulkInsert callers.
	AllowLegacyListViewBulkIngest bool
}

// BulkInsertResult contains the result of a bulk insert operation
type BulkInsertResult struct {
	SuccessCount int      `json:"success_count"`
	FailedCount  int      `json:"failed_count"`
	Errors       []string `json:"errors,omitempty"`
}

// bulkRecordRejection is one rejected record inside a batch, captured while
// the typed error is still intact.
//
// The record map is retained by reference rather than copied: a rejected
// record is abandoned by every path that records one (they all `continue`),
// so nothing mutates it afterwards, and copying every rejected row in a large
// failing batch would be pure cost.
type bulkRecordRejection struct {
	record models.SObject
	err    error
}

// bulkRejectionLog is BulkInsert's per-call rejection accumulator.
//
// It lives OUTSIDE BulkInsertResult on purpose. The result is an exported
// wire shape returned to callers; adding an unexported field to it would
// break any external package constructing it with an unkeyed composite
// literal, and the accumulator is internal bookkeeping rather than part of
// the answer. Keeping them separate also means rejection sites hold the log,
// not the result — so FailedCount cannot be incremented anywhere except
// through rejectRecordWithMessage, which is what makes the capture invariant
// structural rather than a convention a future site has to remember.
//
// Errors and rejections are parallel by construction: one append each, in the
// same method, or neither.
type bulkRejectionLog struct {
	errors     []string
	rejections []bulkRecordRejection
}

// count is the number of rejected records recorded so far.
func (l *bulkRejectionLog) count() int { return len(l.errors) }

// mark returns an opaque position to rewind to. The bulk transaction body is
// re-run on retry and must discard the attempt's captures; without this, a
// record rejected on attempt 1 that then COMMITTED on attempt 2 would keep a
// stale rejection.
func (l *bulkRejectionLog) mark() int { return len(l.errors) }

func (l *bulkRejectionLog) rewind(mark int) {
	// Validate against BOTH slices. They are parallel by construction, but a
	// rewind runs inside the transaction closure, so a future divergence must
	// not surface as a mid-transaction slice-bounds panic.
	if l == nil || mark < 0 || mark > len(l.errors) || mark > len(l.rejections) {
		return
	}
	l.errors = l.errors[:mark]
	l.rejections = l.rejections[:mark]
}

// applyTo publishes the accumulated failures onto the caller-visible result.
func (l *bulkRejectionLog) applyTo(result *BulkInsertResult) {
	if l == nil || result == nil {
		return
	}
	result.FailedCount = len(l.errors)
	result.Errors = append([]string(nil), l.errors...)
}

// rejectRecordWithMessage is the SINGLE place a bulk record is counted as
// failed. Callers that just want the standard "record N: <err>" rendering
// should use rejectRecord.
//
// err may be nil for the one rejection that has no error object (a caller
// supplied permission virtuals with no permission service configured); the
// audit classifier simply produces no row for it, which is correct — that is
// a service-wiring failure, not a user validation rejection.
func (l *bulkRejectionLog) rejectRecordWithMessage(message string, record models.SObject, err error) {
	if l == nil {
		return
	}
	l.errors = append(l.errors, message)
	l.rejections = append(l.rejections, bulkRecordRejection{record: record, err: err})
}

// rejectRecord records a failed record with the standard message rendering.
func (l *bulkRejectionLog) rejectRecord(recordIndex int, record models.SObject, err error) {
	l.rejectRecordWithMessage(fmt.Sprintf("record %d: %v", recordIndex, err), record, err)
}

// bulkRecordAt is the bounds-safe record lookup used by the parent-lock and
// exclusive-sibling helpers, which reject by index into their own slice. A
// missing record degrades the audit row's identity to the unassigned
// sentinel; it never panics the caller's request.
func bulkRecordAt(records []models.SObject, index int) models.SObject {
	if index < 0 || index >= len(records) {
		return nil
	}
	return records[index]
}

func appendBulkInsertRecordError(log *bulkRejectionLog, recordIndex int, record models.SObject, err error) {
	if log == nil {
		return
	}
	message := err.Error()
	if appErrors.IsInternal(err) {
		message = sanitizedInternalErrorMessage
	}
	log.rejectRecordWithMessage(fmt.Sprintf("record %d: %s", recordIndex, message), record, err)
}

// appendBulkInsertBatchUniqueError records an in-batch duplicate on a unique
// field. Unlike a database uniqueness conflict, this case never constructed an
// error object at all — the message was synthesized directly. It now builds a
// real ConflictError so the rejection classifies as `uniqueness` like every
// other uniqueness rejection; the emitted message is unchanged.
//
// The ConflictError is deliberately built WITHOUT the conflicting value: the
// audit path reads only .Field, and leaving Value empty means no value exists
// on the object the audit path can see. The user-facing message below still
// renders the value, as it always has.
func appendBulkInsertBatchUniqueError(log *bulkRejectionLog, recordIndex int, objectName string, record models.SObject, fieldName string, value interface{}) {
	if log == nil {
		return
	}
	log.rejectRecordWithMessage(
		fmt.Sprintf("record %d: duplicate value '%s' for unique field %s within batch", recordIndex, fmt.Sprintf("%v", value), fieldName),
		record,
		appErrors.NewConflictError(objectName, fieldName, ""),
	)
}

func filterBulkInsertValidRecords(records []models.SObject, oldIndexMap, newIndexMap []int) []models.SObject {
	if len(records) == 0 || len(oldIndexMap) == 0 || len(newIndexMap) == len(oldIndexMap) {
		return records
	}
	byOriginalIndex := make(map[int]models.SObject, len(oldIndexMap))
	for i, originalIndex := range oldIndexMap {
		if i < len(records) {
			byOriginalIndex[originalIndex] = records[i]
		}
	}
	filtered := make([]models.SObject, 0, len(newIndexMap))
	for _, originalIndex := range newIndexMap {
		if record, ok := byOriginalIndex[originalIndex]; ok {
			filtered = append(filtered, record)
		}
	}
	return filtered
}

func cloneBulkInsertSObjects(records []models.SObject) []models.SObject {
	out := make([]models.SObject, len(records))
	for i, record := range records {
		out[i] = cloneSObject(record)
	}
	return out
}

func allBulkInsertRowsLanded(records []models.SObject) []bool {
	landedRows := make([]bool, len(records))
	for i := range landedRows {
		landedRows[i] = true
	}
	return landedRows
}

func filterBulkInsertRecordsByFailed(records []models.SObject, failed map[int]struct{}) []models.SObject {
	if len(records) == 0 || len(failed) == 0 {
		return records
	}
	filtered := make([]models.SObject, 0, len(records)-len(failed))
	for i, record := range records {
		if _, ok := failed[i]; ok {
			continue
		}
		filtered = append(filtered, record)
	}
	return filtered
}

func filterBulkInsertIndexMapByFailed(indexMap []int, failed map[int]struct{}) []int {
	if len(indexMap) == 0 || len(failed) == 0 {
		return indexMap
	}
	filtered := make([]int, 0, len(indexMap)-len(failed))
	for i, originalIndex := range indexMap {
		if _, ok := failed[i]; ok {
			continue
		}
		filtered = append(filtered, originalIndex)
	}
	return filtered
}

func createFromSourceGuardMutationPayload(before, after models.SObject, candidateFields map[string]struct{}) models.SObject {
	if len(before) == 0 || len(after) == 0 || len(candidateFields) == 0 {
		return nil
	}
	payload := models.SObject{}
	for fieldName := range candidateFields {
		beforeValue := guardSourceFKString(before[fieldName])
		afterValue := guardSourceFKString(after[fieldName])
		if afterValue == "" || afterValue == beforeValue {
			continue
		}
		payload[fieldName] = after[fieldName]
	}
	if len(payload) == 0 {
		return nil
	}
	return payload
}

func createFromSourceGuardCandidatePayload(data models.SObject, candidateFields map[string]struct{}) models.SObject {
	if len(data) == 0 || len(candidateFields) == 0 {
		return nil
	}
	payload := models.SObject{}
	for fieldName := range candidateFields {
		if guardSourceFKString(data[fieldName]) == "" {
			continue
		}
		payload[fieldName] = data[fieldName]
	}
	if len(payload) == 0 {
		return nil
	}
	return payload
}

func (ps *PersistenceService) filterExclusiveSiblingBulkInsertFinalUniqueRows(ctx context.Context, schema *models.ObjectMetadata, storageRecords, validRecords []models.SObject, indexMap []int, uniqueFields []string, deferredUniqueFields map[string]struct{}, rejectionLog *bulkRejectionLog) ([]models.SObject, []models.SObject, []int, error) {
	if len(storageRecords) == 0 || len(uniqueFields) == 0 {
		return storageRecords, validRecords, indexMap, nil
	}
	baseStorage := cloneBulkInsertSObjects(storageRecords)
	baseValid := cloneBulkInsertSObjects(validRecords)
	baseIndexMap := append([]int(nil), indexMap...)
	for len(baseStorage) > 0 {
		candidateStorage := cloneBulkInsertSObjects(baseStorage)
		candidateValid := cloneBulkInsertSObjects(baseValid)
		assumedLandedRows := allBulkInsertRowsLanded(candidateStorage)
		plans, err := ps.exclusiveSiblingBulkInsertResetPlans(schema, candidateStorage, assumedLandedRows)
		if err != nil {
			return nil, nil, nil, err
		}
		finalByID, err := ps.previewExclusiveSiblingBulkInsertFinalRecords(ctx, schema, candidateValid, assumedLandedRows, exclusiveSiblingBulkInsertLandedRowResetUpdates(plans))
		if err != nil {
			return nil, nil, nil, err
		}
		applyExclusiveSiblingBulkInsertFinalRecords(candidateStorage, assumedLandedRows, finalByID)
		applyExclusiveSiblingBulkInsertFinalRecords(candidateValid, assumedLandedRows, finalByID)

		failedRows := ps.exclusiveSiblingBulkInsertFinalUniqueFailures(ctx, schema, candidateValid, baseIndexMap, uniqueFields, deferredUniqueFields, rejectionLog)
		if len(failedRows) == 0 && len(deferredUniqueFields) > 0 {
			resetExcludes, err := ps.exclusiveSiblingBulkInsertDeferredUniqueResetExcludes(ctx, schema, candidateStorage, assumedLandedRows, deferredUniqueFields)
			if err != nil {
				return nil, nil, nil, err
			}
			failedRows = ps.exclusiveSiblingBulkInsertDeferredUniqueFailures(ctx, schema, candidateValid, baseIndexMap, deferredUniqueFields, resetExcludes, rejectionLog)
		}
		if len(failedRows) == 0 {
			return candidateStorage, candidateValid, baseIndexMap, nil
		}
		baseStorage = filterBulkInsertRecordsByFailed(baseStorage, failedRows)
		baseValid = filterBulkInsertRecordsByFailed(baseValid, failedRows)
		baseIndexMap = filterBulkInsertIndexMapByFailed(baseIndexMap, failedRows)
	}
	return nil, nil, nil, appErrors.NewValidationError("", fmt.Sprintf("all %d records failed validation", rejectionLog.count()))
}

func (ps *PersistenceService) exclusiveSiblingBulkInsertDeferredUniqueResetExcludes(ctx context.Context, schema *models.ObjectMetadata, records []models.SObject, landedRows []bool, deferredUniqueFields map[string]struct{}) (map[string]map[string]string, error) {
	if len(deferredUniqueFields) == 0 {
		return nil, nil
	}
	plans, err := ps.exclusiveSiblingBulkInsertResetPlans(schema, records, landedRows)
	if err != nil || len(plans) == 0 {
		return nil, err
	}
	repo, ok := ps.repo.(exclusiveSiblingResetRepository)
	if !ok {
		return nil, appErrors.NewInternalError("exclusive sibling reset requires repository support", nil)
	}

	fieldsByName := map[string]*models.FieldMetadata{}
	for i := range schema.Fields {
		field := &schema.Fields[i]
		fieldName := strings.ToLower(strings.TrimSpace(field.APIName))
		if fieldName == "" || !field.IsUnique {
			continue
		}
		if _, ok := deferredUniqueFields[fieldName]; ok {
			fieldsByName[fieldName] = field
		}
	}
	if len(fieldsByName) == 0 {
		return nil, nil
	}

	excludes := map[string]map[string]string{}
	for _, plan := range plans {
		if schema == nil || plan.partitionValue == "" || plan.winnerID == "" {
			continue
		}
		clearedRows, err := repo.LockExclusiveSiblingResetRows(ctx, schema, plan.contract.FieldAPIName, plan.contract.PartitionFieldAPIName, plan.partitionValue, plan.winnerID)
		if err != nil {
			return nil, appErrors.NewInternalError("exclusive sibling reset failed", err)
		}
		for _, oldRecord := range clearedRows {
			id := exclusiveSiblingRecordID(oldRecord[constants.FieldID])
			if id == "" {
				continue
			}
			if _, skip := plan.skippedSiblingIDs[id]; skip {
				continue
			}
			_, _, finalRecord, _, err := ps.exclusiveSiblingResetFinalRecord(ctx, schema, oldRecord, plan.contract.FieldAPIName)
			if err != nil {
				return nil, err
			}
			for fieldName, field := range fieldsByName {
				oldValue, ok := oldRecord[field.APIName]
				if !ok || oldValue == nil {
					continue
				}
				if strVal, ok := oldValue.(string); ok && strVal == "" {
					continue
				}
				if ps.areValuesEqual(oldValue, finalRecord[field.APIName]) {
					continue
				}
				if excludes[fieldName] == nil {
					excludes[fieldName] = map[string]string{}
				}
				excludes[fieldName][fmt.Sprintf("%v", oldValue)] = id
			}
		}
	}
	return excludes, nil
}

func (ps *PersistenceService) exclusiveSiblingBulkInsertDeferredUniqueFailures(ctx context.Context, schema *models.ObjectMetadata, records []models.SObject, indexMap []int, deferredUniqueFields map[string]struct{}, resetExcludes map[string]map[string]string, rejectionLog *bulkRejectionLog) map[int]struct{} {
	failedRows := map[int]struct{}{}
	for i, record := range records {
		recordIndex := i
		if i < len(indexMap) {
			recordIndex = indexMap[i]
		}
		if err := ps.checkExclusiveSiblingDeferredUniqueness(ctx, schema, record, deferredUniqueFields, resetExcludes); err != nil {
			rejectionLog.rejectRecord(recordIndex, record, err)
			failedRows[i] = struct{}{}
		}
	}
	return failedRows
}

func (ps *PersistenceService) checkExclusiveSiblingDeferredUniqueness(ctx context.Context, schema *models.ObjectMetadata, record models.SObject, deferredUniqueFields map[string]struct{}, resetExcludes map[string]map[string]string) error {
	if schema == nil || len(deferredUniqueFields) == 0 {
		return nil
	}
	for i := range schema.Fields {
		field := &schema.Fields[i]
		fieldName := strings.ToLower(strings.TrimSpace(field.APIName))
		if fieldName == "" || !field.IsUnique {
			continue
		}
		if _, ok := deferredUniqueFields[fieldName]; !ok {
			continue
		}
		val, ok := record[field.APIName]
		if !ok || val == nil {
			continue
		}
		if strVal, ok := val.(string); ok && strVal == "" {
			continue
		}
		excludeID := ""
		if byValue := resetExcludes[fieldName]; byValue != nil {
			excludeID = byValue[fmt.Sprintf("%v", val)]
		}
		if ps.previewSyntheticUniqueConflict(ctx, schema.APIName, field, val, excludeID) {
			return appErrors.NewConflictError(schema.APIName, field.APIName, fmt.Sprintf("%v", val))
		}
		queryField := field.APIName
		if field.PhysicalName != "" {
			queryField = field.PhysicalName
		}
		exists, err := ps.repo.CheckUniqueness(ctx, schema, queryField, val, excludeID)
		if err != nil {
			return err
		}
		if exists {
			return appErrors.NewConflictError(schema.APIName, field.APIName, fmt.Sprintf("%v", val))
		}
	}
	return nil
}

func (ps *PersistenceService) exclusiveSiblingBulkInsertFinalUniqueFailures(ctx context.Context, schema *models.ObjectMetadata, records []models.SObject, indexMap []int, uniqueFields []string, deferredUniqueFields map[string]struct{}, rejectionLog *bulkRejectionLog) map[int]struct{} {
	failedRows := map[int]struct{}{}
	seen := make(map[string]map[string]struct{}, len(uniqueFields))
	for _, fieldName := range uniqueFields {
		seen[fieldName] = map[string]struct{}{}
	}
	for i, record := range records {
		recordIndex := i
		if i < len(indexMap) {
			recordIndex = indexMap[i]
		}
		if err := ps.checkUniquenessSkippingFields(ctx, schema.APIName, record, schema, "", deferredUniqueFields); err != nil {
			rejectionLog.rejectRecord(recordIndex, record, err)
			failedRows[i] = struct{}{}
			continue
		}
		duplicate := false
		for _, fieldName := range uniqueFields {
			val, ok := record[fieldName]
			if !ok || val == nil {
				continue
			}
			key := fmt.Sprintf("%v", val)
			if _, exists := seen[fieldName][key]; exists {
				appendBulkInsertBatchUniqueError(rejectionLog, recordIndex, schema.APIName, record, fieldName, val)
				failedRows[i] = struct{}{}
				duplicate = true
				break
			}
		}
		if duplicate {
			continue
		}
		for _, fieldName := range uniqueFields {
			val, ok := record[fieldName]
			if !ok || val == nil {
				continue
			}
			seen[fieldName][fmt.Sprintf("%v", val)] = struct{}{}
		}
	}
	return failedRows
}

// BulkInsert creates multiple records in a single transaction
func (ps *PersistenceService) BulkInsert(
	ctx context.Context,
	objectName string,
	records []models.SObject,
	currentUser *models.UserSession,
	options BulkInsertOptions,
) (result BulkInsertResult, err error) {
	// Rejections accumulate here, not on the exported result — see
	// bulkRejectionLog. applyTo publishes them onto result on every return
	// path via this defer, so no exit can report a count without the matching
	// captures.
	//
	// The return values MUST stay named: a deferred write to an unnamed
	// result is written to a local the caller never sees, because `return
	// result, err` has already copied it into the return slot.
	rejectionLog := &bulkRejectionLog{}
	defer func() { rejectionLog.applyTo(&result) }()

	if len(records) == 0 {
		return result, nil
	}

	// 1. Permission check (once for all records)
	schema, err := ps.prepareOperation(ctx, objectName, constants.PermCreate, currentUser)
	if err != nil {
		return result, err
	}
	hasExclusiveSiblingFields := schemaHasExclusiveSiblingFields(schema)
	if ps.objectCanReachExclusiveSiblingFields(ctx, objectName, nil) && (!hasExclusiveSiblingFields || ps.objectGeneratedWritesCanReachExclusiveSiblingFields(ctx, objectName, nil)) {
		return result, appErrors.NewValidationError("ExclusiveSiblingFields", "bulk insert does not support generated ExclusiveSiblingFields writes yet; use single-record Insert or UpsertByExternalID")
	}

	if !constants.GetSecurityContext(ctx).BypassesCRUDandFLS() {
		if !options.AllowLegacyListViewBulkIngest {
			if err := rejectGenericListViewMutation(objectName, "bulk insert"); err != nil {
				return result, err
			}
		}
		if err := rejectToolingOnlyTableWrites(objectName); err != nil {
			return result, err
		}
	}

	// Defense-in-depth: only admins may set passwords on User records.
	if !constants.GetSecurityContext(ctx).BypassesProcessFields() && strings.EqualFold(objectName, constants.TableUser) {
		for _, record := range records {
			if err := ps.rejectUnauthorizedPasswordWrite(ctx, objectName, record, currentUser); err != nil {
				return result, err
			}
		}
	}

	// Calculate comprehensive column list from schema
	storageColumns := getStorageColumns(schema)

	// Identify unique fields for batch-level uniqueness check
	uniqueFields := make([]string, 0)
	for _, f := range schema.Fields {
		if f.IsUnique {
			uniqueFields = append(uniqueFields, f.APIName)
		}
	}
	deferredUniqueFields := exclusiveSiblingDeferredUniqueFields(schema)

	// 2. Get validation rules (once, cached). Runtime mutation requests must
	// fail closed when their request-bound mutation projection is unavailable
	// or incomplete; an empty rule set is a valid value only after that check.
	validationRules, err := ps.metadata.GetValidationRulesForMutation(ctx, objectName)
	if err != nil {
		return result, appErrors.NewInternalError("load mutation validation rules", err)
	}

	// Guard-cache state for the create-from-source bypass guard (issue #1234).
	// These three variables live for the lifetime of ONE BulkInsert call.
	// The LOAD itself happens lazily at the first-reach call site after
	// defaults and lookup validation, so one BulkInsert call issues at most
	// one guard metadata query for this object.
	var (
		guardCandidateFields  map[string]struct{}
		guardCandidateLoaded  bool
		guardCandidateLoadErr error
		guardEntries          []recordsguard.CreateFromSourceGuardEntry
		guardLoaded           bool
		guardLoadErr          error
	)

	// 3. Pre-flight validation and preparation
	type permIntercept struct {
		recordIndex int
		recordID    string
		changes     map[string]bool
	}
	var stagedPerms []permIntercept
	// Permission-intercept branching is metadata-driven via
	// ExtractPermissionGrants — schemas without SystemPermission virtuals
	// return an empty grant map, so no objectName gate is required here.
	// The `ps.permissions != nil` check is still meaningful because the
	// post-write ApplyPermissionChanges call needs the service.
	hasPermissionService := ps.permissions != nil

	preparedRecords := make([]models.SObject, 0, len(records))
	for i, record := range records {
		// Unconditional normalize on the bulk path so bool-like string values
		// ("true"/"1") on virtual permission fields get canonicalized before
		// the intercept / Validate bouncer run. CSV-only conditional
		// normalize was insufficient — non-CSV bulk callers were silently
		// relying on Normalize's old default-to-false coerce, which has been
		// tightened to leave unrecognized types for downstream rejection
		// (schema_utils.go:NormalizeSObject). Double-call is a no-op on
		// already-coerced values.
		record = schemautil.NormalizeSObject(schema, record)
		scrubPolymorphicDiscriminators(schema, record)
		if err := rejectWriteProtectedFieldWrites(ctx, schema, record, protectedFieldWriteOnCreate); err != nil {
			rejectionLog.rejectRecord(i, record, err)
			continue
		}
		if err := rejectUnbypassableMetadataFieldWrites(objectName, record); err != nil {
			rejectionLog.rejectRecord(i, record, err)
			continue
		}
		if !constants.GetSecurityContext(ctx).BypassesCRUDandFLS() {
			if err := rejectImmutableMetadataFieldWrites(objectName, record); err != nil {
				rejectionLog.rejectRecord(i, record, err)
				continue
			}
		}
		if !constants.GetSecurityContext(ctx).BypassesProcessFields() && !options.SkipAutoNumbers {
			if err := ps.rejectExplicitAutoNumberWrites(ctx, objectName, record); err != nil {
				rejectionLog.rejectRecord(i, record, err)
				continue
			}
		}
		// Virtual-field intercept (fail-loud on non-bool). Runs before the
		// capability filter so createable checks only see physical fields;
		// runs after Normalize so bool-like strings are canonicalized first.
		recChanges, recExtractErr := persistence.ExtractPermissionGrants(record, schema)
		if recExtractErr != nil {
			rejectionLog.rejectRecord(i, record, recExtractErr)
			continue
		}
		// Fail explicitly if the caller supplied virtual permission values
		// but the permission service isn't configured (e.g. a test with
		// nil deps). Without this guard, Extract has already deleted the
		// keys from `record` and the later staging gate short-circuits on
		// !hasPermissionService — caller sees a "successful" insert while
		// their grants are silently lost. Raised by Copilot review.
		if !hasPermissionService && len(recChanges) > 0 {
			// No error object has ever existed here, and none is synthesized:
			// this is a service-wiring failure, not a user validation
			// rejection, so it correctly produces no audit row while still
			// being counted as a failed record.
			rejectionLog.rejectRecordWithMessage(fmt.Sprintf(
				"record %d: permission virtual fields supplied but permission service is not configured",
				i), record, nil)
			continue
		}

		// Normalize and validate pre-flow sharing-rule security contracts (same as single-record path)
		if err := ps.runRecordWriteCanonicalizers(ctx, recordWriteValidationBeforeCreate, recordWriteCanonicalizationEvent{
			ObjectName:  objectName,
			Record:      record,
			CurrentUser: currentUser,
		}); err != nil {
			rejectionLog.rejectRecord(i, record, err)
			continue
		}
		// Protected-records guard-field: same gate as the single-record Insert.
		// Prevents bulk-creating rows with the guard pre-set to true (which would
		// fabricate a row that immediately triggers the delete guard).
		if constants.GetSecurityContext(ctx) < constants.SystemContext {
			if err := rejectProtectedGuardFieldWrites(objectName, record); err != nil {
				rejectionLog.rejectRecord(i, record, err)
				continue
			}
		}

		// RESILIENCE: Normalize keys for Long Column Names
		for k, v := range record {
			safeKey := domainSchema.GenerateSafeIdentifier(k, "col")
			if safeKey != k {
				if _, exists := record[safeKey]; !exists {
					record[safeKey] = v
				}
			}
		}

		// Apply field capability check matching SF v63.0 + single-record Insert:
		//   - Intrinsic non-createable formula/system fields → silent drop
		//   - Metadata-declared autonumber fields were rejected earlier by
		//     rejectExplicitAutoNumberWrites
		//   - User-FLS-denied → per-record hard-fail listing ALL denied fields
		//     (SF reports them all in one error, verified against sandbox)
		// Internal/system writes (ServiceContext+) bypass this check.
		if !constants.GetSecurityContext(ctx).BypassesProcessFields() {
			resolver := permissions.NewFieldAccessResolver(ps.permissions)
			var deniedFields []string
			for key := range record {
				for fi := range schema.Fields {
					if schema.Fields[fi].APIName != key {
						continue
					}
					if !resolver.IsIntrinsicallyCreateable(objectName, &schema.Fields[fi]) {
						delete(record, key)
						break
					}
					createable, err := resolver.EffectiveCreateable(ctx, &schema.Fields[fi], objectName, currentUser)
					if err != nil {
						return result, err
					}
					if !createable {
						deniedFields = append(deniedFields, key)
					}
					break
				}
			}
			if len(deniedFields) > 0 {
				flsErr := appErrors.NewFLSDeniedErrorForFields(deniedFields)
				rejectionLog.rejectRecord(i, record, flsErr)
				continue
			}
		}

		// Validate user input BEFORE applyDefaults adds system fields
		// This prevents false rejection of audit fields that are internally set
		if !options.SkipValidation {
			// 0. Strict Schema Validation (Bouncer) - validate user's input only
			if err := ps.schemaValidator.ValidateSchemaCompliance(record, schema, false, options.AllowSystemFields, false); err != nil {
				rejectionLog.rejectRecord(i, record, err)
				continue
			}
		}

		// Guard-trigger image: same as single-record Insert, which evaluates
		// caller/FLS-pruned input before metadata defaults and system fields are
		// added. Defaults may still be written to the row; they must not by
		// themselves turn a plain bulk create into INVALID_OPERATION.
		guardPayload := cloneSObject(record)

		// Caller-payload non-polymorphic FK access / existence gate. This must
		// run before create-from-source guard evaluation and before
		// RecordBeforeCreate automation so caller-supplied missing/inaccessible
		// parents keep the same field-specific rejection boundary as
		// single-record Insert. The transactional final-image pass below still
		// revalidates lookups after before-save flows materialize or repoint
		// values.
		if !options.SkipValidation {
			if err := ps.validateNonPolymorphicLookups(withNonPolymorphicLookupCreateIntent(ctx), guardPayload, schema, currentUser); err != nil {
				rejectionLog.rejectRecord(i, record, err)
				continue
			}
		}

		// Apply defaults (and Generate System Fields logic - respecting input Audit fields)
		prepared := ps.applyDefaults(ctx, record, schema, currentUser)
		applyCountRollupSummaryDefaults(prepared, schema)
		if err := rejectOmittedSiblingMaxDefaultsForBulk(prepared, schema); err != nil {
			appendBulkInsertRecordError(rejectionLog, i, prepared, err)
			continue
		}

		// Create-from-source bypass guard (issue #1234). Mirror the
		// single-record path's trigger image: AFTER the FLS denial loop
		// and non-polymorphic lookup gate (handled earlier in this same
		// per-record block) but evaluated against the pre-default caller
		// payload. The remaining state-dependent FK validation is deferred until
		// AFTER RecordBeforeCreate so before-save materialization can repoint or
		// fill the final FK image before the row is validated and persisted.
		// Runs regardless of SkipValidation because direct-create routing is a
		// configured metadata authority, not optional data validation.
		//
		// Lazy-load the narrow candidate field set first. The full semantic
		// guard registry is loaded only if this record actually sets one of
		// those candidate FKs, so stale unrelated SourceLookupField metadata
		// does not brick ordinary bulk inserts.
		if !guardCandidateLoaded {
			guardCandidateFields, guardCandidateLoadErr = ps.loadCreateFromSourceGuardCandidateFields(ctx, objectName)
			guardCandidateLoaded = true
		}
		if guardCandidateLoadErr != nil {
			rejectionLog.rejectRecordWithMessage(fmt.Sprintf(
				"record %d: load create-from-source candidate fields for %s: %v",
				i, objectName, guardCandidateLoadErr,
			), prepared, guardCandidateLoadErr)
			continue
		}
		if guardSourcePayload := createFromSourceGuardCandidatePayload(guardPayload, guardCandidateFields); len(guardSourcePayload) > 0 {
			if !guardLoaded {
				guardEntries, guardLoadErr = ps.loadCreateFromSourceGuardEntries(ctx, objectName)
				guardLoaded = true
			}
			if err := guardCreateFromSourceContextWithEntries(
				ctx, objectName, guardSourcePayload, guardEntries, guardLoadErr,
			); err != nil {
				// `prepared`, not `record`: applyDefaults above returned a NEW
				// map carrying the minted Id, so passing `record` here would
				// record RecordId=__unassigned__ for a record that does have
				// an id, and would miss both the caller-payload binding and
				// the emission dedup key.
				rejectionLog.rejectRecord(i, prepared, err)
				continue
			}
		}
		// PR #2057 round-8 (Codex Review HIGH): formula compute + VR +
		// uniqueness MOVED into the transaction below, AFTER
		// assignAutoNumbersBatch. Pre-round-8 they ran HERE, BEFORE
		// autonumbers, so a frozen formula referencing an autonumber field
		// (e.g. CONCATENATE(QuoteNumber, "-rev")) materialized "" on the
		// Bulk path while single Insert (which assigns autonumbers BEFORE
		// computeFormulaFields, persistence_insert.go:209→260) materialized
		// the rendered value. Same FormulaMode='frozen' fact, two different
		// on-disk values across writers — Axiom 1.1 strict-parity drift on
		// the public Bulk API + R10 split writer. VR / uniqueness also saw
		// a stale image (the formula value before autonumber rendering),
		// so a rule like "Code__c must start with QN-" passed against ""
		// and then INSERTed the post-render value the rule never saw.
		//
		// Order is now: applyDefaults → assignAutoNumbersBatch →
		// computeFormulaFields → VR → uniqueness → INSERT, matching
		// persistence_insert.go's transactional order step-for-step.

		// applyDefaults already calls generateSystemFields; no need to call again.

		// CRITICAL: User-specific username defaults for _System_User bulk
		// inserts. Password hashing/default stamping is delayed until the
		// transactional final-image pass below so single and bulk create share
		// one authoritative password-validation read path.
		if strings.EqualFold(objectName, constants.TableUser) {
			applyUserUsernameDefault(prepared)
		}

		if hasPermissionService && len(recChanges) > 0 {
			// applyDefaults has now assigned any missing Id; stage under
			// the final record identity.
			recID, _ := prepared[constants.FieldID].(string)
			stagedPerms = append(stagedPerms, permIntercept{recordIndex: len(preparedRecords), recordID: recID, changes: recChanges})
		}

		preparedRecords = append(preparedRecords, prepared)
	}

	// Check if all records failed validation
	if len(preparedRecords) == 0 && rejectionLog.count() > 0 {
		return result, appErrors.NewValidationError("", fmt.Sprintf("all %d records failed validation", rejectionLog.count()))
	}
	if err := ps.validateRecordCacheInvalidationConfiguration(ctx, objectName); err != nil {
		return result, err
	}

	// 4. Execute transactional bulk insert.
	//
	// Order MUST match persistence_insert.go::Insert step-for-step:
	//   assignAutoNumbersBatch → RecordBeforeCreate →
	//   state-dependent validation + polymorphic discriminator injection →
	//   computeFormulaFields → ValidateRecord → checkUniqueness → INSERT.
	// Pre-round-8 the bulk path computed formulas + ran VR / uniqueness
	// in the pre-tx loop above (before autonumbers), so frozen formulas
	// referencing autonumber fields and the VR / uniqueness images all
	// saw a stale pre-autonumber value. The transactional second pass
	// below is the authoritative one; it runs BEFORE INSERT so the row
	// that lands in SQL has been validated against the same image it
	// will materialize.
	//
	// Capture validRecords / indexMap here so post-tx code (cache
	// invalidation, SuccessCount) can reflect the survivors-of-pass-2
	// set, not the pre-tx-prepared set.
	preparedSeedRecords := cloneBulkInsertSObjects(preparedRecords)
	// The transaction body below is re-run on retry (WithRetry, 3 attempts),
	// so each attempt must start from the pre-transaction captures.
	preTxMark := rejectionLog.mark()
	var validRecords []models.SObject
	var indexMap []int // validRecords[k] came from preparedRecords[indexMap[k]]
	err = ps.runRollupAwareTransaction(ctx, objectName, nil, func(txCtx context.Context) error {
		preparedRecords := cloneBulkInsertSObjects(preparedSeedRecords)
		rejectionLog.rewind(preTxMark)
		batchUniqueValues := make(map[string]map[string]bool, len(uniqueFields))
		for _, f := range uniqueFields {
			batchUniqueValues[f] = make(map[string]bool)
		}

		// 1. Autonumber v2: batched engine assignment for every
		// AutoNumberFormat-configured field on the object. MUST run
		// before computeFormulaFields so frozen formulas referencing
		// autonumber fields materialize the rendered value, matching
		// single Insert (persistence_insert.go:209→260).
		if !options.SkipAutoNumbers {
			if err := ps.assignAutoNumbersBatch(txCtx, objectName, preparedRecords); err != nil {
				return err
			}
		}
		// 2. Per-record before-save pass. Bulk create now follows the same
		// materialization boundary as single-record Insert: generated IDs and
		// compound fields are visible to RecordBeforeCreate flows, and any
		// in-place $Record mutations become the image seen by validation,
		// SQL INSERT, formula projection, and parent rollup fanout. Failures
		// stay per-row so one customError does not poison the rest of the
		// Bulk job.
		beforeSaveRecords := make([]models.SObject, 0, len(preparedRecords))
		beforeSaveIndexMap := make([]int, 0, len(preparedRecords))
		for i, prepared := range preparedRecords {
			// Resolve compound parent fields (isCompound=true) from their
			// declared member fields. Runs AFTER autonumber assignment so
			// autonumber-derived members (e.g. QuoteNumber) are rendered.
			// Matches single-Insert ordering at persistence_insert.go.
			ps.resolveCompoundFields(schema, prepared)
			preBeforeSaveGuardPayload := cloneSObject(prepared)
			protectedBaseline, err := captureProtectedCreateBaseline(schema, prepared)
			if err != nil {
				rejectionLog.rejectRecord(i, prepared, err)
				continue
			}

			if !options.SkipFlows {
				if err := ps.publishRecordEvent(txCtx, events.RecordBeforeCreate, objectName, prepared, nil, currentUser); err != nil {
					appendBulkInsertRecordError(rejectionLog, i, prepared, err)
					continue
				}
			}

			// Re-normalize after BeforeSave flows may have written raw values
			// (e.g. plain strings into JSON/MultiPicklist columns), then re-run
			// the same fail-closed write guards single Insert applies before
			// validation and persistence observe the final image.
			prepared = schemautil.NormalizeSObject(schema, prepared)
			preparedRecords[i] = prepared
			scrubPolymorphicDiscriminators(schema, prepared)
			recordID, _ := prepared[constants.FieldID].(string)
			if err := ps.runRecordWriteCanonicalizers(txCtx, recordWriteValidationBeforeCreate, recordWriteCanonicalizationEvent{
				ObjectName:  objectName,
				RecordID:    recordID,
				Record:      prepared,
				CurrentUser: currentUser,
			}); err != nil {
				rejectionLog.rejectRecord(i, prepared, err)
				continue
			}
			if err := rejectUnbypassableMetadataFieldWrites(objectName, prepared); err != nil {
				rejectionLog.rejectRecord(i, prepared, err)
				continue
			}
			if err := protectedBaseline.validate(ctx, schema, prepared); err != nil {
				rejectionLog.rejectRecord(i, prepared, err)
				continue
			}
			if postBeforeSaveGuardPayload := createFromSourceGuardMutationPayload(preBeforeSaveGuardPayload, prepared, guardCandidateFields); len(postBeforeSaveGuardPayload) > 0 {
				if !guardLoaded {
					guardEntries, guardLoadErr = ps.loadCreateFromSourceGuardEntries(txCtx, objectName)
					guardLoaded = true
				}
				if err := guardCreateFromSourceContextWithEntries(
					txCtx, objectName, postBeforeSaveGuardPayload, guardEntries, guardLoadErr,
				); err != nil {
					rejectionLog.rejectRecord(i, prepared, err)
					continue
				}
			}
			if err := validateRuntimeScopedPicklistRequiredValues(schema, prepared); err != nil {
				rejectionLog.rejectRecord(i, prepared, err)
				continue
			}
			if !options.SkipValidation {
				if err := ps.validateNonPolymorphicLookups(withNonPolymorphicLookupCreateIntent(withoutCoordinatedLookupParentLock(txCtx)), prepared, schema, currentUser); err != nil {
					rejectionLog.rejectRecord(i, prepared, err)
					continue
				}
			}
			if err := rejectInvalidSiblingMaxFinalImageForBulk(prepared, schema); err != nil {
				rejectionLog.rejectRecord(i, prepared, err)
				continue
			}
			if !options.SkipValidation {
				if err := ps.validateRecordTypeId(txCtx, objectName, prepared, currentUser); err != nil {
					rejectionLog.rejectRecord(i, prepared, err)
					continue
				}
				if err := ps.validateQueueAssignment(txCtx, objectName, prepared, currentUser); err != nil {
					rejectionLog.rejectRecord(i, prepared, err)
					continue
				}
			}
			normalizeGroupUserRoleReferenceWriteImage(objectName, prepared)
			resolvedTypes, err := ps.validatePolymorphicLookups(txCtx, prepared, schema)
			if err != nil {
				rejectionLog.rejectRecord(i, prepared, err)
				continue
			}
			for fieldName, objType := range resolvedTypes {
				prepared[schemautil.GetPolymorphicTypeColumnName(fieldName)] = objType
			}

			beforeSaveRecords = append(beforeSaveRecords, prepared)
			beforeSaveIndexMap = append(beforeSaveIndexMap, i)
		}
		if len(beforeSaveRecords) == 0 {
			return appErrors.NewValidationError("", fmt.Sprintf("all %d records failed validation", rejectionLog.count()))
		}
		var prelockErr error
		beforeSaveRecords, beforeSaveIndexMap, prelockErr = ps.prelockInitialBulkInsertParents(
			txCtx,
			objectName,
			schema,
			beforeSaveRecords,
			beforeSaveIndexMap,
			rejectionLog,
			options.SkipValidation,
		)
		if prelockErr != nil {
			return prelockErr
		}
		if len(beforeSaveRecords) == 0 {
			return appErrors.NewValidationError("", fmt.Sprintf("all %d records failed validation", rejectionLog.count()))
		}
		pendingCreates := recordWriteValidationPendingCreatesFromSObjects(beforeSaveRecords)

		// 3. Per-record final validation pass: formula compute → VR →
		// uniqueness. Failures drop the record from the INSERT batch and
		// increment FailedCount (preserving partial-success semantics);
		// they do NOT abort the transaction (one bad row in a batch never
		// poisons the rest). All-failed → return ValidationError to roll
		// back the autonumber sequence increments together with the
		// (empty) INSERT.
		validRecords = make([]models.SObject, 0, len(beforeSaveRecords))
		indexMap = make([]int, 0, len(beforeSaveRecords))
		// userCtx built lazily — only validation needs it, and SkipValidation
		// imports must not pay the buildUserContext cost (or its side-effect
		// FindOne calls — pre-round-8 the same lazy pattern lived inside the
		// pre-tx loop). Built once per BulkInsert call and reused for every
		// row's VR.
		var userCtx map[string]interface{}
		for k, prepared := range beforeSaveRecords {
			i := beforeSaveIndexMap[k]
			// Frozen formula compute on the post-autonumber image. Same
			// fetch-error-delete contract as single Insert (round-6 F3):
			// any frozen formula whose relation fetch errored is removed
			// from the prepared row so chunk-wide column selection
			// (record_bulk.go::activeColumnsForRecords) does not include
			// the column for THIS row, matching single Insert's
			// "absent column → no SQL write" semantics byte-for-byte.
			fetchErrFields := computeFormulaFields(txCtx, schema, prepared, ps.repo, ps.metadata, objectName)
			for fieldName := range fetchErrFields {
				delete(prepared, fieldName)
			}

			recordID, _ := prepared[constants.FieldID].(string)
			if err := ps.runRecordWriteValidators(txCtx, recordWriteValidationBeforeCreate, recordWriteValidationEvent{
				ObjectName:     objectName,
				RecordID:       recordID,
				Record:         prepared,
				CurrentUser:    currentUser,
				PendingCreates: pendingCreates,
			}); err != nil {
				rejectionLog.rejectRecord(i, prepared, err)
				continue
			}

			if !options.SkipValidation {
				if userCtx == nil {
					userCtx = ps.buildUserContext(txCtx, currentUser)
				}
				// Validate static rules on the FINAL post-autonumber +
				// post-formula image — VRs see exactly what SQL will write.
				if err := ps.validateStaticDependentPicklists(schema, prepared, nil); err != nil {
					rejectionLog.rejectRecord(i, prepared, err)
					continue
				}
				if err := ps.validateRuntimeScopedPicklists(txCtx, schema, prepared, nil); err != nil {
					rejectionLog.rejectRecord(i, prepared, err)
					continue
				}
				if err := ps.validateRuntimeScopedLookups(txCtx, schema, prepared, nil); err != nil {
					rejectionLog.rejectRecord(i, prepared, err)
					continue
				}
				if err := ps.validator.ValidateRecord(txCtx, prepared, schema, validationRules, nil, userCtx); err != nil {
					rejectionLog.rejectRecord(i, prepared, err)
					continue
				}
				// Database uniqueness on the post-formula value.
				if err := ps.checkUniquenessSkippingFields(txCtx, objectName, prepared, schema, "", deferredUniqueFields); err != nil {
					rejectionLog.rejectRecord(i, prepared, err)
					continue
				}
				// Batch-level uniqueness within this BulkInsert call. ES
				// annotated objects defer this until after parent prelocks and
				// same-batch final-image planning, because those steps can drop
				// rows or rewrite write-materialized unique fields.
				if !hasExclusiveSiblingFields {
					batchDupFound := false
					for _, fieldName := range uniqueFields {
						if val, ok := prepared[fieldName]; ok && val != nil {
							key := fmt.Sprintf("%v", val)
							if batchUniqueValues[fieldName][key] {
								appendBulkInsertBatchUniqueError(rejectionLog, i, objectName, prepared, fieldName, val)
								batchDupFound = true
								break
							}
							batchUniqueValues[fieldName][key] = true
						}
					}
					if batchDupFound {
						continue
					}
				}
			}

			if err := ps.rejectCurrentUserPasswordReuse(txCtx, objectName, prepared, currentUser); err != nil {
				appendBulkInsertRecordError(rejectionLog, i, prepared, err)
				continue
			}
			if strings.EqualFold(objectName, constants.TableUser) {
				if err := applyUserPasswordDefaults(prepared); err != nil {
					rejectionLog.rejectRecord(i, prepared, err)
					continue
				}
			}

			validRecords = append(validRecords, prepared)
			indexMap = append(indexMap, i)
		}

		if len(validRecords) == 0 {
			// Every record failed pass-2 validation. Roll back the
			// autonumber increments by returning an error, matching the
			// pre-tx all-failed gate at the top of this function.
			return appErrors.NewValidationError("", fmt.Sprintf("all %d records failed validation", rejectionLog.count()))
		}
		// 3. Convert valid records to storage format.
		storageRecords := make([]models.SObject, len(validRecords))
		for i, record := range validRecords {
			storageRecords[i] = filterToLogicalFields(schema, record)
		}

		// Determine batch size
		batchSize := options.BatchSize
		if batchSize <= 0 {
			batchSize = 100
		}

		// Prepare repo options
		repoOpts := map[string]interface{}{
			"batchSize":        batchSize,
			"ignoreDuplicates": options.IgnoreDuplicates,
		}

		preLockIndexMap := append([]int(nil), indexMap...)
		var lockErr error
		storageRecords, indexMap, lockErr = ps.filterExclusiveSiblingBulkInsertLockable(txCtx, schema, storageRecords, indexMap, rejectionLog)
		if lockErr != nil {
			return lockErr
		}
		if len(storageRecords) == 0 {
			return appErrors.NewValidationError("", fmt.Sprintf("all %d records failed validation", rejectionLog.count()))
		}
		validRecords = filterBulkInsertValidRecords(validRecords, preLockIndexMap, indexMap)
		if hasExclusiveSiblingFields && !options.SkipValidation && !options.IgnoreDuplicates {
			var uniqueErr error
			storageRecords, validRecords, indexMap, uniqueErr = ps.filterExclusiveSiblingBulkInsertFinalUniqueRows(txCtx, schema, storageRecords, validRecords, indexMap, uniqueFields, deferredUniqueFields, rejectionLog)
			if uniqueErr != nil {
				return uniqueErr
			}
		}
		if hasExclusiveSiblingFields && !options.IgnoreDuplicates && len(deferredUniqueFields) > 0 {
			assumedLandedRows := allBulkInsertRowsLanded(storageRecords)
			preInsertPlans, err := ps.exclusiveSiblingBulkInsertResetPlans(schema, storageRecords, assumedLandedRows)
			if err != nil {
				return err
			}
			if err := ps.applyExclusiveSiblingResets(txCtx, schema, preInsertPlans); err != nil {
				return err
			}
		}

		needsDefaultChildRows := len(schema.DefaultChildRows) > 0
		needsLandingForRollup := hasReferenceFields(schema)
		needsLandingForLifecycle := len(schema.RecordLifecycleCapabilitiesForPhase(string(recordLifecycleAfterCreateInTx))) > 0
		needsLandedRows := len(stagedPerms) > 0 || ps.outbox != nil || needsDefaultChildRows || needsLandingForRollup ||
			hasExclusiveSiblingFields || ps.permissionStateRequiresPessimisticTransaction(objectName) || needsLandingForLifecycle
		landedRows := make([]bool, len(storageRecords))
		if needsLandedRows {
			var landingErr error
			landedRows, landingErr = ps.repo.BulkInsertWithLanding(txCtx, objectName, storageRecords, storageColumns, repoOpts)
			if landingErr != nil {
				return wrapRepositoryMutationError("bulk insert failed", landingErr)
			}
			if len(landedRows) != len(storageRecords) {
				return appErrors.NewInternalError("bulk insert landing result length mismatch", nil)
			}
		} else if err := ps.repo.BulkInsert(txCtx, objectName, storageRecords, storageColumns, repoOpts); err != nil {
			return wrapRepositoryMutationError("bulk insert failed", err)
		}
		for _, rec := range storageRecords {
			if id, ok := rec[constants.FieldID].(string); ok && id != "" {
				InvalidateRecordReadCache(txCtx, objectName, id)
			}
		}
		if err := ps.markGroupMembershipClosureDirtyAfterMutation(txCtx, objectName, groupMembershipClosureMutationBulkInsert, nil); err != nil {
			return appErrors.NewInternalError("group membership closure invalidation failed", err)
		}
		if hasExclusiveSiblingFields {
			plans, err := ps.exclusiveSiblingBulkInsertResetPlans(schema, storageRecords, landedRows)
			if err != nil {
				return err
			}
			finalByID, err := ps.applyExclusiveSiblingBulkInsertLandedRowResets(txCtx, schema, validRecords, landedRows, plans)
			if err != nil {
				return err
			}
			if err := ps.applyExclusiveSiblingResets(txCtx, schema, plans); err != nil {
				return err
			}
			applyExclusiveSiblingBulkInsertFinalRecords(storageRecords, landedRows, finalByID)
			applyExclusiveSiblingBulkInsertFinalRecords(validRecords, landedRows, finalByID)
		}
		// Report every landed row: different rows can change different permission
		// definitions. The shared authority coalesces revisions per root and SQL
		// attempt; stopping after the first global invalidation misses later roots.
		// Ignored rows did not change a definition and must not stale its snapshot.
		for index, landed := range landedRows {
			if !landed {
				continue
			}
			if err := ps.requestPermissionStateAdvanceForMutation(txCtx, objectName, nil, storageRecords[index]); err != nil {
				return appErrors.NewInternalError("permission state advance request failed", err)
			}
		}

		// stagedPerms.recordIndex is the index into preparedRecords (set
		// in pass 1). After pass-2 filtering, the index into landedRows
		// is the index into validRecords. Build an inverse map so a
		// pass-2-dropped record's perms get skipped instead of mis-applied.
		preparedToValidIndex := make(map[int]int, len(indexMap))
		for k, prepIdx := range indexMap {
			preparedToValidIndex[prepIdx] = k
		}

		if needsDefaultChildRows {
			for i, record := range storageRecords {
				if i >= len(landedRows) || !landedRows[i] {
					continue
				}
				if err := ps.applyDefaultChildRows(txCtx, schema, record, currentUser); err != nil {
					return appErrors.NewInternalError("failed to apply default child rows", err)
				}
			}
		}

		// Apply staged permission grants via the junction SSoT. When
		// IgnoreDuplicates=true some records may have been silently skipped
		// by INSERT IGNORE. Use repository landing evidence instead of
		// post-insert existence: an existing row with the same Id belongs to
		// the pre-image and must not receive this batch's virtual grants.
		// ApplyPermissionChanges validates catalog + prereqs, writes diff
		// rows, and records Setup Audit Trail entries per flip.
		for _, pi := range stagedPerms {
			if pi.recordID == "" {
				continue
			}
			validIdx, present := preparedToValidIndex[pi.recordIndex]
			if !present {
				// Record was dropped in pass-2 (VR / uniqueness failure)
				// — its row never landed, do not apply its perm grants.
				continue
			}
			if validIdx < 0 || validIdx >= len(landedRows) || !landedRows[validIdx] {
				continue
			}
			if applyErr := ps.permissions.ApplyPermissionChanges(txCtx, pi.recordID, nil, pi.changes, currentUser); applyErr != nil {
				// Preserve IsInternal errors (catalog / junction write) as
				// 500; wrap plain validation failures as 400.
				if appErrors.IsInternal(applyErr) {
					return applyErr
				}
				return appErrors.NewValidationError("Permissions", applyErr.Error())
			}
		}

		// Dispatch after-create lifecycle only for rows this statement proved it
		// inserted. INSERT IGNORE can report no SQL error for a duplicate row;
		// treating that pre-existing identity as this batch's lifecycle event
		// would let downstream capabilities author effects for a mutation that
		// never occurred. Batch-aware registrations collect these events until
		// the records-owned transaction callback completes, while established
		// single-event registrations retain their immediate behavior.
		for i, record := range storageRecords {
			if i >= len(landedRows) || !landedRows[i] {
				continue
			}
			recordID, _ := record[constants.FieldID].(string)
			if err := ps.runRecordLifecycleHooks(txCtx, recordLifecycleAfterCreateInTx, recordLifecycleEvent{
				ObjectName:  objectName,
				RecordID:    recordID,
				Record:      cloneSObject(record),
				CurrentUser: currentUser,
			}); err != nil {
				return err
			}
		}

		// Rollup fanout: recompute affected parent rollup summary fields
		// for landed child rows. Deduplication by (parentObj, parentID,
		// rollupField) is handled inside ProcessBulkRollups so N children
		// under the same parent trigger one recompute, not N redundant
		// parent writes. Non-landed INSERT IGNORE rows are excluded — they
		// never committed, so their parent rollup must not reflect phantom
		// data.
		//
		// Rollup fanout MUST run inside the transaction for ACID
		// compliance (matching single-record Insert at
		// persistence_insert.go:134). A recompute error fails the
		// transaction in the same spirit as single-record insert.
		//
		// Only call ProcessBulkRollups when landing evidence exists.
		// Without landing, no rows committed and there is nothing to
		// roll up.
		if needsLandedRows {
			if err := ps.rollup.ProcessBulkRollups(txCtx, objectName, validRecords, landedRows); err != nil {
				return wrapRollupHookError("failed to process bulk rollups", err)
			}
		}
		if needsLandedRows && createFinalImageRefreshRequired(schema) {
			finalRecords, err := ps.loadFinalBulkCreateImages(txCtx, objectName, validRecords, landedRows)
			if err != nil {
				return err
			}
			validRecords = finalRecords
		}

		// Enqueue RecordCreated outbox event per record (inside transaction for
		// guaranteed delivery). Matches the single-record Insert path and closes
		// the data-capture gap for Bulk API v2 / CSV imports.
		//
		// IgnoreDuplicates=true requires exact landing evidence: INSERT
		// IGNORE may silently drop rows, and emitting an event for a row that
		// never landed would push a phantom record into downstream replay
		// targets (e.g., SF).
		if ps.outbox != nil {
			projectionRequired := ps.activeFormulaProjector != nil &&
				ps.activeFormulaProjectionRequired(txCtx, objectName, "RecordAfterSave", "Create")
			for i, record := range validRecords {
				if i >= len(landedRows) || !landedRows[i] {
					// Row was silently skipped by INSERT IGNORE; do not emit
					// a phantom event. Reduces result.SuccessCount drift: the
					// skipped row is still counted as "success" for bulk-API
					// compatibility (legacy behavior), but is not replayed.
					continue
				}
				// Slice 3 (#1362): per-row active-projection stamping —
				// same Clear → Mark → project → With sequence as the
				// single-record Insert path (persistence_insert.go).
				//
				// Per-row Clear is essential here even within a single
				// BulkInsert call: each row composes its OWN outbox event,
				// and without clearing the previous row's snapshot from
				// enqueueCtx, the next row's WithActiveProjections(..., nil,
				// nil) (when it has no active fields) would inherit the
				// prior row's stamped snap and FlowTriggerListener would
				// overlay row N-1's active fields onto row N's $Record.
				//
				// Fail-closed rationale: rows are already INSERTed at this
				// point (landedRows[i]==true). A projection failure aborts
				// the whole transaction (return NewInternalError), not just
				// the per-row event — leaving committed rows without
				// corresponding outbox events would create downstream
				// replay drift (e.g. SF sync never sees those records).
				// CSV ingest users expect atomic write semantics for
				// view-rebuild glitches; a full rollback is the correct
				// recovery signal.
				enqueueCtx := txCtx
				if ps.activeFormulaProjector != nil {
					enqueueCtx = eventFeature.ClearActiveProjections(enqueueCtx)
					if projectionRequired {
						enqueueCtx = eventFeature.MarkActiveProjectionAttempted(enqueueCtx)
						recID, _ := record[constants.FieldID].(string)
						// Write-side SystemContext elevation — see analogous
						// block in persistence_insert.go for full rationale.
						projCtx := constants.WithSecurityContext(
							txCtx, constants.SystemContext,
							"active-formula write-side projection: bulk insert (issue #1362)",
						)
						newSnap, projErr := ps.activeFormulaProjector.project(projCtx, currentUser, schema, recID)
						if projErr != nil {
							return appErrors.NewInternalError(
								fmt.Sprintf("active-formula projection (bulk insert index %d, RecordId=%s) failed", i, recID),
								projErr,
							)
						}
						updatedCtx, encErr := eventFeature.WithActiveProjections(enqueueCtx, newSnap, nil)
						if encErr != nil {
							return appErrors.NewInternalError(
								fmt.Sprintf("encode active-projection snapshot (bulk insert index %d) failed", i),
								encErr,
							)
						}
						enqueueCtx = updatedCtx
					}
				}
				outboxPayload := eventFeature.RecordEventPayload{
					ObjectAPIName: objectName,
					Record:        record,
					CurrentUser:   currentUser,
				}
				enqueueCtx, outboxPayload = eventFeature.PrepareRecordCreatedOutboxPayload(enqueueCtx, outboxPayload)
				if err := ps.enqueueRecordEvent(enqueueCtx, events.RecordCreated, outboxPayload); err != nil {
					recID, _ := record[constants.FieldID].(string)
					return appErrors.NewInternalError(
						fmt.Sprintf("failed to enqueue bulk record created event at index %d (RecordId=%s)", i, recID),
						err,
					)
				}
			}
		}

		return nil
	})

	if err != nil {
		return result, err
	}

	ps.invalidateSecurityCachesForRecords(ctx, objectName, validRecords)

	result.SuccessCount = len(validRecords)
	log.Printf("✨ Bulk created %d records in %s (User: %s)", result.SuccessCount, objectName, getUserID(currentUser))

	return result, nil
}

// hasReferenceFields returns true if the schema contains at least one
// reference or master-detail field that could feed a parent rollup summary.
func hasReferenceFields(schema *models.ObjectMetadata) bool {
	if schema == nil {
		return false
	}
	for _, f := range schema.Fields {
		if constants.IsReferenceType(f.Type) {
			return true
		}
	}
	return false
}

// getStorageColumns determines the list of physical database columns from the schema
// This ensures that even if input records are sparse, the bulk insert statement
// includes all relevant columns (setting NULL for missing values).
//
// SQL-generating path: iterates models.PersistableFields so schema-level
// virtuals (for example, SystemPermission projections) never reach the
// INSERT column list. Additional write-policy exclusions still apply
// inside this helper: stored RollupSummary fields are physical columns
// but remain insert-excluded so callers cannot seed derived values on
// create, and non-frozen formulas remain compute-only. Issuing INSERT
// INTO PermissionSet (…, PermissionsViewSetup, …) would fail at TiDB
// with "unknown column" post-P6 because the bitmap columns are gone; the
// intercept in persistence_service/persistence_insert/persistence_bulk_insert
// already strips those keys from the record, so the payload itself will
// not carry them — this guard is the belt that protects the suspenders.
func getStorageColumns(schema *models.ObjectMetadata) []string {
	persistable := models.PersistableFields(schema)
	columns := make([]string, 0, len(persistable)+1)
	columns = append(columns, constants.FieldID) // ID always exists
	seen := map[string]bool{constants.FieldID: true}

	for _, field := range persistable {
		if seen[field.APIName] {
			continue
		}

		normalizedType := constants.NormalizeFieldType(string(field.Type))

		// Stored RollupSummary parent columns are metadata-owned. COUNT
		// rollups are the one create-time exception: the platform initializes
		// them to zero after user payload validation, so a new parent never
		// exposes null for an empty count. Non-COUNT rollups stay insert-
		// excluded so callers cannot seed derived values directly on create.
		if normalizedType == constants.FieldTypeRollupSummary && !isCountRollupSummaryField(field) {
			continue
		}
		if field.Formula != nil || field.ProjectionConfig != nil || normalizedType == constants.FieldTypeFormula {
			storage, err := formulastorage.Resolve(formulastorage.SpecFromField(&field))
			_ = err
			if !storage.RequiresPhysicalColumn() {
				continue
			}
		}

		columns = append(columns, field.APIName)
		seen[field.APIName] = true

		// Add polymorphic type column if applicable.
		if isPolymorphicReferenceField(&field) {
			polyCol := field.APIName + constants.PolymorphicTypeSuffix
			if !seen[polyCol] {
				columns = append(columns, polyCol)
				seen[polyCol] = true
			}
		}
	}
	return columns
}