package records

import (
	"context"
	"fmt"
	"log"
	"strings"
	"sync/atomic"
	"time"

	"github.com/nexuscrm/audit"
	eventFeature "github.com/nexuscrm/events"
	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/perftiming"
	"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/models"
	zlog "github.com/rs/zerolog/log"
)

// --- Insert Operations ---

// insertFinalImageObserver receives and may reject the post-before-save logical
// row image immediately before uniqueness validation. This is the earliest
// point at which an atomic upsert can bind a defaulted identity and classify
// both a precheck-visible concurrent winner and a later database duplicate
// against the same attempted identity without allowing invalid DML.
type insertFinalImageObserver func(models.SObject) error

type insertFinalImageObservationKey struct{}

type insertFinalImageObservation struct {
	claimed atomic.Bool
	observe insertFinalImageObserver
}

type externalIDUpsertInsertErrorPhase uint8

const (
	externalIDUpsertInsertUniquenessPrecheck externalIDUpsertInsertErrorPhase = iota + 1
	externalIDUpsertInsertRepositoryWrite
)

// externalIDUpsertDirectInsertError proves that an error came from the outer
// Insert claimed by one atomic upsert observation, at one exact pre-SQL or SQL
// boundary. The owner pointer prevents an error from a nested Insert, default-
// child write, or lifecycle hook from borrowing another attempt's provenance.
type externalIDUpsertDirectInsertError struct {
	owner *insertFinalImageObservation
	phase externalIDUpsertInsertErrorPhase
	cause error
}

func (e *externalIDUpsertDirectInsertError) Error() string {
	return e.cause.Error()
}

func (e *externalIDUpsertDirectInsertError) Unwrap() error {
	return e.cause
}

func (o *insertFinalImageObservation) wrapDirectInsertError(phase externalIDUpsertInsertErrorPhase, cause error) error {
	if o == nil || cause == nil {
		return cause
	}
	return &externalIDUpsertDirectInsertError{owner: o, phase: phase, cause: cause}
}

func withInsertFinalImageObserver(ctx context.Context, observe insertFinalImageObserver) (context.Context, *insertFinalImageObservation) {
	observation := &insertFinalImageObservation{observe: observe}
	return context.WithValue(ctx, insertFinalImageObservationKey{}, observation), observation
}

func claimInsertFinalImageObserver(ctx context.Context) *insertFinalImageObservation {
	observation, _ := ctx.Value(insertFinalImageObservationKey{}).(*insertFinalImageObservation)
	if observation == nil || observation.observe == nil || !observation.claimed.CompareAndSwap(false, true) {
		return nil
	}
	return observation
}

// Insert creates a new record with ACID transaction guarantees
func (ps *PersistenceService) Insert(
	ctx context.Context,
	objectName string,
	data models.SObject,
	currentUser *models.UserSession,
) (result models.SObject, err error) {
	preparedPermissionAuthoringApply := permissionAuthoringPreparedInsertApplyFromContext(ctx, objectName)
	// The first Insert reached through a context-scoped observer owns it. This
	// claim happens before any before-save hooks, so nested writes inherit an
	// already-claimed observation and cannot overwrite the outer final image.
	insertObservation := claimInsertFinalImageObserver(ctx)

	// Rejected-save audit (#4261). Deferred at the ENTRY point on purpose:
	// Insert rejects from many places (pre-transaction FLS pruning, inside
	// txnBody, after the parent-prelock retry loop), and a hook placed at any
	// one of them would miss the others and would miss whichever return path
	// gets added next. A defer on the named `err` return cannot be bypassed.
	//
	// Placing it AFTER the retry loop rather than inside also gives one row
	// per rejected save rather than one per attempt.
	//
	// The payload field-name snapshot is taken HERE, before defaulting,
	// autonumber assignment, and before-save flows have stamped anything, so
	// it records what the caller actually sent.
	// Install the one-rejection-one-row claim before anything downstream
	// runs, so a nested write inside a before-save hook inherits it.
	ctx = withRejectedSaveClaim(ctx)
	insertRequestedFields := ps.rejectedSaveRequestedFields(data)
	defer func() {
		if err == nil {
			return
		}
		// `data` is rebound to the final post-flow image inside txnBody, so
		// read the id from the value in scope at return time; it is empty
		// only when the rejection preceded Id assignment.
		rejectedID, _ := data[constants.FieldID].(string)
		ps.emitRejectedSaveAudit(ctx, objectName, rejectedID, audit.ActionCreate,
			insertRequestedFields, err, currentUser)
	}()

	start := time.Now()
	defer func() {
		if !perftiming.Enabled(ctx) && err == nil {
			return
		}
		perftiming.Event(ctx, "record_timing", start, err).
			Str("operation", "Insert").
			Str("object", objectName).
			Bool("has_error", err != nil).
			Msg("Persistence timing")
	}()

	stageStart := time.Now()
	var preparedInsert *preparedInsertRequest
	if preparedPermissionAuthoringApply != nil {
		preparedInsert = &preparedInsertRequest{
			schema:               preparedPermissionAuthoringApply.schema,
			data:                 cloneSObject(preparedPermissionAuthoringApply.data),
			preInsertPermChanges: clonePermissionChanges(preparedPermissionAuthoringApply.preInsertPermChanges),
		}
	} else {
		preparedInsert, err = ps.prepareInsertRequest(ctx, objectName, data, currentUser)
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest", objectName, "", stageStart, err, len(data))
		if err != nil {
			return nil, err
		}
	}
	// Runtime's durable command owner reserves one stable Id for this outer
	// create. prepareInsertRequest has copied that Id into preparedInsert.data,
	// so remove the context-only reservation before any lifecycle/default-child
	// hook can issue a nested Insert and accidentally reuse the outer identity.
	ctx = withoutRuntimeReservedCreateID(ctx)
	schema := preparedInsert.schema
	preparedData := preparedInsert.data
	preInsertPermChanges := preparedInsert.preInsertPermChanges
	if err := ps.validateRecordCacheInvalidationConfiguration(ctx, objectName); err != nil {
		return nil, err
	}

	// Execute Transactional Work
	//
	// Salesforce order of execution (Apex Developer Guide):
	//   before-save record-triggered flows
	//   → custom validation rules
	//   → save (uniqueness check, then SQL INSERT)
	//
	// PR #2057 IANTHEREAL blocker 1: pre-fix this code ran
	// compute(v1) → ValidateRecord → checkUniqueness → BeforeCreate → compute(v2) → INSERT,
	// so VRs and uniqueness saw the pre-flow record/formula snapshot while
	// persistence wrote the post-flow snapshot — Axiom 1.1 strict-parity
	// drift on the authoring model + R10 split read path on the formula fact.
	//
	// Fix: run BeforeCreate first, then a SINGLE post-flow compute, then
	// ValidateRecord + checkUniqueness inside the tx so VR/uniqueness see
	// the same record image that SQL writes.
	txnBody := func(txCtx context.Context) error {
		// Establish the parent/ancestor lock order from the prepared input before
		// a synchronous before-save Flow can perform a nested write. The Flow may
		// still repoint a lookup, so prepareInsertBeforeSave repeats discovery on
		// the final image below; that second pass is duplicate-safe and asks the
		// owning transaction boundary to replay only for genuinely new earlier
		// parents. Without this first pass, a Flow can lock a direct rollup parent
		// before an already-known ancestor and force every ordinary insert through
		// a whole-transaction retry.
		stageStart = time.Now()
		if _, err := ps.prelockInitialInsertParents(txCtx, objectName, schema, data); err != nil {
			logRecordStageTiming(txCtx, "Insert", "prelockPreparedInsertParents", objectName, "", stageStart, err, len(data))
			return err
		}
		logRecordStageTiming(txCtx, "Insert", "prelockPreparedInsertParents", objectName, "", stageStart, nil, len(data))

		stageStart = time.Now()
		var finalData models.SObject
		if preparedPermissionAuthoringApply != nil {
			finalData = cloneSObject(preparedPermissionAuthoringApply.data)
		} else {
			finalData, _, _, err = ps.prepareInsertBeforeSave(ctx, txCtx, objectName, data, schema, currentUser, true, insertObservation)
			logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave", objectName, "", stageStart, err, len(data))
			if err != nil {
				return err
			}
		}

		stageStart := time.Now()
		if err := ps.rejectCurrentUserPasswordReuse(txCtx, objectName, finalData, currentUser); err != nil {
			logRecordStageTiming(txCtx, "Insert", "rejectCurrentUserPasswordReuse", objectName, "", stageStart, err, len(finalData))
			return err
		}
		logRecordStageTiming(txCtx, "Insert", "rejectCurrentUserPasswordReuse", objectName, "", stageStart, nil, len(finalData))
		if strings.EqualFold(objectName, constants.TableUser) {
			stageStart = time.Now()
			if err := applyUserPasswordDefaults(finalData); err != nil {
				logRecordStageTiming(txCtx, "Insert", "applyUserPasswordDefaults", objectName, "", stageStart, err, len(finalData))
				return appErrors.NewInternalError("user defaults failed", err)
			}
			logRecordStageTiming(txCtx, "Insert", "applyUserPasswordDefaults", objectName, "", stageStart, nil, len(finalData))
		}
		stageStart = time.Now()
		exclusiveSiblingPlans, err := ps.prepareExclusiveSiblingInsert(txCtx, schema, finalData)
		logRecordStageTiming(txCtx, "Insert", "prepareExclusiveSiblingInsert", objectName, "", stageStart, err, len(finalData))
		if err != nil {
			return err
		}
		logicalData := filterToLogicalFields(schema, finalData)
		data = finalData

		stageStart = time.Now()
		if _, err := ps.prelockFinalInsertParents(txCtx, objectName, schema, finalData); err != nil {
			logRecordStageTiming(txCtx, "Insert", "prelockFinalInsertParents", objectName, "", stageStart, err, len(finalData))
			return err
		}
		logRecordStageTiming(txCtx, "Insert", "prelockFinalInsertParents", objectName, "", stageStart, nil, len(finalData))

		if preparation := permissionAuthoringRecordPreparationFromContext(ctx); preparation != nil {
			return preparation.captureInsert(&permissionAuthoringPreparedInsert{
				objectName:           objectName,
				schema:               schema,
				data:                 cloneSObject(finalData),
				preInsertPermChanges: clonePermissionChanges(preInsertPermChanges),
				currentUser:          currentUser,
			})
		}

		// Permission grants were intercepted at the top of Insert.
		// The actual junction writes + audit run post-insert once the
		// PermissionSet row exists (ApplyPermissionChanges needs the ID).

		// execute insert via Repository
		stageStart = time.Now()
		if err := ps.repo.Insert(txCtx, objectName, logicalData); err != nil {
			logRecordStageTiming(txCtx, "Insert", "repo.Insert", objectName, "", stageStart, err, len(logicalData))
			return wrapRepositoryMutationError(
				"insert failed",
				insertObservation.wrapDirectInsertError(externalIDUpsertInsertRepositoryWrite, err),
			)
		}
		logRecordStageTiming(txCtx, "Insert", "repo.Insert", objectName, "", stageStart, nil, len(logicalData))
		if err := ps.requestPermissionStateAdvanceForRecordMutation(txCtx, objectName, nil, logicalData, preparedPermissionAuthoringApply != nil); err != nil {
			return appErrors.NewInternalError("permission state advance request failed", err)
		}
		markInsertedRowTransactionOwned(txCtx, objectName, logicalData)
		insertedID, _ := logicalData[constants.FieldID].(string)
		InvalidateRecordReadCache(txCtx, objectName, fmt.Sprintf("%v", logicalData[constants.FieldID]))
		stageStart = time.Now()
		if err := ps.applyExclusiveSiblingResets(txCtx, schema, exclusiveSiblingPlans); err != nil {
			logRecordStageTiming(txCtx, "Insert", "applyExclusiveSiblingResets", objectName, "", stageStart, err, len(logicalData))
			return err
		}
		logRecordStageTiming(txCtx, "Insert", "applyExclusiveSiblingResets", objectName, "", stageStart, nil, len(logicalData))
		stageStart = time.Now()
		if err := ps.markGroupMembershipClosureDirtyAfterMutation(txCtx, objectName, groupMembershipClosureMutationInsert, nil); err != nil {
			logRecordStageTiming(txCtx, "Insert", "markGroupMembershipClosureDirtyAfterMutation", objectName, "", stageStart, err, len(logicalData))
			return appErrors.NewInternalError("group membership closure invalidation failed", err)
		}
		logRecordStageTiming(txCtx, "Insert", "markGroupMembershipClosureDirtyAfterMutation", objectName, "", stageStart, nil, len(logicalData))

		// C11: emit CREATE AuditEvent via V2 pipeline. Gated by
		// complete audit v2 fan-out path is ready. No-op outside a tx or when id
		// is missing. The outbox worker expands this head row into a
		// single `__created__` sentinel row so `<Object>History` surfaces
		// a CREATE entry once fan-out is running (§3.6, §5.1).
		stageStart = time.Now()
		if err := ps.emitCreateAuditV2(txCtx, objectName, insertedID, currentUser); err != nil {
			logRecordStageTiming(txCtx, "Insert", "emitCreateAuditV2", objectName, insertedID, stageStart, err, len(logicalData))
			return err
		}
		logRecordStageTiming(txCtx, "Insert", "emitCreateAuditV2", objectName, insertedID, stageStart, nil, len(logicalData))

		// Default-child-row declarations: constant child rows authored on the
		// parent object's metadata. Runs inside the same transaction as the
		// parent insert so any child-write failure rolls back the parent too.
		stageStart = time.Now()
		if err := ps.applyDefaultChildRows(txCtx, schema, logicalData, currentUser); err != nil {
			logRecordStageTiming(txCtx, "Insert", "applyDefaultChildRows", objectName, insertedID, stageStart, err, len(logicalData))
			return appErrors.NewInternalError("failed to apply default child rows", err)
		}
		logRecordStageTiming(txCtx, "Insert", "applyDefaultChildRows", objectName, insertedID, stageStart, nil, len(logicalData))

		// Child-clone rules: post-insert side effects authored on the
		// target object's _System_Object.ChildCloneRules JSON column.
		// Runs inside the outer transaction so any clone failure aborts
		// the parent insert atomically, matching the deleted hardcoded
		// Quote→Order / Opportunity→Quote behavior.
		// Replaces the retired hardcoded cloneQuoteLineItems /
		// cloneOrderProductsFromQuote branches. Greenfield no-op: schemas
		// with nil/empty ChildCloneRules short-circuit in applyChildCloneRules.
		stageStart = time.Now()
		if err := ps.applyChildCloneRules(txCtx, schema, logicalData, currentUser); err != nil {
			logRecordStageTiming(txCtx, "Insert", "applyChildCloneRules", objectName, insertedID, stageStart, err, len(logicalData))
			return appErrors.NewInternalError("failed to apply child-clone rules", err)
		}
		logRecordStageTiming(txCtx, "Insert", "applyChildCloneRules", objectName, insertedID, stageStart, nil, len(logicalData))

		// Apply intercepted permission grants via the junction SSoT.
		// ApplyPermissionChanges validates catalog + prereqs, writes diff
		// rows, and records Setup Audit Trail entries. Runs after repo.Insert
		// so the FK on _System_PermissionSetPermission.PermissionSetId is valid.
		if len(preInsertPermChanges) > 0 {
			recordID, _ := logicalData[constants.FieldID].(string)
			if recordID != "" {
				stageStart = time.Now()
				if applyErr := ps.permissions.ApplyPermissionChanges(txCtx, recordID, nil, preInsertPermChanges, currentUser); applyErr != nil {
					logRecordStageTiming(txCtx, "Insert", "ApplyPermissionChanges", objectName, recordID, stageStart, applyErr, len(preInsertPermChanges))
					// Preserve IsInternal errors (catalog read / junction
					// write failures) as 500; only wrap plain validation
					// errors (unknown permission, prereq) as 400.
					if appErrors.IsInternal(applyErr) {
						return applyErr
					}
					return appErrors.NewValidationError("Permissions", applyErr.Error())
				}
				logRecordStageTiming(txCtx, "Insert", "ApplyPermissionChanges", objectName, recordID, stageStart, nil, len(preInsertPermChanges))
			}
		}

		stageStart = time.Now()
		if err := ps.runRecordLifecycleHooks(txCtx, recordLifecycleAfterCreateInTx, recordLifecycleEvent{
			ObjectName:  objectName,
			RecordID:    insertedID,
			Record:      cloneSObject(logicalData),
			CurrentUser: currentUser,
		}); err != nil {
			logRecordStageTiming(txCtx, "Insert", "runRecordLifecycleHooks.afterCreate", objectName, insertedID, stageStart, err, len(logicalData))
			return err
		}
		logRecordStageTiming(txCtx, "Insert", "runRecordLifecycleHooks.afterCreate", objectName, insertedID, stageStart, nil, len(logicalData))

		// Hook: Rollup Summary (inside transaction for ACID compliance)
		stageStart = time.Now()
		if err := ps.rollup.ProcessInsertRollups(txCtx, objectName, data); err != nil {
			logRecordStageTiming(txCtx, "Insert", "rollup.ProcessInsertRollups", objectName, insertedID, stageStart, err, len(data))
			return wrapRollupHookError("failed to process rollups", err)
		}
		logRecordStageTiming(txCtx, "Insert", "rollup.ProcessInsertRollups", objectName, insertedID, stageStart, nil, len(data))

		if createFinalImageRefreshRequired(schema) {
			stageStart = time.Now()
			finalCreateImage, err := ps.loadFinalCreateImage(txCtx, objectName, data)
			if err != nil {
				logRecordStageTiming(txCtx, "Insert", "loadFinalCreateImage", objectName, insertedID, stageStart, err, len(data))
				return err
			}
			data = finalCreateImage
			logRecordStageTiming(txCtx, "Insert", "loadFinalCreateImage", objectName, insertedID, stageStart, nil, len(data))
		}

		// Enqueue afterCreate event to outbox (inside transaction for guaranteed delivery)
		if ps.outbox != nil {
			enqueueCtx := txCtx
			if ps.activeFormulaProjector != nil {
				// Clear inherited active-projection keys before composing this
				// event. Nested writes triggered inside an after-save flow on
				// the outer record see txCtx with the outer's snapshot already
				// stamped; if THIS record's schema has no active fields,
				// project() returns nil and WithActiveProjections(..., nil, nil)
				// is a no-op, so without an explicit Clear, the outer's
				// active_projection_record key would survive on enqueueCtx and
				// FlowTriggerListener would overlay the wrong record's fields
				// onto THIS event's $Record (Codex #1390 R2 HIGH 1).
				enqueueCtx = eventFeature.ClearActiveProjections(enqueueCtx)
				if ps.activeFormulaProjectionRequired(txCtx, objectName, "RecordAfterSave", "Create") {
					// Project active-formula fields from the formula view for the new
					// record (view is a SECURITY INVOKER SELECT so it reflects the
					// just-inserted row within the same transaction).
					enqueueCtx = eventFeature.MarkActiveProjectionAttempted(enqueueCtx)
					insertedID, _ := data[constants.FieldID].(string)
					// Write-side elevation: the INSERT just passed user CRUD/FLS gates
					// (prepareInsertBeforeSave upstream) and the trigger payload
					// consumer (flow_trigger_listener) runs in SystemContext. The
					// adapter itself does not elevate, so we elevate here only on
					// the write-side path. Read-side reprojection in FlowExecutor
					// stays at the caller's security level (#1381).
					projCtx := constants.WithSecurityContext(
						txCtx, constants.SystemContext,
						"active-formula write-side projection for outbox payload (issue #1362)",
					)
					stageStart = time.Now()
					newSnap, projErr := ps.activeFormulaProjector.project(projCtx, currentUser, schema, insertedID)
					if projErr != nil {
						logRecordStageTiming(txCtx, "Insert", "activeFormulaProjector.project.new", objectName, insertedID, stageStart, projErr, len(data))
						return appErrors.NewInternalError("active-formula projection (insert) failed", projErr)
					}
					logRecordStageTiming(txCtx, "Insert", "activeFormulaProjector.project.new", objectName, insertedID, stageStart, nil, len(newSnap))
					// INSERT has no old record; pass nil for oldRec.
					stageStart = time.Now()
					updatedCtx, encErr := eventFeature.WithActiveProjections(enqueueCtx, newSnap, nil)
					if encErr != nil {
						logRecordStageTiming(txCtx, "Insert", "WithActiveProjections.new", objectName, insertedID, stageStart, encErr, len(newSnap))
						return appErrors.NewInternalError("encode active-projection snapshot (insert) failed", encErr)
					}
					logRecordStageTiming(txCtx, "Insert", "WithActiveProjections.new", objectName, insertedID, stageStart, nil, len(newSnap))
					enqueueCtx = updatedCtx
				}
			}
			stageStart = time.Now()
			outboxPayload := eventFeature.RecordEventPayload{
				ObjectAPIName: objectName,
				Record:        data,
				CurrentUser:   currentUser,
			}
			enqueueCtx, outboxPayload = eventFeature.PrepareRecordCreatedOutboxPayload(enqueueCtx, outboxPayload)
			if err := ps.enqueueRecordEvent(enqueueCtx, events.RecordCreated, outboxPayload); err != nil {
				logRecordStageTiming(txCtx, "Insert", "outbox.EnqueueEvent.RecordCreated", objectName, insertedID, stageStart, err, len(data))
				return appErrors.NewInternalError("failed to enqueue record created event", err)
			}
			logRecordStageTiming(txCtx, "Insert", "outbox.EnqueueEvent.RecordCreated", objectName, insertedID, stageStart, nil, len(data))
		}

		return nil
	}

	// Late-discovered parent-lock retry (mirror of Update /
	// RunInPessimisticTransaction): a synchronous record-triggered (AfterSave)
	// field-update on the new row can write a non-polymorphic lookup field whose
	// parent is discovered only after the child row is locked.
	// prelockInitialInsertParents reads forced parent intents from ctx; on a
	// RetryablePrelockError we force those parents and re-run the whole
	// transaction (fully rolled back between attempts) so the parent is locked
	// before the child. Each attempt starts from a clean clone of the prepared
	// data so before-save side effects (autonumber reservation, BeforeCreate
	// flow mutations) are not double-applied across attempts.
	forcedParentPrelocks := parentLockForcedPrelocksFromContext(ctx)
	canRetryParentPrelock := ps.canRetryParentLockPrelock(ctx)
	for attempt := 0; attempt < 3; attempt++ {
		data = cloneSObject(preparedData)
		retryCtx := withParentLockForcedPrelocks(ctx, forcedParentPrelocks)
		retryCtx = withRetryableInitialInsertParentLock(retryCtx, canRetryParentPrelock)
		// An expected-uniqueness marker applies to THIS insert only. Clearing
		// it here keeps it out of before-save flows and every other nested
		// write reached through txCtx, whose uniqueness conflicts are real.
		// This insert's own hook still sees it: the deferred emit reads the
		// context Insert was called with, not this derived one.
		retryCtx = withoutExpectedUniquenessConflict(retryCtx)
		stageStart = time.Now()
		err = ps.runRollupAwareTransaction(retryCtx, objectName, data, txnBody)
		logRecordStageTiming(retryCtx, "Insert", "runRollupAwareTransaction", objectName, "", stageStart, err, len(data))
		if err == nil {
			break
		}
		retryKeys, shouldRetry := parentLockPrelockRetryKeys(err)
		if !shouldRetry || !canRetryParentPrelock {
			break
		}
		var changed bool
		forcedParentPrelocks, changed = mergeParentLockKeys(forcedParentPrelocks, retryKeys)
		if !changed {
			break
		}
	}

	if err != nil {
		return nil, err
	}
	if permissionAuthoringRecordPreparationFromContext(ctx) != nil {
		return cloneSObject(data), nil
	}

	if !permissionAuthoringAggregateOwnsRecordsConvergence(ctx) {
		ps.invalidateSecurityCaches(ctx, objectName, nil, data)
	}
	id, _ := data[constants.FieldID].(string)
	log.Printf("✨ Created record %s in %s (User: %s)", id, objectName, getUserID(currentUser))

	return data, nil
}

type preparedInsertRequest struct {
	schema               *models.ObjectMetadata
	data                 models.SObject
	preInsertPermChanges map[string]bool
}

// PreviewBeforeInsert runs the create lifecycle through the final pre-save
// record image without inserting the row or firing post-insert side effects.
// It renders auto-number fields against the current counter state without
// reserving those counters; live Insert performs the durable reservation
// inside the write transaction.
func (ps *PersistenceService) PreviewBeforeInsert(
	ctx context.Context,
	objectName string,
	data models.SObject,
	currentUser *models.UserSession,
) (models.SObject, error) {
	reqID := requestIDFromCtx(ctx)
	zlog.Info().Str("request_id", reqID).Str("object", objectName).Str("user", getUserID(currentUser)).Msg("PersistenceService.PreviewBeforeInsert")

	// Stamp the preview-preparation marker on the OUTER ctx so the lookup
	// validators that run inside prepareInsertRequest
	// (validateNonPolymorphicLookups, and later validatePolymorphicLookups
	// inside prepareInsertBeforeSave) can detect "we are inside a no-DML
	// preview" and accept request-local synthetic preview-* IDs registered
	// by an upstream RecordCreate in the same flow run. Without this, a
	// Flow that chains Account-create → Contact-create-with-AccountId-bound-
	// to-the-Account-preview-Id would pass flowVersionCreatePreviewSupported
	// but fail at PreviewBeforeInsert when the validators rejected the
	// preview-* parent ID as a non-durable lookup target.
	previewCtx := constants.WithInsertPreviewPreparation(ctx)
	preparedInsert, err := ps.prepareInsertRequest(previewCtx, objectName, cloneSObject(data), currentUser)
	if err != nil {
		return nil, err
	}

	var finalData models.SObject
	var siblingMaxDefaultedFields []string
	runPreview := func(txCtx context.Context) error {
		prepared, _, defaultedFields, err := ps.prepareInsertBeforeSave(previewCtx, txCtx, objectName, preparedInsert.data, preparedInsert.schema, currentUser, false, nil)
		if err != nil {
			return err
		}
		if err := ps.rejectCurrentUserPasswordReuse(txCtx, objectName, prepared, currentUser); err != nil {
			return err
		}
		if strings.EqualFold(objectName, constants.TableUser) {
			if err := applyUserPasswordDefaults(prepared); err != nil {
				return appErrors.NewInternalError("user defaults failed", err)
			}
		}
		finalData = prepared
		siblingMaxDefaultedFields = defaultedFields
		return nil
	}
	requiresParentLockCoordinator, err := previewInsertRequiresParentLockCoordinator(preparedInsert.schema)
	if err != nil {
		return nil, err
	}
	if requiresParentLockCoordinator {
		err = ps.RunInPessimisticTransaction(previewCtx, func(txCtx context.Context) error {
			return ps.RunWithParentLockCoordinator(txCtx, runPreview)
		})
	} else {
		err = ps.RunInTransaction(previewCtx, runPreview)
	}
	if err != nil {
		return nil, err
	}
	constants.RegisterPreviewSiblingMaxDefaultedFields(previewCtx, siblingMaxDefaultedFields)
	return cloneSObject(finalData), nil
}

func previewInsertRequiresParentLockCoordinator(schema *models.ObjectMetadata) (bool, error) {
	if schema == nil {
		return false, nil
	}
	for i := range schema.Fields {
		field := &schema.Fields[i]
		parentFieldName := siblingMaxParentFieldName(field)
		if parentFieldName == "" {
			continue
		}
		if err := validateSiblingMaxRuntimeConfig(schema, field, parentFieldName); err != nil {
			return false, err
		}
		return true, nil
	}
	return false, nil
}

func (ps *PersistenceService) prepareInsertRequest(
	ctx context.Context,
	objectName string,
	data models.SObject,
	currentUser *models.UserSession,
) (*preparedInsertRequest, error) {
	stageStart := time.Now()
	schema, err := ps.prepareOperation(ctx, objectName, constants.PermCreate, currentUser)
	if err != nil {
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.prepareOperation", objectName, "", stageStart, err, len(data))
		return nil, err
	}
	logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.prepareOperation", objectName, "", stageStart, nil, len(data))

	// _System_RecordShare lock gate: a user may not add a share entry on a
	// record that is currently locked (LockRule evaluates true). Without this
	// gate, ShareRecordModal → dataAPI.createRecord → Insert bypasses the
	// backend lock check that guards Edit/Delete. Trusted contexts (SystemContext+,
	// system admin) bypass, matching the existing mutation gate in
	// checkRecordAccessInternal.
	if constants.GetSecurityContext(ctx) < constants.SystemContext &&
		strings.EqualFold(objectName, constants.TableRecordShare) {
		stageStart = time.Now()
		if err := ps.checkRecordShareParentNotLocked(ctx, data, currentUser); err != nil {
			logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.checkRecordShareParentNotLocked", objectName, "", stageStart, err, len(data))
			return nil, err
		}
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.checkRecordShareParentNotLocked", objectName, "", stageStart, nil, len(data))
	}

	// Defense-in-depth: only admins may set passwords on User records.
	if !constants.GetSecurityContext(ctx).BypassesProcessFields() {
		stageStart = time.Now()
		if err := ps.rejectUnauthorizedPasswordWrite(ctx, objectName, data, currentUser); err != nil {
			logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectUnauthorizedPasswordWrite", objectName, "", stageStart, err, len(data))
			return nil, err
		}
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectUnauthorizedPasswordWrite", objectName, "", stageStart, nil, len(data))
	}

	// Protected-records guard-field: forbid fabricating an "already-protected"
	// record via INSERT. Without this, a user with Profile create rights could
	// POST {"Name":"X","IsSystem":true} and create a row that immediately
	// triggers the delete guard. Same `< SystemContext` threshold as the
	// update path and the delete guard itself — consistent invariants.
	if constants.GetSecurityContext(ctx) < constants.SystemContext {
		stageStart = time.Now()
		if err := rejectProtectedGuardFieldWrites(objectName, data); err != nil {
			logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectProtectedGuardFieldWrites", objectName, "", stageStart, err, len(data))
			return nil, err
		}
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectProtectedGuardFieldWrites", objectName, "", stageStart, nil, len(data))
	}

	if !constants.GetSecurityContext(ctx).BypassesCRUDandFLS() {
		stageStart = time.Now()
		if err := rejectToolingOnlyTableWrites(objectName); err != nil {
			logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectToolingOnlyTableWrites", objectName, "", stageStart, err, len(data))
			return nil, err
		}
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectToolingOnlyTableWrites", objectName, "", stageStart, nil, len(data))
	}

	// Normalize field values (type coercion: multipicklist → JSON array string,
	// strings like "true"/"1" to bool, numbers, etc.). Unknown types on a
	// boolean field are left as-is so ExtractPermissionGrants below (and
	// ValidateSchemaCompliance on physicals) can reject them with a clear
	// error — see schema_utils.go:NormalizeSObject for the strict-bool case.
	stageStart = time.Now()
	data = schemautil.NormalizeSObject(schema, data)
	scrubPolymorphicDiscriminators(schema, data)
	logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.normalizeSObject", objectName, "", stageStart, nil, len(data))
	stageStart = time.Now()
	if err := ps.runRecordWriteCanonicalizers(ctx, recordWriteValidationBeforeCreate, recordWriteCanonicalizationEvent{
		ObjectName:  objectName,
		Record:      data,
		CurrentUser: currentUser,
	}); err != nil {
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.runRecordWriteCanonicalizers", objectName, "", stageStart, err, len(data))
		return nil, err
	}
	logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.runRecordWriteCanonicalizers", objectName, "", stageStart, nil, len(data))
	stageStart = time.Now()
	if err := rejectWriteProtectedFieldWrites(ctx, schema, data, protectedFieldWriteOnCreate); err != nil {
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectWriteProtectedFieldWrites", objectName, "", stageStart, err, len(data))
		return nil, err
	}
	logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectWriteProtectedFieldWrites", objectName, "", stageStart, nil, len(data))

	stageStart = time.Now()
	if err := rejectUnbypassableMetadataFieldWrites(objectName, data); err != nil {
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectUnbypassableMetadataFieldWrites", objectName, "", stageStart, err, len(data))
		return nil, err
	}
	logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectUnbypassableMetadataFieldWrites", objectName, "", stageStart, nil, len(data))
	if !constants.GetSecurityContext(ctx).BypassesCRUDandFLS() {
		stageStart = time.Now()
		if err := rejectImmutableMetadataFieldWrites(objectName, data); err != nil {
			logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectImmutableMetadataFieldWrites", objectName, "", stageStart, err, len(data))
			return nil, err
		}
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectImmutableMetadataFieldWrites", objectName, "", stageStart, nil, len(data))
	}

	if !constants.GetSecurityContext(ctx).BypassesProcessFields() {
		stageStart = time.Now()
		if err := ps.rejectExplicitAutoNumberWrites(ctx, objectName, data); err != nil {
			logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectExplicitAutoNumberWrites", objectName, "", stageStart, err, len(data))
			return nil, err
		}
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.rejectExplicitAutoNumberWrites", objectName, "", stageStart, nil, len(data))
	}

	// Virtual-field intercept: strip schema-declared virtual permission keys
	// (IsVirtual && VirtualKind == SystemPermission) from the payload. Schema-
	// driven, so no objectName branch — objects without permission virtuals
	// return an empty map and leave data untouched. Fails loud on non-bool
	// values so garbage input (which strict Normalize left uncoerced) can't
	// silently become a `false` grant / revoke.
	var preInsertPermChanges map[string]bool
	if ps.permissions != nil {
		stageStart = time.Now()
		extracted, extractErr := persistence.ExtractPermissionGrants(data, schema)
		if extractErr != nil {
			logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.ExtractPermissionGrants", objectName, "", stageStart, extractErr, len(data))
			return nil, appErrors.NewValidationError("Permissions", extractErr.Error())
		}
		preInsertPermChanges = extracted
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.ExtractPermissionGrants", objectName, "", stageStart, nil, len(data))
	}

	// Drop non-createable fields from user-submitted data.
	// Formula/system-managed fields still silently drop; metadata-declared
	// autonumber fields are rejected earlier by rejectExplicitAutoNumberWrites.
	// User-FLS Edit denial hard-fails with INVALID_FIELD_FOR_INSERT_UPDATE.
	if !constants.GetSecurityContext(ctx).BypassesProcessFields() {
		stageStart = time.Now()
		data, err = ps.pruneNonCreateableFields(ctx, objectName, data, schema, currentUser)
		if err != nil {
			logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.pruneNonCreateableFields", objectName, "", stageStart, err, len(data))
			return nil, err
		}
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.pruneNonCreateableFields", objectName, "", stageStart, nil, len(data))
	}

	// Non-polymorphic FK access / existence gate — named owner for the
	// Class-2 cutover-closure the round-4 reviewer required. Runs BEFORE
	// applyInsertInheritance so (a) inaccessible parents reject cleanly
	// with a 400 referencing the field name rather than the old 500 via
	// some downstream path (autonumber token resolve, etc.), and (b)
	// inheritance can then be a pure opportunistic copier instead of a
	// half-validator. Sibling of validatePolymorphicLookups; see
	// nonpoly_lookup_validator.go for the full rationale.
	stageStart = time.Now()
	// This is request preparation, not the transaction's parent-lock planner.
	// In particular, Runtime RecordMutation already supplies an outer
	// pessimistic transaction and coordinator scope. Acquiring a direct lookup
	// lock here would establish a high-water mark before Insert has discovered
	// the complete recursive rollup-parent set. The transaction body below
	// prelocks that complete set, then repeats lookup validation against the
	// locked image, so preparation must remain a read-only validation pass.
	if err := ps.validateNonPolymorphicLookups(withoutCoordinatedLookupParentLock(withNonPolymorphicLookupCreateIntent(ctx)), data, schema, currentUser); err != nil {
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.validateNonPolymorphicLookups", objectName, "", stageStart, err, len(data))
		return nil, err
	}
	logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.validateNonPolymorphicLookups", objectName, "", stageStart, nil, len(data))

	// Create-from-source bypass guard (issue #1234). Rejects a direct
	// sObject create that populates a source-FK owned by a configured
	// standard action whose Flow performs the multi-record fan-out. Runs
	// AFTER pruneNonCreateableFields (FLS denial above) and AFTER
	// validateNonPolymorphicLookups (FK access check above), so FLS errors
	// and inaccessible-parent errors retain precedence over the new
	// INVALID_OPERATION envelope. Runs BEFORE applyInsertInheritance so
	// inheritance never silently completes a create that would have been
	// rejected. No-op when no action owns a (T, SourceFKField) entry, or
	// when the body sets no configured source-FK, or when ctx carries the
	// owning action's id. See guardCreateFromSourceContext for the full
	// semantic contract.
	stageStart = time.Now()
	if err := ps.guardCreateFromSourceContext(ctx, objectName, data); err != nil {
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.guardCreateFromSourceContext", objectName, "", stageStart, err, len(data))
		return nil, err
	}
	logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.guardCreateFromSourceContext", objectName, "", stageStart, nil, len(data))

	// Insert-inheritance: copy fields from a lookup-referenced parent record
	// onto the new record per _System_ObjectFieldMapping rules with
	// ActionType='insertInheritFromLookup'. Runs after the createability
	// filter so system-populated values survive the user-input pruning pass,
	// and before validation so inherited defaults (e.g. Status="Pending"
	// expressed as a constant-default row) participate in ValidationRule
	// evaluation. No-op when the target object has no such rules — the
	// registry read returns zero rows and the applier early-returns.
	// Replaces the deleted hardcoded enrichQuoteFromOpportunity branch
	// (see docs/plans: §4 "Generic persistence hooks", §6.5 hard-fail).
	stageStart = time.Now()
	inheritanceModified, err := ps.applyInsertInheritance(ctx, objectName, data, schema, currentUser)
	if err != nil {
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.applyInsertInheritance", objectName, "", stageStart, err, len(data))
		return nil, err
	}
	logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.applyInsertInheritance", objectName, "", stageStart, nil, len(data))
	if inheritanceModified {
		// Inherited values may have landed as raw strings into JSON /
		// MultiPicklist columns, or introduced values that need the
		// createability pass (e.g. a parent's OwnerId that the user
		// isn't allowed to set). Mirrors the old Quote-branch's second
		// normalize+prune; runs only when rules actually fired.
		stageStart = time.Now()
		data = schemautil.NormalizeSObject(schema, data)
		scrubPolymorphicDiscriminators(schema, data)
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.inheritance.normalizeSObject", objectName, "", stageStart, nil, len(data))
		stageStart = time.Now()
		if err := rejectWriteProtectedFieldWrites(ctx, schema, data, protectedFieldWriteOnCreate); err != nil {
			logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.inheritance.rejectWriteProtectedFieldWrites", objectName, "", stageStart, err, len(data))
			return nil, err
		}
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.inheritance.rejectWriteProtectedFieldWrites", objectName, "", stageStart, nil, len(data))
		if !constants.GetSecurityContext(ctx).BypassesProcessFields() {
			stageStart = time.Now()
			data, err = ps.pruneNonCreateableFields(ctx, objectName, data, schema, currentUser)
			if err != nil {
				logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.inheritance.pruneNonCreateableFields", objectName, "", stageStart, err, len(data))
				return nil, err
			}
			logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.inheritance.pruneNonCreateableFields", objectName, "", stageStart, nil, len(data))
		}
	}

	// Apply metadata-owned defaults after inheritance so seeded
	// insertInheritFromLookup rules outrank schema defaults when both target the
	// same blank field (e.g. Quote.Status Pending vs schema default Draft).
	stageStart = time.Now()
	data = applyFieldDefaults(data, schema)
	logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.applyFieldDefaults", objectName, "", stageStart, nil, len(data))

	// Generate system fields dynamically from metadata (Must happen before validation)
	// These include CreatedById, IsDeleted, etc. — added back by the system, not user.
	stageStart = time.Now()
	systemFields := ps.generateSystemFields(ctx, objectName, data, currentUser, true)

	// Merge data with system fields
	for k, v := range systemFields {
		data[k] = v
	}
	logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.generateSystemFields", objectName, "", stageStart, nil, len(data))

	// Metadata-declared compound parents such as Name may be written by clients
	// through member fields. Synthesize before validation from the canonical
	// source-backed runtime declaration.
	stageStart = time.Now()
	applyCompoundNameSynthesis(schema, data, false)
	logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.applyCompoundNameSynthesis", objectName, "", stageStart, nil, len(data))

	// 0. Strict Schema Validation (Bouncer)
	stageStart = time.Now()
	if err := ps.schemaValidator.ValidateSchemaCompliance(data, schema, false, true, false); err != nil {
		logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.schemaValidator.ValidateSchemaCompliance", objectName, "", stageStart, err, len(data))
		return nil, err
	}
	logRecordStageTiming(ctx, "Insert", "prepareInsertRequest.schemaValidator.ValidateSchemaCompliance", objectName, "", stageStart, nil, len(data))
	applyCountRollupSummaryDefaults(data, schema)

	// CRITICAL: Ensure _System_User username defaults are visible before
	// before-save flows and validation. Password hashing/default stamping is
	// delayed until after before-save flows and password-reuse validation so
	// the final pre-save record image has one authoritative read path.
	if strings.EqualFold(objectName, constants.TableUser) {
		applyUserUsernameDefault(data)
	}

	return &preparedInsertRequest{
		schema:               schema,
		data:                 data,
		preInsertPermChanges: preInsertPermChanges,
	}, nil
}

func (ps *PersistenceService) prepareInsertBeforeSave(
	ctx context.Context,
	txCtx context.Context,
	objectName string,
	data models.SObject,
	schema *models.ObjectMetadata,
	currentUser *models.UserSession,
	reserveAutoNumbers bool,
	insertObservation *insertFinalImageObservation,
) (models.SObject, models.SObject, []string, error) {
	// Autonumber v2: assign rendered identifiers via the engine for every
	// field on this object with AutoNumberFormat set. Live insert reserves
	// counters; no-DML preview renders from the current counter state so
	// compound fields, before-save flows, formulas, and validators see the
	// same class of value without incrementing _System_AutoNumberCounter.
	if reserveAutoNumbers {
		stageStart := time.Now()
		if err := ps.assignAutoNumbers(txCtx, objectName, data); err != nil {
			logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.assignAutoNumbers", objectName, "", stageStart, err, len(data))
			return nil, nil, nil, err
		}
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.assignAutoNumbers", objectName, "", stageStart, nil, len(data))
	} else if constants.HasInsertPreviewPreparation(ctx) {
		stageStart := time.Now()
		if err := ps.assignPreviewAutoNumbers(txCtx, objectName, data); err != nil {
			logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.assignPreviewAutoNumbers", objectName, "", stageStart, err, len(data))
			return nil, nil, nil, err
		}
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.assignPreviewAutoNumbers", objectName, "", stageStart, nil, len(data))
	}

	// Resolve compound parent fields (isCompound=true) from their declared
	// member fields (composedFrom). Runs AFTER autonumber assignment on live
	// insert so autonumber-derived members (e.g. Quote.QuoteNumber) are already
	// rendered. Subsumes the old hardcoded Quote.Name ← QuoteNumber fallback
	// with a metadata-driven resolver. Issue #1582 RFC v1.5 PR C.
	stageStart := time.Now()
	ps.resolveCompoundFields(schema, data)
	logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.resolveCompoundFields", objectName, "", stageStart, nil, len(data))
	protectedBaseline, err := captureProtectedCreateBaseline(schema, data)
	if err != nil {
		return nil, nil, nil, err
	}

	// Publish beforeCreate event (synchronous, can fail transaction).
	// Flows here may mutate `data` (e.g. repointing an FK).
	stageStart = time.Now()
	if err := ps.publishRecordEvent(txCtx, events.RecordBeforeCreate, objectName, data, nil, currentUser); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.publishRecordEvent.RecordBeforeCreate", objectName, "", stageStart, err, len(data))
		return nil, nil, nil, err
	}
	logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.publishRecordEvent.RecordBeforeCreate", objectName, "", stageStart, nil, len(data))

	// Re-normalize after BeforeSave flows may have written raw values
	// (e.g. plain strings into JSON/MultiPicklist columns).
	stageStart = time.Now()
	data = schemautil.NormalizeSObject(schema, data)
	scrubPolymorphicDiscriminators(schema, data)
	if err := ps.runRecordWriteCanonicalizers(txCtx, recordWriteValidationBeforeCreate, recordWriteCanonicalizationEvent{
		ObjectName:  objectName,
		Record:      data,
		CurrentUser: currentUser,
	}); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.normalizeAfterBeforeCreate", objectName, "", stageStart, err, len(data))
		return nil, nil, nil, err
	}
	if err := rejectUnbypassableMetadataFieldWrites(objectName, data); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.rejectUnbypassableMetadataFieldWrites", objectName, "", stageStart, err, len(data))
		return nil, nil, nil, err
	}
	if err := protectedBaseline.validate(ctx, schema, data); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.normalizeAfterBeforeCreate", objectName, "", stageStart, err, len(data))
		return nil, nil, nil, err
	}
	logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.normalizeAfterBeforeCreate", objectName, "", stageStart, nil, len(data))
	// Before-save flows can repoint ordinary single-target lookups. Re-run the
	// same access/existence owner on the final pre-save image before any
	// parent-dependent defaults (sibling-max) or SQL writes observe that value.
	stageStart = time.Now()
	if err := ps.validateNonPolymorphicLookups(withNonPolymorphicLookupCreateIntent(withoutCoordinatedLookupParentLock(txCtx)), data, schema, currentUser); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateNonPolymorphicLookups.prelockless", objectName, "", stageStart, err, len(data))
		return nil, nil, nil, err
	}
	logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateNonPolymorphicLookups.prelockless", objectName, "", stageStart, nil, len(data))
	stageStart = time.Now()
	if _, err := ps.prelockInitialInsertParents(txCtx, objectName, schema, data); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.prelockInitialInsertParents", objectName, "", stageStart, err, len(data))
		return nil, nil, nil, err
	}
	logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.prelockInitialInsertParents", objectName, "", stageStart, nil, len(data))
	stageStart = time.Now()
	if err := ps.validateNonPolymorphicLookups(withNonPolymorphicLookupCreateIntent(withoutCoordinatedLookupParentLock(txCtx)), data, schema, currentUser); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateNonPolymorphicLookups.prelocked", objectName, "", stageStart, err, len(data))
		return nil, nil, nil, err
	}
	logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateNonPolymorphicLookups.prelocked", objectName, "", stageStart, nil, len(data))
	stageStart = time.Now()
	siblingMaxDefaultedFields, err := ps.applySiblingMaxDefaults(txCtx, objectName, data, schema)
	if err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.applySiblingMaxDefaults", objectName, "", stageStart, err, len(data))
		return nil, nil, nil, err
	}
	logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.applySiblingMaxDefaults", objectName, "", stageStart, nil, len(data))
	validationErrors := ps.newClientValidationErrorCollector()
	stageStart = time.Now()
	if err := validateRuntimeScopedPicklistRequiredValues(schema, data); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateRuntimeScopedPicklistRequiredValues", objectName, "", stageStart, err, len(data))
		if terminal := validationErrors.add(err); terminal != nil {
			return nil, nil, nil, terminal
		}
	} else {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateRuntimeScopedPicklistRequiredValues", objectName, "", stageStart, nil, len(data))
	}

	// State-dependent validators run on the FINAL post-flow image
	// (PR #2057 round-6 F2 — Pass B blocker). A BeforeSave flow that
	// mutates RecordTypeId, OwnerId, an Order's QuoteId, or any
	// polymorphic FK gets re-validated here so the row that lands in
	// SQL has been checked against the validators it claims to satisfy.
	stageStart = time.Now()
	if err := ps.validateRecordTypeId(txCtx, objectName, data, currentUser); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateRecordTypeId", objectName, "", stageStart, err, len(data))
		if terminal := validationErrors.add(err); terminal != nil {
			return nil, nil, nil, terminal
		}
	} else {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateRecordTypeId", objectName, "", stageStart, nil, len(data))
	}
	stageStart = time.Now()
	if err := ps.validateQueueAssignment(txCtx, objectName, data, currentUser); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateQueueAssignment", objectName, "", stageStart, err, len(data))
		if terminal := validationErrors.add(err); terminal != nil {
			return nil, nil, nil, terminal
		}
	} else {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateQueueAssignment", objectName, "", stageStart, nil, len(data))
	}
	normalizeGroupUserRoleReferenceWriteImage(objectName, data)
	stageStart = time.Now()
	resolvedTypes, err := ps.validatePolymorphicLookups(txCtx, data, schema)
	if err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validatePolymorphicLookups", objectName, "", stageStart, err, len(data))
		return nil, nil, nil, err
	}
	logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validatePolymorphicLookups", objectName, "", stageStart, nil, len(data))
	// Inject resolved types into data for persistence — must happen AFTER
	// BeforeSave (FK may have been repointed) and BEFORE compute (formulas
	// may reference polymorphic-resolved types) and SQL write.
	for fieldName, objType := range resolvedTypes {
		data[schemautil.GetPolymorphicTypeColumnName(fieldName)] = objType
	}

	// Compute frozen-mode formula fields ONCE on the post-BeforeSave record
	// image. On INSERT the row IS the SQL write payload, so writing back into
	// `data` is sufficient — no inject step needed. Fetch-error map is
	// discarded: an INSERT fetch failure leaves the field absent from `data`
	// (no SQL write).
	stageStart = time.Now()
	_ = computeFormulaFields(txCtx, schema, data, ps.repo, ps.metadata, objectName)
	logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.computeFormulaFields", objectName, "", stageStart, nil, len(data))

	// Validation Rules + uniqueness on the FINAL post-flow + post-compute
	// record image (Salesforce order of execution: VR runs after before-save
	// flows and before save).
	validationRules, err := ps.metadata.GetValidationRulesForMutation(txCtx, objectName)
	if err != nil {
		return nil, nil, nil, appErrors.NewInternalError("load mutation validation rules", err)
	}
	userCtx := ps.buildUserContext(txCtx, currentUser)
	stageStart = time.Now()
	if err := ps.validateStaticDependentPicklists(schema, data, nil); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateStaticDependentPicklists", objectName, "", stageStart, err, len(data))
		if terminal := validationErrors.add(err); terminal != nil {
			return nil, nil, nil, terminal
		}
	} else {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateStaticDependentPicklists", objectName, "", stageStart, nil, len(data))
	}
	stageStart = time.Now()
	if err := ps.validateRuntimeScopedPicklistsAfterRequiredValues(txCtx, schema, data, nil); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateRuntimeScopedPicklists", objectName, "", stageStart, err, len(data))
		if terminal := validationErrors.add(err); terminal != nil {
			return nil, nil, nil, terminal
		}
	} else {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateRuntimeScopedPicklists", objectName, "", stageStart, nil, len(data))
	}
	stageStart = time.Now()
	if err := ps.validateRuntimeScopedLookups(txCtx, schema, data, nil); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateRuntimeScopedLookups", objectName, "", stageStart, err, len(data))
		if terminal := validationErrors.add(err); terminal != nil {
			return nil, nil, nil, terminal
		}
	} else {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validateRuntimeScopedLookups", objectName, "", stageStart, nil, len(data))
	}
	stageStart = time.Now()
	recordID, _ := data[constants.FieldID].(string)
	if err := ps.runRecordWriteValidators(txCtx, recordWriteValidationBeforeCreate, recordWriteValidationEvent{
		ObjectName:  objectName,
		RecordID:    recordID,
		Record:      data,
		CurrentUser: currentUser,
	}); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.runRecordWriteValidators", objectName, recordID, stageStart, err, len(data))
		if terminal := validationErrors.add(err); terminal != nil {
			return nil, nil, nil, terminal
		}
	} else {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.runRecordWriteValidators", objectName, recordID, stageStart, nil, len(data))
	}
	stageStart = time.Now()
	if err := ps.validator.ValidateRecord(txCtx, data, schema, validationRules, nil, userCtx); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validator.ValidateRecord", objectName, "", stageStart, err, len(data))
		if terminal := validationErrors.add(err); terminal != nil {
			return nil, nil, nil, terminal
		}
	} else {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.validator.ValidateRecord", objectName, "", stageStart, nil, len(data))
	}
	if err := validationErrors.combined(); err != nil {
		return nil, nil, nil, err
	}
	if insertObservation != nil {
		if err := insertObservation.observe(filterToLogicalFields(schema, data)); err != nil {
			return nil, nil, nil, err
		}
	}
	stageStart = time.Now()
	if err := ps.checkUniqueness(txCtx, objectName, data, schema, ""); err != nil {
		logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.checkUniqueness", objectName, "", stageStart, err, len(data))
		return nil, nil, nil, insertObservation.wrapDirectInsertError(externalIDUpsertInsertUniquenessPrecheck, err)
	}
	logRecordStageTiming(txCtx, "Insert", "prepareInsertBeforeSave.checkUniqueness", objectName, "", stageStart, nil, len(data))

	return data, filterToLogicalFields(schema, data), siblingMaxDefaultedFields, nil
}

// pruneNonCreateableFields enforces write-path field capability on CREATE.
// Mirrors Salesforce v63.0 (verified against SF sandbox):
//   - Intrinsic non-createable formula/system-managed fields: silently dropped.
//   - Metadata-declared autonumber fields: rejected earlier by
//     rejectExplicitAutoNumberWrites so callers get INVALID_FIELD_FOR_INSERT_UPDATE.
//   - User-FLS Edit denial: hard-fail with INVALID_FIELD_FOR_INSERT_UPDATE.
//     SF returns 400 with the denied field name in the fields array; we
//     propagate via FLSDeniedError so the handler emits the same shape.
func (ps *PersistenceService) pruneNonCreateableFields(
	ctx context.Context,
	objectName string,
	data models.SObject,
	schema *models.ObjectMetadata,
	currentUser *models.UserSession,
) (models.SObject, error) {
	if len(data) == 0 || schema == nil {
		return data, nil
	}

	resolver := permissions.NewFieldAccessResolver(ps.permissions)
	bypassUserFLS := constants.GetSecurityContext(ctx).BypassesCRUDandFLS() ||
		constants.IsScopedInternalAccess(ctx, objectName, constants.PermCreate)
	// Collect ALL FLS-denied fields so the error body matches SF byte-for-byte.
	var deniedFields []string
	for key := range data {
		for i := range schema.Fields {
			if schema.Fields[i].APIName != key {
				continue
			}
			// Intrinsic capability check — silently drop non-createable (formula, autoNumber, system).
			if !resolver.IsIntrinsicallyCreateable(objectName, &schema.Fields[i]) {
				delete(data, key)
				break
			}
			// Intrinsically createable but user lacks FLS Edit — accumulate.
			if !bypassUserFLS {
				createable, err := resolver.EffectiveCreateable(ctx, &schema.Fields[i], objectName, currentUser)
				if err != nil {
					return nil, err
				}
				if !createable {
					deniedFields = append(deniedFields, key)
				}
			}
			break
		}
	}
	if len(deniedFields) > 0 {
		return nil, appErrors.NewFLSDeniedErrorForFields(deniedFields)
	}

	return data, nil
}

// filterToLogicalFields returns a new SObject containing only fields that
// are persisted as physical columns on the object's storage table. Virtual
// fields (schema-declared with IsVirtual=true) are explicitly excluded:
// they have no physical column, so passing them to repo.Insert / repo.Update
// would produce an "unknown column" error at the DB. Upstream write paths
// extract virtuals into a separate kind-specific grants map before calling
// this function; this filter is the belt to that extract's suspenders.
func filterToLogicalFields(schema *models.ObjectMetadata, data models.SObject) models.SObject {
	out := make(models.SObject)

	// Build map of persistable fields only — virtuals are deliberately
	// excluded so they cannot reach a physical SQL write even if an
	// upstream intercept was missed.
	validFields := make(map[string]bool)
	for _, f := range models.PersistableFields(schema) {
		validFields[f.APIName] = true
	}
	// Always include ID
	validFields[constants.FieldID] = true

	// Filter data
	for k, v := range data {
		if validFields[k] {
			out[k] = v
		}
	}
	return out
}

// checkRecordShareParentNotLocked enforces the LockRule gate on the PARENT
// record referenced by a _System_RecordShare row. Share inserts and deletes
// mutate the parent record's access surface; if the parent is locked the
// mutation must be denied regardless of ManageSharing permission.
//
// shareData must contain ObjectApiName and RecordId (present in both the
// insert payload and a loaded share row via Delete).
func (ps *PersistenceService) checkRecordShareParentNotLocked(ctx context.Context, shareData models.SObject, user *models.UserSession) error {
	parentAPIName, _ := shareData[constants.FieldSysRecordShare_ObjectAPIName].(string)
	parentID, _ := shareData[constants.FieldSysRecordShare_RecordID].(string)
	if parentAPIName == "" || parentID == "" {
		// Incomplete share data — allow and let downstream validation handle it.
		return nil
	}
	if ps.permissions == nil {
		return nil
	}
	parentSchema, err := ps.metadata.GetSchemaOrError(ctx, parentAPIName)
	if err != nil {
		// Schema lookup failed — fail closed: deny the share mutation to avoid
		// bypassing the lock gate on a transient metadata error.
		return fmt.Errorf("checkRecordShareParentNotLocked: schema lookup for %q failed: %w", parentAPIName, err)
	}
	if parentSchema == nil {
		return fmt.Errorf("checkRecordShareParentNotLocked: schema for %q not found", parentAPIName)
	}
	parentRecord, err := ps.repo.FindOne(ctx, parentAPIName, parentID)
	if err != nil {
		// Record lookup failed — fail closed: deny to prevent authorization bypass
		// on transient DB errors.
		return fmt.Errorf("checkRecordShareParentNotLocked: parent record %s/%s lookup failed: %w", parentAPIName, parentID, err)
	}
	if parentRecord == nil {
		return fmt.Errorf("checkRecordShareParentNotLocked: parent record %s/%s not found", parentAPIName, parentID)
	}
	// Use fail-closed gate: a user who has ManageSharing but not PermEdit
	// on a read-only parent must still be allowed to manage shares when the
	// record is NOT locked. Using PermEdit was over-restrictive and produced
	// false-positive lock denials for such users. The authoritative gate is
	// whether the record is currently locked (approval-lock or LockRule),
	// not whether the requesting user can edit the record.
	// IsRecordLockedForGate is used instead of CheckRecordAccessAll().IsLocked
	// so that LockRule evaluation errors count as locked (fail-closed), matching
	// the semantics of the edit/delete gate in checkRecordAccessInternal.
	if ps.permissions.IsRecordLockedForGate(ctx, parentSchema, parentRecord, user) {
		return fmt.Errorf("record %s/%s is locked: share mutations are not allowed on a locked record", parentAPIName, parentID)
	}
	return nil
}