// Rejected-save audit (issue #4261).
//
// A save refused by the validation pipeline used to leave zero trace:
// _System_AuditLog carried Outcome/ErrorCode columns, and audit.OutcomeFailure
// had existed since the public-core baseline, but nothing ever produced a
// FAILURE row. Support could only reconstruct "why did the CRM reject my save"
// by diffing the writes that eventually landed against the active rules.
//
// This file is the producer. It hangs off the single-record write entry
// points (Insert, Update, CommitPreparedUpdate, RestoreRecord) rather than off
// the individual checks, because the rejection reaches those entry points as a
// typed error and every check — including the post-BeforeSave re-validation
// pass — funnels through them.
//
// BulkInsert rejects PER RECORD inside a batch and usually returns a nil
// error, so it cannot key off an error return at all; its wiring is a
// follow-up on top of this file.
//
// # Two contracts this file owns
//
//  1. BEST EFFORT. Audit emission must never affect the user's request. Every
//     path here swallows its own errors (and panics) after logging. The
//     mutation test TestRejectedSaveAudit_SinkFailureDoesNotAffectRequest
//     forces the sink to fail and asserts the caller's error is untouched.
//     An audit path that can turn a 400 into a 500 is worse than no audit
//     path.
//
//  2. FIELD NAMES ONLY, NEVER VALUES. The row records which fields were in
//     the caller's payload and which field/rule the rejection names — never
//     what was in them. This is enforced structurally rather than by
//     filtering: we read `.Field` / `.Fields` / `.RuleName` off the typed
//     errors and never touch `.Error()`, `.Message`, or `ConflictError.Value`.
//     Those renderings DO embed values (ConflictError formats
//     "already exists with X='<value>'", the picklist validator formats
//     "value %q is not an active picklist option", and a validation rule's
//     admin-authored ErrorMessage can render merge fields). Because no
//     value-bearing string is ever read, there is nothing to scrub and no
//     scrubber to keep in sync.
package records

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/nexuscrm/audit"
	"github.com/nexuscrm/kernel/pkg/apierr"
	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/shared/pkg/constants"
	sharedErrors "github.com/nexuscrm/shared/pkg/errors"
	"github.com/nexuscrm/shared/pkg/models"
)

const rejectedSaveAuditEnabledEnv = "NEXUSCRM_AUDIT_REJECTED_SAVES_ENABLED"

// rejectedSaveAuditDisabledFromEnv is the operator kill switch. Default-on is
// the point of #4261 — a deployment that forgets to set the flag must still
// have forensics — so it uses the package's shared fail-safe reader.
func rejectedSaveAuditDisabledFromEnv() bool {
	return featureDisabledFromEnv(rejectedSaveAuditEnabledEnv)
}

// Rejection stages recorded in ChangesJSON.rejectionStage.
//
// The taxonomy is derived from the TYPED ERROR, which is structural, and
// deliberately stops there. #4261 proposed splitting `required_field` and
// `picklist` out of general field validation, but both arrive as
// *appErrors.ValidationError and differ only in free-text message prose
// ("is required" / "value %q is not an active picklist option"). Deriving a
// sub-stage would mean parsing that prose — brittle against rewording, and
// the prose is exactly the value-bearing string the PII rule forbids reading.
//
// The distinction is still recoverable downstream without parsing anything:
// a rejection whose named field is ABSENT from requestedFields is a
// missing-required-field; one whose field is PRESENT is a bad value in a
// field the caller actually sent.
const (
	rejectionStageFieldValidation = "field_validation"
	rejectionStageValidationRule  = "validation_rule"
	rejectionStageFLS             = "fls"
	rejectionStageUniqueness      = "uniqueness"
	// rejectionStageDomainRule covers the platform's dialect-neutral write
	// refusals: invalid/inaccessible lookup targets, field-integrity
	// refusals, and guard-blocked operations. These arrive as
	// *sharedErrors.DomainWireError, which is NOT one of the four typed
	// errors the #4261 handoff listed — but it is the type behind some of the
	// commonest rejections there are (an OwnerId pointing at a user who no
	// longer exists, for one), so leaving it unclassified would have kept a
	// large hole open under a surface that now looks complete.
	rejectionStageDomainRule = "domain_rule"
	// rejectionStageLifecycle covers the *apierr.Error 4xx refusals raised by
	// record write validators — role-hierarchy cycles, lifecycle conflicts.
	rejectionStageLifecycle = "lifecycle"
)

// rejectedSave is the classified, value-free identity of one rejection.
type rejectedSave struct {
	stage string
	// fields are the field API names the rejection names. Plural because an
	// FLS denial reports every denied field in one error.
	fields []string
	// rule is the validation-rule developer name, for stage=validation_rule.
	rule string
	// errorCode is the wire error code the caller received.
	errorCode string
}

// classifyRejectedSave maps a write-path error onto a rejection stage, or
// reports false when the error is not a validation rejection at all.
//
// IsInternal is checked FIRST and is not merely an optimization: the
// transaction manager wraps rollback failures as InternalError(cause=businessErr),
// so an unguarded errors.As would unwrap through the InternalError into the
// business cause and record a genuine infrastructure failure as a user
// validation rejection. Same discipline as AsClientValidationError and the
// bulk orchestrator. Infra failures are not rejected saves and get no row.
func classifyRejectedSave(err error) (rejectedSave, bool) {
	if err == nil || appErrors.IsInternal(err) {
		return rejectedSave{}, false
	}

	// Uniqueness surfaces as ConflictError, not as one of the three types the
	// #4261 handoff listed. Only .Field is read; .Value and the rendered
	// message carry the conflicting value and are never touched.
	if conflictErr, ok := appErrors.AsConflict(err); ok {
		return rejectedSave{
			stage:     rejectionStageUniqueness,
			fields:    nonEmptyFieldNames(conflictErr.Field),
			errorCode: conflictErr.Code(),
		}, true
	}

	// AsClientValidationError is reused rather than re-implemented so this
	// classifier and the REST error mapping agree on what counts as a
	// client-visible validation failure.
	typed, ok := appErrors.AsClientValidationError(err)
	if !ok {
		return classifyNonKernelRejection(err)
	}
	leaves, ok := appErrors.AsClientValidationErrors(typed)
	if !ok || len(leaves) == 0 {
		return rejectedSave{}, false
	}
	firstLeaf := leaves[0]
	if failures, ok := appErrors.AsValidationRuleFailures(firstLeaf); ok {
		first := failures[0]
		return rejectedSave{
			stage:     rejectionStageValidationRule,
			rule:      first.RuleName,
			fields:    nonEmptyFieldNames(first.Field),
			errorCode: string(sharedErrors.ErrValidation),
		}, true
	}
	if denied, ok := firstLeaf.(sharedErrors.FieldWriteDeniedConditioner); ok {
		return rejectedSave{
			stage:     rejectionStageFLS,
			fields:    sortedCopy(denied.FieldWriteDeniedFields()),
			errorCode: string(sharedErrors.ErrValidation),
		}, true
	}
	if validation, ok := firstLeaf.(sharedErrors.ValidationFieldConditioner); ok {
		return rejectedSave{
			stage:     rejectionStageFieldValidation,
			fields:    nonEmptyFieldNames(validation.ValidationField()),
			errorCode: string(sharedErrors.ErrValidation),
		}, true
	}
	return rejectedSave{}, false
}

// classifyNonKernelRejection handles the two write-refusal families that live
// outside kernel/pkg/errors and therefore outside AsClientValidationError.
//
// Both reach the hooked entry points as ordinary user-visible 4xx rejections,
// so omitting them would leave the exact blind spot #4261 exists to close —
// and would leave it under a surface that now looks complete, which is worse
// than the original silence: support would read "no FAILURE row" as "no
// rejection happened".
//
// As everywhere in this file, only structural identity is read. Neither
// error's Message is touched: DomainWireError renders the offending record id
// for an invalid reference, and apierr messages are free-form.
func classifyNonKernelRejection(err error) (rejectedSave, bool) {
	var wireErr *sharedErrors.DomainWireError
	if errors.As(err, &wireErr) && wireErr != nil {
		if !wireConditionIsWriteRejection(wireErr.Kind) {
			return rejectedSave{}, false
		}
		return rejectedSave{
			stage:     rejectionStageDomainRule,
			fields:    sortedCopy(wireErr.Fields),
			errorCode: string(wireErr.Kind),
		}, true
	}

	// The external-ID upsert's adoption guard replaces the underlying conflict
	// with a sentinel and keeps the diagnostics as TEXT, not as an error cause
	// — deliberately, so the detail never leaks. That means errors.As finds no
	// *ConflictError here, and without this branch a terminal adoption
	// conflict would produce no row at all: the probe suppression has already
	// silenced the inner attempt, so this is the last chance to record it.
	if errors.Is(err, ports.ErrExternalIDUpsertAdoptionGuardConflict) {
		return rejectedSave{
			stage:     rejectionStageUniqueness,
			errorCode: string(sharedErrors.WireInvalidOperation),
		}, true
	}

	var apiErr *apierr.Error
	if errors.As(err, &apiErr) && apiErr != nil {
		// Only client-fault statuses. A 5xx apierr is infrastructure and is
		// not a rejected save, mirroring the IsInternal guard above.
		if apiErr.Status < 400 || apiErr.Status >= 500 {
			return rejectedSave{}, false
		}
		return rejectedSave{
			stage:     rejectionStageLifecycle,
			errorCode: apiErr.Code,
		}, true
	}

	return rejectedSave{}, false
}

// wireConditionIsWriteRejection selects the DomainWireError kinds that mean
// "the user's write was refused". NotFound / MalformedRequest /
// RequestTooLarge are request-shape or routing failures rather than a save the
// validation pipeline evaluated and declined, so they are left unrecorded.
func wireConditionIsWriteRejection(kind sharedErrors.WireConditionKind) bool {
	switch kind {
	case sharedErrors.WireInvalidReference,
		sharedErrors.WireFieldIntegrity,
		sharedErrors.WireInvalidOperation,
		sharedErrors.WireInvalidField:
		return true
	default:
		return false
	}
}

// rejectedSaveChanges is the ChangesJSON payload. #4261 chose to reuse the
// existing LONGTEXT column with a distinct key set rather than add columns,
// so this ships with no schema change. Slices are sorted so the same logical
// rejection always serializes to the same bytes.
type rejectedSaveChanges struct {
	RejectionStage string `json:"rejectionStage"`
	// Fields are the field API names the rejection names.
	Fields []string `json:"fields,omitempty"`
	// UnknownFieldCount is how many names were dropped because they are not
	// declared fields on this object. See recordableFieldNames — the names
	// themselves are deliberately NOT recorded, because on the unknown-field
	// path they are caller-controlled strings that can be record data.
	UnknownFieldCount int `json:"unknownFieldCount,omitempty"`
	// Rule is the validation-rule developer name.
	Rule string `json:"rule,omitempty"`
	// RequestedFields is the caller's payload field-name set — "what was the
	// user trying to do". #4261: per-field audit rows for SUCCESSFUL saves
	// conflate the caller's payload with fields stamped by before-save flows
	// in the same transaction, which already caused one misattribution during
	// the investigation. For a rejected save nothing was stamped, so this set
	// is exactly the caller's intent.
	RequestedFields []string `json:"requestedFields"`
}

// requestedFieldNames snapshots the caller payload's field NAMES.
//
// Callers must take this snapshot at entry to the write operation, before any
// defaulting, before-save flow, or normalization has run — those stamp fields
// the user never sent, which is the conflation described above. Values are
// dropped here at the boundary, so no value ever enters the audit path.
func requestedFieldNames(record models.Record) []string {
	if len(record) == 0 {
		return nil
	}
	names := make([]string, 0, len(record))
	for name := range record {
		if strings.TrimSpace(name) == "" {
			continue
		}
		names = append(names, name)
	}
	sort.Strings(names)
	return names
}

// maxRecordedFieldNames bounds how many names one row can carry. An object
// with hundreds of fields is legitimate; an unbounded list is not, because
// ChangesJSON is LONGTEXT and these rows are retained for years.
const maxRecordedFieldNames = 200

// recordableFieldNames keeps only names that are DECLARED FIELDS on this
// object, and reports how many were dropped.
//
// This closes the one hole in the "field names, never values" guarantee. The
// guarantee holds trivially for `.Fields` / `.RuleName` off the typed errors,
// which are schema-derived. It does NOT hold for anything sourced from the
// caller's own map keys:
//
//   - The unknown-field bouncer builds its ValidationError with the RAW
//     caller-supplied key as `.Field` (kernel/pkg/validation/schema_validator.go)
//     — the one rejection stage where `.Field` is by construction not a
//     field name.
//   - The CSV bulk path turns file headers straight into map keys
//     (bulk_processor.go). A headerless or off-by-one upload — routine — makes
//     the first DATA row the header, so every key is a customer value:
//     emails, phone numbers, whatever the file holds.
//
// Recording those verbatim would put exactly the PII the issue forbids into a
// long-retention table. Membership in the schema is the only trustworthy
// judge — a shape check would still pass `Jane`. Unknown names collapse to a
// count, which keeps the forensic signal ("the caller sent 3 fields this
// object doesn't have") without persisting the strings.
//
// A nil schema (metadata unavailable) drops every name rather than guessing.
// Losing the field list on an audit row is recoverable; leaking customer data
// into a 78-month table is not.
func (ps *PersistenceService) recordableFieldNames(ctx context.Context, objectName string, names []string) ([]string, int) {
	if len(names) == 0 {
		return nil, 0
	}
	var schema *models.ObjectMetadata
	if ps != nil && ps.metadata != nil {
		schema = ps.metadata.GetSchema(ctx, objectName)
	}
	if schema == nil {
		return nil, len(names)
	}
	kept := make([]string, 0, len(names))
	unknown := 0
	for _, name := range names {
		if schemaDeclaresField(schema, name) {
			kept = append(kept, name)
			continue
		}
		unknown++
	}
	if len(kept) > maxRecordedFieldNames {
		unknown += len(kept) - maxRecordedFieldNames
		kept = kept[:maxRecordedFieldNames]
	}
	if len(kept) == 0 {
		return nil, unknown
	}
	return kept, unknown
}

// schemaDeclaresField reports whether name is a declared field on schema,
// matching the case-insensitive lookup the write paths themselves use.
func schemaDeclaresField(schema *models.ObjectMetadata, name string) bool {
	lowered := strings.ToLower(strings.TrimSpace(name))
	if lowered == "" {
		return false
	}
	if schema.FieldsMap != nil {
		if _, ok := schema.FieldsMap[lowered]; ok {
			return true
		}
		// FieldsMap is authoritative when populated; fall through to the
		// slice only when it is absent so a partially built schema cannot
		// silently drop every name.
		if len(schema.FieldsMap) > 0 {
			return false
		}
	}
	for i := range schema.Fields {
		if strings.EqualFold(schema.Fields[i].APIName, name) {
			return true
		}
	}
	return false
}

func sortedCopy(in []string) []string {
	if len(in) == 0 {
		return nil
	}
	out := append([]string(nil), in...)
	sort.Strings(out)
	return out
}

func nonEmptyFieldNames(name string) []string {
	if strings.TrimSpace(name) == "" {
		return nil
	}
	return []string{name}
}

// --- one rejection, one row ------------------------------------------------
//
// The hooks sit on the write entry points, but an entry point is not the same
// thing as a user's save: the engine re-enters them. A before-save Flow whose
// RecordCreate element writes a child object calls Insert again from inside
// the parent's Insert, and publishRecordEvent deliberately unwraps a typed
// business error out of the EventBus envelope (persistence_events.go) so REST
// can render it as a 400. The consequence is that the SAME error object
// reaches the parent's hook still classifiable, and both frames record it.
//
// Observed before this guard existed: a child rejected by a rule that exists
// only on the child produced two rows, and the second named the child's rule
// against the PARENT object. Support querying the parent then chases a rule
// that object does not have — which defeats the point of the feature more
// thoroughly than the original silence did.
//
// The invariant: a rejection is recorded ONCE, by the write that actually
// evaluated it. The innermost frame classifies and claims; outer frames that
// merely propagate the same error skip.
//
// The claim rides on ctx as a POINTER, so a nested frame's write is visible to
// the outer frame's deferred read even though ctx itself only flows downward.
// Same shape as insertFinalImageObservation in persistence_insert.go.

type rejectedSaveClaimKey struct{}

// rejectedSaveClaim tracks which errors have already produced a row.
//
// The recent-set is bounded because a bulk call records many distinct errors
// and this must not become a per-request leak. A bounded set is sufficient:
// the hazard is an outer frame re-seeing an error an inner frame just
// recorded, and frames unwind immediately.
type rejectedSaveClaim struct {
	mu     sync.Mutex
	recent []error
	// total counts every claim taken, including ones aged out of recent.
	total int
}

const rejectedSaveClaimDepth = 16

func withRejectedSaveClaim(ctx context.Context) context.Context {
	if ctx == nil {
		ctx = context.Background()
	}
	if _, ok := ctx.Value(rejectedSaveClaimKey{}).(*rejectedSaveClaim); ok {
		// An outer write already installed one; inherit it so this frame's
		// emission is visible to that outer frame.
		return ctx
	}
	return context.WithValue(ctx, rejectedSaveClaimKey{}, &rejectedSaveClaim{})
}

func rejectedSaveClaimFrom(ctx context.Context) *rejectedSaveClaim {
	if ctx == nil {
		return nil
	}
	claim, _ := ctx.Value(rejectedSaveClaimKey{}).(*rejectedSaveClaim)
	return claim
}

// recordedCount is how many rejections have been claimed under this holder.
//
// Callers that act as a LAST-RESORT hook (the external-ID upsert) compare it
// across their own span instead of relying on identity dedupe, because an
// error can be re-wrapped on the way out in a way that deliberately cuts the
// cause chain — the adoption guard does exactly that, so the wrapper is not
// errors.Is the inner error and would otherwise be recorded a second time.
func (c *rejectedSaveClaim) recordedCount() int {
	if c == nil {
		return 0
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.total
}

// claim reports whether this error still needs a row, marking it as taken.
// Returns false when some inner frame already recorded it.
func (c *rejectedSaveClaim) claim(err error) bool {
	if c == nil || err == nil {
		return true
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	for _, recorded := range c.recent {
		// errors.Is compares by identity first, which is exactly the case
		// here: the propagating frames share one error object.
		if errors.Is(err, recorded) {
			return false
		}
	}
	c.recent = append(c.recent, err)
	c.total++
	if len(c.recent) > rejectedSaveClaimDepth {
		c.recent = c.recent[len(c.recent)-rejectedSaveClaimDepth:]
	}
	return true
}

// --- what counts as a user's save ------------------------------------------
//
// Two engine-internal writes reach the same entry points a user's save does,
// and recording them would put rows in a forensics table for things no user
// did. Both are suppressed by reading a marker the caller already sets or
// sets in one line — deliberately NOT by building a general "was this a user
// save" framework.
//
// 1. DERIVED RECOMPUTE. UpdateDerived / UpdateDerivedCountRollupValues route
//    a rollup recompute through Update. Nobody pressed a button: the platform
//    is reconciling a parent total after a child changed. Its own doc says it
//    "is not a user-driven update path". Both already mark the context via
//    withDerivedRecompute, so this reads the existing marker rather than
//    adding one.
//
// 2. EXPECTED UNIQUENESS PROBE. The external-ID upsert path does not look
//    before it writes — it attempts the INSERT and uses the uniqueness
//    conflict as its signal that the record already exists, then switches to
//    an UPDATE. That conflict is the mechanism working correctly, and the
//    upsert as a whole usually SUCCEEDS, so a FAILURE row would describe a
//    save that landed fine.
//
//    The suppression is narrow on purpose: it silences ONLY the uniqueness
//    stage. If that same INSERT is refused for a missing required field or a
//    validation rule, the upsert really does fail, the caller really does see
//    an error, and the row is recorded exactly as it should be.

type expectedUniquenessProbeKey struct{}

// withExpectedUniquenessConflict marks a write whose uniqueness conflict is a
// planned outcome rather than a user-visible refusal. Scoped to the single
// attempt the caller wraps; it does not propagate to unrelated writes because
// the caller derives a fresh context for that attempt.
func withExpectedUniquenessConflict(ctx context.Context) context.Context {
	return context.WithValue(ctx, expectedUniquenessProbeKey{}, true)
}

// withoutExpectedUniquenessConflict clears the probe marker for everything
// DOWNSTREAM of the write it was set for.
//
// The marker says "this ONE attempt's uniqueness conflict is planned". A
// context value propagates to every nested write, so without this clear, a
// before-save flow that inserts a child record would have that child's REAL
// uniqueness conflict silenced too — the suppression would leak from one
// attempt into an unbounded subtree, and the blind spot #4261 exists to close
// would reopen underneath it.
//
// The write that owns the marker still sees it: its own hook reads the
// context it was called with, while only the derived context it passes
// downstream is cleared.
func withoutExpectedUniquenessConflict(ctx context.Context) context.Context {
	if ctx == nil || !expectsUniquenessConflict(ctx) {
		return ctx
	}
	return context.WithValue(ctx, expectedUniquenessProbeKey{}, false)
}

func expectsUniquenessConflict(ctx context.Context) bool {
	if ctx == nil {
		return false
	}
	expected, _ := ctx.Value(expectedUniquenessProbeKey{}).(bool)
	return expected
}

// rejectedSaveSuppressed reports whether this rejection belongs to an
// engine-internal write rather than a user's save.
func rejectedSaveSuppressed(ctx context.Context, stage string) bool {
	if ctx == nil {
		return false
	}
	if isDerivedRecomputeWrite(ctx) {
		return true
	}
	return stage == rejectionStageUniqueness && expectsUniquenessConflict(ctx)
}

// rejectedSaveAuditEnabled reports whether this graph produces FAILURE rows.
func (ps *PersistenceService) rejectedSaveAuditEnabled() bool {
	return ps != nil && ps.rejectedSaveAuditSink != nil && !ps.rejectedSaveAuditDisabled
}

// rejectedSaveRequestedFields is the snapshot helper the write entry points
// call. It exists so the snapshot is not taken at all when recording is off.
//
// The snapshot has to be EAGER — taken before defaults, before-save flows,
// and system-field stamping mutate the record — which means it runs on the
// success path too, where it is pure overhead. Gating it here keeps that cost
// at zero for deployments that have turned recording off, and keeps it to one
// small slice per write when it is on. It matters most on the bulk path,
// where the snapshot is per record across the whole batch.
func (ps *PersistenceService) rejectedSaveRequestedFields(record models.Record) []string {
	if !ps.rejectedSaveAuditEnabled() {
		return nil
	}
	return requestedFieldNames(record)
}

// emitRejectedSaveAudit records one FAILURE row for a save the validation
// pipeline refused. It is the ONLY production caller of
// audit.FailureSink.EmitFailure, and it owns the best-effort guarantee:
//
//	IT NEVER RETURNS AN ERROR AND NEVER PANICS INTO THE CALLER.
//
// Callers invoke it from a deferred block on the write entry point, so it
// runs after the business transaction has already concluded — the row is
// written out-of-band and survives the rollback that accompanies the
// rejection.
//
// action is the verb the caller ATTEMPTED (CREATE / UPDATE / UNDELETE). The
// existing AuditAction vocabulary is reused unchanged; Outcome=FAILURE is the
// discriminator, so no new enum value and no new column are introduced.
func (ps *PersistenceService) emitRejectedSaveAudit(
	ctx context.Context,
	objectName string,
	recordID string,
	action audit.AuditAction,
	requestedFields []string,
	rejectionErr error,
	currentUser *models.UserSession,
) {
	// A panic in a deferred audit hook would replace the user's validation
	// error with a crash — the exact failure mode this contract exists to
	// prevent. Recover before anything else runs.
	defer func() {
		if r := recover(); r != nil {
			ps.logRejectedSaveAuditProblem(objectName, "rejected-save audit emission panicked", r)
		}
	}()

	if !ps.rejectedSaveAuditEnabled() {
		return
	}
	// Recursion is impossible by construction — the sink writes raw SQL and
	// never re-enters PersistenceService — but system tables are skipped for
	// the same reason the CREATE/DELETE emitters skip them: their mutations
	// are platform bookkeeping, not user saves, and _System_AuditLog itself
	// must never audit-log its own writes.
	if constants.IsSystemTable(objectName) && objectName != constants.TableUser {
		return
	}

	rejection, ok := classifyRejectedSave(rejectionErr)
	if !ok {
		return
	}
	// Engine-internal writes are not user saves; see rejectedSaveSuppressed.
	if rejectedSaveSuppressed(ctx, rejection.stage) {
		return
	}
	// One rejection, one row. If an inner write already recorded this exact
	// error, this frame is only propagating it — see the claim comment above.
	if !rejectedSaveClaimFrom(ctx).claim(rejectionErr) {
		return
	}

	// Both name lists are filtered against the schema before they are
	// persisted. `rejection.fields` is usually schema-derived already, but the
	// unknown-field bouncer routes a raw caller key through it, and
	// requestedFields is caller map keys by definition. See
	// recordableFieldNames.
	namedFields, unknownNamed := ps.recordableFieldNames(ctx, objectName, rejection.fields)
	recordableRequested, unknownRequested := ps.recordableFieldNames(ctx, objectName, requestedFields)

	changes := rejectedSaveChanges{
		RejectionStage:    rejection.stage,
		Fields:            namedFields,
		UnknownFieldCount: unknownNamed + unknownRequested,
		Rule:              rejection.rule,
		RequestedFields:   recordableRequested,
	}
	changesJSON, err := json.Marshal(changes)
	if err != nil {
		// Losing the payload is not a reason to lose the row: stage and
		// ErrorCode still answer "why was this refused".
		ps.logRejectedSaveAuditProblem(objectName, "rejected-save audit payload marshal failed", err)
		changesJSON = nil
	}

	if strings.TrimSpace(recordID) == "" {
		recordID = audit.RecordIdUnassigned
	}

	event := audit.AuditEvent{
		ObjectApiName: objectName,
		RecordId:      recordID,
		Action:        action,
		ChangedById:   rejectedSaveChangedByID(currentUser),
		Entrypoint:    audit.EntrypointFromContext(ctx),
		Outcome:       audit.OutcomeFailure,
		ErrorCode:     rejection.errorCode,
		ChangesJSON:   string(changesJSON),
		RequestId:     requestIDFromCtx(ctx),
	}

	emitCtx, cancel := rejectedSaveAuditContext(ctx)
	defer cancel()
	if err := ps.rejectedSaveAuditSink.EmitFailure(emitCtx, event); err != nil {
		ps.logRejectedSaveAuditProblem(objectName, "rejected-save audit emission failed", err)
	}
}

// rejectedSaveAuditEmitTimeout bounds one emission's database work.
//
// Short on purpose: this runs on the request path, and a row that cannot be
// written within a couple of seconds is not worth making the user wait for.
// Losing the row is the acceptable failure; blocking the request is not.
const rejectedSaveAuditEmitTimeout = 2 * time.Second

// rejectedSaveAuditContext detaches the emission from the request's
// cancellation, then re-bounds it with its own deadline.
//
// Both halves are load-bearing and they pull in opposite directions:
//
//   - WITHOUT the detach, a save rejected on a request that then times out or
//     whose client disconnects loses its FAILURE row — forensics would vanish
//     precisely in the incident case they exist for.
//   - WITHOUT the new deadline, the detach removes the only bound there was.
//     The emission is synchronous on the request path, so a saturated or
//     unavailable audit pool would hang every rejected request indefinitely
//     and exhaust request goroutines — turning "best effort" into an outage,
//     which is a worse version of the failure mode this contract exists to
//     prevent. Do not assume the pool or driver supplies a timeout; nothing
//     in this path configures one.
//
// The caller is responsible for cancel(): emission is synchronous, so the
// returned context does not outlive it.
func rejectedSaveAuditContext(ctx context.Context) (context.Context, context.CancelFunc) {
	if ctx == nil {
		ctx = context.Background()
	}
	return context.WithTimeout(context.WithoutCancel(ctx), rejectedSaveAuditEmitTimeout)
}

func rejectedSaveChangedByID(currentUser *models.UserSession) string {
	if currentUser != nil && strings.TrimSpace(currentUser.ID) != "" {
		return currentUser.ID
	}
	return constants.SystemUserID
}

// logRejectedSaveAuditProblem is the single place audit-path breakage becomes
// visible. It must stay total: a nil logger degrades to silence rather than
// panicking back into the request.
func (ps *PersistenceService) logRejectedSaveAuditProblem(objectName, message string, cause interface{}) {
	if ps == nil || ps.logger == nil {
		return
	}
	ps.logger.
		WithField("object", objectName).
		WithField("cause", fmt.Sprintf("%v", cause)).
		Warn(message)
}